[CALCITE-7753] CorrelateProjectExtractor corrupts plans with nested Correlates that reuse the same correlation id

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

@@ -18,6 +18,7 @@
 
 import org.apache.calcite.rel.RelHomogeneousShuttle;
 import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Correlate;
 import org.apache.calcite.rel.core.CorrelationId;
 import org.apache.calcite.rel.core.Filter;
 import org.apache.calcite.rel.core.Project;
@@ -115,8 +116,11 @@ private static boolean isDirectFieldAccess(RexNode node, CorrelationId id) {
     boolean isTrivialCorrelation =
         callsWithCorrelationInRight.stream()
             .allMatch(exp -> isDirectFieldAccess(exp, correlate.getCorrelationId()));
+    // A nested correlate on the right re-binding the same id owns the refs below it.
+    boolean rightRebindsCorrelationId =
+        rebindsCorrelationId(right, correlate.getCorrelationId());
     // Early exit condition
-    if (isTrivialCorrelation) {
+    if (isTrivialCorrelation || rightRebindsCorrelationId) {
       if (correlate.getLeft().equals(left) && correlate.getRight().equals(right)) {
         return correlate;
       } else {
@@ -201,6 +205,21 @@ private static boolean isDirectFieldAccess(RexNode node, CorrelationId id) {
     return builder.build();
   }
 
+  /** Returns whether {@code plan} contains a {@link Correlate} that re-binds {@code corrId}. */
+  private static boolean rebindsCorrelationId(RelNode plan, CorrelationId corrId) {
+    final boolean[] found = {false};
+    plan.accept(new RelHomogeneousShuttle() {
+      @Override public RelNode visit(RelNode other) {
+        if (other instanceof Correlate
+            && ((Correlate) other).getCorrelationId().equals(corrId)) {
+          found[0] = true;
+        }
+        return super.visit(other);
+      }
+    });
+    return found[0];
+  }
+
   /**
    * Traverses a plan and finds all simply correlated row expressions with the specified id.
    */

core/src/test/java/org/apache/calcite/sql2rel/RelDecorrelatorTest.java

@@ -2548,4 +2548,43 @@ private RelNode decorrelateSql(String sql) {
         + "            LogicalTableScan(table=[[bookstore, authors]])\n";
     assertThat(after, hasTree(planAfter));
   }
+
+  /** Test case for
+   * <a href="https://issues.apache.org/jira/browse/CALCITE-7753">[CALCITE-7753]
+   * CorrelateProjectExtractor corrupts plans with nested Correlates that reuse
+   * the same correlation id</a>. */
+  @Test void testNestedCorrelatesSharingCorrelationId() {
+    final RelBuilder builder = RelBuilder.create(config().build())
+        .transform(c -> c.withSimplify(false));
+    final Holder<@Nullable RexCorrelVariable> v = Holder.empty();
+    builder.scan("EMP").variable(v::set);
+    final RelNode emp = builder.build();
+
+    final RelNode inner = builder
+        .scan("EMP")
+        .scan("EMP")
+        .filter(builder.equals(builder.field(v.get(), "DEPTNO"), builder.field("DEPTNO")))
+        .correlate(JoinRelType.INNER, v.get().id, builder.field(2, 0, "DEPTNO"))
+        .filter(builder.isNull(builder.field(v.get(), "COMM")))
+        .build();
+
+    final RelNode before = builder
+        .push(emp)
+        .push(inner)
+        .correlate(JoinRelType.LEFT, v.get().id, builder.field(2, 0, "COMM"))
+        .build();
+    final String planBefore = ""
+        + "LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{6}])\n"
+        + "  LogicalTableScan(table=[[scott, EMP]])\n"
+        + "  LogicalFilter(condition=[IS NULL($cor0.COMM)])\n"
+        + "    LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{7}])\n"
+        + "      LogicalTableScan(table=[[scott, EMP]])\n"
+        + "      LogicalFilter(condition=[=($cor0.DEPTNO, $7)])\n"
+        + "        LogicalTableScan(table=[[scott, EMP]])\n";
+    assertThat(before, hasTree(planBefore));
+
+    RelDecorrelator.decorrelateQuery(before, builder,
+        RuleSets.ofList(Collections.emptyList()),
+        RuleSets.ofList(Collections.emptyList()));
+  }
 }