srielau commented on code in PR #57194:
URL: https://github.com/apache/spark/pull/57194#discussion_r3572811485
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2584,8 +2643,305 @@ object AsOfJoin {
direction: AsOfJoinDirection): AsOfJoin = {
val asOfCond = makeAsOfCond(leftAsOf, rightAsOf, tolerance,
allowExactMatches, direction)
val orderingExpr = makeOrderingExpr(leftAsOf, rightAsOf, direction)
+ val (leftSortExprs, rightSortExprs) = matchSortExpressions(leftAsOf,
rightAsOf)
AsOfJoin(left, right, asOfCond, condition, joinType,
- orderingExpr, tolerance.map(t => GreaterThanOrEqual(t,
Literal.default(t.dataType))))
+ orderingExpr, tolerance.map(t => GreaterThanOrEqual(t,
Literal.default(t.dataType))),
+ leftSortExprs = leftSortExprs, rightSortExprs = rightSortExprs)
+ }
+
+ /**
+ * Build an [[AsOfJoin]] from a SQL `MATCH_CONDITION (left_expr op
right_expr)` clause.
+ * Operand normalization is deferred until analysis when join inputs are
resolved.
+ */
+ def fromMatchCondition(
+ left: LogicalPlan,
+ right: LogicalPlan,
+ leftExpr: Expression,
+ operator: MatchComparisonOperator,
+ rightExpr: Expression,
+ condition: Option[Expression],
+ joinType: JoinType,
+ usingColumns: Option[Seq[String]] = None): AsOfJoin = {
+ AsOfJoin(
+ left,
+ right,
+ asOfCondition = Literal.TrueLiteral,
+ condition = condition,
+ joinType = joinType,
+ orderExpression = Literal(0),
+ toleranceAssertion = None,
+ usingColumns = usingColumns,
+ matchComparison = Some(AsOfMatchCondition(leftExpr, operator,
rightExpr)),
+ requiresSortMergeAsOfJoin = true)
+ }
+
+ def resolveMatchComparison(
+ left: LogicalPlan,
+ right: LogicalPlan,
+ leftExpr: Expression,
+ operator: MatchComparisonOperator,
+ rightExpr: Expression): (Expression, Expression, Seq[Expression],
Seq[Expression]) = {
+ val (leftOperand, rightOperand, normalizedOp) =
+ normalizeMatchOperands(left, right, leftExpr, operator, rightExpr)
+ 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 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 leftElem ==
rightElem =>
Review Comment:
**P1: Use structural type equality for array elements**
This gate uses reference equality (`leftElem == rightElem`), while
`matchSortExpressions` (line 2704) uses `leftStruct.sameType(rightStruct)` and
`AsOfJoinValidation.areStructurallyComparableTypes` allows structurally
comparable `ARRAY<STRUCT<...>>` operands with different field names.
For such operands, analysis accepts the query but `buildOrderExpression`
falls through to `buildLeafOrderExpression` on the whole array, which is
incorrect for forward (`<=`/`<`) joins that rely on element-wise distance.
**Suggested fix:**
```scala
case (ArrayType(leftElem, _), ArrayType(rightElem, _))
if DataTypeUtils.sameType(leftElem, rightElem) =>
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -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 = {
Review Comment:
**P1: Regression test for `nullRightRow` fix**
This fix addresses a real bug: `GenericInternalRow` null padding fails
`UnsafeProjection` when right-side catalog columns are `NOT NULL`. The LEFT
OUTER tests use `VALUES` temp views without `NOT NULL` constraints, so this
path isn't exercised in CI.
Consider a test with an explicit `NOT NULL` right-side schema (catalog table
or `CREATE TABLE ... NOT NULL`) where an unmatched left row must produce null
right columns.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -0,0 +1,451 @@
+/*
+ * 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.AnalysisException
+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 require the sort-merge physical operator
+ * (`spark.sql.join.sortMergeAsOfJoin.enabled=true`).
+ */
+class AsOfJoinSortMergeSQLSuite extends QueryTest
+ with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ spark.conf.set(SQLConf.SQL_ASOF_JOIN_ENABLED.key, "true")
+ spark.conf.set(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key, "true")
+ }
+
+ override def afterAll(): Unit = {
+ spark.conf.unset(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key)
+ 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("SQL ASOF JOIN requires sort-merge conf") {
Review Comment:
**P1: Add analysis negative tests**
`AsOfJoinValidation` rejects subqueries, aggregates, window functions,
non-deterministic expressions, incompatible types, and cross-side operands —
but none of these paths are covered by SQL integration tests today (only
parser-level `=` rejection in `AsOfJoinSQLSuite`).
Consider adding `checkError` tests for at least:
- `ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE` (e.g. `MATCH_CONDITION (t.a +
r.b >= r.c)`)
- `ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION` (e.g. subquery or `rand()`
in operand)
- `ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE` (e.g. `INT` vs `STRING`)
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -2546,6 +2552,52 @@ class AstBuilder extends DataTypeAstBuilder
}
}
+ private def asOfMatchConditionFromExpression(
+ expr: Expression): (Expression, MatchComparisonOperator, Expression) = {
+ expr match {
+ case GreaterThanOrEqual(left, right) => (left, GreaterThanOrEqualOp,
right)
+ case GreaterThan(left, right) => (left, GreaterThanOp, right)
+ case LessThanOrEqual(left, right) => (left, LessThanOrEqualOp, right)
+ case LessThan(left, right) => (left, LessThanOp, right)
+ case _ =>
Review Comment:
**P2: Dead error class / misleading parse error for compound
MATCH_CONDITION**
`MATCH_CONDITION (t.a >= u.a AND t.b >= u.b)` hits this branch and surfaces
`PARSE_SYNTAX_ERROR` with `error: ')'`. Meanwhile
`ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR` is defined in
`error-conditions.json` but never thrown.
Consider rejecting non-singular comparisons here with
`ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR` (or a clearer dedicated message).
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -0,0 +1,451 @@
+/*
+ * 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.AnalysisException
+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 require the sort-merge physical operator
+ * (`spark.sql.join.sortMergeAsOfJoin.enabled=true`).
+ */
+class AsOfJoinSortMergeSQLSuite extends QueryTest
+ with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ spark.conf.set(SQLConf.SQL_ASOF_JOIN_ENABLED.key, "true")
+ spark.conf.set(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key, "true")
+ }
+
+ override def afterAll(): Unit = {
+ spark.conf.unset(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key)
+ 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("SQL ASOF JOIN requires sort-merge conf") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT t.trade_time, q.bid_price
+ |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 = 48,
+ stop = 139)))
+ }
+ }
+
+ 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") {
Review Comment:
**P2: Test coverage gaps**
Execution tests cover `>=` and `<=` well but not strict `>` / `<` (different
tie-breaking semantics). Array tests use backward `>=` only; forward array
joins use distance-based early termination in `findBestForwardNearest`, which
is the more fragile path.
Low priority, but one backward `>` and one forward `<=` on `ARRAY<INT>`
would close the gap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
