fix(isthmus): avoid NPE from dangling correlations left by partial decorrelation

Author: mbwhite

isthmus/src/main/java/io/substrait/isthmus/OuterReferenceResolver.java

@@ -66,6 +66,26 @@ public RelNode visit(Filter filter) throws RuntimeException {
     for (CorrelationId id : filter.getVariablesSet()) {
       nestedDepth.putIfAbsent(id, 0);
     }
+    // Also register any CorrelationIds referenced directly in the condition expression.
+    // This covers partially-decorrelated plans where a Calcite optimizer (e.g. HepPlanner)
+    // rewrites correlated IN-subqueries into joins but leaves scalar-aggregate subquery
+    // correlations as RexFieldAccess($corN.col) inside a Filter condition without a
+    // surrounding Correlate node. In that case getVariablesSet() is empty but the condition
+    // still contains RexCorrelVariable references whose IDs must be in nestedDepth before
+    // rexVisitor.visitFieldAccess() is called.
+    filter
+        .getCondition()
+        .accept(
+            new RexShuttle() {
+              @Override
+              public RexNode visitFieldAccess(RexFieldAccess fieldAccess) {
+                if (fieldAccess.getReferenceExpr() instanceof RexCorrelVariable) {
+                  CorrelationId id = ((RexCorrelVariable) fieldAccess.getReferenceExpr()).id;
+                  nestedDepth.putIfAbsent(id, 0);
+                }
+                return fieldAccess;
+              }
+            });
     filter.getCondition().accept(rexVisitor);
     return super.visit(filter);
   }

isthmus/src/main/java/io/substrait/isthmus/expression/RexExpressionConverter.java

