[SPARK-51884][SQL] Part 1.a Add outer scope attributes for SubqueryExpression

### What changes were proposed in this pull request?

- Add OuterScopeReference as a new expression type
- Use OuterScopeReference to differentiate between outerAttrs and outerScopeAttrs
- Add outerScopeAttrs related methods to SubqueryExpression

### Why are the changes needed?

Spark only supports one layer of correlation now and does not support nested correlation.
For example,
```sql
SELECT col1 FROM VALUES (1, 2) t1 (col1, col2) WHERE EXISTS (
 SELECT col1 FROM VALUES (1, 2) t2 (col1, col2) WHERE t2.col2 == MAX(t1.col2)
)GROUP BY col1;
```
is supported and
```sql
SELECT col1 FROM VALUES (1, 2) t1 (col1, col2) WHERE EXISTS (
 SELECT col1 FROM VALUES (1, 2) t2 (col1, col2) WHERE t2.col2 == (
   SELECT MAX(t1.col2)
 )
)GROUP BY col1;
```
is not supported.

The reason spark does not support it is because the Analyzer and Optimizer resolves and plans Subquery in a recursive way.

The definition change for the SubqueryExpression adds the metadata OuterScopeAttrs which helps later rewrites for the Analyzer and Optimizer.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Current UT and Suite

### Was this patch authored or co-authored using generative AI tooling?

No

Closes apache#50838 from AveryQi115/part_1_a_add_a_new_expression_type.

Lead-authored-by: Avery <[email protected]>
Co-authored-by: Avery Qi <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: 2 people

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

@@ -452,6 +452,29 @@ case class OuterReference(e: NamedExpression)
   final override val nodePatterns: Seq[TreePattern] = Seq(OUTER_REFERENCE)
 }
 
+/**
+ * A place holder used to hold a reference that has been resolved to a field outside of the subquery
+ * plan. We use it only for queries containing nested correlation and only in the
+ * SubqueryExpression#outerAttrs to indicate that the correlated columns referenced by
+ * this subquery expression are from the outer scopes, not the subquery plan or
+ * the immediate outer plan of the subquery plan. For example,
+ * SubqueryExpression(outerAttrs=[OuterScopeReference(a)] means a cannot be resolved by the
+ * SubqueryExpression.plan or the plan holding this SubqueryExpression.
+ */
+case class OuterScopeReference(e: NamedExpression)
+  extends LeafExpression with NamedExpression with Unevaluable {
+  override def dataType: DataType = e.dataType
+  override def nullable: Boolean = e.nullable
+  override def prettyName: String = "outerScope"
+
+  override def sql: String = s"$prettyName(${e.sql})"
+  override def name: String = e.name
+  override def qualifier: Seq[String] = e.qualifier
+  override def exprId: ExprId = e.exprId
+  override def toAttribute: Attribute = e.toAttribute
+  override def newInstance(): NamedExpression = OuterScopeReference(e.newInstance())
+}
+
 /**
  * A placeholder used to hold a [[NamedExpression]] that has been temporarily resolved as the
  * reference to a lateral column alias. It will be restored back to [[UnresolvedAttribute]] if

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

@@ -79,14 +79,29 @@ abstract class SubqueryExpression(
     exprId: ExprId,
     joinCond: Seq[Expression],
     hint: Option[HintInfo]) extends PlanExpression[LogicalPlan] {
+
   override lazy val resolved: Boolean = childrenResolved && plan.resolved
+
+  lazy val (outerScopeAttrs, nonOuterScopeAttrs) =
+    outerAttrs.partition(_.exists(_.isInstanceOf[OuterScopeReference]))
+
   override lazy val references: AttributeSet =
-    AttributeSet.fromAttributeSets(outerAttrs.map(_.references))
+    AttributeSet.fromAttributeSets(nonOuterScopeAttrs.map(_.references))
+
   override def children: Seq[Expression] = outerAttrs ++ joinCond
+
   override def withNewPlan(plan: LogicalPlan): SubqueryExpression
+
   def withNewOuterAttrs(outerAttrs: Seq[Expression]): SubqueryExpression
+
+  def getOuterAttrs: Seq[Expression] = outerAttrs
+
+  def getOuterScopeAttrs: Seq[Expression] = outerScopeAttrs
+
   def isCorrelated: Boolean = outerAttrs.nonEmpty
+
   def hint: Option[HintInfo]
+
   def withNewHint(hint: Option[HintInfo]): SubqueryExpression
 }