feat(isthmus)!: resolve outer references natively by rel_anchor

Replace the steps_out bridge with native id-based correlation handling so
isthmus no longer works on offset-based outer references internally.

Producing: OuterReferenceResolver now maps each Calcite CorrelationId to a
plan-wide rel_anchor and its binding relation. SubstraitRelVisitor stamps
that anchor on the binding relation and RexExpressionConverter emits
rel_reference directly. The offset-based depth computation and the
OuterReferenceConverter.toIdBased post-pass are removed.

Consuming: SubstraitRelNodeConverter.Context resolves rel_reference by
anchor, dropping the subquery-depth arithmetic. Incoming offset-based
plans are normalized to the id-based form once at the boundary
(OuterReferenceConverter.toIdBased), so both encodings are still accepted
while the internal machinery is purely id-based.

Calcite's CorrelationId is itself an identity bound to a single relation,
so id-based resolution is a closer and simpler fit than the depth model it
replaces.

BREAKING CHANGE: OuterReferenceResolver no longer computes a
RexFieldAccess->depth map; its apply(RelNode) is replaced by
resolve(RelNode) plus anchorForCorrelationId/anchorForTarget.
SubstraitRelVisitor.popFieldAccessDepthMap/getFieldAccessDepth are
replaced by resolveOuterReferences/getOuterReferenceAnchor, and the
SubstraitRelNodeConverter.Context correlation API is now anchor-based.

Signed-off-by: Niels Pardon <par@zurich.ibm.com>

Author: nielspardon

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

@@ -8,199 +8,131 @@
 import org.apache.calcite.rel.core.CorrelationId;
 import org.apache.calcite.rel.core.Filter;
 import org.apache.calcite.rel.core.Project;
