[CALCITE-6242] Enhance lambda closure

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

@@ -1802,9 +1802,14 @@ private Result toInnerStorageType(Result result, Type storageType) {
         new ParameterExpression[rexLambdaRefs.size()];
     for (int i = 0; i < rexLambdaRefs.size(); i++) {
       final RexLambdaRef rexLambdaRef = rexLambdaRefs.get(i);
+      // Declare lambda parameters as 'final' so that Janino can capture them
+      // from nested anonymous classes (e.g., nested lambdas like x -> y -> x + y).
+      // Janino requires captured local variables to be explicitly final,
+      // unlike javac which supports effectively-final variables.
       parameterExpressions[i] =
           Expressions.parameter(
-              typeFactory.getJavaClass(rexLambdaRef.getType()), rexLambdaRef.getName());
+              Modifier.FINAL, typeFactory.getJavaClass(rexLambdaRef.getType()),
+              rexLambdaRef.getName());
     }
 
     // Generate code for lambda expression body

core/src/main/java/org/apache/calcite/plan/RelOptUtil.java

@@ -70,6 +70,7 @@
 import org.apache.calcite.rex.RexExecutorImpl;
 import org.apache.calcite.rex.RexFieldAccess;
 import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLambda;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexLocalRef;
 import org.apache.calcite.rex.RexNode;
@@ -3363,6 +3364,12 @@ private static RexShuttle pushShuttle(final Project project) {
       @Override public RexNode visitInputRef(RexInputRef ref) {
         return project.getProjects().get(ref.getIndex());
       }
+
+      @Override public RexNode visitLambda(RexLambda lambda) {
+        // Lambda body references are at a different scope level.
+        // Do not remap indices inside lambda body against this project.
+        return lambda;
+      }
     };
   }
 
@@ -3386,6 +3393,12 @@ private static RexShuttle pushShuttle(final Calc calc) {
       @Override public RexNode visitInputRef(RexInputRef ref) {
         return projects.get(ref.getIndex());
       }
+
+      @Override public RexNode visitLambda(RexLambda lambda) {
+        // Lambda body references are at a different scope level.
+        // Do not remap indices inside lambda body against this calc.
+        return lambda;
+      }
     };
   }
 

core/src/main/java/org/apache/calcite/rex/RexBiVisitorImpl.java

@@ -119,8 +119,18 @@ protected RexBiVisitorImpl(boolean deep) {
     return null;
   }
 
+  /**
+   * Visits a lambda expression. When {@code deep} is true, recurses into
+   * the lambda body so that analysis visitors (e.g. InputFinder) can discover
+   * field references inside the lambda. When {@code deep} is false, returns
+   * null without recursing — this is the shallow traversal mode used by
+   * visitors that only need top-level information.
+   */
   @Override public R visitLambda(RexLambda lambda, P arg) {
-    return null;
+    if (!deep) {
+      return null;
+    }
+    return lambda.getExpression().accept(this, arg);
   }
 
   @Override public R visitNodeAndFieldIndex(RexNodeAndFieldIndex nodeAndFieldIndex, P arg) {

core/src/main/java/org/apache/calcite/rex/RexProgramBuilder.java

@@ -910,8 +910,9 @@ private abstract class RegisterShuttle extends RexShuttle {
     }
 
     @Override public RexNode visitLambda(RexLambda lambda) {
-      super.visitLambda(lambda);
-      return registerInternal(lambda);
+      // Lambda body references are at a different scope level.
+      // Do not validate or register lambda body indices against this program's input.
+      return lambda;
     }
   }
 

core/src/main/java/org/apache/calcite/rex/RexVisitorImpl.java

@@ -118,8 +118,17 @@ protected RexVisitorImpl(boolean deep) {
     return null;
   }
 
