[SPARK-55983][SQL] New single-pass analyzer functionality and bugfixes

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

  This PR implements core single-pass resolver infrastructure and bugfixes. The single-pass resolver is an alternative to the traditional fixed-point (iterative) analyzer that resolves SQL plans in a single bottom-up traversal.

  Key changes:

  **New infrastructure:**
  - `OperatorResolutionContext` and `OperatorResolutionContextStack` for tracking operator-level state
  during resolution (HAVING tracking, subquery aggregate push-down, grouping analytics)
  - `NameResolutionParameters` for bundling name resolution flags (LCA access, hidden output, view
  resolution, extract value keys, etc.)
  - `ResolverGuardResult` structured result type replacing boolean return from `ResolverGuard`
  - `NonDeterministicExpressionCheck` as a single-pass-only resolution check
  - `RetainsOriginalJoinOutput` trait for preserving join output when metadata columns change child
  projections
  - `AliasKind` enum for distinguishing alias types during resolution
  - `TryExtractOrdinal` utility for ordinal extraction from expressions

  **Core improvements:**
  - Extended `ExpressionResolutionContext` with window expression tracking (nestedness level, window
  function/spec flags, parent context)
  - Enriched `NameScope` with variable resolution, extract value extraction keys, aggregate expression
  alias lookup, and hidden output improvements
  - Improved `HavingResolver` to handle aggregate expressions extracted from subqueries and window +
  HAVING interaction patterns
  - Extended `ExpressionIdAssigner` with subquery expression ID remapping and outer reference mapping
  - Expanded `ResolverGuard` to support lambda functions, star-with-target, regex columns, COALESCE with
   star, and multi-part TVF names
  - Moved plan rewrite rules (`CleanupAliases`, `PullOutNondeterministic`, `PruneMetadataColumns`) from
  `ResolverRunner` to `Resolver.lookupMetadataAndResolve` for correct per-view config handling
  - Improved `HybridAnalyzer` tentative mode with comprehensive fallback handling
  - Various improvements to aggregate resolution, sort resolution, join resolution, set operation
  resolution, and subquery expression resolution

  ### Why are the changes needed?

  To bring the Apache Spark single-pass analyzer closer to feature parity with the fixed-point analyzer implementation, enabling:
  1. Correct resolution of window expressions, HAVING clauses with windows, and complex aggregate
  patterns
  2. Proper operator-level context tracking during resolution
  3. Better name resolution with variable support, extract value keys, and hidden output handling
  4. Structured guard results for cleaner tentative mode fallback logic
  5. Foundation for future features like nested correlated subquery support

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

  No. The single-pass analyzer is behind feature flags (`spark.sql.analyzer.singlePassResolver.enabled`
  and `spark.sql.analyzer.singlePassResolver.enabled.tentatively`) and is not the default code path.

  ### How was this patch tested?

  - Extended `HybridAnalyzerSuite` with `testDualRun` framework covering dual-run, tentative, and
  single-pass modes
  - New `DataFrameAnalyzerTestGapsSuite` for alias resolution edge cases (implicit aliases, nested alias
   collapsing, autogenerated alias names)
  - New `SinglePassAnalyzerTestUtils` test utility trait
  - Updated `ResolverGuardSuite` with structured result assertions and new feature tests
  (star-with-target, regex columns, COALESCE with star, multi-part TVF names)
  - Updated `ExplicitlyUnsupportedResolverFeatureSuite` with generator edge case tests
  - Updated `ExpressionIdAssignerSuite` for new attribute reference mapping semantics and subquery
  remapping
  - Updated `NameScopeSuite` with variable resolution and extract value key tests
  - Updated `AggregateResolverSuite`, `AliasResolverSuite`, `MetadataResolverSuite`, `ViewResolverSuite`

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

  No.

Closes #54729 from mihailotim-db/resolver-pr1-core.

Authored-by: Mihailo Timotic <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>

Author: mihailotim-db

common/utils/src/main/resources/error/error-conditions.json

