This is an automated email from the ASF dual-hosted git repository.
uros-b 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 028e3f154dc0 [SPARK-59858][SQL] Compare ASOF JOIN STRUCT operands
without rebuilding them
028e3f154dc0 is described below
commit 028e3f154dc0b5dd69636770cdc7bd132039c9e4
Author: Luka Zdravić <[email protected]>
AuthorDate: Tue Oct 6 12:56:42 2026 +0000
[SPARK-59858][SQL] Compare ASOF JOIN STRUCT operands without rebuilding them
### What changes were proposed in this pull request?
SQL ASOF JOIN no longer rebuilds STRUCT operands of `MATCH_CONDITION` from
their fields. It compares
them as they are, since a comparison already matches struct fields by
position.
Each side also picks its own sort keys: a SQL tuple sorts by its parts,
anything else sorts as one value. So the equal-count check in
`SortMergeAsOfJoinExec` is removed.
### Why are the changes needed?
The rebuild turned a NULL struct into `{a: NULL}`, so a NULL struct matched
another NULL struct:
```sql
SELECT r.tag
FROM VALUES (CAST(NULL AS STRUCT<a: INT>), 'l') AS t(k, tag)
ASOF JOIN VALUES (CAST(NULL AS STRUCT<a: INT>), 'rnull') AS r(k, tag)
MATCH_CONDITION (t.k >= r.k);
```
Before: `rnull`. After: no row, the same as a plain `WHERE`.
Splitting struct by its fields also tied NULL with `{NULL, NULL}`, so the
join could return a row that is not the closest.
### Does this PR introduce _any_ user-facing change?
Yes. A NULL STRUCT operand never matches now.
### How was this patch tested?
New tests in `AsOfJoinSortMergeSQLSuite`, with ANSI on and off: NULL
structs (flat and nested), structs of NULL fields, and tuples vs nullable
structs. Three analyzed plans in `join-asof-datatypes.sql.out` now show the
plain comparison.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #59134 from LukaZdravic/asof-null-struct-operand.
Authored-by: Luka Zdravić <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit 43ccc6baa6e55b250faddae0e0a25a431337309c)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../plans/logical/basicLogicalOperators.scala | 90 +++-----------
.../logical/AsOfJoinMatchConditionTypesSuite.scala | 10 --
.../execution/joins/SortMergeAsOfJoinExec.scala | 7 +-
.../analyzer-results/join-asof-datatypes.sql.out | 6 +-
.../spark/sql/AsOfJoinSortMergeSQLSuite.scala | 138 +++++++++++++++++++++
5 files changed, 163 insertions(+), 88 deletions(-)
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 cd6f6f622713..5b5139780131 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
@@ -2866,45 +2866,29 @@ object AsOfJoin {
leftStruct.length == rightStruct.length && leftStruct.nonEmpty
case _ => false
}
-
- /** Whole struct columns with identical schemas sort as a single struct
value. */
- def usesIdenticalStructSort(leftType: DataType, rightType: DataType):
Boolean =
- (leftType, rightType) match {
- case (leftStruct: StructType, rightStruct: StructType) =>
- leftStruct.sameType(rightStruct) && leftStruct.nonEmpty
- case _ => false
- }
}
/**
* Sort-merge ASOF join sorts each side by these expressions (after
equi-keys) so the
* right-side buffer is ordered consistently with the MATCH_CONDITION
comparison (scalar
- * operands are already coerced to a common type; composites are handled
case by case below).
+ * operands are already coerced to a common type).
*
- * 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.
+ * Each side picks its own keys, because the two sides are never compared
key by key. A SQL
+ * tuple `(t.a, t.b)` sorts by its parts. A tuple is never NULL, so this
keeps the comparison
+ * order. Anything else sorts as one value, so a NULL struct comes before a
struct of NULL
+ * fields, the same as in the comparison.
* Array operands sort element-wise; length mismatches follow Spark array
ordering semantics.
*/
def matchSortExpressions(
leftOperand: Expression,
- rightOperand: Expression): (Seq[Expression], Seq[Expression]) = {
- (leftOperand.dataType, rightOperand.dataType) match {
- case (leftStruct: StructType, rightStruct: StructType)
- if MatchConditionTypes.usesIdenticalStructSort(leftStruct,
rightStruct) =>
- 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))
- }
- }
+ rightOperand: Expression): (Seq[Expression], Seq[Expression]) =
+ (matchSortKeys(leftOperand), matchSortKeys(rightOperand))
- /** True for SQL `(col1, col2, ...)` tuple operands, which become
[[CreateNamedStruct]]. */
- private def isSqlTupleStructOperand(operand: Expression): Boolean =
- operand.isInstanceOf[CreateNamedStruct]
+ /** A SQL tuple becomes [[CreateNamedStruct]] and sorts by its parts. */
+ private def matchSortKeys(operand: Expression): Seq[Expression] = operand
match {
+ case tuple: CreateNamedStruct if tuple.valExprs.nonEmpty => tuple.valExprs
+ case other => Seq(other)
+ }
private[catalyst] def normalizeMatchOperands(
leftSet: AttributeSet,
@@ -2943,18 +2927,18 @@ object AsOfJoin {
leftOperand: Expression,
rightOperand: Expression,
operator: MatchComparisonOperator): (Expression, Expression) = {
- val (leftForCompare, rightForCompare) =
- alignOperandsForComparison(leftOperand, rightOperand)
val orderExpression = buildOrderExpression(leftOperand, rightOperand,
operator)
+ // Compare the operands as they are: a comparison already matches struct
fields by position,
+ // and a NULL struct stays NULL, so it never matches.
operator match {
case GreaterThanOrEqualOp =>
- (GreaterThanOrEqual(leftForCompare, rightForCompare), orderExpression)
+ (GreaterThanOrEqual(leftOperand, rightOperand), orderExpression)
case GreaterThanOp =>
- (GreaterThan(leftForCompare, rightForCompare), orderExpression)
+ (GreaterThan(leftOperand, rightOperand), orderExpression)
case LessThanOrEqualOp =>
- (LessThanOrEqual(leftForCompare, rightForCompare), orderExpression)
+ (LessThanOrEqual(leftOperand, rightOperand), orderExpression)
case LessThanOp =>
- (LessThan(leftForCompare, rightForCompare), orderExpression)
+ (LessThan(leftOperand, rightOperand), orderExpression)
}
}
@@ -2979,24 +2963,6 @@ object AsOfJoin {
}
}
- /**
- * 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,
@@ -3131,26 +3097,6 @@ object AsOfJoin {
}
}
- /** Positional struct fields when both operands are the same struct shape. */
- private def decomposeStructOperands(
- leftOperand: Expression,
- rightOperand: Expression): Option[Seq[(Expression, Expression)]] = {
- if (MatchConditionTypes.usesStructDecomposition(
- leftOperand.dataType, rightOperand.dataType)) {
- val leftStruct = leftOperand.dataType.asInstanceOf[StructType]
- val rightStruct = rightOperand.dataType.asInstanceOf[StructType]
- val leftFields = structFieldExprs(leftOperand, leftStruct)
- val rightFields = structFieldExprs(rightOperand, rightStruct)
- if (leftFields.length == rightFields.length) {
- Some(leftFields.zip(rightFields))
- } else {
- None
- }
- } else {
- None
- }
- }
-
private def structFieldExprs(
operand: Expression,
structType: StructType): Seq[Expression] = {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/AsOfJoinMatchConditionTypesSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/AsOfJoinMatchConditionTypesSuite.scala
index 018633767af9..1a32001e6932 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/AsOfJoinMatchConditionTypesSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/AsOfJoinMatchConditionTypesSuite.scala
@@ -118,7 +118,6 @@ class AsOfJoinMatchConditionTypesSuite extends
SparkFunSuite {
Nil)
assert(MatchConditionTypes.areOperandsCompatible(leftStruct, rightStruct))
assert(MatchConditionTypes.usesStructDecomposition(leftStruct,
rightStruct))
- assert(!MatchConditionTypes.usesIdenticalStructSort(leftStruct,
rightStruct))
}
test("nested struct operands are compatible when fields match positionally")
{
@@ -259,15 +258,6 @@ class AsOfJoinMatchConditionTypesSuite extends
SparkFunSuite {
assert(!MatchConditionTypes.usesArrayOrderExpression(emptyArray,
nonEmptyArray))
}
- test("identical struct schemas enable whole-struct sort") {
- val structType = StructType(
- StructField("ts", TimestampType) ::
- StructField("seq", IntegerType) ::
- Nil)
- assert(MatchConditionTypes.usesIdenticalStructSort(structType, structType))
- assert(MatchConditionTypes.usesStructDecomposition(structType, structType))
- }
-
test("struct field count mismatch is not decomposable") {
val leftStruct = StructType(StructField("a", IntegerType) :: Nil)
val rightStruct = StructType(
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 4050da08793d..84eb80616cce 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
@@ -55,9 +55,10 @@ case class SortMergeAsOfJoinExec(
right: SparkPlan,
isSkewJoin: Boolean = false) extends ShuffledJoin with PredicateHelper {
- require(leftSortExprs.nonEmpty && rightSortExprs.nonEmpty &&
- leftSortExprs.length == rightSortExprs.length,
- s"$nodeName requires matching non-empty sort expressions on both sides")
+ // The sides may have different key counts: each side is sorted on its own,
and only the
+ // equi-keys are compared across sides.
+ require(leftSortExprs.nonEmpty && rightSortExprs.nonEmpty,
+ s"$nodeName requires non-empty sort expressions on both sides")
require(Seq(Inner, LeftOuter).exists(joinType == _),
s"$nodeName does not support join type: $joinType")
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-datatypes.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-datatypes.sql.out
index d398440f1edf..ff68c04c7ebe 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-datatypes.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-datatypes.sql.out
@@ -225,7 +225,7 @@ ASOF JOIN (
MATCH_CONDITION (t.k >= r.k)
-- !query analysis
Project [k#x.tag AS tag#x, k#x.inner_seq AS inner_seq#x]
-+- AsOfJoin (named_struct(inner_ts, k#x.inner_ts, inner_seq, k#x.inner_seq,
tag, k#x.tag) >= named_struct(inner_ts, k#x.inner_ts, inner_seq, k#x.inner_seq,
tag, k#x.tag)), Inner
++- AsOfJoin (k#x >= k#x), Inner
:- SubqueryAlias t
: +- LocalRelation [k#x]
+- SubqueryAlias r
@@ -241,7 +241,7 @@ ASOF JOIN VALUES (named_struct('x', 1, 'y', 9)) AS r(k)
MATCH_CONDITION (t.k >= r.k)
-- !query analysis
Project [k#x.y AS matched#x]
-+- AsOfJoin (named_struct(a, k#x.a, b, k#x.b) >= named_struct(x, k#x.x, y,
k#x.y)), Inner
++- AsOfJoin (k#x >= k#x), Inner
:- SubqueryAlias t
: +- LocalRelation [k#x]
+- SubqueryAlias r
@@ -257,7 +257,7 @@ ASOF JOIN VALUES
MATCH_CONDITION (t.k >= r.k)
-- !query analysis
Project [k#x.a AS matched#xL]
-+- AsOfJoin (cast(named_struct(a, k#x.a) as struct<a:bigint>) >=
named_struct(a, k#x.a)), Inner
++- AsOfJoin (cast(k#x as struct<a:bigint>) >= k#x), Inner
:- SubqueryAlias t
: +- LocalRelation [k#x]
+- SubqueryAlias r
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
index 66a3786b2ecf..7791d24a6652 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
@@ -78,6 +78,14 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
|""".stripMargin)
}
+ /** Runs `f` with temp view `r` over `rows`, kept in input order so ties
keep that order. */
+ private def withOrderedRightView(rows: String, columns: String)(f: => Unit):
Unit = {
+ withTempView("r") {
+ sql(s"SELECT * FROM VALUES $rows AS
r($columns)").coalesce(1).createOrReplaceTempView("r")
+ f
+ }
+ }
+
test("INNER ASOF JOIN with TIMESTAMP MATCH_CONDITION") {
setupTradeQuoteViews()
checkSortMergeAsOf(
@@ -440,6 +448,136 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(Timestamp.valueOf("2026-06-29 10:03:00"), 1, "v1.1")))
}
+ gridTest("NULL whole STRUCT column MATCH_CONDITION never matches")(Seq(
+ ("CAST(NULL AS STRUCT<a: INT>)", "named_struct('a', 1)",
"named_struct('a', 5)"),
+ ("CAST(NULL AS STRUCT<a: INT, b: INT>)",
+ "named_struct('a', 1, 'b', 1)", "named_struct('a', 5, 'b', 5)"),
+ ("CAST(NULL AS STRUCT<e: STRUCT<a: INT>>)",
+ "named_struct('e', named_struct('a', 1))", "named_struct('e',
named_struct('a', 5))"),
+ // Non-NULL values of this type are all equal.
+ ("CAST(NULL AS STRUCT<e: STRUCT<>>)",
+ "named_struct('e', named_struct())", "named_struct('e',
named_struct())"))) {
+ case (nullStruct, one, five) =>
+ // A NULL struct must never match, the same as a NULL scalar operand.
+ Seq(true, false).foreach { ansiEnabled =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ def asOfTags(joinType: String, left: String, op: String, rights:
String): DataFrame =
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES ($left, 'l') AS t(k, tag)
+ |$joinType ASOF JOIN VALUES $rights AS r(k, tag)
+ | MATCH_CONDITION (t.k $op r.k)
+ |""".stripMargin)
+ checkSortMergeAsOf(
+ asOfTags("", nullStruct, ">=", s"($nullStruct, 'rnull'), ($one,
'r1')"), Nil)
+ checkSortMergeAsOf(asOfTags("", nullStruct, "<=", s"($one, 'r1')"),
Nil)
+ checkSortMergeAsOf(asOfTags("", five, ">=", s"($nullStruct,
'rnull')"), Nil)
+ // LEFT ASOF JOIN keeps the NULL left row with NULL right columns.
+ checkSortMergeAsOf(
+ asOfTags("LEFT", nullStruct, ">=", s"($nullStruct, 'rnull'),
($one, 'r1')"),
+ Row(null) :: Nil)
+ }
+ }
+ }
+
+ gridTest("whole STRUCT column MATCH_CONDITION matches a struct whose fields
are all NULL")(Seq(
+ ">=" -> Seq(Row("allnull")), "<=" -> Seq(Row("allnull")), ">" -> Nil,
"<" -> Nil)) {
+ case (op, expected) =>
+ // Two structs of NULL fields are equal, and a NULL struct never matches.
+ val allNull = "named_struct('a', CAST(NULL AS INT), 'b', CAST(NULL AS
INT))"
+ val nullStruct = "CAST(NULL AS STRUCT<a: INT, b: INT>)"
+ Seq(true, false).foreach { ansiEnabled =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES ($allNull, 'l') AS t(k, tag)
+ |ASOF JOIN VALUES ($nullStruct, 'rnull'), ($allNull,
'allnull') AS r(k, tag)
+ | MATCH_CONDITION (t.k $op r.k)
+ |""".stripMargin),
+ expected)
+ }
+ }
+ }
+
+ test("nested STRUCT column MATCH_CONDITION keeps a NULL inner struct apart
from NULL fields") {
+ // {e: NULL} sorts before {e: {a: NULL}}, as in a plain comparison.
+ val eNull = "named_struct('e', CAST(NULL AS STRUCT<a: INT>))"
+ val aNull = "named_struct('e', named_struct('a', CAST(NULL AS INT)))"
+ for {
+ (left, op, right) <- Seq((aNull, ">", eNull), (eNull, "<", aNull))
+ ansiEnabled <- Seq(true, false)
+ } {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES ($left, 'l') AS t(k, tag)
+ |ASOF JOIN VALUES ($right, 'r') AS r(k, tag)
+ | MATCH_CONDITION (t.k $op r.k)
+ |""".stripMargin),
+ Row("r") :: Nil)
+ }
+ }
+ }
+
+ test("tuple vs nullable STRUCT column MATCH_CONDITION: a NULL struct does
not end the scan") {
+ // r.s sorts as one value, so NULL comes before {NULL, NULL}. Sorting by
its fields would tie
+ // them, and in one of these input orders the NULL row would end the scan
before r11.
+ val allNull = "(named_struct('a', CAST(NULL AS INT), 'b', CAST(NULL AS
INT)), 'allnull')"
+ val nullStruct = "(CAST(NULL AS STRUCT<a: INT, b: INT>), 'rnull')"
+ val r11 = "(named_struct('a', 1, 'b', 1), 'r11')"
+ for {
+ rights <- Seq(s"$allNull, $nullStruct, $r11", s"$nullStruct, $allNull,
$r11")
+ ansiEnabled <- Seq(true, false)
+ } {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ withOrderedRightView(rights, "s, tag") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.tag
+ |FROM VALUES (1, 2) AS t(a, b)
+ |ASOF JOIN r
+ | MATCH_CONDITION ((t.a, t.b) >= r.s)
+ |""".stripMargin),
+ Row("r11") :: Nil)
+ }
+ }
+ }
+ }
+
+ test("tuple with a nullable inner STRUCT MATCH_CONDITION picks the closest
row") {
+ // NULL and {a: NULL} have the same field, but the comparison ranks NULL
lower. Both input
+ // orders must pick the row equal to the left one, in both directions.
+ val aNull = "named_struct('a', CAST(NULL AS INT))"
+ val nullA = "CAST(NULL AS STRUCT<a: INT>)"
+ val rAnull = s"(1, $aNull, 'r_anull')"
+ val rNull = s"(1, $nullA, 'r_null')"
+ for {
+ rights <- Seq(s"$rAnull, $rNull", s"$rNull, $rAnull")
+ ansiEnabled <- Seq(true, false)
+ } {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ withOrderedRightView(rights, "x, s, tag") {
+ def asOfTag(left: String, op: String): DataFrame =
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (1, $left) AS t(x, s)
+ |ASOF JOIN r
+ | MATCH_CONDITION ((t.x, t.s) $op (r.x, r.s))
+ |""".stripMargin)
+ checkSortMergeAsOf(asOfTag(aNull, ">="), Row("r_anull") :: Nil)
+ checkSortMergeAsOf(asOfTag(nullA, "<="), Row("r_null") :: Nil)
+ }
+ }
+ }
+ }
+
test("whole STRUCT column with an ARRAY field MATCH_CONDITION") {
// The array field is compared as a whole; field 'a' ties, so the array
field decides.
checkSortMergeAsOf(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
