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 c5e2a39aa312 [SPARK-58761][SQL][TEST] Add dedicated e2e test coverage
for DISTINCT over nanosecond-precision timestamp columns
c5e2a39aa312 is described below
commit c5e2a39aa3121415398dbf367ee632077d79aca0
Author: Stevo Mitric <[email protected]>
AuthorDate: Fri Aug 14 07:19:49 2026 +0000
[SPARK-58761][SQL][TEST] Add dedicated e2e test coverage for DISTINCT over
nanosecond-precision timestamp columns
### What changes were proposed in this pull request?
Add end-to-end regression-lock coverage for `DISTINCT` over the
nanosecond-precision timestamp types `TIMESTAMP_NTZ(p)` / `TIMESTAMP_LTZ(p)`
(`p` in `[7, 9]`).
New suite `TimestampNanosDistinctSuiteBase` (with
`TimestampNanosDistinctAnsiOnSuite` / `TimestampNanosDistinctAnsiOffSuite`
subclasses), mirroring `TimestampNanosJoinSuiteBase`. For NTZ and LTZ, across
`p` in `[7, 9]` and both whole-stage-codegen modes, it asserts:
- DISTINCT removes exact duplicates but keeps two values that share
`epochMicros` and differ only within the microsecond, and keeps exactly one
NULL (unlike an equi-join, where NULL never matches NULL);
- a `UNION` of two different nanos precisions widens the column to the
higher precision (`findWiderDateTimeType`) and preserves the distinction.
Also extends the golden files `timestamp-ntz-nanos.sql` /
`timestamp-ltz-nanos.sql` with a DISTINCT case.
### Why are the changes needed?
DISTINCT over a nanosecond-timestamp column already works today -- it rides
on the generic nanos hashing/equality implemented in SPARK-57103 (`Murmur3Hash`
/ `XxHash64` / `HiveHash` over the carrier's `epochMicros: Long` and
`nanosWithinMicro: Short in [0, 999]`) -- but lacked dedicated coverage. These
tests lock the regression: if the nanos hash or equality path is later broken,
sub-microsecond-distinct values collapse and the tests fail loudly. No
production change.
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
`TimestampNanosDistinctAnsiOnSuite` / `TimestampNanosDistinctAnsiOffSuite`
(16 tests, all pass) and the regenerated golden SQL files.
### Was this patch authored or co-authored using generative AI tooling?
Co-Authored-By: Claude Opus 4.8
Closes #57993 from stevomitric/stevomitric/nanos-distinct-tests.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
---
.../analyzer-results/timestamp-ltz-nanos.sql.out | 15 ++
.../analyzer-results/timestamp-ntz-nanos.sql.out | 15 ++
.../sql-tests/inputs/timestamp-ltz-nanos.sql | 9 +
.../sql-tests/inputs/timestamp-ntz-nanos.sql | 9 +
.../sql-tests/results/timestamp-ltz-nanos.sql.out | 15 ++
.../sql-tests/results/timestamp-ntz-nanos.sql.out | 15 ++
.../sql/TimestampNanosDistinctSuiteBase.scala | 183 +++++++++++++++++++++
7 files changed, 261 insertions(+)
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out
index 72b4b63b0a55..93c3ef991b96 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out
@@ -812,6 +812,21 @@ Aggregate [max_by(v#x, k#x) AS max_by(v, k)#x, min_by(v#x,
k#x) AS min_by(v, k)#
+- LocalRelation [v#x, k#x]
+-- !query
+SELECT DISTINCT c FROM VALUES
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC'),
+ (CAST(NULL AS timestamp_ltz(9))) AS t(c)
+ ORDER BY c
+-- !query analysis
+Sort [c#x ASC NULLS FIRST], true
++- Distinct
+ +- Project [c#x]
+ +- SubqueryAlias t
+ +- LocalRelation [c#x]
+
+
-- !query
SELECT unix_nanos(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789 UTC')
-- !query analysis
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out
index a2c519731b7c..02484a99b4b7 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out
@@ -733,6 +733,21 @@ Aggregate [max_by(v#x, k#x) AS max_by(v, k)#x, min_by(v#x,
k#x) AS min_by(v, k)#
+- LocalRelation [v#x, k#x]
+-- !query
+SELECT DISTINCT c FROM VALUES
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999'),
+ (CAST(NULL AS timestamp_ntz(9))) AS t(c)
+ ORDER BY c
+-- !query analysis
+Sort [c#x ASC NULLS FIRST], true
++- Distinct
+ +- Project [c#x]
+ +- SubqueryAlias t
+ +- LocalRelation [c#x]
+
+
-- !query
SELECT unix_nanos(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789')
-- !query analysis
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql
b/sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql
index 859768a514b4..9a172a2fe2ae 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql
@@ -251,6 +251,15 @@ SELECT max_by(v, k), min_by(v, k) FROM VALUES
(TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC', 3),
(TIMESTAMP_LTZ '2020-01-01 00:00:00.000000500 UTC', 2),
(TIMESTAMP_LTZ '2020-01-01 00:00:00.000000007 UTC', CAST(NULL AS INT)) AS
t(v, k);
+-- DISTINCT over a nanosecond column: exact duplicates are removed, two values
sharing epochMicros
+-- but differing within the microsecond are both kept, and NULL survives as a
single row. Three
+-- rows: .000000001, .000000999, NULL. Values render in the session time zone.
+SELECT DISTINCT c FROM VALUES
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC'),
+ (CAST(NULL AS timestamp_ltz(9))) AS t(c)
+ ORDER BY c;
-- SPARK-57527: unix_nanos over nanosecond-precision values returns
DECIMAL(21, 0) nanoseconds since
-- the epoch. The explicit-zone literals below fix the instant directly,
independent of the session
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql
b/sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql
index cc6638e42494..2a571619cf55 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql
@@ -221,6 +221,15 @@ SELECT max_by(v, k), min_by(v, k) FROM VALUES
(TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999', 3),
(TIMESTAMP_NTZ '2020-01-01 00:00:00.000000500', 2),
(TIMESTAMP_NTZ '2020-01-01 00:00:00.000000007', CAST(NULL AS INT)) AS t(v,
k);
+-- DISTINCT over a nanosecond column: exact duplicates are removed, two values
sharing epochMicros
+-- but differing within the microsecond are both kept, and NULL survives as a
single row. Three
+-- rows: .000000001, .000000999, NULL.
+SELECT DISTINCT c FROM VALUES
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999'),
+ (CAST(NULL AS timestamp_ntz(9))) AS t(c)
+ ORDER BY c;
-- SPARK-57527: unix_nanos over nanosecond-precision values returns
DECIMAL(21, 0) nanoseconds since
-- the epoch; NTZ applies no zone shift, so the wall-clock value is read as
the epoch instant. The
diff --git
a/sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out
b/sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out
index 385e8cecaff5..4c036133a551 100644
--- a/sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out
@@ -911,6 +911,21 @@ struct<max_by(v, k):timestamp_ltz(9),min_by(v,
k):timestamp_ltz(9)>
2019-12-31 16:00:00.000000999 2019-12-31 16:00:00.000000001
+-- !query
+SELECT DISTINCT c FROM VALUES
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+ (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC'),
+ (CAST(NULL AS timestamp_ltz(9))) AS t(c)
+ ORDER BY c
+-- !query schema
+struct<c:timestamp_ltz(9)>
+-- !query output
+NULL
+2019-12-31 16:00:00.000000001
+2019-12-31 16:00:00.000000999
+
+
-- !query
SELECT unix_nanos(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789 UTC')
-- !query schema
diff --git
a/sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out
b/sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out
index 40dc2ddc66c0..e10a25456268 100644
--- a/sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out
@@ -821,6 +821,21 @@ struct<max_by(v, k):timestamp_ntz(9),min_by(v,
k):timestamp_ntz(9)>
2020-01-01 00:00:00.000000999 2020-01-01 00:00:00.000000001
+-- !query
+SELECT DISTINCT c FROM VALUES
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+ (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999'),
+ (CAST(NULL AS timestamp_ntz(9))) AS t(c)
+ ORDER BY c
+-- !query schema
+struct<c:timestamp_ntz(9)>
+-- !query output
+NULL
+2020-01-01 00:00:00.000000001
+2020-01-01 00:00:00.000000999
+
+
-- !query
SELECT unix_nanos(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789')
-- !query schema
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/TimestampNanosDistinctSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/TimestampNanosDistinctSuiteBase.scala
new file mode 100644
index 000000000000..5574344520ba
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/TimestampNanosDistinctSuiteBase.scala
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import java.time.{Instant, LocalDateTime}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+/**
+ * End-to-end `DISTINCT` correctness tests over the nanosecond-precision
timestamp types
+ * `TIMESTAMP_NTZ(p)` / `TIMESTAMP_LTZ(p)` (`p` in `[7, 9]`), part of the
nanosecond timestamp
+ * preview (SPARK-56822). Set-distinctness over these types executes today
with no production
+ * change -- it rides on the nanos hashing and equality the types already
implement (SPARK-57103
+ * extended `Murmur3Hash` / `XxHash64` / `HiveHash` to hash the carrier's two
fields,
+ * `epochMicros: Long` and `nanosWithinMicro: Short in [0, 999]`). There is no
coverage organized
+ * around distinctness by column type, so this per-type suite adds it and
locks the regression.
+ *
+ * Headline risk: SUB-MICROSECOND distinctness, mirroring the join suite. The
carrier is
+ * `TimestampNanosVal = (epochMicros: Long, nanosWithinMicro: Short in [0,
999])`. Every value in
+ * these tables shares the SAME epochMicros (1577836800000000, =
2020-01-01T00:00:00Z) and differs
+ * ONLY in nanosWithinMicro. So the micro-level path alone cannot tell them
apart -- correct
+ * distinctness MUST be driven by the full nanos value:
+ * - two values equal in epochMicros but DIFFERENT in nanosWithinMicro must
NOT be deduplicated,
+ * - exact duplicates (equal incl. the sub-microsecond remainder) MUST be
deduplicated,
+ * - NULL, unlike an equi-join, IS a single distinct value -- `DISTINCT`
keeps exactly one NULL.
+ * If the sub-microsecond remainder were ignored, the two distinct
sub-microsecond values here
+ * would fold into one, dropping the distinct-row count from 3 to 2 and
failing these tests loudly.
+ *
+ * Precision-safety: all sub-microsecond remainders are multiples of 100ns
(100 / 900), which are
+ * exact at every p in [7, 9] (`createDataFrame` floors nanosWithinMicro to
(n/100)*100 at p=7 and
+ * (n/10)*10 at p=8). So the SAME inputs and the SAME expected results are
valid verbatim at all
+ * three precisions, and the two distinct remainders never collide even at the
coarsest p=7.
+ *
+ * Each test runs under whole-stage codegen on and off, so the same
sub-microsecond distinctness is
+ * proven on the nanos hash path in both codegen modes, for NTZ and LTZ. The
mixed-precision test
+ * additionally pins that a `UNION` of two different nanos precisions widens
the column to the
+ * higher precision (`findWiderDateTimeType`) and preserves the distinction.
+ *
+ * The nanosecond timestamp types are gated behind a preview flag enabled by
default under tests
+ * (`Utils.isTesting`), so it is not set here. The session time zone is fixed
so the
+ * `TIMESTAMP_LTZ` (`Instant`) values are deterministic. The two subclasses
run every test with
+ * ANSI mode on and off.
+ */
+abstract class TimestampNanosDistinctSuiteBase extends QueryTest with
SharedSparkSession {
+
+ override def sparkConf: SparkConf = super.sparkConf
+ .set(SQLConf.SESSION_LOCAL_TIMEZONE.key, "America/Los_Angeles")
+
+ // Whole-stage codegen on (CODEGEN_ONLY) vs off (NO_CODEGEN). The
hash-aggregate exec backing
+ // DISTINCT is the same in both modes; only the WholeStageCodegenExec
wrapper differs. Mirrors the
+ // join and functions suites.
+ protected val codegenModes: Seq[Seq[(String, String)]] = Seq(
+ Seq(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.CODEGEN_FACTORY_MODE.key -> "CODEGEN_ONLY"),
+ Seq(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
+ SQLConf.CODEGEN_FACTORY_MODE.key -> "NO_CODEGEN"))
+
+ private def cgLabel(cgConf: Seq[(String, String)]): String =
+ if (cgConf.contains(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true"))
"codegen on"
+ else "codegen off"
+
+ // ---- relation builders: single nanos column "k" ----
+ //
+ // Four rows, all sharing epochMicros (2020-01-01T00:00:00Z) and differing
only within the
+ // microsecond:
+ // - two rows with the 100ns value (an exact duplicate): collapse to one
distinct row,
+ // - one row with the 900ns value: same microsecond, distinct
sub-microsecond remainder,
+ // - one NULL row: a single distinct NULL.
+ // Expected three distinct values: 100ns, 900ns, NULL.
+
+ private def ntzDF(p: Int): DataFrame =
+ spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(
+ Row(LocalDateTime.parse("2020-01-01T00:00:00.000000100")),
+ Row(LocalDateTime.parse("2020-01-01T00:00:00.000000100")),
+ Row(LocalDateTime.parse("2020-01-01T00:00:00.000000900")),
+ Row(null))),
+ new StructType().add("k", TimestampNTZNanosType(p)))
+
+ private def ltzDF(p: Int): DataFrame =
+ spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(
+ Row(Instant.parse("2020-01-01T00:00:00.000000100Z")),
+ Row(Instant.parse("2020-01-01T00:00:00.000000100Z")),
+ Row(Instant.parse("2020-01-01T00:00:00.000000900Z")),
+ Row(null))),
+ new StructType().add("k", TimestampLTZNanosType(p)))
+
+ // The external nanosecond remainder of a collected value, whichever family
it came back as
+ // (NTZ -> LocalDateTime, LTZ -> Instant). Every value here shares
epochMicros, so the remainder
+ // alone identifies it.
+ private def nanoOf(v: Any): Int = v match {
+ case ldt: LocalDateTime => ldt.getNano
+ case i: Instant => i.getNano
+ }
+
+ // Each family as (label, builder(p), widerType(p)). DISTINCT is
type-agnostic, so the family only
+ // picks the data builder and (for the mixed-precision widening assertion)
the expected widened
+ // column type.
+ protected val families: Seq[(String, Int => DataFrame, Int => DataType)] =
Seq(
+ ("NTZ", ntzDF, (p: Int) => TimestampNTZNanosType(p)),
+ ("LTZ", ltzDF, (p: Int) => TimestampLTZNanosType(p)))
+
+ // Mixed-precision UNION pairs; the column widens to max(pl, pr).
+ private val mixedPrecisionPairs: Seq[(Int, Int)] = Seq((7, 9), (7, 8), (8,
9))
+
+ // Asserts that `distinct` holds exactly the three expected values: the
100ns and 900ns remainders
+ // and a single NULL. Checking the actual remainders (not just the row
count) pins that the
+ // surviving rows carry the full sub-microsecond value.
+ private def assertDistinctKeys(distinct: DataFrame, p: Int): Unit = {
+ assert(distinct.count() == 3, s"expected 3 distinct values (100ns, 900ns,
NULL) at p=$p")
+ val rows = distinct.collect()
+ assert(rows.count(_.isNullAt(0)) == 1, s"expected exactly one NULL at
p=$p")
+ val remainders = rows.filterNot(_.isNullAt(0)).map(r =>
nanoOf(r.get(0))).toSet
+ assert(remainders == Set(100, 900),
+ s"expected distinct remainders {100, 900} at p=$p, got $remainders")
+ }
+
+ for {
+ (family, builder, widerType) <- families
+ cgConf <- codegenModes
+ } {
+ //
========================================================================================
+ // DISTINCT over a nanos column: exact duplicates are removed,
sub-microsecond-distinct values
+ // are kept, and NULL survives as exactly one row. Expected three distinct
rows: 100ns, 900ns,
+ // NULL. If the sub-microsecond remainder were ignored, 100ns and 900ns
would collapse to one,
+ // dropping the count to 2 and failing this assertion.
+ //
========================================================================================
+ test(s"$family nanos DISTINCT keeps sub-microsecond-distinct values -
${cgLabel(cgConf)}") {
+ withSQLConf(cgConf: _*) {
+ Seq(7, 8, 9).foreach { p =>
+ assertDistinctKeys(builder(p).select("k").distinct(), p)
+ }
+ }
+ }
+
+ //
========================================================================================
+ // DISTINCT over a mixed-precision column: a UNION of the same data at two
different nanos
+ // precisions widens the column to max(pl, pr) (findWiderDateTimeType).
DISTINCT over the
+ // widened column still separates the two sub-microsecond values and still
collapses to three
+ // rows (the union only adds more exact duplicates; all remainders are
multiples of 100ns,
+ // exact at every p in [7, 9]).
+ //
========================================================================================
+ test(s"$family nanos DISTINCT across mixed precisions widens column -
${cgLabel(cgConf)}") {
+ withSQLConf(cgConf: _*) {
+ mixedPrecisionPairs.foreach { case (pl, pr) =>
+ val union = builder(pl).union(builder(pr))
+ assert(union.schema("k").dataType == widerType(math.max(pl, pr)),
+ s"expected column widened to ${widerType(math.max(pl, pr))} for
($pl, $pr)")
+ assertDistinctKeys(union.distinct(), math.max(pl, pr))
+ }
+ }
+ }
+ }
+}
+
+// Runs the nanosecond timestamp DISTINCT tests with ANSI mode enabled
explicitly.
+class TimestampNanosDistinctAnsiOnSuite extends
TimestampNanosDistinctSuiteBase {
+ override def sparkConf: SparkConf =
super.sparkConf.set(SQLConf.ANSI_ENABLED.key, "true")
+}
+
+// Runs the nanosecond timestamp DISTINCT tests with ANSI mode disabled
explicitly.
+class TimestampNanosDistinctAnsiOffSuite extends
TimestampNanosDistinctSuiteBase {
+ override def sparkConf: SparkConf =
super.sparkConf.set(SQLConf.ANSI_ENABLED.key, "false")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
