Attach IncorrectRuntimeStatsMarker on finding an illegal link

apache · Nov 7, 2024 · 2ca6f97 · 2ca6f97
1 parent fbd94d4
commit 2ca6f97
Show file tree

Hide file tree

Showing 4 changed files with 5 additions and 34 deletions.
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/hooks/RuntimeStatsPersistenceCheckerHook.java b/ql/src/java/org/apache/hadoop/hive/ql/hooks/RuntimeStatsPersistenceCheckerHook.java
@@ -38,10 +38,6 @@ public class RuntimeStatsPersistenceCheckerHook implements ExecuteWithHookContex
   public void run(HookContext hookContext) throws Exception {
 
     PlanMapper pm = ((PrivateHookContext) hookContext).getContext().getPlanMapper();
-    if (pm.isBroken()) {
-      LOG.warn("Skip checking signatures. The PlanMapper is broken");
-      return;
-    }
 
     List<OpTreeSignature> sigs = pm.getAll(OpTreeSignature.class);
 

diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/mapper/PlanMapper.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/mapper/PlanMapper.java
@@ -37,6 +37,7 @@
 import org.apache.hadoop.hive.ql.optimizer.signature.OpTreeSignatureFactory;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Sets;
+import org.apache.hadoop.hive.ql.stats.OperatorStats;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -48,9 +49,8 @@
 public class PlanMapper {
   private static final Logger LOG = LoggerFactory.getLogger(PlanMapper.class);
 
-  private final Set<EquivGroup> groups = new HashSet<>();
-  private final Map<Object, EquivGroup> objectMap = new CompositeMap<>(OpTreeSignature.class, AuxOpTreeSignature.class);
-  private boolean isBroken = false;
+  Set<EquivGroup> groups = new HashSet<>();
+  private Map<Object, EquivGroup> objectMap = new CompositeMap<>(OpTreeSignature.class, AuxOpTreeSignature.class);
 
   /**
    * Specialized class which can compare by identity or value; based on the key type.
@@ -222,7 +222,8 @@ private void link(Object o1, Object o2, boolean mayMerge) {
     if (mGroups.size() > 1) {
       if (!mayMerge) {
         LOG.warn("Illegally linking {} and {}", o1, o2);
-        isBroken = true;
+        mGroups.forEach(g -> g.add(new OperatorStats.IncorrectRuntimeStatsMarker()));
+        return;
       }
       EquivGroup newGrp = new EquivGroup();
       newGrp.add(o1);
@@ -253,15 +254,7 @@ private Object getKeyFor(Object o) {
     return o;
   }
 
-  public boolean isBroken() {
-    return isBroken;
-  }
-
   public <T> List<T> getAll(Class<T> clazz) {
-    if (isBroken) {
-      LOG.warn("PlanMapper#getAll is no longer valid. Please use PlanMapper#isBroken to handle the state correctly");
-      return Collections.emptyList();
-    }
     List<T> ret = new ArrayList<>();
     for (EquivGroup g : groups) {
       ret.addAll(g.getAll(clazz));
@@ -288,11 +281,6 @@ public <T> T lookup(Class<T> clazz, Object key) {
   }
 
   public Iterator<EquivGroup> iterateGroups() {
-    if (isBroken) {
-      LOG.warn("PlanMapper#iterateGroups is no longer valid. "
-          + "Please use PlanMapper#isBroken to handle the state correctly");
-      return Collections.emptyIterator();
-    }
     return groups.iterator();
 
   }

diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/mapper/StatsSources.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/mapper/StatsSources.java
@@ -77,11 +77,6 @@ public static StatsSource getStatsSourceContaining(StatsSource currentStatsSourc
 
   private static ImmutableList<PersistedRuntimeStats> extractStatsFromPlanMapper(PlanMapper pm) {
     Builder<PersistedRuntimeStats> li = ImmutableList.builder();
-    if (pm.isBroken()) {
-      LOG.warn("Don't generate any stats. This PlanMapper is broken");
-      return li.build();
-    }
-
     Iterator<EquivGroup> it = pm.iterateGroups();
     while (it.hasNext()) {
       EquivGroup e = it.next();

diff --git a/ql/src/java/org/apache/hadoop/hive/ql/reexec/ReOptimizePlugin.java b/ql/src/java/org/apache/hadoop/hive/ql/reexec/ReOptimizePlugin.java
@@ -102,14 +102,6 @@ public void prepareToReExecute() {
 
   @Override
   public boolean shouldReExecuteAfterCompile(int executionNum, PlanMapper oldPlanMapper, PlanMapper newPlanMapper) {
-    if (oldPlanMapper.isBroken() || newPlanMapper.isBroken()) {
-      LOG.warn(
-          "Giving up a re-execution. The old plan mapper is {}, and the new one is {}",
-          oldPlanMapper.isBroken() ? "broken" : "not broken",
-          newPlanMapper.isBroken() ? "broken" : "not broken");
-      return false;
-    }
-
     boolean planDidChange = !planEquals(oldPlanMapper, newPlanMapper);
     LOG.info("planDidChange: {}", planDidChange);
     return planDidChange;