Revert "[SPARK-35636][SQL] Lambda keys should not be referenced outsi…

…de of the lambda function"

This reverts commit f10fbb2.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/higherOrderFunctions.scala

@@ -84,9 +84,6 @@ case class NamedLambdaVariable(
     AttributeReference(name, dataType, nullable, Metadata.empty)(exprId, Seq.empty)
   }
 
-  // Check if this lambda variable is referenced outside the lambda function it is bound to
-  override def references: AttributeSet = AttributeSet(toAttribute)
-
   override def eval(input: InternalRow): Any = value.get
 
   override def toString: String = s"lambda $name#${exprId.id}$typeSuffix"
@@ -111,13 +108,6 @@ case class LambdaFunction(
   override def dataType: DataType = function.dataType
   override def nullable: Boolean = function.nullable
 
-  // Check if lambda variables bound to this lambda function are referenced in the wrong scope
-  override def references: AttributeSet = if (resolved) {
-    function.references -- AttributeSet(arguments.flatMap(_.references))
-  } else {
-    super.references
-  }
-
   lazy val bound: Boolean = arguments.forall(_.resolved)
 
   override def eval(input: InternalRow): Any = function.eval(input)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/NestedColumnAliasingSuite.scala

@@ -23,10 +23,9 @@ import org.apache.spark.sql.catalyst.SchemaPruningTest
 import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.plans.Cross
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.RuleExecutor
-import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{StringType, StructField, StructType}
 
 class NestedColumnAliasingSuite extends SchemaPruningTest {
 
@@ -685,35 +684,6 @@ class NestedColumnAliasingSuite extends SchemaPruningTest {
     ).analyze
     comparePlans(optimized2, expected2)
   }
-
-  test("SPARK-35636: do not push lambda key out of lambda function") {
-    val rel = LocalRelation(
-      'kvs.map(StringType, new StructType().add("v1", IntegerType)), 'keys.array(StringType))
-    val key = UnresolvedNamedLambdaVariable("key" :: Nil)
-    val lambda = LambdaFunction('kvs.getItem(key).getField("v1"), key :: Nil)
-    val query = rel
-      .limit(5)
-      .select('keys, 'kvs)
-      .limit(5)
-      .select(ArrayTransform('keys, lambda).as("a"))
-      .analyze
-    val optimized = Optimize.execute(query)
-    comparePlans(optimized, query)
-  }
-
-  test("SPARK-35636: do not push down extract value in higher order " +
-    "function that references both sides of a join") {
-    val left = LocalRelation('kvs.map(StringType, new StructType().add("v1", IntegerType)))
-    val right = LocalRelation('keys.array(StringType))
-    val key = UnresolvedNamedLambdaVariable("key" :: Nil)
-    val lambda = LambdaFunction('kvs.getItem(key).getField("v1"), key :: Nil)
-    val query = left
-      .join(right, Cross, None)
-      .select(ArrayTransform('keys, lambda).as("a"))
-      .analyze
-    val optimized = Optimize.execute(query)
-    comparePlans(optimized, query)
-  }
 }
 
 object NestedColumnAliasingSuite {