-import org.apache.calcite.rex.RexCorrelVariable;
-import org.apache.calcite.rex.RexFieldAccess;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexShuttle;
 import org.apache.calcite.rex.RexSubQuery;
 import org.apache.calcite.rex.RexUtil.SubQueryCollector;
 
 /**
- * Resolve correlated variables and compute a depth map for {@link RexFieldAccess}.
+ * Assigns id-based outer-reference anchors to a {@link RelNode} tree.
  *
- * <p>Traverses a {@link RelNode} tree and:
+ * <p>Calcite models correlation with {@link CorrelationId}s, each bound by a single relation (a
+ * {@link Correlate}'s left input, or the input of a {@link Filter}/{@link Project} that declares
+ * the id). This maps directly onto Substrait's id-based outer references: the bound relation
+ * carries a {@code rel_anchor} ({@link io.substrait.relation.Rel#getRelAnchor()}) and every
+ * correlated field reference carries the matching {@code rel_reference} ({@link
+ * io.substrait.expression.FieldReference#outerReferenceRelReference()}).
+ *
+ * <p>This resolver walks the tree (including subquery relations) and produces:
  *
  * <ul>
- *   <li>Tracks nesting depth of {@link CorrelationId}s across filters, projects, subqueries, and
- *       correlates
- *   <li>Computes "steps out" for each {@link RexFieldAccess} referencing a {@link
- *       RexCorrelVariable}
+ *   <li>{@link #anchorForCorrelationId(CorrelationId)} — the anchor a correlated field reference
+ *       should emit;
+ *   <li>{@link #anchorForTarget(RelNode)} — the anchor to stamp on a bound relation.
  * </ul>
  *
- * See OuterReferenceResolver.md for details on how the depth map is computed.
+ * Anchors are plan-wide unique and assigned per bound relation, so multiple correlation ids sharing
+ * a relation share its anchor.
  */
 public class OuterReferenceResolver extends RelNodeVisitor<RelNode, RuntimeException> {
 
-  private final Map<CorrelationId, Integer> nestedDepth;
-  private final Map<RexFieldAccess, Integer> fieldAccessDepthMap;
+  private final Map<CorrelationId, Integer> anchorByCorrelationId = new HashMap<>();
+  private final Map<RelNode, Integer> anchorByTarget = new IdentityHashMap<>();
 
   private final RexVisitor rexVisitor = new RexVisitor(this);
 
-  /** Creates a new resolver with empty depth tracking maps. */
-  public OuterReferenceResolver() {
-    nestedDepth = new HashMap<>();
-    fieldAccessDepthMap = new IdentityHashMap<>();
-  }
+  private int nextAnchor = 1;
+
+  /** Creates a new resolver with empty anchor maps. */
+  public OuterReferenceResolver() {}
 
   /**
-   * Applies the resolver to a {@link RelNode} tree, computing the depth map.
+   * Walks a {@link RelNode} tree, assigning outer-reference anchors.
    *
-   * @param r the root relational node
+   * @param root the root relational node
    * @return the same node after traversal
-   * @throws RuntimeException if the visitor encounters an unrecoverable condition
    */
-  public Map<RexFieldAccess, Integer> apply(RelNode r) {
-    reverseAccept(r);
-    return fieldAccessDepthMap;
+  public RelNode resolve(RelNode root) {
+    reverseAccept(root);
+    return root;
   }
 
   /**
-   * Visits a {@link Filter}, registering any correlation variables and visiting its condition.
+   * Returns the anchor a correlated field reference bound to the given correlation id must emit as
+   * its {@code rel_reference}.
    *
-   * @param filter the filter node
-   * @return the result of {@link RelNodeVisitor#visit(Filter)}
-   * @throws RuntimeException if traversal fails
+   * @param id the correlation id
+   * @return the anchor, or {@code null} if the id was not resolved
    */
-  @Override
-  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);
+  public Integer anchorForCorrelationId(CorrelationId id) {
+    return anchorByCorrelationId.get(id);
   }
 
   /**
-   * Visits a {@link Correlate}, handling correlation depth for both sides.
-   *
-   * <p>Special case: the right side is a correlated subquery in the rel tree (not a REX), so we
-   * manually adjust depth before/after visiting it.
+   * Returns the anchor to stamp on the given relation, if it is the binding point of any outer
+   * reference.
    *
-   * @param correlate the correlate (correlated join) node
-   * @return the correlate node
-   * @throws RuntimeException if traversal fails
+   * @param rel the relation
+   * @return the anchor, or {@code null} if the relation binds no outer reference
    */
-  @Override
-  public RelNode visit(Correlate correlate) throws RuntimeException {
-    for (CorrelationId id : correlate.getVariablesSet()) {
-      nestedDepth.putIfAbsent(id, 0);
-    }
-
-    apply(correlate.getLeft());
-
-    // Correlated join is a special case. The right-rel is a correlated sub-query but not a REX. So,
-    // the RexVisitor cannot be applied to it to correctly compute the depth map. Hence, we need to
-    // manually compute the depth map for the right-rel.
-    nestedDepth.replaceAll((k, v) -> v + 1);
-
-    apply(correlate.getRight()); // look inside sub-queries
-
-    nestedDepth.replaceAll((k, v) -> v - 1);
+  public Integer anchorForTarget(RelNode rel) {
+    return anchorByTarget.get(rel);
+  }
 
-    return correlate;
+  /** Binds {@code id} to {@code target}, allocating {@code target}'s anchor on first use. */
+  private void bind(CorrelationId id, RelNode target) {
+    int anchor = anchorByTarget.computeIfAbsent(target, t -> nextAnchor++);
+    anchorByCorrelationId.put(id, anchor);
   }
 
-  /**
-   * Visits a generic {@link RelNode}, applying the resolver to all the node inputs.
-   *
-   * @param other the node to visit
-   * @return the node
-   * @throws RuntimeException if traversal fails
-   */
   @Override
-  public RelNode visitOther(RelNode other) throws RuntimeException {
-    for (RelNode child : other.getInputs()) {
-      apply(child);
+  public RelNode visit(Filter filter) throws RuntimeException {
+    for (CorrelationId id : filter.getVariablesSet()) {
+      bind(id, filter.getInput());
     }
-    return other;
+    filter.getCondition().accept(rexVisitor);
+    return visitOther(filter);
   }
 
-  /**
-   * Visits a {@link Project}, registering correlation variables and visiting any subqueries within
-   * its expressions.
-   *
-   * @param project the project node
-   * @return the result of {@link RelNodeVisitor#visit(Project)}
-   * @throws RuntimeException if traversal fails
-   */
   @Override
   public RelNode visit(Project project) throws RuntimeException {
     for (CorrelationId id : project.getVariablesSet()) {
-      nestedDepth.putIfAbsent(id, 0);
+      bind(id, project.getInput());
     }
-
     for (RexSubQuery subQuery : SubQueryCollector.collect(project)) {
       subQuery.accept(rexVisitor);
     }
+    return visitOther(project);
+  }
 
-    return super.visit(project);
+  @Override
+  public RelNode visit(Correlate correlate) throws RuntimeException {
+    bind(correlate.getCorrelationId(), correlate.getLeft());
+    reverseAccept(correlate.getLeft());
+    reverseAccept(correlate.getRight());
+    return correlate;
   }
 
-  /** Rex visitor used to track correlation depth within expressions and subqueries. */
+  @Override
+  public RelNode visitOther(RelNode other) throws RuntimeException {
+    for (RelNode child : other.getInputs()) {
+      reverseAccept(child);
+    }
+    return other;
+  }
+
+  /** Rex visitor that descends into subquery relations to reach nested correlations. */
   private static class RexVisitor extends RexShuttle {
     final OuterReferenceResolver referenceResolver;
 
-    /**
-     * Creates a new Rex visitor bound to the given reference resolver.
-     *
-     * @param referenceResolver the parent resolver maintaining depth maps
-     */
     RexVisitor(OuterReferenceResolver referenceResolver) {
       this.referenceResolver = referenceResolver;
     }
 
-    /**
-     * Increments correlation depth when entering a subquery and decrements when exiting.
-     *
-     * @param subQuery the subquery expression
-     * @return the same subquery
-     */
     @Override
     public RexNode visitSubQuery(RexSubQuery subQuery) {
-      referenceResolver.nestedDepth.replaceAll((k, v) -> v + 1);
-
-      referenceResolver.apply(subQuery.rel); // look inside sub-queries
-
-      referenceResolver.nestedDepth.replaceAll((k, v) -> v - 1);
+      referenceResolver.reverseAccept(subQuery.rel);
       return subQuery;
     }
-
-    /**
-     * Records depth for {@link RexFieldAccess} referencing a {@link RexCorrelVariable}.
-     *
-     * @param fieldAccess the field access expression
-     * @return the same field access
-     */
-    @Override
-    public RexNode visitFieldAccess(RexFieldAccess fieldAccess) {
-      if (fieldAccess.getReferenceExpr() instanceof RexCorrelVariable) {
-        CorrelationId id = ((RexCorrelVariable) fieldAccess.getReferenceExpr()).id;
-        if (referenceResolver.nestedDepth.containsKey(id)) {
-          referenceResolver.fieldAccessDepthMap.put(
-              fieldAccess, referenceResolver.nestedDepth.get(id));
-        }
-      }
-      return fieldAccess;
-    }
   }
 }