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 6fbcf0652a35 [SPARK-58124][SQL] Wire sort-merge ASOF JOIN physical
execution
6fbcf0652a35 is described below
commit 6fbcf0652a35d6aee2e1f2e113c68e774b2e6cdd
Author: Serge Rielau <[email protected]>
AuthorDate: Fri Jul 17 11:31:48 2026 +0800
[SPARK-58124][SQL] Wire sort-merge ASOF JOIN physical execution
### What changes were proposed in this pull request?
Wires sort-merge physical execution for SQL ASOF JOIN on top of the parser
(#57251) and analysis (#57264) work.
**`SortMergeAsOfJoinExec`**
- Sort by multi-column `leftSortExprs` / `rightSortExprs` from analysis
(replacing single `leftAsOfExpr` / `rightAsOfExpr`)
- Detect backward scan direction from `asOfCondition` for composite
`MATCH_CONDITION` operands
**Planner / optimizer**
- `AsOfJoinSelection` and `RewriteAsOfJoin` use sort-merge when
`requiresSortMergeAsOfJoin` is set (SQL path) or
`spark.sql.join.sortMergeAsOfJoin.enabled` is on (DataFrame path)
- Typed `case j: AsOfJoin` in `AsOfJoinSelection` (avoids brittle
positional match)
**Analysis / errors**
- Remove the temporary `AS_OF_JOIN.UNSUPPORTED_MATCH_CONDITION_OPERAND`
gate and its error condition
- Remove `AS_OF_JOIN.SORT_MERGE_REQUIRED` — SQL ASOF JOIN no longer
requires the DataFrame sort-merge conf
- Stop populating DataFrame sort expressions in `AsOfJoin.apply()` while
operands may still be unresolved
**Tests**
- Add `AsOfJoinSortMergeSQLSuite` for SQL execution only (scalar, struct,
array, forward/strict, `USING`, left outer, etc.)
- Keep analysis-negative `MATCH_CONDITION` cases in `AsOfJoinSQLSuite` only
(removed duplicates from the execution suite)
- Add parser tests for invalid `<>` / `<=>` operator text; rename
`sqlAsofJoinDisabled` → `sqlAsOfJoinDisabled`
### Why are the changes needed?
Analysis materializes per-side sort expressions for struct/tuple/array
`MATCH_CONDITION` operands. The physical operator and planner must use those
expressions for correct sort-merge ASOF semantics.
### Does this PR introduce _any_ user-facing change?
Yes. With `spark.sql.join.asofJoin.enabled` (default off), valid SQL ASOF
JOIN queries execute via `SortMergeAsOfJoinExec` — no second conf is required
for SQL. `spark.sql.join.sortMergeAsOfJoin.enabled` remains the opt-in for
DataFrame `joinAsOf`.
### How was this patch tested?
```bash
build/sbt -Dscalastyle.skip=true \
"sql/testOnly org.apache.spark.sql.AsOfJoinSQLSuite \
org.apache.spark.sql.AsOfJoinSortMergeSQLSuite \
org.apache.spark.sql.SortMergeAsOfJoinSuite \
org.apache.spark.sql.DataFrameAsOfJoinSuite"
build/sbt -Dscalastyle.skip=true "core/testOnly
org.apache.spark.SparkThrowableSuite"
```
32/32 in `AsOfJoinSQLSuite` + `AsOfJoinSortMergeSQLSuite`;
`SparkThrowableSuite` clean.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Cursor.
Closes #57277 from srielau/SPARK-58124.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit bd511b588e4eb9c5d070d9b651a6d433fd83156d)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 10 -
.../spark/sql/errors/QueryParsingErrors.scala | 2 +-
.../sql/catalyst/analysis/CheckAnalysis.scala | 7 -
.../sql/catalyst/analysis/ResolveAsOfJoin.scala | 26 -
.../sql/catalyst/optimizer/RewriteAsOfJoin.scala | 11 +-
.../spark/sql/catalyst/parser/AstBuilder.scala | 4 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 12 +-
.../sql/catalyst/parser/PlanParserSuite.scala | 32 ++
.../spark/sql/execution/SparkStrategies.scala | 27 +-
.../execution/joins/SortMergeAsOfJoinExec.scala | 70 ++-
.../org/apache/spark/sql/AsOfJoinSQLSuite.scala | 71 +--
.../spark/sql/AsOfJoinSortMergeSQLSuite.scala | 596 +++++++++++++++++++++
12 files changed, 723 insertions(+), 145 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index a58fd0ad924a..b2e0fec02eb1 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -191,11 +191,6 @@
"Invalid as-of join."
],
"subClass" : {
- "SORT_MERGE_REQUIRED" : {
- "message" : [
- "SQL ASOF JOIN requires the sort-merge physical operator. Set
<config> to true."
- ]
- },
"TOLERANCE_IS_NON_NEGATIVE" : {
"message" : [
"The input argument `tolerance` must be non-negative."
@@ -210,11 +205,6 @@
"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/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
index d7554838ead1..80d2d0982526 100644
---
a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
+++
b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
@@ -237,7 +237,7 @@ private[sql] object QueryParsingErrors extends
DataTypeErrorsBase {
ctx)
}
- def sqlAsofJoinDisabled(configKey: String, ctx: ParserRuleContext):
Throwable = {
+ def sqlAsOfJoinDisabled(configKey: String, ctx: ParserRuleContext):
Throwable = {
new ParseException(
errorClass = "UNSUPPORTED_FEATURE.ASOF_JOIN",
messageParameters = Map("config" -> toSQLConf(configKey)),
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 b7394d325f12..ff482e940d3f 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,13 +676,6 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
"joinCondition" -> toSQLExpr(condition),
"conditionType" -> toSQLType(condition.dataType)))
- 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(
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 41c79b2905d6..fb3b5a425e78 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
@@ -80,8 +80,6 @@ object ResolveAsOfJoin extends Rule[LogicalPlan] with
SQLConfHelper {
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(
@@ -160,30 +158,6 @@ private[analysis] object AsOfJoinValidation extends
QueryErrorsBase {
}
}
- 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.
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 8359bed4eb57..8cb66e1cc9d7 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
@@ -49,14 +49,6 @@ import org.apache.spark.sql.catalyst.rules._
*/
object RewriteAsOfJoin extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = {
- // When the sort-merge AS-OF join operator is enabled, skip this rewrite
- // so that the AsOfJoin logical node reaches the planner intact and
- // AsOfJoinSelection can produce a dedicated physical operator.
- // This conf-based gating (rather than excludedRules) is used because
- // the planner strategy and this optimizer rule must be kept in sync:
- // if the rewrite runs, the planner never sees AsOfJoin.
- if (conf.sortMergeAsOfJoinEnabled) return plan
-
plan.transformUpWithNewOutput {
case j @ AsOfJoin(
left,
@@ -72,7 +64,8 @@ object RewriteAsOfJoin extends Rule[LogicalPlan] {
_,
_,
_,
- _) =>
+ _)
+ if !conf.useSortMergeAsOfJoinOperator(j.requiresSortMergeAsOfJoin) =>
val conditionWithOuterReference =
condition.map(And(_,
asOfCondition)).getOrElse(asOfCondition).transformUp {
case a: AttributeReference if left.outputSet.contains(a) =>
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index 1c09cea951d2..f3f963a5d60e 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -2554,6 +2554,8 @@ class AstBuilder extends DataTypeAstBuilder
private def asOfMatchConditionInvalidOperatorText(expr: Expression): String
= expr match {
case EqualTo(_, _) => "="
+ case Not(EqualTo(_, _)) => "<>"
+ case EqualNullSafe(_, _) => "<=>"
case And(_, _) => "AND"
case Or(_, _) => "OR"
case _ => expr.prettyName
@@ -2567,7 +2569,7 @@ class AstBuilder extends DataTypeAstBuilder
base: LogicalPlan,
criteria: AsofJoinCriteriaContext): AsOfJoin = {
if (!conf.sqlAsOfJoinEnabled) {
- throw
QueryParsingErrors.sqlAsofJoinDisabled(SQLConf.SQL_ASOF_JOIN_ENABLED.key, ctx)
+ throw
QueryParsingErrors.sqlAsOfJoinDisabled(SQLConf.SQL_ASOF_JOIN_ENABLED.key, ctx)
}
val joinType = Option(ctx.asofJoinType) match {
case None => Inner
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 8b52f308403c..3d2c618732bc 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
@@ -947,8 +947,8 @@ object SQLConf {
val SQL_ASOF_JOIN_ENABLED =
buildConf("spark.sql.join.asofJoin.enabled")
- .doc("When true, enable SQL ASOF JOIN syntax with MATCH_CONDITION. When
false, " +
- "ASOF JOIN fails at parse time.")
+ .doc("When true, enable SQL ASOF JOIN syntax with MATCH_CONDITION and
plan it with " +
+ "the sort-merge ASOF join physical operator. When false, ASOF JOIN
fails at parse time.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.booleanConf
@@ -8573,6 +8573,14 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def sqlAsOfJoinEnabled: Boolean = getConf(SQL_ASOF_JOIN_ENABLED)
+ /**
+ * Whether to keep [[AsOfJoin]] for the sort-merge physical operator instead
of the
+ * correlated-subquery rewrite. SQL ASOF JOIN always requires sort-merge;
the DataFrame
+ * API is gated separately on [[sortMergeAsOfJoinEnabled]].
+ */
+ def useSortMergeAsOfJoinOperator(requiresSortMergeAsOfJoin: Boolean):
Boolean =
+ sortMergeAsOfJoinEnabled || requiresSortMergeAsOfJoin
+
def enableRadixSort: Boolean = getConf(RADIX_SORT_ENABLED)
def isParquetSchemaMergingEnabled: Boolean =
getConf(PARQUET_SCHEMA_MERGING_ENABLED)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
index 35510686cbe0..2b28aee84307 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
@@ -1031,6 +1031,38 @@ class PlanParserSuite extends AnalysisTest {
}
}
+ test("asof join - not equal match operator") {
+ withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") {
+ checkError(
+ exception = parseException(
+ "select * from t asof join u match_condition (t.a <> u.a)"),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR",
+ sqlState = Some("42K0E"),
+ parameters = Map("operator" -> "<>"),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = "asof join u match_condition (t.a <> u.a)",
+ start = 16,
+ stop = 55)))
+ }
+ }
+
+ test("asof join - null-safe equal match operator") {
+ withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") {
+ checkError(
+ exception = parseException(
+ "select * from t asof join u match_condition (t.a <=> u.a)"),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR",
+ sqlState = Some("42K0E"),
+ parameters = Map("operator" -> "<=>"),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = "asof join u match_condition (t.a <=> u.a)",
+ start = 16,
+ stop = 56)))
+ }
+ }
+
test("asof join - compound match condition rejected") {
withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") {
checkError(
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 022c16c1acc5..996ecac61b72 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
@@ -425,24 +425,28 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
}
/**
- * Plans AS-OF joins using a dedicated sort-merge operator when the
- * conf is enabled.
+ * Plans AS-OF joins using a dedicated sort-merge operator when enabled for
the
+ * DataFrame API, or implicitly for SQL ASOF JOIN
(`requiresSortMergeAsOfJoin`).
*/
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 =>
- val (leftKeys, rightKeys, residual) = condition match {
- case Some(cond) => extractEquiJoinKeys(cond, left, right)
+ case j: AsOfJoin if
conf.useSortMergeAsOfJoinOperator(j.requiresSortMergeAsOfJoin) =>
+ val (leftKeys, rightKeys, residual) = j.condition match {
+ case Some(cond) => extractEquiJoinKeys(cond, j.left, j.right)
case None => (Seq.empty[Expression], Seq.empty[Expression], None)
}
- val (leftAsOf, rightAsOf) = extractAsOfExprs(
- asOfCondition, orderExpression, left, right)
+ val (leftSort, rightSort) =
+ if (j.leftSortExprs.nonEmpty && j.rightSortExprs.nonEmpty) {
+ (j.leftSortExprs, j.rightSortExprs)
+ } else {
+ val (leftAsOf, rightAsOf) = extractAsOfExprs(j.orderExpression,
j.left, j.right)
+ (Seq(leftAsOf), Seq(rightAsOf))
+ }
joins.SortMergeAsOfJoinExec(
- leftKeys, rightKeys, leftAsOf, rightAsOf,
- asOfCondition, orderExpression, joinType, residual,
- planLater(left), planLater(right)) :: Nil
+ leftKeys, rightKeys, leftSort, rightSort,
+ j.asOfCondition, j.orderExpression, j.joinType, residual,
+ planLater(j.left), planLater(j.right)) :: Nil
case _ => Nil
}
@@ -492,7 +496,6 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
* allowExactMatches variations that complicate asOfCondition's shape.
*/
private def extractAsOfExprs(
- asOfCondition: Expression,
orderExpression: Expression,
left: LogicalPlan,
right: LogicalPlan): (Expression, Expression) = {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala
index 955fc7a185c8..5bdfbca78abf 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala
@@ -45,8 +45,8 @@ import org.apache.spark.sql.execution.metric.{SQLMetric,
SQLMetrics}
case class SortMergeAsOfJoinExec(
leftKeys: Seq[Expression],
rightKeys: Seq[Expression],
- leftAsOfExpr: Expression,
- rightAsOfExpr: Expression,
+ leftSortExprs: Seq[Expression],
+ rightSortExprs: Seq[Expression],
asOfCondition: Expression,
orderExpression: Expression,
joinType: JoinType,
@@ -55,6 +55,10 @@ case class SortMergeAsOfJoinExec(
right: SparkPlan,
isSkewJoin: Boolean = false) extends ShuffledJoin {
+ require(leftSortExprs.nonEmpty && rightSortExprs.nonEmpty &&
+ leftSortExprs.length == rightSortExprs.length,
+ s"$nodeName requires matching non-empty sort expressions on both sides")
+
require(Seq(Inner, LeftOuter).exists(joinType == _),
s"$nodeName does not support join type: $joinType")
@@ -79,27 +83,37 @@ case class SortMergeAsOfJoinExec(
}
override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
- val leftOrdering = leftKeys.map(SortOrder(_, Ascending)) :+
- SortOrder(leftAsOfExpr, Ascending)
- val rightOrdering = rightKeys.map(SortOrder(_, Ascending)) :+
- SortOrder(rightAsOfExpr, Ascending)
+ val leftOrdering = leftKeys.map(SortOrder(_, Ascending)) ++
+ leftSortExprs.map(SortOrder(_, Ascending))
+ val rightOrdering = rightKeys.map(SortOrder(_, Ascending)) ++
+ rightSortExprs.map(SortOrder(_, Ascending))
leftOrdering :: rightOrdering :: Nil
}
override def outputOrdering: Seq[SortOrder] = left.outputOrdering
- // Determine scan direction based on the order expression (distance metric).
- // This is a performance heuristic only -- if it misclassifies, the scan
- // still produces the correct result; only the early-termination shortcut
- // is lost.
- //
- // orderExpression is direction-unique by construction:
- // Backward: Subtract(leftAsOf, rightAsOf) -> backward mode
- // Forward: Subtract(rightAsOf, leftAsOf) -> forward mode
- // Nearest: If(...) -> forward mode
- private val isBackwardJoin: Boolean = orderExpression match {
- case Subtract(l, _, _) if l.semanticEquals(leftAsOfExpr) => true
- case _ => false
+ // Backward joins scan the right-side buffer forward and keep the last match.
+ // Detect from asOfCondition so composite MATCH_CONDITION sort keys still
work.
+ private val isBackwardJoin: Boolean =
+ isBackwardAsOfCondition(asOfCondition, left.output, right.output)
+
+ private def isBackwardAsOfCondition(
+ condition: Expression,
+ leftOutput: Seq[Attribute],
+ rightOutput: Seq[Attribute]): Boolean = {
+ val leftAttrs = AttributeSet(leftOutput)
+ val rightAttrs = AttributeSet(rightOutput)
+ def isBackwardPredicate(expr: Expression): Boolean = expr match {
+ case GreaterThanOrEqual(l, r)
+ if l.references.subsetOf(leftAttrs) &&
r.references.subsetOf(rightAttrs) =>
+ true
+ case GreaterThan(l, r)
+ if l.references.subsetOf(leftAttrs) &&
r.references.subsetOf(rightAttrs) =>
+ true
+ case And(c, _) => isBackwardPredicate(c)
+ case _ => false
+ }
+ isBackwardPredicate(condition)
}
protected override def doExecute(): RDD[InternalRow] = {
@@ -165,8 +179,14 @@ private[joins] class SortMergeAsOfJoinScanner(
private val joinedOutput = leftOutput ++ rightOutput
private val joinedRow = new JoinedRow()
- private val resultProjection =
- UnsafeProjection.create(joinedOutput, joinedOutput)
+ // Use nullable bound references so outer-join null padding is safe even when
+ // right attributes are NOT NULL in the catalog schema.
+ private val resultProjection = {
+ val nullableRefs = joinedOutput.zipWithIndex.map { case (attr, i) =>
+ BoundReference(i, attr.dataType, nullable = true)
+ }
+ UnsafeProjection.create(nullableRefs, joinedOutput)
+ }
private val boundAsOfCond = bindReference(asOfCondition, joinedOutput)
private val boundOrderExpr = bindReference(orderExpression, joinedOutput)
@@ -190,7 +210,15 @@ private[joins] class SortMergeAsOfJoinScanner(
private val distanceOrdering =
TypeUtils.getInterpretedOrdering(orderExpression.dataType)
- private val nullRightRow = new GenericInternalRow(rightOutput.length)
+ // Materialize an all-null right row as UnsafeRow. GenericInternalRow cannot
be
+ // passed through identity UnsafeProjection when right columns are NOT NULL.
+ private val nullRightRow: InternalRow = {
+ val nullableRefs = rightOutput.zipWithIndex.map { case (attr, i) =>
+ BoundReference(i, attr.dataType, nullable = true)
+ }
+ val proj = UnsafeProjection.create(nullableRefs, rightOutput)
+ proj(new GenericInternalRow(rightOutput.length))
+ }
// Spill-backed right-side buffer
private val rightGroupBuffer = new ExternalAppendOnlyUnsafeRowArray(
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 215a8a1bbb36..e75ebbe04a04 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
@@ -17,6 +17,8 @@
package org.apache.spark.sql
+import java.sql.Timestamp
+
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin
import org.apache.spark.sql.internal.SQLConf
@@ -25,7 +27,7 @@ import org.apache.spark.sql.test.SharedSparkSession
/**
* 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.
+ * `AsOfJoinSortMergeSQLSuite`.
*/
class AsOfJoinSQLSuite extends QueryTest with SharedSparkSession {
@@ -84,7 +86,7 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
stop = 114)))
}
- test("SQL ASOF JOIN requires sort-merge conf") {
+ test("valid TIMESTAMP MATCH_CONDITION passes analysis") {
setupTradeQuoteViews()
val sqlText =
"""
@@ -93,22 +95,16 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
| 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)))
- }
+ 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("valid TIMESTAMP MATCH_CONDITION passes analysis with sort-merge
enabled") {
+ test("SQL ASOF JOIN uses sort-merge without DataFrame sort-merge conf") {
setupTradeQuoteViews()
val sqlText =
"""
@@ -117,46 +113,9 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
| 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)))
+ withSQLConf(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "false") {
+ checkAnswer(sql(sqlText), Row(Timestamp.valueOf("2026-06-29 10:00:05"),
+ Timestamp.valueOf("2026-06-29 10:00:00")))
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
new file mode 100644
index 000000000000..cb49f26d9c22
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
@@ -0,0 +1,596 @@
+/*
+ * 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
+
+import java.sql.{Date, Timestamp}
+
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.joins.SortMergeAsOfJoinExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * SQL ASOF JOIN tests that exercise the sort-merge physical operator.
+ * SQL ASOF JOIN uses sort-merge implicitly; only
`spark.sql.join.asofJoin.enabled`
+ * needs to be set (unlike the DataFrame API, which still requires
+ * `spark.sql.join.sortMergeAsOfJoin.enabled`).
+ *
+ * Parser and analysis negative paths live in `AsOfJoinSQLSuite`.
+ */
+class AsOfJoinSortMergeSQLSuite extends QueryTest
+ with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ spark.conf.set(SQLConf.SQL_ASOF_JOIN_ENABLED.key, "true")
+ }
+
+ override def afterAll(): Unit = {
+ spark.conf.unset(SQLConf.SQL_ASOF_JOIN_ENABLED.key)
+ super.afterAll()
+ }
+
+ private def assertUsesSortMergeAsOfJoin(query: DataFrame): Unit = {
+ val plan = query.queryExecution.executedPlan
+ assert(collectWithSubqueries(plan) {
+ case _: SortMergeAsOfJoinExec => true
+ }.nonEmpty, s"Expected SortMergeAsOfJoinExec in plan:\n$plan")
+ }
+
+ private def checkSortMergeAsOf(query: => DataFrame, expected: Seq[Row]):
Unit = {
+ val df = query
+ assertUsesSortMergeAsOfJoin(df)
+ checkAnswer(df, expected)
+ }
+
+ private def setupTradeQuoteViews(): Unit = {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW trades(trade_time, symbol, quantity) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:05', 'AAPL', 100),
+ | (TIMESTAMP '2026-06-29 10:00:11', 'AAPL', 200),
+ | (TIMESTAMP '2026-06-29 10:00:12', 'MSFT', 50),
+ | (TIMESTAMP '2026-06-29 09:59:59', 'GOOG', 30)
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW quotes(quote_time, symbol, bid_price) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'AAPL', 180.10),
+ | (TIMESTAMP '2026-06-29 10:00:07', 'AAPL', 180.15),
+ | (TIMESTAMP '2026-06-29 10:00:10', 'AAPL', 180.20),
+ | (TIMESTAMP '2026-06-29 10:00:08', 'MSFT', 420.50)
+ |""".stripMargin)
+ }
+
+ test("INNER ASOF JOIN with TIMESTAMP MATCH_CONDITION") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.trade_time, t.symbol, t.quantity, q.bid_price
+ |FROM trades t
+ |ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol
+ |""".stripMargin),
+ Seq(
+ Row(Timestamp.valueOf("2026-06-29 10:00:05"), "AAPL", 100, 180.10),
+ Row(Timestamp.valueOf("2026-06-29 10:00:11"), "AAPL", 200, 180.20),
+ Row(Timestamp.valueOf("2026-06-29 10:00:12"), "MSFT", 50, 420.50)))
+ }
+
+ test("LEFT ASOF JOIN preserves unmatched left rows") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.trade_time, t.symbol, q.bid_price
+ |FROM trades t
+ |LEFT ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol
+ |ORDER BY t.trade_time
+ |""".stripMargin),
+ Seq(
+ Row(Timestamp.valueOf("2026-06-29 09:59:59"), "GOOG", null),
+ Row(Timestamp.valueOf("2026-06-29 10:00:05"), "AAPL", 180.10),
+ Row(Timestamp.valueOf("2026-06-29 10:00:11"), "AAPL", 180.20),
+ Row(Timestamp.valueOf("2026-06-29 10:00:12"), "MSFT", 420.50)))
+ }
+
+ test("USING is equivalent to ON symbol equality") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT trade_time, symbol, bid_price
+ |FROM trades
+ |ASOF JOIN quotes
+ | MATCH_CONDITION (trades.trade_time >= quotes.quote_time)
+ | USING (symbol)
+ |ORDER BY trade_time
+ |""".stripMargin),
+ Seq(
+ Row(Timestamp.valueOf("2026-06-29 10:00:05"), "AAPL", 180.10),
+ Row(Timestamp.valueOf("2026-06-29 10:00:11"), "AAPL", 180.20),
+ Row(Timestamp.valueOf("2026-06-29 10:00:12"), "MSFT", 420.50)))
+ }
+
+ test("forward match with <=") {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW alerts(alert_time, host) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'db-01')
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW maintenance(window_start, host) AS
+ |VALUES (TIMESTAMP '2026-06-29 08:00:00', 'db-01'),
+ | (TIMESTAMP '2026-06-29 12:00:00', 'db-01')
+ |""".stripMargin)
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT a.alert_time, a.host, m.window_start
+ |FROM alerts a
+ |ASOF JOIN maintenance m
+ | MATCH_CONDITION (a.alert_time <= m.window_start)
+ | ON a.host = m.host
+ |""".stripMargin),
+ Row(
+ Timestamp.valueOf("2026-06-29 10:00:00"),
+ "db-01",
+ Timestamp.valueOf("2026-06-29 12:00:00")) :: Nil)
+ }
+
+ test("DATE scalar MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.d, r.d AS matched_d
+ |FROM VALUES (DATE '2026-06-29') AS t(d) ASOF JOIN
+ | VALUES (DATE '2026-06-28'), (DATE '2026-06-29') AS r(d)
+ | MATCH_CONDITION (t.d >= r.d)
+ |""".stripMargin),
+ Row(Date.valueOf("2026-06-29"), Date.valueOf("2026-06-29")) :: Nil)
+ }
+
+ test("INT scalar MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.k, r.k AS matched_k
+ |FROM VALUES (10) AS t(k) ASOF JOIN VALUES (5), (7) AS r(k)
+ | MATCH_CONDITION (t.k >= r.k)
+ |""".stripMargin),
+ Row(10, 7) :: Nil)
+ }
+
+ test("STRING scalar MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.k, r.k AS matched_k
+ |FROM VALUES ('c') AS t(k) ASOF JOIN VALUES ('a'), ('b') AS r(k)
+ | MATCH_CONDITION (t.k >= r.k)
+ |""".stripMargin),
+ Row("c", "b") :: Nil)
+ }
+
+ test("STRING scalar forward MATCH_CONDITION with <=") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.k, r.k AS matched_k
+ |FROM VALUES ('b') AS t(k) ASOF JOIN VALUES ('c'), ('d'), ('z') AS
r(k)
+ | MATCH_CONDITION (t.k <= r.k)
+ |""".stripMargin),
+ Row("b", "c") :: Nil)
+ }
+
+ test("DECIMAL scalar MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.k, r.k AS matched_k
+ |FROM VALUES (CAST(10.50 AS DECIMAL(10, 2))) AS t(k) ASOF JOIN
+ | VALUES (CAST(5.00 AS DECIMAL(10, 2))),
+ | (CAST(7.25 AS DECIMAL(10, 2))),
+ | (CAST(9.99 AS DECIMAL(10, 2))) AS r(k)
+ | MATCH_CONDITION (t.k >= r.k)
+ |""".stripMargin),
+ Row(
+ new java.math.BigDecimal("10.50"),
+ new java.math.BigDecimal("9.99")) :: Nil)
+ }
+
+ test("DECIMAL scalar forward MATCH_CONDITION with <=") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.k, r.k AS matched_k
+ |FROM VALUES (CAST(10.50 AS DECIMAL(10, 2))) AS t(k) ASOF JOIN
+ | VALUES (CAST(11.00 AS DECIMAL(10, 2))),
+ | (CAST(12.50 AS DECIMAL(10, 2))),
+ | (CAST(15.00 AS DECIMAL(10, 2))) AS r(k)
+ | MATCH_CONDITION (t.k <= r.k)
+ |""".stripMargin),
+ Row(
+ new java.math.BigDecimal("10.50"),
+ new java.math.BigDecimal("11.00")) :: Nil)
+ }
+
+ test("DOUBLE scalar MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.k, r.k AS matched_k
+ |FROM VALUES (CAST(10.5 AS DOUBLE)) AS t(k) ASOF JOIN
+ | VALUES (CAST(5.0 AS DOUBLE)),
+ | (CAST(7.25 AS DOUBLE)),
+ | (CAST(9.99 AS DOUBLE)) AS r(k)
+ | MATCH_CONDITION (t.k >= r.k)
+ |""".stripMargin),
+ Row(10.5, 9.99) :: Nil)
+ }
+
+ test("DOUBLE scalar forward MATCH_CONDITION with <=") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.k, r.k AS matched_k
+ |FROM VALUES (CAST(10.5 AS DOUBLE)) AS t(k) ASOF JOIN
+ | VALUES (CAST(11.0 AS DOUBLE)),
+ | (CAST(12.5 AS DOUBLE)),
+ | (CAST(15.0 AS DOUBLE)) AS r(k)
+ | MATCH_CONDITION (t.k <= r.k)
+ |""".stripMargin),
+ Row(10.5, 11.0) :: Nil)
+ }
+
+ test("STRUCT tuple MATCH_CONDITION") {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW requests(req_ts, seq, service) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 5, 'api'),
+ | (TIMESTAMP '2026-06-29 10:03:00', 1, 'api')
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW deploys(deploy_ts, seq, service, version)
AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 1, 'api', 'v1.0'),
+ | (TIMESTAMP '2026-06-29 10:00:00', 3, 'api', 'v1.1')
+ |""".stripMargin)
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.req_ts, r.seq, d.version
+ |FROM requests r ASOF JOIN deploys d
+ | MATCH_CONDITION ((r.req_ts, r.seq) >= (d.deploy_ts, d.seq))
+ | ON r.service = d.service
+ |ORDER BY r.req_ts, r.seq
+ |""".stripMargin),
+ Seq(
+ Row(Timestamp.valueOf("2026-06-29 10:00:00"), 5, "v1.1"),
+ Row(Timestamp.valueOf("2026-06-29 10:03:00"), 1, "v1.1")))
+ }
+
+ test("scalar leaf STRUCT tuple MATCH_CONDITION") {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW left_nested(inner_ts, inner_seq, tag) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 2, 'a')
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW right_nested(inner_ts, inner_seq, tag) AS
+ |VALUES (TIMESTAMP '2026-06-29 09:00:00', 1, 'a'),
+ | (TIMESTAMP '2026-06-29 10:00:00', 1, 'a')
+ |""".stripMargin)
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.tag, r.inner_seq
+ |FROM left_nested t
+ |ASOF JOIN right_nested r
+ | MATCH_CONDITION ((t.inner_ts, t.inner_seq, t.tag) >= (r.inner_ts,
r.inner_seq, r.tag))
+ |""".stripMargin),
+ Row("a", 1) :: Nil)
+ }
+
+ test("whole STRUCT column MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.k.tag, r.k.inner_seq
+ |FROM VALUES (named_struct(
+ | 'inner_ts', TIMESTAMP '2026-06-29 10:00:00',
+ | 'inner_seq', 2,
+ | 'tag', 'a')) AS t(k)
+ |ASOF JOIN (
+ | SELECT * FROM VALUES
+ | (named_struct(
+ | 'inner_ts', TIMESTAMP '2026-06-29 09:00:00',
+ | 'inner_seq', 1,
+ | 'tag', 'a')),
+ | (named_struct(
+ | 'inner_ts', TIMESTAMP '2026-06-29 10:00:00',
+ | 'inner_seq', 1,
+ | 'tag', 'a')) AS r(k)
+ |) r
+ | MATCH_CONDITION (t.k >= r.k)
+ |""".stripMargin),
+ Row("a", 1) :: Nil)
+ }
+
+ test("nested whole STRUCT column MATCH_CONDITION") {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW requests(req_ts, seq, service) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 5, 'api'),
+ | (TIMESTAMP '2026-06-29 10:03:00', 1, 'api')
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW deploys(deploy_ts, seq, service, version)
AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 1, 'api', 'v1.0'),
+ | (TIMESTAMP '2026-06-29 10:00:00', 3, 'api', 'v1.1')
+ |""".stripMargin)
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.req_ts, r.seq, d.version
+ |FROM (
+ | SELECT named_struct('ts', req_ts, 'seq', seq) AS k, service,
req_ts, seq
+ | FROM requests
+ |) r
+ |ASOF JOIN (
+ | SELECT named_struct('ts', deploy_ts, 'seq', seq) AS k, service,
version, deploy_ts, seq
+ | FROM deploys
+ |) d
+ | MATCH_CONDITION (r.k >= d.k)
+ | ON r.service = d.service
+ |ORDER BY r.req_ts, r.seq
+ |""".stripMargin),
+ Seq(
+ Row(Timestamp.valueOf("2026-06-29 10:00:00"), 5, "v1.1"),
+ Row(Timestamp.valueOf("2026-06-29 10:03:00"), 1, "v1.1")))
+ }
+
+ test("ARRAY<INT> MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.a
+ |FROM VALUES (ARRAY(1, 3)) AS t(a)
+ |ASOF JOIN VALUES (ARRAY(1, 2)), (ARRAY(1, 4)) AS r(a)
+ |MATCH_CONDITION (t.a >= r.a)
+ |""".stripMargin),
+ Row(Seq(1, 2)) :: Nil)
+ }
+
+ test("ARRAY<STRUCT> whole column MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.a
+ |FROM VALUES (ARRAY(named_struct('seq', 1, 'val', 3))) AS t(a)
+ |ASOF JOIN (
+ | SELECT * FROM VALUES
+ | (ARRAY(named_struct('seq', 1, 'val', 2))),
+ | (ARRAY(named_struct('seq', 1, 'val', 4))) AS r(a)
+ |) r
+ | MATCH_CONDITION (t.a >= r.a)
+ |""".stripMargin),
+ Row(Seq(Row(1, 2))) :: Nil)
+ }
+
+ test("STRUCT tuple from scalar columns MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.c1, r.c2
+ |FROM VALUES (1, 2) AS t(c1, c2)
+ |ASOF JOIN VALUES (1, 1) AS r(c1, c2)
+ |MATCH_CONDITION ((t.c1, t.c2) >= (r.c1, r.c2))
+ |""".stripMargin),
+ Row(1, 1) :: Nil)
+ }
+
+ test("MATCH_CONDITION operand arithmetic with ON") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT count(*) AS cnt
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= date_trunc('hour', q.quote_time))
+ | ON t.symbol = q.symbol
+ |""".stripMargin),
+ Row(3L) :: Nil)
+ }
+
+ test("non-equi ON filters right rows before closest match") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.symbol, q.bid_price
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol AND q.bid_price > 200
+ |ORDER BY t.symbol
+ |""".stripMargin),
+ Row("MSFT", 420.50) :: Nil)
+ }
+
+ test("non-equi range predicate in ON") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.symbol, q.bid_price
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON abs(unix_timestamp(t.trade_time) -
unix_timestamp(q.quote_time)) < 30
+ |ORDER BY t.symbol, t.trade_time
+ |""".stripMargin),
+ Seq(
+ Row("AAPL", 180.10),
+ Row("AAPL", 180.20),
+ Row("MSFT", 180.20)))
+ }
+
+ test("disjunction in ON") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT count(*) AS cnt
+ |FROM VALUES
+ | (CAST(NULL AS STRING), TIMESTAMP '2026-06-29 10:00:00') AS
t(symbol, trade_time)
+ |ASOF JOIN VALUES
+ | (CAST(NULL AS STRING), TIMESTAMP '2026-06-29 09:00:00') AS
q(symbol, quote_time)
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol OR (t.symbol IS NULL AND q.symbol IS NULL)
+ |""".stripMargin),
+ Row(1L) :: Nil)
+ }
+
+ test("ON predicate with expression over both tables") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT t.symbol, q.bid_price
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON substr(t.symbol, 1, 1) = substr(q.symbol, 1, 1)
+ |ORDER BY t.symbol, t.trade_time
+ |""".stripMargin),
+ Seq(
+ Row("AAPL", 180.10),
+ Row("AAPL", 180.20),
+ Row("MSFT", 420.50)))
+ }
+
+ test("ON TRUE matches whole right side") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT count(*) AS cnt
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON TRUE
+ |""".stripMargin),
+ Row(3L) :: Nil)
+ }
+
+ test("ON FALSE empties right search space") {
+ setupTradeQuoteViews()
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT count(*) AS inner_cnt
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON FALSE
+ |""".stripMargin),
+ Row(0L) :: Nil)
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT count(*) AS left_cnt
+ |FROM trades t LEFT ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON FALSE
+ |""".stripMargin),
+ Row(4L) :: Nil)
+ }
+
+ test("LEFT ASOF JOIN null-pads NOT NULL right catalog columns") {
+ withTable("right_nn", "left_nn") {
+ sql("CREATE TABLE right_nn (k INT NOT NULL, v STRING NOT NULL) USING
parquet")
+ sql("INSERT INTO right_nn VALUES (3, 'a')")
+ sql("CREATE TABLE left_nn (k INT, label STRING) USING parquet")
+ sql("INSERT INTO left_nn VALUES (2, 'unmatched')")
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT l.k, l.label, r.v
+ |FROM left_nn l LEFT ASOF JOIN right_nn r
+ | MATCH_CONDITION (l.k >= r.k)
+ |""".stripMargin),
+ Row(2, "unmatched", null) :: Nil)
+ }
+ }
+
+ test("strict > MATCH_CONDITION excludes equal right rows") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT q.quote_time
+ |FROM VALUES (TIMESTAMP '2026-06-29 10:00:00', 'X') AS t(trade_time,
symbol)
+ |ASOF JOIN VALUES
+ | (TIMESTAMP '2026-06-29 10:00:00', 'X'),
+ | (TIMESTAMP '2026-06-29 09:00:00', 'X') AS q(quote_time, symbol)
+ | MATCH_CONDITION (t.trade_time > q.quote_time)
+ | ON t.symbol = q.symbol
+ |""".stripMargin),
+ Row(Timestamp.valueOf("2026-06-29 09:00:00")) :: Nil)
+ }
+
+ test("strict < forward MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.ts
+ |FROM VALUES (TIMESTAMP '2026-06-29 10:00:00') AS t(ts)
+ |ASOF JOIN VALUES
+ | (TIMESTAMP '2026-06-29 10:00:00'),
+ | (TIMESTAMP '2026-06-29 11:00:00') AS r(ts)
+ | MATCH_CONDITION (t.ts < r.ts)
+ |""".stripMargin),
+ Row(Timestamp.valueOf("2026-06-29 11:00:00")) :: Nil)
+ }
+
+ test("forward ARRAY<INT> <= MATCH_CONDITION") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.a
+ |FROM VALUES (ARRAY(1, 5)) AS t(a)
+ |ASOF JOIN VALUES (ARRAY(1, 6)), (ARRAY(1, 8)) AS r(a)
+ | MATCH_CONDITION (t.a <= r.a)
+ |""".stripMargin),
+ Row(Seq(1, 6)) :: Nil)
+ }
+
+ test("ARRAY<STRUCT> operands with different field names") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.a
+ |FROM VALUES (ARRAY(named_struct('x', 1, 'y', 3))) AS t(a)
+ |ASOF JOIN VALUES (ARRAY(named_struct('p', 1, 'q', 2))) AS r(a)
+ | MATCH_CONDITION (t.a >= r.a)
+ |""".stripMargin),
+ Row(Seq(Row(1, 2))) :: Nil)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
