Skip to content

Commit 07d9266

Browse files
stevomitricuros-b
authored andcommitted
[SPARK-59442][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. ### What changes were proposed in this pull request? ### Why are the changes needed? ### Does this PR introduce _any_ user-facing change? ### How was this patch tested? ### Was this patch authored or co-authored using generative AI tooling? Closes #58746 from stevomitric/stevomitric/nanos-groupby-null-key. Authored-by: Stevo Mitric <stevomitric2000@gmail.com> Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>
1 parent 87d3841 commit 07d9266

2 files changed

Lines changed: 36 additions & 1 deletion

File tree

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeWriter.java

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,12 @@ public void write(int ordinal, TimestampNanosVal input) {
163163
grow(TimestampNanosRowValues.SIZE_IN_BYTES);
164164
if (input == null) {
165165
BitSetMethods.set(getBuffer(), startingOffset, ordinal);
166+
// Zero the reserved payload so that a null value is byte-identical no matter what stale bytes
167+
// the reused buffer holds. The buffer is not cleared between rows, so without this two null
168+
// keys can carry different bytes and split into separate groups (a nullable nanosecond
169+
// GROUP BY / join key produced several null groups). Mirrors the in-place null-update path
170+
// UnsafeRow#setTimestampNanosPayload, which zeroes the payload the same way.
171+
TimestampNanosRowValues.zeroPayload(getBuffer(), 0, (int) cursor());
166172
} else {
167173
TimestampNanosRowValues.writePayload(
168174
getBuffer(), 0, (int) cursor(), input.epochMicros, input.nanosWithinMicro);

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverterSuite.scala

Lines changed: 30 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.InternalRow
2828
import org.apache.spark.sql.catalyst.util._
2929
import org.apache.spark.sql.types.{IntegerType, LongType, _}
3030
import org.apache.spark.unsafe.array.ByteArrayMethods
31-
import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
31+
import org.apache.spark.unsafe.types.{CalendarInterval, TimestampNanosVal, UTF8String}
3232
import org.apache.spark.util.ArrayImplicits._
3333

3434
class UnsafeRowConverterSuite extends SparkFunSuite with Matchers with ExpressionEvalHelper {
@@ -73,6 +73,35 @@ class UnsafeRowConverterSuite extends SparkFunSuite with Matchers with Expressio
7373
assert(unsafeRow2.getInt(2) === 2)
7474
}
7575

76+
testBothCodegenAndInterpreted(
77+
"null nanosecond timestamp keys are byte-identical regardless of prior rows") {
78+
// The nanosecond timestamp types occupy a 16-byte variable-length payload. The projection
79+
// reuses its output buffer across rows, so a null value must zero that payload; otherwise it
80+
// inherits the previous non-null value's bytes and two null rows compare unequal -- which
81+
// splits a nullable nanosecond GROUP BY / join key into several null groups.
82+
Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9)).foreach { dt =>
83+
val fieldTypes: Array[DataType] = Array(dt)
84+
val converter = UnsafeProjection.create(fieldTypes)
85+
val row = new SpecificInternalRow(fieldTypes.toImmutableArraySeq)
86+
87+
// Dirty the reused buffer with one non-null value, then project a null.
88+
row.update(0, TimestampNanosVal.fromParts(1234567L, 111.toShort))
89+
converter.apply(row)
90+
row.setNullAt(0)
91+
val nullAfterA = converter.apply(row).copy()
92+
93+
// Dirty the buffer with a *different* non-null value, then project a null again.
94+
row.update(0, TimestampNanosVal.fromParts(987654321L, 222.toShort))
95+
converter.apply(row)
96+
row.setNullAt(0)
97+
val nullAfterB = converter.apply(row).copy()
98+
99+
assert(nullAfterA.isNullAt(0) && nullAfterB.isNullAt(0))
100+
assert(nullAfterA == nullAfterB,
101+
s"two null $dt projections must be byte-identical but differed (stale payload)")
102+
}
103+
}
104+
76105
testBothCodegenAndInterpreted("basic conversion with primitive, string and binary types") {
77106
val factory = UnsafeProjection
78107
val fieldTypes: Array[DataType] = Array(LongType, StringType, BinaryType)

0 commit comments

Comments
 (0)