+  /**
+   * Visits a lambda expression. When {@code deep} is true, recurses into
+   * the lambda body to analyze its sub-expressions (critical for InputFinder
+   * to detect field references inside lambda bodies during pushDownJoinConditions).
+   * When {@code deep} is false, returns null without recursing.
+   */
   @Override public R visitLambda(RexLambda lambda) {
-    return null;
+    if (!deep) {
+      return null;
+    }
+    return lambda.getExpression().accept(this);
   }
 
   @Override public R visitLambdaRef(RexLambdaRef lambdaRef) {

core/src/main/java/org/apache/calcite/runtime/CalciteResource.java

@@ -244,9 +244,15 @@ ExInst<SqlValidatorException> columnNotFoundInTableDidYouMean(String a0,
   ExInst<SqlValidatorException> paramNotFoundInFunctionDidYouMean(String a0,
       String a1, String a2);
 
+  @BaseMessage("Lambda closure is not allowed in this conformance: reference to ''{0}'' from enclosing scope")
+  ExInst<SqlValidatorException> lambdaClosureNotAllowed(String identifier);
+
   @BaseMessage("Param ''{0}'' not found in lambda expression ''{1}''")
   ExInst<SqlValidatorException> paramNotFoundInLambdaExpression(String a0, String a1);
 
+  @BaseMessage("Duplicate lambda parameter ''{0}''")
+  ExInst<SqlValidatorException> duplicateLambdaParameter(String paramName);
+
   @BaseMessage("Operand {0} must be a query")
   ExInst<SqlValidatorException> needQueryOp(String a0);
 

core/src/main/java/org/apache/calcite/sql/validate/SqlAbstractConformance.java

@@ -149,6 +149,10 @@ public abstract class SqlAbstractConformance implements SqlConformance {
     return SqlConformanceEnum.DEFAULT.allowQualifyingCommonColumn();
   }
 
+  @Override public boolean allowLambdaClosure() {
+    return SqlConformanceEnum.DEFAULT.allowLambdaClosure();
+  }
+
   @Override public boolean allowAliasUnnestItems() {
     return SqlConformanceEnum.DEFAULT.allowAliasUnnestItems();
   }

core/src/main/java/org/apache/calcite/sql/validate/SqlConformance.java

@@ -610,6 +610,28 @@ default boolean isColonFieldAccessAllowed() {
    */
   boolean allowQualifyingCommonColumn();
 
+  /**
+   * Whether to allow lambda expressions to access variables from enclosing
+   * scopes (closure semantics).
+   *
+   * <p>For example, in a higher-order function context like:
+   *
+   * <blockquote><pre>
+   * SELECT *
+   * FROM t1
+   * JOIN t2 ON EXISTS(t1.arr, x -&gt; x = t2.v)</pre></blockquote>
+   *
+   * <p>The {@code t2.v} from the enclosing scope would be accessible inside
+   * the lambda body if closures are allowed.
+   *
+   * <p>Among the built-in conformance levels, false in
+   * {@link SqlConformanceEnum#STRICT_92},
+   * {@link SqlConformanceEnum#STRICT_99},
+   * {@link SqlConformanceEnum#STRICT_2003};
+   * true otherwise.
+   */
+  boolean allowLambdaClosure();
+
   /**
    * Whether {@code VALUE} is allowed as an alternative to {@code VALUES} in
    * the parser.

core/src/main/java/org/apache/calcite/sql/validate/SqlConformanceEnum.java

@@ -433,6 +433,17 @@ public enum SqlConformanceEnum implements SqlConformance {
     }
   }
 
+  @Override public boolean allowLambdaClosure() {
+    switch (this) {
+    case STRICT_92:
+    case STRICT_99:
+    case STRICT_2003:
+      return false;
+    default:
+      return true;
+    }
+  }
+
   @Override public boolean allowAliasUnnestItems() {
     switch (this) {
     case PRESTO:

core/src/main/java/org/apache/calcite/sql/validate/SqlDelegatingConformance.java

@@ -158,6 +158,10 @@ protected SqlDelegatingConformance(SqlConformance delegate) {
     return delegate.allowQualifyingCommonColumn();
   }
 
+  @Override public boolean allowLambdaClosure() {
+    return delegate.allowLambdaClosure();
+  }
+
   @Override public boolean isValueAllowed() {
     return delegate.isValueAllowed();
   }

core/src/main/java/org/apache/calcite/sql/validate/SqlLambdaScope.java

@@ -21,15 +21,12 @@
 import org.apache.calcite.sql.SqlLambda;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.type.SqlTypeName;
-import org.apache.calcite.util.Litmus;
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.util.HashMap;
 import java.util.Map;
 
-import static com.google.common.base.Preconditions.checkArgument;
-
 import static org.apache.calcite.util.Static.RESOURCE;
 
 /**
@@ -54,29 +51,46 @@ public SqlLambdaScope(
 
   /** True if the identifier matches one of the parameter names. */
   public boolean isParameter(SqlIdentifier id) {
-    return this.parameterTypes.containsKey(id.toString());
+    final SqlNameMatcher nameMatcher = validator.catalogReader.nameMatcher();
+    final String name = id.getSimple();
+    return parameterTypes.keySet().stream()
+        .anyMatch(paramName -> nameMatcher.matches(paramName, name));
   }
 
   @Override public SqlNode getNode() {
     return lambdaExpr;
   }
 
   @Override public SqlQualified fullyQualify(SqlIdentifier identifier) {
-    boolean found = lambdaExpr.getParameters()
-        .stream()
-        .anyMatch(param -> param.equalsDeep(identifier, Litmus.IGNORE));
-    if (found) {
-      return SqlQualified.create(this, 1, null, identifier);
-    } else {
+    if (identifier.isSimple()) {
+      final SqlNameMatcher nameMatcher = validator.catalogReader.nameMatcher();
+      final String name = identifier.getSimple();
+      boolean found = lambdaExpr.getParameters()
+          .stream()
+          .anyMatch(param ->
+              nameMatcher.matches(((SqlIdentifier) param).getSimple(), name));
+      if (found) {
+        return SqlQualified.create(this, 1, null, identifier);
+      }
+    }
+    if (!validator.config().conformance().allowLambdaClosure()) {
       throw validator.newValidationError(identifier,
-          RESOURCE.paramNotFoundInLambdaExpression(identifier.toString(), lambdaExpr.toString()));
+          RESOURCE.lambdaClosureNotAllowed(identifier.toString()));
     }
+    return parent.fullyQualify(identifier);
   }
 
   @Override public @Nullable RelDataType resolveColumn(String columnName, SqlNode ctx) {
-    checkArgument(parameterTypes.containsKey(columnName),
-        "column %s not found", columnName);
-    return parameterTypes.get(columnName);
+    final SqlNameMatcher nameMatcher = validator.catalogReader.nameMatcher();
+    for (Map.Entry<String, RelDataType> entry : parameterTypes.entrySet()) {
+      if (nameMatcher.matches(entry.getKey(), columnName)) {
+        return entry.getValue();
+      }
+    }
+    // Delegate to parent scope for nested lambda closure resolution.
+    // In a nested lambda like x -> EXISTS(arr, y -> x + y), the inner lambda
+    // scope does not contain 'x', but the outer lambda scope does.
+    return parent.resolveColumn(columnName, ctx);
   }
 
   public Map<String, RelDataType> getParameterTypes() {

core/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java

@@ -3300,10 +3300,9 @@ private void registerQuery(
           alias,
           lambdaNamespace,
           forceNullable);
-      operands = call.getOperandList();
-      for (int i = 0; i < operands.size(); i++) {
-        registerOperandSubQueries(parentScope, call, i);
-      }
+      // Register sub-queries inside the body under lambdaScope, so that
+      // nested lambdas can resolve outer lambda parameters.
+      registerOperandSubQueries(lambdaScope, call, 1);
       break;
 
     case WITH:
@@ -6426,6 +6425,30 @@ public void setOriginal(SqlNode expr, SqlNode original) {
     final LambdaNamespace ns =
         getNamespaceOrThrow(lambdaExpr).unwrap(LambdaNamespace.class);
 
+    // Check for duplicate lambda parameter names
+    final SqlNameMatcher nameMatcher = catalogReader.nameMatcher();
+    final Set<String> seen = nameMatcher.createSet();
+    for (SqlNode param : lambdaExpr.getParameters()) {
+      final String name = ((SqlIdentifier) param).getSimple();
+      if (!seen.add(name)) {
+        throw newValidationError(param,
+            RESOURCE.duplicateLambdaParameter(name));
+      }
+      // Check against enclosing lambda scopes: x -> ... x -> ...
+      SqlValidatorScope parentScope = scope.getParent();
+      while (parentScope instanceof DelegatingScope) {
+        if (parentScope instanceof SqlLambdaScope) {
+          final SqlLambdaScope parentLambda = (SqlLambdaScope) parentScope;
+          if (parentLambda.getParameterTypes().keySet().stream()
+              .anyMatch(p -> nameMatcher.matches(p, name))) {
+            throw newValidationError(param,
+                RESOURCE.duplicateLambdaParameter(name));
+          }
+        }
+        parentScope = ((DelegatingScope) parentScope).getParent();
+      }
+    }
+
     deriveType(scope, lambdaExpr.getExpression());
     RelDataType type = deriveTypeImpl(scope, lambdaExpr);
     setValidatedNodeType(lambdaExpr, type);

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

@@ -2362,6 +2362,13 @@ private RexNode convertLambda(Blackboard bb, SqlNode node) {
     final SqlLambdaScope scope = (SqlLambdaScope) validator().getLambdaScope(call);
 
     final Map<String, RexNode> nameToNodeMap = new HashMap<>();
+    // For nested lambdas, inherit the parent blackboard's nameToNodeMap so that
+    // the inner lambda can resolve references to outer lambda parameters.
+    // e.g., in x -> EXISTS(arr, y -> x + y = 4), the inner lambda's blackboard
+    // needs access to "X" from the outer lambda's nameToNodeMap.
+    if (bb.nameToNodeMap != null) {
+      nameToNodeMap.putAll(bb.nameToNodeMap);
+    }
     final List<RexLambdaRef> parameters = new ArrayList<>(scope.getParameterTypes().size());
     final Map<String, RelDataType> parameterTypes = scope.getParameterTypes();
 
@@ -5563,11 +5570,16 @@ void setRoot(List<RelNode> inputs) {
         SqlQualified qualified) {
       if (nameToNodeMap != null && qualified.prefixLength == 1) {
         RexNode node = nameToNodeMap.get(qualified.identifier.names.get(0));
-        if (node == null) {
+        if (node != null) {
+          return Pair.of(node, null);
+        }
+        // If the identifier is not found in nameToNodeMap and the current scope
+        // is a lambda scope, fall through to standard scope resolution to allow
+        // external references (e.g., t2.v in a JOIN ON lambda expression).
+        if (!(scope instanceof SqlLambdaScope)) {
           throw new AssertionError("Unknown identifier '" + qualified.identifier
               + "' encountered while expanding expression");
         }
-        return Pair.of(node, null);
       }
       final SqlNameMatcher nameMatcher =
           scope.getValidator().getCatalogReader().nameMatcher();
@@ -5586,6 +5598,13 @@ void setRoot(List<RelNode> inputs) {
       // preserved.
       final SqlValidatorScope ancestorScope = resolve.scope;
       boolean isParent = ancestorScope != scope;
+      // When in a lambda scope, external references to tables that are part
+      // of the current blackboard's inputs should be resolved locally, not
+      // as correlation variables. The lambda blackboard inherits inputs from
+      // its parent blackboard.
+      if (isParent && scope instanceof SqlLambdaScope && inputs != null) {
+        isParent = false;
+      }
       if ((inputs != null) && !isParent) {
         final LookupContext rels =
             new LookupContext(this, inputs, systemFieldList.size());

core/src/main/resources/org/apache/calcite/runtime/CalciteResource.properties

@@ -390,4 +390,6 @@ CannotInferReturnType=Cannot infer return type for {0}; operand types: {1}
 SelectByCannotWithGroupBy=SELECT BY cannot be used with GROUP BY
 SelectByCannotWithOrderBy=SELECT BY cannot be used with ORDER BY
 DescriptorMustBeIdentifier=The argument of DESCRIPTOR must be an identifier
+LambdaClosureNotAllowed=Lambda closure is not allowed in this conformance: reference to ''{0}'' from enclosing scope
+DuplicateLambdaParameter=Duplicate lambda parameter ''{0}''
 # End CalciteResource.properties

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

@@ -180,6 +180,19 @@ public static void checkActualAndReferenceFiles() {
         .ok();
   }
 
+  /** Test case for nested lambda: inner lambda references outer lambda
+   * parameter. Verifies that 'x' in the inner lambda is resolved from the
+   * outer lambda scope, not treated as a table column name. */
+  @Test void testNestedLambdaExpression() {
+    final String sql =
+        "select \"EXISTS\"(array(1,2,3), x -> \"EXISTS\"(array(1,2,3), y -> x + y = 4))";
+    fixture()
+        .withFactory(c ->
+            c.withOperatorTable(t -> SqlValidatorTest.operatorTableFor(SqlLibrary.SPARK)))
+        .withSql(sql)
+        .ok();
+  }
+
   @Test void testDotLiteralAfterRow() {
     final String sql = "select row(1,2).\"EXPR$1\" from emp";
     sql(sql).ok();