[CALCITE-4967] Support SQL hints for temporal table join

close apache#2664

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCorrelate.java

@@ -52,7 +52,7 @@ public EnumerableCorrelate(RelOptCluster cluster, RelTraitSet traits,
       RelNode left, RelNode right,
       CorrelationId correlationId,
       ImmutableBitSet requiredColumns, JoinRelType joinType) {
-    super(cluster, traits, left, right, correlationId, requiredColumns,
+    super(cluster, traits, ImmutableList.of(), left, right, correlationId, requiredColumns,
         joinType);
   }
 

core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcRules.java

@@ -127,7 +127,7 @@ private JdbcRules() {
       };
 
   static final RelFactories.CorrelateFactory CORRELATE_FACTORY =
-      (left, right, correlationId, requiredColumns, joinType) -> {
+      (left, right, hints, correlationId, requiredColumns, joinType) -> {
         throw new UnsupportedOperationException("JdbcCorrelate");
       };
 

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

@@ -25,6 +25,8 @@
 import org.apache.calcite.rel.RelInput;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.hint.Hintable;
+import org.apache.calcite.rel.hint.RelHint;
 import org.apache.calcite.rel.metadata.RelMetadataQuery;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.sql.validate.SqlValidatorUtil;
@@ -68,12 +70,13 @@
  *
  * @see CorrelationId
  */
-public abstract class Correlate extends BiRel {
+public abstract class Correlate extends BiRel implements Hintable {
   //~ Instance fields --------------------------------------------------------
 
   protected final CorrelationId correlationId;
   protected final ImmutableBitSet requiredColumns;
   protected final JoinRelType joinType;
+  protected final ImmutableList<RelHint> hints;
 
   //~ Constructors -----------------------------------------------------------
 
@@ -91,6 +94,7 @@ public abstract class Correlate extends BiRel {
   protected Correlate(
       RelOptCluster cluster,
       RelTraitSet traitSet,
+      List<RelHint> hints,
       RelNode left,
       RelNode right,
       CorrelationId correlationId,
@@ -101,9 +105,23 @@ protected Correlate(
     this.joinType = requireNonNull(joinType, "joinType");
     this.correlationId = requireNonNull(correlationId, "correlationId");
     this.requiredColumns = requireNonNull(requiredColumns, "requiredColumns");
+    this.hints = ImmutableList.copyOf(hints);
     assert isValid(Litmus.THROW, null);
   }
 
+  @Deprecated // to be removed before 2.0
+  protected Correlate(
+      RelOptCluster cluster,
+      RelTraitSet traitSet,
+      RelNode left,
+      RelNode right,
+      CorrelationId correlationId,
+      ImmutableBitSet requiredColumns,
+      JoinRelType joinType) {
+    this(cluster, traitSet, ImmutableList.of(), left, right,
+        correlationId, requiredColumns, joinType);
+  }
+
   /**
    * Creates a Correlate by parsing serialized output.
    *
@@ -235,4 +253,8 @@ public ImmutableBitSet getRequiredColumns() {
         rowCount /* generate results */ + leftRowCount /* scan left results */,
         0, 0).plus(rescanCost);
   }
+
+  @Override public ImmutableList<RelHint> getHints() {
+    return hints;
+  }
 }

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

@@ -392,16 +392,18 @@ private static class JoinFactoryImpl implements JoinFactory {
    * <p>The result is typically a {@link Correlate}.
    */
   public interface CorrelateFactory {
+
     /**
      * Creates a correlate.
      *
      * @param left             Left input
      * @param right            Right input
+     * @param hints            Hints
      * @param correlationId    Variable name for the row of left input
      * @param requiredColumns  Required columns
      * @param joinType         Join type
      */
-    RelNode createCorrelate(RelNode left, RelNode right,
+    RelNode createCorrelate(RelNode left, RelNode right, List<RelHint> hints,
         CorrelationId correlationId, ImmutableBitSet requiredColumns,
         JoinRelType joinType);
   }
@@ -411,10 +413,10 @@ RelNode createCorrelate(RelNode left, RelNode right,
    * {@link org.apache.calcite.rel.logical.LogicalCorrelate}.
    */
   private static class CorrelateFactoryImpl implements CorrelateFactory {
-    @Override public RelNode createCorrelate(RelNode left, RelNode right,
-        CorrelationId correlationId, ImmutableBitSet requiredColumns,
-        JoinRelType joinType) {
-      return LogicalCorrelate.create(left, right, correlationId,
+
+    @Override public RelNode createCorrelate(RelNode left, RelNode right, List<RelHint> hints,
+        CorrelationId correlationId, ImmutableBitSet requiredColumns, JoinRelType joinType) {
+      return LogicalCorrelate.create(left, right, hints, correlationId,
           requiredColumns, joinType);
     }
   }

core/src/main/java/org/apache/calcite/rel/hint/HintPredicates.java

@@ -50,6 +50,11 @@ public abstract class HintPredicates {
   public static final HintPredicate CALC =
       new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.CALC);
 
+  /** A hint predicate that indicates a hint can only be used to
+   * {@link org.apache.calcite.rel.core.Correlate} nodes. */
+  public static final HintPredicate CORRELATE =
+      new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.CORRELATE);
+
   /**
    * Returns a composed hint predicate that represents a short-circuiting logical
    * AND of an array of hint predicates {@code hintPredicates}.  When evaluating the composed

core/src/main/java/org/apache/calcite/rel/hint/NodeTypeHintPredicate.java

@@ -19,6 +19,7 @@
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Aggregate;
 import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Correlate;
 import org.apache.calcite.rel.core.Join;
 import org.apache.calcite.rel.core.Project;
 import org.apache.calcite.rel.core.TableScan;
@@ -63,7 +64,12 @@ enum NodeType {
     /**
      * The hint would be propagated to the Calc nodes.
      */
-    CALC(Calc.class);
+    CALC(Calc.class),
+
+    /**
+     * The hint would be propagated to the Correlate nodes.
+     */
+    CORRELATE(Correlate.class);
 
     /** Relational expression clazz that the hint can apply to. */
     @SuppressWarnings("ImmutableEnumChecker")

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

@@ -25,8 +25,13 @@
 import org.apache.calcite.rel.core.Correlate;
 import org.apache.calcite.rel.core.CorrelationId;
 import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.hint.RelHint;
 import org.apache.calcite.util.ImmutableBitSet;
 
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -58,6 +63,7 @@ public final class LogicalCorrelate extends Correlate {
   public LogicalCorrelate(
       RelOptCluster cluster,
       RelTraitSet traitSet,
+      List<RelHint> hints,
       RelNode left,
       RelNode right,
       CorrelationId correlationId,
@@ -66,18 +72,33 @@ public LogicalCorrelate(
     super(
         cluster,
         traitSet,
+        hints,
         left,
         right,
         correlationId,
         requiredColumns,
         joinType);
   }
 
+  @Deprecated // to be removed before 2.0
+  public LogicalCorrelate(
+      RelOptCluster cluster,
+      RelTraitSet traitSet,
+      RelNode left,
+      RelNode right,
+      CorrelationId correlationId,
+      ImmutableBitSet requiredColumns,
+      JoinRelType joinType) {
+    this(cluster, traitSet, ImmutableList.of(), left, right,
+        correlationId, requiredColumns, joinType);
+  }
+
   /**
    * Creates a LogicalCorrelate by parsing serialized output.
    */
   public LogicalCorrelate(RelInput input) {
-    this(input.getCluster(), input.getTraitSet(), input.getInputs().get(0),
+    this(input.getCluster(), input.getTraitSet(), ImmutableList.of(),
+        input.getInputs().get(0),
         input.getInputs().get(1),
         new CorrelationId(
             (Integer) requireNonNull(input.get("correlation"), "correlation")),
@@ -86,26 +107,38 @@ public LogicalCorrelate(RelInput input) {
   }
 
   /** Creates a LogicalCorrelate. */
-  public static LogicalCorrelate create(RelNode left, RelNode right,
+  public static LogicalCorrelate create(RelNode left, RelNode right, List<RelHint> hints,
       CorrelationId correlationId, ImmutableBitSet requiredColumns,
       JoinRelType joinType) {
     final RelOptCluster cluster = left.getCluster();
     final RelTraitSet traitSet = cluster.traitSetOf(Convention.NONE);
-    return new LogicalCorrelate(cluster, traitSet, left, right, correlationId,
+    return new LogicalCorrelate(cluster, traitSet, hints, left, right, correlationId,
         requiredColumns, joinType);
   }
 
+  @Deprecated // to be removed before 2.0
+  public static LogicalCorrelate create(RelNode left, RelNode right,
+      CorrelationId correlationId, ImmutableBitSet requiredColumns,
+      JoinRelType joinType) {
+    return create(left, right, ImmutableList.of(), correlationId, requiredColumns, joinType);
+  }
+
   //~ Methods ----------------------------------------------------------------
 
   @Override public LogicalCorrelate copy(RelTraitSet traitSet,
       RelNode left, RelNode right, CorrelationId correlationId,
       ImmutableBitSet requiredColumns, JoinRelType joinType) {
     assert traitSet.containsIfApplicable(Convention.NONE);
-    return new LogicalCorrelate(getCluster(), traitSet, left, right,
+    return new LogicalCorrelate(getCluster(), traitSet, hints, left, right,
         correlationId, requiredColumns, joinType);
   }
 
   @Override public RelNode accept(RelShuttle shuttle) {
     return shuttle.visit(this);
   }
+
+  @Override public RelNode withHints(List<RelHint> hintList) {
+    return new LogicalCorrelate(getCluster(), traitSet, hintList, left, right,
+        correlationId, requiredColumns, joinType);
+  }
 }

core/src/main/java/org/apache/calcite/rel/mutable/MutableRels.java

@@ -60,6 +60,8 @@
 import org.apache.calcite.util.mapping.MappingType;
 import org.apache.calcite.util.mapping.Mappings;
 
+import com.google.common.collect.ImmutableList;
+
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.util.AbstractList;
@@ -297,7 +299,8 @@ public static RelNode fromMutable(MutableRel node, RelBuilder relBuilder) {
     case CORRELATE:
       final MutableCorrelate correlate = (MutableCorrelate) node;
       return LogicalCorrelate.create(fromMutable(correlate.getLeft(), relBuilder),
-          fromMutable(correlate.getRight(), relBuilder), correlate.correlationId,
+          fromMutable(correlate.getRight(), relBuilder),
+          ImmutableList.of(), correlate.correlationId,
           correlate.requiredColumns, correlate.joinType);
     case UNION:
       final MutableUnion union = (MutableUnion) node;

core/src/main/java/org/apache/calcite/rel/rules/JoinToCorrelateRule.java

@@ -124,6 +124,7 @@ protected JoinToCorrelateRule(RelFactories.FilterFactory filterFactory) {
     RelNode newRel =
         LogicalCorrelate.create(left,
             relBuilder.build(),
+            join.getHints(),
             correlationId,
             requiredColumns.build(),
             join.getJoinType());

core/src/main/java/org/apache/calcite/sql2rel/RelStructuredTypeFlattener.java

@@ -482,6 +482,7 @@ public void rewriteRel(LogicalCorrelate rel) {
     LogicalCorrelate newRel =
         LogicalCorrelate.create(getNewForOldRel(rel.getLeft()),
             getNewForOldRel(rel.getRight()),
+            rel.getHints(),
             rel.getCorrelationId(),
             newPos.build(),
             rel.getJoinType());

core/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java

@@ -2756,7 +2756,7 @@ protected RelNode createJoin(
             .union(p.requiredColumns);
       }
 
-      return LogicalCorrelate.create(leftRel, innerRel,
+      return LogicalCorrelate.create(leftRel, innerRel, ImmutableList.of(),
           p.id, requiredCols, joinType);
     }
 

core/src/main/java/org/apache/calcite/tools/RelBuilder.java

@@ -2791,7 +2791,7 @@ public RelBuilder join(JoinRelType joinType, RexNode condition,
       }
       final ImmutableBitSet requiredColumns = RelOptUtil.correlationColumns(id, right.rel);
       join =
-          struct.correlateFactory.createCorrelate(left.rel, right.rel, id,
+          struct.correlateFactory.createCorrelate(left.rel, right.rel, ImmutableList.of(), id,
               requiredColumns, joinType);
     } else {
       RelNode join0 =
@@ -2836,7 +2836,7 @@ public RelBuilder correlate(JoinRelType joinType,
     Frame left = stack.pop();
 
     final RelNode correlate =
-        struct.correlateFactory.createCorrelate(left.rel, right.rel,
+        struct.correlateFactory.createCorrelate(left.rel, right.rel, ImmutableList.of(),
             correlationId, ImmutableBitSet.of(requiredOrdinals), joinType);
 
     final ImmutableList.Builder<Field> fields = ImmutableList.builder();

core/src/test/java/org/apache/calcite/test/RelBuilderTest.java

@@ -4403,6 +4403,10 @@ private static RelNode buildCorrelateWithJoin(JoinRelType type) {
     final RelHint noHashJoinHint = RelHint.builder("NO_HASH_JOIN")
         .inheritPath(0)
         .build();
+    final RelHint hashJoinHint = RelHint.builder("USE_HASH_JOIN")
+        .hintOption("orders")
+        .hintOption("products_temporal")
+        .build();
     final RelBuilder builder = RelBuilder.create(config().build());
     // Equivalent SQL:
     //   SELECT *
@@ -4440,6 +4444,32 @@ private static RelNode buildCorrelateWithJoin(JoinRelType type) {
         .hints(noHashJoinHint)
         .build();
     assertThat(root2, hasHints("[[NO_HASH_JOIN inheritPath:[0]]]"));
+
+    // Equivalent SQL:
+    //   SELECT *
+    //   FROM orders
+    //   JOIN products_temporal FOR SYSTEM_TIME AS OF orders.rowtime
+    //   ON orders.product = products_temporal.id
+    RelNode left = builder.scan("orders").build();
+    RelNode right = builder.scan("products_temporal").build();
+    RexNode period = builder.getRexBuilder().makeFieldAccess(
+        builder.getRexBuilder().makeCorrel(left.getRowType(), new CorrelationId(0)),
+        0);
+    RelNode root3 =
+        builder
+            .push(left)
+            .push(right)
+            .snapshot(period)
+            .correlate(
+                JoinRelType.INNER,
+                new CorrelationId(0),
+                builder.field(2, 0, "ROWTIME"),
+                builder.field(2, 0, "ID"),
+                builder.field(2, 0, "PRODUCT"))
+            .hints(hashJoinHint)
+            .build();
+    assertThat(root3,
+        hasHints("[[USE_HASH_JOIN inheritPath:[] options:[orders, products_temporal]]]"));
   }
 
   @Test void testHintsOnEmptyStack() {

core/src/test/java/org/apache/calcite/test/RelOptRulesTest.java

@@ -1743,7 +1743,7 @@ private void checkProjectCorrelateTransposeRuleSemiOrAntiCorrelate(JoinRelType t
           .project(b.field(0),
               b.getRexBuilder().makeFieldAccess(rexCorrel, 0)).build();
       LogicalCorrelate correlate = new LogicalCorrelate(left.getCluster(),
-          left.getTraitSet(), left, right, correlationId,
+          left.getTraitSet(), ImmutableList.of(), left, right, correlationId,
           ImmutableBitSet.of(0), type);
 
       b.push(correlate);
@@ -4061,6 +4061,7 @@ private HepProgram getTransitiveProgram() {
     CustomCorrelate customCorrelate = new CustomCorrelate(
         logicalCorrelate.getCluster(),
         logicalCorrelate.getTraitSet(),
+        logicalCorrelate.getHints(),
         logicalCorrelate.getLeft(),
         logicalCorrelate.getRight(),
         logicalCorrelate.getCorrelationId(),