[CALCITE-7440] Preserve rel2sql correlation fallback without JDBC regressions

Author: bvolpato

core/src/main/java/org/apache/calcite/rel/rel2sql/RelToSqlConverter.java

@@ -772,6 +772,7 @@ private Builder visitAggregate(Aggregate e, List<Integer> groupKeyList,
     // "select a, b, sum(x) from ( ... ) group by a, b"
     final boolean ignoreClauses = e.getInput() instanceof Project;
     final Result x = visitInput(e, 0, isAnon(), ignoreClauses, clauseSet);
+    parseCorrelTable(e, x);
     final Builder builder = x.builder(e);
     final List<SqlNode> selectList = new ArrayList<>();
     final List<SqlNode> groupByList =

core/src/main/java/org/apache/calcite/rel/rel2sql/SqlImplementor.java

@@ -1587,9 +1587,42 @@ public static SqlNode toSql(RexLiteral literal) {
   }
 
   protected Context getAliasContext(RexCorrelVariable variable) {
-    return requireNonNull(
-        correlTableMap.get(variable.id),
-        () -> "variable " + variable.id + " is not found");
+    Context context = correlTableMap.get(variable.id);
+    if (context == null) {
+      // Some rewrites can introduce a new correlation id that still points to
+      // the same outer scope. Reuse a compatible known context before failing.
+      context = findCompatibleCorrelContext(variable.id, variable.getType().getFieldCount());
+      if (context == null && correlTableMap.isEmpty()) {
+        context =
+            aliasContext(ImmutableMap.of(variable.id.getName(), variable.getType()), true);
+      }
+      if (context != null) {
+        correlTableMap.put(variable.id, context);
+      }
+    }
+    return requireNonNull(context, () -> "variable " + variable.id + " is not found");
+  }
+
+  private @Nullable Context findCompatibleCorrelContext(CorrelationId missingId,
+      int expectedFieldCount) {
+    Context fallback = null;
+    String fallbackName = null;
+    for (Map.Entry<CorrelationId, Context> entry : correlTableMap.entrySet()) {
+      final CorrelationId id = entry.getKey();
+      final Context context = entry.getValue();
+      if (id.equals(missingId)) {
+        continue;
+      }
+      if (context.fieldCount == expectedFieldCount) {
+        return context;
+      }
+      final String name = id.getName();
+      if (fallbackName == null || name.compareTo(fallbackName) < 0) {
+        fallbackName = name;
+        fallback = context;
+      }
+    }
+    return fallback;
   }
 
   /** Simple implementation of {@link Context} that cannot handle sub-queries

core/src/test/java/org/apache/calcite/rel/rel2sql/RelToSqlConverterTest.java

@@ -11887,6 +11887,49 @@ public Sql schema(CalciteAssert.SchemaSpec schemaSpec) {
     sql(sql).schema(CalciteAssert.SchemaSpec.JDBC_SCOTT).ok(expected);
   }
 
+  /** Test case for
+   * <a href="https://issues.apache.org/jira/browse/CALCITE-7440">[CALCITE-7440]
+   * RelToSqlConverter throws NPE when correlation scope is missing after
+   * semi-join rewrites.</a>. */
+  @Test void testPostgresqlRoundTripCorrelatedProjectWithSemiJoinRules() {
+    final String query = "WITH product_keys AS (\n"
+        + "  SELECT p.\"product_id\",\n"
+        + "         (SELECT MAX(p3.\"product_id\")\n"
+        + "          FROM \"foodmart\".\"product\" p3\n"
+        + "          WHERE p3.\"product_id\" = p.\"product_id\") AS \"mx\"\n"
+        + "  FROM \"foodmart\".\"product\" p\n"
+        + ")\n"
+        + "SELECT DISTINCT pk.\"product_id\"\n"
+        + "FROM product_keys pk\n"
+        + "LEFT JOIN \"foodmart\".\"product\" p2 USING (\"product_id\")\n"
+        + "WHERE pk.\"product_id\" IN (\n"
+        + "  SELECT p4.\"product_id\"\n"
+        + "  FROM \"foodmart\".\"product\" p4\n"
+        + ")";
+
+    final RuleSet rules =
+        RuleSets.ofList(CoreRules.FILTER_SUB_QUERY_TO_CORRELATE,
+        CoreRules.PROJECT_SUB_QUERY_TO_CORRELATE,
+        CoreRules.JOIN_SUB_QUERY_TO_CORRELATE,
+        CoreRules.FILTER_SUB_QUERY_TO_MARK_CORRELATE,
+        CoreRules.PROJECT_SUB_QUERY_TO_MARK_CORRELATE,
+        CoreRules.MARK_TO_SEMI_OR_ANTI_JOIN_RULE,
+        CoreRules.PROJECT_TO_SEMI_JOIN,
+        CoreRules.JOIN_TO_SEMI_JOIN,
+        CoreRules.SEMI_JOIN_FILTER_TRANSPOSE,
+        CoreRules.SEMI_JOIN_JOIN_TRANSPOSE);
+
+    final String generated = sql(query).withPostgresql().optimize(rules, null).exec();
+    try {
+      sql(generated).withPostgresql().exec();
+    } catch (Exception e) {
+      throw new AssertionError(
+          "Generated SQL failed PostgreSQL round-trip validation:\n"
+              + generated,
+          e);
+    }
+  }
+
   @Test void testNotBetween() {
     Sql f = fixture().withConvertletTable(new SqlRexConvertletTable() {
       @Override public @Nullable SqlRexConvertlet get(SqlCall call) {