This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 ea9d6a7c9180 [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL
ASOF JOIN
ea9d6a7c9180 is described below
commit ea9d6a7c9180e4fb342643b5f5c5aa2df03c385b
Author: Serge Rielau <[email protected]>
AuthorDate: Thu Jul 16 23:20:44 2026 +0800
[SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN
### What changes were proposed in this pull request?
Second stacked PR for
[SPARK-58092](https://issues.apache.org/jira/browse/SPARK-58092) /
[SPARK-58123](https://issues.apache.org/jira/browse/SPARK-58123).
**Follows #57251** (parser, merged). Rebased onto current `master`; this PR
is a **2-commit** analysis-only delta.
**Analysis rule**
- Add `ResolveAsOfJoin` to materialize `MATCH_CONDITION` into
`asOfCondition`, `orderExpression`, and per-side sort expressions
- Expand `USING (...)` into equi-join predicates via
`NaturalAndUsingJoinResolution`
- Validate MATCH_CONDITION operands: cross-side references, determinism,
and type compatibility (including struct/array)
**Logical plan**
- Extend `AsOfJoin` with `leftSortExprs`, `rightSortExprs`, and
`requiresSortMergeAsOfJoin`
- Add `resolveMatchComparison` / `matchSortExpressions` helpers shared with
the DataFrame API path
- Resolve `matchLeftOperand` / `matchRightOperand` via the standard
analyzer expression walk before materialization
**Gating**
- `CheckAnalysis` rejects SQL ASOF plans when
`spark.sql.join.sortMergeAsOfJoin.enabled=false`
(`AS_OF_JOIN.SORT_MERGE_REQUIRED`)
**Out of scope (follow-up PR)**
- Sort-merge physical operator wiring for multi-column sort keys and SQL
execution tests — **#57277**
### Why are the changes needed?
Parser (#57251) produces unresolved `AsOfJoin` nodes with parsed
`MATCH_CONDITION` operands. Analysis must resolve operands, validate semantics,
and produce the same logical shape the existing DataFrame `joinAsOf` path uses
before physical planning can run.
### Does this PR introduce _any_ user-facing change?
Yes (behind `spark.sql.join.asofJoin.enabled`, default off). With the
parser conf enabled, SQL ASOF JOIN now progresses through analysis: valid
queries become analyzable logical plans; invalid MATCH_CONDITION expressions
fail with structured analysis errors instead of reaching execution unresolved.
### How was this patch tested?
Added analysis negative tests to `AsOfJoinSQLSuite`:
- `AS_OF_JOIN.SORT_MERGE_REQUIRED`
- `ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE` (cross-side and invalid refs)
- `ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION`
- `ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE`
Locally:
```
build/sbt "sql/testOnly org.apache.spark.sql.AsOfJoinSQLSuite" \
"catalyst/testOnly
org.apache.spark.sql.catalyst.optimizer.RewriteAsOfJoinSuite"
```
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Cursor.
Closes #57264 from srielau/SPARK-58123.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 5 +
.../spark/sql/catalyst/analysis/Analyzer.scala | 3 +-
.../sql/catalyst/analysis/CheckAnalysis.scala | 11 +-
.../catalyst/analysis/DeduplicateRelations.scala | 2 +-
.../sql/catalyst/analysis/ResolveAsOfJoin.scala | 230 +++++++++++++++++
.../sql/catalyst/optimizer/RewriteAsOfJoin.scala | 17 +-
.../plans/logical/basicLogicalOperators.scala | 276 ++++++++++++++++++++-
.../sql/catalyst/rules/RuleIdCollection.scala | 1 +
.../spark/sql/errors/QueryCompilationErrors.scala | 10 +
.../spark/sql/execution/SparkStrategies.scala | 2 +-
.../org/apache/spark/sql/AsOfJoinSQLSuite.scala | 183 +++++++++++++-
11 files changed, 730 insertions(+), 10 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 83ace844cfaa..60c990066e24 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -210,6 +210,11 @@
"message" : [
"Unsupported as-of join direction '<direction>'. Supported as-of
join direction include: <supported>."
]
+ },
+ "UNSUPPORTED_MATCH_CONDITION_OPERAND" : {
+ "message" : [
+ "The MATCH_CONDITION operands (<type1> and <type2>) use types that
require multi-column sort-merge ASOF join execution. Only scalar numeric and
datetime operands are currently supported."
+ ]
}
},
"sqlState" : "42604"
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 49fdbd45b91c..7d22ab263059 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
@@ -636,7 +636,8 @@ class Analyzer(
typeCoercionRules() ++
Seq(
ResolveWithCTE,
- ExtractDistributedSequenceID) ++
+ ExtractDistributedSequenceID,
+ ResolveAsOfJoin) ++
Seq(ResolveUpdateEventTimeWatermarkColumn) ++
extendedResolutionRules ++
Seq(NameStreamingSources) : _*),
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index f457ac1ba852..b7394d325f12 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -676,7 +676,14 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
"joinCondition" -> toSQLExpr(condition),
"conditionType" -> toSQLType(condition.dataType)))
- case j @ AsOfJoin(_, _, _, Some(condition), _, _, _, _, _, _, _)
+ case j @ AsOfJoin(_, _, _, _, _, _, _, _, _, _, _, _, _, true)
+ if !SQLConf.get.sortMergeAsOfJoinEnabled =>
+ j.failAnalysis(
+ errorClass = "AS_OF_JOIN.SORT_MERGE_REQUIRED",
+ messageParameters = Map(
+ "config" -> SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key))
+
+ case j @ AsOfJoin(_, _, _, Some(condition), _, _, _, _, _, _, _, _,
_, _)
if condition.dataType != BooleanType =>
throw SparkException.internalError(
msg = s"join condition '${toSQLExpr(condition)}' " +
@@ -684,7 +691,7 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
context = j.origin.getQueryContext,
summary = j.origin.context.summary)
- case j @ AsOfJoin(_, _, _, _, _, _, Some(toleranceAssertion), _, _,
_, _) =>
+ case j @ AsOfJoin(_, _, _, _, _, _, Some(toleranceAssertion), _, _,
_, _, _, _, _) =>
if (!toleranceAssertion.foldable) {
j.failAnalysis(
errorClass = "AS_OF_JOIN.TOLERANCE_IS_UNFOLDABLE",
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
index 18045214e565..ce29706501f2 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
@@ -48,7 +48,7 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
if right.resolved && !j.duplicateResolved &&
noMissingInput(right.plan) =>
j.copy(right = right.withNewPlan(dedupRight(left, right.plan)))
// Resolve duplicate output for AsOfJoin.
- case j @ AsOfJoin(left, right, _, _, _, _, _, _, _, _, _)
+ case j @ AsOfJoin(left, right, _, _, _, _, _, _, _, _, _, _, _, _)
if !j.duplicateResolved && noMissingInput(right) =>
j.copy(right = dedupRight(left, right))
// Resolve duplicate output for NearestByJoin.
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
new file mode 100644
index 000000000000..41c79b2905d6
--- /dev/null
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala
@@ -0,0 +1,230 @@
+/*
+ * 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
+
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.expressions.{
+ Expression,
+ RowOrdering,
+ SubqueryExpression,
+ WindowExpression
+}
+import org.apache.spark.sql.catalyst.expressions.AttributeSet
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.logical.{AsOfJoin, LogicalPlan,
Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AS_OF_JOIN, GENERATOR}
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.errors.QueryErrorsBase
+import org.apache.spark.sql.types.{ArrayType, DataType, DatetimeType,
StringType, StructType}
+
+/**
+ * Resolves SQL [[AsOfJoin]] operators: materializes `MATCH_CONDITION` into
`asOfCondition` and
+ * `orderExpression`, and expands `USING` column lists into equi-join
predicates.
+ */
+object ResolveAsOfJoin extends Rule[LogicalPlan] with SQLConfHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(AS_OF_JOIN), ruleId) {
+ case j @ AsOfJoin(
+ left,
+ right,
+ _,
+ condition,
+ _,
+ _,
+ _,
+ usingColumns,
+ matchLeft,
+ matchOp,
+ matchRight,
+ _,
+ _,
+ _)
+ if left.resolved && right.resolved && condition.forall(_.resolved) =>
+ val (joinBase, usingProjection) = usingColumns match {
+ case Some(cols) if condition.isEmpty =>
+ val (projectList, hiddenList, newCondition) =
+ NaturalAndUsingJoinResolution.computeJoinOutputsAndNewCondition(
+ left,
+ left.output,
+ right,
+ right.output,
+ j.joinType,
+ cols,
+ None,
+ (l, r) => conf.resolver(l, r))
+ (j.copy(condition = newCondition, usingColumns = None),
Some((projectList, hiddenList)))
+ case _ => (j, None)
+ }
+ 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)
+ AsOfJoinValidation.validateMatchConditionPlannerSupport(
+ joinBase, leftOperand, rightOperand)
+ 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
+ }
+ case (None, None, None) => joinBase
+ case _ => joinBase
+ }
+ usingProjection match {
+ case Some((projectList, hiddenList)) =>
+ val project = Project(projectList, resolvedJoin)
+ project.setTagValue(
+ Project.hiddenOutputTag,
+ hiddenList.map(_.markAsQualifiedAccessOnly()))
+ project
+ case None => resolvedJoin
+ }
+ }
+}
+
+private[analysis] object AsOfJoinValidation extends QueryErrorsBase {
+
+ def validateMatchConditionTableReferences(
+ join: AsOfJoin,
+ left: LogicalPlan,
+ right: LogicalPlan,
+ leftExpr: Expression,
+ rightExpr: Expression): Unit = {
+ val leftSet = left.outputSet
+ val rightSet = right.outputSet
+
+ def referencesBothJoinSides(refs: AttributeSet): Boolean = {
+ refs.nonEmpty &&
+ refs.intersect(leftSet).nonEmpty &&
+ refs.intersect(rightSet).nonEmpty
+ }
+
+ val leftRefs = leftExpr.references
+ val rightRefs = rightExpr.references
+ if (referencesBothJoinSides(leftRefs) ||
referencesBothJoinSides(rightRefs)) {
+ join.failAnalysis(
+ errorClass = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE",
+ messageParameters = Map(
+ "refs1" -> toSQLExpr(leftExpr),
+ "refs2" -> toSQLExpr(rightExpr)))
+ }
+ }
+
+ def validateMatchConditionOperands(
+ join: AsOfJoin,
+ leftExpr: Expression,
+ rightExpr: Expression): Unit = {
+ Seq(leftExpr, rightExpr).foreach { expr =>
+ findInvalidMatchConditionExpression(expr).foreach { invalidExpr =>
+ join.failAnalysis(
+ errorClass = "ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION",
+ messageParameters = Map("expr" -> toSQLExpr(invalidExpr)))
+ }
+ }
+ if (!RowOrdering.isOrderable(leftExpr.dataType) ||
+ !RowOrdering.isOrderable(rightExpr.dataType) ||
+ !areMatchConditionTypesCompatible(leftExpr.dataType,
rightExpr.dataType)) {
+ join.failAnalysis(
+ errorClass = "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE",
+ messageParameters = Map(
+ "type1" -> toSQLType(leftExpr.dataType),
+ "type2" -> toSQLType(rightExpr.dataType)))
+ }
+ }
+
+ def validateMatchConditionPlannerSupport(
+ join: AsOfJoin,
+ leftOperand: Expression,
+ rightOperand: Expression): Unit = {
+ if (!areScalarSubtractBasedOperands(leftOperand, rightOperand)) {
+ join.failAnalysis(
+ errorClass = "AS_OF_JOIN.UNSUPPORTED_MATCH_CONDITION_OPERAND",
+ messageParameters = Map(
+ "type1" -> toSQLType(leftOperand.dataType),
+ "type2" -> toSQLType(rightOperand.dataType)))
+ }
+ }
+
+ /**
+ * Until multi-column sort-merge ASOF execution lands (SPARK-58124), the
planner can only
+ * consume MATCH_CONDITION plans whose `orderExpression` is a scalar
`Subtract`. STRING and
+ * composite operands use other distance expressions that `findFromOrder`
cannot parse.
+ */
+ private def areScalarSubtractBasedOperands(
+ leftExpr: Expression,
+ rightExpr: Expression): Boolean = {
+ AsOfJoin.supportsSubtract(leftExpr.dataType) &&
AsOfJoin.supportsSubtract(rightExpr.dataType)
+ }
+
+ /**
+ * Tuple/struct operands may use different field names on each side; compare
field-wise by
+ * position when [[TypeCoercion.findWiderTypeForTwo]] does not apply.
+ */
+ private def areMatchConditionTypesCompatible(t1: DataType, t2: DataType):
Boolean = {
+ if (isIncompatibleMatchConditionPair(t1, t2)) {
+ false
+ } else {
+ TypeCoercion.findWiderTypeForTwo(t1, t2).isDefined ||
+ areStructurallyComparableTypes(t1, t2)
+ }
+ }
+
+ private def isIncompatibleMatchConditionPair(t1: DataType, t2: DataType):
Boolean = {
+ def isString(dt: DataType): Boolean = dt.isInstanceOf[StringType]
+ def isTemporal(dt: DataType): Boolean = dt.isInstanceOf[DatetimeType]
+ (isTemporal(t1) && isString(t2)) || (isString(t1) && isTemporal(t2))
+ }
+
+ private def areStructurallyComparableTypes(t1: DataType, t2: DataType):
Boolean = {
+ (t1, t2) match {
+ case (s1: StructType, s2: StructType) if s1.length == s2.length =>
+ s1.zip(s2).forall { case (f1, f2) =>
+ RowOrdering.isOrderable(f1.dataType) &&
+ RowOrdering.isOrderable(f2.dataType) &&
+ areMatchConditionTypesCompatible(f1.dataType, f2.dataType)
+ }
+ case (ArrayType(e1, _), ArrayType(e2, _)) =>
+ RowOrdering.isOrderable(e1) && RowOrdering.isOrderable(e2) &&
+ areMatchConditionTypesCompatible(e1, e2)
+ case _ => false
+ }
+ }
+
+ private def findInvalidMatchConditionExpression(expr: Expression):
Option[Expression] = {
+ expr.collect {
+ case e: SubqueryExpression => e
+ case e: AggregateExpression => e
+ case e: WindowExpression => e
+ case e if e.containsPattern(GENERATOR) => e
+ case e if !e.deterministic => e
+ }.headOption
+ }
+}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoin.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoin.scala
index 31625b25b378..8359bed4eb57 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoin.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoin.scala
@@ -59,7 +59,20 @@ object RewriteAsOfJoin extends Rule[LogicalPlan] {
plan.transformUpWithNewOutput {
case j @ AsOfJoin(
- left, right, asOfCondition, condition, joinType, orderExpression, _,
_, _, _, _) =>
+ left,
+ right,
+ asOfCondition,
+ condition,
+ joinType,
+ orderExpression,
+ _,
+ _,
+ _,
+ _,
+ _,
+ _,
+ _,
+ _) =>
val conditionWithOuterReference =
condition.map(And(_,
asOfCondition)).getOrElse(asOfCondition).transformUp {
case a: AttributeReference if left.outputSet.contains(a) =>
@@ -70,7 +83,7 @@ object RewriteAsOfJoin extends Rule[LogicalPlan] {
val orderExpressionWithOuterReference = orderExpression.transformUp {
case a: AttributeReference if left.outputSet.contains(a) =>
OuterReference(a)
- }
+ }
val rightStruct = CreateStruct(right.output)
val nearestRight = MinBy(rightStruct,
orderExpressionWithOuterReference)
.toAggregateExpression()
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 244614a2c962..c2e42df00b68 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
@@ -2537,7 +2537,10 @@ case class AsOfJoin(
usingColumns: Option[Seq[String]] = None,
matchLeftOperand: Option[Expression] = None,
matchOperator: Option[MatchComparisonOperator] = None,
- matchRightOperand: Option[Expression] = None)
+ matchRightOperand: Option[Expression] = None,
+ leftSortExprs: Seq[Expression] = Nil,
+ rightSortExprs: Seq[Expression] = Nil,
+ requiresSortMergeAsOfJoin: Boolean = false)
extends BinaryNode {
require(Seq(Inner, LeftOuter).contains(joinType),
@@ -2621,7 +2624,276 @@ object AsOfJoin {
usingColumns = usingColumns,
matchLeftOperand = Some(leftExpr),
matchOperator = Some(operator),
- matchRightOperand = Some(rightExpr))
+ matchRightOperand = Some(rightExpr),
+ requiresSortMergeAsOfJoin = true)
+ }
+
+ private[catalyst] def materializeMatchComparison(
+ leftOperand: Expression,
+ rightOperand: Expression,
+ normalizedOp: MatchComparisonOperator)
+ : (Expression, Expression, Seq[Expression], Seq[Expression]) = {
+ val (asOfCondition, orderExpression) =
+ buildMatchExpressions(leftOperand, rightOperand, normalizedOp)
+ val (leftSortExprs, rightSortExprs) = matchSortExpressions(leftOperand,
rightOperand)
+ (asOfCondition, orderExpression, leftSortExprs, rightSortExprs)
+ }
+
+ /**
+ * Sort-merge ASOF join sorts each side by these expressions (after
equi-keys) so the
+ * right-side buffer is ordered consistently with MATCH_CONDITION
lexicographic comparison.
+ *
+ * SQL tuple literals `(t.a, t.b)` are flattened to scalar leaves. Whole
struct columns
+ * (`t.k >= r.k`) sort by the struct value directly so nested struct shapes
stay intact.
+ */
+ def matchSortExpressions(
+ leftOperand: Expression,
+ rightOperand: Expression): (Seq[Expression], Seq[Expression]) = {
+ (leftOperand.dataType, rightOperand.dataType) match {
+ case (leftStruct: StructType, rightStruct: StructType)
+ if leftStruct.sameType(rightStruct) && leftStruct.nonEmpty =>
+ if (isSqlTupleStructOperand(leftOperand) ||
isSqlTupleStructOperand(rightOperand)) {
+ val pairs = collectStructLeafPairs(leftOperand, rightOperand,
leftStruct)
+ (pairs.map(_._1), pairs.map(_._2))
+ } else {
+ (Seq(leftOperand), Seq(rightOperand))
+ }
+ case _ =>
+ (Seq(leftOperand), Seq(rightOperand))
+ }
+ }
+
+ /** True for SQL `(col1, col2, ...)` tuple operands, which become
[[CreateNamedStruct]]. */
+ private def isSqlTupleStructOperand(operand: Expression): Boolean =
+ operand.isInstanceOf[CreateNamedStruct]
+
+ private[catalyst] def normalizeMatchOperands(
+ left: LogicalPlan,
+ right: LogicalPlan,
+ expr1: Expression,
+ operator: MatchComparisonOperator,
+ expr2: Expression): (Expression, Expression, MatchComparisonOperator) = {
+ val leftSet = left.outputSet
+ val rightSet = right.outputSet
+ val expr1Side = operandJoinSide(expr1, leftSet, rightSet)
+ val expr2Side = operandJoinSide(expr2, leftSet, rightSet)
+ (expr1Side, expr2Side) match {
+ case (Some(true), Some(false)) => (expr1, expr2, operator)
+ case (Some(false), Some(true)) => (expr2, expr1, operator.flip)
+ case _ =>
+ throw
QueryCompilationErrors.asOfJoinMatchConditionTableReferenceError(expr1, expr2)
+ }
+ }
+
+ private def operandJoinSide(
+ expr: Expression,
+ leftSet: AttributeSet,
+ rightSet: AttributeSet): Option[Boolean] = {
+ val refs = expr.references
+ if (refs.isEmpty) {
+ None
+ } else if (refs.subsetOf(leftSet)) {
+ Some(true)
+ } else if (refs.subsetOf(rightSet)) {
+ Some(false)
+ } else {
+ None
+ }
+ }
+
+ private def buildMatchExpressions(
+ leftOperand: Expression,
+ rightOperand: Expression,
+ operator: MatchComparisonOperator): (Expression, Expression) = {
+ val (leftForCompare, rightForCompare) =
+ alignOperandsForComparison(leftOperand, rightOperand)
+ val orderExpression = buildOrderExpression(leftOperand, rightOperand,
operator)
+ operator match {
+ case GreaterThanOrEqualOp =>
+ (GreaterThanOrEqual(leftForCompare, rightForCompare), orderExpression)
+ case GreaterThanOp =>
+ (GreaterThan(leftForCompare, rightForCompare), orderExpression)
+ case LessThanOrEqualOp =>
+ (LessThanOrEqual(leftForCompare, rightForCompare), orderExpression)
+ case LessThanOp =>
+ (LessThan(leftForCompare, rightForCompare), orderExpression)
+ }
+ }
+
+ private def buildOrderExpression(
+ leftOperand: Expression,
+ rightOperand: Expression,
+ operator: MatchComparisonOperator): Expression = {
+ (leftOperand.dataType, rightOperand.dataType) match {
+ case (ArrayType(leftElem, _), ArrayType(rightElem, _))
+ if DataTypeUtils.sameType(leftElem, rightElem) =>
+ buildArrayOrderExpression(leftOperand, rightOperand, leftElem,
operator)
+ case (leftStruct: StructType, rightStruct: StructType)
+ if leftStruct.length == rightStruct.length && leftStruct.nonEmpty =>
+ buildFlattenedStructOrderExpression(
+ leftOperand, rightOperand, leftStruct, operator)
+ case _ =>
+ buildLeafOrderExpression(leftOperand, rightOperand, operator)
+ }
+ }
+
+ /**
+ * Tuple/struct operands may use different field names on each side. Rewrite
them to positional
+ * structs with matching schemas so comparison and ordering type-check.
+ */
+ private def alignOperandsForComparison(
+ leftOperand: Expression,
+ rightOperand: Expression): (Expression, Expression) = {
+ decomposeStructOperands(leftOperand, rightOperand) match {
+ case Some(pairs) =>
+ val aligned = pairs.map { case (left, right) =>
+ alignOperandsForComparison(left, right)
+ }
+ (CreateStruct(aligned.map(_._1)), CreateStruct(aligned.map(_._2)))
+ case None =>
+ (leftOperand, rightOperand)
+ }
+ }
+
+ private def buildLeafOrderExpression(
+ leftOperand: Expression,
+ rightOperand: Expression,
+ operator: MatchComparisonOperator): Expression = {
+ if (supportsSubtract(leftOperand.dataType)) {
+ operator match {
+ case GreaterThanOrEqualOp | GreaterThanOp =>
+ Subtract(leftOperand, rightOperand)
+ case LessThanOrEqualOp | LessThanOp =>
+ Subtract(rightOperand, leftOperand)
+ }
+ } else {
+ buildSignedComparisonDistance(leftOperand, rightOperand, operator)
+ }
+ }
+
+ private[catalyst] def supportsSubtract(dataType: DataType): Boolean = {
+ dataType match {
+ case _: NumericType | _: DayTimeIntervalType | _: YearMonthIntervalType |
+ _: CalendarIntervalType | _: TimestampType | _: TimestampNTZType |
_: DateType =>
+ true
+ case _ =>
+ false
+ }
+ }
+
+ private def buildSignedComparisonDistance(
+ leftOperand: Expression,
+ rightOperand: Expression,
+ operator: MatchComparisonOperator): Expression = {
+ val (greaterValue, lesserValue) = operator match {
+ case GreaterThanOrEqualOp | GreaterThanOp => (Literal(1), Literal(-1))
+ case LessThanOrEqualOp | LessThanOp => (Literal(-1), Literal(1))
+ }
+ If(
+ EqualTo(leftOperand, rightOperand),
+ Literal(0),
+ If(GreaterThan(leftOperand, rightOperand), greaterValue, lesserValue))
+ }
+
+ private def buildArrayOrderExpression(
+ leftOperand: Expression,
+ rightOperand: Expression,
+ elementType: DataType,
+ operator: MatchComparisonOperator): Expression = {
+ elementType match {
+ case struct: StructType =>
+ val leftElement = NamedLambdaVariable("left_elem", struct, nullable =
true)
+ val rightElement = NamedLambdaVariable("right_elem", struct, nullable
= true)
+ val leafDiffs = collectStructLeafPairs(leftElement, rightElement,
struct).map {
+ case (left, right) => buildLeafOrderExpression(left, right, operator)
+ }
+ val elementOrder = wrapCompositeOrderExpression(
+ leafDiffs,
+ ArrayType(struct, containsNull = true))
+ ZipWith(
+ leftOperand,
+ rightOperand,
+ LambdaFunction(elementOrder, Seq(leftElement, rightElement)))
+ case _ =>
+ val leftElement = NamedLambdaVariable("left_elem", elementType,
nullable = true)
+ val rightElement = NamedLambdaVariable("right_elem", elementType,
nullable = true)
+ val elementOrder = buildLeafOrderExpression(leftElement, rightElement,
operator)
+ ZipWith(
+ leftOperand,
+ rightOperand,
+ LambdaFunction(elementOrder, Seq(leftElement, rightElement)))
+ }
+ }
+
+ private def buildFlattenedStructOrderExpression(
+ leftOperand: Expression,
+ rightOperand: Expression,
+ structType: StructType,
+ operator: MatchComparisonOperator): Expression = {
+ val leafDiffs = collectStructLeafPairs(leftOperand, rightOperand,
structType).map {
+ case (left, right) => buildLeafOrderExpression(left, right, operator)
+ }
+ wrapCompositeOrderExpression(leafDiffs, structType)
+ }
+
+ private def collectStructLeafPairs(
+ leftOperand: Expression,
+ rightOperand: Expression,
+ structType: StructType): Seq[(Expression, Expression)] = {
+ structFieldExprs(leftOperand, structType)
+ .zip(structFieldExprs(rightOperand, structType))
+ .flatMap {
+ case (left, right) =>
+ (left.dataType, right.dataType) match {
+ case (leftStruct: StructType, rightStruct: StructType)
+ if leftStruct.length == rightStruct.length &&
leftStruct.nonEmpty =>
+ collectStructLeafPairs(left, right, leftStruct)
+ case _ =>
+ Seq((left, right))
+ }
+ }
+ }
+
+ private def wrapCompositeOrderExpression(
+ diffs: Seq[Expression],
+ compositeType: DataType): Expression = {
+ diffs match {
+ case Seq(single) => single
+ case _ =>
+ compositeType match {
+ case _: ArrayType => CreateArray(diffs)
+ case _ => CreateStruct(diffs)
+ }
+ }
+ }
+
+ /** Positional struct fields when both operands are the same struct shape. */
+ private def decomposeStructOperands(
+ leftOperand: Expression,
+ rightOperand: Expression): Option[Seq[(Expression, Expression)]] = {
+ (leftOperand.dataType, rightOperand.dataType) match {
+ case (leftStruct: StructType, rightStruct: StructType)
+ if leftStruct.length == rightStruct.length && leftStruct.nonEmpty =>
+ val leftFields = structFieldExprs(leftOperand, leftStruct)
+ val rightFields = structFieldExprs(rightOperand, rightStruct)
+ if (leftFields.length == rightFields.length) {
+ Some(leftFields.zip(rightFields))
+ } else {
+ None
+ }
+ case _ => None
+ }
+ }
+
+ private def structFieldExprs(
+ operand: Expression,
+ structType: StructType): Seq[Expression] = {
+ operand match {
+ case ns: CreateNamedStruct => ns.valExprs
+ case _ =>
+ structType.indices.map(index =>
+ GetStructField(operand, index, Some(structType(index).name)))
+ }
}
private def makeAsOfCond(
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
index 12fcf3ed9aa3..2953c09e183d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
@@ -101,6 +101,7 @@ object RuleIdCollection {
"org.apache.spark.sql.catalyst.analysis.ResolveCursors" ::
"org.apache.spark.sql.catalyst.analysis.ResolveFetchCursor" ::
"org.apache.spark.sql.catalyst.analysis.ResolveBinBy" ::
+ "org.apache.spark.sql.catalyst.analysis.ResolveAsOfJoin" ::
"org.apache.spark.sql.catalyst.analysis.ResolveSetVariable" ::
"org.apache.spark.sql.catalyst.analysis.ResolveTableConstraints" ::
"org.apache.spark.sql.catalyst.analysis.ResolveTableSpec" ::
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 34242fe25f41..d6f9de4cab54 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
@@ -4971,6 +4971,16 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map.empty)
}
+ def asOfJoinMatchConditionTableReferenceError(
+ expr1: Expression,
+ expr2: Expression): Throwable = {
+ new AnalysisException(
+ errorClass = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE",
+ messageParameters = Map(
+ "refs1" -> toSQLExpr(expr1),
+ "refs2" -> toSQLExpr(expr2)))
+ }
+
def nestedSequentialStreamingUnionError(): Throwable = {
new AnalysisException(
errorClass = "NESTED_SEQUENTIAL_STREAMING_UNION",
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index 1329e0d5d6b3..022c16c1acc5 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -431,7 +431,7 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
object AsOfJoinSelection extends Strategy with PredicateHelper {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case j @ AsOfJoin(left, right, asOfCondition, condition, joinType,
- orderExpression, _, _, _, _, _) if conf.sortMergeAsOfJoinEnabled =>
+ orderExpression, _, _, _, _, _, _, _, _) if
conf.sortMergeAsOfJoinEnabled =>
val (leftKeys, rightKeys, residual) = condition match {
case Some(cond) => extractEquiJoinKeys(cond, left, right)
case None => (Seq.empty[Expression], Seq.empty[Expression], None)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala
index d8fead52af94..215a8a1bbb36 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala
@@ -18,11 +18,12 @@
package org.apache.spark.sql
import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
/**
- * SQL ASOF JOIN surface tests (parser and feature gating).
+ * SQL ASOF JOIN surface tests (parser, analysis, and feature gating).
* Execution semantics and complex MATCH_CONDITION types are covered by
* `AsOfJoinSortMergeSQLSuite`, which requires sort-merge ASOF join.
*/
@@ -38,6 +39,19 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
super.afterAll()
}
+ private def setupTradeQuoteViews(): Unit = {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW trades(trade_time, symbol) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:05', 'AAPL')
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW quotes(quote_time, symbol) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'AAPL')
+ |""".stripMargin)
+ }
+
test("equality operator is rejected in MATCH_CONDITION") {
sql(
"""
@@ -69,4 +83,171 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
start = 24,
stop = 114)))
}
+
+ test("SQL ASOF JOIN requires sort-merge conf") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT t.trade_time, q.quote_time
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ withSQLConf(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "false") {
+ checkError(
+ exception = intercept[AnalysisException](sql(sqlText)),
+ condition = "AS_OF_JOIN.SORT_MERGE_REQUIRED",
+ parameters = Map("config" ->
SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = """ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol""".stripMargin,
+ start = 49,
+ stop = 140)))
+ }
+ }
+
+ test("valid TIMESTAMP MATCH_CONDITION passes analysis with sort-merge
enabled") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT t.trade_time, q.quote_time
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ withSQLConf(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "true") {
+ val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst {
+ case j: AsOfJoin => j
+ }.get
+ assert(asOfJoin.asOfCondition.resolved)
+ assert(asOfJoin.leftSortExprs.nonEmpty)
+ assert(asOfJoin.rightSortExprs.nonEmpty)
+ assert(asOfJoin.matchLeftOperand.isEmpty)
+ }
+ }
+
+ test("MATCH_CONDITION rejects STRING operands until composite sort-merge
lands") {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW left_s(k) AS VALUES ('c')
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW right_s(k) AS VALUES ('a'), ('b')
+ |""".stripMargin)
+ val sqlText =
+ """
+ |SELECT l.k
+ |FROM left_s l ASOF JOIN right_s r
+ | MATCH_CONDITION (l.k >= r.k)
+ |""".stripMargin
+ withSQLConf(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "true") {
+ checkError(
+ exception = intercept[AnalysisException](sql(sqlText)),
+ condition = "AS_OF_JOIN.UNSUPPORTED_MATCH_CONDITION_OPERAND",
+ sqlState = Some("42604"),
+ parameters = Map(
+ "type1" -> "\"STRING\"",
+ "type2" -> "\"STRING\""),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = """ASOF JOIN right_s r
+ | MATCH_CONDITION (l.k >= r.k)""".stripMargin,
+ start = 26,
+ stop = 75)))
+ }
+ }
+
+ test("MATCH_CONDITION rejects cross-side operand references") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT count(*)
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time + q.quote_time >= q.quote_time)
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ checkError(
+ exception = intercept[AnalysisException](sql(sqlText)),
+ condition = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE",
+ sqlState = Some("42K0E"),
+ parameters = Map(
+ "refs1" -> "\"(trade_time + quote_time)\"",
+ "refs2" -> "\"quote_time\""),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = """ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time + q.quote_time >=
q.quote_time)
+ | ON t.symbol = q.symbol""".stripMargin,
+ start = 31,
+ stop = 137)))
+ }
+
+ test("MATCH_CONDITION rejects invalid table references") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT count(*)
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.symbol >= t.symbol)
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ checkError(
+ exception = intercept[AnalysisException](sql(sqlText)),
+ condition = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE",
+ sqlState = Some("42K0E"),
+ parameters = Map(
+ "refs1" -> "\"symbol\"",
+ "refs2" -> "\"symbol\""))
+ }
+
+ test("MATCH_CONDITION rejects non-deterministic expressions") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT t.trade_time
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (rand() >= q.quote_time)
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ checkError(
+ exception = intercept[AnalysisException](sql(sqlText)),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION",
+ sqlState = Some("42903"),
+ parameters = Map("expr" -> "\"rand()\""),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = """ASOF JOIN quotes q
+ | MATCH_CONDITION (rand() >= q.quote_time)
+ | ON t.symbol = q.symbol""".stripMargin,
+ start = 35,
+ stop = 120)))
+ }
+
+ test("MATCH_CONDITION rejects incompatible operand types") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT t.trade_time
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.symbol)
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ checkError(
+ exception = intercept[AnalysisException](sql(sqlText)),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE",
+ sqlState = Some("42K09"),
+ parameters = Map(
+ "type1" -> "\"TIMESTAMP\"",
+ "type2" -> "\"STRING\""),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = """ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.symbol)
+ | ON t.symbol = q.symbol""".stripMargin,
+ start = 35,
+ stop = 122)))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
