This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 6f0c4791a1f4 [SPARK-59749][SQL] Accept empty STRUCT operands in ASOF
JOIN MATCH_CONDITION
6f0c4791a1f4 is described below
commit 6f0c4791a1f4bceaa9749c913dc80cef47918a5b
Author: Luka Zdravić <[email protected]>
AuthorDate: Wed Sep 30 13:07:18 2026 +0000
[SPARK-59749][SQL] Accept empty STRUCT operands in ASOF JOIN MATCH_CONDITION
### What changes were proposed in this pull request?
ASOF JOIN `MATCH_CONDITION` now accepts operands that are, or contain, an
empty struct (`STRUCT<>`). The operand check was "orderable and contains no
empty struct". It is now only "orderable", the same rule `min_by` and `max_by`
use for their ordering column. An array of empty structs compares each element
as one value.
Now accepted (same type on both sides): `STRUCT<>`, `STRUCT<x:
STRUCT<>>`, `ARRAY<STRUCT<>>`. An empty struct against a non-empty struct stays
rejected.
While testing these operands, I found and fixed one more bug. An
`ARRAY<STRUCT>` operand whose fields have different types (for example `INT`
and `TIMESTAMP`) passed the type check, then failed with
`DATATYPE_MISMATCH.DATA_DIFF_TYPES`. The per-element distance now keeps the
field distances in a struct instead of an array.
Forward matches (`<`, `<=`) stay as they are on master. SPARK-59829
(#59102) changes them to keep the first match.
### Why are the changes needed?
The comparison operator already accepts these pairs: `named_struct() >=
named_struct()` is `true`, because two empty structs are equal. ASOF rejected
them with an error that says the operands must be orderable, but `STRUCT<>` is
orderable.
The `ARRAY<STRUCT>` bug gave a confusing error for a valid query.
### Does this PR introduce _any_ user-facing change?
Yes. This query failed and now runs:
```sql
SELECT * FROM VALUES (named_struct()) AS l(s)
ASOF JOIN VALUES (named_struct()) AS r(s)
MATCH_CONDITION (l.s >= r.s);
-- before: [ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE] ... Got: "STRUCT<>"
and "STRUCT<>".
```
Two empty structs are equal, so `>=` and `<=` match, and `>` and `<` find
no match. A NULL empty struct never matches.
`ARRAY<STRUCT>` operands with fields of different types now run instead
of failing with `DATATYPE_MISMATCH.DATA_DIFF_TYPES`.
### How was this patch tested?
Unit tests in `AsOfJoinMatchConditionTypesSuite`, `ResolveAsOfJoinSuite`,
and `AsOfJoinSQLSuite` cover the accepted and rejected type pairs and the array
element distance. Execution tests in `AsOfJoinSortMergeSQLSuite` check the
matched row for all four operators, for: an empty struct, a struct with only an
empty field, a NULL empty struct, a struct with an empty field and an int field
(column and tuple form), arrays of empty structs, and `ARRAY<STRUCT>` operands
with fields of diff [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #59016 from LukaZdravic/SPARK-59749-asof-empty-struct.
Authored-by: Luka Zdravić <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
---
.../plans/logical/basicLogicalOperators.scala | 30 ++----
.../catalyst/analysis/ResolveAsOfJoinSuite.scala | 38 ++++++++
.../logical/AsOfJoinMatchConditionTypesSuite.scala | 40 +++++---
.../analyzer-results/join-asof-datatypes.sql.out | 78 ++++++++++++++++
.../sql-tests/inputs/join-asof-datatypes.sql | 39 ++++++++
.../sql-tests/results/join-asof-datatypes.sql.out | 63 +++++++++++++
.../org/apache/spark/sql/AsOfJoinSQLSuite.scala | 10 +-
.../spark/sql/AsOfJoinSortMergeSQLSuite.scala | 101 +++++++++++++++++++++
8 files changed, 358 insertions(+), 41 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 e4c5d955cf81..dd44a8b25912 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
@@ -2772,8 +2772,7 @@ object AsOfJoin {
*/
private[catalyst] object MatchConditionTypes {
- def isValidOperandType(dataType: DataType): Boolean =
- RowOrdering.isOrderable(dataType) && !containsEmptyStructType(dataType)
+ def isValidOperandType(dataType: DataType): Boolean =
RowOrdering.isOrderable(dataType)
/** Whether the `>=` this join builds can compare the two operands. */
def areOperandsCompatible(leftType: DataType, rightType: DataType):
Boolean = {
@@ -2866,13 +2865,6 @@ object AsOfJoin {
leftStruct.sameType(rightStruct) && leftStruct.nonEmpty
case _ => false
}
-
- private def containsEmptyStructType(dataType: DataType): Boolean =
dataType match {
- case struct: StructType =>
- struct.isEmpty || struct.exists(field =>
containsEmptyStructType(field.dataType))
- case ArrayType(elementType, _) => containsEmptyStructType(elementType)
- case _ => false
- }
}
/**
@@ -3057,15 +3049,14 @@ object AsOfJoin {
val leftArray = castArrayElementType(leftOperand, elementType)
val rightArray = castArrayElementType(rightOperand, elementType)
elementType match {
- case struct: StructType =>
+ // An empty struct has no fields to split into, so it takes the
whole-value arm below.
+ case struct: StructType if struct.nonEmpty =>
val leftElement = NamedLambdaVariable("left_elem", struct, nullable =
true)
val rightElement = NamedLambdaVariable("right_elem", struct, nullable
= true)
val leafDiffs = collectStructLeafPairs(leftElement, rightElement,
struct).map {
case (left, right) => buildLeafOrderExpression(left, right, operator)
}
- val elementOrder = wrapCompositeOrderExpression(
- leafDiffs,
- ArrayType(struct, containsNull = true))
+ val elementOrder = wrapCompositeOrderExpression(leafDiffs)
ZipWith(
leftArray,
rightArray,
@@ -3103,7 +3094,7 @@ object AsOfJoin {
val leafDiffs = collectStructLeafPairs(leftOperand, rightOperand,
structType).map {
case (left, right) => buildLeafOrderExpression(left, right, operator)
}
- wrapCompositeOrderExpression(leafDiffs, structType)
+ wrapCompositeOrderExpression(leafDiffs)
}
private def collectStructLeafPairs(
@@ -3123,16 +3114,11 @@ object AsOfJoin {
}
}
- private def wrapCompositeOrderExpression(
- diffs: Seq[Expression],
- compositeType: DataType): Expression = {
+ /** Groups multiple field distances in a struct because their types may
differ. */
+ private def wrapCompositeOrderExpression(diffs: Seq[Expression]): Expression
= {
diffs match {
case Seq(single) => single
- case _ =>
- compositeType match {
- case _: ArrayType => CreateArray(diffs)
- case _ => CreateStruct(diffs)
- }
+ case _ => CreateStruct(diffs)
}
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoinSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoinSuite.scala
index 80674833065c..04d572e695dc 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoinSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoinSuite.scala
@@ -81,6 +81,20 @@ class ResolveAsOfJoinSuite extends AnalysisTest {
private val leftStruct: LogicalPlan = LocalRelation(lstruct)
private val rightStruct: LogicalPlan = LocalRelation(rstruct)
+ // Arrays of empty structs: each element has no fields, so it is compared as
one value.
+ private val lemptyArr = AttributeReference("ea",
ArrayType(StructType(Nil)))()
+ private val remptyArr = AttributeReference("ea",
ArrayType(StructType(Nil)))()
+ private val leftEmptyArr: LogicalPlan = LocalRelation(lemptyArr)
+ private val rightEmptyArr: LogicalPlan = LocalRelation(remptyArr)
+
+ // Arrays of structs whose fields differ in type, so their per-field
distances differ in type.
+ private val mixedElement =
+ StructType(StructField("f", IntegerType) :: StructField("ts",
TimestampType) :: Nil)
+ private val lmixedArr = AttributeReference("ma", ArrayType(mixedElement))()
+ private val rmixedArr = AttributeReference("ma", ArrayType(mixedElement))()
+ private val leftMixedArr: LogicalPlan = LocalRelation(lmixedArr)
+ private val rightMixedArr: LogicalPlan = LocalRelation(rmixedArr)
+
/** Build an [[AsOfJoin]] from a `MATCH_CONDITION` whose operands are
already resolved. */
private def asOf(
leftExpr: Expression = la,
@@ -186,6 +200,30 @@ class ResolveAsOfJoinSuite extends AnalysisTest {
assert(body.isInstanceOf[Subtract], s"expected a Subtract element
distance, got $body")
}
+ test("materializes an array of empty structs into a whole-element signed
distance") {
+ val resolved = ResolveAsOfJoin.apply(
+ asOf(leftExpr = lemptyArr, rightExpr = remptyArr, l = leftEmptyArr, r =
rightEmptyArr))
+ .asInstanceOf[AsOfJoin]
+ val zip = resolved.orderExpression.asInstanceOf[ZipWith]
+ // An empty struct has no fields to split, so each element gets one
whole-value distance.
+ val body = zip.function.asInstanceOf[LambdaFunction].function
+ assert(body.isInstanceOf[If], s"expected a signed element distance, got
$body")
+ assert(body.dataType == IntegerType)
+ }
+
+ test("materializes an ARRAY<STRUCT> with mixed field types into a struct
element distance") {
+ val resolved = ResolveAsOfJoin.apply(
+ asOf(leftExpr = lmixedArr, rightExpr = rmixedArr, l = leftMixedArr, r =
rightMixedArr))
+ .asInstanceOf[AsOfJoin]
+ val zip = resolved.orderExpression.asInstanceOf[ZipWith]
+ // The INT and TIMESTAMP field distances differ in type, so one array
could not hold both.
+ val body = zip.function.asInstanceOf[LambdaFunction].function
+ assert(body.isInstanceOf[CreateNamedStruct], s"expected a struct element
distance, got $body")
+ val fields = body.asInstanceOf[CreateNamedStruct].valExprs
+ assert(fields.size == 2)
+ assert(fields.forall(_.isInstanceOf[Subtract]), s"expected per-field
Subtracts, got $fields")
+ }
+
test("materializes a positional struct operand into a flattened per-field
distance") {
val resolved = ResolveAsOfJoin.apply(
asOf(leftExpr = lstruct, rightExpr = rstruct, l = leftStruct, r =
rightStruct))
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 c8063f35ad9d..018633767af9 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
@@ -18,7 +18,6 @@
package org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.catalyst.expressions.RowOrdering
import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin.MatchConditionTypes
import org.apache.spark.sql.types._
@@ -231,17 +230,33 @@ class AsOfJoinMatchConditionTypesSuite extends
SparkFunSuite {
assert(!MatchConditionTypes.usesArrayOrderExpression(leftArray,
diffNameArray))
}
- test("empty struct operands are invalid") {
+ test("empty struct operands are valid and compare as a whole value") {
val emptyStruct = StructType(Nil)
- assert(!MatchConditionTypes.isValidOperandType(emptyStruct))
- assert(!MatchConditionTypes.areOperandsCompatible(emptyStruct,
emptyStruct))
+ assert(MatchConditionTypes.isValidOperandType(emptyStruct))
+ assert(MatchConditionTypes.areOperandsCompatible(emptyStruct, emptyStruct))
+ // Splitting into zero fields would compare two empty literals, so a NULL
operand would match.
assert(!MatchConditionTypes.usesStructDecomposition(emptyStruct,
emptyStruct))
}
- test("nested empty struct operands are invalid") {
+ test("nested empty struct operands are valid") {
val nestedEmptyStruct = StructType(StructField("x", StructType(Nil)) ::
Nil)
- assert(!MatchConditionTypes.isValidOperandType(nestedEmptyStruct))
- assert(!MatchConditionTypes.areOperandsCompatible(nestedEmptyStruct,
nestedEmptyStruct))
+ assert(MatchConditionTypes.isValidOperandType(nestedEmptyStruct))
+ assert(MatchConditionTypes.areOperandsCompatible(nestedEmptyStruct,
nestedEmptyStruct))
+ assert(MatchConditionTypes.usesStructDecomposition(nestedEmptyStruct,
nestedEmptyStruct))
+ }
+
+ test("empty struct and non-empty struct operands are incompatible") {
+ val emptyStruct = StructType(Nil)
+ val nonEmptyStruct = StructType(StructField("a", IntegerType) :: Nil)
+ assert(!MatchConditionTypes.areOperandsCompatible(emptyStruct,
nonEmptyStruct))
+ assert(!MatchConditionTypes.areOperandsCompatible(nonEmptyStruct,
emptyStruct))
+ val nestedEmpty = StructType(StructField("x", emptyStruct) :: Nil)
+ val nestedNonEmpty = StructType(StructField("x", nonEmptyStruct) :: Nil)
+ assert(!MatchConditionTypes.areOperandsCompatible(nestedEmpty,
nestedNonEmpty))
+ val emptyArray = ArrayType(emptyStruct)
+ val nonEmptyArray = ArrayType(nonEmptyStruct)
+ assert(!MatchConditionTypes.areOperandsCompatible(emptyArray,
nonEmptyArray))
+ assert(!MatchConditionTypes.usesArrayOrderExpression(emptyArray,
nonEmptyArray))
}
test("identical struct schemas enable whole-struct sort") {
@@ -300,14 +315,11 @@ class AsOfJoinMatchConditionTypesSuite extends
SparkFunSuite {
assert(!MatchConditionTypes.areOperandsCompatible(leftStruct, rightStruct))
}
- test("array operands with empty struct elements are invalid") {
- // An array whose element contains an empty struct is invalid even though
the array itself
- // is orderable (the ArrayType arm of containsEmptyStructType).
+ test("array operands with empty struct elements are valid") {
val arrayOfEmptyStruct = ArrayType(StructType(Nil))
- // Pin orderability so the rejection is due to the empty struct, not
non-orderability.
- assert(RowOrdering.isOrderable(arrayOfEmptyStruct))
- assert(!MatchConditionTypes.isValidOperandType(arrayOfEmptyStruct))
- assert(!MatchConditionTypes.areOperandsCompatible(arrayOfEmptyStruct,
arrayOfEmptyStruct))
+ assert(MatchConditionTypes.isValidOperandType(arrayOfEmptyStruct))
+ assert(MatchConditionTypes.areOperandsCompatible(arrayOfEmptyStruct,
arrayOfEmptyStruct))
+ assert(MatchConditionTypes.usesArrayOrderExpression(arrayOfEmptyStruct,
arrayOfEmptyStruct))
}
test("array operands with incompatible struct element field types are
rejected") {
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 161005b5bef4..d398440f1edf 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
@@ -264,6 +264,20 @@ Project [k#x.a AS matched#xL]
+- LocalRelation [k#x]
+-- !query
+SELECT t.id, r.tag AS matched
+FROM VALUES (1, named_struct()) AS t(id, k) ASOF JOIN
+ VALUES (named_struct(), 'hit') AS r(k, tag)
+ MATCH_CONDITION (t.k >= r.k)
+-- !query analysis
+Project [id#x, tag#x AS matched#x]
++- AsOfJoin (k#x >= k#x), Inner
+ :- SubqueryAlias t
+ : +- LocalRelation [id#x, k#x]
+ +- SubqueryAlias r
+ +- LocalRelation [k#x, tag#x]
+
+
-- !query
SELECT t.k, r.k AS matched_k
FROM VALUES (CAST(10 AS TINYINT)) AS t(k) ASOF JOIN VALUES (CAST(5 AS BIGINT))
AS r(k)
@@ -462,3 +476,67 @@ Project [c1#x, c2#x]
: +- LocalRelation [c1#x, c2#x]
+- SubqueryAlias r
+- LocalRelation [c1#x, c2#x]
+
+
+-- !query
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a >= r.a)
+-- !query analysis
+Project [size(a#x, false) AS matched_size#x]
++- AsOfJoin (a#x >= a#x), Inner
+ :- SubqueryAlias t
+ : +- LocalRelation [a#x]
+ +- SubqueryAlias r
+ +- LocalRelation [a#x]
+
+
+-- !query
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a > r.a)
+-- !query analysis
+Project [size(a#x, false) AS matched_size#x]
++- AsOfJoin (a#x > a#x), Inner
+ :- SubqueryAlias t
+ : +- LocalRelation [a#x]
+ +- SubqueryAlias r
+ +- LocalRelation [a#x]
+
+
+-- !query
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a <= r.a)
+-- !query analysis
+Project [size(a#x, false) AS matched_size#x]
++- AsOfJoin (a#x <= a#x), Inner
+ :- SubqueryAlias t
+ : +- LocalRelation [a#x]
+ +- SubqueryAlias r
+ +- LocalRelation [a#x]
+
+
+-- !query
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a < r.a)
+-- !query analysis
+Project [size(a#x, false) AS matched_size#x]
++- AsOfJoin (a#x < a#x), Inner
+ :- SubqueryAlias t
+ : +- LocalRelation [a#x]
+ +- SubqueryAlias r
+ +- LocalRelation [a#x]
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/join-asof-datatypes.sql
b/sql/core/src/test/resources/sql-tests/inputs/join-asof-datatypes.sql
index c8ae66b7d444..87c4a8b09158 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/join-asof-datatypes.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/join-asof-datatypes.sql
@@ -120,6 +120,12 @@ ASOF JOIN VALUES
(named_struct('a', CAST(4 AS BIGINT))) AS r(k)
MATCH_CONDITION (t.k >= r.k);
+-- FVT-ASOF-4-012c: empty STRUCT operands (SPARK-59749); two empty structs are
equal
+SELECT t.id, r.tag AS matched
+FROM VALUES (1, named_struct()) AS t(id, k) ASOF JOIN
+ VALUES (named_struct(), 'hit') AS r(k, tag)
+ MATCH_CONDITION (t.k >= r.k);
+
-- FVT-ASOF-4-013: coercion TINYINT vs BIGINT
SELECT t.k, r.k AS matched_k
FROM VALUES (CAST(10 AS TINYINT)) AS t(k) ASOF JOIN VALUES (CAST(5 AS BIGINT))
AS r(k)
@@ -207,6 +213,39 @@ 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));
+-- FVT-ASOF-4-018b: ARRAY of empty STRUCT (SPARK-59749) with >=. The elements
tie, so length
+-- orders the arrays. Sizes 1 and 2 pass; the closest is size 2.
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a >= r.a);
+
+-- FVT-ASOF-4-018c: ARRAY of empty STRUCT with >. Only size 1 passes.
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a > r.a);
+
+-- FVT-ASOF-4-018d: ARRAY of empty STRUCT with <=. Sizes 2 and 3 pass; the
closest is size 2.
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a <= r.a);
+
+-- FVT-ASOF-4-018e: ARRAY of empty STRUCT with <. Only size 3 passes.
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a < r.a);
+
-- FVT-ASOF-4-019: MAP operand rejection — covered by FVT-ASOF-3-010
-- FVT-ASOF-4-020: ARRAY<MAP> rejection — covered by FVT-ASOF-3-011
diff --git
a/sql/core/src/test/resources/sql-tests/results/join-asof-datatypes.sql.out
b/sql/core/src/test/resources/sql-tests/results/join-asof-datatypes.sql.out
index be069595721a..4b3aa32582ba 100644
--- a/sql/core/src/test/resources/sql-tests/results/join-asof-datatypes.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/join-asof-datatypes.sql.out
@@ -207,6 +207,17 @@ struct<matched:bigint>
1
+-- !query
+SELECT t.id, r.tag AS matched
+FROM VALUES (1, named_struct()) AS t(id, k) ASOF JOIN
+ VALUES (named_struct(), 'hit') AS r(k, tag)
+ MATCH_CONDITION (t.k >= r.k)
+-- !query schema
+struct<id:int,matched:string>
+-- !query output
+1 hit
+
+
-- !query
SELECT t.k, r.k AS matched_k
FROM VALUES (CAST(10 AS TINYINT)) AS t(k) ASOF JOIN VALUES (CAST(5 AS BIGINT))
AS r(k)
@@ -361,3 +372,55 @@ FROM VALUES (1, 2) AS t(c1, c2) ASOF JOIN VALUES (1, 1) AS
r(c1, c2)
struct<c1:int,c2:int>
-- !query output
1 1
+
+
+-- !query
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a >= r.a)
+-- !query schema
+struct<matched_size:int>
+-- !query output
+2
+
+
+-- !query
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a > r.a)
+-- !query schema
+struct<matched_size:int>
+-- !query output
+1
+
+
+-- !query
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a <= r.a)
+-- !query schema
+struct<matched_size:int>
+-- !query output
+2
+
+
+-- !query
+SELECT size(r.a) AS matched_size
+FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a) ASOF JOIN
+ VALUES (ARRAY(named_struct())),
+ (ARRAY(named_struct(), named_struct())),
+ (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ MATCH_CONDITION (t.a < r.a)
+-- !query schema
+struct<matched_size:int>
+-- !query output
+3
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 edbc6d866747..f9cdcbeeea5d 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
@@ -511,11 +511,11 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
assert(asOfJoin.asOfCondition.resolved)
}
- test("MATCH_CONDITION rejects empty STRUCT operands") {
+ test("MATCH_CONDITION rejects an empty STRUCT paired with a non-empty
STRUCT") {
val sqlText =
"""
|SELECT *
- |FROM VALUES (named_struct()) AS t(s) ASOF JOIN VALUES
(named_struct()) AS r(s)
+ |FROM VALUES (named_struct()) AS t(s) ASOF JOIN VALUES
(named_struct('a', 1)) AS r(s)
| MATCH_CONDITION (t.s >= r.s)
|""".stripMargin
checkError(
@@ -524,12 +524,12 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
sqlState = Some("42K09"),
parameters = Map(
"type1" -> "\"STRUCT<>\"",
- "type2" -> "\"STRUCT<>\""),
+ "type2" -> "\"STRUCT<a: INT NOT NULL>\""),
queryContext = Array(
ExpectedContext(
- fragment = """ASOF JOIN VALUES (named_struct()) AS r(s)
+ fragment = """ASOF JOIN VALUES (named_struct('a', 1)) AS r(s)
| MATCH_CONDITION (t.s >= r.s)""".stripMargin,
start = 47,
- stop = 118)))
+ stop = 124)))
}
}
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 ee2debcbfc92..200800f08ab3 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
@@ -435,6 +435,67 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(Row(5, Seq(1, 1))) :: Nil)
}
+ test("empty STRUCT column MATCH_CONDITION") {
+ // Empty structs are equal, also when nested, so inclusive operators match
and strict do not.
+ for {
+ emptyStruct <- Seq("named_struct()", "named_struct('e', named_struct())")
+ (op, tag) <- Seq(">=" -> "hit", "<=" -> "hit", ">" -> null, "<" -> null)
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES ($emptyStruct) AS t(s)
+ |LEFT ASOF JOIN VALUES ($emptyStruct, 'hit') AS r(s, tag)
+ | MATCH_CONDITION (t.s $op r.s)
+ |""".stripMargin),
+ Row(tag) :: Nil)
+ }
+ }
+
+ test("NULL empty STRUCT operand never matches") {
+ // A NULL operand fails every comparison, though any two non-NULL empty
structs are equal.
+ for {
+ (leftValue, rightValue) <- Seq(
+ "CAST(NULL AS STRUCT<>)" -> "named_struct()",
+ "named_struct()" -> "CAST(NULL AS STRUCT<>)")
+ op <- Seq(">=", ">", "<=", "<")
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES ($leftValue) AS t(s)
+ |LEFT ASOF JOIN VALUES ($rightValue, 'hit') AS r(s, tag)
+ | MATCH_CONDITION (t.s $op r.s)
+ |""".stripMargin),
+ Row(null) :: Nil)
+ }
+ }
+
+ test("STRUCT with an empty STRUCT field MATCH_CONDITION") {
+ // The empty field always ties, so field 'seq' alone picks the nearest
match to 5.
+ for {
+ (left, right) <- Seq("t.k" -> "r.k", "(t.k.e, t.k.seq)" -> "(r.k.e,
r.k.seq)")
+ (op, seq) <- Seq(">=" -> 5, ">" -> 4, "<=" -> 5, "<" -> 6)
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.k.seq
+ |FROM VALUES (named_struct('e', named_struct(), 'seq', 5)) AS t(k)
+ |ASOF JOIN VALUES
+ | (named_struct('e', named_struct(), 'seq', 3)),
+ | (named_struct('e', named_struct(), 'seq', 4)),
+ | (named_struct('e', named_struct(), 'seq', 5)),
+ | (named_struct('e', named_struct(), 'seq', 6)),
+ | (named_struct('e', named_struct(), 'seq', 7)) AS r(k)
+ | MATCH_CONDITION ($left $op $right)
+ |""".stripMargin),
+ Row(seq) :: Nil)
+ }
+ }
+
test("ARRAY<INT> MATCH_CONDITION") {
checkSortMergeAsOf(
sql(
@@ -547,6 +608,24 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(Seq(Row(1, 4))) :: Nil)
}
+ test("ARRAY of empty STRUCT MATCH_CONDITION") {
+ // Empty struct elements always tie, so the arrays order by length alone.
+ Seq(">=" -> 2, ">" -> 1, "<=" -> 2, "<" -> 3).foreach { case (op, size) =>
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT size(r.a)
+ |FROM VALUES (ARRAY(named_struct(), named_struct())) AS t(a)
+ |ASOF JOIN VALUES
+ | (ARRAY(named_struct())),
+ | (ARRAY(named_struct(), named_struct())),
+ | (ARRAY(named_struct(), named_struct(), named_struct())) AS r(a)
+ | MATCH_CONDITION (t.a $op r.a)
+ |""".stripMargin),
+ Row(size) :: Nil)
+ }
+ }
+
test("STRUCT tuple from scalar columns MATCH_CONDITION") {
checkSortMergeAsOf(
sql(
@@ -728,6 +807,28 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(Seq(1, 6)) :: Nil)
}
+ test("ARRAY<STRUCT> MATCH_CONDITION with fields of different types") {
+ // Each element has an INT distance and an INTERVAL distance.
+ def element(first: String, time: String): String =
+ s"ARRAY(named_struct('f', $first, 'ts', TIMESTAMP '2026-06-29 $time'))"
+ for {
+ first <- Seq("1", "named_struct()")
+ (op, tag) <- Seq(">=" -> "r9", ">" -> "r9", "<=" -> "r11", "<" -> "r11")
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (${element(first, "10:00:00")}) AS t(a)
+ |ASOF JOIN VALUES
+ | (${element(first, "09:00:00")}, 'r9'),
+ | (${element(first, "11:00:00")}, 'r11') AS r(a, tag)
+ | MATCH_CONDITION (t.a $op r.a)
+ |""".stripMargin),
+ Row(tag) :: Nil)
+ }
+ }
+
test("ARRAY<STRUCT> operands with different field names") {
checkSortMergeAsOf(
sql(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