@@ -224,7 +224,23 @@ public Expression visitFieldAccess(RexFieldAccess fieldAccess) {
     switch (kind) {
       case CORREL_VARIABLE:
         {
-          int stepsOut = relVisitor.getFieldAccessDepth(fieldAccess);
+          Integer stepsOut = relVisitor.getFieldAccessDepth(fieldAccess);
+          // Substrait requires steps_out >= 1 for an outer reference
+          // (proto/substrait/algebra.proto, OuterReference.steps_out: "Must be >= 1.").
+          // A null or 0 depth means the correlation does not step out of any subquery,
+          // typically a plan that was only partially decorrelated (the owning Correlate
+          // was rewritten into a join but the $cor reference remains in this Filter).
+          // We can't represent that with steps_out, so fail clearly rather than emit
+          // an invalid steps_out=0 reference. Id-based outer-reference resolution
+          // would handle this and is the proper long-term fix.
+          if (stepsOut == null || stepsOut == 0) {
+            throw new UnsupportedOperationException(
+                String.format(
+                    "Cannot convert outer reference %s: it does not step out of any "
+                        + "subquery (steps_out=%s). This usually indicates a plan that was "
+                        + "only partially decorrelated; Substrait requires steps_out >= 1.",
+                    fieldAccess, stepsOut));
+          }
 
           return FieldReference.newRootStructOuterReference(
               fieldAccess.getField().getIndex(),

isthmus/src/test/java/io/substrait/isthmus/OuterReferenceResolverTest.java

@@ -7,6 +7,7 @@
 import org.apache.calcite.rel.core.JoinRelType;
 import org.apache.calcite.rex.RexCorrelVariable;
 import org.apache.calcite.rex.RexFieldAccess;
+import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.tools.RelBuilder;
 import org.apache.calcite.util.Holder;
@@ -142,4 +143,69 @@ void nestedApplyJoinQuery() throws SqlParseException {
     validateOuterRef(fieldAccessDepthMap, "$cor0", "SS_ITEM_SK", 2);
     validateOuterRef(fieldAccessDepthMap, "$cor1", "I_ITEM_SK", 1);
   }
+
+  /**
+   * Regression test for the partially-decorrelated Filter case.
+   *
+   * <p>When a Calcite optimizer (e.g. HepPlanner) rewrites a correlated IN-subquery into a join, it
+   * can produce a {@link org.apache.calcite.rel.core.Filter} whose condition contains a {@link
+   * RexFieldAccess} referencing a {@link RexCorrelVariable} ({@code $cor0.field}), but whose {@link
+   * RelNode#getVariablesSet()} is empty — because the enclosing {@link
+   * org.apache.calcite.rel.core.Correlate} node has already been replaced by a regular join.
+   *
+   * <p>Before the fix, {@code OuterReferenceResolver.visit(Filter)} only iterated {@code
+   * filter.getVariablesSet()}, so the {@link org.apache.calcite.rel.core.CorrelationId} was never
+   * registered in {@code nestedDepth}. The subsequent {@code rexVisitor.visitFieldAccess()} call
+   * found {@code !nestedDepth.containsKey(id)} and silently skipped the access, producing an empty
+   * {@code fieldAccessDepthMap} instead of the expected entry.
+   *
+   * <p>The fix pre-scans the Filter condition for {@link RexCorrelVariable} references and
+   * registers their IDs before delegating to the rex visitor.
+   */
+  @Test
+  void filterWithCorrelVariableButEmptyVariablesSet() throws SqlParseException {
+    // Build the partially-decorrelated pattern:
+    //   JOIN (inner side has a Filter whose condition references $cor0 but variablesSet is empty)
+    //
+    // SQL equivalent of what a post-optimisation plan looks like:
+    //   SELECT ss_sold_date_sk FROM store_sales
+    //   JOIN item ON item.i_item_sk = store_sales.ss_item_sk   ← decorrelated join
+    //   WHERE item.i_item_sk = $cor0.SS_ITEM_SK               ← Filter still has the correl ref
+    //
+    // We deliberately call .filter(condition) without passing the CorrelationId so that
+    // getVariablesSet() returns an empty set, reproducing the post-decorrelation shape.
+
+    final Holder<RexCorrelVariable> cor0 = Holder.empty();
+
+    // Push STORE_SALES and capture the correlation variable for it.
+    tpcDsRelBuilder.scan("tpcds", "STORE_SALES").variable(cor0::set);
+
+    // Push ITEM on top, then build the filter condition while ITEM is the top-of-stack.
+    // The condition equates item.I_ITEM_SK to the correlated $cor0.SS_ITEM_SK.
+    tpcDsRelBuilder.scan("tpcds", "ITEM");
+    RexNode correlCondition =
+        tpcDsRelBuilder.equals(
+            tpcDsRelBuilder.field("I_ITEM_SK"), tpcDsRelBuilder.field(cor0.get(), "SS_ITEM_SK"));
+
+    final RelNode calciteRel =
+        tpcDsRelBuilder
+            // Intentionally NOT passing cor0.get().id to filter() → getVariablesSet() is empty.
+            .filter(correlCondition)
+            .join(JoinRelType.INNER, tpcDsRelBuilder.literal(true))
+            .project(tpcDsRelBuilder.field("SS_SOLD_DATE_SK"))
+            .build();
+
+    // Resolver registers the dangling correlation (no longer silently dropped → no NPE).
+    final Map<RexFieldAccess, Integer> fieldAccessDepthMap = buildOuterFieldRefMap(calciteRel);
+    Assertions.assertEquals(1, fieldAccessDepthMap.size());
+    validateOuterRef(fieldAccessDepthMap, "$cor0", "SS_ITEM_SK", 0);
+
+    // steps_out=0 is not a valid Substrait outer reference (proto requires >= 1),
+    // so conversion must fail clearly rather than emit an invalid plan.
+    UnsupportedOperationException ex =
+        Assertions.assertThrows(
+            UnsupportedOperationException.class,
+            () -> SubstraitRelVisitor.convert(calciteRel, converterProvider));
+    Assertions.assertTrue(ex.getMessage().contains("steps_out"));
+  }
 }