[CALCITE-2018] Queries failed with AssertionError: rel has lower cost…

… than best cost of subset. (Inspired from pr: apache#552 which is committed by vvysotskyi).

Summary:
1.Replaced Map usage by Table for query metadata caching. It allows finding cached metadata query for concrete RelNode instance more easily.
2. Cached metadata values are removed for current RelSubset and all parent RelNodes when RelSubset.propagateCostImprovements() is called and best rel node was changed.
3. Made changes in RelSet.mergeWith() method to call RelSubset.propagateCostImprovements() when resulting best is known instead of just assigning RelSubset.best.
Note:
1. Does not import following points from pr because it will increase the lifetime of RelMetadataQuery.
Made changes to use single RelMetadataQuery instance during planning between the first call of RelOptCluster.getMetadataQuery() and RelOptCluster.invalidateMetadataQuery() call.

Test Plan: NO

Reviewers: 鲁尼, 晓令, 玉兆

Reviewed By: 晓令

Subscribers: P577102

Differential Revision: https://aone.alibaba-inc.com/code/D642962

core/src/main/java/org/apache/calcite/plan/hep/HepPlanner.java

@@ -61,6 +61,8 @@
 import java.util.Map;
 import java.util.Set;
 
+import static org.apache.calcite.rel.metadata.RelMdUtil.clearCache;
+
 /**
  * HepPlanner is a heuristic implementation of the {@link RelOptPlanner}
  * interface.
@@ -866,6 +868,7 @@ private void contractVertices(
         }
         parentRel.replaceInput(i, preservedVertex);
       }
+      clearCache(parentRel);
       graph.removeEdge(parent, discardedVertex);
       graph.addEdge(parent, preservedVertex);
       updateVertex(parent, parentRel);
@@ -930,9 +933,9 @@ private RelNode buildFinalPlan(HepRelVertex vertex) {
       }
       child = buildFinalPlan((HepRelVertex) child);
       rel.replaceInput(i, child);
-      rel.recomputeDigest();
     }
-
+    clearCache(rel);
+    rel.recomputeDigest();
     return rel;
   }
 

core/src/main/java/org/apache/calcite/plan/volcano/RelSet.java

@@ -33,7 +33,9 @@
 
 import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.IdentityHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -316,16 +318,18 @@ void mergeWith(
     boolean existed = planner.allSets.remove(otherSet);
     assert existed : "merging with a dead otherSet";
 
+    final Map<RelSubset, RelNode> changedSubsets = new IdentityHashMap();
+
     // merge subsets
     for (RelSubset otherSubset : otherSet.subsets) {
       planner.ruleQueue.subsetImportances.remove(otherSubset);
       RelSubset subset =
           getOrCreateSubset(
               otherSubset.getCluster(),
               otherSubset.getTraitSet());
+      // collect RelSubset instances, whose best should be changed
       if (otherSubset.bestCost.isLt(subset.bestCost)) {
-        subset.bestCost = otherSubset.bestCost;
-        subset.best = otherSubset.best;
+        changedSubsets.put(subset, otherSubset.best);
       }
       for (RelNode otherRel : otherSubset.getRels()) {
         planner.reregister(this, otherRel);
@@ -335,6 +339,19 @@ void mergeWith(
     // Has another set merged with this?
     assert equivalentSet == null;
 
+    // calls propagateCostImprovements() for RelSubset instances,
+    // whose best should be changed to checks whether that
+    // subset's parents have gotten cheaper.
+    final Set<RelSubset> activeSet = new HashSet<>();
+    final RelMetadataQuery mq = rel.getCluster().getMetadataQuery();
+    for (Map.Entry<RelSubset, RelNode> subsetBestPair : changedSubsets.entrySet()) {
+      RelSubset relSubset = subsetBestPair.getKey();
+      relSubset.propagateCostImprovements(
+              planner, mq, subsetBestPair.getValue(),
+              activeSet);
+    }
+    assert activeSet.isEmpty();
+
     // Update all rels which have a child in the other set, to reflect the
     // fact that the child has been renamed.
     //
@@ -353,8 +370,6 @@ void mergeWith(
     }
 
     // Make sure the cost changes as a result of merging are propagated.
-    final Set<RelSubset> activeSet = new HashSet<>();
-    final RelMetadataQuery mq = rel.getCluster().getMetadataQuery();
     for (RelNode parentRel : getParentRels()) {
       final RelSubset parentSubset = planner.getSubset(parentRel);
       parentSubset.propagateCostImprovements(

core/src/main/java/org/apache/calcite/plan/volcano/RelSubset.java

@@ -200,7 +200,8 @@ Set<RelNode> getParents() {
     final Set<RelNode> list = new LinkedHashSet<>();
     for (RelNode parent : set.getParentRels()) {
       for (RelSubset rel : inputSubsets(parent)) {
-        if (rel.set == set && traitSet.satisfies(rel.getTraitSet())) {
+        // see usage of this method in propagateCostImprovements0()
+        if (rel == this) {
           list.add(parent);
         }
       }
@@ -342,12 +343,17 @@ void propagateCostImprovements0(VolcanoPlanner planner, RelMetadataQuery mq,
 
         bestCost = cost;
         best = rel;
+        // since best was changed, cached metadata for this subset should be removed
+        mq.clearCache(this);
 
         // Lower cost means lower importance. Other nodes will change
         // too, but we'll get to them later.
         planner.ruleQueue.recompute(this);
         for (RelNode parent : getParents()) {
+          // removes parent cached metadata since its input was changed
+          mq.clearCache(parent);
           final RelSubset parentSubset = planner.getSubset(parent);
+          // parent subset will clear its cache in propagateCostImprovements0 method itself
           parentSubset.propagateCostImprovements(planner, mq, parent,
               activeSet);
         }

core/src/main/java/org/apache/calcite/plan/volcano/VolcanoPlanner.java

@@ -91,6 +91,8 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import static org.apache.calcite.rel.metadata.RelMdUtil.clearCache;
+
 /**
  * VolcanoPlanner optimizes queries by transforming expressions selectively
  * according to a dynamic programming algorithm.
@@ -858,10 +860,9 @@ public RelSubset register(
     }
     final RelSubset subset = registerImpl(rel, set);
 
-    // FIXME https://aone.alibaba-inc.com/task/14681173
-    // if (LOGGER.isDebugEnabled()) {
-    //  validate();
-    // }
+    if (LOGGER.isDebugEnabled()) {
+      validate();
+    }
 
     return subset;
   }
@@ -898,13 +899,17 @@ protected void validate() {
               + "] is in wrong set [" + set + "]");
         }
         for (RelNode rel : subset.getRels()) {
-          RelOptCost relCost = getCost(rel, rel.getCluster().getMetadataQuery());
-          if (relCost.isLt(subset.bestCost)) {
-            throw new AssertionError(
-                "rel [" + rel.getDescription()
-                + "] has lower cost " + relCost
-                + " than best cost " + subset.bestCost
-                + " of subset [" + subset.getDescription() + "]");
+          // FIXME https://issues.apache.org/jira/browse/CALCITE-2166,
+          // https://aone.alibaba-inc.com/task/16508844
+          if (rel != subset.best) {
+            RelOptCost relCost = getCost(rel, rel.getCluster().getMetadataQuery());
+            if (relCost.isLt(subset.bestCost)) {
+              throw new AssertionError(
+                      "rel [" + rel.getDescription()
+                      + "] has lower cost " + relCost
+                      + " than best cost " + subset.bestCost
+                      + " of subset [" + subset.getDescription() + "]");
+            }
           }
         }
       }
@@ -1372,6 +1377,7 @@ private boolean fixUpInputs(RelNode rel) {
         }
       }
     }
+    clearCache(rel);
     return changeCount > 0;
   }
 

core/src/main/java/org/apache/calcite/rel/metadata/JaninoRelMetadataProvider.java

@@ -266,10 +266,9 @@ private static <M extends Metadata> MetadataHandler<M> load3(
             .append(method.i)
             .append(")");
       }
-      buff.append(", r");
       safeArgList(buff, method.e)
           .append(");\n")
-          .append("    final Object v = mq.map.get(key);\n")
+          .append("    final Object v = mq.map.get(r, key);\n")
           .append("    if (v != null) {\n")
           .append("      if (v == ")
           .append(NullSentinel.class.getName())
@@ -287,7 +286,7 @@ private static <M extends Metadata> MetadataHandler<M> load3(
           .append(method.e.getReturnType().getName())
           .append(") v;\n")
           .append("    }\n")
-          .append("    mq.map.put(key,")
+          .append("    mq.map.put(r, key,")
           .append(NullSentinel.class.getName())
           .append(".ACTIVE);\n")
           .append("    try {\n")
@@ -298,14 +297,14 @@ private static <M extends Metadata> MetadataHandler<M> load3(
           .append("_(r, mq");
       argList(buff, method.e)
           .append(");\n")
-          .append("      mq.map.put(key, ")
+          .append("      mq.map.put(r, key, ")
           .append(NullSentinel.class.getName())
           .append(".mask(x));\n")
           .append("      return x;\n")
           .append("    } catch (")
           .append(Exception.class.getName())
           .append(" e) {\n")
-          .append("      mq.map.remove(key);\n")
+          .append("      mq.map.row(r).clear();\n")
           .append("      throw e;\n")
           .append("    }\n")
           .append("  }\n")

core/src/main/java/org/apache/calcite/rel/metadata/ReflectiveRelMetadataProvider.java

@@ -178,7 +178,7 @@ private static RelMetadataProvider reflectiveSource(
                   }
                   key1 = FlatLists.copyOf(args2);
                 }
-                if (mq.map.put(key1, NullSentinel.INSTANCE) != null) {
+                if (mq.map.put(rel, key1, NullSentinel.INSTANCE) != null) {
                   throw CyclicMetadataException.INSTANCE;
                 }
                 try {
@@ -188,7 +188,7 @@ private static RelMetadataProvider reflectiveSource(
                   Util.throwIfUnchecked(e.getCause());
                   throw new RuntimeException(e.getCause());
                 } finally {
-                  mq.map.remove(key1);
+                  mq.map.remove(rel, key1);
                 }
               });
       methodsMap.put(key, function);

core/src/main/java/org/apache/calcite/rel/metadata/RelMdUtil.java

@@ -850,6 +850,15 @@ public static boolean checkInputForCollationAndLimit(RelMetadataQuery mq,
     }
     return alreadySorted && alreadySmaller;
   }
+
+  /**
+   * Removes cached metadata values for specified RelNode.
+   *
+   * @param rel RelNode whose cached metadata should be removed
+   */
+  public static void clearCache(RelNode rel) {
+    rel.getCluster().getMetadataQuery().clearCache(rel);
+  }
 }
 
 // End RelMdUtil.java

core/src/main/java/org/apache/calcite/rel/metadata/RelMetadataQuery.java

@@ -27,15 +27,15 @@
 import org.apache.calcite.sql.SqlExplainLevel;
 import org.apache.calcite.util.ImmutableBitSet;
 
+import com.google.common.collect.HashBasedTable;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Multimap;
+import com.google.common.collect.Table;
 
 import java.lang.reflect.Proxy;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 
@@ -76,7 +76,7 @@
  */
 public class RelMetadataQuery {
   /** Set of active metadata queries, and cache of previous results. */
-  public final Map<List, Object> map = new HashMap<>();
+  public final Table<RelNode, List, Object> map = HashBasedTable.create();
 
   public final JaninoRelMetadataProvider metadataProvider;
 
@@ -838,6 +838,15 @@ public boolean isVisibleInExplain(RelNode rel,
     }
   }
 
+  /**
+   * Removes cached metadata values for specified RelNode.
+   *
+   * @param rel RelNode whose cached metadata should be removed
+   */
+  public void clearCache(RelNode rel) {
+    map.row(rel).clear();
+  }
+
   private static Double validatePercentage(Double result) {
     assert isPercentage(result, true);
     return result;