srielau commented on code in PR #57194:
URL: https://github.com/apache/spark/pull/57194#discussion_r3580182011
##########
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:
Fixed in 2994aa0 — switched to `DataTypeUtils.sameType(leftElem,
rightElem)`. Added an `ARRAY<STRUCT>` test with different field names on each
side to cover this path.
##########
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:
Addressed in 2994aa0 and d426ab1. Added `checkError` tests in
`AsOfJoinSortMergeSQLSuite` for:
- `ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE` (same-side and cross-side
operands)
- `ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION` (`rand()`)
- `ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE` (TIMESTAMP vs STRING)
Cross-side operands like `(t.trade_time + q.quote_time >= q.quote_time)`
previously surfaced an internal error; d426ab1 validates both-join-input
references early and now raises `TABLE_REFERENCE`.
##########
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:
Added in 2994aa0: `LEFT ASOF JOIN null-pads NOT NULL right catalog columns`
uses a parquet table with `NOT NULL` columns and an unmatched left row to
exercise the `nullRightRow` / `UnsafeProjection` path.
##########
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:
Fixed in 2994aa0. `AstBuilder` now throws
`ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR` for `=`, `AND`, `OR`, and other
non-singular comparisons. Covered in `AsOfJoinSQLSuite`, `PlanParserSuite`.
##########
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:
Added in 2994aa0:
- `strict > MATCH_CONDITION excludes equal right rows`
- `strict < forward MATCH_CONDITION`
- `forward ARRAY<INT> <= MATCH_CONDITION`
--
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]
