From 903a78a21f836722561d125f0e14d5a0b353da49 Mon Sep 17 00:00:00 2001 From: Stevo Mitric Date: Fri, 11 Sep 2026 16:00:17 +0000 Subject: [PATCH 1/2] [SPARK-XXXXX][SQL] Zero the reserved payload when writing a null nanosecond timestamp to UnsafeRow ### What changes were proposed in this pull request? `UnsafeWriter.write(int, TimestampNanosVal)` stores a nanosecond timestamp as a 16-byte variable-length payload (modeled on `CalendarInterval`) and reserves that space even for a null value so the slot can be updated in place later. On the null branch it set the null bit but left the reserved 16 bytes untouched. The writer's buffer is reused across rows, so a null nanosecond value inherited whatever bytes the previously written row left in that slot. Two logically-equal null rows could therefore produce different `UnsafeRow` byte contents, breaking the invariant that equal rows encode identically -- the invariant that `UnsafeRow` hashing and equality rely on. This is the root cause of a nullable nanosecond-timestamp GROUP BY / join key splitting its NULLs across multiple groups. The fix zeroes the reserved payload on the null branch, mirroring the in-place update path (`UnsafeRow.setTimestampNTZNanos` / `setTimestampLTZNanos`), which already calls `TimestampNanosRowValues.zeroPayload`. ### Why are the changes needed? Correctness: a null nanosecond timestamp must encode canonically so that null grouping/join keys compare and hash identically. Without it, `GROUP BY` (and any key-based operator) over a nullable `TIMESTAMP_NTZ(p)` / `TIMESTAMP_LTZ(p)` column can silently scatter NULL rows across several groups. The array path (`UnsafeArrayWriter`) is unaffected: it writes a null element through `setNull8Bytes`, which zeroes the element's offset-and-size slot (size 0). ### Does this PR introduce any user-facing change? No change in a default configuration -- the nanosecond timestamp types are behind the `spark.sql.timestampNanosTypes.enabled` preview flag (off by default). With the flag enabled, null nanosecond keys now group and compare canonically. ### How was this patch tested? New `UnsafeRowConverterSuite` test asserting that two null nanosecond projections built after different non-null values are byte-identical, in both the interpreted and the codegen paths. It fails without this change and passes with it. Co-authored-by: Isaac --- .../expressions/codegen/UnsafeWriter.java | 5 +++ .../expressions/UnsafeRowConverterSuite.scala | 31 ++++++++++++++++++- 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java index 53f1ea442cd46..ed0620e97b665 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java @@ -163,6 +163,11 @@ public void write(int ordinal, TimestampNanosVal input) { grow(TimestampNanosRowValues.SIZE_IN_BYTES); if (input == null) { BitSetMethods.set(getBuffer(), startingOffset, ordinal); + // Zero the reserved payload so that a null value is byte-identical no matter what stale bytes + // the reused buffer holds. The buffer is not cleared between rows, so without this two null + // keys can carry different bytes and split into separate groups (a nullable nanosecond + // GROUP BY / join key produced several null groups). Mirrors UnsafeRow#setTimestampNanos. + TimestampNanosRowValues.zeroPayload(getBuffer(), 0, (int) cursor()); } else { TimestampNanosRowValues.writePayload( getBuffer(), 0, (int) cursor(), input.epochMicros, input.nanosWithinMicro); diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverterSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverterSuite.scala index 37ef843665fa5..8ffece81a474b 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverterSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverterSuite.scala @@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.util._ import org.apache.spark.sql.types.{IntegerType, LongType, _} import org.apache.spark.unsafe.array.ByteArrayMethods -import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String} +import org.apache.spark.unsafe.types.{CalendarInterval, TimestampNanosVal, UTF8String} import org.apache.spark.util.ArrayImplicits._ class UnsafeRowConverterSuite extends SparkFunSuite with Matchers with ExpressionEvalHelper { @@ -73,6 +73,35 @@ class UnsafeRowConverterSuite extends SparkFunSuite with Matchers with Expressio assert(unsafeRow2.getInt(2) === 2) } + testBothCodegenAndInterpreted( + "null nanosecond timestamp keys are byte-identical regardless of prior rows") { + // The nanosecond timestamp types occupy a 16-byte variable-length payload. The projection + // reuses its output buffer across rows, so a null value must zero that payload; otherwise it + // inherits the previous non-null value's bytes and two null rows compare unequal -- which + // splits a nullable nanosecond GROUP BY / join key into several null groups. + Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9)).foreach { dt => + val fieldTypes: Array[DataType] = Array(dt) + val converter = UnsafeProjection.create(fieldTypes) + val row = new SpecificInternalRow(fieldTypes.toImmutableArraySeq) + + // Dirty the reused buffer with one non-null value, then project a null. + row.update(0, TimestampNanosVal.fromParts(1234567L, 111.toShort)) + converter.apply(row) + row.setNullAt(0) + val nullAfterA = converter.apply(row).copy() + + // Dirty the buffer with a *different* non-null value, then project a null again. + row.update(0, TimestampNanosVal.fromParts(987654321L, 222.toShort)) + converter.apply(row) + row.setNullAt(0) + val nullAfterB = converter.apply(row).copy() + + assert(nullAfterA.isNullAt(0) && nullAfterB.isNullAt(0)) + assert(nullAfterA == nullAfterB, + s"two null $dt projections must be byte-identical but differed (stale payload)") + } + } + testBothCodegenAndInterpreted("basic conversion with primitive, string and binary types") { val factory = UnsafeProjection val fieldTypes: Array[DataType] = Array(LongType, StringType, BinaryType) From 8487af51c87b4d95255d1ebe7b9586eb59674d7c Mon Sep 17 00:00:00 2001 From: Stevo Mitric Date: Fri, 11 Sep 2026 16:52:19 +0000 Subject: [PATCH 2/2] [SPARK-56822][SQL] Address review: correct the mirrored-method reference in the comment The added comment referenced a nonexistent `UnsafeRow#setTimestampNanos`; the method it mirrors is the private `UnsafeRow#setTimestampNanosPayload`, which zeroes the payload on the null-update path. Correct the cross-reference so it is greppable. Co-authored-by: Isaac --- .../spark/sql/catalyst/expressions/codegen/UnsafeWriter.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java index ed0620e97b665..20ec4c2bb0c6c 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java @@ -166,7 +166,8 @@ public void write(int ordinal, TimestampNanosVal input) { // Zero the reserved payload so that a null value is byte-identical no matter what stale bytes // the reused buffer holds. The buffer is not cleared between rows, so without this two null // keys can carry different bytes and split into separate groups (a nullable nanosecond - // GROUP BY / join key produced several null groups). Mirrors UnsafeRow#setTimestampNanos. + // GROUP BY / join key produced several null groups). Mirrors the in-place null-update path + // UnsafeRow#setTimestampNanosPayload, which zeroes the payload the same way. TimestampNanosRowValues.zeroPayload(getBuffer(), 0, (int) cursor()); } else { TimestampNanosRowValues.writePayload(