This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 169c7ecb8883 [SPARK-60517][SQL] Fix mixed timestamp ordering in SQL
ASOF joins
169c7ecb8883 is described below
commit 169c7ecb8883401d04fee3c06317c63e923f481c
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Tue Oct 27 05:14:37 2026 +0000
[SPARK-60517][SQL] Fix mixed timestamp ordering in SQL ASOF joins
### What changes were proposed in this pull request?
Fixes [SPARK-60517](https://issues.apache.org/jira/browse/SPARK-60517).
Fix SQL ASOF sorting for mixed `TIMESTAMP_LTZ` and `TIMESTAMP_NTZ` keys by
converting right-side NTZ leaves to LTZ when the session time zone can change
their order. Apply this recursively to tuples, structs, and arrays, preserving
each leaf's timestamp precision, other field types, and composite null
semantics.
Left sort keys do not need this timestamp cast with the current scanner: it
searches the right buffer independently for each left row. The left sort cast
is deferred to the streaming-cursor PR, which will need left keys sorted in the
comparison's timestamp order too.
LTZ sort keys keep their original precision. In fixed-offset zones such as
UTC or `+05:30`, right NTZ sort keys also remain uncast because conversion
preserves their order. Comparisons use Spark's normal type coercion, avoiding
an extra timestamp-only cast in composite comparisons.
A separate PR can optimize computed keys by evaluating each key once in a
logical projection and reusing the resulting column for sorting and matching.
Each helper projection needs its own statistics and logical link for AQE.
### Why are the changes needed?
SQL ASOF compares mixed timestamp keys as LTZ values but sorts them in
their original types. Converting NTZ values to LTZ can change their order
across a daylight-saving gap, so the scan can select the wrong row.
For example:
```sql
SET spark.sql.join.asofJoin.enabled = true;
SET spark.sql.session.timeZone = America/Los_Angeles;
SELECT r.id
FROM VALUES (TIMESTAMP_LTZ '2024-03-10 10:05:00Z') AS l(ts)
ASOF JOIN VALUES
(TIMESTAMP_NTZ '2024-03-10 02:50:00', 1),
(TIMESTAMP_NTZ '2024-03-10 03:00:00', 2),
(TIMESTAMP_NTZ '2024-03-10 03:15:00', 3) AS r(ts, id)
MATCH_CONDITION (l.ts <= r.ts);
```
This returns `1` instead of `3`. The right-side NTZ values sort as `02:50`,
`03:00`, `03:15`, but convert to `10:50Z`, `10:00Z`, `10:15Z`. The fix sorts by
those converted instants. The same problem affects backward matching and
timestamp fields inside composite keys.
### Does this PR introduce _any_ user-facing change?
Yes. SQL ASOF joins with mixed LTZ/NTZ keys return the correct match when
time-zone conversion changes the key order, including for composite keys.
### How was this patch tested?
Added DST regressions for both directions and timestamp orientations, with
scalar, tuple, struct, array, nested, and NULL-struct keys. The tests cover
both analyzers; top-level arrays use the fixed-point analyzer because
single-pass lambda ordering is already unsupported. Additional checks cover
microsecond/nanosecond combinations, raw sort keys in fixed-offset zones,
uncast left timestamp sort keys, and a single composite cast in the match
condition. Strengthened the FVT-ASOF-4-016 g [...]
All 145 tests passed on the final changes:
- `AsOfJoinMatchConditionTypesSuite`: 34 tests.
- Full `AsOfJoinSortMergeSQLSuite` and `AsOfJoinSQLSuite`: 109 tests.
- `SQLQueryTestSuite`, filtered to `join-asof-datatypes.sql`: execution and
analyzer goldens.
Catalyst production/test and SQL test scalastyle checks passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Closes #59753 from vranes/asof-sql-timestamp-ordering.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../plans/logical/basicLogicalOperators.scala | 81 +++++++----
.../logical/AsOfJoinMatchConditionTypesSuite.scala | 105 +++++++++++++-
.../analyzer-results/join-asof-datatypes.sql.out | 71 ++++++++--
.../sql-tests/inputs/join-asof-datatypes.sql | 39 +++++-
.../sql-tests/results/join-asof-datatypes.sql.out | 65 ++++++++-
.../spark/sql/AsOfJoinSortMergeSQLSuite.scala | 155 ++++++++++++++++++++-
6 files changed, 467 insertions(+), 49 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 a08214404c6e..6dfa5c38b1b2 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
@@ -2755,36 +2755,68 @@ object AsOfJoin {
rightOperand: Expression,
normalizedOp: MatchComparisonOperator)
: (Expression, Expression, Seq[Expression], Seq[Expression]) = {
- val (coercedLeft, coercedRight) = coerceMatchLeafOperands(leftOperand,
rightOperand)
+ val (coercedLeft, coercedRight) = coerceMatchStringOperands(leftOperand,
rightOperand)
val (asOfCondition, orderExpression) =
buildMatchExpressions(coercedLeft, coercedRight, normalizedOp)
- val (leftSortExprs, rightSortExprs) = matchSortExpressions(coercedLeft,
coercedRight)
+ val timeZone = DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone)
+ val rightTarget = if (timeZone.getRules.isFixedOffset) {
+ // A fixed offset preserves the order of NTZ values, including inside
composite keys.
+ coercedRight.dataType
+ } else {
+ rightTimestampSortType(coercedLeft.dataType, coercedRight.dataType)
+ }
+ // The scan searches the right buffer independently for each left row.
+ val (leftSortExprs, rightSortExprs) = matchSortExpressions(
+ coercedLeft, castMatchOperand(coercedRight, rightTarget))
(asOfCondition, orderExpression, leftSortExprs, rightSortExprs)
}
- /**
- * Casts a scalar operand pair to the comparison operator's common type so
the comparison,
- * ordering, and sort keys agree (needed for string vs
DATE/TIMESTAMP/number, where the buffer
- * would otherwise sort by a different type). STRUCT/ARRAY operands are left
uncoerced.
- */
- private def coerceMatchLeafOperands(
+ /** Coerces string/non-string operands so sorting and comparison use the
same value domain. */
+ private def coerceMatchStringOperands(
leftOperand: Expression,
rightOperand: Expression): (Expression, Expression) = {
- val leftType = leftOperand.dataType
- val rightType = rightOperand.dataType
- if (MatchConditionTypes.isCompositeOperand(leftType, rightType)) {
- (leftOperand, rightOperand)
- } else {
- MatchConditionTypes.stringComparisonCommonType(leftType, rightType)
match {
- case Some(commonType) =>
- (castMatchOperand(leftOperand, commonType),
castMatchOperand(rightOperand, commonType))
- case None => (leftOperand, rightOperand)
- }
+ val commonType = MatchConditionTypes.stringComparisonCommonType(
+ leftOperand.dataType, rightOperand.dataType)
+ commonType match {
+ case Some(commonType) =>
+ (castMatchOperand(leftOperand, commonType),
castMatchOperand(rightOperand, commonType))
+ case None => (leftOperand, rightOperand)
}
}
- private def castMatchOperand(operand: Expression, targetType: DataType):
Expression =
- if (operand.dataType == targetType) operand else Cast(operand, targetType)
+ /**
+ * Returns right sort types with mixed NTZ leaves converted to LTZ at their
own precision.
+ * NTZ-to-LTZ conversion can change order across DST gaps; precision
widening preserves it.
+ */
+ private def rightTimestampSortType(
+ leftType: DataType,
+ rightType: DataType): DataType = (leftType, rightType) match {
+ case (left: StructType, right: StructType) if left.length == right.length
=>
+ StructType(left.fields.zip(right.fields).map { case (l, r) =>
+ r.copy(dataType = rightTimestampSortType(l.dataType, r.dataType))
+ })
+ case (left: ArrayType, right: ArrayType) =>
+ right.copy(elementType = rightTimestampSortType(left.elementType,
right.elementType))
+ case (left, TimestampNTZType) if TimestampFamily.isLtz(left) =>
+ TimestampType
+ case (left, right: TimestampNTZNanosType) if TimestampFamily.isLtz(left) =>
+ TimestampLTZNanosType(right.precision)
+ case _ => rightType
+ }
+
+ private def castMatchOperand(operand: Expression, targetType: DataType):
Expression = {
+ (operand, targetType) match {
+ case _ if operand.dataType == targetType => operand
+ case (tuple: CreateNamedStruct, struct: StructType) =>
+ val values = tuple.valExprs.zip(struct.fields).map { case (value,
field) =>
+ castMatchOperand(value, field.dataType)
+ }
+ CreateNamedStruct(tuple.nameExprs.zip(values).flatMap { case (name,
value) =>
+ Seq(name, value)
+ })
+ case _ => Cast(operand, targetType)
+ }
+ }
/**
* Shared MATCH_CONDITION operand type rules used by analysis validation and
by expression
@@ -2807,13 +2839,13 @@ object AsOfJoin {
}
}
- /** True when either operand is a STRUCT or ARRAY, left uncoerced under
the strict rule. */
- private[catalyst] def isCompositeOperand(
+ /** True when either operand is a STRUCT or ARRAY. */
+ private def isCompositeOperand(
leftType: DataType,
rightType: DataType): Boolean =
Seq(leftType, rightType).exists(t => t.isInstanceOf[StructType] ||
t.isInstanceOf[ArrayType])
- /** True when exactly one operand is a string, the only pair that needs
comparison coercion. */
+ /** True when exactly one operand is a string. */
private def isExactlyOneStringPair(leftType: DataType, rightType:
DataType): Boolean =
leftType.isInstanceOf[StringType] != rightType.isInstanceOf[StringType]
@@ -2881,8 +2913,7 @@ object AsOfJoin {
/**
* 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).
+ * right-side buffer is ordered consistently with the MATCH_CONDITION
comparison.
*
* 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
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 1a32001e6932..868349d36112 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,10 +18,113 @@
package org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
CreateNamedStruct, Literal}
+import org.apache.spark.sql.catalyst.plans.GreaterThanOrEqualOp
import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin.MatchConditionTypes
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
-class AsOfJoinMatchConditionTypesSuite extends SparkFunSuite {
+class AsOfJoinMatchConditionTypesSuite extends SparkFunSuite with
SQLConfHelper {
+
+ test("mixed timestamp sort keys preserve precision and convert NTZ only when
needed") {
+ val keyTypes: Seq[DataType => DataType] = Seq(
+ identity,
+ dt => StructType(Seq(StructField("ts", dt))),
+ dt => ArrayType(dt),
+ dt => StructType(Seq(
+ StructField("items", ArrayType(StructType(Seq(StructField("ts",
dt))))))))
+ for {
+ (timeZone, convertNtz) <- Seq(
+ ("UTC", false), ("+05:30", false), ("America/Los_Angeles", true))
+ ltzType <- Seq(TimestampType, TimestampLTZNanosType(9))
+ (ntzType, convertedType) <- Seq(
+ (TimestampNTZType, TimestampType), (TimestampNTZNanosType(7),
TimestampLTZNanosType(7)))
+ (leftType, rightType, rightTarget) <- Seq(
+ (ltzType, ntzType, convertedType), (ntzType, ltzType, ltzType))
+ keyType <- keyTypes
+ } {
+ val leftOperand = AttributeReference("l", keyType(leftType))()
+ val rightOperand = AttributeReference("r", keyType(rightType))()
+ val (condition, _, leftKeys, rightKeys) =
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> timeZone) {
+ AsOfJoin.materializeMatchComparison(
+ leftOperand = leftOperand,
+ rightOperand = rightOperand,
+ normalizedOp = GreaterThanOrEqualOp
+ )
+ }
+ assert(condition.children == Seq(leftOperand, rightOperand))
+ assert(leftKeys == Seq(leftOperand))
+ assert(rightKeys.map(_.dataType) == Seq(keyType(if (convertNtz)
rightTarget else rightType)))
+ if (!convertNtz || TimestampFamily.isLtz(rightType)) {
+ assert(rightKeys == Seq(rightOperand))
+ }
+ }
+ }
+
+ test("mixed timestamp tuples keep separate sort keys") {
+ val metadata = new MetadataBuilder().putString("comment", "timestamp
key").build()
+ for {
+ (leftType, rightType) <- Seq(
+ (TimestampType, TimestampNTZType), (TimestampNTZType, TimestampType))
+ } {
+ val leftTimestamp = AttributeReference(
+ name = "ts", dataType = leftType, metadata = metadata)()
+ val rightTimestamp = AttributeReference(
+ name = "ts", dataType = rightType, metadata = metadata)()
+ val leftNumber = AttributeReference("n", IntegerType)()
+ val rightNumber = AttributeReference("n", LongType)()
+ val (_, _, leftKeys, rightKeys) =
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key ->
"America/Los_Angeles") {
+ AsOfJoin.materializeMatchComparison(
+ leftOperand = CreateNamedStruct(Seq(
+ Literal("ts"), leftTimestamp, Literal("n"), leftNumber)),
+ rightOperand = CreateNamedStruct(Seq(
+ Literal("ts"), rightTimestamp, Literal("n"), rightNumber)),
+ normalizedOp = GreaterThanOrEqualOp
+ )
+ }
+ assert(leftKeys.length == 2)
+ assert(rightKeys.length == 2)
+ assert(leftKeys.head == leftTimestamp)
+ assert(rightKeys.head.dataType == TimestampType)
+ assert(leftKeys(1) == leftNumber)
+ assert(rightKeys(1) == rightNumber)
+ }
+ }
+
+ test("timestamp coercion preserves other composite field types and
properties") {
+ val timestampField = StructField(
+ name = "ts",
+ dataType = TimestampNTZType,
+ nullable = false,
+ metadata = new MetadataBuilder().putString("comment", "timestamp
key").build()
+ )
+ val rightFields = Seq(
+ timestampField,
+ StructField(name = "d", dataType = DecimalType(38, 30), nullable =
false),
+ StructField(name = "n", dataType = IntegerType, nullable = false))
+ val leftFields = Seq(
+ StructField(name = "ts", dataType = TimestampType),
+ StructField(name = "d", dataType = DecimalType(38, 30)),
+ StructField(name = "n", dataType = LongType))
+ val leftType = ArrayType(StructType(leftFields), containsNull = true)
+ val rightType = ArrayType(StructType(rightFields), containsNull = false)
+ val expectedRight = ArrayType(
+ StructType(timestampField.copy(dataType = TimestampType) +:
rightFields.tail),
+ containsNull = false)
+ val (_, _, leftKeys, rightKeys) =
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles")
{
+ AsOfJoin.materializeMatchComparison(
+ leftOperand = AttributeReference("l", leftType)(),
+ rightOperand = AttributeReference("r", rightType)(),
+ normalizedOp = GreaterThanOrEqualOp
+ )
+ }
+ assert(leftKeys.map(_.dataType) == Seq(leftType))
+ assert(rightKeys.map(_.dataType) == Seq(expectedRight))
+ }
test("scalar types widen via TypeCoercion") {
assert(MatchConditionTypes.areOperandsCompatible(IntegerType, LongType))
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 ff68c04c7ebe..90369a62efdb 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
@@ -319,17 +319,70 @@ Project [d#x, ts#x AS matched_ts#x]
-- !query
-SELECT t.ts, r.ts_ntz AS matched_ts_ntz
-FROM VALUES (TIMESTAMP '2026-06-29 10:00:00') AS t(ts) ASOF JOIN
- VALUES (TIMESTAMP_NTZ '2026-06-29 09:00:00') AS r(ts_ntz)
- MATCH_CONDITION (t.ts >= r.ts_ntz)
+SET spark.sql.session.timeZone=America/Los_Angeles
-- !query analysis
-Project [ts#x, ts_ntz#x AS matched_ts_ntz#x]
-+- AsOfJoin (ts#x >= cast(ts_ntz#x as timestamp)), Inner
- :- SubqueryAlias t
- : +- LocalRelation [ts#x]
+SetCommand (spark.sql.session.timeZone,Some(America/Los_Angeles))
+
+
+-- !query
+SELECT l.id, r.id
+FROM VALUES
+ (TIMESTAMP_LTZ '2024-03-10 10:05:00Z', 1),
+ (TIMESTAMP_LTZ '2024-03-10 10:55:00Z', 2) AS l(ts, id)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:00:00', 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3) AS r(ts, id)
+MATCH_CONDITION (l.ts <= r.ts)
+-- !query analysis
+Project [id#x, id#x]
++- AsOfJoin (ts#x <= cast(ts#x as timestamp)), LeftOuter
+ :- SubqueryAlias l
+ : +- LocalRelation [ts#x, id#x]
+- SubqueryAlias r
- +- LocalRelation [ts_ntz#x]
+ +- LocalRelation [ts#x, id#x]
+
+
+-- !query
+SELECT l.id, r.id
+FROM VALUES
+ (TIMESTAMP_LTZ '2024-03-10 10:05:00Z', 1),
+ (TIMESTAMP_LTZ '2024-03-10 10:55:00Z', 2) AS l(ts, id)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:00:00', 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3) AS r(ts, id)
+MATCH_CONDITION (l.ts >= r.ts)
+-- !query analysis
+Project [id#x, id#x]
++- AsOfJoin (ts#x >= cast(ts#x as timestamp)), LeftOuter
+ :- SubqueryAlias l
+ : +- LocalRelation [ts#x, id#x]
+ +- SubqueryAlias r
+ +- LocalRelation [ts#x, id#x]
+
+
+-- !query
+SELECT r.id
+FROM VALUES (TIMESTAMP_LTZ '2024-03-10 10:15:00Z', 2L) AS l(ts, n)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1, 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 1, 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3, 3) AS r(ts, n, id)
+MATCH_CONDITION ((l.ts, l.n) <= (r.ts, r.n))
+-- !query analysis
+Project [id#x]
++- AsOfJoin (named_struct(ts, ts#x, n, n#xL) <= cast(named_struct(ts, ts#x, n,
n#x) as struct<ts:timestamp,n:bigint>)), LeftOuter
+ :- SubqueryAlias l
+ : +- LocalRelation [ts#x, n#xL]
+ +- SubqueryAlias r
+ +- LocalRelation [ts#x, n#x, id#x]
+
+
+-- !query
+RESET spark.sql.session.timeZone
+-- !query analysis
+ResetCommand spark.sql.session.timeZone
-- !query
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 87c4a8b09158..6116b77139e0 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
@@ -142,11 +142,40 @@ FROM VALUES (DATE '2026-06-29') AS t(d) ASOF JOIN
VALUES (TIMESTAMP '2026-06-29 10:00:00'), (TIMESTAMP '2026-06-28
08:00:00') AS r(ts)
MATCH_CONDITION (t.d >= r.ts);
--- FVT-ASOF-4-016: coercion TIMESTAMP vs TIMESTAMP_NTZ
-SELECT t.ts, r.ts_ntz AS matched_ts_ntz
-FROM VALUES (TIMESTAMP '2026-06-29 10:00:00') AS t(ts) ASOF JOIN
- VALUES (TIMESTAMP_NTZ '2026-06-29 09:00:00') AS r(ts_ntz)
- MATCH_CONDITION (t.ts >= r.ts_ntz);
+-- FVT-ASOF-4-016: coercion TIMESTAMP vs TIMESTAMP_NTZ across a DST gap
+SET spark.sql.session.timeZone=America/Los_Angeles;
+
+-- The DST gap maps 02:50 to 10:50Z, after 03:00 (10:00Z) and 03:15 (10:15Z).
+SELECT l.id, r.id
+FROM VALUES
+ (TIMESTAMP_LTZ '2024-03-10 10:05:00Z', 1),
+ (TIMESTAMP_LTZ '2024-03-10 10:55:00Z', 2) AS l(ts, id)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:00:00', 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3) AS r(ts, id)
+MATCH_CONDITION (l.ts <= r.ts);
+
+SELECT l.id, r.id
+FROM VALUES
+ (TIMESTAMP_LTZ '2024-03-10 10:05:00Z', 1),
+ (TIMESTAMP_LTZ '2024-03-10 10:55:00Z', 2) AS l(ts, id)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:00:00', 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3) AS r(ts, id)
+MATCH_CONDITION (l.ts >= r.ts);
+
+-- Timestamp coercion also applies within a tuple that widens INT to BIGINT.
+SELECT r.id
+FROM VALUES (TIMESTAMP_LTZ '2024-03-10 10:15:00Z', 2L) AS l(ts, n)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1, 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 1, 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3, 3) AS r(ts, n, id)
+MATCH_CONDITION ((l.ts, l.n) <= (r.ts, r.n));
+
+RESET spark.sql.session.timeZone;
-- FVT-ASOF-4-016a: coercion DATE vs STRING (SPARK-59527), coerced like the >=
operator
SELECT t.d, r.s AS matched_s
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 4b3aa32582ba..f5d696fbf867 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
@@ -250,14 +250,67 @@ struct<d:date,matched_ts:timestamp>
-- !query
-SELECT t.ts, r.ts_ntz AS matched_ts_ntz
-FROM VALUES (TIMESTAMP '2026-06-29 10:00:00') AS t(ts) ASOF JOIN
- VALUES (TIMESTAMP_NTZ '2026-06-29 09:00:00') AS r(ts_ntz)
- MATCH_CONDITION (t.ts >= r.ts_ntz)
+SET spark.sql.session.timeZone=America/Los_Angeles
-- !query schema
-struct<ts:timestamp,matched_ts_ntz:timestamp_ntz>
+struct<key:string,value:string>
-- !query output
-2026-06-29 10:00:00 2026-06-29 09:00:00
+spark.sql.session.timeZone America/Los_Angeles
+
+
+-- !query
+SELECT l.id, r.id
+FROM VALUES
+ (TIMESTAMP_LTZ '2024-03-10 10:05:00Z', 1),
+ (TIMESTAMP_LTZ '2024-03-10 10:55:00Z', 2) AS l(ts, id)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:00:00', 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3) AS r(ts, id)
+MATCH_CONDITION (l.ts <= r.ts)
+-- !query schema
+struct<id:int,id:int>
+-- !query output
+1 3
+2 NULL
+
+
+-- !query
+SELECT l.id, r.id
+FROM VALUES
+ (TIMESTAMP_LTZ '2024-03-10 10:05:00Z', 1),
+ (TIMESTAMP_LTZ '2024-03-10 10:55:00Z', 2) AS l(ts, id)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:00:00', 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3) AS r(ts, id)
+MATCH_CONDITION (l.ts >= r.ts)
+-- !query schema
+struct<id:int,id:int>
+-- !query output
+1 2
+2 1
+
+
+-- !query
+SELECT r.id
+FROM VALUES (TIMESTAMP_LTZ '2024-03-10 10:15:00Z', 2L) AS l(ts, n)
+LEFT ASOF JOIN VALUES
+ (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1, 1),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 1, 2),
+ (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3, 3) AS r(ts, n, id)
+MATCH_CONDITION ((l.ts, l.n) <= (r.ts, r.n))
+-- !query schema
+struct<id:int>
+-- !query output
+3
+
+
+-- !query
+RESET spark.sql.session.timeZone
+-- !query schema
+struct<>
+-- !query output
+
-- !query
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 ba68e9b7cde6..f927388f659a 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
@@ -19,7 +19,9 @@ package org.apache.spark.sql
import java.sql.{Date, Timestamp}
+import org.apache.spark.sql.catalyst.expressions.Cast
import org.apache.spark.sql.catalyst.plans.{AsOfJoinDirection, Backward,
Forward}
+import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.joins.SortMergeAsOfJoinExec
@@ -136,11 +138,12 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
for {
singlePass <- Seq(false, true)
sortMergeEnabled <- Seq(false, true)
- } yield (singlePass, sortMergeEnabled)
- ) { case (singlePass, sortMergeEnabled) =>
+ timeZone <- Seq("UTC", "+05:30")
+ } yield (singlePass, sortMergeEnabled, timeZone)
+ ) { case (singlePass, sortMergeEnabled, timeZone) =>
withSQLConf(
SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> sortMergeEnabled.toString,
- SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> timeZone,
SQLConf.TIMESTAMP_TYPE.key -> "TIMESTAMP_LTZ",
SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key ->
singlePass.toString,
SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED_TENTATIVELY.key ->
"false",
@@ -159,11 +162,157 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
|""".stripMargin
)
assertUsesSortMergeAsOfJoin(query)
+ val join = query.queryExecution.analyzed.collectFirst { case j: AsOfJoin
=> j }.get
+ assert(join.leftSortExprs == Seq(join.left.output.head))
+ assert(join.rightSortExprs == Seq(join.right.output.head))
// Single-pass analysis does not support the deserializer used by
checkToRDD.
QueryTest.checkAnswer(df = query, expectedAnswer = Seq(Row(2)),
checkToRDD = false)
}
}
+ gridTest("SQL ASOF JOIN sorts mixed timestamp keys across DST gaps")(
+ Seq(false, true)
+ ) { singlePass =>
+ withSQLConf(
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles",
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key ->
singlePass.toString,
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED_TENTATIVELY.key ->
"false",
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key ->
"false",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLE_ASOF_JOIN_RESOLUTION.key
-> "true"
+ ) {
+ val ltzRows =
+ """(TIMESTAMP_LTZ '2024-03-10 10:05:00Z', 1),
+ |(TIMESTAMP_LTZ '2024-03-10 10:55:00Z', 2)""".stripMargin
+ // 02:50 becomes 10:50Z, after both 03:00 (10:00Z) and 03:15 (10:15Z).
+ val ntzRows =
+ """(TIMESTAMP_NTZ '2024-03-10 02:50:00', 1),
+ |(TIMESTAMP_NTZ '2024-03-10 03:00:00', 2),
+ |(TIMESTAMP_NTZ '2024-03-10 03:15:00', 3)""".stripMargin
+ val keyExpressions = Seq(
+ "scalar" -> "ts",
+ "tuple" -> "ts",
+ "struct" -> "named_struct('ts', ts, 'n', 0)",
+ "array" -> "array(ts)",
+ "nested" -> "named_struct('items', array(named_struct('ts', ts)))")
+ for {
+ (shape, key) <- keyExpressions
+ // Top-level ARRAY operands require lambda resolution in the
fixed-point analyzer.
+ if !singlePass || shape != "array"
+ ltzOnLeft <- Seq(false, true)
+ op <- Seq(">=", "<=")
+ } {
+ withClue(s"shape=$shape, ltzOnLeft=$ltzOnLeft, op=$op: ") {
+ val (leftRows, rightRows) = if (ltzOnLeft) (ltzRows, ntzRows) else
(ntzRows, ltzRows)
+ val leftKey = if (shape == "tuple") "(l.k, 0)" else "l.k"
+ val rightKey = if (shape == "tuple") "(r.k, 0)" else "r.k"
+ val query = sql(
+ s"""SELECT l.id, r.id
+ |FROM (SELECT $key AS k, id FROM VALUES $leftRows AS v(ts, id))
l
+ |LEFT ASOF JOIN
+ | (SELECT $key AS k, id FROM VALUES $rightRows AS v(ts, id)) r
+ |MATCH_CONDITION ($leftKey $op $rightKey)
+ |""".stripMargin
+ )
+ val expected = (ltzOnLeft, op == ">=") match {
+ case (true, true) => Seq(Row(1, 2), Row(2, 1))
+ case (true, false) => Seq(Row(1, 3), Row(2, null))
+ case (false, true) => Seq(Row(1, 1), Row(2, null), Row(3, 1))
+ case (false, false) => Seq(Row(1, 2), Row(2, 1), Row(3, 2))
+ }
+ assertUsesSortMergeAsOfJoin(query)
+ QueryTest.checkAnswer(df = query, expectedAnswer = expected,
checkToRDD = false)
+ }
+ }
+
+ for (op <- Seq(">=", "<=")) {
+ for ((ltzType, ntzType, leftFraction, beforeFraction, afterFraction)
<- Seq(
+ ("TIMESTAMP_LTZ(9)", "TIMESTAMP_NTZ(7)", "000000150", "0000001",
"0000002"),
+ ("TIMESTAMP_LTZ", "TIMESTAMP_NTZ(7)", "000001", "0000009",
"0000011"),
+ ("TIMESTAMP_LTZ(9)", "TIMESTAMP_NTZ", "000000150", "000000",
"000001"))) withClue(
+ s"ltzType=$ltzType, ntzType=$ntzType, op=$op: ") {
+ val query = sql(
+ s"""SELECT r.id
+ |FROM VALUES (CAST('2024-03-10 10:15:00.${leftFraction}Z' AS
$ltzType)) l(ts)
+ |LEFT ASOF JOIN VALUES
+ | (CAST('2024-03-10 02:50:00.$beforeFraction' AS $ntzType), 1),
+ | (CAST('2024-03-10 03:15:00.$beforeFraction' AS $ntzType), 2),
+ | (CAST('2024-03-10 03:15:00.$afterFraction' AS $ntzType), 3)
r(ts, id)
+ |MATCH_CONDITION (l.ts $op r.ts)
+ |""".stripMargin
+ )
+ assertUsesSortMergeAsOfJoin(query)
+ QueryTest.checkAnswer(
+ df = query,
+ expectedAnswer = Seq(Row(if (op == ">=") 2 else 3)),
+ checkToRDD = false
+ )
+ }
+
+ val structKey = "named_struct('ts', ts, 'n', n)"
+ val compositeKeys = Seq(
+ "struct" -> structKey,
+ "array" -> s"array($structKey)",
+ "nested" -> s"named_struct('items', array($structKey))")
+ for {
+ (shape, key) <- compositeKeys
+ if !singlePass || shape != "array"
+ } withClue(s"mixed timestamp and integral fields, shape=$shape,
op=$op: ") {
+ val query = sql(
+ s"""SELECT r.id
+ |FROM (SELECT $key AS k
+ | FROM VALUES (TIMESTAMP_LTZ '2024-03-10 10:15:00Z', 2L) v(ts,
n)) l
+ |LEFT ASOF JOIN (SELECT $key AS k, id FROM VALUES
+ | (TIMESTAMP_NTZ '2024-03-10 02:50:00', 1, 1),
+ | (TIMESTAMP_NTZ '2024-03-10 03:15:00', 1, 2),
+ | (TIMESTAMP_NTZ '2024-03-10 03:15:00', 3, 3) v(ts, n, id)) r
+ |MATCH_CONDITION (l.k $op r.k)
+ |""".stripMargin
+ )
+ assertUsesSortMergeAsOfJoin(query)
+ val join = query.queryExecution.analyzed.collectFirst { case j:
AsOfJoin => j }.get
+ val casts = join.asOfCondition.collect { case c: Cast => c }
+ assert(casts.size == 1, join.asOfCondition.treeString)
+ QueryTest.checkAnswer(
+ df = query,
+ expectedAnswer = Seq(Row(if (op == ">=") 2 else 3)),
+ checkToRDD = false
+ )
+ }
+ }
+
+ def nullStructRows(timestampType: String): Seq[String] = Seq(
+ s"(CAST(NULL AS STRUCT<ts: $timestampType>), 1)",
+ s"(named_struct('ts', CAST(NULL AS $timestampType)), 2)")
+
+ for {
+ (leftType, rightType) <- Seq(
+ ("TIMESTAMP_LTZ", "TIMESTAMP_NTZ"), ("TIMESTAMP_NTZ",
"TIMESTAMP_LTZ"))
+ op <- Seq(">=", "<=")
+ reverseRightRows <- Seq(false, true)
+ } {
+ withClue(s"leftType=$leftType, op=$op,
reverseRightRows=$reverseRightRows: ") {
+ val leftRows = nullStructRows(leftType).mkString(", ")
+ val rightRows = nullStructRows(rightType)
+ val orderedRightRows = if (reverseRightRows) rightRows.reverse else
rightRows
+ val query = sql(
+ s"""SELECT l.id, r.id
+ |FROM VALUES $leftRows AS l(k, id)
+ |LEFT ASOF JOIN VALUES ${orderedRightRows.mkString(", ")} AS
r(k, id)
+ |MATCH_CONDITION (l.k $op r.k)
+ |""".stripMargin
+ )
+ assertUsesSortMergeAsOfJoin(query)
+ QueryTest.checkAnswer(
+ df = query,
+ expectedAnswer = Seq(Row(1, null), Row(2, 2)),
+ checkToRDD = false
+ )
+ }
+ }
+ }
+ }
+
test("LEFT ASOF JOIN preserves unmatched left rows") {
setupTradeQuoteViews()
checkSortMergeAsOf(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