@@ -2476,6 +2476,13 @@
           "Single-pass analyzer output schema:",
           "<singlePassOutputSchema>"
         ]
+      },
+      "SINGLE_PASS_FAILED_FIXED_POINT_SUCCEEDED" : {
+        "message" : [
+          "Single-pass resolution failed, but fixed-point resolution succeeded.",
+          "Fixed-point analyzer output:",
+          "<fixedPointOutput>"
+        ]
       }
     },
     "sqlState" : "XX000"

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AggregateExpressionResolver.scala

@@ -18,8 +18,17 @@
 package org.apache.spark.sql.catalyst.analysis.resolver
 
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.expressions.{Expression, OuterReference, SubExprUtils}
-import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ListAgg}
+import org.apache.spark.sql.catalyst.analysis.TypeCoercionValidation
+import org.apache.spark.sql.catalyst.expressions.{
+  Expression,
+  OuterReference,
+  SubExprUtils
+}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{
+  AggregateExpression,
+  AggregateFunction,
+  ListAgg
+}
 import org.apache.spark.sql.catalyst.util.toPrettySQL
 import org.apache.spark.sql.errors.QueryCompilationErrors
 
@@ -39,16 +48,18 @@ class AggregateExpressionResolver(
   private val expressionResolutionContextStack =
     expressionResolver.getExpressionResolutionContextStack
   private val subqueryRegistry = operatorResolver.getSubqueryRegistry
-  private val autoGeneratedAliasProvider = new AutoGeneratedAliasProvider(
-    expressionResolver.getExpressionIdAssigner
-  )
+  private val autoGeneratedAliasProvider = expressionResolver.getAutoGeneratedAliasProvider
 
   /**
    * Resolves the given [[AggregateExpression]] originating from [[ExpressionResolver]] by
    * resolving its children recursively and validating the resolved expression.
    */
   override def resolve(aggregateExpression: AggregateExpression): Expression = {
-    expressionResolutionContextStack.peek().resolvingTreeUnderAggregateExpression = true
+    val expressionResolutionContext = expressionResolutionContextStack.peek()
+    expressionResolutionContext.resolvingTreeUnderAggregateExpression = true
+    if (expressionResolutionContext.resolvingWindowFunction) {
+      expressionResolutionContext.windowFunctionNestednessLevel += 1
+    }
     val aggregateExpressionWithChildrenResolved =
       withResolvedChildren(aggregateExpression, expressionResolver.resolve _)
         .asInstanceOf[AggregateExpression]
@@ -76,25 +87,84 @@ class AggregateExpressionResolver(
    *  - Resolution:
    *    1. Update the [[ExpressionResolver.expressionResolutionContextStack]];
    *    2. Handle [[OuterReference]] in [[AggregateExpression]], if there are any (see
-   *    `handleOuterAggregateExpression`);
+   *       `handleOuterAggregateExpression`);
+   *    3. Confirm that we are resolving an actual aggregate expression, not a window aggregate
+   *       function by assessing the `windowFunctionNestednessLevel`:
+   *
+   *          - In this query, `windowFunctionNestednessLevel` is `1` when resolving `SUM()`,
+   *            meaning we are resolving a window aggregate function:
+   *
+   *            {{{ SELECT SUM() OVER() FROM VALUES 1, 2; }}}
+   *
+   *          - In this query, `windowFunctionNestednessLevel` is `2` when resolving `SUM(col1)`,
+   *            meaning we are resolving an aggregate expression:
+   *
+   *            {{{
+   *            SELECT
+   *              SUM(SUM(col1)) OVER (PARTITION BY SUM(col1))
+   *            FROM
+   *              VALUES (1, 2)
+   *            GROUP BY col2;
+   *            }}}
+   *
+   *          - In this query, `windowFunctionNestednessLevel` is `0` when resolving `SUM(col1)`,
+   *            meaning we are resolving an aggregate expression:
+   *
+   *            {{{ SELECT SUM(col1) FROM VALUES (1, 2); }}}
+   *
+   *       If the expression is not a window aggregate function, set
+   *       `hasAggregateExpressionsOutsideWindow` to `true` and collect it as window source
+   *       expression.
+   *
+   *       In this query:
+   *
+   *       {{{ SELECT SUM(SUM(col1)) OVER () FROM VALUES (1, 2) GROUP BY col2; }}}
+   *
+   *       `SUM(col1)` is collected, whereas `SUM(SUM(col1))` is not because the latter is a
+   *       window aggregate function.  As a result, `SUM(col1)` is handled by [[Aggregate]],
+   *       while `SUM(SUM(col1))` is handled  by [[Window]]:
+   *
+   *       Window [sum(_w0#2) windowspecdefinition...#3]
+   *       +- Aggregate [col2#1], [sum(col1#0) AS _w0#2]
    *  - Validation:
-   *   1. [[ListAgg]] is not allowed in DISTINCT aggregates if the order value is ambiguous
+   *   1. Validate that the type coercion is doen properly for the underlying aggregate function.
+   *   2. [[ListAgg]] is not allowed in DISTINCT aggregates if the order value is ambiguous
    *      after deduplication. When [[SQLConf.LISTAGG_ALLOW_DISTINCT_CAST_WITH_ORDER]] is
    *      enabled, a mismatch is tolerated if the child value uniquely determines the order
    *      value (see [[ListAgg.hasDistinctOrderAmbiguity]]);
-   *   2. Nested aggregate functions are not allowed;
-   *   3. Nondeterministic expressions in the subtree of a related aggregate function are not
+   *   3. [[PythonUDAF]] is not allowed in [[Pivot.aggregates]];
+   *   4. Nested aggregate functions are not allowed, unless the outer function represents
+   *      [[WindowExpression.windowFunction]]. For example the following query is considered
+   *      valid, because `SUM(SUM(col1))` is followed by an `OVER` clause, making it a nesting
+   *      of aggregate expression in a window expression:
+   *
+   *      {{{
+   *      SELECT
+   *      SUM(SUM(col1)) OVER (PARTITION BY col2)
+   *      FROM (VALUES (1, 'a'), (2, 'b'))
+   *      GROUP BY col2
+   *      }}}
+   *
+   *      On the other hand, this query should be invalid, because `SUM(SUM(col1))` isn't followed
+   *      by an `OVER` clause, therefore it should be treated like aggregate expression nesting:
+   *
+   *      {{{
+   *      SELECT
+   *      SUM(SUM(SUM(col1)) OVER (PARTITION BY col2)
+   *      FROM (VALUES (1, 'a'), (2, 'b'))
+   *      GROUP BY col2
+   *      }}}
+   *
+   *   5. Nondeterministic expressions in the subtree of a related aggregate function are not
    *      allowed;
-   *   4. The mix of outer and local references is not allowed;
+   *   6. The mix of outer and local references is not allowed;
    */
   private def handleAggregateExpressionWithChildrenResolved(
       aggregateExpressionWithChildrenResolved: AggregateExpression): Expression = {
     val expressionResolutionContext = expressionResolutionContextStack.peek()
 
     validateResolvedAggregateExpression(aggregateExpressionWithChildrenResolved)
 
-    expressionResolutionContext.hasAggregateExpressions = true
-
     // There are two different cases that we handle regarding the value of the flag:
     //
     //   - We have an attribute under an `AggregateExpression`:
@@ -108,20 +178,31 @@ class AggregateExpressionResolver(
     //     It would be `true` as described above.
     expressionResolutionContext.hasAttributeOutsideOfAggregateExpressions = false
 
-    if (expressionResolutionContext.hasOuterReferences) {
+    val resolvedAggregateExpression = if (expressionResolutionContext.hasOuterReferences) {
       handleOuterAggregateExpression(aggregateExpressionWithChildrenResolved)
     } else {
       aggregateExpressionWithChildrenResolved
     }
+
+    if (expressionResolutionContext.windowFunctionNestednessLevel != 1) {
+      expressionResolutionContext.hasAggregateExpressionsOutsideWindow = true
+    }
+
+    resolvedAggregateExpression
   }
 
   private def validateResolvedAggregateExpression(aggregateExpression: AggregateExpression): Unit =
     aggregateExpression match {
+      case _ @AggregateExpression(aggregateFunction: AggregateFunction, _, _, _, _)
+          if aggregateFunction.checkInputDataTypes().isFailure =>
+        TypeCoercionValidation.failOnTypeCheckResult(aggregateFunction)
       case agg @ AggregateExpression(listAgg: ListAgg, _, _, _, _)
           if agg.isDistinct && listAgg.hasDistinctOrderAmbiguity =>
         listAgg.throwDistinctOrderError()
       case _ =>
-        if (expressionResolutionContextStack.peek().hasAggregateExpressions) {
+        val expressionResolutionContext = expressionResolutionContextStack.peek()
+        if (expressionResolutionContext.windowFunctionNestednessLevel != 1 &&
+          expressionResolutionContext.hasAggregateExpressionsOutsideWindow) {
           throwNestedAggregateFunction(aggregateExpression)
         }
 
@@ -145,8 +226,9 @@ class AggregateExpressionResolver(
    *  - If outer aggregates are allowed, replace the [[AggregateExpression]] with an
    *    [[OuterReference]] to the auto-generated [[Alias]] that we created in case the subtree
    *    without [[OuterReference]]s can't be found in the outer
-   *    [[Aggregate.aggregateExpressions]] list. Otherwise, use the [[Alias]] from the outer
-   *    [[Aggregate]]. This alias will later be injected into the outer [[Aggregate]];
+   *    [[Aggregate.aggregateExpressions]] list or a semantically equal expression wasn't already
+   *    added. Otherwise, use the [[Alias]] from the outer [[Aggregate]]. This alias will later be
+   *    injected into the outer [[Aggregate]];
    *  - Store the name that needs to be used for the [[OuterReference]] in
    *    [[OuterReference.SINGLE_PASS_SQL_STRING_OVERRIDE]] and
    *    [[OuterReference.SINGLE_PASS_OUTER_AGGREGATE_ALIAS_NAME_OVERRIDE]], both computed from the
@@ -202,20 +284,27 @@ class AggregateExpressionResolver(
       child = aggregateExpressionWithStrippedOuterReferences
     )
 
-    val (_, referencedAggregateExpressionAlias) =
-      aggregateExpressionsExtractor.collectFirstAggregateExpression(
-        aggregateExpressionWithStrippedOuterReferences
-      )
-
-    referencedAggregateExpressionAlias match {
-      case Some(alias) =>
-        subqueryRegistry.currentScope.addAliasForOuterAggregateExpression(alias)
-        OuterReference(alias.toAttribute)
+    subqueryRegistry.currentScope.getExtractedAggregateExpression(
+      aggregateExpressionWithStrippedOuterReferences
+    ) match {
+      case Some(alreadyExtractedAggregateExpression) =>
+        OuterReference(alreadyExtractedAggregateExpression.toAttribute)
       case None =>
-        subqueryRegistry.currentScope.addAliasForOuterAggregateExpression(
-          outerAggregateExpressionAlias
-        )
-        OuterReference(outerAggregateExpressionAlias.toAttribute)
+        val (_, referencedAggregateExpressionAlias) =
+          aggregateExpressionsExtractor.collectFirstAggregateExpression(
+            aggregateExpressionWithStrippedOuterReferences
+          )
+
+        referencedAggregateExpressionAlias match {
+          case Some(alias) =>
+            subqueryRegistry.currentScope.addExtractedAggregateExpression(alias)
+            OuterReference(alias.toAttribute)
+          case None =>
+            subqueryRegistry.currentScope.addExtractedAggregateExpression(
+              outerAggregateExpressionAlias
+            )
+            OuterReference(outerAggregateExpressionAlias.toAttribute)
+        }
     }
   }
 
@@ -236,4 +325,5 @@ class AggregateExpressionResolver(
       origin = nonDeterministicChild.origin
     )
   }
+
 }