You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2024/12/18 08:05:25 UTC
(spark) branch master updated: [SPARK-50472][SQL] Introduce initial implementation of the single-pass Analyzer
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 76672cbf366c [SPARK-50472][SQL] Introduce initial implementation of the single-pass Analyzer
76672cbf366c is described below
commit 76672cbf366ca4f11ae40da35229d670b205c780
Author: Vladimir Golubev <vl...@databricks.com>
AuthorDate: Wed Dec 18 16:05:01 2024 +0800
[SPARK-50472][SQL] Introduce initial implementation of the single-pass Analyzer
### What changes were proposed in this pull request?
This is the initial PR for the single-pass Analyzer project.
Coverage:
- Scans of csv/json/parquet/orc/hive tables and files
- `VALUES`
- `WHERE`
- `LIMIT`
- Trivial star expansion
- Literals
- Aliases
- Binary arithmetic
- Binary logic
Main components:
`HybridAnalyzer`:
The analysis is now routed between the fixed-point and single-pass by the `HybridAnalyzer`. There are 3 modes:
- Fixed-point. In that case `HybridAnalyzer` just calls `executeAndTrack`. This is the default one.
- Dual. This one is activated by the `spark.sql.analyzer.singlePassResolver.dualRunWithLegacy` flag and involves a filtering logic based on the `ResolverGuard`. First, the query is always analyzed by the fixed-point `Analyzer`, and then, if it's supported by the single-pass, it's analyzed by the `Resolver`.
- Single-pass. In that case `HybridAnalyzer` calls the single-pass `Resolver` without any filtering or routing. This is useful for debugging.
`ResolverGuard`:
This component allows `HybridAnalyzer` to figure out whether a SQL query/DataFrame program is supported by the single-pass `Resolver`. Here we do a simple pass over the operator/expression tree with a short-circuited boolean logic.
`Resolver`:
This is the main entry point for the single-pass analysis. All the operators are resolved here. If the new operator is introduced into the Catalyst, it should be handled in the `resolve(...)` pattern match. The operator is either resolved here, or the resolution is delegated to the extensions.
`ExpressionResolver`:
Similar to the `Resolver`, this is the main entry point for the expression analysis. The resolution either happens here, or is delegated to specific resolvers (e.g. `BinaryArithmeticResolver`, `TimeZoneAwareExpressionResolver`). The `Resolver` calls the `ExpressionResolver` to resolve expressions nested in operators.
`MetadataResolver`:
The `MetadataResolver` is used by the `Resolver` to look up relation metadata and store it in a map by `RelationId`. There are two modes:
- `AnalyzerBridgeState` from the fixed-point `Analyzer` is available. In that case we just reuse the existing relation metadata from the fixed-point run. This is done to avoid latency/consistency issues.
- Otherwise, we collect the metadata using `RelationResolution`.
`ResolverExtension`:
This is a trait that allows extending the `Resolver`. Its `resolveOperator` must deterministically match the unresolved plan node - a conflict between the configured extensions is disallowed.
`NameScope`:
This component controls the visibility of names (tables, columns, aliases), star expansion for tables and structs and operator output handling. A new `NameScope` is created when a new scope is introduced (`Project`, `Aggregate`). We add operator output to the `NameScope` to later look up unresolved attributes and aliases. The `NameScope`s are stacked together to further support correlated names.
`DataSourceResolver`/`DataSourceWithHiveResolver`:
Those extensions allow us to resolve relation nodes (e.g. `UnresolvedCatalogRelation`, `LogicalRelation`, `HiveTableRelation`) defined outside of `catalyst` package to avoid depending on those packages (that would require a cyclic import).
`ResolutionValidator`/`ExpressionResolutionValidator`:
These components work together on the resolved tree and assert invariants that should be true no matter which SQL query/DataFrame program was passed to the single-pass analyzer. For example, we validate that expressions used in project lists actually exist in the lower operator outputs (using `AttributeScopeStack`). Every `resolve*` method in `Resolver`/`ExpressionResolver` must have its `validate*` counterpart in the `ResolutionValidator`/`ExpressionResolutionValidator`.
`TracksResolvedNodes`:
This is another validation mechanism which asserts that the analysis is actually single-pass - we store a reference to any resolved node in a set, and prior to the resolution check that a reference to the passed node hasn't been previously analyzed.
Thanks mihailoale-db and mihailotim-db for working on that!
### Why are the changes needed?
We want to build a new single-pass Analyzer to replace the fixed-point one.
SPIP Jira: https://issues.apache.org/jira/browse/SPARK-49834
SPIP: https://docs.google.com/document/d/1dWxvrJV-0joGdLtWbvJ0uNyTocDMJ90rPRNWa4T56Og
### Does this PR introduce _any_ user-facing change?
No, the single-pass Analyzer is disabled by default (under the `spark.sql.analyzer.singlePassResolver.enabled` and `spark.sql.analyzer.singlePassResolver.dualRunWithLegacy` flags).
### How was this patch tested?
- New test suites
- Ran some suites with `spark.sql.analyzer.singlePassResolver.dualRunWithLegacy` enabled
### Was this patch authored or co-authored using generative AI tooling?
Copilot.
Closes #49029 from vladimirg-db/vladimirg-db/single-pass-analyzer/initial-state.
Authored-by: Vladimir Golubev <vl...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../src/main/resources/error/error-conditions.json | 45 ++
.../spark/sql/catalyst/analysis/Analyzer.scala | 92 ++-
.../catalyst/analysis/resolver/AliasResolver.scala | 142 +++++
.../analysis/resolver/AnalyzerBridgeState.scala | 40 ++
.../analysis/resolver/AttributeScopeStack.scala | 89 +++
.../resolver/BinaryArithmeticResolver.scala | 200 +++++++
.../resolver/BridgedRelationsProvider.scala | 50 ++
.../resolver/ConditionalExpressionResolver.scala | 65 ++
.../resolver/CreateNamedStructResolver.scala | 61 ++
.../resolver/DelegatesResolutionToExtensions.scala | 70 +++
.../ExplicitlyUnsupportedResolverFeature.scala | 57 ++
.../resolver/ExpressionResolutionValidator.scala | 367 ++++++++++++
.../analysis/resolver/ExpressionResolver.scala | 340 +++++++++++
.../analysis/resolver/FunctionResolver.scala | 91 +++
.../analysis/resolver/HybridAnalyzer.scala | 218 +++++++
.../catalyst/analysis/resolver/IdentifierMap.scala | 37 ++
.../analysis/resolver/KeyTransformingMap.scala | 46 ++
.../resolver/LimitExpressionResolver.scala | 114 ++++
.../analysis/resolver/MetadataResolver.scala | 133 +++++
.../sql/catalyst/analysis/resolver/NameScope.scala | 393 ++++++++++++
.../catalyst/analysis/resolver/NameTarget.scala | 81 +++
.../catalyst/analysis/resolver/PlanLogger.scala | 58 ++
.../analysis/resolver/PredicateResolver.scala | 112 ++++
.../resolver/ProducesUnresolvedSubtree.scala | 52 ++
.../catalyst/analysis/resolver/RelationId.scala | 31 +
.../resolver/RelationMetadataProvider.scala | 86 +++
.../analysis/resolver/ResolutionValidator.scala | 153 +++++
.../sql/catalyst/analysis/resolver/Resolver.scala | 364 ++++++++++++
.../analysis/resolver/ResolverExtension.scala | 43 ++
.../catalyst/analysis/resolver/ResolverGuard.scala | 283 +++++++++
.../resolver/ResolvesExpressionChildren.scala | 33 ++
.../resolver/ResolvesOperatorChildren.scala | 36 ++
.../analysis/resolver/TimeAddResolver.scala | 72 +++
.../resolver/TimezoneAwareExpressionResolver.scala | 73 +++
.../analysis/resolver/TracksResolvedNodes.scala | 51 ++
.../analysis/resolver/TreeNodeResolver.scala | 30 +
.../analysis/resolver/TypeCoercionResolver.scala | 50 ++
.../analysis/resolver/UnaryMinusResolver.scala | 60 ++
.../spark/sql/catalyst/analysis/unresolved.scala | 19 +-
.../sql/catalyst/catalog/TempVariableManager.scala | 4 +
.../spark/sql/errors/QueryCompilationErrors.scala | 77 +++
.../org/apache/spark/sql/internal/SQLConf.scala | 70 +++
.../resolver/LimitExpressionResolverSuite.scala | 94 +++
.../resolver/ResolutionValidatorSuite.scala | 265 +++++++++
.../TimezoneAwareExpressionResolverSuite.scala | 58 ++
.../resolver/TypeCoercionResolverSuite.scala | 68 +++
.../execution/datasources/DataSourceResolver.scala | 67 +++
.../sql/execution/datasources/FileResolver.scala | 55 ++
.../spark/sql/execution/datasources/rules.scala | 13 +-
.../sql/internal/BaseSessionStateBuilder.scala | 10 +
...ExplicitlyUnsupportedResolverFeatureSuite.scala | 91 +++
.../analysis/resolver/HybridAnalyzerSuite.scala | 404 +++++++++++++
.../analysis/resolver/MetadataResolverSuite.scala | 277 +++++++++
.../sql/analysis/resolver/NameScopeSuite.scala | 659 +++++++++++++++++++++
.../sql/analysis/resolver/ResolverGuardSuite.scala | 207 +++++++
.../sql/analysis/resolver/ResolverSuite.scala | 157 +++++
.../resolver/TracksResolvedNodesSuite.scala | 133 +++++
.../datasources/DataSourceResolverSuite.scala | 119 ++++
.../execution/datasources/FileResolverSuite.scala | 97 +++
.../sql/hive/DataSourceWithHiveResolver.scala | 57 ++
.../spark/sql/hive/HiveSessionStateBuilder.scala | 9 +
.../sql/hive/DataSourceWithHiveResolverSuite.scala | 107 ++++
.../sql/hive/PartitionedTablePerfStatsSuite.scala | 3 +
63 files changed, 7425 insertions(+), 13 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index fca6a0d451de..55265fa6dfc9 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -70,6 +70,12 @@
],
"sqlState" : "42000"
},
+ "AMBIGUOUS_RESOLVER_EXTENSION" : {
+ "message" : [
+ "The single-pass analyzer cannot process this query or command because the extension choice for <operator> is ambiguous: <extensions>."
+ ],
+ "sqlState" : "XX000"
+ },
"ARITHMETIC_OVERFLOW" : {
"message" : [
"<message>.<alternative> If necessary set <config> to \"false\" to bypass this error."
@@ -1659,6 +1665,39 @@
],
"sqlState" : "22000"
},
+ "HYBRID_ANALYZER_EXCEPTION" : {
+ "message" : [
+ "An failure occurred when attempting to resolve a query or command with both the legacy fixed-point analyzer as well as the single-pass resolver."
+ ],
+ "subClass" : {
+ "FIXED_POINT_FAILED_SINGLE_PASS_SUCCEEDED" : {
+ "message" : [
+ "Fixed-point resolution failed, but single-pass resolution succeeded.",
+ "Single-pass analyzer output:",
+ "<singlePassOutput>"
+ ]
+ },
+ "LOGICAL_PLAN_COMPARISON_MISMATCH" : {
+ "message" : [
+ "Outputs of fixed-point and single-pass analyzers do not match.",
+ "Fixed-point analyzer output:",
+ "<fixedPointOutput>",
+ "Single-pass analyzer output:",
+ "<singlePassOutput>"
+ ]
+ },
+ "OUTPUT_SCHEMA_COMPARISON_MISMATCH" : {
+ "message" : [
+ "Output schemas of fixed-point and single-pass analyzers do not match.",
+ "Fixed-point analyzer output schema:",
+ "<fixedPointOutputSchema>",
+ "Single-pass analyzer output schema:",
+ "<singlePassOutputSchema>"
+ ]
+ }
+ },
+ "sqlState" : "XX000"
+ },
"IDENTIFIER_TOO_MANY_NAME_PARTS" : {
"message" : [
"<identifier> is not a valid identifier as it has more than 2 name parts."
@@ -5659,6 +5698,12 @@
},
"sqlState" : "0A000"
},
+ "UNSUPPORTED_SINGLE_PASS_ANALYZER_FEATURE" : {
+ "message" : [
+ "The single-pass analyzer cannot process this query or command because it does not yet support <feature>."
+ ],
+ "sqlState" : "0A000"
+ },
"UNSUPPORTED_STREAMING_OPERATOR_WITHOUT_WATERMARK" : {
"message" : [
"<outputMode> output mode not supported for <statefulOperator> on streaming DataFrames/DataSets without watermark."
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 1c31c45d2729..7c97fccade01 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
@@ -28,6 +28,13 @@ import scala.util.{Failure, Random, Success, Try}
import org.apache.spark.{SparkException, SparkThrowable, SparkUnsupportedOperationException}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst._
+import org.apache.spark.sql.catalyst.analysis.resolver.{
+ AnalyzerBridgeState,
+ HybridAnalyzer,
+ Resolver => OperatorResolver,
+ ResolverExtension,
+ ResolverGuard
+}
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.encoders.OuterScopes
import org.apache.spark.sql.catalyst.expressions._
@@ -146,7 +153,26 @@ case class AnalysisContext(
// lookup a temporary function. And export to the view metadata.
referredTempFunctionNames: mutable.Set[String] = mutable.Set.empty,
referredTempVariableNames: Seq[Seq[String]] = Seq.empty,
- outerPlan: Option[LogicalPlan] = None)
+ outerPlan: Option[LogicalPlan] = None,
+
+ /**
+ * This is a bridge state between this fixed-point [[Analyzer]] and a single-pass [[Resolver]].
+ * It's managed ([[setSinglePassResolverBridgeState]] method) by the [[HybridAnalyzer]] - the
+ * goal is to preserve it correctly between the fixed-point and single-pass runs.
+ * [[AnalysisContext.reset]] simply propagates it to prevent it from being reset in
+ * [[Analyzer.execute]]. Normally it's always [[None]], unless
+ * [[ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER]] is set to [[true]].
+ *
+ * See [[AnalyzerBridgeState]] and [[HybridAnalyzer]] for more info.
+ */
+ private var singlePassResolverBridgeState: Option[AnalyzerBridgeState] = None) {
+
+ def setSinglePassResolverBridgeState(bridgeState: Option[AnalyzerBridgeState]): Unit =
+ singlePassResolverBridgeState = bridgeState
+
+ def getSinglePassResolverBridgeState: Option[AnalyzerBridgeState] =
+ singlePassResolverBridgeState
+}
object AnalysisContext {
private val value = new ThreadLocal[AnalysisContext]() {
@@ -154,7 +180,16 @@ object AnalysisContext {
}
def get: AnalysisContext = value.get()
- def reset(): Unit = value.remove()
+
+ def reset(): Unit = {
+ // We need to preserve the single-pass resolver bridge state here, since it's managed by the
+ // [[HybridAnalyzer]] (set or reset to `None`) to avoid it being reset in [[execute]].
+ // It acts as a bridge between the single-pass and fixed-point analyzers in the absence of any
+ // other explicit state.
+ val prevSinglePassResolverBridgeState = value.get.getSinglePassResolverBridgeState
+ value.remove()
+ value.get.setSinglePassResolverBridgeState(prevSinglePassResolverBridgeState)
+ }
private def set(context: AnalysisContext): Unit = value.set(context)
@@ -219,9 +254,15 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
def executeAndCheck(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {
if (plan.analyzed) return plan
AnalysisHelper.markInAnalyzer {
- val analyzed = executeAndTrack(plan, tracker)
- checkAnalysis(analyzed)
- analyzed
+ new HybridAnalyzer(
+ this,
+ new ResolverGuard(catalogManager),
+ new OperatorResolver(
+ catalogManager,
+ singlePassResolverExtensions,
+ singlePassMetadataResolverExtensions
+ )
+ ).apply(plan, tracker)
}
}
@@ -245,6 +286,20 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
errorOnExceed = true,
maxIterationsSetting = SQLConf.ANALYZER_MAX_ITERATIONS.key)
+ /**
+ * Extensions for the single-pass analyzer.
+ *
+ * See [[ResolverExtension]] for more info.
+ */
+ val singlePassResolverExtensions: Seq[ResolverExtension] = Nil
+
+ /**
+ * Extensions used for early resolution of the single-pass analyzer.
+ *
+ * See [[ResolverExtension]] for more info.
+ */
+ val singlePassMetadataResolverExtensions: Seq[ResolverExtension] = Nil
+
/**
* Override to provide additional rules for the "Resolution" batch.
*/
@@ -1018,7 +1073,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
case i @ InsertIntoStatement(table, _, _, _, _, _, _) =>
val relation = table match {
case u: UnresolvedRelation if !u.isStreaming =>
- relationResolution.resolveRelation(u).getOrElse(u)
+ resolveRelation(u).getOrElse(u)
case other => other
}
@@ -1035,7 +1090,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
case write: V2WriteCommand =>
write.table match {
case u: UnresolvedRelation if !u.isStreaming =>
- relationResolution.resolveRelation(u).map(unwrapRelationPlan).map {
+ resolveRelation(u).map(unwrapRelationPlan).map {
case v: View => throw QueryCompilationErrors.writeIntoViewNotAllowedError(
v.desc.identifier, write)
case r: DataSourceV2Relation => write.withNewTable(r)
@@ -1050,12 +1105,12 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
case u: UnresolvedRelation =>
- relationResolution.resolveRelation(u).map(resolveViews).getOrElse(u)
+ resolveRelation(u).map(resolveViews).getOrElse(u)
case r @ RelationTimeTravel(u: UnresolvedRelation, timestamp, version)
if timestamp.forall(ts => ts.resolved && !SubqueryExpression.hasSubquery(ts)) =>
val timeTravelSpec = TimeTravelSpec.create(timestamp, version, conf.sessionLocalTimeZone)
- relationResolution.resolveRelation(u, timeTravelSpec).getOrElse(r)
+ resolveRelation(u, timeTravelSpec).getOrElse(r)
case u @ UnresolvedTable(identifier, cmd, suggestAlternative) =>
lookupTableOrView(identifier).map {
@@ -1119,6 +1174,25 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
}
+
+ def resolveRelation(
+ unresolvedRelation: UnresolvedRelation,
+ timeTravelSpec: Option[TimeTravelSpec] = None): Option[LogicalPlan] = {
+ relationResolution
+ .resolveRelation(
+ unresolvedRelation,
+ timeTravelSpec
+ )
+ .map { relation =>
+ // We put the synchronously resolved relation into the [[AnalyzerBridgeState]] for
+ // it to be later reused by the single-pass [[Resolver]] to avoid resolving the relation
+ // metadata twice.
+ AnalysisContext.get.getSinglePassResolverBridgeState.map { bridgeState =>
+ bridgeState.relationsWithResolvedMetadata.put(unresolvedRelation, relation)
+ }
+ relation
+ }
+ }
}
/** Handle INSERT INTO for DSv2 */
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AliasResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AliasResolver.scala
new file mode 100644
index 000000000000..7b652437dbd8
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AliasResolver.scala
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.{AliasResolution, UnresolvedAlias}
+import org.apache.spark.sql.catalyst.expressions.{
+ Alias,
+ Cast,
+ CreateNamedStruct,
+ Expression,
+ NamedExpression
+}
+
+/**
+ * Resolver class that resolves unresolved aliases and handles user-specified aliases.
+ */
+class AliasResolver(expressionResolver: ExpressionResolver, scopes: NameScopeStack)
+ extends TreeNodeResolver[UnresolvedAlias, Expression]
+ with ResolvesExpressionChildren {
+
+ /**
+ * Resolves [[UnresolvedAlias]] by handling two specific cases:
+ * - Alias(CreateNamedStruct(...)) - instead of calling [[CreateNamedStructResolver]] which will
+ * clean up its inner aliases, we manually resolve [[CreateNamedStruct]]'s children, because we
+ * need to preserve inner aliases until after the alias name is computed. This is a hack because
+ * fixed-point analyzer computes [[Alias]] name before removing inner aliases.
+ * - Alias(...) - recursively call [[ExpressionResolver]] to resolve the child expression.
+ *
+ * After the children are resolved, call [[AliasResolution]] to compute the alias name. Finally,
+ * clean up inner aliases from [[CreateNamedStruct]].
+ */
+ override def resolve(unresolvedAlias: UnresolvedAlias): NamedExpression = {
+ val aliasWithResolvedChildren = withResolvedChildren(
+ unresolvedAlias, {
+ case createNamedStruct: CreateNamedStruct =>
+ withResolvedChildren(createNamedStruct, expressionResolver.resolve)
+ case other => expressionResolver.resolve(other)
+ }
+ )
+
+ val resolvedAlias =
+ AliasResolution.resolve(aliasWithResolvedChildren).asInstanceOf[NamedExpression]
+
+ scopes.top.addAlias(resolvedAlias.name)
+ AliasResolver.cleanupAliases(resolvedAlias)
+ }
+
+ /**
+ * Handle already resolved [[Alias]] nodes, i.e. user-specified aliases. We disallow stacking
+ * of [[Alias]] nodes by collapsing them so that only the top node remains.
+ *
+ * For an example query like:
+ *
+ * {{{ SELECT 1 AS a }}}
+ *
+ * parsed plan will be:
+ *
+ * Project [Alias(1, a)]
+ * +- OneRowRelation
+ *
+ */
+ def handleResolvedAlias(alias: Alias): Alias = {
+ val aliasWithResolvedChildren = withResolvedChildren(alias, expressionResolver.resolve)
+ scopes.top.addAlias(aliasWithResolvedChildren.name)
+ AliasResolver.collapseAlias(aliasWithResolvedChildren)
+ }
+}
+
+object AliasResolver {
+
+ /**
+ * For a query like:
+ *
+ * {{{ SELECT STRUCT(1 AS a, 2 AS b) AS st }}}
+ *
+ * After resolving [[CreateNamedStruct]] the plan will be:
+ * CreateNamedStruct(Seq("a", Alias(1, "a"), "b", Alias(2, "b")))
+ *
+ * For a query like:
+ *
+ * {{{ df.select($"col1".cast("int").cast("double")) }}}
+ *
+ * After resolving top-most [[Alias]] the plan will be:
+ * Alias(Cast(Alias(Cast(col1, int), col1)), double), col1)
+ *
+ * Both examples contain inner aliases that are not expected in the analyzed logical plan,
+ * therefore need to be removed. However, in both examples inner aliases are necessary in order
+ * for the outer alias to compute its name. To achieve this, we delay removal of inner aliases
+ * until after the outer alias name is computed.
+ *
+ * For cases where there are no dependencies on inner alias, inner alias should be removed by the
+ * resolver that produces it.
+ */
+ private def cleanupAliases(namedExpression: NamedExpression): NamedExpression =
+ namedExpression
+ .withNewChildren(namedExpression.children.map {
+ case cast @ Cast(alias: Alias, _, _, _) =>
+ cast.copy(child = alias.child)
+ case createNamedStruct: CreateNamedStruct =>
+ CreateNamedStructResolver.cleanupAliases(createNamedStruct)
+ case other => other
+ })
+ .asInstanceOf[NamedExpression]
+
+ /**
+ * If an [[Alias]] node appears on top of another [[Alias]], remove the bottom one. Here we don't
+ * handle a case where a node of different type appears between two [[Alias]] nodes: in this
+ * case, removal of inner alias (if it is unnecessary) should be handled by respective node's
+ * resolver, in order to preserve the bottom-up contract.
+ */
+ private def collapseAlias(alias: Alias): Alias =
+ alias.child match {
+ case innerAlias: Alias =>
+ val metadata = if (alias.metadata.isEmpty) {
+ None
+ } else {
+ Some(alias.metadata)
+ }
+ alias.copy(child = innerAlias.child)(
+ exprId = alias.exprId,
+ qualifier = alias.qualifier,
+ explicitMetadata = metadata,
+ nonInheritableMetadataKeys = alias.nonInheritableMetadataKeys
+ )
+ case _ => alias
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AnalyzerBridgeState.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AnalyzerBridgeState.scala
new file mode 100644
index 000000000000..d3e93c82dfa2
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AnalyzerBridgeState.scala
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import java.util.HashMap
+
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+
+/**
+ * The [[AnalyzerBridgeState]] is a state passed from legacy [[Analyzer]] to the single-pass
+ * [[Resolver]].
+ *
+ * @param relationsWithResolvedMetadata A map from [[UnresolvedRelation]] to the relations with
+ * resolved metadata. It allows us to reuse the relation metadata and avoid duplicate
+ * catalog/table lookups in dual-run mode (when
+ * [[ANALYZER_SINGLE_PASS_RESOLVER_RELATION_BRIDGING_ENABLED]] is true).
+ */
+case class AnalyzerBridgeState(
+ relationsWithResolvedMetadata: AnalyzerBridgeState.RelationsWithResolvedMetadata =
+ new AnalyzerBridgeState.RelationsWithResolvedMetadata)
+
+object AnalyzerBridgeState {
+ type RelationsWithResolvedMetadata = HashMap[UnresolvedRelation, LogicalPlan]
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AttributeScopeStack.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AttributeScopeStack.scala
new file mode 100644
index 000000000000..6f9d6defd2ed
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AttributeScopeStack.scala
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import java.util.ArrayDeque
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet}
+
+/**
+ * The [[AttributeScopeStack]] is used to validate that the attribute which was encountered by the
+ * [[ExpressionResolutionValidator]] is in the current operator's visibility scope. We use
+ * [[AttributeSet]] as scope implementation here to check the equality of attributes based on their
+ * expression IDs.
+ *
+ * E.g. for the following SQL query:
+ * {{{
+ * SELECT a, a, a + col2 FROM (SELECT col1 as a, col2 FROM VALUES (1, 2));
+ * }}}
+ *
+ * Having the following logical plan:
+ * {{{
+ * Project [a#2, a#2, (a#2 + col2#1) AS (a + col2)#3]
+ * +- SubqueryAlias __auto_generated_subquery_name
+ * +- Project [col1#0 AS a#2, col2#1]
+ * +- LocalRelation [col1#0, col2#1]
+ * }}}
+ *
+ * The [[LocalRelation]] outputs attributes with IDs #0 and #1, which can be referenced by the lower
+ * [[Project]]. This [[Project]] produces a new attribute ID #2 for an alias and retains the old
+ * ID #1 for col2. The upper [[Project]] references `a` twice using the same ID #2 and produces a
+ * new ID #3 for an alias of `a + col2`.
+ */
+class AttributeScopeStack {
+ private val stack = new ArrayDeque[AttributeSet]
+ push()
+
+ /**
+ * Get the relevant attribute scope in the context of the current operator.
+ */
+ def top: AttributeSet = {
+ stack.peek()
+ }
+
+ /**
+ * Overwrite current relevant scope with a sequence of attributes which is an output of some
+ * operator. `attributes` can have duplicate IDs if the output of the operator contains multiple
+ * occurrences of the same attribute.
+ */
+ def overwriteTop(attributes: Seq[Attribute]): Unit = {
+ stack.pop()
+ stack.push(AttributeSet(attributes))
+ }
+
+ /**
+ * Execute `body` in the context of a fresh attribute scope. Used by [[Project]] and [[Aggregate]]
+ * validation code since those operators introduce a new scope with fresh expression IDs.
+ */
+ def withNewScope[R](body: => R): Unit = {
+ push()
+ try {
+ body
+ } finally {
+ pop()
+ }
+ }
+
+ private def push(): Unit = {
+ stack.push(AttributeSet(Seq.empty))
+ }
+
+ private def pop(): Unit = {
+ stack.pop()
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/BinaryArithmeticResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/BinaryArithmeticResolver.scala
new file mode 100644
index 000000000000..7d9c6752094d
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/BinaryArithmeticResolver.scala
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.{
+ AnsiStringPromotionTypeCoercion,
+ AnsiTypeCoercion,
+ BinaryArithmeticWithDatetimeResolver,
+ DecimalPrecisionTypeCoercion,
+ DivisionTypeCoercion,
+ IntegralDivisionTypeCoercion,
+ StringPromotionTypeCoercion,
+ TypeCoercion
+}
+import org.apache.spark.sql.catalyst.expressions.{
+ Add,
+ BinaryArithmetic,
+ DateAdd,
+ Divide,
+ Expression,
+ Multiply,
+ Subtract,
+ SubtractDates
+}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DateType, StringType}
+
+/**
+ * [[BinaryArithmeticResolver]] is invoked by [[ExpressionResolver]] in order to resolve
+ * [[BinaryArithmetic]] nodes. During resolution, calling [[BinaryArithmeticWithDatetimeResolver]]
+ * and applying type coercion can result in [[BinaryArithmetic]] producing some other type of node
+ * or a subtree of nodes. In such cases a downwards traversal is necessary, but not going deeper
+ * than the original expression's children, since all nodes below that point are guaranteed to be
+ * already resolved.
+ *
+ * For example, given a query:
+ *
+ * SELECT '4 11:11' - INTERVAL '4 22:12' DAY TO MINUTE
+ *
+ * [[BinaryArithmeticResolver]] is called for the following expression:
+ *
+ * Subtract(
+ * Literal('4 11:11', StringType),
+ * Literal(Interval('4 22:12' DAY TO MINUTE), DayTimeIntervalType(0,2))
+ * )
+ *
+ * After calling [[BinaryArithmeticWithDatetimeResolver]] and applying type coercion,
+ * the expression is transformed into:
+ *
+ * Cast(
+ * DatetimeSub(
+ * TimeAdd(
+ * Literal('4 11:11', StringType),
+ * UnaryMinus(
+ * Literal(Interval('4 22:12' DAY TO MINUTE), DayTimeIntervalType(0,2))
+ * )
+ * )
+ * )
+ * )
+ *
+ * A single [[Subtract]] node is replaced with a subtree of nodes. In order to resolve this subtree
+ * we need to invoke [[ExpressionResolver]] recursively on the top-most node's children. The
+ * top-most node itself is not resolved recursively in order to avoid recursive calls to
+ * [[BinaryArithmeticResolver]] and other sub-resolvers. To prevent a case where we resolve the
+ * same node twice, we need to mark nodes that will act as a limit for the downwards traversal by
+ * applying a [[ExpressionResolver.SINGLE_PASS_SUBTREE_BOUNDARY]] tag to them. These children
+ * along with all the nodes below them are guaranteed to be resolved at this point. When
+ * [[ExpressionResolver]] reaches one of the tagged nodes, it returns identity rather than
+ * resolving it. Finally, after resolving the subtree, we need to resolve the top-most node itself,
+ * which in this case means applying a timezone, if necessary.
+ */
+class BinaryArithmeticResolver(
+ expressionResolver: ExpressionResolver,
+ timezoneAwareExpressionResolver: TimezoneAwareExpressionResolver)
+ extends TreeNodeResolver[BinaryArithmetic, Expression]
+ with ProducesUnresolvedSubtree {
+
+ private val shouldTrackResolvedNodes =
+ conf.getConf(SQLConf.ANALYZER_SINGLE_PASS_TRACK_RESOLVED_NODES_ENABLED)
+
+ private val typeCoercionRules: Seq[Expression => Expression] =
+ if (conf.ansiEnabled) {
+ BinaryArithmeticResolver.ANSI_TYPE_COERCION_RULES
+ } else {
+ BinaryArithmeticResolver.TYPE_COERCION_RULES
+ }
+ private val typeCoercionResolver: TypeCoercionResolver =
+ new TypeCoercionResolver(timezoneAwareExpressionResolver, typeCoercionRules)
+
+ override def resolve(unresolvedBinaryArithmetic: BinaryArithmetic): Expression = {
+ val binaryArithmeticWithResolvedChildren: BinaryArithmetic =
+ withResolvedChildren(unresolvedBinaryArithmetic, expressionResolver.resolve)
+ val binaryArithmeticWithResolvedSubtree: Expression =
+ withResolvedSubtree(binaryArithmeticWithResolvedChildren, expressionResolver.resolve) {
+ transformBinaryArithmeticNode(binaryArithmeticWithResolvedChildren)
+ }
+ val binaryArithmeticWithResolvedTimezone = timezoneAwareExpressionResolver.withResolvedTimezone(
+ binaryArithmeticWithResolvedSubtree,
+ conf.sessionLocalTimeZone
+ )
+ reallocateKnownNodesForTracking(binaryArithmeticWithResolvedTimezone)
+ }
+
+ /**
+ * Transform [[BinaryArithmetic]] node by calling [[BinaryArithmeticWithDatetimeResolver]] and
+ * applying type coercion. Initial node can be replaced with some other type of node or a subtree
+ * of nodes.
+ */
+ private def transformBinaryArithmeticNode(binaryArithmetic: BinaryArithmetic): Expression = {
+ val binaryArithmeticWithDateTypeReplaced: Expression =
+ replaceDateType(binaryArithmetic)
+ val binaryArithmeticWithTypeCoercion: Expression =
+ typeCoercionResolver.resolve(binaryArithmeticWithDateTypeReplaced)
+ // In case that original expression's children types are DateType and StringType, fixed-point
+ // fails to resolve the expression with a single application of
+ // [[BinaryArithmeticWithDatetimeResolver]]. Therefore, single-pass resolver needs to invoke
+ // [[BinaryArithmeticWithDatetimeResolver.resolve]], type coerce and only after that fix the
+ // date/string case. Instead of invoking [[BinaryArithmeticWithDatetimeResolver]] again, we
+ // handle the case directly.
+ (
+ binaryArithmetic.left.dataType,
+ binaryArithmetic.right.dataType
+ ) match {
+ case (_: DateType, _: StringType) =>
+ binaryArithmeticWithTypeCoercion match {
+ case add: Add => DateAdd(add.left, add.right)
+ case subtract: Subtract => SubtractDates(subtract.left, subtract.right)
+ case other => other
+ }
+ case _ => binaryArithmeticWithTypeCoercion
+ }
+ }
+
+ /**
+ * When DateType like operand is given to [[BinaryArithmetic]], apply
+ * [[BinaryArithmeticWithDatetimeResolver]] in order to replace the [[BinaryArithmetic]] with
+ * the appropriate equivalent for DateTime types.
+ */
+ private def replaceDateType(expression: Expression) = expression match {
+ case arithmetic @ (_: Add | _: Subtract | _: Multiply | _: Divide) =>
+ BinaryArithmeticWithDatetimeResolver.resolve(arithmetic)
+ case other => other
+ }
+
+ /**
+ * Since [[TracksResolvedNodes]] requires all the expressions in the tree to be unique objects,
+ * we reallocate the known nodes in [[ANALYZER_SINGLE_PASS_TRACK_RESOLVED_NODES_ENABLED]] mode,
+ * otherwise we preserve the old object to avoid unnecessary memory allocations.
+ */
+ private def reallocateKnownNodesForTracking(expression: Expression): Expression = {
+ if (shouldTrackResolvedNodes) {
+ expression match {
+ case add: Add => add.copy()
+ case subtract: Subtract => subtract.copy()
+ case multiply: Multiply => multiply.copy()
+ case divide: Divide => divide.copy()
+ case _ => expression
+ }
+ } else {
+ expression
+ }
+ }
+}
+
+object BinaryArithmeticResolver {
+ // Ordering in the list of type coercions should be in sync with the list in [[TypeCoercion]].
+ private val TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ StringPromotionTypeCoercion.apply,
+ DecimalPrecisionTypeCoercion.apply,
+ DivisionTypeCoercion.apply,
+ IntegralDivisionTypeCoercion.apply,
+ TypeCoercion.ImplicitTypeCoercion.apply,
+ TypeCoercion.DateTimeOperationsTypeCoercion.apply
+ )
+
+ // Ordering in the list of type coercions should be in sync with the list in [[AnsiTypeCoercion]].
+ private val ANSI_TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ AnsiStringPromotionTypeCoercion.apply,
+ DecimalPrecisionTypeCoercion.apply,
+ DivisionTypeCoercion.apply,
+ IntegralDivisionTypeCoercion.apply,
+ AnsiTypeCoercion.ImplicitTypeCoercion.apply,
+ AnsiTypeCoercion.AnsiDateTimeOperationsTypeCoercion.apply
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/BridgedRelationsProvider.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/BridgedRelationsProvider.scala
new file mode 100644
index 000000000000..bc7a9df064c3
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/BridgedRelationsProvider.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.RelationResolution
+import org.apache.spark.sql.connector.catalog.CatalogManager
+
+/**
+ * The [[BridgedRelationMetadataProvider]] is a [[RelationMetadataProvider]] that just reuses
+ * resolved metadata from the [[AnalyzerBridgeState]]. This is used in the single-pass [[Resolver]]
+ * to avoid duplicate catalog/table lookups in dual-run mode, so metadata is simply reused from the
+ * fixed-point [[Analyzer]] run. We strictly rely on the [[AnalyzerBridgeState]] to avoid any
+ * blocking calls here.
+ */
+class BridgedRelationMetadataProvider(
+ override val catalogManager: CatalogManager,
+ override val relationResolution: RelationResolution,
+ analyzerBridgeState: AnalyzerBridgeState
+) extends RelationMetadataProvider {
+ override val relationsWithResolvedMetadata = getRelationsFromBridgeState(analyzerBridgeState)
+
+ private def getRelationsFromBridgeState(
+ analyzerBridgeState: AnalyzerBridgeState): RelationsWithResolvedMetadata = {
+ val result = new RelationsWithResolvedMetadata
+ analyzerBridgeState.relationsWithResolvedMetadata.forEach(
+ (unresolvedRelation, relationWithResolvedMetadata) => {
+ result.put(
+ relationIdFromUnresolvedRelation(unresolvedRelation),
+ relationWithResolvedMetadata
+ )
+ }
+ )
+ result
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ConditionalExpressionResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ConditionalExpressionResolver.scala
new file mode 100644
index 000000000000..75ba1b7a01a5
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ConditionalExpressionResolver.scala
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.{AnsiTypeCoercion, TypeCoercion}
+import org.apache.spark.sql.catalyst.expressions.{ConditionalExpression, Expression}
+
+/**
+ * Resolver for [[If]], [[CaseWhen]] and [[Coalesce]] expressions.
+ */
+class ConditionalExpressionResolver(
+ expressionResolver: ExpressionResolver,
+ timezoneAwareExpressionResolver: TimezoneAwareExpressionResolver)
+ extends TreeNodeResolver[ConditionalExpression, Expression]
+ with ResolvesExpressionChildren
+ with SQLConfHelper {
+
+ private val typeCoercionRules: Seq[Expression => Expression] =
+ if (conf.ansiEnabled) {
+ ConditionalExpressionResolver.ANSI_TYPE_COERCION_RULES
+ } else {
+ ConditionalExpressionResolver.TYPE_COERCION_RULES
+ }
+ private val typeCoercionResolver: TypeCoercionResolver =
+ new TypeCoercionResolver(timezoneAwareExpressionResolver, typeCoercionRules)
+
+ override def resolve(unresolvedConditionalExpression: ConditionalExpression): Expression = {
+ val conditionalExpressionWithResolvedChildren =
+ withResolvedChildren(unresolvedConditionalExpression, expressionResolver.resolve)
+
+ typeCoercionResolver.resolve(conditionalExpressionWithResolvedChildren)
+ }
+}
+
+object ConditionalExpressionResolver {
+ // Ordering in the list of type coercions should be in sync with the list in [[TypeCoercion]].
+ private val TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ TypeCoercion.CaseWhenTypeCoercion.apply,
+ TypeCoercion.FunctionArgumentTypeCoercion.apply,
+ TypeCoercion.IfTypeCoercion.apply
+ )
+
+ // Ordering in the list of type coercions should be in sync with the list in [[AnsiTypeCoercion]].
+ private val ANSI_TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ AnsiTypeCoercion.CaseWhenTypeCoercion.apply,
+ AnsiTypeCoercion.FunctionArgumentTypeCoercion.apply,
+ AnsiTypeCoercion.IfTypeCoercion.apply
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/CreateNamedStructResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/CreateNamedStructResolver.scala
new file mode 100644
index 000000000000..12c3c71b5e8b
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/CreateNamedStructResolver.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, CreateNamedStruct, Expression}
+
+/**
+ * Resolves [[CreateNamedStruct]] nodes by recursively resolving children. If [[CreateNamedStruct]]
+ * is not directly under an [[Alias]], removes aliases from struct fields. Otherwise, let
+ * [[AliasResolver]] handle the removal.
+ */
+class CreateNamedStructResolver(expressionResolver: ExpressionResolver)
+ extends TreeNodeResolver[CreateNamedStruct, Expression]
+ with ResolvesExpressionChildren {
+
+ override def resolve(createNamedStruct: CreateNamedStruct): Expression = {
+ val createNamedStructWithResolvedChildren =
+ withResolvedChildren(createNamedStruct, expressionResolver.resolve)
+ CreateNamedStructResolver.cleanupAliases(createNamedStructWithResolvedChildren)
+ }
+}
+
+object CreateNamedStructResolver {
+
+ /**
+ * For a query like:
+ *
+ * {{{ SELECT STRUCT(1 AS a, 2 AS b) }}}
+ *
+ * [[CreateNamedStruct]] will be: CreateNamedStruct(Seq("a", Alias(1, "a"), "b", Alias(2, "b")))
+ *
+ * Because inner aliases are not expected in the analyzed logical plan, we need to remove them
+ * here. However, we only do so if [[CreateNamedStruct]] is not directly under an [[Alias]], in
+ * which case the removal will be handled by [[AliasResolver]]. This is because in single-pass,
+ * [[Alias]] is resolved after [[CreateNamedStruct]] and in order to compute the correct output
+ * name, it needs to know complete structure of the child.
+ */
+ def cleanupAliases(createNamedStruct: CreateNamedStruct): CreateNamedStruct = {
+ createNamedStruct
+ .withNewChildren(createNamedStruct.children.map {
+ case a: Alias if a.metadata.isEmpty => a.child
+ case other => other
+ })
+ .asInstanceOf[CreateNamedStruct]
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/DelegatesResolutionToExtensions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/DelegatesResolutionToExtensions.scala
new file mode 100644
index 000000000000..7d57e4683df4
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/DelegatesResolutionToExtensions.scala
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.errors.QueryCompilationErrors
+
+/**
+ * The [[DelegatesResolutionToExtensions]] is a trait which provides a method to delegate the
+ * resolution of unresolved operators to a list of [[ResolverExtension]]s.
+ */
+trait DelegatesResolutionToExtensions {
+
+ protected val extensions: Seq[ResolverExtension]
+
+ /**
+ * Find the suitable extension for `unresolvedOperator` resolution and resolve it with that
+ * extension. Usually extensions return resolved relation nodes, so we generically update the name
+ * scope without matching for specific relations, for simplicity.
+ *
+ * We match the extension once to reduce the number of
+ * [[ResolverExtension.resolveOperator.isDefinedAt]] calls, because those can be expensive.
+ *
+ * @returns `Some(resolutionResult)` if the extension was found and `unresolvedOperator` was
+ * resolved, `None` otherwise.
+ *
+ * @throws `AMBIGUOUS_RESOLVER_EXTENSION` if there were several matched extensions for this
+ * operator.
+ */
+ def tryDelegateResolutionToExtension(unresolvedOperator: LogicalPlan): Option[LogicalPlan] = {
+ var resolutionResult: Option[LogicalPlan] = None
+ var matchedExtension: Option[ResolverExtension] = None
+ extensions.foreach { extension =>
+ matchedExtension match {
+ case None =>
+ resolutionResult = extension.resolveOperator.lift(unresolvedOperator)
+
+ if (resolutionResult.isDefined) {
+ matchedExtension = Some(extension)
+ }
+ case Some(matchedExtension) =>
+ if (extension.resolveOperator.isDefinedAt(unresolvedOperator)) {
+ throw QueryCompilationErrors
+ .ambiguousResolverExtension(
+ unresolvedOperator,
+ Seq(matchedExtension, extension).map(_.getClass.getSimpleName)
+ )
+ .withPosition(unresolvedOperator.origin)
+ }
+ }
+ }
+
+ resolutionResult
+ }
+}
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
new file mode 100644
index 000000000000..e6279c974039
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExplicitlyUnsupportedResolverFeature.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+/**
+ * This is an addon to [[ResolverGuard]] functionality for features that cannot be determined by
+ * only looking at the unresolved plan. [[Resolver]] will throw this control-flow exception
+ * when it encounters some explicitly unsupported feature. Later behavior depends on the value of
+ * [[HybridAnalyzer.checkSupportedSinglePassFeatures]] flag:
+ * - If it is true: It will later be caught by [[HybridAnalyzer]] to abort single-pass
+ * analysis without comparing single-pass and fixed-point results. The motivation for this
+ * feature is the same as for the [[ResolverGuard]] - we want to have an explicit allowlist of
+ * unimplemented features that we are aware of, and `UNSUPPORTED_SINGLE_PASS_ANALYZER_FEATURE`
+ * will signal us the rest of the gaps.
+ * - If it is false: It will be thrown by the [[HybridAnalyzer]] in order to get better sense
+ * of coverage.
+ *
+ * For example, [[UnresolvedRelation]] can be intermediately resolved by [[ResolveRelations]] as
+ * [[UnresolvedCatalogRelation]] or a [[View]] (among all others). Say that for now the views
+ * are not implemented, and we are aware of that, so [[ExplicitlyUnsupportedResolverFeature]] will
+ * be thrown in the middle of the single-pass analysis to abort it.
+ */
+class ExplicitlyUnsupportedResolverFeature(reason: String)
+ extends Exception(
+ s"The single-pass analyzer cannot process this query or command because it does not yet " +
+ s"support $reason."
+ ) {
+ override def getStackTrace(): Array[StackTraceElement] = new Array[StackTraceElement](0)
+ override def fillInStackTrace(): Throwable = this
+}
+
+/**
+ * This object contains all the metadata on explicitly unsupported resolver features.
+ */
+object ExplicitlyUnsupportedResolverFeature {
+ val OPERATORS = Set(
+ "org.apache.spark.sql.catalyst.plans.logical.View",
+ "org.apache.spark.sql.catalyst.streaming.StreamingRelationV2",
+ "org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation",
+ "org.apache.spark.sql.execution.streaming.StreamingRelation"
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolutionValidator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolutionValidator.scala
new file mode 100644
index 000000000000..8c80992e2fa2
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolutionValidator.scala
@@ -0,0 +1,367 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.expressions.{
+ Alias,
+ ArrayDistinct,
+ ArrayInsert,
+ ArrayJoin,
+ ArrayMax,
+ ArrayMin,
+ ArraysZip,
+ AttributeReference,
+ BinaryExpression,
+ ConditionalExpression,
+ CreateArray,
+ CreateMap,
+ CreateNamedStruct,
+ Expression,
+ ExtractANSIIntervalDays,
+ GetArrayStructFields,
+ GetMapValue,
+ GetStructField,
+ Literal,
+ MapConcat,
+ MapContainsKey,
+ MapEntries,
+ MapFromEntries,
+ MapKeys,
+ MapValues,
+ NamedExpression,
+ Predicate,
+ RuntimeReplaceable,
+ StringRPad,
+ StringToMap,
+ TimeZoneAwareExpression,
+ UnaryMinus
+}
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * The [[ExpressionResolutionValidator]] performs the validation work on the expression tree for the
+ * [[ResolutionValidator]]. These two components work together recursively validating the
+ * logical plan. You can find more info in the [[ResolutionValidator]] scaladoc.
+ */
+class ExpressionResolutionValidator(resolutionValidator: ResolutionValidator) {
+
+ /**
+ * Validate resolved expression tree. The principle is the same as
+ * [[ResolutionValidator.validate]].
+ */
+ def validate(expression: Expression): Unit = {
+ expression match {
+ case attributeReference: AttributeReference =>
+ validateAttributeReference(attributeReference)
+ case alias: Alias =>
+ validateAlias(alias)
+ case getMapValue: GetMapValue =>
+ validateGetMapValue(getMapValue)
+ case binaryExpression: BinaryExpression =>
+ validateBinaryExpression(binaryExpression)
+ case extractANSIIntervalDay: ExtractANSIIntervalDays =>
+ validateExtractANSIIntervalDays(extractANSIIntervalDay)
+ case literal: Literal =>
+ validateLiteral(literal)
+ case predicate: Predicate =>
+ validatePredicate(predicate)
+ case stringRPad: StringRPad =>
+ validateStringRPad(stringRPad)
+ case unaryMinus: UnaryMinus =>
+ validateUnaryMinus(unaryMinus)
+ case getStructField: GetStructField =>
+ validateGetStructField(getStructField)
+ case createNamedStruct: CreateNamedStruct =>
+ validateCreateNamedStruct(createNamedStruct)
+ case getArrayStructFields: GetArrayStructFields =>
+ validateGetArrayStructFields(getArrayStructFields)
+ case createMap: CreateMap =>
+ validateCreateMap(createMap)
+ case stringToMap: StringToMap =>
+ validateStringToMap(stringToMap)
+ case mapContainsKey: MapContainsKey =>
+ validateMapContainsKey(mapContainsKey)
+ case mapConcat: MapConcat =>
+ validateMapConcat(mapConcat)
+ case mapKeys: MapKeys =>
+ validateMapKeys(mapKeys)
+ case mapValues: MapValues =>
+ validateMapValues(mapValues)
+ case mapEntries: MapEntries =>
+ validateMapEntries(mapEntries)
+ case mapFromEntries: MapFromEntries =>
+ validateMapFromEntries(mapFromEntries)
+ case createArray: CreateArray =>
+ validateCreateArray(createArray)
+ case arrayDistinct: ArrayDistinct =>
+ validateArrayDistinct(arrayDistinct)
+ case arrayInsert: ArrayInsert =>
+ validateArrayInsert(arrayInsert)
+ case arrayJoin: ArrayJoin =>
+ validateArrayJoin(arrayJoin)
+ case arrayMax: ArrayMax =>
+ validateArrayMax(arrayMax)
+ case arrayMin: ArrayMin =>
+ validateArrayMin(arrayMin)
+ case arraysZip: ArraysZip =>
+ validateArraysZip(arraysZip)
+ case conditionalExpression: ConditionalExpression =>
+ validateConditionalExpression(conditionalExpression)
+ case runtimeReplaceable: RuntimeReplaceable =>
+ validateRuntimeReplaceable(runtimeReplaceable)
+ case timezoneExpression: TimeZoneAwareExpression =>
+ validateTimezoneExpression(timezoneExpression)
+ }
+ }
+
+ def validateProjectList(projectList: Seq[NamedExpression]): Unit = {
+ projectList.foreach {
+ case attributeReference: AttributeReference =>
+ validateAttributeReference(attributeReference)
+ case alias: Alias =>
+ validateAlias(alias)
+ }
+ }
+
+ private def validatePredicate(predicate: Predicate) = {
+ predicate.children.foreach(validate)
+ assert(
+ predicate.dataType == BooleanType,
+ s"Output type of a predicate must be a boolean, but got: ${predicate.dataType.typeName}"
+ )
+ assert(
+ predicate.checkInputDataTypes().isSuccess,
+ "Input types of a predicate must be valid, but got: " +
+ predicate.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateStringRPad(stringRPad: StringRPad) = {
+ validate(stringRPad.first)
+ validate(stringRPad.second)
+ validate(stringRPad.third)
+ assert(
+ stringRPad.checkInputDataTypes().isSuccess,
+ "Input types of rpad must be valid, but got: " +
+ stringRPad.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateAttributeReference(attributeReference: AttributeReference): Unit = {
+ assert(
+ resolutionValidator.attributeScopeStack.top.contains(attributeReference),
+ s"Attribute $attributeReference is missing from attribute scope: " +
+ s"${resolutionValidator.attributeScopeStack.top}"
+ )
+ }
+
+ private def validateAlias(alias: Alias): Unit = {
+ validate(alias.child)
+ }
+
+ private def validateBinaryExpression(binaryExpression: BinaryExpression): Unit = {
+ validate(binaryExpression.left)
+ validate(binaryExpression.right)
+ assert(
+ binaryExpression.checkInputDataTypes().isSuccess,
+ "Input types of a binary expression must be valid, but got: " +
+ binaryExpression.children.map(_.dataType.typeName).mkString(", ")
+ )
+
+ binaryExpression match {
+ case timezoneExpression: TimeZoneAwareExpression =>
+ assert(timezoneExpression.timeZoneId.nonEmpty, "Timezone expression must have a timezone")
+ case _ =>
+ }
+ }
+
+ private def validateConditionalExpression(conditionalExpression: ConditionalExpression): Unit =
+ conditionalExpression.children.foreach(validate)
+
+ private def validateExtractANSIIntervalDays(
+ extractANSIIntervalDays: ExtractANSIIntervalDays): Unit = {
+ validate(extractANSIIntervalDays.child)
+ }
+
+ private def validateLiteral(literal: Literal): Unit = {}
+
+ private def validateUnaryMinus(unaryMinus: UnaryMinus): Unit = {
+ validate(unaryMinus.child)
+ assert(
+ unaryMinus.checkInputDataTypes().isSuccess,
+ "Input types of a unary minus must be valid, but got: " +
+ unaryMinus.child.dataType.typeName.mkString(", ")
+ )
+ }
+
+ private def validateGetStructField(getStructField: GetStructField): Unit = {
+ validate(getStructField.child)
+ }
+
+ private def validateCreateNamedStruct(createNamedStruct: CreateNamedStruct): Unit = {
+ createNamedStruct.children.foreach(validate)
+ assert(
+ createNamedStruct.checkInputDataTypes().isSuccess,
+ "Input types of CreateNamedStruct must be valid, but got: " +
+ createNamedStruct.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateGetArrayStructFields(getArrayStructFields: GetArrayStructFields): Unit = {
+ validate(getArrayStructFields.child)
+ }
+
+ private def validateGetMapValue(getMapValue: GetMapValue): Unit = {
+ validate(getMapValue.child)
+ validate(getMapValue.key)
+ assert(
+ getMapValue.checkInputDataTypes().isSuccess,
+ "Input types of GetMapValue must be valid, but got: " +
+ getMapValue.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateCreateMap(createMap: CreateMap): Unit = {
+ createMap.children.foreach(validate)
+ assert(
+ createMap.checkInputDataTypes().isSuccess,
+ "Input types of CreateMap must be valid, but got: " +
+ createMap.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateStringToMap(stringToMap: StringToMap): Unit = {
+ validate(stringToMap.text)
+ validate(stringToMap.pairDelim)
+ validate(stringToMap.keyValueDelim)
+ }
+
+ private def validateMapContainsKey(mapContainsKey: MapContainsKey): Unit = {
+ validate(mapContainsKey.left)
+ validate(mapContainsKey.right)
+ assert(
+ mapContainsKey.checkInputDataTypes().isSuccess,
+ "Input types of MapContainsKey must be valid, but got: " +
+ mapContainsKey.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateMapConcat(mapConcat: MapConcat): Unit = {
+ mapConcat.children.foreach(validate)
+ assert(
+ mapConcat.checkInputDataTypes().isSuccess,
+ "Input types of MapConcat must be valid, but got: " +
+ mapConcat.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateMapKeys(mapKeys: MapKeys): Unit = {
+ validate(mapKeys.child)
+ }
+
+ private def validateMapValues(mapValues: MapValues): Unit = {
+ validate(mapValues.child)
+ }
+
+ private def validateMapEntries(mapEntries: MapEntries): Unit = {
+ validate(mapEntries.child)
+ }
+
+ private def validateMapFromEntries(mapFromEntries: MapFromEntries): Unit = {
+ mapFromEntries.children.foreach(validate)
+ assert(
+ mapFromEntries.checkInputDataTypes().isSuccess,
+ "Input types of MapFromEntries must be valid, but got: " +
+ mapFromEntries.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateCreateArray(createArray: CreateArray): Unit = {
+ createArray.children.foreach(validate)
+ assert(
+ createArray.checkInputDataTypes().isSuccess,
+ "Input types of CreateArray must be valid, but got: " +
+ createArray.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateArrayDistinct(arrayDistinct: ArrayDistinct): Unit = {
+ validate(arrayDistinct.child)
+ assert(
+ arrayDistinct.checkInputDataTypes().isSuccess,
+ "Input types of ArrayDistinct must be valid, but got: " +
+ arrayDistinct.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateArrayInsert(arrayInsert: ArrayInsert): Unit = {
+ validate(arrayInsert.srcArrayExpr)
+ validate(arrayInsert.posExpr)
+ validate(arrayInsert.itemExpr)
+ assert(
+ arrayInsert.checkInputDataTypes().isSuccess,
+ "Input types of ArrayInsert must be valid, but got: " +
+ arrayInsert.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateArrayJoin(arrayJoin: ArrayJoin): Unit = {
+ validate(arrayJoin.array)
+ validate(arrayJoin.delimiter)
+ if (arrayJoin.nullReplacement.isDefined) {
+ validate(arrayJoin.nullReplacement.get)
+ }
+ }
+
+ private def validateArrayMax(arrayMax: ArrayMax): Unit = {
+ validate(arrayMax.child)
+ assert(
+ arrayMax.checkInputDataTypes().isSuccess,
+ "Input types of ArrayMax must be valid, but got: " +
+ arrayMax.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateArrayMin(arrayMin: ArrayMin): Unit = {
+ validate(arrayMin.child)
+ assert(
+ arrayMin.checkInputDataTypes().isSuccess,
+ "Input types of ArrayMin must be valid, but got: " +
+ arrayMin.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateArraysZip(arraysZip: ArraysZip): Unit = {
+ arraysZip.children.foreach(validate)
+ arraysZip.names.foreach(validate)
+ assert(
+ arraysZip.checkInputDataTypes().isSuccess,
+ "Input types of ArraysZip must be valid, but got: " +
+ arraysZip.children.map(_.dataType.typeName).mkString(", ")
+ )
+ }
+
+ private def validateRuntimeReplaceable(runtimeReplaceable: RuntimeReplaceable): Unit = {
+ runtimeReplaceable.children.foreach(validate)
+ }
+
+ private def validateTimezoneExpression(timezoneExpression: TimeZoneAwareExpression): Unit = {
+ timezoneExpression.children.foreach(validate)
+ assert(timezoneExpression.timeZoneId.nonEmpty, "Timezone expression must have a timezone")
+ }
+}
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
new file mode 100644
index 000000000000..0a9e2b9c5a87
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolver.scala
@@ -0,0 +1,340 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.{
+ withPosition,
+ FunctionResolution,
+ UnresolvedAlias,
+ UnresolvedAttribute,
+ UnresolvedFunction,
+ UnresolvedStar
+}
+import org.apache.spark.sql.catalyst.expressions.{
+ Alias,
+ AttributeReference,
+ BinaryArithmetic,
+ ConditionalExpression,
+ CreateNamedStruct,
+ Expression,
+ ExtractANSIIntervalDays,
+ InheritAnalysisRules,
+ Literal,
+ NamedExpression,
+ Predicate,
+ RuntimeReplaceable,
+ TimeAdd,
+ TimeZoneAwareExpression,
+ UnaryMinus
+}
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.MetadataBuilder
+
+/**
+ * The [[ExpressionResolver]] is used by the [[Resolver]] during the analysis to resolve
+ * expressions.
+ *
+ * The functions here generally traverse unresolved [[Expression]] nodes recursively,
+ * constructing and returning the resolved [[Expression]] nodes bottom-up.
+ * This is the primary entry point for implementing expression analysis,
+ * wherein the [[resolve]] method accepts a fully unresolved [[Expression]] and returns
+ * a fully resolved [[Expression]] in response with all data types and attribute
+ * reference ID assigned for valid requests. This resolver also takes responsibility
+ * to detect any errors in the initial SQL query or DataFrame and return appropriate
+ * error messages including precise parse locations wherever possible.
+ *
+ * @param resolver [[Resolver]] is passed from the parent to resolve other
+ * operators which are nested in expressions.
+ * @param scopes [[NameScopeStack]] to resolve the expression tree in the correct scope.
+ * @param functionResolution [[FunctionResolution]] to resolve function expressions.
+ */
+class ExpressionResolver(
+ resolver: Resolver,
+ scopes: NameScopeStack,
+ functionResolution: FunctionResolution)
+ extends TreeNodeResolver[Expression, Expression]
+ with ProducesUnresolvedSubtree
+ with ResolvesExpressionChildren
+ with TracksResolvedNodes[Expression] {
+ private val shouldTrackResolvedNodes =
+ conf.getConf(SQLConf.ANALYZER_SINGLE_PASS_TRACK_RESOLVED_NODES_ENABLED)
+ private val aliasResolver = new AliasResolver(this, scopes)
+ private val createNamedStructResolver = new CreateNamedStructResolver(this)
+ private val timezoneAwareExpressionResolver = new TimezoneAwareExpressionResolver(this)
+ private val conditionalExpressionResolver =
+ new ConditionalExpressionResolver(this, timezoneAwareExpressionResolver)
+ private val predicateResolver =
+ new PredicateResolver(this, timezoneAwareExpressionResolver)
+ private val binaryArithmeticResolver = {
+ new BinaryArithmeticResolver(
+ this,
+ timezoneAwareExpressionResolver
+ )
+ }
+ private val functionResolver = new FunctionResolver(
+ this,
+ timezoneAwareExpressionResolver,
+ functionResolution
+ )
+ private val timeAddResolver = new TimeAddResolver(this, timezoneAwareExpressionResolver)
+ private val unaryMinusResolver = new UnaryMinusResolver(this, timezoneAwareExpressionResolver)
+
+ /**
+ * This method is an expression analysis entry point. The method first checks if the expression
+ * has already been resolved (necessary because of partially-unresolved subtrees, see
+ * [[ProducesUnresolvedSubtree]]). If not already resolved, method takes an unresolved
+ * [[Expression]] and chooses the right `resolve*` method using pattern matching on the
+ * `unresolvedExpression` type. This pattern matching enumerates all the expression node types
+ * that are supported by the single-pass analysis.
+ * When developers introduce a new [[Expression]] type to the Catalyst, they should implement
+ * a corresponding `resolve*` method in the [[ExpressionResolver]] and add it to this pattern
+ * match list.
+ *
+ * [[resolve]] will be called recursively during the expression tree traversal eventually
+ * producing a fully resolved expression subtree or a descriptive error message.
+ *
+ * [[resolve]] can recursively call `resolver` to resolve nested operators (e.g. scalar
+ * subqueries):
+ *
+ * {{{ SELECT * FROM VALUES (1), (2) WHERE col1 IN (SELECT 1); }}}
+ *
+ * In this case `IN` is an expression and `SELECT 1` is a nested operator tree for which
+ * the [[ExpressionResolver]] would invoke the [[Resolver]].
+ */
+ override def resolve(unresolvedExpression: Expression): Expression =
+ if (unresolvedExpression
+ .getTagValue(ExpressionResolver.SINGLE_PASS_SUBTREE_BOUNDARY)
+ .nonEmpty) {
+ unresolvedExpression
+ } else {
+ throwIfNodeWasResolvedEarlier(unresolvedExpression)
+
+ val resolvedExpr = unresolvedExpression match {
+ case unresolvedBinaryArithmetic: BinaryArithmetic =>
+ binaryArithmeticResolver.resolve(unresolvedBinaryArithmetic)
+ case unresolvedExtractANSIIntervalDays: ExtractANSIIntervalDays =>
+ resolveExtractANSIIntervalDays(unresolvedExtractANSIIntervalDays)
+ case unresolvedNamedExpression: NamedExpression =>
+ resolveNamedExpression(unresolvedNamedExpression)
+ case unresolvedFunction: UnresolvedFunction =>
+ functionResolver.resolve(unresolvedFunction)
+ case unresolvedLiteral: Literal =>
+ resolveLiteral(unresolvedLiteral)
+ case unresolvedPredicate: Predicate =>
+ predicateResolver.resolve(unresolvedPredicate)
+ case unresolvedTimeAdd: TimeAdd =>
+ timeAddResolver.resolve(unresolvedTimeAdd)
+ case unresolvedUnaryMinus: UnaryMinus =>
+ unaryMinusResolver.resolve(unresolvedUnaryMinus)
+ case createNamedStruct: CreateNamedStruct =>
+ createNamedStructResolver.resolve(createNamedStruct)
+ case unresolvedConditionalExpression: ConditionalExpression =>
+ conditionalExpressionResolver.resolve(unresolvedConditionalExpression)
+ case unresolvedRuntimeReplaceable: RuntimeReplaceable =>
+ resolveRuntimeReplaceable(unresolvedRuntimeReplaceable)
+ case unresolvedTimezoneExpression: TimeZoneAwareExpression =>
+ timezoneAwareExpressionResolver.resolve(unresolvedTimezoneExpression)
+ case _ =>
+ withPosition(unresolvedExpression) {
+ throwUnsupportedSinglePassAnalyzerFeature(unresolvedExpression)
+ }
+ }
+
+ markNodeAsResolved(resolvedExpr)
+
+ resolvedExpr
+ }
+
+ private def resolveNamedExpression(
+ unresolvedNamedExpression: Expression,
+ isTopOfProjectList: Boolean = false): Expression =
+ unresolvedNamedExpression match {
+ case alias: Alias =>
+ aliasResolver.handleResolvedAlias(alias)
+ case unresolvedAlias: UnresolvedAlias =>
+ aliasResolver.resolve(unresolvedAlias)
+ case unresolvedAttribute: UnresolvedAttribute =>
+ resolveAttribute(unresolvedAttribute, isTopOfProjectList)
+ case unresolvedStar: UnresolvedStar =>
+ withPosition(unresolvedStar) {
+ throwInvalidStarUsageError(unresolvedStar)
+ }
+ case attributeReference: AttributeReference =>
+ handleResolvedAttributeReference(attributeReference)
+ case _ =>
+ withPosition(unresolvedNamedExpression) {
+ throwUnsupportedSinglePassAnalyzerFeature(unresolvedNamedExpression)
+ }
+ }
+
+ /**
+ * The [[Project]] list can contain different unresolved expressions before the resolution, which
+ * will be resolved using generic [[resolve]]. However, [[UnresolvedStar]] is a special case,
+ * because it is expanded into a sequence of [[NamedExpression]]s. Because of that this method
+ * returns a sequence and doesn't conform to generic [[resolve]] interface - it's called directly
+ * from the [[Resolver]] during [[Project]] resolution.
+ *
+ * The output sequence can be larger than the input sequence due to [[UnresolvedStar]] expansion.
+ */
+ def resolveProjectList(unresolvedProjectList: Seq[NamedExpression]): Seq[NamedExpression] = {
+ unresolvedProjectList.flatMap {
+ case unresolvedStar: UnresolvedStar =>
+ resolveStar(unresolvedStar)
+ case other =>
+ Seq(resolveNamedExpression(other, isTopOfProjectList = true).asInstanceOf[NamedExpression])
+ }
+ }
+
+ /**
+ * [[UnresolvedAttribute]] resolution relies on [[NameScope]] to lookup the attribute by its
+ * multipart name. The resolution can result in three different outcomes which are handled in the
+ * [[NameTarget.pickCandidate]]:
+ *
+ * - No results from the [[NameScope]] mean that the attribute lookup failed as in:
+ * {{{ SELECT col1 FROM (SELECT 1 as col2); }}}
+ *
+ * - Several results from the [[NameScope]] mean that the reference is ambiguous as in:
+ * {{{ SELECT col1 FROM (SELECT 1 as col1), (SELECT 2 as col1); }}}
+ *
+ * - Single result from the [[NameScope]] means that the attribute was found as in:
+ * {{{ SELECT col1 FROM VALUES (1); }}}
+ *
+ * If the attribute is at the top of the project list (which is indicated by
+ * [[isTopOfProjectList]]), we preserve the [[Alias]] or remove it otherwise.
+ */
+ private def resolveAttribute(
+ unresolvedAttribute: UnresolvedAttribute,
+ isTopOfProjectList: Boolean): Expression =
+ withPosition(unresolvedAttribute) {
+ if (scopes.top.isExistingAlias(unresolvedAttribute.nameParts.head)) {
+ // Temporarily disable referencing aliases until we support LCA resolution.
+ throw new ExplicitlyUnsupportedResolverFeature("unsupported expression: LateralColumnAlias")
+ }
+
+ val nameTarget: NameTarget = scopes.top.matchMultipartName(unresolvedAttribute.nameParts)
+
+ val candidate = nameTarget.pickCandidate(unresolvedAttribute)
+ if (isTopOfProjectList && nameTarget.aliasName.isDefined) {
+ Alias(candidate, nameTarget.aliasName.get)()
+ } else {
+ candidate
+ }
+ }
+
+ /**
+ * [[AttributeReference]] is already resolved if it's passed to us from DataFrame `col(...)`
+ * function, for example.
+ */
+ private def handleResolvedAttributeReference(attributeReference: AttributeReference) =
+ tryStripAmbiguousSelfJoinMetadata(attributeReference)
+
+ /**
+ * [[ExtractANSIIntervalDays]] resolution doesn't require any specific resolution logic apart
+ * from resolving its children.
+ */
+ private def resolveExtractANSIIntervalDays(
+ unresolvedExtractANSIIntervalDays: ExtractANSIIntervalDays) =
+ withResolvedChildren(unresolvedExtractANSIIntervalDays, resolve)
+
+ /**
+ * [[UnresolvedStar]] resolution relies on the [[NameScope]]'s ability to get the attributes by a
+ * multipart name ([[UnresolvedStar]]'s `target` field):
+ *
+ * - Star target is defined:
+ *
+ * {{{
+ * SELECT t.* FROM VALUES (1) AS t;
+ * ->
+ * Project [col1#19]
+ * }}}
+ *
+ *
+ * - Star target is not defined:
+ *
+ * {{{
+ * SELECT * FROM (SELECT 1 as col1), (SELECT 2 as col2);
+ * ->
+ * Project [col1#19, col2#20]
+ * }}}
+ */
+ def resolveStar(unresolvedStar: UnresolvedStar): Seq[NamedExpression] =
+ withPosition(unresolvedStar) {
+ scopes.top.expandStar(unresolvedStar)
+ }
+
+ /**
+ * [[Literal]] resolution doesn't require any specific resolution logic at this point.
+ *
+ * Since [[TracksResolvedNodes]] requires all the expressions in the tree to be unique objects,
+ * we reallocate the literal in [[ANALYZER_SINGLE_PASS_TRACK_RESOLVED_NODES_ENABLED]] mode,
+ * otherwise we preserve the old object to avoid unnecessary memory allocations.
+ */
+ private def resolveLiteral(literal: Literal): Expression = {
+ if (shouldTrackResolvedNodes) {
+ literal.copy()
+ } else {
+ literal
+ }
+ }
+
+ /**
+ * When [[RuntimeReplaceable]] is mixed in with [[InheritAnalysisRules]], child expression will
+ * be runtime replacement. In that case we need to resolve the children of the expression.
+ * otherwise, no resolution is necessary because replacement is already resolved.
+ */
+ private def resolveRuntimeReplaceable(unresolvedRuntimeReplaceable: RuntimeReplaceable) =
+ unresolvedRuntimeReplaceable match {
+ case inheritAnalysisRules: InheritAnalysisRules =>
+ withResolvedChildren(inheritAnalysisRules, resolve)
+ case other => other
+ }
+
+ /**
+ * [[DetectAmbiguousSelfJoin]] rule in the fixed-point Analyzer detects ambiguous references in
+ * self-joins based on special metadata added by [[Dataset]] code (see SPARK-27547). Just strip
+ * this for now since we don't support joins yet.
+ */
+ private def tryStripAmbiguousSelfJoinMetadata(attributeReference: AttributeReference) = {
+ val metadata = attributeReference.metadata
+ if (ExpressionResolver.AMBIGUOUS_SELF_JOIN_METADATA.exists(metadata.contains(_))) {
+ val metadataBuilder = new MetadataBuilder().withMetadata(metadata)
+ for (metadataKey <- ExpressionResolver.AMBIGUOUS_SELF_JOIN_METADATA) {
+ metadataBuilder.remove(metadataKey)
+ }
+ attributeReference.withMetadata(metadataBuilder.build())
+ } else {
+ attributeReference
+ }
+ }
+
+ private def throwUnsupportedSinglePassAnalyzerFeature(unresolvedExpression: Expression): Nothing =
+ throw QueryCompilationErrors.unsupportedSinglePassAnalyzerFeature(
+ s"${unresolvedExpression.getClass} expression resolution"
+ )
+
+ private def throwInvalidStarUsageError(unresolvedStar: UnresolvedStar): Nothing =
+ // TODO(vladimirg-db): Use parent operator name instead of "query"
+ throw QueryCompilationErrors.invalidStarUsageError("query", Seq(unresolvedStar))
+}
+
+object ExpressionResolver {
+ private val AMBIGUOUS_SELF_JOIN_METADATA = Seq("__dataset_id", "__col_position")
+ val SINGLE_PASS_SUBTREE_BOUNDARY = TreeNodeTag[Unit]("single_pass_subtree_boundary")
+}
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
new file mode 100644
index 000000000000..b7311b83e872
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolver.scala
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.{
+ AnsiTypeCoercion,
+ CollationTypeCoercion,
+ FunctionResolution,
+ TypeCoercion,
+ UnresolvedFunction,
+ UnresolvedStar
+}
+import org.apache.spark.sql.catalyst.expressions.Expression
+
+/**
+ * A resolver for [[UnresolvedFunction]]s that resolves functions to concrete [[Expression]]s.
+ * It resolves the children of the function first by calling [[ExpressionResolver.resolve]] on them
+ * if they are not [[UnresolvedStar]]s. If the children are [[UnresolvedStar]]s, it resolves them
+ * using [[ExpressionResolver.resolveStar]]. Examples are following:
+ *
+ * - Function doesn't contain any [[UnresolvedStar]]:
+ * {{{ SELECT ARRAY(col1) FROM VALUES (1); }}}
+ * it is resolved only using [[ExpressionResolver.resolve]].
+ * - Function contains [[UnresolvedStar]]:
+ * {{{ SELECT ARRAY(*) FROM VALUES (1); }}}
+ * it is resolved using [[ExpressionResolver.resolveStar]].
+ *
+ * It applies appropriate [[TypeCoercion]] (or [[AnsiTypeCoercion]]) rules after resolving the
+ * function using the [[FunctionResolution]] code.
+ */
+class FunctionResolver(
+ expressionResolver: ExpressionResolver,
+ timezoneAwareExpressionResolver: TimezoneAwareExpressionResolver,
+ functionResolution: FunctionResolution)
+ extends TreeNodeResolver[UnresolvedFunction, Expression]
+ with ProducesUnresolvedSubtree {
+
+ private val typeCoercionRules: Seq[Expression => Expression] =
+ if (conf.ansiEnabled) {
+ FunctionResolver.ANSI_TYPE_COERCION_RULES
+ } else {
+ FunctionResolver.TYPE_COERCION_RULES
+ }
+ private val typeCoercionResolver: TypeCoercionResolver =
+ new TypeCoercionResolver(timezoneAwareExpressionResolver, typeCoercionRules)
+
+ override def resolve(unresolvedFunction: UnresolvedFunction): Expression = {
+ val functionWithResolvedChildren =
+ unresolvedFunction.copy(arguments = unresolvedFunction.arguments.flatMap {
+ case s: UnresolvedStar => expressionResolver.resolveStar(s)
+ case other => Seq(expressionResolver.resolve(other))
+ })
+ val resolvedFunction = functionResolution.resolveFunction(functionWithResolvedChildren)
+ typeCoercionResolver.resolve(resolvedFunction)
+ }
+}
+
+object FunctionResolver {
+ // Ordering in the list of type coercions should be in sync with the list in [[TypeCoercion]].
+ private val TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ CollationTypeCoercion.apply,
+ TypeCoercion.InTypeCoercion.apply,
+ TypeCoercion.FunctionArgumentTypeCoercion.apply,
+ TypeCoercion.IfTypeCoercion.apply,
+ TypeCoercion.ImplicitTypeCoercion.apply
+ )
+
+ // Ordering in the list of type coercions should be in sync with the list in [[AnsiTypeCoercion]].
+ private val ANSI_TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ CollationTypeCoercion.apply,
+ AnsiTypeCoercion.InTypeCoercion.apply,
+ AnsiTypeCoercion.FunctionArgumentTypeCoercion.apply,
+ AnsiTypeCoercion.IfTypeCoercion.apply,
+ AnsiTypeCoercion.ImplicitTypeCoercion.apply
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/HybridAnalyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/HybridAnalyzer.scala
new file mode 100644
index 000000000000..8a1a8602368e
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/HybridAnalyzer.scala
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.sql.catalyst.{QueryPlanningTracker, SQLConfHelper}
+import org.apache.spark.sql.catalyst.analysis.{AnalysisContext, Analyzer}
+import org.apache.spark.sql.catalyst.plans.NormalizePlan
+import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper, LogicalPlan}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * The HybridAnalyzer routes the unresolved logical plan between the legacy Analyzer and
+ * a single-pass Analyzer when the query that we are processing is being run from unit tests
+ * depending on the testing flags set and the structure of this unresolved logical plan:
+ * - If the "spark.sql.analyzer.singlePassResolver.soloRunEnabled" is "true", the
+ * [[HybridAnalyzer]] will unconditionally run the single-pass Analyzer, which would
+ * usually result in some unexpected behavior and failures. This flag is used only for
+ * development.
+ * - If the "spark.sql.analyzer.singlePassResolver.dualRunEnabled" is "true", the
+ * [[HybridAnalyzer]] will invoke the legacy analyzer and optionally _also_ the fixed-point
+ * one depending on the structure of the unresolved plan. This decision is based on which
+ * features are supported by the single-pass Analyzer, and the checking is implemented in
+ * the [[ResolverGuard]]. After that we validate the results using the following
+ * logic:
+ * - If the fixed-point Analyzer fails and the single-pass one succeeds, we throw an
+ * appropriate exception (please check the
+ * [[QueryCompilationErrors.fixedPointFailedSinglePassSucceeded]] method)
+ * - If both the fixed-point and the single-pass Analyzers failed, we throw the exception
+ * from the fixed-point Analyzer.
+ * - If the single-pass Analyzer failed, we throw an exception from its failure.
+ * - If both the fixed-point and the single-pass Analyzers succeeded, we compare the logical
+ * plans and output schemas, and return the resolved plan from the fixed-point Analyzer.
+ * - Otherwise we run the legacy analyzer.
+ * */
+class HybridAnalyzer(legacyAnalyzer: Analyzer, resolverGuard: ResolverGuard, resolver: Resolver)
+ extends SQLConfHelper {
+ private var singlePassResolutionDuration: Option[Long] = None
+ private var fixedPointResolutionDuration: Option[Long] = None
+
+ def apply(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {
+ val dualRun =
+ conf.getConf(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER) &&
+ resolverGuard.apply(plan)
+
+ withTrackedAnalyzerBridgeState(dualRun) {
+ if (dualRun) {
+ resolveInDualRun(plan, tracker)
+ } else if (conf.getConf(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED)) {
+ resolveInSinglePass(plan)
+ } else {
+ resolveInFixedPoint(plan, tracker)
+ }
+ }
+ }
+
+ def getSinglePassResolutionDuration: Option[Long] = singlePassResolutionDuration
+
+ def getFixedPointResolutionDuration: Option[Long] = fixedPointResolutionDuration
+
+ /**
+ * Call `body` in the context of tracked [[AnalyzerBridgeState]]. Set the new bridge state
+ * depending on whether we are in dual-run mode or not:
+ * - If [[dualRun]] and [[ANALYZER_SINGLE_PASS_RESOLVER_RELATION_BRIDGING_ENABLED]] are true,
+ * create and set a new [[AnalyzerBridgeState]].
+ * - Otherwise, reset [[AnalyzerBridgeState]].
+ *
+ * Finally, set the bridge state back to the previous one after the `body` is executed to avoid
+ * disrupting the possible upper-level [[Analyzer]] invocation in case it's recursive
+ * [[Analyzer]] call.
+ * */
+ private def withTrackedAnalyzerBridgeState(dualRun: Boolean)(
+ body: => LogicalPlan): LogicalPlan = {
+ val bridgeRelations = dualRun && conf.getConf(
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_RELATION_BRIDGING_ENABLED
+ )
+
+ val prevSinglePassResolverBridgeState = AnalysisContext.get.getSinglePassResolverBridgeState
+
+ AnalysisContext.get.setSinglePassResolverBridgeState(if (bridgeRelations) {
+ Some(new AnalyzerBridgeState)
+ } else {
+ None
+ })
+
+ try {
+ body
+ } finally {
+ AnalysisContext.get.setSinglePassResolverBridgeState(prevSinglePassResolverBridgeState)
+ }
+ }
+
+ /**
+ * This method is used to run both the legacy Analyzer and single-pass Analyzer,
+ * and then compare the results or check the errors. For more context please check the
+ * [[HybridAnalyzer]] scaladoc.
+ * */
+ private def resolveInDualRun(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {
+ var fixedPointException: Option[Throwable] = None
+ val fixedPointResult = try {
+ val (resolutionDuration, result) = recordDuration {
+ Some(resolveInFixedPoint(plan, tracker))
+ }
+ fixedPointResolutionDuration = Some(resolutionDuration)
+ result
+ } catch {
+ case NonFatal(e) =>
+ fixedPointException = Some(e)
+ None
+ }
+
+ var singlePassException: Option[Throwable] = None
+ val singlePassResult = try {
+ val (resolutionDuration, result) = recordDuration {
+ Some(resolveInSinglePass(plan))
+ }
+ singlePassResolutionDuration = Some(resolutionDuration)
+ result
+ } catch {
+ case NonFatal(e) =>
+ singlePassException = Some(e)
+ None
+ }
+
+ fixedPointException match {
+ case Some(fixedPointEx) =>
+ singlePassException match {
+ case Some(_) =>
+ throw fixedPointEx
+ case None =>
+ throw QueryCompilationErrors.fixedPointFailedSinglePassSucceeded(
+ singlePassResult.get,
+ fixedPointEx
+ )
+ }
+ case None =>
+ singlePassException match {
+ case Some(singlePassEx: ExplicitlyUnsupportedResolverFeature) =>
+ fixedPointResult.get
+ case Some(singlePassEx) =>
+ throw singlePassEx
+ case None =>
+ validateLogicalPlans(fixedPointResult.get, singlePassResult.get)
+ fixedPointResult.get
+ }
+ }
+ }
+
+ /**
+ * This method is used to run the single-pass Analyzer which will return the resolved plan
+ * or throw an exception if the resolution fails. Both cases are handled in the caller method.
+ * */
+ private def resolveInSinglePass(plan: LogicalPlan): LogicalPlan = {
+ val resolvedPlan = resolver.lookupMetadataAndResolve(
+ plan,
+ analyzerBridgeState = AnalysisContext.get.getSinglePassResolverBridgeState
+ )
+ if (conf.getConf(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_VALIDATION_ENABLED)) {
+ val validator = new ResolutionValidator
+ validator.validatePlan(resolvedPlan)
+ }
+ resolvedPlan
+ }
+
+ /**
+ * This method is used to run the legacy Analyzer which will return the resolved plan
+ * or throw an exception if the resolution fails. Both cases are handled in the caller method.
+ * */
+ private def resolveInFixedPoint(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {
+ val resolvedPlan = legacyAnalyzer.executeAndTrack(plan, tracker)
+ QueryPlanningTracker.withTracker(tracker) {
+ legacyAnalyzer.checkAnalysis(resolvedPlan)
+ }
+ resolvedPlan
+ }
+
+ private def validateLogicalPlans(fixedPointResult: LogicalPlan, singlePassResult: LogicalPlan) = {
+ if (fixedPointResult.schema != singlePassResult.schema) {
+ throw QueryCompilationErrors.hybridAnalyzerOutputSchemaComparisonMismatch(
+ fixedPointResult.schema,
+ singlePassResult.schema
+ )
+ }
+ if (normalizePlan(fixedPointResult) != normalizePlan(singlePassResult)) {
+ throw QueryCompilationErrors.hybridAnalyzerLogicalPlanComparisonMismatch(
+ fixedPointResult,
+ singlePassResult
+ )
+ }
+ }
+
+ private def normalizePlan(plan: LogicalPlan) = AnalysisHelper.allowInvokingTransformsInAnalyzer {
+ NormalizePlan(plan)
+ }
+
+ private def recordDuration[T](thunk: => T): (Long, T) = {
+ val start = System.nanoTime()
+ val res = thunk
+ (System.nanoTime() - start, res)
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/IdentifierMap.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/IdentifierMap.scala
new file mode 100644
index 000000000000..899eb7d71e81
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/IdentifierMap.scala
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import java.util.Locale
+
+/**
+ * The [[IdentifierMap]] is an implementation of a [[KeyTransformingMap]] that uses SQL/DataFrame
+ * identifiers as keys. The implementation is case-insensitive for keys.
+ */
+private class IdentifierMap[V] extends KeyTransformingMap[String, V] {
+ override def mapKey(key: String): String = key.toLowerCase(Locale.ROOT)
+}
+
+/**
+ * The [[OptionalIdentifierMap]] is an implementation of a [[KeyTransformingMap]] that uses optional
+ * SQL/DataFrame identifiers as keys. The implementation is case-insensitive for non-empty keys.
+ */
+private class OptionalIdentifierMap[V] extends KeyTransformingMap[Option[String], V] {
+ override def mapKey(key: Option[String]): Option[String] =
+ key.map(_.toLowerCase(Locale.ROOT))
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/KeyTransformingMap.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/KeyTransformingMap.scala
new file mode 100644
index 000000000000..ff6e118fcc3c
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/KeyTransformingMap.scala
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import scala.collection.mutable
+
+/**
+ * The [[KeyTransformingMap]] is a partial implementation of [[mutable.Map]] that transforms input
+ * keys with a custom [[mapKey]] method.
+ */
+private abstract class KeyTransformingMap[K, V] {
+ private val impl = new mutable.HashMap[K, V]
+
+ def get(key: K): Option[V] = impl.get(mapKey(key))
+
+ def contains(key: K): Boolean = impl.contains(mapKey(key))
+
+ def iterator: Iterator[(K, V)] = impl.iterator
+
+ def +=(kv: (K, V)): this.type = {
+ impl += (mapKey(kv._1) -> kv._2)
+ this
+ }
+
+ def -=(key: K): this.type = {
+ impl -= mapKey(key)
+ this
+ }
+
+ def mapKey(key: K): K
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/LimitExpressionResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/LimitExpressionResolver.scala
new file mode 100644
index 000000000000..a25616ba50b6
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/LimitExpressionResolver.scala
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.errors.QueryErrorsBase
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * The [[LimitExpressionResolver]] is a resolver that resolves a [[LocalLimit]] or [[GlobalLimit]]
+ * expression and performs all the necessary validation.
+ */
+class LimitExpressionResolver(expressionResolver: TreeNodeResolver[Expression, Expression])
+ extends TreeNodeResolver[Expression, Expression]
+ with QueryErrorsBase {
+
+ /**
+ * Resolve a limit expression of [[GlobalLimit]] or [[LocalLimit]] and perform validation.
+ */
+ override def resolve(unresolvedLimitExpression: Expression): Expression = {
+ val resolvedLimitExpression = expressionResolver.resolve(unresolvedLimitExpression)
+ validateLimitExpression(resolvedLimitExpression, expressionName = "limit")
+ resolvedLimitExpression
+ }
+
+ /**
+ * Validate a resolved limit expression of [[GlobalLimit]] or [[LocalLimit]]:
+ * - The expression has to be foldable
+ * - The result data type has to be [[IntegerType]]
+ * - The evaluated expression has to be non-null
+ * - The evaluated expression has to be positive
+ *
+ * The `foldable` check is implemented in some expressions
+ * as a recursive expression tree traversal.
+ * It is not an ideal approach for the single-pass [[ExpressionResolver]],
+ * but __is__ practical, since:
+ * - We have to call `eval` here anyway, and it's recursive
+ * - In practice `LIMIT` expression trees are very small
+ */
+ private def validateLimitExpression(expression: Expression, expressionName: String): Unit = {
+ if (!expression.foldable) {
+ throwInvalidLimitLikeExpressionIsUnfoldable(expressionName, expression)
+ }
+ if (expression.dataType != IntegerType) {
+ throwInvalidLimitLikeExpressionDataType(expressionName, expression)
+ }
+ expression.eval() match {
+ case null =>
+ throwInvalidLimitLikeExpressionIsNull(expressionName, expression)
+ case value: Int if value < 0 =>
+ throwInvalidLimitLikeExpressionIsNegative(expressionName, expression, value)
+ case _ =>
+ }
+ }
+
+ private def throwInvalidLimitLikeExpressionIsUnfoldable(
+ name: String,
+ expression: Expression): Nothing =
+ throw new AnalysisException(
+ errorClass = "INVALID_LIMIT_LIKE_EXPRESSION.IS_UNFOLDABLE",
+ messageParameters = Map(
+ "name" -> name,
+ "expr" -> toSQLExpr(expression)
+ ),
+ origin = expression.origin
+ )
+
+ private def throwInvalidLimitLikeExpressionDataType(
+ name: String,
+ expression: Expression): Nothing =
+ throw new AnalysisException(
+ errorClass = "INVALID_LIMIT_LIKE_EXPRESSION.DATA_TYPE",
+ messageParameters = Map(
+ "name" -> name,
+ "expr" -> toSQLExpr(expression),
+ "dataType" -> toSQLType(expression.dataType)
+ ),
+ origin = expression.origin
+ )
+
+ private def throwInvalidLimitLikeExpressionIsNull(name: String, expression: Expression): Nothing =
+ throw new AnalysisException(
+ errorClass = "INVALID_LIMIT_LIKE_EXPRESSION.IS_NULL",
+ messageParameters = Map("name" -> name, "expr" -> toSQLExpr(expression)),
+ origin = expression.origin
+ )
+
+ private def throwInvalidLimitLikeExpressionIsNegative(
+ name: String,
+ expression: Expression,
+ value: Int): Nothing =
+ throw new AnalysisException(
+ errorClass = "INVALID_LIMIT_LIKE_EXPRESSION.IS_NEGATIVE",
+ messageParameters =
+ Map("name" -> name, "expr" -> toSQLExpr(expression), "v" -> toSQLValue(value, IntegerType)),
+ origin = expression.origin
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/MetadataResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/MetadataResolver.scala
new file mode 100644
index 000000000000..e1334fc56575
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/MetadataResolver.scala
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import java.util.ArrayDeque
+
+import org.apache.spark.sql.catalyst.analysis.{withPosition, RelationResolution, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.expressions.{Expression, PlanExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper, LogicalPlan}
+import org.apache.spark.sql.connector.catalog.CatalogManager
+
+/**
+ * The [[MetadataResolver]] performs relation metadata resolution based on the unresolved plan
+ * at the start of the analysis phase. Usually it does RPC calls to some table catalog and to table
+ * metadata itself.
+ *
+ * [[RelationsWithResolvedMetadata]] is a map from relation ID to the relations with resolved
+ * metadata. It's produced by [[resolve]] and is used later in [[Resolver]] to replace
+ * [[UnresolvedRelation]]s.
+ *
+ * This object is one-shot per SQL query or DataFrame program resolution.
+ */
+class MetadataResolver(
+ override val catalogManager: CatalogManager,
+ override val relationResolution: RelationResolution,
+ override val extensions: Seq[ResolverExtension] = Seq.empty)
+ extends RelationMetadataProvider
+ with DelegatesResolutionToExtensions {
+ override val relationsWithResolvedMetadata = new RelationsWithResolvedMetadata
+
+ /**
+ * Resolves the relation metadata for `unresolvedPlan`. Usually this involves several blocking
+ * calls for the [[UnresolvedRelation]]s present in that tree. During the `unresolvedPlan`
+ * traversal we fill [[relationsWithResolvedMetadata]] with resolved metadata by relation id.
+ * This map will be used to resolve the plan in single-pass by the [[Resolver]] using
+ * [[getRelationWithResolvedMetadata]]. If the generic metadata resolution using
+ * [[RelationResolution]] wasn't successful, we resort to using [[extensions]].
+ * Otherwise, we fail with an exception.
+ */
+ def resolve(unresolvedPlan: LogicalPlan): Unit = {
+ traverseLogicalPlanTree(unresolvedPlan) { unresolvedOperator =>
+ unresolvedOperator match {
+ case unresolvedRelation: UnresolvedRelation =>
+ val relationId = relationIdFromUnresolvedRelation(unresolvedRelation)
+
+ if (!relationsWithResolvedMetadata.containsKey(relationId)) {
+ val relationWithResolvedMetadata = resolveRelation(unresolvedRelation).orElse {
+ // In case the generic metadata resolution returned `None`, we try to check if any
+ // of the [[extensions]] matches this `unresolvedRelation`, and resolve it using
+ // that extension.
+ tryDelegateResolutionToExtension(unresolvedRelation)
+ }
+
+ relationWithResolvedMetadata match {
+ case Some(relationWithResolvedMetadata) =>
+ relationsWithResolvedMetadata.put(
+ relationId,
+ relationWithResolvedMetadata
+ )
+ case None =>
+ withPosition(unresolvedRelation) {
+ unresolvedRelation.tableNotFound(unresolvedRelation.multipartIdentifier)
+ }
+ }
+ }
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Resolves the metadata for the given unresolved relation and returns a relation with the
+ * resolved metadata. This method is blocking.
+ */
+ private def resolveRelation(unresolvedRelation: UnresolvedRelation): Option[LogicalPlan] =
+ AnalysisHelper.allowInvokingTransformsInAnalyzer {
+ relationResolution.resolveRelation(
+ u = unresolvedRelation
+ )
+ }
+
+ /**
+ * Traverse the logical plan tree from `root` in a pre-order DFS manner and apply `visitor` to
+ * each node.
+ */
+ private def traverseLogicalPlanTree(root: LogicalPlan)(visitor: LogicalPlan => Unit) = {
+ val stack = new ArrayDeque[Either[LogicalPlan, Expression]]
+ stack.push(Left(root))
+
+ while (!stack.isEmpty) {
+ stack.pop() match {
+ case Left(logicalPlan) =>
+ visitor(logicalPlan)
+
+ for (child <- logicalPlan.children) {
+ stack.push(Left(child))
+ }
+ for (expression <- logicalPlan.expressions) {
+ stack.push(Right(expression))
+ }
+ case Right(expression) =>
+ for (child <- expression.children) {
+ stack.push(Right(child))
+ }
+
+ expression match {
+ case planExpression: PlanExpression[_] =>
+ planExpression.plan match {
+ case plan: LogicalPlan =>
+ stack.push(Left(plan))
+ case _ =>
+ }
+ case _ =>
+ }
+ }
+ }
+ }
+}
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
new file mode 100644
index 000000000000..8abf4e04b883
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/NameScope.scala
@@ -0,0 +1,393 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import java.util.{ArrayDeque, ArrayList, HashSet}
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.{Resolver => NameComparator, UnresolvedStar}
+import org.apache.spark.sql.catalyst.expressions.{
+ Alias,
+ Attribute,
+ AttributeSeq,
+ Expression,
+ NamedExpression
+}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+
+/**
+ * The [[NameScope]] is used during the analysis to control the visibility of names: plan names
+ * and output attributes. New [[NameScope]] can be created both in the [[Resolver]] and in
+ * the [[ExpressionResolver]] using the [[NameScopeStack]] api. The name resolution for identifiers
+ * is case-insensitive.
+ *
+ * In this example:
+ *
+ * {{{
+ * WITH table_1_cte AS (
+ * SELECT
+ * col1,
+ * col2,
+ * col2
+ * FROM
+ * table_1
+ * )
+ * SELECT
+ * table_1_cte.col1,
+ * table_2.col1
+ * FROM
+ * table_1_cte
+ * INNER JOIN
+ * table_2
+ * ON
+ * table_1_cte.col2 = table_2.col3
+ * ;
+ * }}}
+ *
+ * there are two named subplans in the scope: table_1_cte -> [col1, col2, col2] and
+ * table_2 -> [col1, col3].
+ *
+ * State breakout:
+ * - `planOutputs`: list of named plan outputs. Order matters here (e.g. to correctly expand `*`).
+ * Can contain duplicate names, since it's possible to select same column twice, or to select
+ * columns with the same name from different relations. [[OptionalIdentifierMap]] is used here,
+ * since some plans don't have an explicit name, so output attributes from those plans will reside
+ * under the `None` key.
+ * In our example it will be {{{ [(table_1_cte, [col1, col2, col2]), (table_2, [col1, col3])] }}}
+ *
+ * - `planNameToOffset`: mapping from plan output names to their offsets in the `planOutputs` array.
+ * It's used to lookup attributes by plan output names (multipart names are not supported yet).
+ * In our example it will be {{{ [table_1_cte -> 0, table_2 -> 1] }}}
+ */
+class NameScope extends SQLConfHelper {
+ private val planOutputs = new ArrayList[PlanOutput]()
+ private val planNameToOffset = new OptionalIdentifierMap[Int]
+ private val nameComparator: NameComparator = conf.resolver
+ private val existingAliases = new HashSet[String]
+
+ /**
+ * Register the named plan output in this [[NameScope]]. The named plan is usually a
+ * [[NamedRelation]]. `attributes` sequence can contain duplicate names both for this named plan
+ * and for the scope in general, despite the fact that their further resolution _may_ throw an
+ * error in case of ambiguous reference. After calling this method, the code can lookup the
+ * attributes using `get*` methods of this [[NameScope]].
+ *
+ * Duplicate plan names are merged into the same [[PlanOutput]]. For example, this query:
+ *
+ * {{{ SELECT t.* FROM (SELECT * FROM VALUES (1)) as t, (SELECT * FROM VALUES (2)) as t; }}}
+ *
+ * will have the following output schema:
+ *
+ * {{{ [col1, col1] }}}
+ *
+ * Same logic applies for the unnamed plan outputs. This query:
+ *
+ * {{{ SELECT * FROM (SELECT * FROM VALUES (1)), (SELECT * FROM VALUES (2)); }}}
+ *
+ * will have the same output schema:
+ *
+ * {{{ [col1, col1] }}}
+ *
+ * @param name The name of this named plan.
+ * @param attributes The output of this named plan. Can contain duplicate names.
+ */
+ def update(name: String, attributes: Seq[Attribute]): Unit = {
+ update(attributes, Some(name))
+ }
+
+ /**
+ * Register the unnamed plan output in this [[NameScope]]. Some examples of the unnamed plan are
+ * [[Project]] and [[Aggregate]].
+ *
+ * See the [[update]] method for more details.
+ *
+ * @param attributes The output of the unnamed plan. Can contain duplicate names.
+ */
+ def +=(attributes: Seq[Attribute]): Unit = {
+ update(attributes)
+ }
+
+ /**
+ * Get all the attributes from all the plans registered in this [[NameScope]]. The output can
+ * contain duplicate names. This is used for star (`*`) resolution.
+ */
+ def getAllAttributes: Seq[Attribute] = {
+ val attributes = new mutable.ArrayBuffer[Attribute]
+
+ planOutputs.forEach(planOutput => {
+ attributes.appendAll(planOutput.attributes)
+ })
+
+ attributes.toSeq
+ }
+
+ /**
+ * Expand the [[UnresolvedStar]] using `planOutputs`. The expected use case for this method is
+ * star expansion inside [[Project]]. Since [[Project]] has only one child, we assert that the
+ * size of `planOutputs` is 1, otherwise the query is malformed.
+ *
+ * Some examples of queries with a star:
+ *
+ * - Star without a target:
+ * {{{ SELECT * FROM VALUES (1, 2, 3) AS t(a, b, c); }}}
+ * - Star with a multipart name target:
+ * {{{ SELECT catalog1.database1.table1.* FROM catalog1.database1.table1; }}}
+ * - Star with a struct target:
+ * {{{ SELECT d.* FROM VALUES (named_struct('a', 1, 'b', 2)) AS t(d); }}}
+ * - Star as an argument to a function:
+ * {{{ SELECT concat_ws('', *) AS result FROM VALUES (1, 2, 3) AS t(a, b, c); }}}
+ *
+ * It is resolved by correctly resolving the star qualifier.
+ * Please check [[UnresolvedStarBase.expandStar]] for more details.
+ *
+ * @param unresolvedStar [[UnresolvedStar]] to expand.
+ * @return The output of a plan expanded from the star.
+ */
+ def expandStar(unresolvedStar: UnresolvedStar): Seq[NamedExpression] = {
+ if (planOutputs.size != 1) {
+ throw QueryCompilationErrors.invalidStarUsageError("query", Seq(unresolvedStar))
+ }
+
+ planOutputs.get(0).expandStar(unresolvedStar)
+ }
+
+ /**
+ * Get all matched attributes by a multipart name. It returns [[Attribute]]s when we resolve a
+ * simple column or an alias name from a lower operator. However this function can also return
+ * [[Alias]]es in case we access a struct field or a map value using some key.
+ *
+ * Example that contains those major use-cases:
+ *
+ * {{{
+ * SELECT col1, a, col2.field, col3.struct.field, col4.key
+ * FROM (SELECT *, col5 AS a FROM t);
+ * }}}
+ *
+ * has a Project list that looks like this:
+ *
+ * {{{
+ * AttributeReference(col1),
+ * AttributeReference(a),
+ * Alias(col2.field, field),
+ * Alias(col3.struct.field, field),
+ * Alias(col4[CAST(key AS INT)], key)
+ * }}}
+ *
+ * Also, see [[AttributeSeq.resolve]] for more details.
+ *
+ * Since there can be several identical attribute names for several named plans, this function
+ * can return multiple values:
+ * - 0 values: No matched attributes
+ * - 1 value: Unique attribute matched
+ * - 1+ values: Ambiguity, several attributes matched
+ *
+ * One example of a query with an attribute that has a multipart name:
+ *
+ * {{{ SELECT catalog1.database1.table1.col1 FROM catalog1.database1.table1; }}}
+ *
+ * @param multipartName Multipart attribute name. Can be of several forms:
+ * - `catalog.database.table.column`
+ * - `database.table.column`
+ * - `table.column`
+ * - `column`
+ * @return All the attributes matched by the `multipartName`, encapsulated in a [[NameTarget]].
+ */
+ def matchMultipartName(multipartName: Seq[String]): NameTarget = {
+ val candidates = new mutable.ArrayBuffer[Expression]
+ val allAttributes = new mutable.ArrayBuffer[Attribute]
+ var aliasName: Option[String] = None
+
+ planOutputs.forEach(planOutput => {
+ allAttributes.appendAll(planOutput.attributes)
+ val nameTarget = planOutput.matchMultipartName(multipartName)
+ if (nameTarget.aliasName.isDefined) {
+ aliasName = nameTarget.aliasName
+ }
+ candidates.appendAll(nameTarget.candidates)
+ })
+
+ NameTarget(candidates.toSeq, aliasName, allAttributes.toSeq)
+ }
+
+ /**
+ * Add an alias, by name, to the list of existing aliases.
+ */
+ def addAlias(aliasName: String): Unit = existingAliases.add(aliasName.toLowerCase())
+
+ /**
+ * Returns whether an alias exists in the current scope.
+ */
+ def isExistingAlias(aliasName: String): Boolean =
+ existingAliases.contains(aliasName.toLowerCase())
+
+ private def update(attributes: Seq[Attribute], name: Option[String] = None): Unit = {
+ planNameToOffset.get(name) match {
+ case Some(index) =>
+ val prevPlanOutput = planOutputs.get(index)
+ planOutputs.set(
+ index,
+ new PlanOutput(prevPlanOutput.attributes ++ attributes, name, nameComparator)
+ )
+ case None =>
+ val index = planOutputs.size
+ planOutputs.add(new PlanOutput(attributes, name, nameComparator))
+ planNameToOffset += (name -> index)
+ }
+ }
+}
+
+/**
+ * The [[NameScopeStack]] is a stack of [[NameScope]]s managed by the [[Resolver]] and the
+ * [[ExpressionResolver]]. Usually a top scope is used for name resolution, but in case of
+ * correlated subqueries we can lookup names in the parent scopes. Low-level scope creation is
+ * managed internally, and only high-level api like [[withNewScope]] is available to the resolvers.
+ * Freshly-created [[NameScopeStack]] contains an empty root [[NameScope]].
+ */
+class NameScopeStack extends SQLConfHelper {
+ private val stack = new ArrayDeque[NameScope]
+ push()
+
+ /**
+ * Get the top scope, which is a default choice for name resolution.
+ */
+ def top: NameScope = {
+ stack.peek()
+ }
+
+ /**
+ * Completely overwrite the top scope state with a named plan output.
+ *
+ * See [[NameScope.update]] for more details.
+ */
+ def overwriteTop(name: String, attributes: Seq[Attribute]): Unit = {
+ val newScope = new NameScope
+ newScope.update(name, attributes)
+
+ stack.pop()
+ stack.push(newScope)
+ }
+
+ /**
+ * Completely overwrite the top scope state with an unnamed plan output.
+ *
+ * See [[NameScope.+=]] for more details.
+ */
+ def overwriteTop(attributes: Seq[Attribute]): Unit = {
+ val newScope = new NameScope
+ newScope += attributes
+
+ stack.pop()
+ stack.push(newScope)
+ }
+
+ /**
+ * Execute `body` in a context of a fresh scope. It's used during the [[Project]] or the
+ * [[Aggregate]] resolution to avoid calling [[push]] and [[pop]] explicitly.
+ */
+ def withNewScope[R](body: => R): R = {
+ push()
+ try {
+ body
+ } finally {
+ pop()
+ }
+ }
+
+ /**
+ * Push a new scope to the stack. Introduced by the [[Project]] or the [[Aggregate]].
+ */
+ private def push(): Unit = {
+ stack.push(new NameScope)
+ }
+
+ /**
+ * Pop a scope from the stack. Called when the resolution process for the pushed scope is done.
+ */
+ private def pop(): Unit = {
+ stack.pop()
+ }
+}
+
+/**
+ * [[PlanOutput]] represents a sequence of attributes from a plan ([[NamedRelation]], [[Project]],
+ * [[Aggregate]], etc).
+ *
+ * It is created from `attributes`, which is an output of a named plan, optional plan `name` and a
+ * resolver provided by the [[NameScopeStack]].
+ *
+ * @param attributes Plan output. Can contain duplicate names.
+ * @param name Plan name. Non-empty for named plans like [[NamedRelation]] or [[SubqueryAlias]],
+ * `None` otherwise.
+ */
+class PlanOutput(
+ val attributes: Seq[Attribute],
+ val name: Option[String],
+ val nameComparator: NameComparator) {
+
+ /**
+ * attributesForResolution is an [[AttributeSeq]] that is used for resolution of
+ * multipart attribute names. It's created from the `attributes` when [[NameScope]] is updated.
+ */
+ private val attributesForResolution: AttributeSeq =
+ AttributeSeq.fromNormalOutput(attributes)
+
+ /**
+ * Find attributes by the multipart name.
+ *
+ * See [[NameScope.matchMultipartName]] for more details.
+ *
+ * @param multipartName Multipart attribute name.
+ * @return Matched attributes or [[Seq.empty]] otherwise.
+ */
+ def matchMultipartName(multipartName: Seq[String]): NameTarget = {
+ val (candidates, nestedFields) =
+ attributesForResolution.getCandidatesForResolution(multipartName, nameComparator)
+ val resolvedCandidates = attributesForResolution.resolveCandidates(
+ multipartName,
+ nameComparator,
+ candidates,
+ nestedFields
+ )
+ resolvedCandidates match {
+ case Seq(Alias(child, aliasName)) =>
+ NameTarget(Seq(child), Some(aliasName))
+ case other =>
+ NameTarget(other, None)
+ }
+ }
+
+ /**
+ * Method to expand an unresolved star. See [[NameScope.expandStar]] for more details.
+ *
+ * @param unresolvedStar Star to resolve.
+ * @return Attributes expanded from the star.
+ */
+ def expandStar(unresolvedStar: UnresolvedStar): Seq[NamedExpression] = {
+ unresolvedStar.expandStar(
+ childOperatorOutput = attributes,
+ childOperatorMetadataOutput = Seq.empty,
+ resolve =
+ (nameParts, nameComparator) => attributesForResolution.resolve(nameParts, nameComparator),
+ suggestedAttributes = attributes,
+ resolver = nameComparator,
+ cleanupNestedAliasesDuringStructExpansion = true
+ )
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/NameTarget.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/NameTarget.scala
new file mode 100644
index 000000000000..3b31c9b1a911
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/NameTarget.scala
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression}
+import org.apache.spark.sql.catalyst.util.StringUtils.orderSuggestedIdentifiersBySimilarity
+import org.apache.spark.sql.errors.QueryCompilationErrors
+
+/**
+ * Class that represents results of name resolution or star expansion. It encapsulates:
+ * - `candidates` - A list of candidates that are possible matches for a given name.
+ * - `aliasName` - If the candidates size is 1 and it's type is `ExtractValue` (which means that
+ * it's a recursive type), then the `aliasName` should be the name with which the candidate is
+ * aliased. Otherwise, `aliasName` should be `None`.
+ * - `allAttributes` - A list of all attributes which is used to generate suggestions for
+ * unresolved column error.
+ *
+ * Example:
+ *
+ * - Attribute resolution:
+ * {{{ SELECT col1 FROM VALUES (1); }}} will have a [[NameTarget]] with a single candidate `col1`.
+ * `aliasName` would be `None` in this case because the column is not of recursive type.
+ *
+ * - Recursive attribute resolution:
+ * {{{ SELECT col1.col1 FROM VALUES(STRUCT(1,2), 3) }}} will have a [[NameTarget]] with a
+ * single candidate `col1` and an `aliasName` of `Some("col1")`.
+ */
+case class NameTarget(
+ candidates: Seq[Expression],
+ aliasName: Option[String] = None,
+ allAttributes: Seq[Attribute] = Seq.empty) {
+
+ /**
+ * Picks a candidate from the list of candidates based on the given unresolved attribute.
+ * Its behavior is as follows (based on the number of candidates):
+ *
+ * - If there is only one candidate, it will be returned.
+ *
+ * - If there are multiple candidates, an ambiguous reference error will be thrown.
+ *
+ * - If there are no candidates, an unresolved column error will be thrown.
+ */
+ def pickCandidate(unresolvedAttribute: UnresolvedAttribute): Expression = {
+ candidates match {
+ case Seq() =>
+ throwUnresolvedColumnError(unresolvedAttribute)
+ case Seq(candidate) =>
+ candidate
+ case _ =>
+ throw QueryCompilationErrors.ambiguousReferenceError(
+ unresolvedAttribute.name,
+ candidates.collect { case attribute: AttributeReference => attribute }
+ )
+ }
+ }
+
+ private def throwUnresolvedColumnError(unresolvedAttribute: UnresolvedAttribute): Nothing =
+ throw QueryCompilationErrors.unresolvedColumnError(
+ unresolvedAttribute.name,
+ proposal = orderSuggestedIdentifiersBySimilarity(
+ unresolvedAttribute.name,
+ candidates = allAttributes.map(attribute => attribute.qualifier :+ attribute.name)
+ )
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/PlanLogger.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/PlanLogger.scala
new file mode 100644
index 000000000000..fcf1eab0c04a
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/PlanLogger.scala
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.internal.{Logging, MDC, MessageWithContext}
+import org.apache.spark.internal.LogKeys.QUERY_PLAN
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.sideBySide
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * [[PlanLogger]] is used by the [[Resolver]] to log intermediate resolution results.
+ */
+class PlanLogger extends Logging {
+ private val logLevel = SQLConf.get.planChangeLogLevel
+
+ /**
+ * Logs the transition from the `unresolvedPlan` to the `resolvedPlan`.
+ */
+ def log(unresolvedPlan: LogicalPlan, resolvedPlan: LogicalPlan): Unit = {
+ logBasedOnLevel(() => createMessage(unresolvedPlan, resolvedPlan))
+ }
+
+ private def createMessage(
+ unresolvedPlan: LogicalPlan,
+ resolvedPlan: LogicalPlan): MessageWithContext =
+ log"""
+ |=== Unresolved/resolved operator subtree ===
+ |${MDC(
+ QUERY_PLAN,
+ sideBySide(unresolvedPlan.treeString, resolvedPlan.treeString).mkString("\n")
+ )}
+ """.stripMargin
+
+ private def logBasedOnLevel(createMessage: () => MessageWithContext): Unit = logLevel match {
+ case "TRACE" => logTrace(createMessage().message)
+ case "DEBUG" => logDebug(createMessage().message)
+ case "INFO" => logInfo(createMessage())
+ case "WARN" => logWarning(createMessage())
+ case "ERROR" => logError(createMessage())
+ case _ => logTrace(createMessage().message)
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/PredicateResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/PredicateResolver.scala
new file mode 100644
index 000000000000..d94559496d04
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/PredicateResolver.scala
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.{
+ AnsiStringPromotionTypeCoercion,
+ AnsiTypeCoercion,
+ ApplyCharTypePaddingHelper,
+ BooleanEqualityTypeCoercion,
+ CollationTypeCoercion,
+ DecimalPrecisionTypeCoercion,
+ DivisionTypeCoercion,
+ IntegralDivisionTypeCoercion,
+ StringPromotionTypeCoercion,
+ TypeCoercion
+}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Predicate, StringRPad}
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Resolver class for resolving all [[Predicate]] expressions. Recursively resolves all children
+ * and applies selected type coercions to the expression.
+ */
+class PredicateResolver(
+ expressionResolver: ExpressionResolver,
+ timezoneAwareExpressionResolver: TimezoneAwareExpressionResolver)
+ extends TreeNodeResolver[Predicate, Expression]
+ with ResolvesExpressionChildren {
+
+ private val typeCoercionRules = if (conf.ansiEnabled) {
+ PredicateResolver.ANSI_TYPE_COERCION_RULES
+ } else {
+ PredicateResolver.TYPE_COERCION_RULES
+ }
+ private val typeCoercionResolver =
+ new TypeCoercionResolver(timezoneAwareExpressionResolver, typeCoercionRules)
+
+ override def resolve(unresolvedPredicate: Predicate): Expression = {
+ val predicateWithResolvedChildren =
+ withResolvedChildren(unresolvedPredicate, expressionResolver.resolve)
+ val predicateWithTypeCoercion = typeCoercionResolver.resolve(predicateWithResolvedChildren)
+ val predicateWithCharTypePadding = {
+ ApplyCharTypePaddingHelper.singleNodePaddingForStringComparison(
+ predicateWithTypeCoercion,
+ !conf.getConf(SQLConf.LEGACY_NO_CHAR_PADDING_IN_PREDICATE)
+ )
+ }
+ predicateWithCharTypePadding.children.collectFirst {
+ case rpad: StringRPad => rpad
+ } match {
+ // In the fixed-point Analyzer [[ApplyCharTypePadding]] is called after [[ResolveAliases]]
+ // and therefore padding doesn't affect the alias. In single-pass resolver we need to call
+ // this code before we resolve the alias, which will cause the alias to include the pad in
+ // its name:
+ //
+ // fixed-point:
+ // expression: rpad('12', 3, ' ') = '12 '
+ // alias: '12' = '12 '
+ //
+ // single-pass:
+ // expression: rpad('12', 3, ' ') = '12 '
+ // alias: rpad('12', 3, ' ') = '12 '
+ //
+ // Disabling this case until the aliasing is fixed.
+ case Some(_) => throw new ExplicitlyUnsupportedResolverFeature("CharTypePaddingAliasing")
+
+ case _ => predicateWithCharTypePadding
+ }
+ }
+}
+
+object PredicateResolver {
+ // Ordering in the list of type coercions should be in sync with the list in [[TypeCoercion]].
+ private val TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ CollationTypeCoercion.apply,
+ TypeCoercion.InTypeCoercion.apply,
+ StringPromotionTypeCoercion.apply,
+ DecimalPrecisionTypeCoercion.apply,
+ BooleanEqualityTypeCoercion.apply,
+ DivisionTypeCoercion.apply,
+ IntegralDivisionTypeCoercion.apply,
+ TypeCoercion.ImplicitTypeCoercion.apply,
+ TypeCoercion.DateTimeOperationsTypeCoercion.apply
+ )
+
+ // Ordering in the list of type coercions should be in sync with the list in [[AnsiTypeCoercion]].
+ private val ANSI_TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ CollationTypeCoercion.apply,
+ AnsiTypeCoercion.InTypeCoercion.apply,
+ AnsiStringPromotionTypeCoercion.apply,
+ DecimalPrecisionTypeCoercion.apply,
+ DivisionTypeCoercion.apply,
+ IntegralDivisionTypeCoercion.apply,
+ AnsiTypeCoercion.ImplicitTypeCoercion.apply,
+ AnsiTypeCoercion.AnsiDateTimeOperationsTypeCoercion.apply
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ProducesUnresolvedSubtree.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ProducesUnresolvedSubtree.scala
new file mode 100644
index 000000000000..8d85804a9363
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ProducesUnresolvedSubtree.scala
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+
+/**
+ * A mixin trait for expression resolvers that as part of their resolution, replace single node
+ * with a subtree of nodes. This step is necessary because the underlying legacy code that is being
+ * called produces partially-unresolved subtrees. In order to resolve the subtree a callback
+ * resolver is called recursively. This callback must ensure that no node is resolved twice in
+ * order to not break the single-pass invariant. This is done by tagging the limits of this
+ * traversal with [[ExpressionResolver.SINGLE_PASS_SUBTREE_BOUNDARY]] tag. This tag is applied to
+ * the original expression's children, which are guaranteed to be resolved at the time of given
+ * expression's resolution. When callback resolver encounters the node that is tagged, it should
+ * return identity instead of trying to resolve it.
+ */
+trait ProducesUnresolvedSubtree extends ResolvesExpressionChildren {
+
+ /**
+ * Helper method used to resolve a subtree that is generated as part of the resolution of some
+ * node. Method ensures that the downwards traversal never visits previously resolved nodes by
+ * tracking the limits of the traversal with a tag. Invokes a resolver callback to resolve
+ * children, but DOES NOT resolve the root of the subtree.
+ */
+ protected def withResolvedSubtree(
+ expression: Expression,
+ expressionResolver: Expression => Expression)(body: => Expression): Expression = {
+ expression.children.foreach { child =>
+ child.setTagValue(ExpressionResolver.SINGLE_PASS_SUBTREE_BOUNDARY, ())
+ }
+
+ val result = body
+
+ withResolvedChildren(result, expressionResolver)
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RelationId.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RelationId.scala
new file mode 100644
index 000000000000..eab97dd8345c
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RelationId.scala
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * The [[RelationId]] is a unique identifier for a relation. It is used to lookup the relations
+ * which were processed by the [[MetadataResolver]] to substitute the unresolved relations in single
+ * pass during the analysis phase.
+ */
+case class RelationId(
+ multipartIdentifier: Seq[String],
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty,
+ isStreaming: Boolean = false
+)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RelationMetadataProvider.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RelationMetadataProvider.scala
new file mode 100644
index 000000000000..cf352842fd10
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RelationMetadataProvider.scala
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import java.util.HashMap
+
+import org.apache.spark.sql.catalyst.analysis.{withPosition, RelationResolution, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.connector.catalog.LookupCatalog
+import org.apache.spark.util.ArrayImplicits._
+
+/**
+ * [[RelationMetadataProvider]] provides relations with resolved metadata based on the
+ * corresponding [[UnresolvedRelation]]s. It is used by [[Resolver]] to replace
+ * [[UnresolvedRelation]] with a specific [[LogicalPlan]] with resolved metadata, e.g. with
+ * [[UnresolvedCatalogRelation]] or [[View]].
+ */
+trait RelationMetadataProvider extends LookupCatalog {
+ type RelationsWithResolvedMetadata = HashMap[RelationId, LogicalPlan]
+
+ /**
+ * [[relationResolution]] is used by the [[RelationMetadataProvider]] to expand relation
+ * identifiers in [[relationIdFromUnresolvedRelation]].
+ */
+ protected val relationResolution: RelationResolution
+
+ /**
+ * [[relationsWithResolvedMetadata]] is a map from relation ID to the specific [[LogicalPlan]]
+ * with resolved metadata, like [[UnresolvedCatalogRelation]] or [[View]]. It's filled by the
+ * specific [[RelationMetadataProvider]] implementation and is queried in
+ * [[getRelationWithResolvedMetadata]].
+ */
+ protected val relationsWithResolvedMetadata: RelationsWithResolvedMetadata
+
+ /**
+ * Get the [[LogicalPlan]] with resolved metadata for the given [[UnresolvedRelation]].
+ *
+ * [[java.util.HashMap]] returns `null` if the key is not found, so we wrap it in an [[Option]].
+ */
+ def getRelationWithResolvedMetadata(
+ unresolvedRelation: UnresolvedRelation): Option[LogicalPlan] = {
+ Option(
+ relationsWithResolvedMetadata.get(
+ relationIdFromUnresolvedRelation(unresolvedRelation)
+ )
+ )
+ }
+
+ /**
+ * Returns the [[RelationId]] for the given [[UnresolvedRelation]]. Here we use
+ * [[relationResolution]] to expand the [[UnresolvedRelation]] identifier fully, so that our
+ * [[RelationId]] uniquely identifies the [[unresolvedRelation]].
+ *
+ * This method is public, because it's used in [[MetadataResolverSuite]].
+ */
+ def relationIdFromUnresolvedRelation(unresolvedRelation: UnresolvedRelation): RelationId = {
+ relationResolution.expandIdentifier(unresolvedRelation.multipartIdentifier) match {
+ case CatalogAndIdentifier(catalog, ident) =>
+ RelationId(
+ multipartIdentifier =
+ Seq(catalog.name()) ++ ident.namespace().toImmutableArraySeq ++ Seq(ident.name()),
+ options = unresolvedRelation.options,
+ isStreaming = unresolvedRelation.isStreaming
+ )
+ case _ =>
+ withPosition(unresolvedRelation) {
+ unresolvedRelation.tableNotFound(unresolvedRelation.multipartIdentifier)
+ }
+ }
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolutionValidator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolutionValidator.scala
new file mode 100644
index 000000000000..6c4de2e6e58d
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolutionValidator.scala
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, ResolvedInlineTable}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference}
+import org.apache.spark.sql.catalyst.plans.logical.{
+ Filter,
+ GlobalLimit,
+ LocalLimit,
+ LocalRelation,
+ LogicalPlan,
+ OneRowRelation,
+ Project,
+ SubqueryAlias
+}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * The [[ResolutionValidator]] performs the validation work after the logical plan tree is
+ * resolved by the [[Resolver]]. Each `resolve*` method in the [[Resolver]] must
+ * have its `validate*` counterpart in the [[ResolutionValidator]]. The validation code asserts the
+ * conditions that must never be false no matter which SQL query or DataFrame program was provided.
+ * The validation approach is single-pass, post-order, complementary to the resolution process.
+ */
+class ResolutionValidator {
+ private val expressionResolutionValidator = new ExpressionResolutionValidator(this)
+
+ private[resolver] var attributeScopeStack = new AttributeScopeStack
+
+ /**
+ * Validate the resolved logical `plan` - assert invariants that should never be false no
+ * matter which SQL query or DataFrame program was provided. New operators must be added here as
+ * soon as [[Resolver]] supports them. We check this by throwing an exception for
+ * unknown operators.
+ */
+ def validatePlan(plan: LogicalPlan): Unit = wrapErrors(plan) {
+ validate(plan)
+ }
+
+ private def validate(operator: LogicalPlan): Unit = {
+ operator match {
+ case project: Project =>
+ validateProject(project)
+ case filter: Filter =>
+ validateFilter(filter)
+ case subqueryAlias: SubqueryAlias =>
+ validateSubqueryAlias(subqueryAlias)
+ case globalLimit: GlobalLimit =>
+ validateGlobalLimit(globalLimit)
+ case localLimit: LocalLimit =>
+ validateLocalLimit(localLimit)
+ case inlineTable: ResolvedInlineTable =>
+ validateInlineTable(inlineTable)
+ case localRelation: LocalRelation =>
+ validateRelation(localRelation)
+ case oneRowRelation: OneRowRelation =>
+ validateRelation(oneRowRelation)
+ // [[LogicalRelation]], [[HiveTableRelation]] and other specific relations can't be imported
+ // because of a potential circular dependency, so we match a generic Catalyst
+ // [[MultiInstanceRelation]] instead.
+ case multiInstanceRelation: MultiInstanceRelation =>
+ validateRelation(multiInstanceRelation)
+ }
+ }
+
+ private def validateProject(project: Project): Unit = {
+ attributeScopeStack.withNewScope {
+ validate(project.child)
+ expressionResolutionValidator.validateProjectList(project.projectList)
+ }
+
+ handleOperatorOutput(project)
+ }
+
+ private def validateFilter(filter: Filter): Unit = {
+ validate(filter.child)
+
+ assert(
+ filter.condition.dataType == BooleanType,
+ s"Output type of a filter must be a boolean, but got: ${filter.condition.dataType.typeName}"
+ )
+ expressionResolutionValidator.validate(filter.condition)
+ }
+
+ private def validateSubqueryAlias(subqueryAlias: SubqueryAlias): Unit = {
+ validate(subqueryAlias.child)
+
+ handleOperatorOutput(subqueryAlias)
+ }
+
+ private def validateGlobalLimit(globalLimit: GlobalLimit): Unit = {
+ validate(globalLimit.child)
+ expressionResolutionValidator.validate(globalLimit.limitExpr)
+ }
+
+ private def validateLocalLimit(localLimit: LocalLimit): Unit = {
+ validate(localLimit.child)
+ expressionResolutionValidator.validate(localLimit.limitExpr)
+ }
+
+ private def validateInlineTable(inlineTable: ResolvedInlineTable): Unit = {
+ inlineTable.rows.foreach(row => {
+ row.foreach(expression => {
+ expressionResolutionValidator.validate(expression)
+ })
+ })
+
+ handleOperatorOutput(inlineTable)
+ }
+
+ private def validateRelation(relation: LogicalPlan): Unit = {
+ handleOperatorOutput(relation)
+ }
+
+ private def handleOperatorOutput(operator: LogicalPlan): Unit = {
+ attributeScopeStack.overwriteTop(operator.output)
+
+ operator.output.foreach(attribute => {
+ assert(
+ attribute.isInstanceOf[AttributeReference],
+ s"Output of an operator must be a reference to an attribute, but got: " +
+ s"${attribute.getClass.getSimpleName}"
+ )
+ expressionResolutionValidator.validate(attribute)
+ })
+ }
+
+ private def wrapErrors[R](plan: LogicalPlan)(body: => R): Unit = {
+ try {
+ body
+ } catch {
+ case ex: Throwable =>
+ throw QueryCompilationErrors.resolutionValidationError(ex, plan)
+ }
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/Resolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/Resolver.scala
new file mode 100644
index 000000000000..b0e6828a97a0
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/Resolver.scala
@@ -0,0 +1,364 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.EvaluateUnresolvedInlineTable
+import org.apache.spark.sql.catalyst.analysis.{
+ withPosition,
+ FunctionResolution,
+ NamedRelation,
+ RelationResolution,
+ ResolvedInlineTable,
+ UnresolvedInlineTable,
+ UnresolvedRelation
+}
+import org.apache.spark.sql.catalyst.plans.logical.{
+ Filter,
+ GlobalLimit,
+ LocalLimit,
+ LocalRelation,
+ LogicalPlan,
+ OneRowRelation,
+ Project,
+ SubqueryAlias
+}
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase}
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * The Resolver implements a single-pass bottom-up analysis algorithm in the Catalyst.
+ *
+ * The functions here generally traverse the [[LogicalPlan]] nodes recursively,
+ * constructing and returning the resolved [[LogicalPlan]] nodes bottom-up.
+ * This is the primary entry point for implementing SQL and DataFrame plan analysis,
+ * wherein the [[resolve]] method accepts a fully unresolved [[LogicalPlan]] and returns
+ * a fully resolved [[LogicalPlan]] in response with all data types and attribute
+ * reference ID assigned for valid requests. This resolver also takes responsibility
+ * to detect any errors in the initial SQL query or DataFrame and return appropriate
+ * error messages including precise parse locations wherever possible.
+ *
+ * The Resolver is a one-shot object per each SQL/DataFrame logical plan, the calling code must
+ * re-create it for every new analysis run.
+ *
+ * @param catalogManager [[CatalogManager]] for relation and identifier resolution.
+ * @param extensions A list of [[ResolverExtension]] that can resolve external operators.
+ */
+class Resolver(
+ catalogManager: CatalogManager,
+ override val extensions: Seq[ResolverExtension] = Seq.empty,
+ metadataResolverExtensions: Seq[ResolverExtension] = Seq.empty)
+ extends TreeNodeResolver[LogicalPlan, LogicalPlan]
+ with QueryErrorsBase
+ with ResolvesOperatorChildren
+ with TracksResolvedNodes[LogicalPlan]
+ with DelegatesResolutionToExtensions {
+ private val scopes = new NameScopeStack
+ private val relationResolution = Resolver.createRelationResolution(catalogManager)
+ private val functionResolution = new FunctionResolution(catalogManager, relationResolution)
+ private val expressionResolver = new ExpressionResolver(this, scopes, functionResolution)
+ private val limitExpressionResolver = new LimitExpressionResolver(expressionResolver)
+ private val planLogger = new PlanLogger
+
+ /**
+ * [[relationMetadataProvider]] is used to resolve metadata for relations. It's initialized with
+ * the default implementation [[MetadataResolver]] here and is called in
+ * [[lookupMetadataAndResolve]] on the unresolved logical plan to visit it (both operators and
+ * expressions) to resolve the metadata and populate its internal state. It's later queried by
+ * [[resolveRelation]] to get the plan with resolved metadata (for example, a [[View]] or an
+ * [[UnresolvedCatalogRelation]]) based on the [[UnresolvedRelation]].
+ *
+ * If the [[AnalyzerBridgeState]] is provided, we reset this provider to the
+ * [[BridgedRelationMetadataProvider]] and later stick to it forever without resorting to the
+ * actual blocking metadata resolution.
+ */
+ private var relationMetadataProvider: RelationMetadataProvider = new MetadataResolver(
+ catalogManager,
+ relationResolution,
+ metadataResolverExtensions
+ )
+
+ /**
+ * This method is an analysis entry point. It resolves the metadata and invokes [[resolve]],
+ * which does most of the analysis work.
+ */
+ def lookupMetadataAndResolve(
+ unresolvedPlan: LogicalPlan,
+ analyzerBridgeState: Option[AnalyzerBridgeState] = None): LogicalPlan = {
+ relationMetadataProvider = analyzerBridgeState match {
+ case Some(analyzerBridgeState) =>
+ new BridgedRelationMetadataProvider(
+ catalogManager,
+ relationResolution,
+ analyzerBridgeState
+ )
+ case None =>
+ relationMetadataProvider
+ }
+
+ relationMetadataProvider match {
+ case metadataResolver: MetadataResolver =>
+ metadataResolver.resolve(unresolvedPlan)
+ case _ =>
+ }
+
+ resolve(unresolvedPlan)
+ }
+
+ /**
+ * This method takes an unresolved [[LogicalPlan]] and chooses the right `resolve*` method using
+ * pattern matching on the `unresolvedPlan` type. This pattern matching enumerates all the
+ * operator node types that are supported by the single-pass analysis.
+ *
+ * When developers introduce a new unresolved node type to the Catalyst, they should implement
+ * a corresponding `resolve*` method in the [[Resolver]] and add it to this pattern match
+ * list.
+ *
+ * [[resolve]] will be called recursively during the unresolved plan traversal eventually
+ * producing a fully resolved plan or a descriptive error message.
+ */
+ override def resolve(unresolvedPlan: LogicalPlan): LogicalPlan = {
+ throwIfNodeWasResolvedEarlier(unresolvedPlan)
+
+ val resolvedPlan =
+ unresolvedPlan match {
+ case unresolvedProject: Project =>
+ resolveProject(unresolvedProject)
+ case unresolvedFilter: Filter =>
+ resolveFilter(unresolvedFilter)
+ case unresolvedSubqueryAlias: SubqueryAlias =>
+ resolveSubqueryAlias(unresolvedSubqueryAlias)
+ case unresolvedGlobalLimit: GlobalLimit =>
+ resolveGlobalLimit(unresolvedGlobalLimit)
+ case unresolvedLocalLimit: LocalLimit =>
+ resolveLocalLimit(unresolvedLocalLimit)
+ case unresolvedRelation: UnresolvedRelation =>
+ resolveRelation(unresolvedRelation)
+ case unresolvedInlineTable: UnresolvedInlineTable =>
+ resolveInlineTable(unresolvedInlineTable)
+ // See the reason why we have to match both [[LocalRelation]] and [[ResolvedInlineTable]]
+ // in the [[resolveInlineTable]] scaladoc
+ case resolvedInlineTable: ResolvedInlineTable =>
+ updateNameScopeWithPlanOutput(resolvedInlineTable)
+ case localRelation: LocalRelation =>
+ updateNameScopeWithPlanOutput(localRelation)
+ case unresolvedOneRowRelation: OneRowRelation =>
+ updateNameScopeWithPlanOutput(unresolvedOneRowRelation)
+ case _ =>
+ tryDelegateResolutionToExtension(unresolvedPlan).getOrElse {
+ handleUnmatchedOperator(unresolvedPlan)
+ }
+ }
+
+ markNodeAsResolved(resolvedPlan)
+ planLogger.log(unresolvedPlan, resolvedPlan)
+ resolvedPlan
+ }
+
+ /**
+ * [[Project]] introduces a new scope to resolve its subtree and project list expressions. After
+ * those are resolved in the child scope we overwrite current scope with resolved [[Project]]'s
+ * output to expose new names to the parent operators.
+ */
+ private def resolveProject(unresolvedProject: Project): LogicalPlan = {
+ val resolvedProject = scopes.withNewScope {
+ val resolvedChild = resolve(unresolvedProject.child)
+ val resolvedProjectList =
+ expressionResolver.resolveProjectList(unresolvedProject.projectList)
+ Project(resolvedProjectList, resolvedChild)
+ }
+
+ withPosition(unresolvedProject) {
+ scopes.overwriteTop(resolvedProject.output)
+ }
+
+ resolvedProject
+ }
+
+ /**
+ * [[Filter]] has a single child and a single condition and we resolve them in this respective
+ * order.
+ */
+ private def resolveFilter(unresolvedFilter: Filter): LogicalPlan = {
+ val resolvedChild = resolve(unresolvedFilter.child)
+ val resolvedCondition = expressionResolver.resolve(unresolvedFilter.condition)
+
+ val resolvedFilter = Filter(resolvedCondition, resolvedChild)
+ if (resolvedFilter.condition.dataType != BooleanType) {
+ withPosition(unresolvedFilter) {
+ throwDatatypeMismatchFilterNotBoolean(resolvedFilter)
+ }
+ }
+
+ resolvedFilter
+ }
+
+ /**
+ * [[SubqueryAlias]] has a single child and an identifier. We need to resolve the child and update
+ * the scope with the output, since upper expressions can reference [[SubqueryAlias]]es output by
+ * its identifier.
+ */
+ private def resolveSubqueryAlias(unresolvedSubqueryAlias: SubqueryAlias): LogicalPlan = {
+ val resolvedSubqueryAlias =
+ SubqueryAlias(unresolvedSubqueryAlias.identifier, resolve(unresolvedSubqueryAlias.child))
+ withPosition(unresolvedSubqueryAlias) {
+ scopes.overwriteTop(unresolvedSubqueryAlias.alias, resolvedSubqueryAlias.output)
+ }
+ resolvedSubqueryAlias
+ }
+
+ /**
+ * Resolve [[GlobalLimit]]. We have to resolve its child and resolve and validate its limit
+ * expression.
+ */
+ private def resolveGlobalLimit(unresolvedGlobalLimit: GlobalLimit): LogicalPlan = {
+ val resolvedChild = resolve(unresolvedGlobalLimit.child)
+
+ val resolvedLimitExpr = withPosition(unresolvedGlobalLimit) {
+ limitExpressionResolver.resolve(unresolvedGlobalLimit.limitExpr)
+ }
+
+ GlobalLimit(resolvedLimitExpr, resolvedChild)
+ }
+
+ /**
+ * Resolve [[LocalLimit]]. We have to resolve its child and resolve and validate its limit
+ * expression.
+ */
+ private def resolveLocalLimit(unresolvedLocalLimit: LocalLimit): LogicalPlan = {
+ val resolvedChild = resolve(unresolvedLocalLimit.child)
+
+ val resolvedLimitExpr = withPosition(unresolvedLocalLimit) {
+ limitExpressionResolver.resolve(unresolvedLocalLimit.limitExpr)
+ }
+
+ LocalLimit(resolvedLimitExpr, resolvedChild)
+ }
+
+ /**
+ * [[UnresolvedRelation]] was previously looked up by the [[MetadataResolver]] and now we need to:
+ * - Get the specific relation with metadata from `relationsWithResolvedMetadata`, like
+ * [[UnresolvedCatalogRelation]], or throw an error if it wasn't found
+ * - Resolve it further, usually using extensions, like [[DataSourceResolver]]
+ */
+ private def resolveRelation(unresolvedRelation: UnresolvedRelation): LogicalPlan = {
+ relationMetadataProvider.getRelationWithResolvedMetadata(unresolvedRelation) match {
+ case Some(relationWithResolvedMetadata) =>
+ planLogger.log(unresolvedRelation, relationWithResolvedMetadata)
+
+ withPosition(unresolvedRelation) {
+ resolve(relationWithResolvedMetadata)
+ }
+ case None =>
+ withPosition(unresolvedRelation) {
+ unresolvedRelation.tableNotFound(unresolvedRelation.multipartIdentifier)
+ }
+ }
+ }
+
+ /**
+ * [[UnresolvedInlineTable]] resolution requires all the rows to be resolved first. After that we
+ * use [[EvaluateUnresolvedInlineTable]] and try to evaluate the row expressions if possible to
+ * get [[LocalRelation]] right away. Sometimes it's not possible because of expressions like
+ * `current_date()` which are evaluated in the optimizer (SPARK-46380).
+ *
+ * Note: By default if all the inline table expressions can be evaluated eagerly, the parser
+ * would produce a [[LocalRelation]] and the analysis would just skip this step and go straight
+ * to `resolveLocalRelation` (SPARK-48967, SPARK-49269).
+ */
+ private def resolveInlineTable(unresolvedInlineTable: UnresolvedInlineTable): LogicalPlan = {
+ val withResolvedExpressions = UnresolvedInlineTable(
+ unresolvedInlineTable.names,
+ unresolvedInlineTable.rows.map(row => {
+ row.map(expressionResolver.resolve(_))
+ })
+ )
+
+ val resolvedRelation = EvaluateUnresolvedInlineTable
+ .evaluateUnresolvedInlineTable(withResolvedExpressions)
+
+ withPosition(unresolvedInlineTable) {
+ resolve(resolvedRelation)
+ }
+ }
+
+ /**
+ * To finish the operator resolution we add its output to the current scope. This is usually
+ * done for relations. [[NamedRelation]]'s output should be added to the scope under its name.
+ */
+ private def updateNameScopeWithPlanOutput(relation: LogicalPlan): LogicalPlan = {
+ withPosition(relation) {
+ relation match {
+ case namedRelation: NamedRelation =>
+ scopes.top.update(namedRelation.name, namedRelation.output)
+ case _ =>
+ scopes.top += relation.output
+ }
+ }
+ relation
+ }
+
+ override def tryDelegateResolutionToExtension(
+ unresolvedOperator: LogicalPlan): Option[LogicalPlan] = {
+ val resolutionResult = super.tryDelegateResolutionToExtension(unresolvedOperator)
+ resolutionResult.map { resolvedOperator =>
+ updateNameScopeWithPlanOutput(resolvedOperator)
+ }
+ }
+
+ /**
+ * Check if the unresolved operator is explicitly unsupported and throw
+ * [[ExplicitlyUnsupportedResolverFeature]] in that case. Otherwise, throw
+ * [[QueryCompilationErrors.unsupportedSinglePassAnalyzerFeature]].
+ */
+ private def handleUnmatchedOperator(unresolvedOperator: LogicalPlan): Nothing = {
+ if (ExplicitlyUnsupportedResolverFeature.OPERATORS.contains(
+ unresolvedOperator.getClass.getName
+ )) {
+ throw new ExplicitlyUnsupportedResolverFeature(
+ s"unsupported operator: ${unresolvedOperator.getClass.getName}"
+ )
+ }
+ throw QueryCompilationErrors
+ .unsupportedSinglePassAnalyzerFeature(
+ s"${unresolvedOperator.getClass} operator resolution"
+ )
+ .withPosition(unresolvedOperator.origin)
+ }
+
+ private def throwDatatypeMismatchFilterNotBoolean(filter: Filter): Nothing =
+ throw new AnalysisException(
+ errorClass = "DATATYPE_MISMATCH.FILTER_NOT_BOOLEAN",
+ messageParameters = Map(
+ "sqlExpr" -> filter.expressions.map(toSQLExpr).mkString(","),
+ "filter" -> toSQLExpr(filter.condition),
+ "type" -> toSQLType(filter.condition.dataType)
+ )
+ )
+}
+
+object Resolver {
+
+ /**
+ * Create a new instance of the [[RelationResolution]].
+ */
+ def createRelationResolution(catalogManager: CatalogManager): RelationResolution = {
+ new RelationResolution(catalogManager)
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverExtension.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverExtension.scala
new file mode 100644
index 000000000000..8bed881ec97a
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverExtension.scala
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+
+/**
+ * The [[ResolverExtension]] is a main interface for single-pass analysis extensions in Catalyst.
+ * External code that needs specific node types to be resolved has to implement this trait and
+ * inject the implementation into the [[Analyzer.singlePassResolverExtensions]].
+ *
+ * Note that resolver extensions are responsible for creating attribute references with IDs that
+ * are unique from any other subplans. This should be straightforward in most cases because
+ * creating new attribute references will assign [[NamedExpression.newExprId]] by default.
+ */
+trait ResolverExtension {
+
+ /**
+ * Resolve the operator if it's supported by this extension. This method is called by the
+ * single-pass [[Resolver]] on all the configured extensions when it exhausted its match list
+ * for the known node types.
+ *
+ * Guarantees:
+ * - The implementation can rely on children being resolved
+ * - We commit to performing the partial function check only at most once per unresolved operator
+ */
+ def resolveOperator: PartialFunction[LogicalPlan, LogicalPlan]
+}
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
new file mode 100644
index 000000000000..b3b3d4def602
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala
@@ -0,0 +1,283 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.{
+ ResolvedInlineTable,
+ UnresolvedAlias,
+ UnresolvedAttribute,
+ UnresolvedFunction,
+ UnresolvedInlineTable,
+ UnresolvedRelation,
+ UnresolvedStar
+}
+import org.apache.spark.sql.catalyst.expressions.{
+ Alias,
+ AttributeReference,
+ BinaryArithmetic,
+ Cast,
+ ConditionalExpression,
+ CreateNamedStruct,
+ Expression,
+ Literal,
+ Predicate,
+ SubqueryExpression
+}
+import org.apache.spark.sql.catalyst.plans.logical.{
+ Filter,
+ GlobalLimit,
+ LocalLimit,
+ LocalRelation,
+ LogicalPlan,
+ OneRowRelation,
+ Project,
+ SubqueryAlias
+}
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.internal.SQLConf.HiveCaseSensitiveInferenceMode
+
+/**
+ * [[ResolverGuard]] is a class that checks if the operator that is yet to be analyzed
+ * only consists of operators and expressions that are currently supported by the
+ * single-pass analyzer.
+ *
+ * This is a one-shot object and should not be reused after [[apply]] call.
+ */
+class ResolverGuard(catalogManager: CatalogManager) extends SQLConfHelper {
+
+ /**
+ * Check the top level operator of the parsed operator.
+ */
+ def apply(operator: LogicalPlan): Boolean =
+ checkConfValues() && checkVariables() && checkOperator(operator)
+
+ /**
+ * Check if all the operators are supported. For implemented ones, recursively check
+ * their children. For unimplemented ones, return false.
+ */
+ private def checkOperator(operator: LogicalPlan): Boolean = operator match {
+ case project: Project =>
+ checkProject(project)
+ case filter: Filter =>
+ checkFilter(filter)
+ case subqueryAlias: SubqueryAlias =>
+ checkSubqueryAlias(subqueryAlias)
+ case globalLimit: GlobalLimit =>
+ checkGlobalLimit(globalLimit)
+ case localLimit: LocalLimit =>
+ checkLocalLimit(localLimit)
+ case unresolvedRelation: UnresolvedRelation =>
+ checkUnresolvedRelation(unresolvedRelation)
+ case unresolvedInlineTable: UnresolvedInlineTable =>
+ checkUnresolvedInlineTable(unresolvedInlineTable)
+ case resolvedInlineTable: ResolvedInlineTable =>
+ checkResolvedInlineTable(resolvedInlineTable)
+ case localRelation: LocalRelation =>
+ checkLocalRelation(localRelation)
+ case oneRowRelation: OneRowRelation =>
+ checkOneRowRelation(oneRowRelation)
+ case _ =>
+ false
+ }
+
+ /**
+ * Method used to check if expressions are supported by the new analyzer.
+ * For LeafNode types, we return true or false. For other ones, check their children.
+ */
+ private def checkExpression(expression: Expression): Boolean = {
+ expression match {
+ case alias: Alias =>
+ checkAlias(alias)
+ case unresolvedBinaryArithmetic: BinaryArithmetic =>
+ checkUnresolvedBinaryArithmetic(unresolvedBinaryArithmetic)
+ case unresolvedConditionalExpression: ConditionalExpression =>
+ checkUnresolvedConditionalExpression(unresolvedConditionalExpression)
+ case unresolvedCast: Cast =>
+ checkUnresolvedCast(unresolvedCast)
+ case unresolvedStar: UnresolvedStar =>
+ checkUnresolvedStar(unresolvedStar)
+ case unresolvedAlias: UnresolvedAlias =>
+ checkUnresolvedAlias(unresolvedAlias)
+ case unresolvedAttribute: UnresolvedAttribute =>
+ checkUnresolvedAttribute(unresolvedAttribute)
+ case unresolvedPredicate: Predicate =>
+ checkUnresolvedPredicate(unresolvedPredicate)
+ case literal: Literal =>
+ checkLiteral(literal)
+ case attributeReference: AttributeReference =>
+ checkAttributeReference(attributeReference)
+ case createNamedStruct: CreateNamedStruct =>
+ checkCreateNamedStruct(createNamedStruct)
+ case unresolvedFunction: UnresolvedFunction =>
+ checkUnresolvedFunction(unresolvedFunction)
+ case _ =>
+ false
+ }
+ }
+
+ private def checkProject(project: Project) = {
+ checkOperator(project.child) && project.projectList.forall(checkExpression)
+ }
+
+ private def checkFilter(unresolvedFilter: Filter) =
+ checkOperator(unresolvedFilter.child) && checkExpression(unresolvedFilter.condition)
+
+ private def checkSubqueryAlias(subqueryAlias: SubqueryAlias) =
+ subqueryAlias.identifier.qualifier.isEmpty && checkOperator(subqueryAlias.child)
+
+ private def checkGlobalLimit(globalLimit: GlobalLimit) =
+ checkOperator(globalLimit.child) && checkExpression(globalLimit.limitExpr)
+
+ private def checkLocalLimit(localLimit: LocalLimit) =
+ checkOperator(localLimit.child) && checkExpression(localLimit.limitExpr)
+
+ private def checkUnresolvedInlineTable(unresolvedInlineTable: UnresolvedInlineTable) =
+ unresolvedInlineTable.rows.forall(_.forall(checkExpression))
+
+ private def checkUnresolvedRelation(unresolvedRelation: UnresolvedRelation) = true
+
+ private def checkResolvedInlineTable(resolvedInlineTable: ResolvedInlineTable) =
+ resolvedInlineTable.rows.forall(_.forall(checkExpression))
+
+ // Usually we don't check outputs of operators in unresolved plans, but in this case
+ // [[LocalRelation]] is resolved in the parser.
+ private def checkLocalRelation(localRelation: LocalRelation) =
+ localRelation.output.forall(checkExpression)
+
+ private def checkOneRowRelation(oneRowRelation: OneRowRelation) = true
+
+ private def checkAlias(alias: Alias) = checkExpression(alias.child)
+
+ private def checkUnresolvedBinaryArithmetic(unresolvedBinaryArithmetic: BinaryArithmetic) =
+ checkExpression(unresolvedBinaryArithmetic.left) &&
+ checkExpression(unresolvedBinaryArithmetic.right)
+
+ private def checkUnresolvedConditionalExpression(
+ unresolvedConditionalExpression: ConditionalExpression) =
+ unresolvedConditionalExpression.children.forall(checkExpression)
+
+ private def checkUnresolvedCast(cast: Cast) = checkExpression(cast.child)
+
+ private def checkUnresolvedStar(unresolvedStar: UnresolvedStar) = true
+
+ private def checkUnresolvedAlias(unresolvedAlias: UnresolvedAlias) =
+ checkExpression(unresolvedAlias.child)
+
+ private def checkUnresolvedAttribute(unresolvedAttribute: UnresolvedAttribute) =
+ !ResolverGuard.UNSUPPORTED_ATTRIBUTE_NAMES.contains(unresolvedAttribute.nameParts.head)
+
+ private def checkUnresolvedPredicate(unresolvedPredicate: Predicate) = {
+ unresolvedPredicate match {
+ case _: SubqueryExpression => false
+ case other =>
+ other.children.forall(checkExpression)
+ }
+ }
+
+ private def checkAttributeReference(attributeReference: AttributeReference) = true
+
+ private def checkCreateNamedStruct(createNamedStruct: CreateNamedStruct) = {
+ createNamedStruct.children.forall(checkExpression)
+ }
+
+ private def checkUnresolvedFunction(unresolvedFunction: UnresolvedFunction) =
+ ResolverGuard.SUPPORTED_FUNCTION_NAMES.contains(
+ unresolvedFunction.nameParts.head
+ ) && unresolvedFunction.children.forall(checkExpression)
+
+ private def checkLiteral(literal: Literal) = true
+
+ private def checkConfValues() =
+ // Case sensitive analysis is not supported.
+ !conf.caseSensitiveAnalysis &&
+ // Case-sensitive inference is not supported for Hive table schema.
+ conf.caseSensitiveInferenceMode == HiveCaseSensitiveInferenceMode.NEVER_INFER
+
+ private def checkVariables() = catalogManager.tempVariableManager.isEmpty
+}
+
+object ResolverGuard {
+
+ private val UNSUPPORTED_ATTRIBUTE_NAMES = {
+ val map = new IdentifierMap[Unit]()
+
+ /**
+ * Some SQL functions can be called without the braces and thus they are found in the
+ * parsed operator as UnresolvedAttributes. This list contains the names of those functions
+ * so we can reject them. Find more information in [[ColumnResolutionHelper.literalFunctions]].
+ */
+ map += ("current_date", ())
+ map += ("current_timestamp", ())
+ map += ("current_user", ())
+ map += ("user", ())
+ map += ("session_user", ())
+ map += ("grouping__id", ())
+
+ /**
+ * Metadata column resolution is not supported for now
+ */
+ map += ("_metadata", ())
+
+ map
+ }
+
+ /**
+ * Most of the functions are not supported, but we allow some explicitly supported ones.
+ */
+ private val SUPPORTED_FUNCTION_NAMES = {
+ val map = new IdentifierMap[Unit]()
+ map += ("array", ())
+ // map += ("array_agg", ()) - until aggregate expressions are supported
+ map += ("array_append", ())
+ map += ("array_compact", ())
+ map += ("array_contains", ())
+ map += ("array_distinct", ())
+ map += ("array_except", ())
+ map += ("array_insert", ())
+ map += ("array_intersect", ())
+ map += ("array_join", ())
+ map += ("array_max", ())
+ map += ("array_min", ())
+ map += ("array_position", ())
+ map += ("array_prepend", ())
+ map += ("array_remove", ())
+ map += ("array_repeat", ())
+ map += ("array_size", ())
+ // map += ("array_sort", ()) - until lambda functions are supported
+ map += ("array_union", ())
+ map += ("arrays_overlap", ())
+ map += ("arrays_zip", ())
+ map += ("coalesce", ())
+ map += ("if", ())
+ map += ("map", ())
+ map += ("map_concat", ())
+ map += ("map_contains_key", ())
+ map += ("map_entries", ())
+ // map += ("map_filter", ()) - until lambda functions are supported
+ map += ("map_from_arrays", ())
+ map += ("map_from_entries", ())
+ map += ("map_keys", ())
+ map += ("map_values", ())
+ // map += ("map_zip_with", ()) - until lambda functions are supported
+ map += ("named_struct", ())
+ map += ("sort_array", ())
+ map += ("str_to_map", ())
+ map
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesExpressionChildren.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesExpressionChildren.scala
new file mode 100644
index 000000000000..c170941ce534
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesExpressionChildren.scala
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+
+trait ResolvesExpressionChildren {
+
+ /**
+ * Resolves generic [[Expression]] children and returns its copy with children resolved.
+ */
+ protected def withResolvedChildren[ExpressionType <: Expression](
+ unresolvedExpression: ExpressionType,
+ resolveChild: Expression => Expression): ExpressionType = {
+ val newChildren = unresolvedExpression.children.map(resolveChild(_))
+ unresolvedExpression.withNewChildren(newChildren).asInstanceOf[ExpressionType]
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesOperatorChildren.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesOperatorChildren.scala
new file mode 100644
index 000000000000..0f548c3c5585
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesOperatorChildren.scala
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+
+/**
+ * A mixin trait for all operator resolvers that need to resolve their children.
+ */
+trait ResolvesOperatorChildren {
+
+ /**
+ * Resolves generic [[LogicalPlan]] children and returns its copy with children resolved.
+ */
+ protected def withResolvedChildren[OperatorType <: LogicalPlan](
+ unresolvedOperator: OperatorType,
+ resolve: LogicalPlan => LogicalPlan): OperatorType = {
+ val newChildren = unresolvedOperator.children.map(resolve(_))
+ unresolvedOperator.withNewChildren(newChildren).asInstanceOf[OperatorType]
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimeAddResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimeAddResolver.scala
new file mode 100644
index 000000000000..bf27f6459872
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimeAddResolver.scala
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.{
+ AnsiStringPromotionTypeCoercion,
+ AnsiTypeCoercion,
+ StringPromotionTypeCoercion,
+ TypeCoercion
+}
+import org.apache.spark.sql.catalyst.expressions.{Expression, TimeAdd}
+
+/**
+ * Helper resolver for [[TimeAdd]] which is produced by resolving [[BinaryArithmetic]] nodes.
+ */
+class TimeAddResolver(
+ expressionResolver: ExpressionResolver,
+ timezoneAwareExpressionResolver: TimezoneAwareExpressionResolver)
+ extends TreeNodeResolver[TimeAdd, Expression]
+ with ResolvesExpressionChildren {
+
+ private val typeCoercionRules: Seq[Expression => Expression] =
+ if (conf.ansiEnabled) {
+ TimeAddResolver.ANSI_TYPE_COERCION_RULES
+ } else {
+ TimeAddResolver.TYPE_COERCION_RULES
+ }
+ private val typeCoercionResolver: TypeCoercionResolver =
+ new TypeCoercionResolver(timezoneAwareExpressionResolver, typeCoercionRules)
+
+ override def resolve(unresolvedTimeAdd: TimeAdd): Expression = {
+ val timeAddWithResolvedChildren: TimeAdd =
+ withResolvedChildren(unresolvedTimeAdd, expressionResolver.resolve)
+ val timeAddWithTypeCoercion: Expression = typeCoercionResolver
+ .resolve(timeAddWithResolvedChildren)
+ timezoneAwareExpressionResolver.withResolvedTimezone(
+ timeAddWithTypeCoercion,
+ conf.sessionLocalTimeZone
+ )
+ }
+}
+
+object TimeAddResolver {
+ // Ordering in the list of type coercions should be in sync with the list in [[TypeCoercion]].
+ private val TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ StringPromotionTypeCoercion.apply,
+ TypeCoercion.ImplicitTypeCoercion.apply,
+ TypeCoercion.DateTimeOperationsTypeCoercion.apply
+ )
+
+ // Ordering in the list of type coercions should be in sync with the list in [[AnsiTypeCoercion]].
+ private val ANSI_TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ AnsiStringPromotionTypeCoercion.apply,
+ AnsiTypeCoercion.ImplicitTypeCoercion.apply,
+ AnsiTypeCoercion.AnsiDateTimeOperationsTypeCoercion.apply
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimezoneAwareExpressionResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimezoneAwareExpressionResolver.scala
new file mode 100644
index 000000000000..a45e9e41cbfb
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimezoneAwareExpressionResolver.scala
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.expressions.{Expression, TimeZoneAwareExpression}
+
+/**
+ * Resolves [[TimeZoneAwareExpressions]] by applying the session's local timezone.
+ *
+ * This class is responsible for resolving [[TimeZoneAwareExpression]]s by first resolving their
+ * children and then applying the session's local timezone. Additionally, ensures that any tags from
+ * the original expression are preserved during the resolution process.
+ *
+ * @constructor Creates a new TimezoneAwareExpressionResolver with the given expression resolver.
+ * @param expressionResolver The [[ExpressionResolver]] used to resolve child expressions.
+ */
+class TimezoneAwareExpressionResolver(expressionResolver: TreeNodeResolver[Expression, Expression])
+ extends TreeNodeResolver[TimeZoneAwareExpression, Expression]
+ with ResolvesExpressionChildren {
+
+ /**
+ * Resolves a [[TimeZoneAwareExpression]] by resolving its children and applying a timezone.
+ *
+ * @param unresolvedTimezoneExpression The [[TimeZoneAwareExpression]] to resolve.
+ * @return A resolved [[Expression]] with the session's local timezone applied.
+ */
+ override def resolve(unresolvedTimezoneExpression: TimeZoneAwareExpression): Expression = {
+ val expressionWithResolvedChildren =
+ withResolvedChildren(unresolvedTimezoneExpression, expressionResolver.resolve)
+ withResolvedTimezoneCopyTags(expressionWithResolvedChildren, conf.sessionLocalTimeZone)
+ }
+
+ /**
+ * Applies a timezone to a [[TimeZoneAwareExpression]] while preserving original tags.
+ *
+ * This method is particularly useful for cases like resolving [[Cast]] expressions where tags
+ * such as [[USER_SPECIFIED_CAST]] need to be preserved.
+ *
+ * @param expression The [[TimeZoneAwareExpression]] to apply the timezone to.
+ * @param timeZoneId The timezone ID to apply.
+ * @return A new [[TimeZoneAwareExpression]] with the specified timezone and original tags.
+ */
+ def withResolvedTimezoneCopyTags(expression: Expression, timeZoneId: String): Expression = {
+ val withTimeZone = withResolvedTimezone(expression, timeZoneId)
+ withTimeZone.copyTagsFrom(expression)
+ withTimeZone
+ }
+
+ /**
+ * Apply timezone to [[TimeZoneAwareExpression]] expressions.
+ */
+ def withResolvedTimezone(expression: Expression, timeZoneId: String): Expression =
+ expression match {
+ case timezoneExpression: TimeZoneAwareExpression if timezoneExpression.timeZoneId.isEmpty =>
+ timezoneExpression.withTimeZone(timeZoneId)
+ case other => other
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TracksResolvedNodes.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TracksResolvedNodes.scala
new file mode 100644
index 000000000000..dd86bf843b4e
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TracksResolvedNodes.scala
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import java.util.IdentityHashMap
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.trees.TreeNode
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Trait for top-level resolvers that is used to keep track of resolved nodes and throw an error if
+ * a node is resolved more than once. This is only used in tests because of the memory overhead of
+ * using a set to track resolved nodes.
+ */
+trait TracksResolvedNodes[TreeNodeType <: TreeNode[TreeNodeType]] extends SQLConfHelper {
+ // Using Map because IdentityHashSet is not available in Scala
+ private val seenResolvedNodes = new IdentityHashMap[TreeNodeType, Unit]
+
+ private val shouldTrackResolvedNodes =
+ conf.getConf(SQLConf.ANALYZER_SINGLE_PASS_TRACK_RESOLVED_NODES_ENABLED)
+
+ protected def throwIfNodeWasResolvedEarlier(node: TreeNodeType): Unit =
+ if (shouldTrackResolvedNodes && seenResolvedNodes.containsKey(node)) {
+ throw SparkException.internalError(
+ s"Single-pass resolver attempted to resolve the same node more than once: $node"
+ )
+ }
+
+ protected def markNodeAsResolved(node: TreeNodeType): Unit = {
+ if (shouldTrackResolvedNodes) {
+ seenResolvedNodes.put(node, ())
+ }
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TreeNodeResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TreeNodeResolver.scala
new file mode 100644
index 000000000000..5991585995ca
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TreeNodeResolver.scala
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.trees.TreeNode
+
+/**
+ * Base class for [[TreeNode]] resolvers. All resolvers should extend this class with
+ * specific [[UnresolvedNode]] and [[ResolvedNode]] types.
+ */
+trait TreeNodeResolver[UnresolvedNode <: TreeNode[_], ResolvedNode <: TreeNode[_]]
+ extends SQLConfHelper {
+ def resolve(unresolvedNode: UnresolvedNode): ResolvedNode
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TypeCoercionResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TypeCoercionResolver.scala
new file mode 100644
index 000000000000..cf4c2ef0d750
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TypeCoercionResolver.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.expressions.{Cast, Expression}
+
+/**
+ * [[TypeCoercionResolver]] is used by other resolvers to uniformly apply type coercions to all
+ * expressions. [[TypeCoercionResolver]] takes in a sequence of type coercion transformations that
+ * should be applied to an expression and applies them in order. Finally, [[TypeCoercionResolver]]
+ * applies timezone to expression's children, as a child could be replaced with Cast(child, type),
+ * therefore [[Cast]] resolution is needed. Timezone is applied only on children that have been
+ * re-instantiated by [[TypeCoercionResolver]], because otherwise children have already been
+ * resolved.
+ */
+class TypeCoercionResolver(
+ timezoneAwareExpressionResolver: TimezoneAwareExpressionResolver,
+ typeCoercionRules: Seq[Expression => Expression])
+ extends TreeNodeResolver[Expression, Expression] {
+
+ override def resolve(expression: Expression): Expression = {
+ val oldChildren = expression.children
+
+ val withTypeCoercion = typeCoercionRules.foldLeft(expression) {
+ case (expr, rule) => rule.apply(expr)
+ }
+
+ val newChildren = withTypeCoercion.children.zip(oldChildren).map {
+ case (newChild: Cast, oldChild) if !newChild.eq(oldChild) =>
+ timezoneAwareExpressionResolver.withResolvedTimezone(newChild, conf.sessionLocalTimeZone)
+ case (newChild, _) => newChild
+ }
+ withTypeCoercion.withNewChildren(newChildren)
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/UnaryMinusResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/UnaryMinusResolver.scala
new file mode 100644
index 000000000000..739d7cf43c18
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/UnaryMinusResolver.scala
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.analysis.{AnsiTypeCoercion, TypeCoercion}
+import org.apache.spark.sql.catalyst.expressions.{Expression, UnaryMinus}
+
+/**
+ * Resolver for [[UnaryMinus]]. Resolves children and applies type coercion to target node.
+ */
+class UnaryMinusResolver(
+ expressionResolver: ExpressionResolver,
+ timezoneAwareExpressionResolver: TimezoneAwareExpressionResolver)
+ extends TreeNodeResolver[UnaryMinus, Expression]
+ with ResolvesExpressionChildren {
+
+ private val typeCoercionRules: Seq[Expression => Expression] =
+ if (conf.ansiEnabled) {
+ UnaryMinusResolver.ANSI_TYPE_COERCION_RULES
+ } else {
+ UnaryMinusResolver.TYPE_COERCION_RULES
+ }
+ private val typeCoercionResolver: TypeCoercionResolver =
+ new TypeCoercionResolver(timezoneAwareExpressionResolver, typeCoercionRules)
+
+ override def resolve(unresolvedUnaryMinus: UnaryMinus): Expression = {
+ val unaryMinusWithResolvedChildren: UnaryMinus =
+ withResolvedChildren(unresolvedUnaryMinus, expressionResolver.resolve)
+ typeCoercionResolver.resolve(unaryMinusWithResolvedChildren)
+ }
+}
+
+object UnaryMinusResolver {
+ // Ordering in the list of type coercions should be in sync with the list in [[TypeCoercion]].
+ private val TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ TypeCoercion.ImplicitTypeCoercion.apply,
+ TypeCoercion.DateTimeOperationsTypeCoercion.apply
+ )
+
+ // Ordering in the list of type coercions should be in sync with the list in [[AnsiTypeCoercion]].
+ private val ANSI_TYPE_COERCION_RULES: Seq[Expression => Expression] = Seq(
+ AnsiTypeCoercion.ImplicitTypeCoercion.apply,
+ AnsiTypeCoercion.AnsiDateTimeOperationsTypeCoercion.apply
+ )
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
index 0a73b6b85674..b14f4be534a3 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
@@ -507,7 +507,9 @@ abstract class UnresolvedStarBase(target: Option[Seq[String]]) extends Star with
childOperatorMetadataOutput: Seq[Attribute],
resolve: (Seq[String], Resolver) => Option[NamedExpression],
suggestedAttributes: Seq[Attribute],
- resolver: Resolver): Seq[NamedExpression] = {
+ resolver: Resolver,
+ cleanupNestedAliasesDuringStructExpansion: Boolean = false
+ ): Seq[NamedExpression] = {
// If there is no table specified, use all non-hidden input attributes.
if (target.isEmpty) return childOperatorOutput
@@ -528,11 +530,22 @@ abstract class UnresolvedStarBase(target: Option[Seq[String]]) extends Star with
// (i.e. [name].* is both a table and a struct), the struct path can always be qualified.
val attribute = resolve(target.get, resolver)
if (attribute.isDefined) {
+ // If cleanupNestedAliasesDuringStructExpansion is true, we remove nested aliases during
+ // struct expansion. This is something which is done in the CleanupAliases rule but for the
+ // single-pass analyzer it has to be done here to avoid additional tree traversals.
+ val normalizedAttribute = if (cleanupNestedAliasesDuringStructExpansion) {
+ attribute.get match {
+ case a: Alias => a.child
+ case other => other
+ }
+ } else {
+ attribute.get
+ }
// This target resolved to an attribute in child. It must be a struct. Expand it.
- attribute.get.dataType match {
+ normalizedAttribute.dataType match {
case s: StructType => s.zipWithIndex.map {
case (f, i) =>
- val extract = GetStructField(attribute.get, i)
+ val extract = GetStructField(normalizedAttribute, i)
Alias(extract, f.name)()
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/TempVariableManager.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/TempVariableManager.scala
index abe6cede0c55..2c262da1f444 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/TempVariableManager.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/TempVariableManager.scala
@@ -63,6 +63,10 @@ class TempVariableManager extends DataTypeErrorsBase {
def clear(): Unit = synchronized {
variables.clear()
}
+
+ def isEmpty: Boolean = synchronized {
+ variables.isEmpty
+ }
}
case class VariableDefinition(defaultValueSQL: String, currentValue: Literal)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index b673d5a04315..e55d4d1f9523 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -4136,6 +4136,83 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
)
}
+ def unsupportedSinglePassAnalyzerFeature(feature: String): AnalysisException = {
+ new AnalysisException(
+ errorClass = "UNSUPPORTED_SINGLE_PASS_ANALYZER_FEATURE",
+ messageParameters = Map("feature" -> feature)
+ )
+ }
+
+ def ambiguousResolverExtension(
+ operator: LogicalPlan,
+ extensionNames: Seq[String]): AnalysisException = {
+ new AnalysisException(
+ errorClass = "AMBIGUOUS_RESOLVER_EXTENSION",
+ messageParameters = Map(
+ "operator" -> operator.getClass.getName,
+ "extensions" -> extensionNames.mkString(", ")
+ )
+ )
+ }
+
+ def fixedPointFailedSinglePassSucceeded(
+ singlePassResult: LogicalPlan,
+ fixedPointException: Throwable): Throwable = {
+ new ExtendedAnalysisException(
+ new AnalysisException(
+ errorClass = "HYBRID_ANALYZER_EXCEPTION.FIXED_POINT_FAILED_SINGLE_PASS_SUCCEEDED",
+ messageParameters = Map("singlePassOutput" -> singlePassResult.toString),
+ cause = Some(fixedPointException)
+ ),
+ plan = singlePassResult
+ )
+ }
+
+ def hybridAnalyzerOutputSchemaComparisonMismatch(
+ fixedPointOutputSchema: StructType,
+ singlePassOutputSchema: StructType): Throwable = {
+
+ def structToString(struct: StructType) =
+ struct.fields.map(structFieldToStringWithMetadata(_)).mkString(",")
+
+ def structFieldToStringWithMetadata(sf: StructField) =
+ s"(${sf.name},${sf.dataType},${sf.nullable},${sf.metadata})"
+
+ new AnalysisException(
+ errorClass = "HYBRID_ANALYZER_EXCEPTION.OUTPUT_SCHEMA_COMPARISON_MISMATCH",
+ messageParameters = Map(
+ "fixedPointOutputSchema" -> structToString(fixedPointOutputSchema),
+ "singlePassOutputSchema" -> structToString(singlePassOutputSchema)
+ )
+ )
+ }
+
+ def hybridAnalyzerLogicalPlanComparisonMismatch(
+ fixedPointOutput: LogicalPlan,
+ singlePassOutput: LogicalPlan): Throwable = {
+ new AnalysisException(
+ errorClass = "HYBRID_ANALYZER_EXCEPTION.LOGICAL_PLAN_COMPARISON_MISMATCH",
+ messageParameters = Map(
+ "fixedPointOutput" -> fixedPointOutput.toString,
+ "singlePassOutput" -> singlePassOutput.toString
+ )
+ )
+ }
+
+ def resolutionValidationError(cause: Throwable, plan: LogicalPlan): Throwable = {
+ new ExtendedAnalysisException(
+ new AnalysisException(
+ errorClass = "INTERNAL_ERROR",
+ cause = Some(cause),
+ messageParameters = Map(
+ "message" -> ("The analysis phase failed with an internal error. Reason: " +
+ cause.getMessage)
+ )
+ ),
+ plan = plan
+ )
+ }
+
def avroNotLoadedSqlFunctionsUnusable(functionName: String): Throwable = {
new AnalysisException(
errorClass = "AVRO_NOT_LOADED_SQL_FUNCTIONS_UNUSABLE",
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 e69fa996f6f2..31282d43bbce 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
@@ -247,6 +247,76 @@ object SQLConf {
.intConf
.createWithDefault(100)
+ val ANALYZER_SINGLE_PASS_RESOLVER_ENABLED =
+ buildConf("spark.sql.analyzer.singlePassResolver.enabled")
+ .internal()
+ .doc(
+ "When true, use the single-pass Resolver instead of the fixed-point Analyzer. " +
+ "This is an alternative Analyzer framework, which resolves the parsed logical plan in a " +
+ "single post-order traversal. It uses ExpressionResolver to resolve expressions and " +
+ "NameScope to control the visibility of names. In contrast to the current fixed-point " +
+ "framework, subsequent in-tree traversals are disallowed. Most of the fixed-point " +
+ "Analyzer code is reused in the form of specific node transformation functions " +
+ "(AliasResolution.resolve, FunctionResolution.resolveFunction, etc)." +
+ "This feature is currently under development."
+ )
+ .version("4.0.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ val ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER =
+ buildConf("spark.sql.analyzer.singlePassResolver.dualRunWithLegacy")
+ .internal()
+ .doc("When true, run both analyzers to check if single-pass Analyzer correctly produces" +
+ " the same analyzed plan as the fixed-point Analyzer for the existing set of features" +
+ " defined in the ResolverGuard")
+ .version("4.0.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ val ANALYZER_SINGLE_PASS_RESOLVER_VALIDATION_ENABLED =
+ buildConf("spark.sql.analyzer.singlePassResolver.validationEnabled")
+ .internal()
+ .doc(
+ "When true, validate the Resolver output with ResolutionValidator. " +
+ "The ResolutionValidator validates the resolved logical plan tree in one pass " +
+ "and asserts the internal contracts. It uses the ExpressionResolutionValidator " +
+ "internally to validate resolved expression trees in the same manner."
+ )
+ .version("4.0.0")
+ .booleanConf
+ .createWithDefault(true)
+
+ val ANALYZER_SINGLE_PASS_TRACK_RESOLVED_NODES_ENABLED =
+ buildConf("spark.sql.analyzer.singlePassResolver.trackResolvedNodes.enabled")
+ .internal()
+ .doc(
+ "When true, keep track of resolved nodes in order to assert that the single-pass " +
+ "invariant is never broken. While true, if a resolver attempts to resolve the same node " +
+ "twice, INTERNAL_ERROR exception is thrown. Used only for testing due to memory impact " +
+ "of storing each node in a HashSet."
+ )
+ .version("4.0.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ val ANALYZER_SINGLE_PASS_RESOLVER_RELATION_BRIDGING_ENABLED =
+ buildConf("spark.sql.analyzer.singlePassResolver.relationBridging.enabled")
+ .internal()
+ .doc(
+ "When set to true, the single-pass Resolver will reuse the relation metadata that was " +
+ "previously resolved in fixed-point run. This makes sense only in " +
+ "ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER mode. In that case HybridAnalyzer " +
+ "enables the AnalyzerBridgeState and passes it to the single-pass Analyzer after the " +
+ "fixed-point run is complete. Single-pass Resolver uses this AnalyzerBridgeState to " +
+ "construct a special RelationMetadataProvider implementation - " +
+ "BridgedRelationMetadataProvider. This component simply reuses cached relation metadata " +
+ "and avoids any blocking calls (catalog RPCs or table metadata reads)."
+ )
+ .version("4.0.0")
+ .booleanConf
+ .createWithDefault(Utils.isTesting)
+
val MULTI_COMMUTATIVE_OP_OPT_THRESHOLD =
buildConf("spark.sql.analyzer.canonicalization.multiCommutativeOpMemoryOptThreshold")
.internal()
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/LimitExpressionResolverSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/LimitExpressionResolverSuite.scala
new file mode 100644
index 000000000000..fdab4df379a7
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/LimitExpressionResolverSuite.scala
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, Expression, Literal}
+import org.apache.spark.sql.errors.QueryErrorsBase
+import org.apache.spark.sql.types.IntegerType
+
+class LimitExpressionResolverSuite extends SparkFunSuite with QueryErrorsBase {
+
+ private class IdentityExpressionResolver extends TreeNodeResolver[Expression, Expression] {
+ override def resolve(expression: Expression): Expression = expression
+ }
+
+ private val expressionResolver = new IdentityExpressionResolver
+ private val limitExpressionResolver = new LimitExpressionResolver(expressionResolver)
+
+ test("Basic LIMIT without errors") {
+ val expr = Literal(42, IntegerType)
+ assert(limitExpressionResolver.resolve(expr) == expr)
+ }
+
+ test("Unfoldable LIMIT") {
+ val col = AttributeReference(name = "foo", dataType = IntegerType)()
+ checkError(
+ exception = intercept[AnalysisException] {
+ limitExpressionResolver.resolve(col)
+ },
+ condition = "INVALID_LIMIT_LIKE_EXPRESSION.IS_UNFOLDABLE",
+ parameters = Map("name" -> "limit", "expr" -> toSQLExpr(col))
+ )
+ }
+
+ test("LIMIT with non-integer") {
+ val anyNonInteger = Literal("42")
+ checkError(
+ exception = intercept[AnalysisException] {
+ limitExpressionResolver.resolve(anyNonInteger)
+ },
+ condition = "INVALID_LIMIT_LIKE_EXPRESSION.DATA_TYPE",
+ parameters = Map(
+ "name" -> "limit",
+ "expr" -> toSQLExpr(anyNonInteger),
+ "dataType" -> toSQLType(anyNonInteger.dataType)
+ )
+ )
+ }
+
+ test("LIMIT with null") {
+ val expr = Cast(Literal(null), IntegerType)
+ checkError(
+ exception = intercept[AnalysisException] {
+ limitExpressionResolver.resolve(expr)
+ },
+ condition = "INVALID_LIMIT_LIKE_EXPRESSION.IS_NULL",
+ parameters = Map(
+ "name" -> "limit",
+ "expr" -> toSQLExpr(expr)
+ )
+ )
+ }
+
+ test("LIMIT with negative integer") {
+ val expr = Literal(-1, IntegerType)
+ checkError(
+ exception = intercept[AnalysisException] {
+ limitExpressionResolver.resolve(expr)
+ },
+ condition = "INVALID_LIMIT_LIKE_EXPRESSION.IS_NEGATIVE",
+ parameters = Map(
+ "name" -> "limit",
+ "expr" -> toSQLExpr(expr),
+ "v" -> toSQLValue(-1, IntegerType)
+ )
+ )
+ }
+}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolutionValidatorSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolutionValidatorSuite.scala
new file mode 100644
index 000000000000..922e94ea442b
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolutionValidatorSuite.scala
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import scala.collection.immutable
+import scala.reflect.runtime.universe.typeOf
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.expressions.{
+ Add,
+ Alias,
+ AttributeReference,
+ Cast,
+ GreaterThan,
+ Literal,
+ NamedExpression,
+ TimeAdd
+}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation, LogicalPlan, Project}
+import org.apache.spark.sql.types.{
+ BooleanType,
+ DayTimeIntervalType,
+ DecimalType,
+ IntegerType,
+ StringType,
+ TimestampType
+}
+
+class ResolutionValidatorSuite extends SparkFunSuite with SQLConfHelper {
+ private val resolveMethodNamesToIgnore = Seq(
+ // [[Resolver]] accepts [[UnresolvedInlineTable]], [[ResolvedInlineTable]] and
+ // [[LocalRelation]], but produces only [[ResolvedInlineTable]] and [[LocalRelation]], so
+ // we omit one of them here.
+ // See [[Resolver.resolveInlineTable]] scaladoc for more info.
+ "resolveResolvedInlineTable"
+ )
+
+ private val colInteger = AttributeReference(name = "colInteger", dataType = IntegerType)()
+ private val colBoolean = AttributeReference(name = "colBoolean", dataType = BooleanType)()
+ private val colTimestamp = AttributeReference(name = "colTimestamp", dataType = TimestampType)()
+
+ test("All resolve* methods must have validate* counterparts") {
+ val actualMethodNames = typeOf[ResolutionValidator].decls
+ .collect {
+ case decl if decl.isMethod => decl.name.toString
+ }
+ .filter(name => {
+ name.startsWith("validate")
+ })
+ val actualMethodNamesSet = immutable.HashSet(actualMethodNames.toSeq: _*)
+
+ val resolveMethodNamesToIgnoreSet = immutable.HashSet(resolveMethodNamesToIgnore: _*)
+
+ typeOf[Resolver].decls
+ .collect {
+ case decl if decl.isMethod => decl.name.toString
+ }
+ .filter(name => {
+ name.startsWith("resolve") && !resolveMethodNamesToIgnoreSet.contains(name)
+ })
+ .map(name => {
+ "validate" + name.stripPrefix("resolve")
+ })
+ .foreach(name => {
+ assert(actualMethodNamesSet.contains(name), name)
+ })
+ }
+
+ test("Project") {
+ validate(
+ Project(
+ projectList = Seq(colInteger, colBoolean, colInteger),
+ child = LocalRelation(output = Seq(colInteger, colBoolean))
+ )
+ )
+ validate(
+ Project(
+ projectList = Seq(colInteger),
+ child = LocalRelation(output = colBoolean)
+ ),
+ error = Some("Project list contains nonexisting attribute")
+ )
+ }
+
+ test("Filter") {
+ validate(
+ Project(
+ projectList = Seq(colBoolean),
+ child = Filter(
+ condition = colBoolean,
+ child = LocalRelation(output = colBoolean)
+ )
+ )
+ )
+ validate(
+ Project(
+ projectList = Seq(colInteger),
+ child = Filter(
+ condition = colInteger,
+ child = LocalRelation(output = colInteger)
+ )
+ ),
+ error = Some("Non-boolean condition")
+ )
+ validate(
+ Project(
+ projectList = Seq(colBoolean),
+ child = Filter(
+ condition = AttributeReference(name = "colBooleanOther", dataType = BooleanType)(),
+ child = LocalRelation(output = colBoolean)
+ )
+ ),
+ error = Some("Condition references nonexisting attribute")
+ )
+ }
+
+ test("Predicate") {
+ validate(
+ Project(
+ projectList = Seq(colInteger),
+ child = Filter(
+ condition = GreaterThan(colInteger, colInteger),
+ child = LocalRelation(output = colInteger)
+ )
+ )
+ )
+ validate(
+ Project(
+ projectList = Seq(colInteger),
+ child = Filter(
+ condition = GreaterThan(colInteger, colBoolean),
+ child = LocalRelation(output = Seq(colInteger, colBoolean))
+ )
+ ),
+ error = Some("Input data types mismatch")
+ )
+ }
+
+ test("BinaryExpression") {
+ validate(
+ Project(
+ projectList = Seq(
+ Alias(
+ child = Add(
+ left = Literal(5),
+ right = Literal(1)
+ ),
+ "Add"
+ )(NamedExpression.newExprId)
+ ),
+ child = LocalRelation(output = colInteger)
+ )
+ )
+ validate(
+ Project(
+ projectList = Seq(
+ Alias(
+ child = Add(
+ left = Literal(5),
+ right = Literal("1")
+ ),
+ "AddWrongInputTypes"
+ )(NamedExpression.newExprId)
+ ),
+ child = LocalRelation(output = colInteger)
+ ),
+ error = Some("checkInputDataTypes mismatch")
+ )
+ validate(
+ Project(
+ projectList = Seq(
+ Alias(
+ child = TimeAdd(
+ start = Cast(
+ child = Literal("2024-10-01"),
+ dataType = TimestampType,
+ timeZoneId = Option(conf.sessionLocalTimeZone)
+ ),
+ interval = Cast(
+ child = Literal(1),
+ dataType = DayTimeIntervalType(DayTimeIntervalType.DAY, DayTimeIntervalType.DAY),
+ timeZoneId = Option(conf.sessionLocalTimeZone)
+ )
+ ),
+ "AddNoTimezone"
+ )(NamedExpression.newExprId)
+ ),
+ child = LocalRelation(output = colInteger)
+ ),
+ error = Some("TimezoneId is not set for TimeAdd")
+ )
+ }
+
+ test("TimeZoneAwareExpression") {
+ validate(
+ Project(
+ projectList = Seq(
+ Alias(
+ Cast(
+ child = colInteger,
+ dataType = DecimalType.USER_DEFAULT,
+ timeZoneId = Option(conf.sessionLocalTimeZone)
+ ),
+ "withTimezone"
+ )(NamedExpression.newExprId)
+ ),
+ child = LocalRelation(output = colInteger)
+ )
+ )
+ validate(
+ Project(
+ projectList = Seq(
+ Alias(
+ Cast(
+ child = colTimestamp,
+ dataType = StringType
+ ),
+ "withoutTimezone"
+ )(NamedExpression.newExprId)
+ ),
+ child = LocalRelation(output = colTimestamp)
+ ),
+ error = Some("TimezoneId is not set")
+ )
+ }
+
+ def validate(plan: LogicalPlan, error: Option[String] = None): Unit = {
+ def errorWrapper(error: String)(body: => Unit): Unit = {
+ withClue(error) {
+ intercept[Throwable] {
+ body
+ }
+ }
+ }
+
+ def noopWrapper(body: => Unit) = {
+ body
+ }
+
+ val wrapper = error
+ .map(error => { errorWrapper(error) _ })
+ .getOrElse { noopWrapper _ }
+
+ val validator = new ResolutionValidator
+ wrapper {
+ validator.validatePlan(plan)
+ }
+ }
+}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimezoneAwareExpressionResolverSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimezoneAwareExpressionResolverSuite.scala
new file mode 100644
index 000000000000..d5c5387d4b76
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimezoneAwareExpressionResolverSuite.scala
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.{
+ AttributeReference,
+ Cast,
+ Expression,
+ TimeZoneAwareExpression
+}
+import org.apache.spark.sql.types.{IntegerType, StringType}
+
+class TimezoneAwareExpressionResolverSuite extends SparkFunSuite {
+
+ class HardCodedExpressionResolver(resolvedExpression: Expression)
+ extends TreeNodeResolver[Expression, Expression] {
+ override def resolve(expression: Expression): Expression = resolvedExpression
+ }
+
+ private val unresolvedChild =
+ AttributeReference(name = "unresolvedChild", dataType = StringType)()
+ private val resolvedChild = AttributeReference(name = "resolvedChild", dataType = IntegerType)()
+ private val castExpression = Cast(child = unresolvedChild, dataType = IntegerType)
+ private val expressionResolver = new HardCodedExpressionResolver(resolvedChild)
+ private val timezoneAwareExpressionResolver = new TimezoneAwareExpressionResolver(
+ expressionResolver
+ )
+
+ test("TimeZoneAwareExpression resolution") {
+ assert(castExpression.children.head == unresolvedChild)
+ assert(castExpression.timeZoneId.isEmpty)
+ assert(castExpression.getTagValue(Cast.USER_SPECIFIED_CAST).isEmpty)
+
+ castExpression.setTagValue(Cast.USER_SPECIFIED_CAST, ())
+ val resolvedExpression =
+ timezoneAwareExpressionResolver.resolve(castExpression).asInstanceOf[TimeZoneAwareExpression]
+
+ assert(resolvedExpression.children.head == resolvedChild)
+ assert(resolvedExpression.timeZoneId.nonEmpty)
+ assert(resolvedExpression.getTagValue(Cast.USER_SPECIFIED_CAST).nonEmpty)
+ }
+}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/TypeCoercionResolverSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/TypeCoercionResolverSuite.scala
new file mode 100644
index 000000000000..7e0107147c9a
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/TypeCoercionResolverSuite.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.AnsiTypeCoercion
+import org.apache.spark.sql.catalyst.expressions.{Add, Cast, Expression, Literal}
+import org.apache.spark.sql.types.{DoubleType, IntegerType}
+
+class TypeCoercionResolverSuite extends SparkFunSuite with SQLConfHelper {
+
+ class HardCodedExpressionResolver(resolvedExpression: Expression)
+ extends TreeNodeResolver[Expression, Expression] {
+ override def resolve(expression: Expression): Expression = resolvedExpression
+ }
+
+ private val integerChild = Literal(1, IntegerType)
+ private val doubleChild = Literal(1.1, DoubleType)
+ private val castIntegerChild = Cast(child = integerChild, dataType = DoubleType)
+ private val expressionResolver = new HardCodedExpressionResolver(castIntegerChild)
+ private val timezoneAwareExpressionResolver = new TimezoneAwareExpressionResolver(
+ expressionResolver
+ )
+ private val typeCoercionRules = Seq(
+ AnsiTypeCoercion.ImplicitTypeCasts.transform
+ )
+ private val typeCoercionResolver =
+ new TypeCoercionResolver(timezoneAwareExpressionResolver, typeCoercionRules)
+
+ test("TypeCoercion resolution - with children reinstantiation") {
+ val expression = Add(left = doubleChild, right = integerChild)
+ val resolvedExpression = typeCoercionResolver.resolve(expression).asInstanceOf[Add]
+ // left child remains the same
+ assert(resolvedExpression.left == doubleChild)
+ // right first gets resolved to castIntegerChild. However, after the Cast gets
+ // re-resolved with timezone, it won't be equal to castIntegerChild because of re-instantiation
+ assert(resolvedExpression.right.isInstanceOf[Cast])
+ val newRightChild = resolvedExpression.right.asInstanceOf[Cast]
+ assert(newRightChild != castIntegerChild)
+ assert(newRightChild.timeZoneId.nonEmpty)
+ // not a user-specified cast
+ assert(newRightChild.getTagValue(Cast.USER_SPECIFIED_CAST).isEmpty)
+ }
+
+ test("TypeCoercion resolution - no children reinstantiation") {
+ val expression = Add(left = doubleChild, right = castIntegerChild)
+ val resolvedExpression = typeCoercionResolver.resolve(expression).asInstanceOf[Add]
+ assert(resolvedExpression.left == doubleChild)
+ // Cast that isn't a product of type coercion resolution won't be re-instantiated with timezone
+ assert(resolvedExpression.right == castIntegerChild)
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceResolver.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceResolver.scala
new file mode 100644
index 000000000000..3a2a3207a01f
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceResolver.scala
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.analysis.resolver.{
+ ExplicitlyUnsupportedResolverFeature,
+ ResolverExtension
+}
+import org.apache.spark.sql.catalyst.catalog.UnresolvedCatalogRelation
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.execution.streaming.StreamingRelation
+
+/**
+ * The [[DataSourceResolver]] is a [[Resolver]] extension that resolves nodes defined in the
+ * [[datasources]] package. We have it as an extension to avoid cyclic dependencies between
+ * [[resolver]] and [[datasources]] packages.
+ */
+class DataSourceResolver(sparkSession: SparkSession) extends ResolverExtension {
+ private val findDataSourceTable = new FindDataSourceTable(sparkSession)
+
+ /**
+ * Resolve [[UnresolvedCatalogRelation]]:
+ * - Reuse [[FindDataSourceTable]] code to resolve [[UnresolvedCatalogRelation]]
+ * - Create a new instance of [[LogicalRelation]] to regenerate the expression IDs
+ * - Explicitly disallow [[StreamingRelation]] and [[StreamingRelationV2]] for now
+ * - [[FileResolver]], which is a [[ResolverExtension]], introduces a new [[LogicalPlan]] node
+ * which resolution has to be handled here (further resolution of it doesn't need any specific
+ * resolution except adding it's attributes to the scope).
+ */
+ override def resolveOperator: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case unresolvedCatalogRelation: UnresolvedCatalogRelation =>
+ val result = findDataSourceTable.resolveUnresolvedCatalogRelation(unresolvedCatalogRelation)
+ result match {
+ case logicalRelation: LogicalRelation =>
+ logicalRelation.newInstance()
+ case streamingRelation: StreamingRelation =>
+ throw new ExplicitlyUnsupportedResolverFeature(
+ s"unsupported operator: ${streamingRelation.getClass.getName}"
+ )
+ case streamingRelationV2: StreamingRelationV2 =>
+ throw new ExplicitlyUnsupportedResolverFeature(
+ s"unsupported operator: ${streamingRelationV2.getClass.getName}"
+ )
+ case other =>
+ other
+ }
+ case logicalRelation: LogicalRelation =>
+ logicalRelation.newInstance()
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileResolver.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileResolver.scala
new file mode 100644
index 000000000000..44102da752c2
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileResolver.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.analysis.resolver.ResolverExtension
+import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper, LogicalPlan}
+
+/**
+ * The [[FileResolver]] is a [[MetadataResolver]] extension that resolves [[UnresolvedRelation]]
+ * which is created out of file. It reuses the code from [[ResolveSQLOnFile]] to resolve it
+ * properly.
+ *
+ * We have it as an extension to avoid cyclic dependencies between [[resolver]] and [[datasources]]
+ * packages.
+ */
+class FileResolver(sparkSession: SparkSession) extends ResolverExtension {
+ private val resolveSQLOnFile = new ResolveSQLOnFile(sparkSession)
+
+ /**
+ * [[ResolveSQLOnFile]] code that is reused to resolve [[UnresolvedRelation]] has
+ * [[ExpressionEncoder.resolveAndBind]] on its path which introduces another call to
+ * the analyzer which is acceptable as it is called on the leaf node of the plan. That's why we
+ * have to allow invoking transforms in the single-pass analyzer.
+ */
+ object UnresolvedRelationResolution {
+ def unapply(operator: LogicalPlan): Option[LogicalPlan] =
+ AnalysisHelper.allowInvokingTransformsInAnalyzer {
+ resolveSQLOnFile.UnresolvedRelationResolution.unapply(operator)
+ }
+ }
+
+ /**
+ * Reuse [[ResolveSQLOnFile]] code to resolve [[UnresolvedRelation]] made out of file.
+ */
+ override def resolveOperator: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case UnresolvedRelationResolution(resolvedRelation) =>
+ resolvedRelation
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
index f1033c5cded1..c78f9702557c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
@@ -53,7 +53,7 @@ class ResolveSQLOnFile(sparkSession: SparkSession) extends Rule[LogicalPlan] {
object UnresolvedRelationResolution {
def unapply(plan: LogicalPlan): Option[LogicalPlan] = {
- plan match {
+ val result = plan match {
case u: UnresolvedRelation if maybeSQLFile(u) =>
try {
val ds = resolveDataSource(u)
@@ -75,6 +75,17 @@ class ResolveSQLOnFile(sparkSession: SparkSession) extends Rule[LogicalPlan] {
case _ =>
None
}
+ result.foreach(resolvedRelation => plan match {
+ case unresolvedRelation: UnresolvedRelation =>
+ // We put the resolved relation into the [[AnalyzerBridgeState]] for
+ // it to be later reused by the single-pass [[Resolver]] to avoid resolving the
+ // relation metadata twice.
+ AnalysisContext.get.getSinglePassResolverBridgeState.map { bridgeState =>
+ bridgeState.relationsWithResolvedMetadata.put(unresolvedRelation, resolvedRelation)
+ }
+ case _ =>
+ })
+ result
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala
index 59a873ef982f..bfd591552911 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala
@@ -20,6 +20,7 @@ import org.apache.spark.annotation.Unstable
import org.apache.spark.sql.{ExperimentalMethods, SparkSession, UDFRegistration, _}
import org.apache.spark.sql.artifact.ArtifactManager
import org.apache.spark.sql.catalyst.analysis.{Analyzer, EvalSubqueriesForTimeTravel, FunctionRegistry, InvokeProcedures, ReplaceCharWithVarchar, ResolveSessionCatalog, ResolveTranspose, TableFunctionRegistry}
+import org.apache.spark.sql.catalyst.analysis.resolver.ResolverExtension
import org.apache.spark.sql.catalyst.catalog.{FunctionExpressionBuilder, SessionCatalog}
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.optimizer.Optimizer
@@ -199,6 +200,15 @@ abstract class BaseSessionStateBuilder(
protected def analyzer: Analyzer = new Analyzer(catalogManager) {
override val hintResolutionRules: Seq[Rule[LogicalPlan]] =
customHintResolutionRules
+
+ override val singlePassResolverExtensions: Seq[ResolverExtension] = Seq(
+ new DataSourceResolver(session)
+ )
+
+ override val singlePassMetadataResolverExtensions: Seq[ResolverExtension] = Seq(
+ new FileResolver(session)
+ )
+
override val extendedResolutionRules: Seq[Rule[LogicalPlan]] =
new FindDataSourceTable(session) +:
new ResolveSQLOnFile(session) +:
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ExplicitlyUnsupportedResolverFeatureSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ExplicitlyUnsupportedResolverFeatureSuite.scala
new file mode 100644
index 000000000000..7fd7d570ecfc
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ExplicitlyUnsupportedResolverFeatureSuite.scala
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.analysis.resolver
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.analysis.resolver.Resolver
+import org.apache.spark.sql.test.SharedSparkSession
+
+class ExplicitlyUnsupportedResolverFeatureSuite extends QueryTest with SharedSparkSession {
+ test("Unsupported table types") {
+ withTable("csv_table") {
+ spark.sql("CREATE TABLE csv_table (col1 INT) USING CSV;").collect()
+ checkResolution("SELECT * FROM csv_table;", shouldPass = true)
+ }
+ withTable("json_table") {
+ spark.sql("CREATE TABLE json_table (col1 INT) USING JSON;").collect()
+ checkResolution("SELECT * FROM json_table;", shouldPass = true)
+ }
+ withTable("parquet_table") {
+ spark.sql("CREATE TABLE parquet_table (col1 INT) USING PARQUET;").collect()
+ checkResolution("SELECT * FROM parquet_table;", shouldPass = true)
+ }
+ withTable("orc_table") {
+ spark.sql("CREATE TABLE orc_table (col1 INT) USING ORC;").collect()
+ checkResolution("SELECT * FROM orc_table;", shouldPass = true)
+ }
+ }
+
+ test("Unsupported view types") {
+ withTable("src_table") {
+ spark.sql("CREATE TABLE src_table (col1 INT) USING PARQUET;").collect()
+
+ withView("temporary_view") {
+ spark.sql("CREATE TEMPORARY VIEW temporary_view AS SELECT * FROM src_table;").collect()
+ checkResolution("SELECT * FROM temporary_view;")
+ }
+
+ withView("persistent_view") {
+ spark.sql("CREATE VIEW persistent_view AS SELECT * FROM src_table;").collect()
+ checkResolution("SELECT * FROM persistent_view;")
+ }
+ }
+ }
+
+ test("Unsupported char type padding") {
+ withTable("char_type_padding") {
+ spark.sql(s"CREATE TABLE t1 (c1 CHAR(3), c2 STRING) USING PARQUET")
+ checkResolution("SELECT c1 = '12', c1 = '12 ', c1 = '12 ' FROM t1 WHERE c2 = '12'")
+ }
+ }
+
+ test("Unsupported lateral column alias") {
+ checkResolution("SELECT 1 AS a, a AS b")
+ checkResolution("SELECT sum(1), `sum(1)` + 1 AS a")
+ }
+
+ private def checkResolution(sqlText: String, shouldPass: Boolean = false): Unit = {
+ def noopWrapper(body: => Unit) = body
+
+ val wrapper = if (shouldPass) {
+ noopWrapper _
+ } else {
+ intercept[Throwable] _
+ }
+
+ val unresolvedPlan = spark.sql(sqlText).queryExecution.logical
+
+ val resolver = new Resolver(
+ spark.sessionState.catalogManager,
+ extensions = spark.sessionState.analyzer.singlePassResolverExtensions
+ )
+ wrapper {
+ resolver.lookupMetadataAndResolve(unresolvedPlan)
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/HybridAnalyzerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/HybridAnalyzerSuite.scala
new file mode 100644
index 000000000000..587725093f0e
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/HybridAnalyzerSuite.scala
@@ -0,0 +1,404 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.analysis.resolver
+
+import org.scalactic.source.Position
+import org.scalatest.Tag
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.{
+ AliasIdentifier,
+ ExtendedAnalysisException,
+ QueryPlanningTracker
+}
+import org.apache.spark.sql.catalyst.analysis.{
+ AnalysisContext,
+ Analyzer,
+ UnresolvedAttribute,
+ UnresolvedStar
+}
+import org.apache.spark.sql.catalyst.analysis.resolver.{
+ AnalyzerBridgeState,
+ ExplicitlyUnsupportedResolverFeature,
+ HybridAnalyzer,
+ Resolver,
+ ResolverGuard
+}
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.plans.logical.{
+ LocalRelation,
+ LogicalPlan,
+ Project,
+ SubqueryAlias
+}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, MetadataBuilder}
+
+class HybridAnalyzerSuite extends QueryTest with SharedSparkSession {
+ private val col1Integer = AttributeReference("col1", IntegerType)()
+ private val col2Integer = AttributeReference("col2", IntegerType)()
+ private val col2IntegerWithMetadata = AttributeReference(
+ "col2",
+ IntegerType,
+ metadata = (new MetadataBuilder).putString("comment", "this is an integer").build()
+ )()
+
+ private def validateSinglePassResolverBridgeState(bridgeRelations: Boolean): Unit = {
+ assert(bridgeRelations == AnalysisContext.get.getSinglePassResolverBridgeState.isDefined)
+ }
+
+ private class BrokenResolver(ex: Throwable, bridgeRelations: Boolean)
+ extends Resolver(spark.sessionState.catalogManager) {
+ override def lookupMetadataAndResolve(
+ plan: LogicalPlan,
+ analyzerBridgeState: Option[AnalyzerBridgeState] = None): LogicalPlan = {
+ validateSinglePassResolverBridgeState(bridgeRelations)
+ throw ex
+ }
+ }
+
+ private class ValidatingResolver(bridgeRelations: Boolean)
+ extends Resolver(spark.sessionState.catalogManager) {
+ override def lookupMetadataAndResolve(
+ plan: LogicalPlan,
+ analyzerBridgeState: Option[AnalyzerBridgeState] = None): LogicalPlan = {
+ validateSinglePassResolverBridgeState(bridgeRelations)
+ super.lookupMetadataAndResolve(plan, analyzerBridgeState)
+ }
+ }
+
+ private class HardCodedResolver(resolvedPlan: LogicalPlan, bridgeRelations: Boolean)
+ extends Resolver(spark.sessionState.catalogManager) {
+ override def lookupMetadataAndResolve(
+ plan: LogicalPlan,
+ analyzerBridgeState: Option[AnalyzerBridgeState] = None): LogicalPlan = {
+ validateSinglePassResolverBridgeState(bridgeRelations)
+ resolvedPlan
+ }
+ }
+
+ private class ValidatingAnalyzer(bridgeRelations: Boolean)
+ extends Analyzer(spark.sessionState.catalogManager) {
+ override def executeAndTrack(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {
+ validateSinglePassResolverBridgeState(bridgeRelations)
+ super.executeAndTrack(plan, tracker)
+ }
+ }
+
+ private class BrokenAnalyzer(ex: Throwable, bridgeRelations: Boolean)
+ extends Analyzer(spark.sessionState.catalogManager) {
+ override def executeAndTrack(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {
+ validateSinglePassResolverBridgeState(bridgeRelations)
+ throw ex
+ }
+ }
+
+ private class CustomAnalyzer(customCode: () => Unit, bridgeRelations: Boolean)
+ extends Analyzer(spark.sessionState.catalogManager) {
+ override def executeAndTrack(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {
+ validateSinglePassResolverBridgeState(bridgeRelations)
+ customCode()
+ super.executeAndTrack(plan, tracker)
+ }
+ }
+
+ override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(
+ implicit pos: Position): Unit = {
+ super.test(testName) {
+ withSQLConf(
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "true"
+ ) {
+ testFun
+ }
+ }
+ }
+
+ test("Both fixed-point and single-pass analyzers pass") {
+ val plan: LogicalPlan = {
+ Project(
+ Seq(UnresolvedStar(None)),
+ LocalRelation(col1Integer)
+ )
+ }
+ val resolvedPlan =
+ Project(
+ Seq(col1Integer),
+ LocalRelation(Seq(col1Integer))
+ )
+ assert(
+ new HybridAnalyzer(
+ new ValidatingAnalyzer(bridgeRelations = true),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new ValidatingResolver(bridgeRelations = true)
+ ).apply(plan, null)
+ ==
+ resolvedPlan
+ )
+ }
+
+ test("Fixed-point analyzer passes, single-pass analyzer fails") {
+ val plan: LogicalPlan =
+ Project(Seq(UnresolvedStar(None)), LocalRelation(col1Integer))
+ checkError(
+ exception = intercept[AnalysisException](
+ new HybridAnalyzer(
+ new ValidatingAnalyzer(bridgeRelations = true),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new BrokenResolver(
+ QueryCompilationErrors.unsupportedSinglePassAnalyzerFeature("test"),
+ bridgeRelations = true
+ )
+ ).apply(plan, null)
+ ),
+ condition = "UNSUPPORTED_SINGLE_PASS_ANALYZER_FEATURE",
+ parameters = Map("feature" -> "test")
+ )
+ }
+
+ test("Fixed-point analyzer fails, single-pass analyzer passes") {
+ val plan: LogicalPlan =
+ Project(
+ Seq(UnresolvedAttribute("nonexistent_col")),
+ LocalRelation(col1Integer)
+ )
+ val resolvedPlan =
+ Project(
+ Seq(col1Integer),
+ LocalRelation(Seq(col1Integer))
+ )
+ checkError(
+ exception = intercept[AnalysisException](
+ new HybridAnalyzer(
+ new ValidatingAnalyzer(bridgeRelations = true),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new HardCodedResolver(resolvedPlan, bridgeRelations = true)
+ ).apply(plan, null)
+ ),
+ condition = "HYBRID_ANALYZER_EXCEPTION.FIXED_POINT_FAILED_SINGLE_PASS_SUCCEEDED",
+ parameters = Map("singlePassOutput" -> resolvedPlan.toString)
+ )
+ }
+
+ test("Both fixed-point and single-pass analyzers fail") {
+ val plan: LogicalPlan =
+ Project(
+ Seq(UnresolvedAttribute("nonexistent_col")),
+ LocalRelation(col1Integer)
+ )
+ checkError(
+ exception = intercept[ExtendedAnalysisException](
+ new HybridAnalyzer(
+ new ValidatingAnalyzer(bridgeRelations = true),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new ValidatingResolver(bridgeRelations = true)
+ ).apply(plan, null)
+ ),
+ condition = "UNRESOLVED_COLUMN.WITH_SUGGESTION",
+ parameters = Map(
+ "objectName" -> "`nonexistent_col`",
+ "proposal" -> "`col1`"
+ )
+ )
+ }
+
+ test("Plan mismatch") {
+ val plan: LogicalPlan =
+ Project(
+ Seq(UnresolvedAttribute("col1")),
+ SubqueryAlias(
+ AliasIdentifier("t", Seq.empty),
+ LocalRelation(Seq(col1Integer))
+ )
+ )
+ val resolvedPlan =
+ Project(
+ Seq(col1Integer),
+ LocalRelation(Seq(col1Integer))
+ )
+ val expectedResolvedPlan =
+ Project(
+ Seq(col1Integer),
+ SubqueryAlias(
+ AliasIdentifier("t", Seq.empty),
+ LocalRelation(Seq(col1Integer))
+ )
+ )
+ checkError(
+ exception = intercept[AnalysisException](
+ new HybridAnalyzer(
+ new ValidatingAnalyzer(bridgeRelations = true),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new HardCodedResolver(resolvedPlan, bridgeRelations = true)
+ ).apply(plan, null)
+ ),
+ condition = "HYBRID_ANALYZER_EXCEPTION.LOGICAL_PLAN_COMPARISON_MISMATCH",
+ parameters = Map(
+ "singlePassOutput" -> resolvedPlan.toString,
+ "fixedPointOutput" -> expectedResolvedPlan.toString
+ )
+ )
+ }
+
+ test("Missing metadata in output schema") {
+ val plan: LogicalPlan =
+ Project(
+ Seq(UnresolvedAttribute("col2")),
+ LocalRelation(col2IntegerWithMetadata)
+ )
+ val resolvedPlan =
+ Project(
+ Seq(col2Integer),
+ LocalRelation(Seq(col2Integer))
+ )
+ checkError(
+ exception = intercept[AnalysisException](
+ new HybridAnalyzer(
+ new ValidatingAnalyzer(bridgeRelations = true),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new HardCodedResolver(resolvedPlan, bridgeRelations = true)
+ ).apply(plan, null)
+ ),
+ condition = "HYBRID_ANALYZER_EXCEPTION.OUTPUT_SCHEMA_COMPARISON_MISMATCH",
+ parameters = Map(
+ "singlePassOutputSchema" -> "(col2,IntegerType,true,{})",
+ "fixedPointOutputSchema" -> "(col2,IntegerType,true,{\"comment\":\"this is an integer\"})"
+ )
+ )
+ }
+
+ test("Explicitly unsupported resolver feature") {
+ val plan: LogicalPlan = {
+ Project(
+ Seq(UnresolvedStar(None)),
+ LocalRelation(col1Integer)
+ )
+ }
+ checkAnswer(
+ new HybridAnalyzer(
+ new ValidatingAnalyzer(bridgeRelations = true),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new BrokenResolver(
+ new ExplicitlyUnsupportedResolverFeature("FAILURE"),
+ bridgeRelations = true
+ )
+ ).apply(plan, null),
+ plan
+ )
+ }
+
+ test("Fixed-point only run") {
+ val plan = Project(
+ Seq(UnresolvedStar(None)),
+ LocalRelation(col1Integer)
+ )
+ val resolvedPlan = Project(
+ Seq(col1Integer),
+ LocalRelation(Seq(col1Integer))
+ )
+ assert(
+ withSQLConf(
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false"
+ ) {
+ new HybridAnalyzer(
+ new ValidatingAnalyzer(bridgeRelations = false),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new BrokenResolver(
+ new Exception("Single-pass resolver should not be invoked"),
+ bridgeRelations = false
+ )
+ ).apply(plan, null)
+ }
+ ==
+ resolvedPlan
+ )
+ }
+
+ test("Single-pass only run") {
+ val plan = Project(
+ Seq(UnresolvedStar(None)),
+ LocalRelation(col1Integer)
+ )
+ val resolvedPlan = Project(
+ Seq(col1Integer),
+ LocalRelation(Seq(col1Integer))
+ )
+ assert(
+ withSQLConf(
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key -> "true"
+ ) {
+ new HybridAnalyzer(
+ new BrokenAnalyzer(
+ new Exception("Fixed-point analyzer should not be invoked"),
+ bridgeRelations = false
+ ),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new ValidatingResolver(bridgeRelations = false)
+ ).apply(plan, null)
+ }
+ ==
+ resolvedPlan
+ )
+ }
+
+ test("Nested invocations") {
+ val plan = Project(
+ Seq(UnresolvedStar(None)),
+ LocalRelation(col1Integer)
+ )
+ val resolvedPlan = Project(
+ Seq(col1Integer),
+ LocalRelation(Seq(col1Integer))
+ )
+
+ val nestedAnalysis = () => {
+ assert(
+ withSQLConf(
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key -> "true"
+ ) {
+ new HybridAnalyzer(
+ new BrokenAnalyzer(
+ new Exception("Fixed-point analyzer should not be invoked"),
+ bridgeRelations = false
+ ),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new ValidatingResolver(bridgeRelations = false)
+ ).apply(plan, null)
+ }
+ ==
+ resolvedPlan
+ )
+ }
+
+ assert(
+ new HybridAnalyzer(
+ new CustomAnalyzer(
+ customCode = () => { nestedAnalysis() },
+ bridgeRelations = true
+ ),
+ new ResolverGuard(spark.sessionState.catalogManager),
+ new ValidatingResolver(bridgeRelations = true)
+ ).apply(plan, null)
+ ==
+ resolvedPlan
+ )
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/MetadataResolverSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/MetadataResolverSuite.scala
new file mode 100644
index 000000000000..5fd21d7543b3
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/MetadataResolverSuite.scala
@@ -0,0 +1,277 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.analysis.resolver
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.{AliasIdentifier, TableIdentifier}
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.analysis.resolver.{
+ AnalyzerBridgeState,
+ BridgedRelationMetadataProvider,
+ MetadataResolver,
+ RelationId,
+ Resolver
+}
+import org.apache.spark.sql.catalyst.catalog.UnresolvedCatalogRelation
+import org.apache.spark.sql.catalyst.expressions.{Expression, PlanExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
+import org.apache.spark.sql.execution.datasources.{FileResolver, HadoopFsRelation, LogicalRelation}
+import org.apache.spark.sql.test.{SharedSparkSession, SQLTestUtils}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+class MetadataResolverSuite extends QueryTest with SharedSparkSession with SQLTestUtils {
+ private val keyValueTableSchema = StructType(
+ Seq(
+ StructField("key", IntegerType, true),
+ StructField("value", StringType, true)
+ )
+ )
+ private val fileTableSchema = StructType(
+ Seq(
+ StructField("id", LongType, true)
+ )
+ )
+
+ test("Single CSV relation") {
+ withTable("src_csv") {
+ spark.sql("CREATE TABLE src_csv (key INT, value STRING) USING CSV;").collect()
+
+ checkResolveUnresolvedCatalogRelation(
+ sqlText = "SELECT * FROM src_csv",
+ expectedTableData = Seq(createTableData("src_csv"))
+ )
+ }
+ }
+
+ test("Single ORC relation") {
+ withTable("src_orc") {
+ spark.sql("CREATE TABLE src_orc (key INT, value STRING) USING ORC;").collect()
+
+ checkResolveUnresolvedCatalogRelation(
+ sqlText = "SELECT * FROM src_orc",
+ expectedTableData = Seq(createTableData("src_orc"))
+ )
+ }
+ }
+
+ test("Relation inside an EXISTS subquery") {
+ withTable("src") {
+ spark.sql("CREATE TABLE src (key INT, value STRING) USING PARQUET;").collect()
+
+ checkResolveUnresolvedCatalogRelation(
+ sqlText = "SELECT * FROM VALUES (1) WHERE EXISTS (SELECT col1 FROM src)",
+ expectedTableData = Seq(createTableData("src"))
+ )
+ }
+ }
+
+ test("Relation inside an IN subquery") {
+ withTable("src") {
+ spark.sql("CREATE TABLE src (key INT, value STRING) USING PARQUET;").collect()
+
+ checkResolveUnresolvedCatalogRelation(
+ sqlText = "SELECT * FROM VALUES (1) WHERE col1 IN (SELECT col1 FROM src)",
+ expectedTableData = Seq(createTableData("src"))
+ )
+ }
+ }
+
+ test("Relation inside a nested subquery expression") {
+ withTable("src") {
+ spark.sql("CREATE TABLE src (key INT, value STRING) USING PARQUET;").collect()
+
+ checkResolveUnresolvedCatalogRelation(
+ sqlText = """
+ SELECT
+ col1 + (
+ SELECT 35 * (
+ SELECT key FROM src LIMIT 1
+ ) * col1 FROM VALUES (2)
+ )
+ FROM
+ VALUES (1)
+ """,
+ expectedTableData = Seq(createTableData("src"))
+ )
+ }
+ }
+
+ test("Relation from a file") {
+ val df = spark.range(100).toDF()
+ withTempPath(f => {
+ df.write.json(f.getCanonicalPath)
+ checkResolveLogicalRelation(
+ sqlText = s"select id from json.`${f.getCanonicalPath}`",
+ expectedTableData = Seq(
+ RelationId(
+ multipartIdentifier = Seq("spark_catalog", "json", s"${f.getCanonicalPath}")
+ ) -> TestTableData(
+ name = s"file:${f.getCanonicalPath}",
+ schema = fileTableSchema
+ )
+ )
+ )
+ })
+ }
+
+ test("Relation bridged from legacy Analyzer") {
+ withTable("src") {
+ spark.sql("CREATE TABLE src (key INT, value STRING) USING PARQUET;").collect()
+
+ val analyzerBridgeState = new AnalyzerBridgeState
+ analyzerBridgeState.relationsWithResolvedMetadata.put(
+ UnresolvedRelation(Seq("src")),
+ createUnresolvedCatalogRelation("src")
+ )
+
+ checkResolveUnresolvedCatalogRelation(
+ sqlText = "SELECT * FROM src",
+ expectedTableData = Seq(createTableData("src")),
+ analyzerBridgeState = Some(analyzerBridgeState)
+ )
+ }
+ }
+
+ test("Relation not bridged from legacy Analyzer") {
+ withTable("src") {
+ spark.sql("CREATE TABLE src (key INT, value STRING) USING PARQUET;").collect()
+
+ checkResolveUnresolvedCatalogRelation(
+ sqlText = "SELECT * FROM src",
+ expectedTableData = Seq.empty,
+ analyzerBridgeState = Some(new AnalyzerBridgeState)
+ )
+ }
+ }
+
+ private def checkResolveUnresolvedCatalogRelation(
+ sqlText: String,
+ expectedTableData: Seq[(RelationId, TestTableData)],
+ analyzerBridgeState: Option[AnalyzerBridgeState] = None): Unit = {
+ checkResolve(
+ sqlText,
+ expectedTableData,
+ relation =>
+ relation.asInstanceOf[UnresolvedCatalogRelation].tableMeta.identifier.unquotedString,
+ relation => relation.asInstanceOf[UnresolvedCatalogRelation].tableMeta.schema,
+ analyzerBridgeState
+ )
+ }
+
+ private def checkResolveLogicalRelation(
+ sqlText: String,
+ expectedTableData: Seq[(RelationId, TestTableData)],
+ analyzerBridgeState: Option[AnalyzerBridgeState] = None): Unit = {
+ checkResolve(
+ sqlText,
+ expectedTableData,
+ relation =>
+ relation
+ .asInstanceOf[LogicalRelation]
+ .relation
+ .asInstanceOf[HadoopFsRelation]
+ .location
+ .rootPaths
+ .mkString(","),
+ relation => relation.asInstanceOf[LogicalRelation].relation.schema,
+ analyzerBridgeState
+ )
+ }
+
+ private def checkResolve(
+ sqlText: String,
+ expectedTableData: Seq[(RelationId, TestTableData)],
+ getTableName: LogicalPlan => String,
+ getTableSchema: LogicalPlan => StructType,
+ analyzerBridgeState: Option[AnalyzerBridgeState]): Unit = {
+ val unresolvedPlan = spark.sql(sqlText).queryExecution.logical
+
+ val metadataResolver = analyzerBridgeState match {
+ case Some(analyzerBridgeState) =>
+ new BridgedRelationMetadataProvider(
+ spark.sessionState.catalogManager,
+ Resolver.createRelationResolution(spark.sessionState.catalogManager),
+ analyzerBridgeState
+ )
+ case None =>
+ val metadataResolver = new MetadataResolver(
+ spark.sessionState.catalogManager,
+ Resolver.createRelationResolution(spark.sessionState.catalogManager),
+ Seq(new FileResolver(spark))
+ )
+ metadataResolver.resolve(unresolvedPlan)
+ metadataResolver
+ }
+
+ val actualTableData = new mutable.HashMap[RelationId, TestTableData]
+
+ def findUnresolvedRelations(unresolvedPlan: LogicalPlan): Unit = unresolvedPlan.foreach {
+ case unresolvedRelation: UnresolvedRelation =>
+ metadataResolver.getRelationWithResolvedMetadata(unresolvedRelation) match {
+ case Some(plan) =>
+ val relationId = metadataResolver.relationIdFromUnresolvedRelation(unresolvedRelation)
+ val relation = plan match {
+ case SubqueryAlias(_, relation) => relation
+ case relation => relation
+ }
+
+ actualTableData(relationId) =
+ TestTableData(getTableName(relation), getTableSchema(relation))
+ case None =>
+ }
+ case unresolvedPlan =>
+ def traverseExpressions(expression: Expression): Unit = expression match {
+ case planExpression: PlanExpression[_] =>
+ planExpression.plan match {
+ case plan: LogicalPlan =>
+ findUnresolvedRelations(plan)
+ case _ =>
+ }
+ case expression =>
+ expression.children.foreach(traverseExpressions)
+ }
+
+ unresolvedPlan.expressions.foreach(traverseExpressions)
+ }
+
+ findUnresolvedRelations(unresolvedPlan)
+
+ assert(actualTableData == mutable.HashMap(expectedTableData: _*))
+ }
+
+ private def createTableData(name: String) =
+ RelationId(
+ multipartIdentifier = Seq("spark_catalog", "default", name)
+ ) -> TestTableData(
+ name = s"spark_catalog.default.$name",
+ schema = keyValueTableSchema
+ )
+
+ private def createUnresolvedCatalogRelation(name: String) = SubqueryAlias(
+ AliasIdentifier(name),
+ UnresolvedCatalogRelation(
+ spark.sessionState.catalog.getTableMetadata(TableIdentifier(name)),
+ CaseInsensitiveStringMap.empty
+ )
+ )
+
+ private case class TestTableData(name: String, schema: StructType)
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/NameScopeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/NameScopeSuite.scala
new file mode 100644
index 000000000000..ec744af89f00
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/NameScopeSuite.scala
@@ -0,0 +1,659 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.analysis.resolver
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.UnresolvedStar
+import org.apache.spark.sql.catalyst.analysis.resolver.{NameScope, NameScopeStack, NameTarget}
+import org.apache.spark.sql.catalyst.expressions.{
+ AttributeReference,
+ GetArrayItem,
+ GetArrayStructFields,
+ GetMapValue,
+ GetStructField,
+ Literal
+}
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.types.{
+ ArrayType,
+ BooleanType,
+ IntegerType,
+ MapType,
+ StringType,
+ StructField,
+ StructType
+}
+
+class NameScopeSuite extends PlanTest with SQLConfHelper {
+ private val col1Integer = AttributeReference(name = "col1", dataType = IntegerType)()
+ private val col1IntegerOther = AttributeReference(name = "col1", dataType = IntegerType)()
+ private val col2Integer = AttributeReference(name = "col2", dataType = IntegerType)()
+ private val col3Boolean = AttributeReference(name = "col3", dataType = BooleanType)()
+ private val col4String = AttributeReference(name = "col4", dataType = StringType)()
+ private val col5String = AttributeReference(name = "col5", dataType = StringType)()
+ private val col6IntegerWithQualifier = AttributeReference(
+ name = "col6",
+ dataType = IntegerType
+ )(qualifier = Seq("catalog", "database", "table"))
+ private val col6IntegerOtherWithQualifier = AttributeReference(
+ name = "col6",
+ dataType = IntegerType
+ )(qualifier = Seq("catalog", "database", "table"))
+ private val col7StringWithQualifier = AttributeReference(
+ name = "col7",
+ dataType = IntegerType
+ )(qualifier = Seq("catalog", "database", "table"))
+ private val col8Struct = AttributeReference(
+ name = "col8",
+ dataType = StructType(Seq(StructField("field", IntegerType, true)))
+ )()
+ private val col9NestedStruct = AttributeReference(
+ name = "col9",
+ dataType = StructType(
+ Seq(
+ StructField(
+ "field",
+ StructType(
+ Seq(
+ StructField("subfield", IntegerType)
+ )
+ )
+ )
+ )
+ )
+ )()
+ private val col10Map = AttributeReference(
+ name = "col10",
+ dataType = MapType(StringType, IntegerType)
+ )()
+ private val col11MapWithStruct = AttributeReference(
+ name = "col11",
+ dataType = MapType(
+ StringType,
+ StructType(Seq(StructField("field", StringType)))
+ )
+ )()
+ private val col12Array = AttributeReference(
+ name = "col12",
+ dataType = ArrayType(IntegerType)
+ )()
+ private val col13ArrayWithStruct = AttributeReference(
+ name = "col13",
+ dataType = ArrayType(
+ StructType(Seq(StructField("field", StringType)))
+ )
+ )()
+
+ test("Empty scope") {
+ val nameScope = new NameScope
+
+ assert(nameScope.getAllAttributes.isEmpty)
+
+ assert(nameScope.matchMultipartName(Seq("col1")) == NameTarget(candidates = Seq.empty))
+ }
+
+ test("Single unnamed plan") {
+ val nameScope = new NameScope
+
+ nameScope += Seq(col1Integer, col2Integer, col3Boolean)
+
+ assert(nameScope.getAllAttributes == Seq(col1Integer, col2Integer, col3Boolean))
+
+ assert(
+ nameScope.matchMultipartName(Seq("col1")) == NameTarget(
+ candidates = Seq(col1Integer),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col2")) == NameTarget(
+ candidates = Seq(col2Integer),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col3")) == NameTarget(
+ candidates = Seq(col3Boolean),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col4")) == NameTarget(
+ candidates = Seq.empty,
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean)
+ )
+ )
+ }
+
+ test("Several unnamed plans") {
+ val nameScope = new NameScope
+
+ nameScope += Seq(col1Integer)
+ nameScope += Seq(col2Integer, col3Boolean)
+ nameScope += Seq(col4String)
+
+ assert(nameScope.getAllAttributes == Seq(col1Integer, col2Integer, col3Boolean, col4String))
+
+ assert(
+ nameScope.matchMultipartName(Seq("col1")) == NameTarget(
+ candidates = Seq(col1Integer),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col2")) == NameTarget(
+ candidates = Seq(col2Integer),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col3")) == NameTarget(
+ candidates = Seq(col3Boolean),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col4")) == NameTarget(
+ candidates = Seq(col4String),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col5")) == NameTarget(
+ candidates = Seq.empty,
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ }
+
+ test("Single named plan") {
+ val nameScope = new NameScope
+
+ nameScope("table1") = Seq(col1Integer, col2Integer, col3Boolean)
+
+ assert(nameScope.getAllAttributes == Seq(col1Integer, col2Integer, col3Boolean))
+
+ assert(
+ nameScope.matchMultipartName(Seq("col1")) == NameTarget(
+ candidates = Seq(col1Integer),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col2")) == NameTarget(
+ candidates = Seq(col2Integer),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col3")) == NameTarget(
+ candidates = Seq(col3Boolean),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col4")) == NameTarget(
+ candidates = Seq.empty,
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean)
+ )
+ )
+ }
+
+ test("Several named plans") {
+ val nameScope = new NameScope
+
+ nameScope("table1") = Seq(col1Integer)
+ nameScope("table2") = Seq(col2Integer, col3Boolean)
+ nameScope("table2") = Seq(col4String)
+ nameScope("table3") = Seq(col5String)
+
+ assert(
+ nameScope.getAllAttributes == Seq(
+ col1Integer,
+ col2Integer,
+ col3Boolean,
+ col4String,
+ col5String
+ )
+ )
+
+ assert(
+ nameScope.matchMultipartName(Seq("col1")) == NameTarget(
+ candidates = Seq(col1Integer),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String, col5String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col2")) == NameTarget(
+ candidates = Seq(col2Integer),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String, col5String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col3")) == NameTarget(
+ candidates = Seq(col3Boolean),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String, col5String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col4")) == NameTarget(
+ candidates = Seq(col4String),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String, col5String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col5")) == NameTarget(
+ candidates = Seq(col5String),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String, col5String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col6")) == NameTarget(
+ candidates = Seq.empty,
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String, col5String)
+ )
+ )
+ }
+
+ test("Named and unnamed plans with case insensitive comparison") {
+ val col1Integer = AttributeReference(name = "Col1", dataType = IntegerType)()
+ val col2Integer = AttributeReference(name = "col2", dataType = IntegerType)()
+ val col3Boolean = AttributeReference(name = "coL3", dataType = BooleanType)()
+ val col4String = AttributeReference(name = "Col4", dataType = StringType)()
+
+ val nameScope = new NameScope
+
+ nameScope("TaBle1") = Seq(col1Integer)
+ nameScope("table2") = Seq(col2Integer, col3Boolean)
+ nameScope += Seq(col4String)
+
+ assert(nameScope.getAllAttributes == Seq(col1Integer, col2Integer, col3Boolean, col4String))
+
+ assert(
+ nameScope.matchMultipartName(Seq("cOL1")) == NameTarget(
+ candidates = Seq(col1Integer.withName("cOL1")),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("CoL2")) == NameTarget(
+ candidates = Seq(col2Integer.withName("CoL2")),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col3")) == NameTarget(
+ candidates = Seq(col3Boolean.withName("col3")),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("COL4")) == NameTarget(
+ candidates = Seq(col4String.withName("COL4")),
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ assert(
+ nameScope.matchMultipartName(Seq("col5")) == NameTarget(
+ candidates = Seq.empty,
+ allAttributes = Seq(col1Integer, col2Integer, col3Boolean, col4String)
+ )
+ )
+ }
+
+ test("Duplicate attribute names from one plan") {
+ val nameScope = new NameScope
+
+ nameScope("table1") = Seq(col1Integer, col1Integer)
+ nameScope("table1") = Seq(col1IntegerOther)
+
+ assert(nameScope.getAllAttributes == Seq(col1Integer, col1Integer, col1IntegerOther))
+
+ nameScope.matchMultipartName(Seq("col1")) == NameTarget(
+ candidates = Seq(col1Integer, col1IntegerOther)
+ )
+ }
+
+ test("Duplicate attribute names from several plans") {
+ val nameScope = new NameScope
+
+ nameScope("table1") = Seq(col1Integer, col1IntegerOther)
+ nameScope("table2") = Seq(col1Integer, col1IntegerOther)
+
+ assert(
+ nameScope.getAllAttributes == Seq(
+ col1Integer,
+ col1IntegerOther,
+ col1Integer,
+ col1IntegerOther
+ )
+ )
+
+ nameScope.matchMultipartName(Seq("col1")) == NameTarget(
+ candidates = Seq(
+ col1Integer,
+ col1IntegerOther,
+ col1Integer,
+ col1IntegerOther
+ )
+ )
+ }
+
+ test("Expand star") {
+ val nameScope = new NameScope
+
+ nameScope("table") =
+ Seq(col6IntegerWithQualifier, col6IntegerOtherWithQualifier, col7StringWithQualifier)
+
+ Seq(Seq("table"), Seq("database", "table"), Seq("catalog", "database", "table"))
+ .foreach(tableQualifier => {
+ assert(
+ nameScope.expandStar(UnresolvedStar(Some(tableQualifier)))
+ == Seq(col6IntegerWithQualifier, col6IntegerOtherWithQualifier, col7StringWithQualifier)
+ )
+ })
+
+ checkError(
+ exception = intercept[AnalysisException](
+ nameScope.expandStar(UnresolvedStar(Some(Seq("database", "table_fail"))))
+ ),
+ condition = "CANNOT_RESOLVE_STAR_EXPAND",
+ parameters = Map(
+ "targetString" -> "`database`.`table_fail`",
+ "columns" -> "`col6`, `col6`, `col7`"
+ )
+ )
+
+ nameScope("table2") = Seq(col6IntegerWithQualifier)
+
+ checkError(
+ exception = intercept[AnalysisException](
+ nameScope.expandStar(UnresolvedStar(Some(Seq("table2"))))
+ ),
+ condition = "INVALID_USAGE_OF_STAR_OR_REGEX",
+ parameters = Map(
+ "elem" -> "'*'",
+ "prettyName" -> "query"
+ )
+ )
+ }
+
+ test("Multipart attribute names") {
+ val nameScope = new NameScope
+
+ nameScope("table") = Seq(col6IntegerWithQualifier)
+
+ for (multipartIdentifier <- Seq(
+ Seq("catalog", "database", "table", "col6"),
+ Seq("database", "table", "col6"),
+ Seq("table", "col6")
+ )) {
+ assert(
+ nameScope.matchMultipartName(multipartIdentifier) == NameTarget(
+ candidates = Seq(
+ col6IntegerWithQualifier
+ ),
+ allAttributes = Seq(col6IntegerWithQualifier)
+ )
+ )
+ }
+
+ for (multipartIdentifier <- Seq(
+ Seq("catalog.database.table", "col6"),
+ Seq("`database`.`table`.`col6`"),
+ Seq("table.col6")
+ )) {
+ assert(
+ nameScope.matchMultipartName(multipartIdentifier) == NameTarget(
+ candidates = Seq.empty,
+ allAttributes = Seq(col6IntegerWithQualifier)
+ )
+ )
+ }
+ }
+
+ test("Nested fields") {
+ val nameScope = new NameScope
+
+ nameScope("table") = Seq(
+ col8Struct,
+ col9NestedStruct,
+ col10Map,
+ col11MapWithStruct,
+ col12Array,
+ col13ArrayWithStruct
+ )
+
+ var matchedStructs = nameScope.matchMultipartName(Seq("col8", "field"))
+ assert(
+ matchedStructs == NameTarget(
+ candidates = Seq(
+ GetStructField(col8Struct, 0, Some("field"))
+ ),
+ aliasName = Some("field"),
+ allAttributes = Seq(
+ col8Struct,
+ col9NestedStruct,
+ col10Map,
+ col11MapWithStruct,
+ col12Array,
+ col13ArrayWithStruct
+ )
+ )
+ )
+
+ matchedStructs = nameScope.matchMultipartName(Seq("col9", "field", "subfield"))
+ assert(
+ matchedStructs == NameTarget(
+ candidates = Seq(
+ GetStructField(
+ GetStructField(
+ col9NestedStruct,
+ 0,
+ Some("field")
+ ),
+ 0,
+ Some("subfield")
+ )
+ ),
+ aliasName = Some("subfield"),
+ allAttributes = Seq(
+ col8Struct,
+ col9NestedStruct,
+ col10Map,
+ col11MapWithStruct,
+ col12Array,
+ col13ArrayWithStruct
+ )
+ )
+ )
+
+ var matchedMaps = nameScope.matchMultipartName(Seq("col10", "key"))
+ assert(
+ matchedMaps == NameTarget(
+ candidates = Seq(GetMapValue(col10Map, Literal("key"))),
+ aliasName = Some("key"),
+ allAttributes = Seq(
+ col8Struct,
+ col9NestedStruct,
+ col10Map,
+ col11MapWithStruct,
+ col12Array,
+ col13ArrayWithStruct
+ )
+ )
+ )
+
+ matchedMaps = nameScope.matchMultipartName(Seq("col11", "key"))
+ assert(
+ matchedMaps == NameTarget(
+ candidates = Seq(GetMapValue(col11MapWithStruct, Literal("key"))),
+ aliasName = Some("key"),
+ allAttributes = Seq(
+ col8Struct,
+ col9NestedStruct,
+ col10Map,
+ col11MapWithStruct,
+ col12Array,
+ col13ArrayWithStruct
+ )
+ )
+ )
+
+ var matchedArrays = nameScope.matchMultipartName(Seq("col12", "element"))
+ assert(
+ matchedArrays == NameTarget(
+ candidates = Seq(GetArrayItem(col12Array, Literal("element"))),
+ aliasName = Some("element"),
+ allAttributes = Seq(
+ col8Struct,
+ col9NestedStruct,
+ col10Map,
+ col11MapWithStruct,
+ col12Array,
+ col13ArrayWithStruct
+ )
+ )
+ )
+
+ matchedArrays = nameScope.matchMultipartName(Seq("col13", "field"))
+ assert(
+ matchedArrays == NameTarget(
+ candidates = Seq(
+ GetArrayStructFields(
+ col13ArrayWithStruct,
+ StructField("field", StringType, true),
+ 0,
+ 1,
+ true
+ )
+ ),
+ aliasName = Some("field"),
+ allAttributes = Seq(
+ col8Struct,
+ col9NestedStruct,
+ col10Map,
+ col11MapWithStruct,
+ col12Array,
+ col13ArrayWithStruct
+ )
+ )
+ )
+
+ nameScope("table2") = Seq(col8Struct)
+ matchedStructs = nameScope.matchMultipartName(Seq("col8", "field"))
+ assert(
+ matchedStructs == NameTarget(
+ candidates = Seq(
+ GetStructField(
+ col8Struct,
+ 0,
+ Some("field")
+ ),
+ GetStructField(
+ col8Struct,
+ 0,
+ Some("field")
+ )
+ ),
+ aliasName = Some("field"),
+ allAttributes = Seq(
+ col8Struct,
+ col9NestedStruct,
+ col10Map,
+ col11MapWithStruct,
+ col12Array,
+ col13ArrayWithStruct,
+ col8Struct
+ )
+ )
+ )
+ }
+}
+
+class NameScopeStackSuite extends PlanTest {
+ private val col1Integer = AttributeReference(name = "col1", dataType = IntegerType)()
+ private val col2String = AttributeReference(name = "col2", dataType = StringType)()
+ private val col3Integer = AttributeReference(name = "col3", dataType = IntegerType)()
+ private val col4String = AttributeReference(name = "col4", dataType = StringType)()
+
+ test("Empty stack") {
+ val stack = new NameScopeStack
+
+ assert(stack.top.getAllAttributes.isEmpty)
+ }
+
+ test("Overwrite top of the stack containing single scope") {
+ val stack = new NameScopeStack
+
+ stack.top.update("table1", Seq(col1Integer, col2String))
+ assert(stack.top.getAllAttributes == Seq(col1Integer, col2String))
+
+ stack.overwriteTop("table2", Seq(col3Integer, col4String))
+ assert(stack.top.getAllAttributes == Seq(col3Integer, col4String))
+
+ stack.overwriteTop(Seq(col2String))
+ assert(stack.top.getAllAttributes == Seq(col2String))
+ }
+
+ test("Overwrite top of the stack containing several scopes") {
+ val stack = new NameScopeStack
+
+ stack.top.update("table2", Seq(col3Integer))
+
+ stack.withNewScope {
+ assert(stack.top.getAllAttributes.isEmpty)
+
+ stack.top.update("table1", Seq(col1Integer, col2String))
+ assert(stack.top.getAllAttributes == Seq(col1Integer, col2String))
+
+ stack.overwriteTop("table2", Seq(col3Integer, col4String))
+ assert(stack.top.getAllAttributes == Seq(col3Integer, col4String))
+
+ stack.overwriteTop(Seq(col2String))
+ assert(stack.top.getAllAttributes == Seq(col2String))
+ }
+ }
+
+ test("Scope stacking") {
+ val stack = new NameScopeStack
+
+ stack.top.update("table1", Seq(col1Integer))
+
+ stack.withNewScope {
+ stack.top.update("table2", Seq(col2String))
+
+ stack.withNewScope {
+ stack.top.update("table3", Seq(col3Integer))
+
+ stack.withNewScope {
+ stack.top.update("table4", Seq(col4String))
+
+ assert(stack.top.getAllAttributes == Seq(col4String))
+ }
+
+ assert(stack.top.getAllAttributes == Seq(col3Integer))
+ }
+
+ assert(stack.top.getAllAttributes == Seq(col2String))
+ }
+
+ assert(stack.top.getAllAttributes == Seq(col1Integer))
+ }
+}
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
new file mode 100644
index 000000000000..d512adbb0af3
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.analysis.resolver
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.analysis.resolver.ResolverGuard
+import org.apache.spark.sql.test.SharedSparkSession
+
+class ResolverGuardSuite extends QueryTest with SharedSparkSession {
+
+ // Queries that should pass the OperatorResolverGuard
+
+ test("Select * from an inline table") {
+ checkResolverGuard("SELECT * FROM VALUES(1,2,3)", shouldPass = true)
+ }
+
+ test("Select the named parameters from an inline table") {
+ checkResolverGuard("SELECT col1,col2,col3 FROM VALUES(1,2,3)", shouldPass = true)
+ }
+
+ test("Inline table as a top level operator") {
+ checkResolverGuard("VALUES(1,2,3)", shouldPass = true)
+ }
+
+ test("Select one row") {
+ checkResolverGuard("SELECT 'Hello world!'", shouldPass = true)
+ }
+
+ test("Where clause with a literal") {
+ checkResolverGuard(
+ "SELECT * FROM VALUES(1, 2, false), (3, 4, true) WHERE true",
+ shouldPass = true
+ )
+ }
+
+ test("Where clause with an attribute") {
+ checkResolverGuard(
+ "SELECT * FROM VALUES(1, 2, false), (3, 4, true) WHERE col3",
+ shouldPass = true
+ )
+ }
+
+ test("Explicit cast with auto-alias") {
+ checkResolverGuard(
+ "SELECT CAST(1 AS DECIMAL(3,2))",
+ shouldPass = true
+ )
+ }
+
+ test("Multipart attribute name") {
+ checkResolverGuard("SELECT table.col1 FROM VALUES(1) AS table", shouldPass = true)
+ }
+
+ test("Predicates") {
+ checkResolverGuard("SELECT true and false", shouldPass = true)
+ checkResolverGuard("SELECT true or false", shouldPass = true)
+ checkResolverGuard(
+ "SELECT col1 from VALUES(1,2) where true and false or true",
+ shouldPass = true
+ )
+ checkResolverGuard("SELECT 1 = 2", shouldPass = true)
+ checkResolverGuard("SELECT 1 != 2", shouldPass = true)
+ checkResolverGuard("SELECT 1 IN (1,2,3)", shouldPass = true)
+ checkResolverGuard("SELECT 1 NOT IN (1,2,3)", shouldPass = true)
+ checkResolverGuard("SELECT 1 IS NULL", shouldPass = true)
+ checkResolverGuard("SELECT 1 IS NOT NULL", shouldPass = true)
+ checkResolverGuard("SELECT INTERVAL '1' DAY > INTERVAL '1' HOUR", shouldPass = true)
+ }
+
+ test("Star target") {
+ checkResolverGuard("SELECT table.* FROM VALUES(1) as table", shouldPass = true)
+ }
+
+ test("Binary arithmetic") {
+ checkResolverGuard("SELECT col1+col2 FROM VALUES(1,2)", shouldPass = true)
+ checkResolverGuard("SELECT 1 + 2.3 / 2 - 3 DIV 2 + 3.0 * 10.0", shouldPass = true)
+ checkResolverGuard(
+ "SELECT TIMESTAMP'2011-11-11 11:11:11' - TIMESTAMP'2011-11-11 11:11:10'",
+ shouldPass = true
+ )
+ checkResolverGuard(
+ "SELECT DATE'2020-01-01' - TIMESTAMP'2019-10-06 10:11:12.345678'",
+ shouldPass = true
+ )
+ checkResolverGuard("SELECT DATE'2012-01-01' - INTERVAL 3 HOURS", shouldPass = true)
+ checkResolverGuard(
+ "SELECT DATE'2012-01-01' + INTERVAL '12:12:12' HOUR TO SECOND",
+ shouldPass = true
+ )
+ checkResolverGuard("SELECT DATE'2012-01-01' + 1", shouldPass = true)
+ checkResolverGuard("SELECT 2 * INTERVAL 2 YEAR", shouldPass = true)
+ }
+
+ test("Supported recursive types") {
+ Seq("ARRAY", "MAP", "STRUCT").foreach { typeName =>
+ checkResolverGuard(
+ s"SELECT col1 FROM VALUES($typeName(1,2),3)",
+ shouldPass = true
+ )
+ }
+ }
+
+ test("Recursive types related functions") {
+ checkResolverGuard("SELECT NAMED_STRUCT('a', 1)", shouldPass = true)
+ checkResolverGuard("SELECT MAP_CONTAINS_KEY(MAP(1, 'a', 2, 'b'), 2)", shouldPass = true)
+ checkResolverGuard("SELECT ARRAY_CONTAINS(ARRAY(1, 2, 3), 2);", shouldPass = true)
+ }
+
+ test("Conditional expressions") {
+ checkResolverGuard("SELECT COALESCE(NULL, 1)", shouldPass = true)
+ checkResolverGuard("SELECT col1, IF(col1 > 1, 1, 0) FROM VALUES(1,2),(2,3)", shouldPass = true)
+ checkResolverGuard(
+ "SELECT col1, CASE WHEN col1 > 1 THEN 1 ELSE 0 END FROM VALUES(1,2),(2,3)",
+ shouldPass = true
+ )
+ }
+
+ test("User specified alias") {
+ checkResolverGuard("SELECT 1 AS alias", shouldPass = true)
+ }
+
+ // Queries that shouldn't pass the OperatorResolverGuard
+
+ test("Select from table") {
+ withTable("test_table") {
+ sql("CREATE TABLE test_table (col1 INT, col2 INT)")
+ checkResolverGuard("SELECT * FROM test_table", shouldPass = true)
+ }
+ }
+
+ test("Single-layer subquery") {
+ checkResolverGuard("SELECT * FROM (SELECT * FROM VALUES(1))", shouldPass = true)
+ }
+
+ test("Multi-layer subquery") {
+ checkResolverGuard("SELECT * FROM (SELECT * FROM (SELECT * FROM VALUES(1)))", shouldPass = true)
+ }
+
+ test("Scalar subquery") {
+ checkResolverGuard("SELECT (SELECT * FROM VALUES(1))", shouldPass = false)
+ }
+
+ test("EXISTS subquery") {
+ checkResolverGuard(
+ "SELECT * FROM VALUES (1) WHERE EXISTS (SELECT * FROM VALUES(1))",
+ shouldPass = false
+ )
+ }
+
+ test("IN subquery") {
+ checkResolverGuard(
+ "SELECT * FROM VALUES (1) WHERE col1 IN (SELECT * FROM VALUES(1))",
+ shouldPass = false
+ )
+ }
+
+ test("Function") {
+ checkResolverGuard("SELECT current_date()", shouldPass = false)
+ }
+
+ test("Function without the braces") {
+ checkResolverGuard("SELECT current_date", shouldPass = false)
+ }
+
+ test("Session variables") {
+ withSessionVariable {
+ checkResolverGuard("SELECT session_variable", shouldPass = false)
+ }
+ }
+
+ test("Case sensitive analysis") {
+ withSQLConf("spark.sql.caseSensitive" -> "true") {
+ checkResolverGuard("SELECT 1", shouldPass = false)
+ }
+ }
+
+ private def checkResolverGuard(query: String, shouldPass: Boolean): Unit = {
+ val resolverGuard = new ResolverGuard(spark.sessionState.catalogManager)
+ assert(
+ resolverGuard.apply(sql(query).queryExecution.logical) == shouldPass
+ )
+ }
+
+ private def withSessionVariable(body: => Unit): Unit = {
+ sql("DECLARE session_variable = 1;")
+ try {
+ body
+ } finally {
+ sql("DROP TEMPORARY VARIABLE session_variable;")
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverSuite.scala
new file mode 100644
index 000000000000..057724758d33
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverSuite.scala
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.analysis.resolver
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
+import org.apache.spark.sql.catalyst.analysis.resolver.{Resolver, ResolverExtension}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.IntegerType
+
+class ResolverSuite extends QueryTest with SharedSparkSession {
+ private val col1Integer = AttributeReference("col1", IntegerType)()
+
+ test("Node matched the extension") {
+ val resolver = createResolver(
+ Seq(
+ new NoopResolver,
+ new TestRelationResolver
+ )
+ )
+
+ val result = resolver.lookupMetadataAndResolve(
+ Project(
+ Seq(UnresolvedAttribute("col1")),
+ TestRelation(resolutionDone = false, output = Seq(col1Integer))
+ )
+ )
+ assert(
+ result == Project(
+ Seq(col1Integer),
+ TestRelation(resolutionDone = true, output = Seq(col1Integer))
+ )
+ )
+ }
+
+ test("Node didn't match the extension") {
+ val resolver = createResolver(
+ Seq(
+ new NoopResolver,
+ new TestRelationResolver
+ )
+ )
+
+ checkError(
+ exception = intercept[AnalysisException](
+ resolver.lookupMetadataAndResolve(
+ Project(
+ Seq(UnresolvedAttribute("col1")),
+ UnknownRelation(output = Seq(col1Integer))
+ )
+ )
+ ),
+ condition = "UNSUPPORTED_SINGLE_PASS_ANALYZER_FEATURE",
+ parameters = Map(
+ "feature" -> ("class " +
+ "org.apache.spark.sql.analysis.resolver.ResolverSuite$UnknownRelation operator resolution")
+ )
+ )
+ }
+
+ test("Ambiguous extensions") {
+ val resolver = createResolver(
+ Seq(
+ new NoopResolver,
+ new TestRelationResolver,
+ new TestRelationBrokenResolver
+ )
+ )
+
+ checkError(
+ exception = intercept[AnalysisException](
+ resolver.lookupMetadataAndResolve(
+ Project(
+ Seq(UnresolvedAttribute("col1")),
+ TestRelation(resolutionDone = false, output = Seq(col1Integer))
+ )
+ )
+ ),
+ condition = "AMBIGUOUS_RESOLVER_EXTENSION",
+ parameters = Map(
+ "operator" -> "org.apache.spark.sql.analysis.resolver.ResolverSuite$TestRelation",
+ "extensions" -> "TestRelationResolver, TestRelationBrokenResolver"
+ )
+ )
+ }
+
+ private def createResolver(extensions: Seq[ResolverExtension] = Seq.empty): Resolver = {
+ new Resolver(spark.sessionState.catalogManager, extensions)
+ }
+
+ private class TestRelationResolver extends ResolverExtension {
+ var timesCalled = 0
+
+ override def resolveOperator: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case testNode: TestRelation if countTimesCalled() =>
+ testNode.copy(resolutionDone = true)
+ }
+
+ private def countTimesCalled(): Boolean = {
+ timesCalled += 1
+ assert(timesCalled == 1)
+ true
+ }
+ }
+
+ private class TestRelationBrokenResolver extends ResolverExtension {
+ override def resolveOperator: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case testNode: TestRelation =>
+ assert(false)
+ testNode
+ }
+ }
+
+ private class NoopResolver extends ResolverExtension {
+ override def resolveOperator: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case node: LogicalPlan if false =>
+ assert(false)
+ node
+ }
+ }
+
+ private case class TestRelation(
+ resolutionDone: Boolean,
+ override val output: Seq[Attribute],
+ override val children: Seq[LogicalPlan] = Seq.empty)
+ extends LogicalPlan {
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[LogicalPlan]): TestRelation =
+ copy(children = newChildren)
+ }
+
+ private case class UnknownRelation(
+ override val output: Seq[Attribute],
+ override val children: Seq[LogicalPlan] = Seq.empty)
+ extends LogicalPlan {
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[LogicalPlan]): UnknownRelation =
+ copy(children = newChildren)
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/TracksResolvedNodesSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/TracksResolvedNodesSuite.scala
new file mode 100644
index 000000000000..28ccebc89bc5
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/TracksResolvedNodesSuite.scala
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.analysis.resolver
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.analysis.FunctionResolution
+import org.apache.spark.sql.catalyst.analysis.resolver.{
+ ExpressionResolver,
+ NameScopeStack,
+ Resolver
+}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, ExprId}
+import org.apache.spark.sql.catalyst.plans.logical.{OneRowRelation, Project}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{BooleanType, StringType}
+
+class TracksResolvedNodesSuite extends QueryTest with SharedSparkSession {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ spark.conf.set(SQLConf.ANALYZER_SINGLE_PASS_TRACK_RESOLVED_NODES_ENABLED.key, "true")
+ }
+
+ test("Single-pass contract preserved for equal expressions with different memory addresses") {
+ val expressionResolver = createExpressionResolver()
+ val columnObjFirst =
+ AttributeReference(name = "column", dataType = BooleanType)(exprId = ExprId(0))
+ val columnObjSecond =
+ AttributeReference(name = "column", dataType = BooleanType)(exprId = ExprId(0))
+
+ expressionResolver.resolve(columnObjFirst)
+ expressionResolver.resolve(columnObjSecond)
+ }
+
+ test("Single-pass contract broken for operators") {
+ val resolver = createResolver()
+
+ val project = Project(
+ projectList = Seq(),
+ child = Project(
+ projectList = Seq(),
+ child = OneRowRelation()
+ )
+ )
+
+ val resolvedProject = resolver.lookupMetadataAndResolve(project)
+
+ checkError(
+ exception = intercept[SparkException]({
+ resolver.lookupMetadataAndResolve(resolvedProject.children.head)
+ }),
+ condition = "INTERNAL_ERROR",
+ parameters = Map(
+ "message" -> ("Single-pass resolver attempted to resolve the same " +
+ "node more than once: Project\n+- OneRowRelation\n")
+ )
+ )
+ checkError(
+ exception = intercept[SparkException]({
+ resolver.lookupMetadataAndResolve(resolvedProject)
+ }),
+ condition = "INTERNAL_ERROR",
+ parameters = Map(
+ "message" -> ("Single-pass resolver attempted to resolve the same " +
+ "node more than once: Project\n+- Project\n +- OneRowRelation\n")
+ )
+ )
+ }
+
+ test("Single-pass contract broken for expressions") {
+ val expressionResolver = createExpressionResolver()
+
+ val cast = Cast(
+ child = AttributeReference(name = "column", dataType = BooleanType)(exprId = ExprId(0)),
+ dataType = StringType
+ )
+
+ val resolvedCast = expressionResolver.resolve(cast)
+
+ checkError(
+ exception = intercept[SparkException]({
+ expressionResolver.resolve(resolvedCast.children.head)
+ }),
+ condition = "INTERNAL_ERROR",
+ parameters = Map(
+ "message" -> ("Single-pass resolver attempted " +
+ "to resolve the same node more than once: column#0")
+ )
+ )
+ checkError(
+ exception = intercept[SparkException]({
+ expressionResolver.resolve(resolvedCast)
+ }),
+ condition = "INTERNAL_ERROR",
+ parameters = Map(
+ "message" -> ("Single-pass resolver attempted " +
+ "to resolve the same node more than once: cast(column#0 as string)")
+ )
+ )
+ }
+
+ private def createResolver(): Resolver = {
+ new Resolver(spark.sessionState.catalogManager)
+ }
+
+ private def createExpressionResolver(): ExpressionResolver = {
+ new ExpressionResolver(
+ createResolver(),
+ new NameScopeStack,
+ new FunctionResolution(
+ spark.sessionState.catalogManager,
+ Resolver.createRelationResolution(spark.sessionState.catalogManager)
+ )
+ )
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/DataSourceResolverSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/DataSourceResolverSuite.scala
new file mode 100644
index 000000000000..016c1e2f5457
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/DataSourceResolverSuite.scala
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.analysis.resolver.{MetadataResolver, Resolver}
+import org.apache.spark.sql.catalyst.catalog.UnresolvedCatalogRelation
+import org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+
+class DataSourceResolverSuite extends QueryTest with SharedSparkSession {
+ private val keyValueTableSchema = StructType(
+ Seq(
+ StructField("key", IntegerType, true),
+ StructField("value", StringType, true)
+ )
+ )
+
+ test("CSV relation") {
+ withTable("src_csv") {
+ spark.sql("CREATE TABLE src_csv (key INT, value STRING) USING CSV;").collect()
+
+ checkResolveOperator(
+ sqlText = "SELECT * FROM src_csv",
+ expectedTableName = "spark_catalog.default.src_csv",
+ expectedTableSchema = keyValueTableSchema
+ )
+ }
+ }
+
+ test("JSON relation") {
+ withTable("src_json") {
+ spark.sql("CREATE TABLE src_json (key INT, value STRING) USING JSON;").collect()
+
+ checkResolveOperator(
+ sqlText = "SELECT * FROM src_json",
+ expectedTableName = "spark_catalog.default.src_json",
+ expectedTableSchema = keyValueTableSchema
+ )
+ }
+ }
+
+ test("PARQUET relation") {
+ withTable("src_parquet") {
+ spark.sql("CREATE TABLE src_parquet (key INT, value STRING) USING PARQUET;").collect()
+
+ checkResolveOperator(
+ sqlText = "SELECT * FROM src_parquet",
+ expectedTableName = "spark_catalog.default.src_parquet",
+ expectedTableSchema = keyValueTableSchema
+ )
+ }
+ }
+
+ test("ORC relation") {
+ withTable("src_orc") {
+ spark.sql("CREATE TABLE src_orc (key INT, value STRING) USING ORC;").collect()
+
+ checkResolveOperator(
+ sqlText = "SELECT * FROM src_orc",
+ expectedTableName = "spark_catalog.default.src_orc",
+ expectedTableSchema = keyValueTableSchema
+ )
+ }
+ }
+
+ private def checkResolveOperator(
+ sqlText: String,
+ expectedTableName: String,
+ expectedTableSchema: StructType) = {
+ val metadataResolver = new MetadataResolver(
+ spark.sessionState.catalogManager,
+ Resolver.createRelationResolution(spark.sessionState.catalogManager)
+ )
+ val dataSourceResolver = new DataSourceResolver(spark)
+
+ val unresolvedPlan = spark.sql(sqlText).queryExecution.logical
+
+ metadataResolver.resolve(unresolvedPlan)
+
+ val unresolvedRelations = unresolvedPlan.collect {
+ case unresolvedRelation: UnresolvedRelation => unresolvedRelation
+ }
+ assert(unresolvedRelations.size == 1)
+
+ val partiallyResolvedRelation = metadataResolver
+ .getRelationWithResolvedMetadata(unresolvedRelations.head)
+ .get
+ .asInstanceOf[SubqueryAlias]
+ .child
+ assert(partiallyResolvedRelation.isInstanceOf[UnresolvedCatalogRelation])
+
+ val result = dataSourceResolver.resolveOperator(partiallyResolvedRelation)
+
+ val logicalRelation = result.asInstanceOf[LogicalRelation]
+ assert(
+ logicalRelation.catalogTable.get.identifier.unquotedString
+ == expectedTableName
+ )
+ assert(logicalRelation.relation.schema == expectedTableSchema)
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileResolverSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileResolverSuite.scala
new file mode 100644
index 000000000000..1d1b228028bd
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileResolverSuite.scala
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{LongType, StringType, StructType}
+
+class FileResolverSuite extends QueryTest with SharedSparkSession {
+ private val tableSchema = new StructType().add("id", LongType)
+ private val csvTableSchema = new StructType().add("_c0", StringType)
+
+ test("JSON file format") {
+ val df = spark.range(100).toDF()
+ withTempPath(f => {
+ df.write.json(f.getCanonicalPath)
+ checkResolveOperator(
+ sqlText = s"select id from json.`${f.getCanonicalPath}`",
+ expectedTablePath = s"file:${f.getCanonicalPath}",
+ expectedTableSchema = tableSchema
+ )
+ })
+ }
+
+ test("PARQUET file format") {
+ val df = spark.range(100).toDF()
+ withTempPath(f => {
+ df.write.parquet(f.getCanonicalPath)
+ checkResolveOperator(
+ sqlText = s"select id from parquet.`${f.getCanonicalPath}`",
+ expectedTablePath = s"file:${f.getCanonicalPath}",
+ expectedTableSchema = tableSchema
+ )
+ })
+ }
+
+ test("ORC file format") {
+ val df = spark.range(100).toDF()
+ withTempPath(f => {
+ df.write.orc(f.getCanonicalPath)
+ checkResolveOperator(
+ sqlText = s"select id from ORC.`${f.getCanonicalPath}`",
+ expectedTablePath = s"file:${f.getCanonicalPath}",
+ expectedTableSchema = tableSchema
+ )
+ })
+ }
+
+ test("CSV file format") {
+ val df = spark.range(100).toDF()
+ withTempPath(f => {
+ df.write.csv(f.getCanonicalPath)
+ checkResolveOperator(
+ sqlText = s"select _c0 from csv.`${f.getCanonicalPath}`",
+ expectedTablePath = s"file:${f.getCanonicalPath}",
+ expectedTableSchema = csvTableSchema
+ )
+ })
+ }
+
+ private def checkResolveOperator(
+ sqlText: String,
+ expectedTablePath: String,
+ expectedTableSchema: StructType) = {
+ val fileResolver = new FileResolver(spark)
+
+ val unresolvedPlan = spark.sql(sqlText).queryExecution.logical
+
+ val result = fileResolver.resolveOperator(
+ unresolvedPlan.asInstanceOf[Project].child.asInstanceOf[UnresolvedRelation]
+ )
+
+ val logicalRelation = result.asInstanceOf[LogicalRelation]
+ assert(
+ logicalRelation.relation.asInstanceOf[HadoopFsRelation].location.rootPaths.mkString(",") ==
+ expectedTablePath
+ )
+ assert(logicalRelation.relation.schema == expectedTableSchema)
+ }
+}
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/DataSourceWithHiveResolver.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/DataSourceWithHiveResolver.scala
new file mode 100644
index 000000000000..842faba66cc3
--- /dev/null
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/DataSourceWithHiveResolver.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hive
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.datasources.{DataSourceResolver, LogicalRelation}
+
+/**
+ * [[DataSourceWithHiveResolver]] is a [[DataSourceResolver]] that additionally handles
+ * [[HiveTableRelation]] conversion using [[RelationConversions]].
+ */
+class DataSourceWithHiveResolver(sparkSession: SparkSession, hiveCatalog: HiveSessionCatalog)
+ extends DataSourceResolver(sparkSession) {
+ private val relationConversions = RelationConversions(hiveCatalog)
+
+ /**
+ * Invoke [[DataSourceResolver]] to resolve the input operator. If [[DataSourceResolver]] produces
+ * [[HiveTableRelation]], convert it to [[LogicalRelation]] if possible.
+ */
+ override def resolveOperator: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case operator: LogicalPlan if super.resolveOperator.isDefinedAt(operator) =>
+ val relationAfterDataSourceResolver = super.resolveOperator(operator)
+
+ relationAfterDataSourceResolver match {
+ case hiveTableRelation: HiveTableRelation =>
+ resolveHiveTableRelation(hiveTableRelation)
+ case other => other
+ }
+ }
+
+ private def resolveHiveTableRelation(hiveTableRelation: HiveTableRelation): LogicalPlan = {
+ if (relationConversions.doConvertHiveTableRelationForRead(hiveTableRelation)) {
+ val logicalRelation: LogicalRelation =
+ relationConversions.convertHiveTableRelationForRead(hiveTableRelation)
+ logicalRelation.newInstance()
+ } else {
+ hiveTableRelation.newInstance()
+ }
+ }
+}
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 dbeb8607facc..a1cf27510838 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
@@ -26,6 +26,7 @@ import org.apache.hadoop.hive.ql.udf.generic.{AbstractGenericUDAFResolver, Gener
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.analysis.{Analyzer, EvalSubqueriesForTimeTravel, InvokeProcedures, ReplaceCharWithVarchar, ResolveSessionCatalog, ResolveTranspose}
+import org.apache.spark.sql.catalyst.analysis.resolver.ResolverExtension
import org.apache.spark.sql.catalyst.catalog.{ExternalCatalogWithListener, InvalidUDFClassException}
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -84,6 +85,14 @@ class HiveSessionStateBuilder(
* A logical query plan `Analyzer` with rules specific to Hive.
*/
override protected def analyzer: Analyzer = new Analyzer(catalogManager) {
+ override val singlePassResolverExtensions: Seq[ResolverExtension] = Seq(
+ new DataSourceWithHiveResolver(session, catalog)
+ )
+
+ override val singlePassMetadataResolverExtensions: Seq[ResolverExtension] = Seq(
+ new FileResolver(session)
+ )
+
override val extendedResolutionRules: Seq[Rule[LogicalPlan]] =
new ResolveHiveSerdeTable(session) +:
new FindDataSourceTable(session) +:
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/DataSourceWithHiveResolverSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/DataSourceWithHiveResolverSuite.scala
new file mode 100644
index 000000000000..cb26354521b0
--- /dev/null
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/DataSourceWithHiveResolverSuite.scala
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hive
+
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.analysis.resolver.{MetadataResolver, Resolver}
+import org.apache.spark.sql.catalyst.catalog.{HiveTableRelation, UnresolvedCatalogRelation}
+import org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.hive.HiveUtils
+import org.apache.spark.sql.hive.test.TestHiveSingleton
+import org.apache.spark.sql.test.SQLTestUtils
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+
+class DataSourceWithHiveResolverSuite extends TestHiveSingleton with SQLTestUtils {
+ private val keyValueTableSchema = StructType(
+ Seq(
+ StructField("key", IntegerType, true),
+ StructField("value", StringType, true)
+ )
+ )
+
+ test("ORC table resolution") {
+ withTable("src_orc") {
+ spark.sql("CREATE TABLE src_orc (key INT, value STRING) STORED AS ORC")
+
+ checkResolveOperator(
+ sqlText = "SELECT * FROM src_orc",
+ expectedTableName = "spark_catalog.default.src_orc",
+ expectedTableSchema = keyValueTableSchema,
+ convertedToLogicalRelation = true
+ )
+ }
+ }
+
+ test("ORC table resolution without conversion") {
+ withSQLConf(HiveUtils.CONVERT_METASTORE_ORC.key -> "false") {
+ withTable("src_orc_no_conversion") {
+ spark.sql("CREATE TABLE src_orc_no_conversion (key INT, value STRING) STORED AS ORC")
+
+ checkResolveOperator(
+ sqlText = "SELECT * FROM src_orc_no_conversion",
+ expectedTableName = "spark_catalog.default.src_orc_no_conversion",
+ expectedTableSchema = keyValueTableSchema,
+ convertedToLogicalRelation = false
+ )
+ }
+ }
+ }
+
+ private def checkResolveOperator(
+ sqlText: String,
+ expectedTableName: String,
+ expectedTableSchema: StructType,
+ convertedToLogicalRelation: Boolean) = {
+ val metadataResolver = new MetadataResolver(
+ spark.sessionState.catalogManager,
+ Resolver.createRelationResolution(spark.sessionState.catalogManager)
+ )
+ val dataSourceWithHiveResolver = new DataSourceWithHiveResolver(
+ spark,
+ spark.sessionState.catalog.asInstanceOf[HiveSessionCatalog]
+ )
+
+ val unresolvedPlan = spark.sql(sqlText).queryExecution.logical
+
+ metadataResolver.resolve(unresolvedPlan)
+
+ val unresolvedRelations = unresolvedPlan.collect {
+ case unresolvedRelation: UnresolvedRelation => unresolvedRelation
+ }
+ assert(unresolvedRelations.size == 1)
+
+ val partiallyResolvedRelation = metadataResolver
+ .getRelationWithResolvedMetadata(unresolvedRelations.head)
+ .get
+ .asInstanceOf[SubqueryAlias]
+ .child
+ assert(partiallyResolvedRelation.isInstanceOf[UnresolvedCatalogRelation])
+
+ dataSourceWithHiveResolver.resolveOperator(partiallyResolvedRelation) match {
+ case logicalRelation: LogicalRelation =>
+ assert(convertedToLogicalRelation)
+ assert(logicalRelation.catalogTable.get.identifier.unquotedString == expectedTableName)
+ assert(logicalRelation.relation.schema == expectedTableSchema)
+ case hiveTableRelation: HiveTableRelation =>
+ assert(!convertedToLogicalRelation)
+ assert(hiveTableRelation.tableMeta.identifier.unquotedString == expectedTableName)
+ assert(hiveTableRelation.tableMeta.schema == expectedTableSchema)
+ }
+ }
+}
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/PartitionedTablePerfStatsSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/PartitionedTablePerfStatsSuite.scala
index f7e453a1dbde..b67370f6eb9f 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/PartitionedTablePerfStatsSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/PartitionedTablePerfStatsSuite.scala
@@ -34,6 +34,9 @@ class PartitionedTablePerfStatsSuite
override def beforeEach(): Unit = {
super.beforeEach()
+ // Hive operation counters are doubled in dual-analyzer mode.
+ hiveContext.sparkSession.conf.set(
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key, "false")
FileStatusCache.resetForTesting()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org