This is an automated email from the ASF dual-hosted git repository.
dtenedor pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new ee9367f18364 [SPARK-56596][SQL] Enable dual runs for single-pass
analyzer
ee9367f18364 is described below
commit ee9367f18364c4cfe1926868ebd2814a9bd1ec0c
Author: Mihailo Timotic <[email protected]>
AuthorDate: Thu Jul 23 09:37:57 2026 -0700
[SPARK-56596][SQL] Enable dual runs for single-pass analyzer
### What changes were proposed in this pull request?
This PR turns on **dual-run** of the single-pass analyzer by default in
test environments, so the existing test suite continuously validates the
single-pass resolver against the fixed-point analyzer, and it includes the
resolver/test fixes needed to make the two agree across the suite.
**Config defaults (`SQLConf`)**
- `spark.sql.analyzer.singlePassResolver.dualRunWithLegacy`: `false` ->
`Utils.isTesting`. In test runs, every query that passes the single-pass
resolver guard is analyzed by **both** resolvers and the resulting
logical plans are compared (after `NormalizePlan`); a divergence fails with
`LOGICAL_PLAN_COMPARISON_MISMATCH`.
-
`spark.sql.analyzer.singlePassResolver.returnSinglePassResultInDualRun`:
`Utils.isTesting` -> `false`. The dual run now returns the **fixed-point**
result, so analyzer output is unchanged -- the single-pass plan is
validated, never used for output.
- `spark.sql.analyzer.singlePassResolver.dualRunSampleRate` (non-testing
default): `0.001` -> `0.1`.
- The "tentatively enabled" single-pass flag now defaults from the
`SPARK_SINGLE_PASS_ANALYZER_ENABLED_TENTATIVELY` environment variable instead
of a
hard `false`.
**Resolver-guard scoping** (`ResolverGuard`) -- route shapes single-pass
does not support yet to fixed-point only:
- Stop matching higher-order functions and their lambda machinery
(`LambdaFunction`, `UnresolvedNamedLambdaVariable`, `NamedLambdaVariable`), and
treat generator and higher-order function names as unsupported.
- Stop matching grouping-analytics (`BaseGroupingSets`).
- Reject session-qualified user-defined functions.
- Reject stars whose target is an unsupported (metadata) attribute name
(`checkStarTarget`).
- Reject the `prioritizeOrdinalResolutionInSort=false` and
`persistentCatalogFirst=true` configurations.
**Single-pass / fixed-point resolution alignment**
- `ExpressionResolver` + `ExtractValueResolver`: resolve
`UnresolvedExtractValue` through a dedicated resolver using
`ExtractValue.applyOrNull`; when
the base is `NullType` (e.g. missing columns from schema evolution),
alias the resulting null with the extraction's text (e.g. `col[0]`) so the
output name matches the fixed-point analyzer. The extraction key is
threaded through multipart-name resolution.
- `NameScope` / `NameScopeStack` + `SubqueryScope`: expand a targeted
star (e.g. `t1.*`) from the nearest outer scope inside `Project` / `Aggregate`
/ `CollectMetrics`, wrapping the results in `OuterReference`, matching
fixed-point correlated-star behavior.
- `Analyzer`: propagate the single-pass resolver bridge state into the
view-analysis `AnalysisContext`.
- `FunctionResolver` / `FunctionResolverUtils`: share star-in-argument
handling via the `FunctionResolverUtils` trait -- expand star arguments and
align `count(*)` -> `count(1)` / `count(table.*)` handling and
function-name matching with fixed-point.
- `QueryPlan`: wrap subquery outer-reference normalization in
`allowInvokingTransformsInAnalyzer` so the dual-run plan comparison does not
trip the
in-analyzer transform assertion.
- `AttributeSet`: make `intersect` preserve the ordering of `other`, so
normalized plans compare deterministically.
- `View` (`basicLogicalOperators`): remove the redundant
`require(!isTempViewStoringAnalyzedPlan || child.resolved)`. The same invariant
(SPARK-34546) is already enforced authoritatively by
`TemporaryViewRelation` where the analyzed plan is stored, and a `View`
carrying an analyzed
plan is only ever built from `TemporaryViewRelation.plan`
(`SessionCatalog.getTempViewPlan`). The constructor copy was redundant and
additionally
tripped during the dual-run comparison, which normalizes expression ids
and can transiently make `child.resolved` false (e.g. a self-join whose sides
lose their distinct ids).
- `ExplicitlyUnsupportedResolverFeature`: mark metric-view logical nodes
(`MetricViewPlaceholder`, `ResolvedMetricView`) explicitly unsupported.
**Tests**
- New golden tests for the surfaced behaviors:
`replacing-missing-expression-with-alias.sql`,
`cte-with-multi-children-operators.sql`,
`having-and-order-by-recursive-type-name-resolution.sql`,
`order-by-and-having-on-top-of-aggregate-with-join.sql`; regenerated
`order-by.sql.out` and
`udf/postgreSQL/udf-join.sql.out`.
- `CollationTypePrecedenceSuite`: split a collation test into fixed-point
and single-pass variants where the two legitimately produce different (but
equivalent) aliasing.
- `HiveSessionStateBuilder`: register `DetermineTableStats` among the
single-pass resolver extensions (Hive-only rule).
- Connect / declarative-pipelines tests (`EndToEndAPISuite`,
`SparkDeclarativePipelinesServerTest`, `PipelineTest`): disable dual run for the
pipelines path, which single-pass does not support yet.
- `V1WriteFallbackSuite`, `SQLQuerySuite`: minor adjustments.
### Why are the changes needed?
The single-pass resolver is being developed as an eventual replacement
for the fixed-point analyzer. Running it in dual-run mode across the entire
test suite -- comparing its output to the fixed-point analyzer on every query
while still returning the fixed-point result -- gives continuous, broad
validation of single-pass correctness with no risk to behavior, and surfaces
divergences early. This PR makes that the default in test environments and
fixes the divergences it surfaced.
### Does this PR introduce _any_ user-facing change?
No. In non-test (production) environments dual-run remains off by
default, and even when dual-run is active the analyzer returns the fixed-point
result (`returnSinglePassResultInDualRun=false`), so query results and behavior
are unchanged. The single-pass plan is only compared, never used for output.
### How was this patch tested?
This change is itself test infrastructure: enabling dual-run makes the
existing catalyst and SQL test suites validate the single-pass resolver against
the fixed-point analyzer on every query. It was validated by running those
suites under the new default and resolving every plan-comparison mismatch (the
resolver fixes and golden/test updates above). New golden-file tests cover the
specific behaviors that were surfaced.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #56468 from
mihailotim-db/mihailo-timotic_data/enable-single-pass-dual-run.
Authored-by: Mihailo Timotic <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>
(cherry picked from commit a8211c3268929689ce29bfe2ea19da3c8a63f7db)
Signed-off-by: Daniel Tenedorio <[email protected]>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 1 +
.../ExplicitlyUnsupportedResolverFeature.scala | 4 +-
.../analysis/resolver/ExpressionResolver.scala | 63 ++++++++++++----
.../analysis/resolver/ExtractValueResolver.scala | 25 ++++++-
.../analysis/resolver/FunctionResolver.scala | 12 +++-
.../analysis/resolver/FunctionResolverUtils.scala | 34 ++-------
.../sql/catalyst/analysis/resolver/NameScope.scala | 84 +++++++++++++++++++++-
.../catalyst/analysis/resolver/ResolverGuard.scala | 71 +++++++++---------
.../catalyst/analysis/resolver/SubqueryScope.scala | 5 ++
.../sql/catalyst/expressions/AttributeSet.scala | 2 +-
.../spark/sql/catalyst/plans/QueryPlan.scala | 8 ++-
.../plans/logical/basicLogicalOperators.scala | 2 -
.../org/apache/spark/sql/internal/SQLConf.scala | 9 +--
.../sql/connect/pipelines/EndToEndAPISuite.scala | 2 +
.../SparkDeclarativePipelinesServerTest.scala | 7 ++
.../sql-tests/analyzer-results/order-by.sql.out | 2 +-
...replacing-missing-expression-with-alias.sql.out | 60 ++++++++++++++++
.../udf/postgreSQL/udf-join.sql.out | 6 +-
.../inputs/cte-with-multi-children-operators.sql | 7 ++
...and-order-by-recursive-type-name-resolution.sql | 8 +++
...by-and-having-on-top-of-aggregate-with-join.sql | 3 +
.../replacing-missing-expression-with-alias.sql | 18 +++++
...replacing-missing-expression-with-alias.sql.out | 64 +++++++++++++++++
.../scala/org/apache/spark/sql/SQLQuerySuite.scala | 9 ++-
.../scala/org/apache/spark/sql/SetPathSuite.scala | 25 ++++---
.../sql/analysis/resolver/ResolverGuardSuite.scala | 61 ++++++++++++++++
.../collation/CollationTypePrecedenceSuite.scala | 20 +++++-
.../spark/sql/connector/V1WriteFallbackSuite.scala | 4 +-
.../spark/sql/hive/HiveSessionStateBuilder.scala | 1 +
.../spark/sql/pipelines/utils/PipelineTest.scala | 4 ++
30 files changed, 509 insertions(+), 112 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 215c859154a0..c77e5e16accb 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -253,6 +253,7 @@ object AnalysisContext {
referredTempFunctionNames =
mutable.Set(viewDesc.viewReferredTempFunctionNames: _*),
referredTempVariableNames = viewDesc.viewReferredTempVariableNames,
collation = viewDesc.collation)
+
context.setSinglePassResolverBridgeState(originContext.getSinglePassResolverBridgeState)
set(context)
try f finally { set(originContext) }
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExplicitlyUnsupportedResolverFeature.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExplicitlyUnsupportedResolverFeature.scala
index c71e62453004..fad95a9555fb 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExplicitlyUnsupportedResolverFeature.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExplicitlyUnsupportedResolverFeature.scala
@@ -51,6 +51,8 @@ object ExplicitlyUnsupportedResolverFeature {
val OPERATORS = Set(
"org.apache.spark.sql.catalyst.streaming.StreamingRelationV2",
"org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation",
- "org.apache.spark.sql.execution.streaming.StreamingRelation"
+ "org.apache.spark.sql.execution.streaming.StreamingRelation",
+ "org.apache.spark.sql.metricview.logical.MetricViewPlaceholder",
+ "org.apache.spark.sql.metricview.logical.ResolvedMetricView"
)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolver.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolver.scala
index 306c8513ee32..c763d8323438 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolver.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolver.scala
@@ -32,6 +32,7 @@ import org.apache.spark.sql.catalyst.analysis.{
TypeCoercionValidation,
UnresolvedAlias,
UnresolvedAttribute,
+ UnresolvedExtractValue,
UnresolvedFunction,
UnresolvedHaving,
UnresolvedOrdinal,
@@ -48,7 +49,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{
Sort
}
import org.apache.spark.sql.catalyst.trees.CurrentOrigin
-import org.apache.spark.sql.catalyst.util.CollationFactory
+import org.apache.spark.sql.catalyst.util.{toPrettySQL, CollationFactory}
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.util.Utils
@@ -149,6 +150,7 @@ class ExpressionResolver(
private val ordinalResolver = new OrdinalResolver(resolver)
private val lcaResolver = new LateralColumnAliasResolver(this, resolver)
private val semiStructuredExtractResolver = new
SemiStructuredExtractResolver(this)
+ private val extractValueResolver = new ExtractValueResolver(this)
/**
* Get the expression tree traversal stack.
@@ -320,6 +322,8 @@ class ExpressionResolver(
resolveExpressionGenerically(unresolvedDateAddYMInterval)
case extractIntervalPart: ExtractIntervalPart[_] =>
resolveExpressionGenerically(extractIntervalPart)
+ case unresolvedExtractValue: UnresolvedExtractValue =>
+ extractValueResolver.resolve(unresolvedExtractValue)
case unresolvedNamedExpression: NamedExpression =>
resolveNamedExpression(unresolvedNamedExpression)
case unresolvedFunction: UnresolvedFunction =>
@@ -458,7 +462,7 @@ class ExpressionResolver(
resolvedElementContext.hasAggregateExpressionsOutsideWindow
hasLateralColumnAlias |= resolvedElementContext.hasLateralColumnAlias
- Seq(resolvedElement.asInstanceOf[NamedExpression])
+
Seq(aliasOuterReferenceIfNeeded(resolvedElement).asInstanceOf[NamedExpression])
}
ResolvedProjectList(
@@ -535,11 +539,13 @@ class ExpressionResolver(
val resolvedAggregateExpressions =
unresolvedAggregateExpressionsWithStarsExpanded.zipWithIndex.flatMap {
case (expression, index) =>
- val (resolvedElement, resolvedElementContext) =
resolveExpressionTreeInOperatorImpl(
- expression,
- unresolvedAggregate,
- shouldPreserveAlias = true
- )
+ val (unaliasedResolvedElement, resolvedElementContext) =
+ resolveExpressionTreeInOperatorImpl(
+ expression,
+ unresolvedAggregate,
+ shouldPreserveAlias = true
+ )
+ val resolvedElement =
aliasOuterReferenceIfNeeded(unaliasedResolvedElement)
hasLateralColumnAlias |= resolvedElementContext.hasLateralColumnAlias
@@ -570,6 +576,28 @@ class ExpressionResolver(
)
}
+ /**
+ * Wrap a top-level [[OuterReference]] of a subquery's Project/Aggregate
output in an [[Alias]]
+ * so that it gets a fresh ExprId in the subquery's scope. Without the
[[Alias]] the outer
+ * attribute's ExprId leaks through the operator's output (since
[[OuterReference.toAttribute]]
+ * strips the wrapper). Any other expression (including an [[Alias]] from
struct star expansion or
+ * an explicit alias) is left unchanged.
+ *
+ * The alias name follows the same rule as [[AliasResolution.resolve]]: when
the reference was
+ * replaced by an outer aggregate alias, [[AggregateExpressionResolver]]
records the original
+ * attribute's name in
[[OuterReference.SINGLE_PASS_OUTER_AGGREGATE_ALIAS_NAME_OVERRIDE]], which
+ * we honor here.
+ */
+ private def aliasOuterReferenceIfNeeded(resolvedElement: Expression):
Expression =
+ resolvedElement match {
+ case outerReference: OuterReference =>
+ val aliasName = outerReference
+
.getTagValue(OuterReference.SINGLE_PASS_OUTER_AGGREGATE_ALIAS_NAME_OVERRIDE)
+ .getOrElse(toPrettySQL(outerReference.e))
+ autoGeneratedAliasProvider.newAlias(child = outerReference, name =
Some(aliasName))
+ case other => other
+ }
+
/**
* Resolve grouping expressions in [[Aggregate]] operator.
*
@@ -734,7 +762,7 @@ class ExpressionResolver(
private def expandStar(star: Star): Seq[NamedExpression] =
withPosition(star) {
validateStarParentOperator(star)
- scopes.current.expandStar(star)
+ scopes.expandStar(star)
}
/**
@@ -868,6 +896,8 @@ class ExpressionResolver(
private def resolveMultipartName(multipartName: Seq[String]): NameTarget = {
val expressionResolutionContext = expressionResolutionContextStack.peek()
val viewResolutionContext =
resolver.getViewResolver.getViewResolutionContext
+ val extractValueExtractionKey =
+
expressionResolutionContext.parentContext.flatMap(_.extractValueExtractionKey)
scopes.resolveMultipartName(
multipartName = multipartName,
@@ -885,7 +915,8 @@ class ExpressionResolver(
resolvingView = viewResolutionContext.isDefined,
resolvingExecuteImmediate = false,
referredTempVariableNames =
-
viewResolutionContext.map(_.referredTempVariableNames).getOrElse(Seq.empty)
+
viewResolutionContext.map(_.referredTempVariableNames).getOrElse(Seq.empty),
+ extractValueExtractionKey = extractValueExtractionKey
)
)
}
@@ -921,8 +952,9 @@ class ExpressionResolver(
/**
* Process the candidate expression from the [[NameTarget]] resolution.
*
- * - If the candidate is a [[CurrentDate]], we complete the resolution it
using the
- * [[TimezoneAwareExpressionResolver]] to ensure that it is timezone-aware.
+ * - If the candidate is a timezone-aware literal function ([[CurrentDate]]
or [[CurrentTime]]),
+ * complete the resolution using the [[TimezoneAwareExpressionResolver]]
so that the expression
+ * is fully resolved.
* - If the candidate is an [[Attribute]], we pre-emptively collect it for
window source
* expressions.
* - If the candidate is an [[ExtractValue]], we handle it to ensure that
recursive types are
@@ -930,8 +962,8 @@ class ExpressionResolver(
*/
private def processCandidateFromNameTarget(candidate: Expression):
Expression = {
candidate match {
- case currentDate: CurrentDate =>
- timezoneAwareExpressionResolver.resolve(currentDate)
+ case timezoneAware: TimeZoneAwareExpression if
isCurrentDateTimeFunction(timezoneAware) =>
+ timezoneAwareExpressionResolver.resolve(timezoneAware)
case attribute: Attribute =>
attribute
case extractValue: ExtractValue =>
@@ -941,6 +973,11 @@ class ExpressionResolver(
}
}
+ private def isCurrentDateTimeFunction(expression: Expression): Boolean =
expression match {
+ case _: CurrentDate | _: CurrentTime => true
+ case _ => false
+ }
+
/**
* Coerces recursive types ([[ExtractValue]] expressions) in a bottom up
manner.
*/
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExtractValueResolver.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExtractValueResolver.scala
index dad863f20aed..86b295c75447 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExtractValueResolver.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExtractValueResolver.scala
@@ -19,6 +19,8 @@ package org.apache.spark.sql.catalyst.analysis.resolver
import org.apache.spark.sql.catalyst.analysis.UnresolvedExtractValue
import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression, ExtractValue}
+import org.apache.spark.sql.catalyst.util.toPrettySQL
+import org.apache.spark.sql.types.NullType
/**
* Resolver for [[UnresolvedExtractValue]]. Resolve the
[[UnresolvedExtractValue]] by resolving its
@@ -44,6 +46,7 @@ class ExtractValueResolver(expressionResolver:
ExpressionResolver)
private val traversals = expressionResolver.getExpressionTreeTraversals
private val expressionResolutionContextStack =
expressionResolver.getExpressionResolutionContextStack
+ private val autoGeneratedAliasProvider =
expressionResolver.getAutoGeneratedAliasProvider
/**
* Resolves [[UnresolvedExtractValue]] by first resolving its extraction key
and then its child.
@@ -52,6 +55,14 @@ class ExtractValueResolver(expressionResolver:
ExpressionResolver)
*
* Handle the resolved [[ExtractValue]] by type coercing it and collecting
it for window
* resolution, if needed.
+ *
+ * When the base is a `NullType` (e.g. from schema evolution with missing
columns),
+ * [[ExtractValue.applyOrNull]] propagates NULL by returning a bare
`Literal(null, NullType)`
+ * instead of an [[ExtractValue]]. That literal would otherwise be
auto-aliased as `NULL`,
+ * diverging from the fixed-point analyzer which aliases it with the
extraction's text
+ * (e.g. `col[0]`) in [[ColumnResolutionHelper]]. We mirror that here so
both analyzers produce
+ * the same output name; [[CleanupAliases]] trims the alias where it is not
a top-level
+ * projection output.
*/
def resolve(unresolvedExtractValue: UnresolvedExtractValue): Expression = {
val resolvedExtractionKey =
expressionResolver.resolve(unresolvedExtractValue.extraction)
@@ -63,7 +74,7 @@ class ExtractValueResolver(expressionResolver:
ExpressionResolver)
expressionResolutionContextStack.peek().extractValueExtractionKey = None
}
- val resolvedExtractValue = ExtractValue.apply(
+ val resolvedExtractValue = ExtractValue.applyOrNull(
child = resolvedChild,
extraction = resolvedExtractionKey,
resolver = conf.resolver
@@ -71,6 +82,18 @@ class ExtractValueResolver(expressionResolver:
ExpressionResolver)
resolvedExtractValue match {
case extractValue: ExtractValue =>
handleResolvedExtractValue(extractValue)
+ case nullExtraction if resolvedChild.dataType == NullType =>
+ autoGeneratedAliasProvider.newAlias(
+ child = nullExtraction,
+ name = Some(
+ toPrettySQL(
+ unresolvedExtractValue.copy(
+ child = resolvedChild,
+ extraction = resolvedExtractionKey
+ )
+ )
+ )
+ )
case other => other
}
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolver.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolver.scala
index dd70963a7984..74bb90ed6137 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolver.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolver.scala
@@ -22,6 +22,7 @@ import scala.util.Random
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis.{
FunctionResolution,
+ SQLFunctionExpression,
UnresolvedFunction,
UnresolvedSeed
}
@@ -59,13 +60,14 @@ import org.apache.spark.sql.catalyst.util.CollationFactory
* expression is [[TimeZoneAwareExpression]], apply timezone.
*/
class FunctionResolver(
- expressionResolver: ExpressionResolver,
+ protected val expressionResolver: ExpressionResolver,
protected val functionResolution: FunctionResolution,
aggregateExpressionResolver: AggregateExpressionResolver,
binaryArithmeticResolver: BinaryArithmeticResolver)
extends TreeNodeResolver[UnresolvedFunction, Expression]
with ProducesUnresolvedSubtree
- with CoercesExpressionTypes {
+ with CoercesExpressionTypes
+ with FunctionResolverUtils {
private val random = new Random()
private val traversals = expressionResolver.getExpressionTreeTraversals
@@ -131,9 +133,11 @@ class FunctionResolver(
expressionResolutionContext.windowFunctionNestednessLevel += 1
}
+ val unresolvedFunctionWithExpandedStarArgs =
handleStarInArguments(unresolvedFunction)
+
val functionWithResolvedChildren =
withResolvedChildren(
- unresolvedExpression = unresolvedFunction,
+ unresolvedExpression = unresolvedFunctionWithExpandedStarArgs,
resolveChild = expressionResolver.resolve _
).asInstanceOf[UnresolvedFunction]
@@ -148,6 +152,8 @@ class FunctionResolver(
private def handlePartiallyResolvedFunction(partiallyResolvedFunction:
Expression): Expression = {
val expressionResolutionContext = expressionResolutionContextStack.peek()
val resolvedFunction = partiallyResolvedFunction match {
+ case _: SQLFunctionExpression =>
+ throw new ExplicitlyUnsupportedResolverFeature("SQL function
expression")
case collate @ Collate(_, collation: UnresolvedCollation) =>
val withResolvedCollation = collate.copy(
collation = ResolvedCollation(
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolverUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolverUtils.scala
index 3c5a3f1832e8..1ae1c9863f65 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolverUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolverUtils.scala
@@ -17,8 +17,6 @@
package org.apache.spark.sql.catalyst.analysis.resolver
-import java.util.Locale
-
import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.analysis.{
FunctionResolution,
@@ -98,12 +96,9 @@ trait FunctionResolverUtils {
* Method used to determine whether the given function is non-distinct
`count` function,
* with optional normalization.
*/
- private def isNonDistinctCount(
- unresolvedFunction: UnresolvedFunction,
- normalizeFunctionName: Boolean = true
- ): Boolean = {
+ private def isNonDistinctCount(unresolvedFunction: UnresolvedFunction):
Boolean = {
!unresolvedFunction.isDistinct &&
- isCount(unresolvedFunction, normalizeFunctionName) &&
+ isCount(unresolvedFunction) &&
!isUnqualifiedCountShadowedByTemp(unresolvedFunction)
}
@@ -119,17 +114,8 @@ trait FunctionResolverUtils {
.isTemporaryFunction(FunctionIdentifier(unresolvedFunction.nameParts.head))
}
- private def isCount(
- unresolvedFunction: UnresolvedFunction,
- normalizeFunctionName: Boolean = true
- ): Boolean = {
- val isCountName = if (normalizeFunctionName) {
- unresolvedFunction.nameParts.head.toLowerCase(Locale.ROOT) == "count"
- } else {
- unresolvedFunction.nameParts.head == "count"
- }
-
- unresolvedFunction.nameParts.length == 1 && isCountName
+ private def isCount(unresolvedFunction: UnresolvedFunction): Boolean = {
+
FunctionResolution.isUnqualifiedOrBuiltinFunctionName(unresolvedFunction.nameParts,
"count")
}
/**
@@ -150,19 +136,13 @@ trait FunctionResolverUtils {
/**
* Throws an exception according to
[[SQLConf.ALLOW_STAR_WITH_SINGLE_TABLE_IDENTIFIER_IN_COUNT]].
*
- * Note that check for function name is case-sensitive. Even when flag is
false we allow
- * `COUNT(tableName.*)` but block `count(tableName.*)`. This is the same
behavior as in
- * fixed-point analyzer, and clearly it is a bug. If this is ever fixed it
should be done in both
- * analyzers simultaneously.
- *
* See [[handleStarInArguments]]
*/
private def assertSingleTableStarNotInCountFunction(
unresolvedFunction: UnresolvedFunction): Unit = {
- if (!conf.allowStarWithSingleTableIdentifierInCount && isCount(
- unresolvedFunction = unresolvedFunction,
- normalizeFunctionName = false
- ) && unresolvedFunction.arguments.length == 1) {
+ if (!conf.allowStarWithSingleTableIdentifierInCount &&
+ isCount(unresolvedFunction) &&
+ unresolvedFunction.arguments.length == 1) {
unresolvedFunction.arguments.head match {
case star: UnresolvedStar if
scopes.current.isStarQualifiedByTable(star) =>
throw QueryCompilationErrors
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/NameScope.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/NameScope.scala
index 2c39fe71e62e..ad232adfd273 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/NameScope.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/NameScope.scala
@@ -22,6 +22,7 @@ import java.util.{ArrayDeque, ArrayList, HashMap, HashSet,
LinkedHashMap}
import scala.collection.mutable
import scala.jdk.CollectionConverters._
+import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.analysis.{
ExpandStarParameters,
@@ -42,7 +43,7 @@ import org.apache.spark.sql.catalyst.expressions.{
NamedExpression,
OuterReference
}
-import org.apache.spark.sql.catalyst.plans.logical.Aggregate
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, CollectMetrics,
Project}
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.internal.SQLConf
@@ -421,6 +422,25 @@ class NameScope(
)
}
+ /**
+ * Checks whether the given [[Star]] can be resolved in this scope without
performing the actual
+ * expansion. This is used to decide whether to attempt outer scope
expansion: if the star can
+ * be resolved in the current scope, there is no need to probe outer scopes.
+ *
+ * For targeted stars (e.g. `t1.*`), this checks whether the target is
resolvable as either a
+ * table qualifier or a struct name. Untargeted stars and other [[Star]]
subtypes always resolve
+ * in the current scope (untargeted stars expand to the full output, and
other subtypes like
+ * [[UnresolvedRegex]] are handled directly by [[expandStar]]).
+ */
+ def canExpandStarFromCurrentScope(star: Star): Boolean = {
+ star match {
+ case starBase: UnresolvedStarBase if starBase.target.isDefined =>
+ isStarQualifiedByTable(starBase) ||
+ resolveNameInStarExpansion(starBase.target.get,
nameComparator).isDefined
+ case _ => true
+ }
+ }
+
/**
* Find attributes in this [[NameScope]] that match a provided one-part
`name`.
*
@@ -1363,6 +1383,60 @@ class NameScopeStack(
}
}
+ /**
+ * Expand a [[Star]] expression. The resolution strategy is:
+ * 1. If the current scope can resolve the star
([[NameScope.canExpandStarFromCurrentScope]]),
+ * expand it there.
+ * 2. Otherwise, try to expand from the nearest outer scope
([[expandStarFromOuterScope]]),
+ * wrapping results in [[OuterReference]].
+ * 3. If neither succeeds, fall back to [[NameScope.expandStar]] on the
current scope solely
+ * to throw the appropriate error with the correct context (e.g.
available column names).
+ */
+ def expandStar(star: Star): Seq[NamedExpression] = {
+ if (current.canExpandStarFromCurrentScope(star)) {
+ current.expandStar(star)
+ } else {
+ val fromOuterScope = expandStarFromOuterScope(star)
+ if (fromOuterScope.nonEmpty) {
+ fromOuterScope
+ } else {
+ current.expandStar(star)
+ }
+ }
+ }
+
+ private def expandStarFromOuterScope(star: Star): Seq[NamedExpression] = {
+ (star, subqueryRegistry.currentScope.parentOperator) match {
+ case (starBase: UnresolvedStarBase, Some(_: Project | _: Aggregate | _:
CollectMetrics))
+ if starBase.target.isDefined =>
+ outerScopes.headOption match {
+ case Some(outerScope) if
outerScope.canExpandStarFromCurrentScope(star) =>
+ outerScope.expandStar(star).map { namedExpression =>
+ val wrapped = wrapCandidateInOuterReference(
+ candidate = namedExpression,
+ outerScope = outerScope,
+ subqueryScope = subqueryRegistry.currentScope,
+ outerScopeLevel = 1
+ )
+
+ wrapped match {
+ case alias: Alias =>
+ alias
+ case named: NamedExpression =>
+ Alias(named, toPrettySQL(namedExpression))()
+ case other =>
+ throw SparkException.internalError(
+ s"Expected NamedExpression from outer star expansion, got
" +
+ s"${other.getClass.getSimpleName}"
+ )
+ }
+ }
+ case _ => Seq.empty
+ }
+ case _ => Seq.empty
+ }
+ }
+
/**
* If `spark.sql.optimizer.supportNestedCorrelatedSubqueries.enabled` is set
to false, find
* the nearest outer scope and return it if we are in a subquery. Otherwise,
returns all outer
@@ -1405,6 +1479,14 @@ class NameScopeStack(
outerReference
case other => other
}
+ case alias: Alias =>
+ val wrappedChild = wrapCandidateInOuterReference(
+ candidate = alias.child,
+ outerScope = outerScope,
+ subqueryScope = subqueryScope,
+ outerScopeLevel = outerScopeLevel
+ )
+ alias.withNewChildren(Seq(wrappedChild))
case attribute: Attribute =>
val outerReference =
tryReplaceOuterReferenceAttributeWithAlias(attribute, outerScope)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala
index 3ba0121dbb64..99c79aaee3f6 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.catalyst.analysis.{
UnresolvedInlineTable,
UnresolvedOrdinal,
UnresolvedRelation,
- UnresolvedStar,
+ UnresolvedStarBase,
UnresolvedStarExceptOrReplace,
UnresolvedSubqueryColumnAliases
}
@@ -219,16 +219,8 @@ class ResolverGuard(
checkNamedParameter(namedParameter)
case getStructField: GetStructField =>
checkGetStructField(getStructField)
- case lambdaFunction: LambdaFunction =>
- checkLambdaFunction(lambdaFunction)
- case unresolvedNamedLambdaVariable: UnresolvedNamedLambdaVariable =>
- checkUnresolvedNamedLambdaVariable(unresolvedNamedLambdaVariable)
- case namedLambdaVariable: NamedLambdaVariable =>
- checkNamedLambdaVariable(namedLambdaVariable)
case NamePlaceholder =>
checkNamePlaceholder()
- case baseGroupingSets: BaseGroupingSets =>
- checkBaseGroupingSets(baseGroupingSets)
case expression if isGenerallySupportedExpression(expression) =>
expression.children.collectFirst { case CheckExpression(reason) =>
reason }
case _ =>
@@ -265,9 +257,7 @@ class ResolverGuard(
checkProjectHiddenOutputTag(project)
.orElse(checkOperator(project.child))
.orElse {
-
project.projectList.filterNot(_.isInstanceOf[UnresolvedStar]).collectFirst {
- case CheckExpression(reason) => reason
- }
+ project.projectList.collectFirst { case CheckExpression(reason) =>
reason }
}
}
@@ -277,9 +267,7 @@ class ResolverGuard(
aggregate.groupingExpressions.collectFirst { case
CheckExpression(reason) => reason }
}
.orElse {
-
aggregate.aggregateExpressions.filterNot(_.isInstanceOf[UnresolvedStar]).collectFirst
{
- case CheckExpression(reason) => reason
- }
+ aggregate.aggregateExpressions.collectFirst { case
CheckExpression(reason) => reason }
}
}
@@ -392,11 +380,28 @@ class ResolverGuard(
* [[UnresolvedStarExceptOrReplace]] is handled separately, because it's a
leaf expression,
* but it has replacement expressions, that could be non-trivial.
*/
- private def checkStar(star: Star) = star match {
- case starExceptOrReplace: UnresolvedStarExceptOrReplace =>
- starExceptOrReplace.replacements.collectFirst { case
CheckExpressionSeq(reason) => reason }
- case star =>
- star.children.collectFirst { case CheckExpression(reason) => reason }
+ private def checkStar(star: Star) =
+ checkStarTarget(star).orElse {
+ star match {
+ case starExceptOrReplace: UnresolvedStarExceptOrReplace =>
+ starExceptOrReplace.replacements.collectFirst {
+ case CheckExpressionSeq(reason) => reason
+ }
+ case star =>
+ star.children.collectFirst { case CheckExpression(reason) => reason }
+ }
+ }
+
+ private def checkStarTarget(star: Star): Option[String] = star match {
+ case starBase: UnresolvedStarBase =>
+ starBase.target.flatMap { targetParts =>
+ targetParts
+ .find(name =>
ResolverGuard.UNSUPPORTED_ATTRIBUTE_NAMES.contains(name))
+ .map(unsupportedName =>
+ s"unsupported attribute name
'${unsupportedName.toLowerCase(Locale.ROOT)}'"
+ )
+ }
+ case _ => None
}
private def checkNamedParameter(namedParameter: NamedParameter) = None
@@ -405,21 +410,8 @@ class ResolverGuard(
checkExpression(getStructField.child)
}
- private def checkLambdaFunction(lambdaFunction: LambdaFunction) = {
- lambdaFunction.children.collectFirst { case CheckExpression(reason) =>
reason }
- }
-
- private def checkUnresolvedNamedLambdaVariable(
- unresolvedNamedLambdaVariable: UnresolvedNamedLambdaVariable) = None
-
- private def checkNamedLambdaVariable(namedLambdaVariable:
NamedLambdaVariable) = None
-
private def checkNamePlaceholder() = None
- private def checkBaseGroupingSets(baseGroupingSets: BaseGroupingSets) = {
- baseGroupingSets.children.collectFirst { case CheckExpression(reason) =>
reason }
- }
-
private def checkUnresolvedAlias(unresolvedAlias: UnresolvedAlias) =
checkExpression(unresolvedAlias.child)
@@ -486,14 +478,13 @@ class ResolverGuard(
} else if (FunctionResolution.sessionNamespaceKind(nameParts)
.contains(org.apache.spark.sql.catalyst.catalog.SessionCatalog.Builtin)) {
// Explicitly builtin-qualified: reject if unsupported, else check
children
- if (ResolverGuard.UNSUPPORTED_FUNCTION_NAMES.contains(funcName)) {
+ if (isUnsupportedFunction(funcName)) {
Some(s"unsupported function ${funcName}")
} else {
unresolvedFunction.children.collectFirst { case
CheckExpression(reason) => reason }
}
} else if (FunctionResolution.sessionNamespaceKind(nameParts).isDefined) {
- // Session-qualified: allow through (PATH + system-first)
- unresolvedFunction.children.collectFirst { case CheckExpression(reason)
=> reason }
+ Some("session-qualified user-defined function")
} else {
Some("multi-part function name")
}
@@ -625,6 +616,10 @@ class ResolverGuard(
Some("legacyCTEPrecedencePolicy")
} else if (conf.getConfString("pipelines.id", null) != null) {
Some("dlt")
+ } else if (!conf.getConf(SQLConf.PRIORITIZE_ORDINAL_RESOLUTION_IN_SORT)) {
+ Some("prioritizeOrdinalResolutionInSortDisabled")
+ } else if (conf.getConf(SQLConf.PERSISTENT_CATALOG_FIRST)) {
+ Some("persistentCatalogFirst")
} else {
None
}
@@ -639,7 +634,9 @@ class ResolverGuard(
}
private def isUnsupportedFunction(name: String): Boolean = {
- ResolverGuard.UNSUPPORTED_FUNCTION_NAMES.contains(name)
+ ResolverGuard.UNSUPPORTED_FUNCTION_NAMES.contains(name) ||
+ ResolverGuard.GENERATOR_FUNCTION_NAMES.contains(name) ||
+ ResolverGuard.HIGHER_ORDER_FUNCTIONS.contains(name)
}
private def isBuiltinFunction(singlePartName: String) = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/SubqueryScope.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/SubqueryScope.scala
index 2dfbbe65b722..6f83dd0fb6b2 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/SubqueryScope.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/SubqueryScope.scala
@@ -93,9 +93,13 @@ import
org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPl
* @param aggregateExpressionsExtractor
[[GroupingAndAggregateExpressionsExtractor]] object defined
* in case the outer parent operator is a partially resolved HAVING. Used to
extract the
* aggregate expressions from the outer [[Aggregate]] operator.
+ * @param parentOperator the operator hosting this subquery expression. Used by
+ * [[NameScopeStack.expandStarFromOuterScope]] to gate star expansion from
outer scopes to
+ * `Project`, `Aggregate`, or `CollectMetrics` operators, matching the
fixed-point behavior.
*/
class SubqueryScope(
val aggregateExpressionsExtractor:
Option[GroupingAndAggregateExpressionsExtractor] = None,
+ val parentOperator: Option[LogicalPlan] = None,
private val outerOperatorResolutionContext: OperatorResolutionContext =
new OperatorResolutionContext,
private val outerReferenceScopeLevels: HashMap[ExprId, Int] = new
HashMap[ExprId, Int]) {
@@ -167,6 +171,7 @@ class SubqueryRegistry {
stack.push(
new SubqueryScope(
aggregateExpressionsExtractor =
groupingAndAggregateExpressionsExtractor,
+ parentOperator = Some(parentOperator),
outerOperatorResolutionContext = outerOperatorResolutionContext
)
)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala
index d958cba27933..4a034f4fada4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala
@@ -142,7 +142,7 @@ class AttributeSet private (private val baseSet:
mutable.LinkedHashSet[Attribute
* `this` and `other`.
*/
def intersect(other: AttributeSet): AttributeSet =
- new AttributeSet(baseSet.intersect(other.baseSet))
+ new AttributeSet(other.baseSet.filter(baseSet.contains))
override def foreach[U](f: (Attribute) => U): Unit =
baseSet.map(_.a).foreach(f)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
index 8ca3f76c3b88..785131875114 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
@@ -26,6 +26,7 @@ import scala.util.control.NonFatal
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper
import org.apache.spark.sql.catalyst.rules.RuleId
import org.apache.spark.sql.catalyst.rules.UnknownRuleId
import org.apache.spark.sql.catalyst.trees.{AlwaysProcess, CurrentOrigin,
TreeNode, TreeNodeTag}
@@ -798,9 +799,10 @@ object QueryPlan extends PredicateHelper {
e.transformUp {
case s: PlanExpression[QueryPlan[_] @unchecked] =>
// Normalize the outer references in the subquery plan.
- val normalizedPlan = s.plan.transformAllExpressionsWithPruning(
- _.containsPattern(OUTER_REFERENCE)) {
- case OuterReference(r) =>
OuterReference(QueryPlan.normalizeExpressions(r, input))
+ val normalizedPlan = AnalysisHelper.allowInvokingTransformsInAnalyzer {
+
s.plan.transformAllExpressionsWithPruning(_.containsPattern(OUTER_REFERENCE)) {
+ case OuterReference(r) =>
OuterReference(QueryPlan.normalizeExpressions(r, input))
+ }
}
s.withNewPlan(normalizedPlan)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index cd66d747380a..38d107cd2260 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -907,8 +907,6 @@ case class View(
isTempView: Boolean,
child: LogicalPlan,
options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty)
extends UnaryNode {
- require(!isTempViewStoringAnalyzedPlan || child.resolved)
-
override def output: Seq[Attribute] = child.output
override def metadataOutput: Seq[Attribute] = Nil
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 965b80620968..5af33da7ae8e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -410,7 +410,8 @@ object SQLConf {
)
.version("4.1.0")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(
+
sys.env.get("SPARK_SINGLE_PASS_ANALYZER_ENABLED_TENTATIVELY").contains("true"))
val ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER =
buildConf("spark.sql.analyzer.singlePassResolver.dualRunWithLegacy")
@@ -422,7 +423,7 @@ object SQLConf {
)
.version("4.0.0")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(Utils.isTesting)
val ANALYZER_DUAL_RUN_RETURN_SINGLE_PASS_RESULT =
buildConf("spark.sql.analyzer.singlePassResolver.returnSinglePassResultInDualRun")
@@ -434,7 +435,7 @@ object SQLConf {
)
.version("4.0.0")
.booleanConf
- .createWithDefault(Utils.isTesting)
+ .createWithDefault(false)
val ANALYZER_DUAL_RUN_SAMPLE_RATE =
buildConf("spark.sql.analyzer.singlePassResolver.dualRunSampleRate")
@@ -447,7 +448,7 @@ object SQLConf {
)
.version("4.1.0")
.doubleConf
- .createWithDefault(if (Utils.isTesting) 1.0 else 0.001)
+ .createWithDefault(if (Utils.isTesting) 1.0 else 0.1)
val ANALYZER_SINGLE_PASS_RESOLVER_EXPOSE_RESOLVER_GUARD_FAILURE =
buildConf("spark.sql.analyzer.singlePassResolver.exposeResolverGuardFailure")
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/EndToEndAPISuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/EndToEndAPISuite.scala
index f674b45bb072..4acc60e2ac71 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/EndToEndAPISuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/EndToEndAPISuite.scala
@@ -24,6 +24,7 @@ import scala.concurrent.duration.Duration
import org.apache.spark.api.python.PythonUtils
import org.apache.spark.sql.connect.{PythonTestDepsChecker,
SparkConnectServerTest}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.pipelines.utils.{APITest, PipelineReference,
PipelineSourceFile, PipelineTest, TestPipelineConfiguration, TestPipelineSpec}
case class PipelineReferenceImpl(executionProcess: Process) extends
PipelineReference
@@ -165,6 +166,7 @@ class EndToEndAPISuite extends PipelineTest with APITest
with SparkConnectServer
|storage: "file://${projectDir.resolve("storage").toAbsolutePath}"
|configuration:
| "spark.remote": "sc://localhost:$serverPort"
+ | "${SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key}":
"false"
|libraries:
|$libraries
|""".stripMargin
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerTest.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerTest.scala
index 80fbfe581886..384ad8ff7328 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerTest.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerTest.scala
@@ -19,15 +19,22 @@ package org.apache.spark.sql.connect.pipelines
import scala.collection.mutable.ArrayBuffer
+import org.apache.spark.SparkConf
import org.apache.spark.connect.{proto => sc}
import org.apache.spark.connect.proto.{PipelineCommand, PipelineEvent}
import org.apache.spark.sql.connect.{SparkConnectServerTest,
SparkConnectTestUtils}
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
import org.apache.spark.sql.connect.service.{SessionHolder, SessionKey,
SparkConnectService}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.pipelines.utils.{PipelineTest, StorageRootMixin}
class SparkDeclarativePipelinesServerTest extends SparkConnectServerTest with
StorageRootMixin {
+ override def sparkConf: SparkConf = {
+ super.sparkConf
+ .set(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key,
"false")
+ }
+
override def afterEach(): Unit = {
SparkConnectService.sessionManager
.getIsolatedSessionIfPresent(SessionKey(defaultUserId, defaultSessionId))
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/order-by.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/order-by.sql.out
index abaaf9c9fb8c..6e37ab836e3c 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/order-by.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/order-by.sql.out
@@ -111,7 +111,7 @@ SELECT a, (SELECT b FROM VALUES (1, 2) AS innerTestData
(all, b) ORDER BY ALL) F
Project [a#x, scalar-subquery#x [] AS scalarsubquery()#x]
: +- Project [b#x]
: +- Sort [ALL#x ASC NULLS FIRST], true
-: +- Project [b#x, ALL#x]
+: +- Project [b#x, all#x]
: +- SubqueryAlias innerTestData
: +- LocalRelation [all#x, b#x]
+- SubqueryAlias testdata
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/replacing-missing-expression-with-alias.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/replacing-missing-expression-with-alias.sql.out
index a353c9e8ed01..846c28faf180 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/replacing-missing-expression-with-alias.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/replacing-missing-expression-with-alias.sql.out
@@ -1,4 +1,10 @@
-- Automatically generated by SQLQueryTestSuite
+-- !query
+SET
spark.sql.analyzer.singlePassResolver.preventUsingAliasesFromNonDirectChildren
= true
+-- !query analysis
+SetCommand
(spark.sql.analyzer.singlePassResolver.preventUsingAliasesFromNonDirectChildren,Some(true))
+
+
-- !query
SELECT col1 + 1 AS a FROM VALUES(1) GROUP BY a ORDER BY col1 + 1
-- !query analysis
@@ -180,6 +186,18 @@ Sort [b#x ASC NULLS FIRST, col2#x ASC NULLS FIRST, c#x ASC
NULLS FIRST], true
+- LocalRelation [col1#x, col2#x]
+-- !query
+SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=false
+-- !query analysis
+SetCommand
(spark.sql.analyzer.singlePassResolver.dualRunWithLegacy,Some(false))
+
+
+-- !query
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false
+-- !query analysis
+SetCommand
(spark.sql.analyzer.singlePassResolver.enabledTentatively,Some(false))
+
+
-- !query
SELECT col1 AS a FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1 ORDER BY
col1
-- !query analysis
@@ -202,3 +220,45 @@ Project [a#x, b#x]
+- Project [col1#x, col2#x, col1#x AS a#x]
+- Aggregate [col1#x, col2#x], [col1#x, col2#x]
+- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=true
+-- !query analysis
+SetCommand
(spark.sql.analyzer.singlePassResolver.enabledTentatively,Some(true))
+
+
+-- !query
+SELECT col1 AS a FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1 ORDER BY
col1
+-- !query analysis
+Project [a#x]
++- Sort [col1#x ASC NULLS FIRST], true
+ +- Project [a#x, col1#x]
+ +- Filter (col2#x > 1)
+ +- Aggregate [col1#x, col2#x], [col1#x AS a#x, col2#x, col1#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+SELECT col1 AS a, a AS b FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1
ORDER BY col1
+-- !query analysis
+Project [a#x, b#x]
++- Sort [col1#x ASC NULLS FIRST], true
+ +- Project [a#x, b#x, col1#x]
+ +- Filter (col2#x > 1)
+ +- Project [a#x, a#x AS b#x, col2#x, col1#x]
+ +- Project [col1#x, col1#x AS a#x, col2#x]
+ +- Aggregate [col1#x, col2#x], [col1#x, col2#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=true
+-- !query analysis
+SetCommand (spark.sql.analyzer.singlePassResolver.dualRunWithLegacy,Some(true))
+
+
+-- !query
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false
+-- !query analysis
+SetCommand
(spark.sql.analyzer.singlePassResolver.enabledTentatively,Some(false))
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/udf/postgreSQL/udf-join.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/udf/postgreSQL/udf-join.sql.out
index 9edacd126afe..f424e804176d 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/udf/postgreSQL/udf-join.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/udf/postgreSQL/udf-join.sql.out
@@ -545,7 +545,7 @@ SELECT udf(udf('')) AS `xxx`, udf(i), udf(j), udf(t), udf(k)
-- !query analysis
Project [xxx#x, udf(i)#x, udf(j)#x, udf(t)#x, udf(k)#x]
+- Sort [cast(udf(cast(cast(udf(cast(i#x as string)) as int) as string)) as
int) ASC NULLS FIRST, cast(udf(cast(k#x as string)) as int) ASC NULLS FIRST,
cast(udf(cast(t#x as string)) as string) ASC NULLS FIRST], true
- +- Project [cast(udf(cast(cast(udf(cast( as string)) as string) as string))
as string) AS xxx#x, cast(udf(cast(i#x as string)) as int) AS udf(i)#x,
cast(udf(cast(j#x as string)) as int) AS udf(j)#x, cast(udf(cast(t#x as
string)) as string) AS udf(t)#x, cast(udf(cast(k#x as string)) as int) AS
udf(k)#x, i#x, k#x, t#x]
+ +- Project [cast(udf(cast(cast(udf(cast( as string)) as string) as string))
as string) AS xxx#x, cast(udf(cast(i#x as string)) as int) AS udf(i)#x,
cast(udf(cast(j#x as string)) as int) AS udf(j)#x, cast(udf(cast(t#x as
string)) as string) AS udf(t)#x, cast(udf(cast(k#x as string)) as int) AS
udf(k)#x, i#x, t#x, k#x]
+- Project [i#x, j#x, t#x, k#x]
+- Join LeftOuter, (i#x = i#x)
:- SubqueryAlias spark_catalog.default.j1_tbl
@@ -561,7 +561,7 @@ SELECT udf('') AS `xxx`, udf(i), udf(j), udf(t), udf(k)
-- !query analysis
Project [xxx#x, udf(i)#x, udf(j)#x, udf(t)#x, udf(k)#x]
+- Sort [cast(udf(cast(i#x as string)) as int) ASC NULLS FIRST,
cast(udf(cast(cast(udf(cast(k#x as string)) as int) as string)) as int) ASC
NULLS FIRST, cast(udf(cast(t#x as string)) as string) ASC NULLS FIRST], true
- +- Project [cast(udf(cast( as string)) as string) AS xxx#x,
cast(udf(cast(i#x as string)) as int) AS udf(i)#x, cast(udf(cast(j#x as
string)) as int) AS udf(j)#x, cast(udf(cast(t#x as string)) as string) AS
udf(t)#x, cast(udf(cast(k#x as string)) as int) AS udf(k)#x, i#x, k#x, t#x]
+ +- Project [cast(udf(cast( as string)) as string) AS xxx#x,
cast(udf(cast(i#x as string)) as int) AS udf(i)#x, cast(udf(cast(j#x as
string)) as int) AS udf(j)#x, cast(udf(cast(t#x as string)) as string) AS
udf(t)#x, cast(udf(cast(k#x as string)) as int) AS udf(k)#x, i#x, t#x, k#x]
+- Project [i#x, j#x, t#x, k#x]
+- Join LeftOuter, (i#x = i#x)
:- SubqueryAlias spark_catalog.default.j1_tbl
@@ -603,7 +603,7 @@ SELECT udf('') AS `xxx`, udf(i), udf(j), udf(udf(t)), udf(k)
-- !query analysis
Project [xxx#x, udf(i)#x, udf(j)#x, udf(udf(t))#x, udf(k)#x]
+- Sort [cast(udf(cast(cast(udf(cast(i#x as string)) as int) as string)) as
int) ASC NULLS FIRST, cast(udf(cast(k#x as string)) as int) ASC NULLS FIRST,
cast(udf(cast(t#x as string)) as string) ASC NULLS FIRST], true
- +- Project [cast(udf(cast( as string)) as string) AS xxx#x,
cast(udf(cast(i#x as string)) as int) AS udf(i)#x, cast(udf(cast(j#x as
string)) as int) AS udf(j)#x, cast(udf(cast(cast(udf(cast(t#x as string)) as
string) as string)) as string) AS udf(udf(t))#x, cast(udf(cast(k#x as string))
as int) AS udf(k)#x, i#x, k#x, t#x]
+ +- Project [cast(udf(cast( as string)) as string) AS xxx#x,
cast(udf(cast(i#x as string)) as int) AS udf(i)#x, cast(udf(cast(j#x as
string)) as int) AS udf(j)#x, cast(udf(cast(cast(udf(cast(t#x as string)) as
string) as string)) as string) AS udf(udf(t))#x, cast(udf(cast(k#x as string))
as int) AS udf(k)#x, i#x, t#x, k#x]
+- Project [coalesce(i#x, i#x) AS i#x, j#x, t#x, k#x]
+- Join FullOuter, (i#x = i#x)
:- SubqueryAlias spark_catalog.default.j1_tbl
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/cte-with-multi-children-operators.sql
b/sql/core/src/test/resources/sql-tests/inputs/cte-with-multi-children-operators.sql
index 731ade96c40f..42cdcbd5c08f 100644
---
a/sql/core/src/test/resources/sql-tests/inputs/cte-with-multi-children-operators.sql
+++
b/sql/core/src/test/resources/sql-tests/inputs/cte-with-multi-children-operators.sql
@@ -1,3 +1,10 @@
+-- We run these tests with single-pass analyzer separately because in
fixed-point we sometimes push
+-- the WithCte node below the multi-children operators (not consistent)
whereas in the single-pass
+-- resolver we always keep it above.
+
+--SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false
+--SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=false
+
-- This test tests CTEs with multi-children operators (Union, Intersect,
Except, Join etc).
CREATE TEMPORARY VIEW t1 AS SELECT * FROM VALUES
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/having-and-order-by-recursive-type-name-resolution.sql
b/sql/core/src/test/resources/sql-tests/inputs/having-and-order-by-recursive-type-name-resolution.sql
index de3f5c8cc43f..0f411de350fe 100644
---
a/sql/core/src/test/resources/sql-tests/inputs/having-and-order-by-recursive-type-name-resolution.sql
+++
b/sql/core/src/test/resources/sql-tests/inputs/having-and-order-by-recursive-type-name-resolution.sql
@@ -1,3 +1,11 @@
+-- We run these tests with fixed-point analyzer separately because for some
cases with recursive
+-- types, fixed-point analyzer either fails to resolve the query completely or
resolves it to a
+-- slightly different logical plan due to
spark.sql.analyzer.singlePassResolver.preventUsingAliasesFromNonDirectChildren
+-- not being set in all tests.
+
+--SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false
+--SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=false
+
-- This test file contains queries that test recursive types name resolution
in ORDER BY and HAVING clauses.
-- Alias type: String, Table column type: Struct
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/order-by-and-having-on-top-of-aggregate-with-join.sql
b/sql/core/src/test/resources/sql-tests/inputs/order-by-and-having-on-top-of-aggregate-with-join.sql
index 4e50f4c4a365..cad48b64e278 100644
---
a/sql/core/src/test/resources/sql-tests/inputs/order-by-and-having-on-top-of-aggregate-with-join.sql
+++
b/sql/core/src/test/resources/sql-tests/inputs/order-by-and-having-on-top-of-aggregate-with-join.sql
@@ -1,3 +1,6 @@
+--SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false
+--SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=false
+
CREATE OR REPLACE TEMPORARY VIEW v1 AS SELECT * FROM VALUES
(1, 2),
(3, 5),
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/replacing-missing-expression-with-alias.sql
b/sql/core/src/test/resources/sql-tests/inputs/replacing-missing-expression-with-alias.sql
index 0f238033fada..02f571996cb9 100644
---
a/sql/core/src/test/resources/sql-tests/inputs/replacing-missing-expression-with-alias.sql
+++
b/sql/core/src/test/resources/sql-tests/inputs/replacing-missing-expression-with-alias.sql
@@ -1,3 +1,5 @@
+SET
spark.sql.analyzer.singlePassResolver.preventUsingAliasesFromNonDirectChildren
= true;
+
-- Replace expression with alias that has semantically equal child if
expression is not in output
SELECT col1 + 1 AS a FROM VALUES(1) GROUP BY a ORDER BY col1 + 1;
SELECT col1 + 1 AS a, a AS b FROM VALUES(1) GROUP BY a ORDER BY col1 + 1;
@@ -28,6 +30,22 @@ SELECT col2 AS b, col2 FROM VALUES(1,2) GROUP BY 1,2 HAVING
col2 > 0 ORDER BY AL
SELECT col2 AS b, col2, b as c FROM VALUES(1,2) GROUP BY 1,2 ORDER BY ALL;
SELECT col2 AS b, col2, b as c FROM VALUES(1,2) GROUP BY 1,2 HAVING col2 > 0
ORDER BY ALL;
+-- We run the following tests in fixed-point and single-pass modes separately
because there is a difference
+-- in results. Difference is order in inner Project nodes which is usually
normalized when we compare results between
+-- those two modes but in golden files we have to run them one after another.
+
-- Fixed point doesn't know to correctly replace `col1` with `a` because
HAVING adds an artificial Project node
+
+-- Fixed-point mode
+SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=false;
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false;
SELECT col1 AS a FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1 ORDER BY
col1;
SELECT col1 AS a, a AS b FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1
ORDER BY col1;
+
+-- Single-pass mode
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=true;
+SELECT col1 AS a FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1 ORDER BY
col1;
+SELECT col1 AS a, a AS b FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1
ORDER BY col1;
+
+SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=true;
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false;
diff --git
a/sql/core/src/test/resources/sql-tests/results/replacing-missing-expression-with-alias.sql.out
b/sql/core/src/test/resources/sql-tests/results/replacing-missing-expression-with-alias.sql.out
index f3a6b0424b65..c52c3bdd41e6 100644
---
a/sql/core/src/test/resources/sql-tests/results/replacing-missing-expression-with-alias.sql.out
+++
b/sql/core/src/test/resources/sql-tests/results/replacing-missing-expression-with-alias.sql.out
@@ -1,4 +1,12 @@
-- Automatically generated by SQLQueryTestSuite
+-- !query
+SET
spark.sql.analyzer.singlePassResolver.preventUsingAliasesFromNonDirectChildren
= true
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.analyzer.singlePassResolver.preventUsingAliasesFromNonDirectChildren
true
+
+
-- !query
SELECT col1 + 1 AS a FROM VALUES(1) GROUP BY a ORDER BY col1 + 1
-- !query schema
@@ -162,6 +170,46 @@ struct<b:int,col2:int,c:int>
2 2 2
+-- !query
+SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=false
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.analyzer.singlePassResolver.dualRunWithLegacy false
+
+
+-- !query
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.analyzer.singlePassResolver.enabledTentatively false
+
+
+-- !query
+SELECT col1 AS a FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1 ORDER BY
col1
+-- !query schema
+struct<a:int>
+-- !query output
+1
+
+
+-- !query
+SELECT col1 AS a, a AS b FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1
ORDER BY col1
+-- !query schema
+struct<a:int,b:int>
+-- !query output
+1 1
+
+
+-- !query
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=true
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.analyzer.singlePassResolver.enabledTentatively true
+
+
-- !query
SELECT col1 AS a FROM VALUES(1,2) GROUP BY col1, col2 HAVING col2 > 1 ORDER BY
col1
-- !query schema
@@ -176,3 +224,19 @@ SELECT col1 AS a, a AS b FROM VALUES(1,2) GROUP BY col1,
col2 HAVING col2 > 1 OR
struct<a:int,b:int>
-- !query output
1 1
+
+
+-- !query
+SET spark.sql.analyzer.singlePassResolver.dualRunWithLegacy=true
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.analyzer.singlePassResolver.dualRunWithLegacy true
+
+
+-- !query
+SET spark.sql.analyzer.singlePassResolver.enabledTentatively=false
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.analyzer.singlePassResolver.enabledTentatively false
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index 5786199cb376..d7908279deb5 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -5062,7 +5062,8 @@ class SQLQuerySuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
for (confValue <- Seq(false, true)) {
withSQLConf(
- SQLConf.UNION_IS_RESOLVED_WHEN_DUPLICATES_PER_CHILD_RESOLVED.key ->
confValue.toString
+ SQLConf.UNION_IS_RESOLVED_WHEN_DUPLICATES_PER_CHILD_RESOLVED.key ->
confValue.toString,
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key ->
confValue.toString
) {
val analyzedPlan = sql(
"""SELECT
@@ -5133,7 +5134,11 @@ class SQLQuerySuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
checkAnswer(sql(query), Row(1, 1))
}
- withSQLConf(SQLConf.PREFER_COLUMN_OVER_LCA_IN_ARRAY_INDEX.key -> "false") {
+ withSQLConf(
+ SQLConf.PREFER_COLUMN_OVER_LCA_IN_ARRAY_INDEX.key -> "false",
+ // Single-pass analyzer doesn't support this legacy behavior.
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false"
+ ) {
checkAnswer(sql(query), Row(1, 2))
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala
index b40e3799a9c0..75ece63af193 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala
@@ -19,10 +19,11 @@ package org.apache.spark.sql
import org.apache.spark.SparkIllegalArgumentException
import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException
+import
org.apache.spark.sql.catalyst.analysis.resolver.ExplicitlyUnsupportedResolverFeature
import org.apache.spark.sql.connector.catalog.InMemoryCatalog
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{IntegerType, LongType}
+import org.apache.spark.sql.types.LongType
/**
* Tests for SET PATH command and session path management.
@@ -991,10 +992,11 @@ class SetPathSuite extends SharedSparkSession {
// Setup (`CREATE TEMPORARY FUNCTION`, `SET PATH`) and execution (Dataset
collect via
// checkAnswer, which inserts a `DeserializeToObject` node the single-pass
analyzer
// does not yet support) are run under the fixed-point analyzer; only the
actual
- // count(*) analysis is run under the single-pass analyzer, and we assert
against the
- // analyzed plan's output schema. The builtin count returns BIGINT
(rewrite applied);
- // the temp count(INT) returns INT (rewrite suppressed and the star
expansion routes
- // through the temp), so the schema's first-field dataType tells us which
branch fired.
+ // count(*) analysis is run under the single-pass analyzer. When the
shortcut fires the
+ // analyzed output is the BIGINT builtin count; when it is suppressed the
star expands
+ // and resolves through the temp SQL `count`, which the single-pass
analyzer does not
+ // support and reports as an `ExplicitlyUnsupportedResolverFeature`. Which
of the two
+ // outcomes occurs tells us whether the rewrite was applied.
withPathEnabled {
sql("CREATE TEMPORARY FUNCTION count(x INT) RETURNS INT RETURN x + 100")
try {
@@ -1011,12 +1013,15 @@ class SetPathSuite extends SharedSparkSession {
sql("SET PATH = system.session, system.builtin")
- // PATH session-first: the gate reports true, the rewrite is
suppressed, the star
- // expands against `a`, and the temp count(INT) wins; analyzed output
is INT.
+ // PATH session-first: the gate reports true, the rewrite is
suppressed, and the
+ // star expands against `a` and resolves through the temp SQL `count`.
The
+ // single-pass analyzer does not support SQL functions, so it signals
the fallback
+ // via `ExplicitlyUnsupportedResolverFeature` -- reaching this branch
confirms the
+ // rewrite was suppressed (otherwise the builtin count would have
resolved cleanly).
withSQLConf(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key ->
"true") {
- val tpe =
spark.sql(countStarSql).queryExecution.analyzed.schema.head.dataType
- assert(tpe == IntegerType,
- s"Expected INT (temp count; rewrite suppressed); got: $tpe")
+ intercept[ExplicitlyUnsupportedResolverFeature] {
+ spark.sql(countStarSql).queryExecution.analyzed
+ }
}
} finally {
sql("SET PATH = DEFAULT_PATH")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala
index 5a538b469441..6c61c7797a6c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala
@@ -394,6 +394,67 @@ class ResolverGuardSuite extends ResolverGuardSuiteBase {
}
}
+ test("Session-qualified function is rejected") {
+ checkResolverGuard(
+ "SELECT session.some_func(1)",
+ unsupportedReason = Some("session-qualified user-defined function")
+ )
+ }
+
+ test("Generator functions are unsupported") {
+ Seq(
+ "explode(array(1, 2, 3))",
+ "posexplode(array(1, 2, 3))",
+ "inline(array(struct(1, 'a')))"
+ ).foreach { call =>
+ val name = call.takeWhile(_ != '(')
+ checkResolverGuard(s"SELECT $call", unsupportedReason =
Some(s"unsupported function $name"))
+ }
+ }
+
+ test("Higher-order functions are unsupported") {
+ checkResolverGuard(
+ "SELECT transform(array(1, 2), x -> x + 1)",
+ unsupportedReason = Some("unsupported function transform")
+ )
+ checkResolverGuard(
+ "SELECT filter(array(1, 2), x -> x > 1)",
+ unsupportedReason = Some("unsupported function filter")
+ )
+ checkResolverGuard(
+ "SELECT aggregate(array(1, 2), 0, (acc, x) -> acc + x)",
+ unsupportedReason = Some("unsupported function aggregate")
+ )
+ }
+
+ test("Star target with unsupported metadata attribute name") {
+ withTable("t") {
+ sql("CREATE TABLE t (col INT) USING parquet")
+ checkResolverGuard(
+ "SELECT _metadata.* FROM t",
+ unsupportedReason = Some("unsupported attribute name '_metadata'")
+ )
+ }
+ }
+
+ test("prioritizeOrdinalResolutionInSort disabled is unsupported") {
+ withSQLConf("spark.sql.prioritizeOrdinalResolutionInSort.enabled" ->
"false") {
+ checkResolverGuard(
+ "SELECT 1",
+ unsupportedReason = Some("configuration:
prioritizeOrdinalResolutionInSortDisabled")
+ )
+ }
+ }
+
+ test("persistentCatalogFirst is unsupported") {
+ withSQLConf("spark.sql.legacy.persistentCatalogFirst" -> "true") {
+ checkResolverGuard(
+ "SELECT 1",
+ unsupportedReason = Some("configuration: persistentCatalogFirst")
+ )
+ }
+ }
+
}
trait ResolverGuardSuiteBase extends SharedSparkSession {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationTypePrecedenceSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationTypePrecedenceSuite.scala
index 05d0b40a0f02..605e838b8c5a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationTypePrecedenceSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationTypePrecedenceSuite.scala
@@ -53,6 +53,22 @@ class CollationTypePrecedenceSuite extends
SharedSparkSession {
assert(DataType.equalsIgnoreNullability(querySchema, expectedSchema))
}
+ private def testFixedPointAndSinglePass(testName: String)(testBody: =>
Unit): Unit = {
+ test(s"$testName (fixed-point analyzer)") {
+
withSQLConf(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key ->
"false") {
+ testBody
+ }
+ }
+
+ test(s"$testName (single-pass analyzer)") {
+ withSQLConf(
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED_TENTATIVELY.key ->
"true",
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key ->
"false") {
+ testBody
+ }
+ }
+ }
+
test("explicit collation propagates up") {
checkAnswer(
sql(s"SELECT COLLATION('a' collate unicode)"),
@@ -228,7 +244,7 @@ class CollationTypePrecedenceSuite extends
SharedSparkSession {
)
}
- test("in subquery expression") {
+ testFixedPointAndSinglePass("in subquery expression") {
val tableName = "subquery_tbl"
withTable(tableName) {
sql(s"""
@@ -284,7 +300,7 @@ class CollationTypePrecedenceSuite extends
SharedSparkSession {
}
}
- test("scalar subquery") {
+ testFixedPointAndSinglePass("scalar subquery") {
val tableName = "scalar_subquery_tbl"
withTable(tableName) {
sql(s"""
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/V1WriteFallbackSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/V1WriteFallbackSuite.scala
index 3e48c5222e6f..16a94bef2973 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/V1WriteFallbackSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/V1WriteFallbackSuite.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.datasources.DataSourceUtils
import org.apache.spark.sql.execution.datasources.v2.{AppendDataExecV1,
OverwriteByExpressionExecV1}
import org.apache.spark.sql.functions.lit
-import org.apache.spark.sql.internal.SQLConf.{OPTIMIZER_MAX_ITERATIONS,
V2_SESSION_CATALOG_IMPLEMENTATION}
+import
org.apache.spark.sql.internal.SQLConf.{ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER,
OPTIMIZER_MAX_ITERATIONS, V2_SESSION_CATALOG_IMPLEMENTATION}
import org.apache.spark.sql.sources._
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.StructType
@@ -144,6 +144,8 @@ class V1WriteFallbackSuite extends SharedSparkSession with
BeforeAndAfter {
.master("local[1]")
.withExtensions(_.injectPostHocResolutionRule(_ => OnlyOnceRule))
.withExtensions(_.injectOptimizerRule(_ => OnlyOnceOptimizerRule))
+ // Dual-running Analyzers would lead to rule being invoked twice
+ .config(ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key, "false")
.config(OPTIMIZER_MAX_ITERATIONS.key, "1")
.config(V2_SESSION_CATALOG_IMPLEMENTATION.key,
classOf[V1FallbackTableCatalog].getName)
.getOrCreate()
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala
index 44cfd339e15c..3331dff50e35 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala
@@ -99,6 +99,7 @@ class HiveSessionStateBuilder(
override val singlePassPostHocResolutionRules: Seq[Rule[LogicalPlan]] =
DetectAmbiguousSelfJoin +:
ApplyCharTypePadding +:
+ new DetermineTableStats(session) +:
singlePassCustomPostHocResolutionRules
override val singlePassExtendedResolutionChecks: Seq[LogicalPlan => Unit]
= {
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala
index e0be6e42451c..95302e89da28 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala
@@ -32,6 +32,7 @@ import org.apache.spark.sql.{Column, DataFrame, QueryTest,
Row, TypedColumn}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.execution._
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.pipelines.graph.{DataflowGraph,
PipelineUpdateContextImpl, SqlGraphRegistrationContext}
import org.apache.spark.sql.pipelines.utils.PipelineTest.cleanupMetastore
@@ -113,6 +114,8 @@ abstract class PipelineTest
protected override def beforeEach(): Unit = {
super.beforeEach()
+ spark.sessionState.conf.setConf(
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER, false)
cleanupMetastore(spark)
(catalogInPipelineSpec, databaseInPipelineSpec) match {
case (Some(catalog), Some(schema)) =>
@@ -124,6 +127,7 @@ abstract class PipelineTest
protected override def afterEach(): Unit = {
cleanupMetastore(spark)
+
spark.sessionState.conf.unsetConf(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER)
super.afterEach()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
