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 2026/08/03 07:00:18 UTC
(spark) branch branch-4.x updated: [SPARK-58450][SQL] Add single-pass analyzer support for ASOF JOIN
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 49af81720470 [SPARK-58450][SQL] Add single-pass analyzer support for ASOF JOIN
49af81720470 is described below
commit 49af81720470caf690fa2ad7d07bb278003d4b12
Author: Serge Rielau <se...@rielau.com>
AuthorDate: Mon Aug 3 14:59:38 2026 +0800
[SPARK-58450][SQL] Add single-pass analyzer support for ASOF JOIN
### What changes were proposed in this pull request?
Add single-pass (Analyzer++) support for ASOF JOIN.
- Extract shared join-child resolution, join-condition resolution, and hidden-output helpers into a new `JoinLikeResolver` trait used by `JoinResolver`.
- Add `AsOfJoinResolver` that resolves SQL `MATCH_CONDITION` / `USING` ASOF joins and DataFrame-created ASOF joins in one pass.
- Share MATCH_CONDITION materialization between the fixed-point `ResolveAsOfJoin` rule and the single-pass resolver via `AsOfJoinMatchConditionResolution`.
- Wire ASOF JOIN into resolver routing, `ResolverGuard`, dual-run view confs, and `ResolutionValidator`.
- Add `spark.sql.analyzer.singlePassResolver.enableAsOfJoinResolution` (default `true`).
### Why are the changes needed?
ASOF JOIN is already available in Spark SQL / DataFrame APIs, but the single-pass analyzer still falls back to the fixed-point path for `AsOfJoin`. This closes that gap so ASOF JOIN plans can be resolved by Analyzer++ with plan parity under dual-run.
### Does this PR introduce _any_ user-facing change?
No intentional behavior change for queries that already work under the fixed-point analyzer. ASOF JOIN is now eligible for single-pass resolution when
`spark.sql.analyzer.singlePassResolver.enableAsOfJoinResolution` is true (default).
### How was this patch tested?
- `build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.analysis.resolver.ResolutionValidatorSuite'`
- `build/sbt 'sql/testOnly org.apache.spark.sql.analysis.resolver.ResolverGuardSuite -- -z "ASOF JOIN"'`
- Existing ASOF JOIN SQL goldens / suites exercise dual-run plan parity when single-pass is enabled.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.5
Closes #57654 from srielau/SPARK-58450.
Authored-by: Serge Rielau <se...@rielau.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit c14268e050e2b6a6d8b70511e366eee21877a492)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/catalyst/analysis/ResolveAsOfJoin.scala | 104 ++++--
.../analysis/resolver/AsOfJoinResolver.scala | 367 +++++++++++++++++++++
.../analysis/resolver/HybridAnalyzer.scala | 3 +-
.../analysis/resolver/JoinLikeResolver.scala | 275 +++++++++++++++
.../catalyst/analysis/resolver/JoinResolver.scala | 86 +----
.../analysis/resolver/ResolutionValidator.scala | 32 ++
.../sql/catalyst/analysis/resolver/Resolver.scala | 5 +-
.../catalyst/analysis/resolver/ResolverGuard.scala | 51 +++
.../plans/logical/basicLogicalOperators.scala | 29 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 10 +
...ExplicitlyUnsupportedResolverFeatureSuite.scala | 22 ++
.../sql/analysis/resolver/ResolverGuardSuite.scala | 22 ++
12 files changed, 891 insertions(+), 115 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala
index f09612d5902b..f5f828fc6ed6 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala
@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.{
}
import org.apache.spark.sql.catalyst.expressions.AttributeSet
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.MatchComparisonOperator
import org.apache.spark.sql.catalyst.plans.logical.{AsOfJoin, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin.MatchConditionTypes
import org.apache.spark.sql.catalyst.rules.Rule
@@ -73,24 +74,23 @@ object ResolveAsOfJoin extends Rule[LogicalPlan] with SQLConfHelper {
}
val resolvedJoin = (matchLeft, matchOp, matchRight) match {
case (Some(leftExpr), Some(operator), Some(rightExpr)) =>
- AsOfJoinValidation.validateMatchConditionTableReferences(
- joinBase, left, right, leftExpr, rightExpr)
- if (leftExpr.resolved && rightExpr.resolved) {
- AsOfJoinValidation.validateMatchConditionOperands(joinBase, leftExpr, rightExpr)
- val (leftOperand, rightOperand, normalizedOp) =
- AsOfJoin.normalizeMatchOperands(left, right, leftExpr, operator, rightExpr)
- val (asOfCondition, orderExpression, leftSortExprs, rightSortExprs) =
- AsOfJoin.materializeMatchComparison(leftOperand, rightOperand, normalizedOp)
- joinBase.copy(
- asOfCondition = asOfCondition,
- orderExpression = orderExpression,
- leftSortExprs = leftSortExprs,
- rightSortExprs = rightSortExprs,
- matchLeftOperand = None,
- matchOperator = None,
- matchRightOperand = None)
- } else {
- joinBase
+ AsOfJoinMatchConditionResolution.materialize(
+ join = joinBase,
+ leftSet = left.outputSet,
+ rightSet = right.outputSet,
+ leftOperand = leftExpr,
+ operator = operator,
+ rightOperand = rightExpr) match {
+ case Some(materialized) =>
+ joinBase.copy(
+ asOfCondition = materialized.asOfCondition,
+ orderExpression = materialized.orderExpression,
+ leftSortExprs = materialized.leftSortExpressions,
+ rightSortExprs = materialized.rightSortExpressions,
+ matchLeftOperand = None,
+ matchOperator = None,
+ matchRightOperand = None)
+ case None => joinBase
}
case (None, None, None) => joinBase
case _ => joinBase
@@ -107,17 +107,75 @@ object ResolveAsOfJoin extends Rule[LogicalPlan] with SQLConfHelper {
}
}
+/**
+ * The executable [[AsOfJoin]] fields that a SQL `MATCH_CONDITION` clause materializes into.
+ */
+private[analysis] case class MaterializedMatchCondition(
+ asOfCondition: Expression,
+ orderExpression: Expression,
+ leftSortExpressions: Seq[Expression],
+ rightSortExpressions: Seq[Expression])
+
+/**
+ * Validates a SQL `MATCH_CONDITION` clause and materializes it into the executable [[AsOfJoin]]
+ * fields, shared by the fixed-point [[ResolveAsOfJoin]] rule and the single-pass
+ * [[org.apache.spark.sql.catalyst.analysis.resolver.AsOfJoinResolver]] so the two analyzers
+ * cannot diverge on match-condition validation or normalization.
+ */
+private[analysis] object AsOfJoinMatchConditionResolution {
+
+ /**
+ * Returns [[None]] while either operand is still unresolved, which the fixed-point analyzer
+ * reaches on iterations before the operands resolve. Table-reference validation runs
+ * regardless, as it only needs the operands' attribute references.
+ */
+ def materialize(
+ join: AsOfJoin,
+ leftSet: AttributeSet,
+ rightSet: AttributeSet,
+ leftOperand: Expression,
+ operator: MatchComparisonOperator,
+ rightOperand: Expression): Option[MaterializedMatchCondition] = {
+ AsOfJoinValidation.validateMatchConditionTableReferences(
+ join = join,
+ leftSet = leftSet,
+ rightSet = rightSet,
+ leftExpr = leftOperand,
+ rightExpr = rightOperand)
+
+ if (leftOperand.resolved && rightOperand.resolved) {
+ AsOfJoinValidation.validateMatchConditionOperands(join, leftOperand, rightOperand)
+ val (normalizedLeft, normalizedRight, normalizedOperator) =
+ AsOfJoin.normalizeMatchOperands(
+ leftSet = leftSet,
+ rightSet = rightSet,
+ expr1 = leftOperand,
+ operator = operator,
+ expr2 = rightOperand)
+ val (asOfCondition, orderExpression, leftSortExpressions, rightSortExpressions) =
+ AsOfJoin.materializeMatchComparison(
+ leftOperand = normalizedLeft,
+ rightOperand = normalizedRight,
+ normalizedOp = normalizedOperator)
+ Some(MaterializedMatchCondition(
+ asOfCondition = asOfCondition,
+ orderExpression = orderExpression,
+ leftSortExpressions = leftSortExpressions,
+ rightSortExpressions = rightSortExpressions))
+ } else {
+ None
+ }
+ }
+}
+
private[analysis] object AsOfJoinValidation extends QueryErrorsBase {
def validateMatchConditionTableReferences(
join: AsOfJoin,
- left: LogicalPlan,
- right: LogicalPlan,
+ leftSet: AttributeSet,
+ rightSet: AttributeSet,
leftExpr: Expression,
rightExpr: Expression): Unit = {
- val leftSet = left.outputSet
- val rightSet = right.outputSet
-
def referencesBothJoinSides(refs: AttributeSet): Boolean = {
refs.nonEmpty &&
refs.intersect(leftSet).nonEmpty &&
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AsOfJoinResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AsOfJoinResolver.scala
new file mode 100644
index 000000000000..d7216e4f304f
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AsOfJoinResolver.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.analysis.{
+ AnalysisErrorAt,
+ AsOfJoinMatchConditionResolution,
+ MaterializedMatchCondition,
+ NaturalAndUsingJoinResolution
+}
+import org.apache.spark.sql.catalyst.expressions.{
+ Attribute,
+ AttributeSet,
+ Expression,
+ LambdaFunction,
+ NamedExpression
+}
+import org.apache.spark.sql.catalyst.plans.JoinType
+import org.apache.spark.sql.catalyst.plans.logical.{AsOfJoin, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.util._
+
+/**
+ * Resolves [[AsOfJoin]] operators, including SQL `MATCH_CONDITION` and `USING` clauses.
+ *
+ * An ASOF JOIN matches each left-side row to the closest right-side row whose order key is
+ * less than or equal to (or, for a reversed match condition, greater than or equal to) the
+ * left row's order key, optionally within a tolerance and equality keys (`USING` / `ON`).
+ * Unlike a regular join, at most one right row is chosen per left row based on temporal (or
+ * other ordered) proximity rather than arbitrary key equality.
+ *
+ * {{{
+ * SELECT * FROM trades ASOF JOIN quotes
+ * MATCH_CONDITION (trades.time >= quotes.time) USING (symbol)
+ *
+ * Project [symbol, time, ...]
+ * +- AsOfJoin (trades.time >= quotes.time), (trades.symbol = quotes.symbol), Inner
+ * }}}
+ */
+class AsOfJoinResolver(
+ override val resolver: Resolver,
+ override val expressionResolver: ExpressionResolver)
+ extends TreeNodeResolver[AsOfJoin, LogicalPlan]
+ with JoinLikeResolver {
+
+ override def resolve(unresolvedAsOfJoin: AsOfJoin): LogicalPlan = {
+ val (resolvedLeft, leftNameScope) = resolveJoinChild(
+ unresolvedOperator = unresolvedAsOfJoin,
+ child = unresolvedAsOfJoin.left
+ )
+
+ val (resolvedRight, rightNameScope) = resolveJoinChild(
+ unresolvedOperator = unresolvedAsOfJoin,
+ child = unresolvedAsOfJoin.right
+ )
+
+ ExpressionIdAssigner.assertOutputsHaveNoConflictingExpressionIds(
+ Seq(leftNameScope.output, rightNameScope.output)
+ )
+
+ expressionIdAssigner.createMappingFromChildMappings(
+ newOutputIds = leftNameScope.getOutputIds ++ rightNameScope.getOutputIds
+ )
+
+ val partiallyResolved = unresolvedAsOfJoin.copy(
+ left = resolvedLeft,
+ right = resolvedRight
+ )
+
+ val resolvedCondition = resolveJoinCondition(
+ unresolvedJoin = unresolvedAsOfJoin,
+ unresolvedCondition = partiallyResolved.condition,
+ leftNameScope = leftNameScope,
+ rightNameScope = rightNameScope,
+ collectInvalidExpressions = true
+ )
+
+ val (usingOutput, conditionWithUsingColumns) = resolveUsingColumns(
+ unresolvedAsOfJoin = unresolvedAsOfJoin,
+ partiallyResolved = partiallyResolved,
+ resolvedCondition = resolvedCondition,
+ leftNameScope = leftNameScope,
+ rightNameScope = rightNameScope
+ )
+
+ val resolvedJoin = resolveAsOfExpressions(
+ unresolvedAsOfJoin = unresolvedAsOfJoin,
+ partiallyResolved = partiallyResolved.copy(
+ condition = conditionWithUsingColumns,
+ usingColumns = None
+ ),
+ leftNameScope = leftNameScope,
+ rightNameScope = rightNameScope
+ )
+
+ usingOutput match {
+ case Some((outputList, hiddenList)) =>
+ buildUsingProject(
+ unresolvedAsOfJoin = unresolvedAsOfJoin,
+ resolvedJoin = resolvedJoin,
+ outputList = outputList,
+ hiddenList = hiddenList,
+ rightNameScope = rightNameScope
+ )
+ case None =>
+ overwriteJoinOutputScope(
+ joinType = unresolvedAsOfJoin.joinType,
+ leftNameScope = leftNameScope,
+ rightNameScope = rightNameScope
+ )
+ cteRegistry.currentScope.tryPutWithCTE(
+ unresolvedOperator = unresolvedAsOfJoin,
+ resolvedOperator = resolvedJoin
+ )
+ }
+ }
+
+ private def resolveUsingColumns(
+ unresolvedAsOfJoin: AsOfJoin,
+ partiallyResolved: AsOfJoin,
+ resolvedCondition: Option[Expression],
+ leftNameScope: NameScope,
+ rightNameScope: NameScope)
+ : (Option[(Seq[NamedExpression], Seq[Attribute])], Option[Expression]) = {
+ partiallyResolved.usingColumns match {
+ case Some(columns) if resolvedCondition.isEmpty =>
+ val (outputList, hiddenList, newCondition) =
+ NaturalAndUsingJoinResolution.computeJoinOutputsAndNewCondition(
+ left = partiallyResolved.left,
+ leftOutput = leftNameScope.output,
+ right = partiallyResolved.right,
+ rightOutput = rightNameScope.output,
+ joinType = partiallyResolved.joinType,
+ joinNames = columns,
+ condition = None,
+ resolveName = conf.resolver
+ )
+ val resolvedUsingCondition = resolveJoinCondition(
+ unresolvedJoin = unresolvedAsOfJoin,
+ unresolvedCondition = newCondition,
+ leftNameScope = leftNameScope,
+ rightNameScope = rightNameScope,
+ collectInvalidExpressions = true
+ )
+ (Some((outputList, hiddenList)), resolvedUsingCondition)
+ case _ =>
+ (None, resolvedCondition)
+ }
+ }
+
+ private def buildUsingProject(
+ unresolvedAsOfJoin: AsOfJoin,
+ resolvedJoin: AsOfJoin,
+ outputList: Seq[NamedExpression],
+ hiddenList: Seq[Attribute],
+ rightNameScope: NameScope): Project = {
+ val resolvedOutputList = outputList.map { expression =>
+ resolveExpressionInJoin(unresolvedAsOfJoin, expression).asInstanceOf[NamedExpression]
+ }
+ val outputAttributes = resolvedOutputList.map(_.toAttribute)
+ val filteredHiddenOutput = filterHiddenOutputMetadataForJoin(
+ joinType = unresolvedAsOfJoin.joinType,
+ oldHiddenOutput = scopes.current.hiddenOutput,
+ rightHiddenOutput = rightNameScope.hiddenOutput
+ )
+ val newHiddenOutput = computeHiddenOutputForJoin(
+ mainOutput = outputAttributes,
+ oldHiddenOutput = filteredHiddenOutput,
+ extraHiddenOutput = hiddenList
+ )
+ scopes.overwriteCurrent(
+ output = Some(outputAttributes),
+ hiddenOutput = Some(newHiddenOutput)
+ )
+
+ val qualifiedAccessOnlyColumns = newHiddenOutput.filter(_.qualifiedAccessOnly)
+ val projectList =
+ if (unresolvedAsOfJoin.containsTag(ResolverTag.TOP_LEVEL_OPERATOR)) {
+ resolvedOutputList
+ } else {
+ resolvedOutputList ++ qualifiedAccessOnlyColumns
+ }
+
+ operatorResolutionContextStack.current.baseOperator = Some(resolvedJoin)
+ val project = Project(projectList, resolvedJoin)
+ project.setTagValue(Project.hiddenOutputTag, qualifiedAccessOnlyColumns)
+ project
+ }
+
+ private def resolveAsOfExpressions(
+ unresolvedAsOfJoin: AsOfJoin,
+ partiallyResolved: AsOfJoin,
+ leftNameScope: NameScope,
+ rightNameScope: NameScope): AsOfJoin = {
+ (
+ partiallyResolved.matchLeftOperand,
+ partiallyResolved.matchOperator,
+ partiallyResolved.matchRightOperand
+ ) match {
+ case (Some(unresolvedLeftOperand), Some(operator), Some(unresolvedRightOperand)) =>
+ val leftOperand = resolveExpressionInJoin(unresolvedAsOfJoin, unresolvedLeftOperand)
+ val rightOperand = resolveExpressionInJoin(unresolvedAsOfJoin, unresolvedRightOperand)
+ AsOfJoinMatchConditionResolution.materialize(
+ join = partiallyResolved,
+ leftSet = AttributeSet(leftNameScope.output),
+ rightSet = AttributeSet(rightNameScope.output),
+ leftOperand = leftOperand,
+ operator = operator,
+ rightOperand = rightOperand
+ ) match {
+ case Some(materialized) =>
+ materializedMatchConditionToJoin(
+ unresolvedAsOfJoin = unresolvedAsOfJoin,
+ partiallyResolved = partiallyResolved,
+ materialized = materialized
+ )
+ case None =>
+ partiallyResolved
+ }
+ case (None, None, None) =>
+ resolvePreMaterializedExpressions(unresolvedAsOfJoin, partiallyResolved)
+ case _ =>
+ partiallyResolved
+ }
+ }
+
+ /**
+ * Places the materialized `MATCH_CONDITION` expressions on the join. They are freshly
+ * constructed trees, so each one still has to go through expression resolution before it can be
+ * placed in the resolved plan.
+ */
+ private def materializedMatchConditionToJoin(
+ unresolvedAsOfJoin: AsOfJoin,
+ partiallyResolved: AsOfJoin,
+ materialized: MaterializedMatchCondition): AsOfJoin = {
+ throwIfLambdaBasedOrdering(materialized.orderExpression)
+
+ val asOfCondition = resolveExpressionInJoin(unresolvedAsOfJoin, materialized.asOfCondition)
+ val orderExpression =
+ resolveExpressionInJoin(unresolvedAsOfJoin, materialized.orderExpression)
+ val leftSortExpressions =
+ materialized.leftSortExpressions.map(resolveExpressionInJoin(unresolvedAsOfJoin, _))
+ val rightSortExpressions =
+ materialized.rightSortExpressions.map(resolveExpressionInJoin(unresolvedAsOfJoin, _))
+
+ partiallyResolved.copy(
+ asOfCondition = asOfCondition,
+ orderExpression = orderExpression,
+ matchLeftOperand = None,
+ matchOperator = None,
+ matchRightOperand = None,
+ leftSortExprs = leftSortExpressions,
+ rightSortExprs = rightSortExpressions
+ )
+ }
+
+ private def resolvePreMaterializedExpressions(
+ unresolvedAsOfJoin: AsOfJoin,
+ partiallyResolved: AsOfJoin): AsOfJoin = {
+ throwIfLambdaBasedOrdering(partiallyResolved.orderExpression)
+
+ val resolvedTolerance =
+ partiallyResolved.toleranceAssertion.map(resolveExpressionInJoin(unresolvedAsOfJoin, _))
+ validateTolerance(unresolvedAsOfJoin, resolvedTolerance)
+
+ val asOfCondition =
+ resolveExpressionInJoin(unresolvedAsOfJoin, partiallyResolved.asOfCondition)
+ val orderExpression =
+ resolveExpressionInJoin(unresolvedAsOfJoin, partiallyResolved.orderExpression)
+ val leftSortExpressions =
+ partiallyResolved.leftSortExprs.map(resolveExpressionInJoin(unresolvedAsOfJoin, _))
+ val rightSortExpressions =
+ partiallyResolved.rightSortExprs.map(resolveExpressionInJoin(unresolvedAsOfJoin, _))
+
+ partiallyResolved.copy(
+ asOfCondition = asOfCondition,
+ orderExpression = orderExpression,
+ toleranceAssertion = resolvedTolerance,
+ leftSortExprs = leftSortExpressions,
+ rightSortExprs = rightSortExpressions
+ )
+ }
+
+ /**
+ * `MATCH_CONDITION` operands that are ordered element-wise (`ARRAY` operands) materialize into
+ * an ordering expression built on top of a [[LambdaFunction]]:
+ *
+ * {{{
+ * -- MATCH_CONDITION (t.a >= r.a) with ARRAY<INT> operands
+ * zip_with(t.a, r.a, lambdafunction((lambda left_elem - lambda right_elem), ...))
+ * }}}
+ *
+ * The single-pass [[ExpressionResolver]] doesn't support lambda expressions, so bail out and
+ * let the fixed-point analyzer resolve those queries.
+ */
+ private def throwIfLambdaBasedOrdering(orderExpression: Expression): Unit = {
+ if (orderExpression.exists(_.isInstanceOf[LambdaFunction])) {
+ throw new ExplicitlyUnsupportedResolverFeature(
+ "MATCH_CONDITION with a lambda-based ordering expression"
+ )
+ }
+ }
+
+ private def resolveExpressionInJoin(
+ unresolvedAsOfJoin: AsOfJoin,
+ unresolvedExpression: Expression): Expression = {
+ expressionResolver.resolveExpressionTreeInOperator(
+ unresolvedExpression,
+ unresolvedAsOfJoin
+ )
+ }
+
+ private def validateTolerance(
+ unresolvedAsOfJoin: AsOfJoin,
+ toleranceAssertion: Option[Expression]): Unit = {
+ toleranceAssertion.foreach { assertion =>
+ if (!assertion.foldable) {
+ unresolvedAsOfJoin.failAnalysis(
+ errorClass = "AS_OF_JOIN.TOLERANCE_IS_UNFOLDABLE",
+ messageParameters = Map.empty
+ )
+ }
+ if (!assertion.eval().asInstanceOf[Boolean]) {
+ unresolvedAsOfJoin.failAnalysis(
+ errorClass = "AS_OF_JOIN.TOLERANCE_IS_NON_NEGATIVE",
+ messageParameters = Map.empty
+ )
+ }
+ }
+ }
+
+ private def overwriteJoinOutputScope(
+ joinType: JoinType,
+ leftNameScope: NameScope,
+ rightNameScope: NameScope): Unit = {
+ val newOutput = AsOfJoin.computeOutput(
+ joinType = joinType,
+ leftOutput = leftNameScope.output,
+ rightOutput = rightNameScope.output
+ )
+
+ val filteredHiddenOutput = filterHiddenOutputMetadataForJoin(
+ joinType = joinType,
+ oldHiddenOutput = scopes.current.hiddenOutput,
+ rightHiddenOutput = rightNameScope.hiddenOutput
+ )
+
+ scopes.overwriteCurrent(
+ output = Some(newOutput),
+ hiddenOutput = Some(computeHiddenOutputForJoin(newOutput, filteredHiddenOutput))
+ )
+ }
+}
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
index 3fc643859730..ad3d88e42afc 100644
--- 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
@@ -354,7 +354,8 @@ object HybridAnalyzer {
SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_RELATION_BRIDGING_ENABLED.key,
SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_RUN_EXTENDED_RESOLUTION_CHECKS.key,
SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_THROW_FROM_RESOLVER_GUARD.key,
- SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_VALIDATION_ENABLED.key
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_VALIDATION_ENABLED.key,
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLE_ASOF_JOIN_RESOLUTION.key
)
/**
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/JoinLikeResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/JoinLikeResolver.scala
new file mode 100644
index 000000000000..9cc87bf7f00d
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/JoinLikeResolver.scala
@@ -0,0 +1,275 @@
+/*
+ * 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.HashSet
+
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.{withPosition, AnalysisErrorAt}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, ExprId}
+import org.apache.spark.sql.catalyst.plans.{
+ ExistenceJoin,
+ FullOuter,
+ JoinType,
+ LeftAnti,
+ LeftOuter,
+ LeftSemi,
+ LeftSingle,
+ RightOuter
+}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.errors.QueryErrorsBase
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * Shared resolution mechanics for join-like binary operators, mixed into [[JoinResolver]] and
+ * other join-like resolvers: resolving two children in isolated multi-child scopes, computing a
+ * combined hidden output, filtering hidden metadata columns by join type, and resolving a boolean
+ * join condition. These routines carry subtle scope, CTE, hidden-output, metadata-nullability and
+ * [[ExprId]] behavior; keeping them here ensures join-like resolvers stay aligned and fixes cannot
+ * drift between them.
+ *
+ * For example, in the following query:
+ *
+ * {{{
+ * SELECT * FROM t1 JOIN t2 ON t1.key = t2.key;
+ * }}}
+ *
+ * the plan is:
+ *
+ * {{{
+ * Project [key#1, key#2]
+ * +- Join Inner, (key#1 = key#2)
+ * :- SubqueryAlias t1
+ * : +- Relation t1[key#1]
+ * +- SubqueryAlias t2
+ * +- Relation t2[key#2]
+ * }}}
+ *
+ * `t1` and `t2` are each resolved by [[resolveJoinChild]] in their own [[NameScope]], and the
+ * condition `key#1 = key#2` is resolved by [[resolveJoinCondition]] against the union of the two
+ * child scopes.
+ */
+trait JoinLikeResolver extends SQLConfHelper with QueryErrorsBase {
+
+ protected val resolver: Resolver
+ protected val expressionResolver: ExpressionResolver
+
+ protected def scopes: NameScopeStack = resolver.getNameScopes
+ protected def cteRegistry: CteRegistry = resolver.getCteRegistry
+ protected def operatorResolutionContextStack: OperatorResolutionContextStack =
+ resolver.getOperatorResolutionContextStack
+ protected def expressionIdAssigner: ExpressionIdAssigner =
+ expressionResolver.getExpressionIdAssigner
+
+ /**
+ * Resolves a single join child in the context of a) new [[NameScope]] b) new
+ * [[ExpressionIdAssigner]] mapping c) new [[CteScope]] for the multi-child operator. Returns the
+ * resolved child together with its [[NameScope]], which the caller uses to compute the join
+ * output.
+ */
+ protected def resolveJoinChild(
+ unresolvedOperator: LogicalPlan,
+ child: LogicalPlan): (LogicalPlan, NameScope) = {
+ expressionIdAssigner.pushMapping()
+ scopes.pushScope()
+ cteRegistry.pushScopeForMultiChildOperator(
+ unresolvedOperator = unresolvedOperator,
+ unresolvedChild = child
+ )
+
+ try {
+ val resolvedChild = resolver.resolve(child)
+ (resolvedChild, scopes.current)
+ } finally {
+ cteRegistry.popScope()
+ scopes.popScope()
+ expressionIdAssigner.popMapping(collectChildMapping = true)
+ }
+ }
+
+ /**
+ * Resolves the join condition against __all__ attributes from child scopes. We overwrite the
+ * current scope first to prepare for
+ * [[ExpressionResolver.resolveExpressionTreeInOperator]]. The join will actually produce a
+ * different output than the one set here, so an additional overwrite with the correct values is
+ * needed afterwards. Two overwrites are necessary because the condition is resolved from
+ * original children outputs, whereas the join output will either not contain all attributes or
+ * their nullabilities will be different.
+ *
+ * `collectInvalidExpressions` controls whether unsupported expressions (aggregate / window /
+ * generator, etc.) found in the just-resolved condition are thrown immediately as
+ * `UNSUPPORTED_EXPR_FOR_OPERATOR`. [[JoinResolver]] leaves this off and relies on the generic
+ * post-resolution check in [[Resolver.validateResolvedOperatorGenerically]], which inspects
+ * [[ExpressionResolver.getLastInvalidExpressionsInTheContextOfOperator]] once, after the last
+ * expression tree of the operator is resolved. Callers that resolve further expression trees
+ * after the condition must turn this on, or those later trees overwrite the "last invalid
+ * expressions" snapshot before the generic check runs.
+ */
+ protected def resolveJoinCondition(
+ unresolvedJoin: LogicalPlan,
+ unresolvedCondition: Option[Expression],
+ leftNameScope: NameScope,
+ rightNameScope: NameScope,
+ collectInvalidExpressions: Boolean = false): Option[Expression] = {
+ scopes.overwriteCurrent(
+ output = Some(leftNameScope.output ++ rightNameScope.output),
+ hiddenOutput = Some(leftNameScope.hiddenOutput ++ rightNameScope.hiddenOutput)
+ )
+
+ val resolvedCondition = unresolvedCondition.map { condition =>
+ expressionResolver.resolveExpressionTreeInOperator(
+ condition,
+ unresolvedJoin
+ )
+ }
+
+ validateJoinConditionDataType(resolvedCondition, unresolvedJoin)
+
+ if (collectInvalidExpressions) {
+ val invalidExpressions =
+ expressionResolver.getLastInvalidExpressionsInTheContextOfOperator
+ if (invalidExpressions.nonEmpty) {
+ withPosition(unresolvedJoin) {
+ resolver.throwUnsupportedExprForOperator(
+ operator = unresolvedJoin,
+ invalidExpressions = invalidExpressions
+ )
+ }
+ }
+ }
+
+ resolvedCondition
+ }
+
+ private def validateJoinConditionDataType(
+ condition: Option[Expression],
+ unresolvedJoin: LogicalPlan): Unit = {
+ condition match {
+ case Some(condition) =>
+ if (condition.dataType != BooleanType) {
+ unresolvedJoin.failAnalysis(
+ errorClass = "JOIN_CONDITION_IS_NOT_BOOLEAN_TYPE",
+ messageParameters = Map(
+ "joinCondition" -> toSQLExpr(condition),
+ "conditionType" -> toSQLType(condition.dataType)
+ )
+ )
+ }
+ case None =>
+ }
+ }
+
+ /**
+ * Computes the new hidden output for a join. The result contains attributes from `mainOutput`,
+ * followed by `extraHiddenOutput` (marked as qualified access only), followed by qualified access
+ * only attributes from `oldHiddenOutput`. All attributes must be unique: `mainOutput` takes
+ * precedence over hidden output, and `extraHiddenOutput` takes precedence over `oldHiddenOutput`.
+ *
+ * For regular joins, `extraHiddenOutput` is empty and the result is simply `mainOutput` plus the
+ * qualified access only portion of `oldHiddenOutput`. For NATURAL / USING joins,
+ * `extraHiddenOutput` contains additional attributes from
+ * [[org.apache.spark.sql.catalyst.analysis.NaturalAndUsingJoinResolution]] that must be marked as
+ * qualified access only and take precedence over `oldHiddenOutput` so that name resolution in
+ * downstream operators (e.g. `Sort`, `Filter`) disambiguates correctly.
+ */
+ protected def computeHiddenOutputForJoin(
+ mainOutput: Seq[Attribute],
+ oldHiddenOutput: Seq[Attribute],
+ extraHiddenOutput: Seq[Attribute] = Seq.empty): Seq[Attribute] = {
+ val mainOutputLookup = new HashSet[ExprId](mainOutput.size)
+ mainOutput.foreach { attribute =>
+ mainOutputLookup.add(attribute.exprId)
+ }
+
+ val extraHiddenOutputLookup = new HashSet[ExprId](extraHiddenOutput.size)
+ extraHiddenOutput.foreach { attribute =>
+ extraHiddenOutputLookup.add(attribute.exprId)
+ }
+
+ val filteredExtraHiddenOutput = extraHiddenOutput.collect {
+ case attribute if !mainOutputLookup.contains(attribute.exprId) =>
+ attribute.markAsQualifiedAccessOnly()
+ }
+
+ val filteredOldHiddenOutput = oldHiddenOutput.filter { attribute =>
+ !mainOutputLookup.contains(attribute.exprId) &&
+ !extraHiddenOutputLookup.contains(attribute.exprId) &&
+ (attribute.qualifiedAccessOnly || attribute.isMetadataCol)
+ }
+
+ mainOutput ++ filteredExtraHiddenOutput ++ filteredOldHiddenOutput
+ }
+
+ /**
+ * Filters metadata columns from hidden output based on join type.
+ *
+ * For [[ExistenceJoin]] and left-existence joins ([[LeftSemi]], [[LeftAnti]]), right-side
+ * metadata columns are removed, matching [[org.apache.spark.sql.catalyst.plans.logical.Join]]'s
+ * `metadataOutput`, which propagates only the left side's metadata output for these join types.
+ * For outer joins, metadata columns on the nullable side have their nullability set to `true`,
+ * mirroring the adjustment applied to the main output -- metadata columns bypass it because they
+ * live in hidden output instead. For all other join types, metadata columns from both sides are
+ * kept as-is.
+ */
+ protected def filterHiddenOutputMetadataForJoin(
+ joinType: JoinType,
+ oldHiddenOutput: Seq[Attribute],
+ rightHiddenOutput: Seq[Attribute]): Seq[Attribute] = {
+ val rightMetadataIds = new HashSet[ExprId]()
+ rightHiddenOutput.foreach { attribute =>
+ if (attribute.isMetadataCol) {
+ rightMetadataIds.add(attribute.exprId)
+ }
+ }
+
+ joinType match {
+ case _: ExistenceJoin | LeftSemi | LeftAnti =>
+ oldHiddenOutput.filter(attribute =>
+ !attribute.isMetadataCol || !rightMetadataIds.contains(attribute.exprId)
+ )
+ case LeftOuter | LeftSingle =>
+ oldHiddenOutput.map { attribute =>
+ if (attribute.isMetadataCol && rightMetadataIds.contains(attribute.exprId)) {
+ attribute.withNullability(true)
+ } else {
+ attribute
+ }
+ }
+ case RightOuter =>
+ oldHiddenOutput.map { attribute =>
+ if (attribute.isMetadataCol && !rightMetadataIds.contains(attribute.exprId)) {
+ attribute.withNullability(true)
+ } else {
+ attribute
+ }
+ }
+ case FullOuter =>
+ oldHiddenOutput.map { attribute =>
+ if (attribute.isMetadataCol) {
+ attribute.withNullability(true)
+ } else {
+ attribute
+ }
+ }
+ case _ =>
+ oldHiddenOutput
+ }
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/JoinResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/JoinResolver.scala
index 27a55c7f6b9f..1fbd43074154 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/JoinResolver.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/JoinResolver.scala
@@ -19,24 +19,22 @@ package org.apache.spark.sql.catalyst.analysis.resolver
import java.util.HashSet
-import org.apache.spark.sql.catalyst.analysis.{AnalysisErrorAt, NaturalAndUsingJoinResolution}
+import org.apache.spark.sql.catalyst.analysis.NaturalAndUsingJoinResolution
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, ExprId, NamedExpression}
import org.apache.spark.sql.catalyst.plans.{JoinType, NaturalJoin, UsingJoin}
import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.util._
-import org.apache.spark.sql.types.BooleanType
/**
* Resolves [[Join]] operator by resolving its left and right children and its join condition. If
* the unresolved join is [[NaturalJoin]] or [[UsingJoin]], the resulting operator will be
* [[Project]], otherwise it will be [[Join]].
*/
-class JoinResolver(resolver: Resolver, expressionResolver: ExpressionResolver)
- extends TreeNodeResolver[Join, LogicalPlan] {
- private val scopes = resolver.getNameScopes
- private val expressionIdAssigner = expressionResolver.getExpressionIdAssigner
- private val cteRegistry = resolver.getCteRegistry
- private val operatorResolutionContextStack = resolver.getOperatorResolutionContextStack
+class JoinResolver(
+ override val resolver: Resolver,
+ override val expressionResolver: ExpressionResolver)
+ extends TreeNodeResolver[Join, LogicalPlan]
+ with JoinLikeResolver {
/**
* Resolves [[Join]] operator:
@@ -52,12 +50,12 @@ class JoinResolver(resolver: Resolver, expressionResolver: ExpressionResolver)
*/
override def resolve(unresolvedJoin: Join): LogicalPlan = {
val (resolvedLeftOperator: LogicalPlan, leftNameScope: NameScope) = resolveJoinChild(
- unresolvedJoin = unresolvedJoin,
+ unresolvedOperator = unresolvedJoin,
child = unresolvedJoin.left
)
val (resolvedRightOperator: LogicalPlan, rightNameScope: NameScope) = resolveJoinChild(
- unresolvedJoin = unresolvedJoin,
+ unresolvedOperator = unresolvedJoin,
child = unresolvedJoin.right
)
@@ -82,26 +80,6 @@ class JoinResolver(resolver: Resolver, expressionResolver: ExpressionResolver)
)
}
- private def resolveJoinChild(
- unresolvedJoin: Join,
- child: LogicalPlan): (LogicalPlan, NameScope) = {
- expressionIdAssigner.pushMapping()
- scopes.pushScope()
- cteRegistry.pushScopeForMultiChildOperator(
- unresolvedOperator = unresolvedJoin,
- unresolvedChild = child
- )
-
- try {
- val resolvedChild = resolver.resolve(child)
- (resolvedChild, scopes.current)
- } finally {
- cteRegistry.popScope()
- scopes.popScope()
- expressionIdAssigner.popMapping(collectChildMapping = true)
- }
- }
-
/**
* If the type of join is [[NaturalJoin]] or [[UsingJoin]], perform additional transformations in
* [[commonNaturalJoinProcessing]]. Otherwise, overwrite current name scope output with the
@@ -375,52 +353,4 @@ class JoinResolver(resolver: Resolver, expressionResolver: ExpressionResolver)
rightNameScope.output.map(_.name)
)
}
-
- /**
- * Resolves join condition by __all__ attributes from child scopes. We need to overwrite current
- * scope first to prepare for [[resolveExpressionTreeInOperator]]. [[Join]] will actually produce
- * different output than the one we are setting here, so additional overwrite with correct values
- * will be needed. Two overwrites are necessary because condition is resolved from original
- * children outputs, whereas output of [[Join]] will either not contain all attributes or their
- * nullabilities will be different.
- */
- private def resolveJoinCondition(
- unresolvedJoin: Join,
- unresolvedCondition: Option[Expression],
- leftNameScope: NameScope,
- rightNameScope: NameScope) = {
- scopes.overwriteCurrent(
- output = Some(leftNameScope.output ++ rightNameScope.output),
- hiddenOutput = Some(leftNameScope.hiddenOutput ++ rightNameScope.hiddenOutput)
- )
-
- val resolvedCondition = unresolvedCondition.map { condition =>
- expressionResolver.resolveExpressionTreeInOperator(
- condition,
- unresolvedJoin
- )
- }
-
- validateJoinConditionDataType(resolvedCondition, unresolvedJoin)
-
- resolvedCondition
- }
-
- private def validateJoinConditionDataType(
- condition: Option[Expression],
- unresolvedJoin: Join): Unit = {
- condition match {
- case Some(condition) =>
- if (condition.dataType != BooleanType) {
- unresolvedJoin.failAnalysis(
- errorClass = "JOIN_CONDITION_IS_NOT_BOOLEAN_TYPE",
- messageParameters = Map(
- "joinCondition" -> toSQLExpr(condition),
- "conditionType" -> toSQLType(condition.dataType)
- )
- )
- }
- case None =>
- }
- }
}
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
index 6bdf2d4b0615..452f6d9555ae 100644
--- 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
@@ -101,6 +101,8 @@ class ResolutionValidator {
validateSort(sort)
case join: Join =>
validateJoin(join)
+ case asOfJoin: AsOfJoin =>
+ validateAsOfJoin(asOfJoin)
case repartition: Repartition =>
validateRepartition(repartition)
case repartitionByExpression: RepartitionByExpression =>
@@ -349,6 +351,36 @@ class ResolutionValidator {
handleOperatorOutput(join)
}
+ private def validateAsOfJoin(asOfJoin: AsOfJoin): Unit = {
+ // The inner scope keeps the per-child output overwrites done by `handleOperatorOutput` out
+ // of the outer scope, which holds the combined join output the join expressions resolve
+ // against.
+ attributeScopeStack.pushScope()
+ try {
+ attributeScopeStack.pushScope()
+ try {
+ validate(asOfJoin.left)
+ validate(asOfJoin.right)
+ assert(asOfJoin.left.outputSet.intersect(asOfJoin.right.outputSet).isEmpty)
+ } finally {
+ attributeScopeStack.popScope()
+ }
+
+ attributeScopeStack.overwriteCurrent(asOfJoin.left.output ++ asOfJoin.right.output)
+
+ expressionResolutionValidator.validate(asOfJoin.asOfCondition)
+ expressionResolutionValidator.validate(asOfJoin.orderExpression)
+ asOfJoin.condition.foreach(expressionResolutionValidator.validate)
+ asOfJoin.toleranceAssertion.foreach(expressionResolutionValidator.validate)
+ asOfJoin.leftSortExprs.foreach(expressionResolutionValidator.validate)
+ asOfJoin.rightSortExprs.foreach(expressionResolutionValidator.validate)
+ } finally {
+ attributeScopeStack.popScope()
+ }
+
+ handleOperatorOutput(asOfJoin)
+ }
+
private def validateSupervisingCommand(supervisingCommand: SupervisingCommand): Unit = {}
private def handleOperatorOutput(operator: LogicalPlan): Unit = {
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
index 9aac3cb726ec..5d9a029b7fb6 100644
--- 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
@@ -118,6 +118,7 @@ class Resolver(
private val filterResolver = new FilterResolver(this, expressionResolver)
private val sortResolver = new SortResolver(this, expressionResolver)
private val joinResolver = new JoinResolver(this, expressionResolver)
+ private val asOfJoinResolver = new AsOfJoinResolver(this, expressionResolver)
private val havingResolver = new HavingResolver(this, expressionResolver)
/**
@@ -265,6 +266,8 @@ class Resolver(
unresolvedPlan match {
case unresolvedJoin: Join =>
joinResolver.resolve(unresolvedJoin)
+ case unresolvedAsOfJoin: AsOfJoin =>
+ asOfJoinResolver.resolve(unresolvedAsOfJoin)
case unresolvedWith: UnresolvedWith =>
resolveWith(unresolvedWith)
case withCte: WithCTE =>
@@ -871,7 +874,7 @@ class Resolver(
summary = operator.origin.context.summary()
)
- private def throwUnsupportedExprForOperator(
+ private[resolver] def throwUnsupportedExprForOperator(
operator: LogicalPlan,
invalidExpressions: Seq[Expression]): Nothing = {
throw new AnalysisException(
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala
index 99c79aaee3f6..80270cd4653d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala
@@ -111,6 +111,11 @@ class ResolverGuard(
checkFilter(filter)
case join: Join =>
checkJoin(join)
+ case asOfJoin: AsOfJoin
+ if conf.getConf(
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLE_ASOF_JOIN_RESOLUTION
+ ) =>
+ checkAsOfJoin(asOfJoin)
case unresolvedSubqueryColumnAliases: UnresolvedSubqueryColumnAliases =>
checkUnresolvedSubqueryColumnAliases(unresolvedSubqueryColumnAliases)
case subqueryAlias: SubqueryAlias =>
@@ -284,6 +289,52 @@ class ResolverGuard(
}
}
+ private def checkAsOfJoin(asOfJoin: AsOfJoin) = {
+ checkOperator(asOfJoin.left)
+ .orElse {
+ checkOperator(asOfJoin.right)
+ }
+ .orElse {
+ asOfJoin.condition match {
+ case Some(condition) => checkExpression(condition)
+ case None => None
+ }
+ }
+ .orElse {
+ asOfJoin.matchLeftOperand match {
+ case Some(expression) => checkExpression(expression)
+ case None => None
+ }
+ }
+ .orElse {
+ asOfJoin.matchRightOperand match {
+ case Some(expression) => checkExpression(expression)
+ case None => None
+ }
+ }
+ .orElse {
+ checkExpression(asOfJoin.asOfCondition)
+ }
+ .orElse {
+ checkExpression(asOfJoin.orderExpression)
+ }
+ .orElse {
+ asOfJoin.toleranceAssertion match {
+ case Some(expression) => checkExpression(expression)
+ case None => None
+ }
+ }
+ .orElse {
+ checkExpressions(asOfJoin.leftSortExprs)
+ }
+ .orElse {
+ checkExpressions(asOfJoin.rightSortExprs)
+ }
+ }
+
+ private def checkExpressions(expressions: Seq[Expression]): Option[String] =
+ expressions.iterator.map(checkExpression).collectFirst { case Some(reason) => reason }
+
private def checkFilter(unresolvedFilter: Filter) =
checkOperator(unresolvedFilter.child).orElse(checkExpression(unresolvedFilter.condition))
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index 38d107cd2260..2edb7f6fc60c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -2547,14 +2547,8 @@ case class AsOfJoin(
override protected def stringArgs: Iterator[Any] = super.stringArgs.take(5)
- override def output: Seq[Attribute] = {
- joinType match {
- case LeftOuter =>
- left.output ++ right.output.map(_.withNullability(true))
- case _ =>
- left.output ++ right.output
- }
- }
+ override def output: Seq[Attribute] =
+ AsOfJoin.computeOutput(joinType, left.output, right.output)
def duplicateResolved: Boolean = left.outputSet.intersect(right.outputSet).isEmpty
@@ -2583,6 +2577,19 @@ case class AsOfJoin(
object AsOfJoin {
+ /**
+ * Computes the output attributes of an [[AsOfJoin]] given its join type and child outputs.
+ */
+ def computeOutput(
+ joinType: JoinType,
+ leftOutput: Seq[Attribute],
+ rightOutput: Seq[Attribute]): Seq[Attribute] = joinType match {
+ case LeftOuter =>
+ leftOutput ++ rightOutput.map(_.withNullability(true))
+ case _ =>
+ leftOutput ++ rightOutput
+ }
+
def apply(
left: LogicalPlan,
right: LogicalPlan,
@@ -2751,13 +2758,11 @@ object AsOfJoin {
operand.isInstanceOf[CreateNamedStruct]
private[catalyst] def normalizeMatchOperands(
- left: LogicalPlan,
- right: LogicalPlan,
+ leftSet: AttributeSet,
+ rightSet: AttributeSet,
expr1: Expression,
operator: MatchComparisonOperator,
expr2: Expression): (Expression, Expression, MatchComparisonOperator) = {
- val leftSet = left.outputSet
- val rightSet = right.outputSet
val expr1Side = operandJoinSide(expr1, leftSet, rightSet, syntacticIsLeft = true)
val expr2Side = operandJoinSide(expr2, leftSet, rightSet, syntacticIsLeft = false)
(expr1Side, expr2Side) match {
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 b35ad05091c7..217f6f68a7b5 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
@@ -574,6 +574,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val ANALYZER_SINGLE_PASS_RESOLVER_ENABLE_ASOF_JOIN_RESOLUTION =
+ buildConf("spark.sql.analyzer.singlePassResolver.enableAsOfJoinResolution")
+ .internal()
+ .doc("When true, enables ASOF JOIN resolution in single-pass analyzer. " +
+ "Otherwise, resolution falls back to fixed-point analyzer.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
+
val MULTI_COMMUTATIVE_OP_OPT_THRESHOLD =
buildConf("spark.sql.analyzer.canonicalization.multiCommutativeOpMemoryOptThreshold")
.internal()
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
index 0a9c1c07d1bd..3cf2613cf21d 100644
--- 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
@@ -22,6 +22,7 @@ import org.apache.spark.sql.catalyst.analysis.resolver.{
Resolver
}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
class ExplicitlyUnsupportedResolverFeatureSuite extends SharedSparkSession {
@@ -44,6 +45,27 @@ class ExplicitlyUnsupportedResolverFeatureSuite extends SharedSparkSession {
}
}
+ test("ASOF JOIN MATCH_CONDITION operands requiring element-wise ordering") {
+ withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") {
+ checkResolution(
+ """SELECT t.a, r.a FROM VALUES (ARRAY(1, 3)) AS t(a)
+ |ASOF JOIN VALUES (ARRAY(1, 2)) AS r(a) MATCH_CONDITION (t.a >= r.a);""".stripMargin,
+ expectedMessage = Some("MATCH_CONDITION with a lambda-based ordering expression")
+ )
+ checkResolution(
+ """SELECT t.a, r.a FROM VALUES (ARRAY(named_struct('seq', 1))) AS t(a)
+ |ASOF JOIN VALUES (ARRAY(named_struct('seq', 2))) AS r(a)
+ |MATCH_CONDITION (t.a >= r.a);""".stripMargin,
+ expectedMessage = Some("MATCH_CONDITION with a lambda-based ordering expression")
+ )
+ checkResolution(
+ """SELECT t.k, r.k FROM VALUES (10) AS t(k)
+ |ASOF JOIN VALUES (5) AS r(k) MATCH_CONDITION (t.k >= r.k);""".stripMargin,
+ shouldPass = true
+ )
+ }
+ }
+
private def checkResolution(
sqlText: String,
shouldPass: Boolean = false,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala
index 6c61c7797a6c..26e95313e594 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/ResolverGuardSuite.scala
@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.analysis.resolver.{
}
import org.apache.spark.sql.catalyst.expressions.Literal
import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
class ResolverGuardSuite extends ResolverGuardSuiteBase {
@@ -455,6 +456,27 @@ class ResolverGuardSuite extends ResolverGuardSuiteBase {
}
}
+ gridTest("ASOF JOIN")(Seq(true, false)) { enabled =>
+ val asOfJoinQuery =
+ "SELECT t.symbol, q.bid_price FROM " +
+ "VALUES (TIMESTAMP '2026-06-29 10:00:05', 'AAPL') AS t(trade_time, symbol) " +
+ "ASOF JOIN VALUES (TIMESTAMP '2026-06-29 10:00:00', 'AAPL', 180.10) " +
+ "AS q(quote_time, symbol, bid_price) " +
+ "MATCH_CONDITION (t.trade_time >= q.quote_time) ON t.symbol = q.symbol"
+ val expectedReason = if (enabled) {
+ None
+ } else {
+ Some("class org.apache.spark.sql.catalyst.plans.logical.AsOfJoin operator resolution")
+ }
+ withSQLConf(
+ SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLE_ASOF_JOIN_RESOLUTION.key ->
+ enabled.toString
+ ) {
+ checkResolverGuard(asOfJoinQuery, expectedReason)
+ }
+ }
+
}
trait ResolverGuardSuiteBase extends SharedSparkSession {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org