[CALCITE-3963] Maintains logical properties at RelSet (equivalent gro…

cassandra/src/test/java/org/apache/calcite/test/CassandraAdapterTest.java

@@ -100,8 +100,8 @@ static void load(Session session) {
         .returns("tweet_id=f3c329de-d05b-11e5-b58b-90e2ba530b12\n"
                + "tweet_id=f3dbb03a-d05b-11e5-b58b-90e2ba530b12\n")
         .explainContains("PLAN=CassandraToEnumerableConverter\n"
-                + "  CassandraLimit(fetch=[2])\n"
-                + "    CassandraProject(tweet_id=[$2])\n"
+                + "  CassandraProject(tweet_id=[$2])\n"
+                + "    CassandraLimit(fetch=[2])\n"
                 + "      CassandraFilter(condition=[=($0, '!PUBLIC!')])\n");
   }
 

core/src/main/java/org/apache/calcite/plan/AbstractRelOptPlanner.java

@@ -225,6 +225,10 @@ public long getRelMetadataTimestamp(RelNode rel) {
   @Override public void prune(RelNode rel) {
   }
 
+  @Override public boolean isPruned(RelNode rel) {
+    return false;
+  }
+
   public void registerClass(RelNode node) {
     final Class<? extends RelNode> clazz = node.getClass();
     if (classes.add(clazz)) {

core/src/main/java/org/apache/calcite/plan/RelOptPlanner.java

@@ -306,6 +306,13 @@ RelNode register(
    */
   void prune(RelNode rel);
 
+  /**
+   * Check whether or not a RelNode is pruned
+   * @param rel the node to check
+   * @return true if rel is pruned, false if otherwise
+   */
+  boolean isPruned(RelNode rel);
+
   /**
    * Registers a class of RelNode. If this class of RelNode has been seen
    * before, does nothing.

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

@@ -23,6 +23,7 @@
 import org.apache.calcite.plan.RelTrait;
 import org.apache.calcite.plan.RelTraitDef;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.PhysicalNode;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.convert.Converter;
@@ -73,7 +74,14 @@ class RelSet {
    * set.
    */
   RelSet equivalentSet;
-  RelNode rel;
+
+  /**
+   * The first RelNode added to the set or the RelNode with highest confidence
+   * level of estimated statistics.
+   * The logical properties of the RelSet, including row count, uniqueness, etc,
+   * are determined by this RelNode.
+   */
+  RelNode originalRel;
 
   /**
    * The position indicator of rel node that is to be processed.
@@ -211,6 +219,7 @@ public RelSubset add(RelNode rel) {
     final RelSubset subset = getOrCreateSubset(
         rel.getCluster(), traitSet, rel.isEnforcer());
     subset.add(rel);
+    checkAndUpdateOriginalRel(rel);
     return subset;
   }
 
@@ -350,15 +359,16 @@ void addInternal(RelNode rel) {
         postEquivalenceEvent(planner, rel);
       }
     }
-    if (this.rel == null) {
-      this.rel = rel;
+    if (this.originalRel == null) {
+      this.originalRel = rel;
     } else {
       // Row types must be the same, except for field names.
       RelOptUtil.verifyTypeEquivalence(
-          this.rel,
+          this.originalRel,
           rel,
           this);
     }
+    checkAndUpdateOriginalRel(rel);
   }
 
   /**
@@ -383,7 +393,7 @@ void mergeWith(
     assert otherSet.equivalentSet == null;
     LOGGER.trace("Merge set#{} into set#{}", otherSet.id, id);
     otherSet.equivalentSet = this;
-    RelOptCluster cluster = rel.getCluster();
+    RelOptCluster cluster = originalRel.getCluster();
     RelMetadataQuery mq = cluster.getMetadataQuery();
 
     // remove from table
@@ -486,4 +496,50 @@ void mergeWith(
       planner.fireRules(subset);
     }
   }
+
+  private void checkAndUpdateOriginalRel(RelNode newRel) {
+    if (newRel instanceof LogicalNode && this.originalRel instanceof LogicalNode
+        && ((LogicalNode) newRel).getStatsEstimateConfidence().compareTo(
+            ((LogicalNode) this.originalRel).getStatsEstimateConfidence()) > 0) {
+      this.originalRel = newRel;
+
+      // Invalidate parent sets original rel meta cache since child set properties might
+      // have been changed.
+      HashSet<RelSet> newParentSets = getParentSets();
+      HashSet<RelSet> allParentSets = new HashSet<>();
+      while (!newParentSets.isEmpty()) {
+        allParentSets.addAll(newParentSets);
+        HashSet<RelSet> validParents = new HashSet<>();
+        for (RelSet set : newParentSets) {
+          HashSet<RelSet> parents = set.getParentSets();
+          for (RelSet s : parents) {
+            // To handle cycle references between sets
+            if (!allParentSets.contains(s)) {
+              validParents.add(s);
+            }
+          }
+        }
+        newParentSets = validParents;
+      }
+      for (RelSet set : allParentSets) {
+        final RelMetadataQuery mq = set.originalRel.getCluster().getMetadataQuery();
+        mq.clearCache(set.originalRel);
+      }
+    }
+  }
+
+  private HashSet<RelSet> getParentSets() {
+    HashSet<RelSet> results = new HashSet<>();
+    VolcanoPlanner planner = (VolcanoPlanner) this.originalRel.getCluster().getPlanner();
+    assert planner != null;
+
+    for (RelNode rel : getParentRels()) {
+      RelSet set = planner.getSet(rel);
+      if (set.id != this.id && !planner.isPruned(rel)) {
+        results.add(set);
+      }
+    }
+
+    return results;
+  }
 }

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

@@ -185,7 +185,7 @@ public RelNode getBest() {
   }
 
   public RelNode getOriginal() {
-    return set.rel;
+    return set.originalRel;
   }
 
   public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
@@ -207,7 +207,7 @@ public double estimateRowCount(RelMetadataQuery mq) {
     if (best != null) {
       return mq.getRowCount(best);
     } else {
-      return mq.getRowCount(set.rel);
+      return mq.getRowCount(set.originalRel);
     }
   }
 
@@ -234,7 +234,7 @@ public double estimateRowCount(RelMetadataQuery mq) {
   }
 
   @Override protected RelDataType deriveRowType() {
-    return set.rel.getRowType();
+    return set.originalRel.getRowType();
   }
 
   /**
@@ -318,7 +318,7 @@ void add(RelNode rel) {
 
     // If this isn't the first rel in the set, it must have compatible
     // row type.
-    if (set.rel != null) {
+    if (set.originalRel != null) {
       RelOptUtil.equal("rowtype of new rel", rel.getRowType(),
           "rowtype of set", getRowType(), Litmus.THROW);
     }

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

@@ -848,6 +848,10 @@ RelNode changeTraitsUsingConverters(
     prunedNodes.add(rel);
   }
 
+  @Override public boolean isPruned(RelNode rel) {
+    return prunedNodes.contains(rel);
+  }
+
   /**
    * Dumps the internal state of this VolcanoPlanner to a writer.
    *

core/src/main/java/org/apache/calcite/rel/LogicalNode.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.rel;
+
+/**
+ * Represents a logical node in Calcite
+ */
+public interface LogicalNode extends RelNode {
+
+  /**
+   * Confidence levels of statistics estimation
+   */
+  enum StatsEstimateConfidenceLevel {
+    None,
+    Low,
+    Medium,
+    High
+  }
+
+  /**
+   * Get confidence level of statistics estimation
+   * @return Confidence level of statistics estimation
+   */
+  default StatsEstimateConfidenceLevel getStatsEstimateConfidence() {
+    return StatsEstimateConfidenceLevel.Medium;
+  }
+}

core/src/main/java/org/apache/calcite/rel/logical/LogicalAggregate.java

@@ -19,6 +19,7 @@
 import org.apache.calcite.plan.Convention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelInput;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelShuttle;
@@ -43,7 +44,7 @@
  * <li>{@link org.apache.calcite.rel.rules.AggregateReduceFunctionsRule}.
  * </ul>
  */
-public final class LogicalAggregate extends Aggregate {
+public final class LogicalAggregate extends Aggregate implements LogicalNode {
   //~ Constructors -----------------------------------------------------------
 
   /**

core/src/main/java/org/apache/calcite/rel/logical/LogicalCalc.java

@@ -20,6 +20,7 @@
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelOptUtil;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelCollationTraitDef;
 import org.apache.calcite.rel.RelDistributionTraitDef;
@@ -63,7 +64,7 @@
  *     merges two {@code LogicalCalc}s
  * </ul>
  */
-public final class LogicalCalc extends Calc {
+public final class LogicalCalc extends Calc implements LogicalNode {
   //~ Static fields/initializers ---------------------------------------------
 
   //~ Constructors -----------------------------------------------------------

core/src/main/java/org/apache/calcite/rel/logical/LogicalCorrelate.java

@@ -20,6 +20,7 @@
 import org.apache.calcite.plan.Convention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelInput;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelShuttle;
@@ -41,7 +42,7 @@
  *
  * @see org.apache.calcite.rel.core.CorrelationId
  */
-public final class LogicalCorrelate extends Correlate {
+public final class LogicalCorrelate extends Correlate implements LogicalNode {
   //~ Instance fields --------------------------------------------------------
 
   //~ Constructors -----------------------------------------------------------

core/src/main/java/org/apache/calcite/rel/logical/LogicalExchange.java

@@ -19,6 +19,7 @@
 import org.apache.calcite.plan.Convention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelDistribution;
 import org.apache.calcite.rel.RelDistributionTraitDef;
 import org.apache.calcite.rel.RelInput;
@@ -30,7 +31,7 @@
  * Sub-class of {@link Exchange} not
  * targeted at any particular engine or calling convention.
  */
-public final class LogicalExchange extends Exchange {
+public final class LogicalExchange extends Exchange implements LogicalNode {
   private LogicalExchange(RelOptCluster cluster, RelTraitSet traitSet,
       RelNode input, RelDistribution distribution) {
     super(cluster, traitSet, input, distribution);

core/src/main/java/org/apache/calcite/rel/logical/LogicalFilter.java

@@ -19,6 +19,7 @@
 import org.apache.calcite.plan.Convention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelCollationTraitDef;
 import org.apache.calcite.rel.RelDistributionTraitDef;
 import org.apache.calcite.rel.RelInput;
@@ -42,7 +43,8 @@
  * Sub-class of {@link org.apache.calcite.rel.core.Filter}
  * not targeted at any particular engine or calling convention.
  */
-public final class LogicalFilter extends Filter {
+public final class LogicalFilter extends Filter implements LogicalNode {
+
   private final ImmutableSet<CorrelationId> variablesSet;
 
   //~ Constructors -----------------------------------------------------------

core/src/main/java/org/apache/calcite/rel/logical/LogicalIntersect.java

@@ -19,6 +19,7 @@
 import org.apache.calcite.plan.Convention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelInput;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelShuttle;
@@ -30,7 +31,7 @@
  * Sub-class of {@link org.apache.calcite.rel.core.Intersect}
  * not targeted at any particular engine or calling convention.
  */
-public final class LogicalIntersect extends Intersect {
+public final class LogicalIntersect extends Intersect implements LogicalNode {
   //~ Constructors -----------------------------------------------------------
 
   /**

core/src/main/java/org/apache/calcite/rel/logical/LogicalJoin.java

@@ -19,6 +19,7 @@
 import org.apache.calcite.plan.Convention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelInput;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelShuttle;
@@ -54,7 +55,7 @@
  *
  * </ul>
  */
-public final class LogicalJoin extends Join {
+public final class LogicalJoin extends Join implements LogicalNode {
   //~ Instance fields --------------------------------------------------------
 
   // NOTE jvs 14-Mar-2006:  Normally we don't use state like this

core/src/main/java/org/apache/calcite/rel/logical/LogicalMatch.java

@@ -19,6 +19,7 @@
 import org.apache.calcite.plan.Convention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelShuttle;
@@ -35,7 +36,7 @@
  * Sub-class of {@link Match}
  * not targeted at any particular engine or calling convention.
  */
-public class LogicalMatch extends Match {
+public class LogicalMatch extends Match implements LogicalNode {
 
   /**
    * Creates a LogicalMatch.

core/src/main/java/org/apache/calcite/rel/logical/LogicalMinus.java

@@ -19,6 +19,7 @@
 import org.apache.calcite.plan.Convention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.LogicalNode;
 import org.apache.calcite.rel.RelInput;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelShuttle;
@@ -30,7 +31,7 @@
  * Sub-class of {@link org.apache.calcite.rel.core.Minus}
  * not targeted at any particular engine or calling convention.
  */
-public final class LogicalMinus extends Minus {
+public final class LogicalMinus extends Minus implements LogicalNode {
   //~ Constructors -----------------------------------------------------------
 
   /**