You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/07/23 14:58:55 UTC
(spark) branch master updated: [SPARK-57828][SQL] Add vectorized Parquet reader support for nanosecond-precision timestamps
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 905aaae710c9 [SPARK-57828][SQL] Add vectorized Parquet reader support for nanosecond-precision timestamps
905aaae710c9 is described below
commit 905aaae710c9d40f5111bf27345f4f9e28f6cdd8
Author: Anupam Yadav <an...@gmail.com>
AuthorDate: Thu Jul 23 22:58:39 2026 +0800
[SPARK-57828][SQL] Add vectorized Parquet reader support for nanosecond-precision timestamps
### What changes were proposed in this pull request?
Adds vectorized Parquet reader support for nanosecond-precision timestamp columns (INT64 `TIMESTAMP(NANOS)`). A new `TimestampNanosVectorUpdater` in `TimestampNanosParquetOps` (routed through the types framework's `getVectorUpdater` hook) decomposes epoch-nanoseconds into the two-child column-vector representation (`epochMicros: Long`, `nanosWithinMicro: Short`), matching `DateTimeUtils.epochNanosToTimestampNanos` and the row-based `newConverter` path exactly (including the dictionary [...]
### Why are the changes needed?
Part of nanosecond-precision timestamp support (SPARK-56822). The vectorized reader is the default fast path; nanosecond-precision timestamp columns should be readable through it - producing identical results to the row-based reader - rather than falling back to row-by-row reads.
### Does this PR introduce _any_ user-facing change?
Yes - nanosecond-precision timestamp columns in Parquet are now read via the vectorized reader (a performance improvement); results are identical to the row-based reader.
### How was this patch tested?
`ParquetTimestampNanosSuite` was upgraded to run under both readers (`withAllParquetReaders`), plus a new vectorized-vs-row-based parity test over edge-case nanosecond values (positive, pre-epoch, exact-second, nulls) at precisions 7/8/9 for both NTZ and LTZ. `TimestampNanosParquetOpsSuite` adds ops-contract tests pinning `getVectorUpdater` (accepts INT64 `TIMESTAMP(NANOS)`; rejects `TIMESTAMP(MICROS)`, raw INT64, and `TIME(NANOS)`). `ParquetIOSuite` adds a dictionary-encoding test fo [...]
### Was this patch authored or co-authored using generative AI tooling?
Authored with assistance by Claude Opus 4.8.
Closes #56942 from yadavay-amzn/SPARK-57828.
Authored-by: Anupam Yadav <an...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../types/ops/TimestampNanosParquetOps.scala | 90 ++++++++-
.../datasources/parquet/ParquetIOSuite.scala | 41 ++++
.../parquet/ParquetTimestampNanosSuite.scala | 221 +++++++++++++--------
.../types/ops/TimestampNanosParquetOpsSuite.scala | 44 ++++
4 files changed, 304 insertions(+), 92 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
index 4b5f01d17374..6eb19f5f34b0 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
+import org.apache.parquet.column.{ColumnDescriptor, Dictionary}
import org.apache.parquet.io.api.{Converter, RecordConsumer}
import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
import org.apache.parquet.schema.LogicalTypeAnnotation.{TimestampLogicalTypeAnnotation, TimeUnit}
@@ -26,7 +27,9 @@ import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater, ParentContainerUpdater, ParquetPrimitiveConverter}
+import org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater, ParentContainerUpdater, ParquetPrimitiveConverter, ParquetVectorUpdater, VectorizedValuesReader}
+import org.apache.spark.sql.execution.vectorized.WritableColumnVector
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataType, TimestampLTZNanosType, TimestampNTZNanosType}
import org.apache.spark.unsafe.types.TimestampNanosVal
@@ -53,9 +56,9 @@ import org.apache.spark.unsafe.types.TimestampNanosVal
*
* TIMESTAMP(NANOS) postdates Spark's switch to the proleptic Gregorian calendar, so the values are
* exempt from datetime rebasing (the rebase modes only cover DATE, TIMESTAMP_MILLIS and
- * TIMESTAMP_MICROS). Vectorized read is not supported: the value is a 16-byte composite rather
- * than a single long slot, so `isBatchReadSupported` stays false (the trait default) and reads go
- * through the row-based converter.
+ * TIMESTAMP_MICROS). The vectorized reader decomposes the INT64 epoch-nanos into the two-child
+ * column vector (epochMicros: Long, nanosWithinMicro: Short) via [[TimestampNanosVectorUpdater]],
+ * routed through the types framework's `getVectorUpdater` hook.
*
* @see ParquetTypeOps for the dispatch contract
* @since 4.3.0
@@ -77,6 +80,22 @@ private[parquet] trait TimestampNanosParquetOps extends ParquetTypeOps {
// The Parquet TIMESTAMP `isAdjustedToUTC` flag: LTZ is UTC-adjusted, NTZ is not.
private def isAdjustedToUTC: Boolean = !isNtz
+ // ==================== Vectorized Read Support ====================
+
+ override def isBatchReadSupported(sqlConf: SQLConf): Boolean = true
+
+ // Only a canonical INT64 TIMESTAMP(NANOS) column can be vectorized-decoded as a nanos timestamp.
+ // Return None for anything else so the factory falls through to its clean
+ // SchemaColumnConvertNotSupportedException instead of silently mis-reading.
+ override def getVectorUpdater(descriptor: ColumnDescriptor): Option[ParquetVectorUpdater] = {
+ val parquetType = descriptor.getPrimitiveType
+ if (TimestampNanosParquetOps.isNanosTimestamp(parquetType)) {
+ Some(new TimestampNanosVectorUpdater(precision))
+ } else {
+ None
+ }
+ }
+
// ==================== Schema Conversion ====================
override def convertToParquetType(
@@ -167,3 +186,66 @@ private[ops] object TimestampNanosParquetOps {
case _ => false
})
}
+
+/**
+ * Vectorized (batch) updater for nanosecond-precision timestamps: reads an INT64 epoch-nanos
+ * column and decomposes each value into the two-child column vector (epochMicros: Long,
+ * nanosWithinMicro: Short). Sub-microsecond digits are truncated to the requested precision.
+ * Mirrors the row-based `newConverter` path which calls
+ * `DateTimeUtils.epochNanosToTimestampNanos(value, precision)`.
+ *
+ * No datetime rebase is applied (TIMESTAMP(NANOS) postdates the proleptic Gregorian switch).
+ * No timezone conversion is applied at the storage level.
+ *
+ * Owned by the type's ops ([[TimestampNanosParquetOps]]) and routed through the types
+ * framework's `getVectorUpdater` hook (no factory-level branch).
+ */
+private[ops] class TimestampNanosVectorUpdater(precision: Int) extends ParquetVectorUpdater {
+
+ // Truncation uses the shared DateTimeUtils.truncateNanosWithinMicroToPrecision helper so the
+ // vectorized path stays in lock-step with the row-based reader (epochNanosToTimestampNanos).
+ // We decompose floorDiv/floorMod inline rather than calling epochNanosToTimestampNanos directly
+ // because that method returns a heap-allocated TimestampNanosVal per value, which would add
+ // GC pressure on the hot vectorized decode loop. The truncation itself is a pure Int->Int
+ // operation with no allocation.
+
+ private def putTimestampNanos(
+ offset: Int, values: WritableColumnVector, epochNanos: Long): Unit = {
+ val epochMicros = Math.floorDiv(epochNanos, 1000L)
+ val rawNanosWithinMicro = Math.floorMod(epochNanos, 1000L).toInt
+ val nanosWithinMicro =
+ DateTimeUtils.truncateNanosWithinMicroToPrecision(rawNanosWithinMicro, precision).toShort
+ values.getChild(0).putLong(offset, epochMicros)
+ values.getChild(1).putShort(offset, nanosWithinMicro)
+ }
+
+ override def readValues(
+ total: Int,
+ offset: Int,
+ values: WritableColumnVector,
+ valuesReader: VectorizedValuesReader): Unit = {
+ var i = 0
+ while (i < total) {
+ putTimestampNanos(offset + i, values, valuesReader.readLong())
+ i += 1
+ }
+ }
+
+ override def skipValues(total: Int, valuesReader: VectorizedValuesReader): Unit =
+ valuesReader.skipLongs(total)
+
+ override def readValue(
+ offset: Int,
+ values: WritableColumnVector,
+ valuesReader: VectorizedValuesReader): Unit =
+ putTimestampNanos(offset, values, valuesReader.readLong())
+
+ override def decodeSingleDictionaryId(
+ offset: Int,
+ values: WritableColumnVector,
+ dictionaryIds: WritableColumnVector,
+ dictionary: Dictionary): Unit = {
+ val epochNanos = dictionary.decodeToLong(dictionaryIds.getDictId(offset))
+ putTimestampNanos(offset, values, epochNanos)
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
index 4439062ccfdc..b9ac9df8e31d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
@@ -2369,6 +2369,47 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession {
}
}
}
+
+ test("SPARK-57828: vectorized read of TIMESTAMP(NANOS) with dictionary encoding") {
+ // Exercises TimestampNanosVectorUpdater.decodeSingleDictionaryId by writing a file with
+ // dictionary encoding enabled and a nullable column (OPTIONAL repetition, which forces
+ // single-value decode paths via def-level interleaving). Analogous to the TIME(NANOS)
+ // dictionary test above (SPARK-55444).
+ val schema = MessageTypeParser.parseMessageType(
+ """message root {
+ | optional int64 ts_nanos(TIMESTAMP(NANOS,true));
+ |}""".stripMargin)
+ val readSchema = new StructType()
+ .add("ts_nanos", TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION))
+
+ // 1_000_000_500 ns = epoch micros 1000000 + 500 ns within micro
+ // Instant: 1000000 micros = 1.000000 seconds, plus 500 ns -> 1.000000500 s
+ val epochNanosValue = 1000000500L
+ val expectedInstant = java.time.Instant.ofEpochSecond(1L, 500L)
+
+ for (dictEnabled <- Seq(true, false)) {
+ withTempDir { dir =>
+ val tablePath = new Path(s"${dir.getCanonicalPath}/ts_nanos_dict.parquet")
+ val numRecords = 100
+ val writer = createParquetWriter(schema, tablePath, dictionaryEnabled = dictEnabled)
+ (0 until numRecords).foreach { i =>
+ val record = new SimpleGroup(schema)
+ // Every 7th row is null: interleaves null/value runs to force single-value path.
+ if (i % 7 != 0) record.add(0, epochNanosValue)
+ writer.write(record)
+ }
+ writer.close()
+
+ withAllParquetReaders {
+ val df = spark.read.schema(readSchema).parquet(tablePath.toString)
+ val expected = (0 until numRecords).map { i =>
+ if (i % 7 == 0) Row(null) else Row(expectedInstant)
+ }
+ checkAnswer(df, expected)
+ }
+ }
+ }
+ }
}
class JobCommitFailureParquetOutputCommitter(outputPath: Path, context: TaskAttemptContext)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
index b3e8ecdd5868..e62d952a9eb9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
@@ -100,37 +100,37 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
test("SPARK-57102: read a foreign TIMESTAMP(NANOS) file as nanosecond timestamp types") {
withNanosEnabled {
- withSQLConf(
- SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
- SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
- val values = Seq(Some(123L), Some(1000000123L), Some(-1L), None)
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ withAllParquetReaders {
+ val values = Seq(Some(123L), Some(1000000123L), Some(-1L), None)
- withTempPath { dir =>
- val file = new File(dir, "ntz.parquet")
- writeForeignNanosParquet(file, isAdjustedToUTC = false, values)
- val read = spark.read.parquet(file.getCanonicalPath)
- assert(read.schema("ts").dataType === TimestampNTZNanosType(9))
- checkAnswer(read, spark.sql(
- """SELECT * FROM VALUES
- | (TIMESTAMP_NTZ '1970-01-01 00:00:00.000000123'),
- | (TIMESTAMP_NTZ '1970-01-01 00:00:01.000000123'),
- | (TIMESTAMP_NTZ '1969-12-31 23:59:59.999999999'),
- | (CAST(NULL AS TIMESTAMP_NTZ(9)))
- | AS t(ts)""".stripMargin).collect().toSeq)
- }
+ withTempPath { dir =>
+ val file = new File(dir, "ntz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, values)
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampNTZNanosType(9))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:00.000000123'),
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:01.000000123'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.999999999'),
+ | (CAST(NULL AS TIMESTAMP_NTZ(9)))
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
- withTempPath { dir =>
- val file = new File(dir, "ltz.parquet")
- writeForeignNanosParquet(file, isAdjustedToUTC = true, values)
- val read = spark.read.parquet(file.getCanonicalPath)
- assert(read.schema("ts").dataType === TimestampLTZNanosType(9))
- checkAnswer(read, spark.sql(
- """SELECT * FROM VALUES
- | (TIMESTAMP_LTZ '1970-01-01 00:00:00.000000123'),
- | (TIMESTAMP_LTZ '1970-01-01 00:00:01.000000123'),
- | (TIMESTAMP_LTZ '1969-12-31 23:59:59.999999999'),
- | (CAST(NULL AS TIMESTAMP_LTZ(9)))
- | AS t(ts)""".stripMargin).collect().toSeq)
+ withTempPath { dir =>
+ val file = new File(dir, "ltz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = true, values)
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampLTZNanosType(9))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:00.000000123'),
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:01.000000123'),
+ | (TIMESTAMP_LTZ '1969-12-31 23:59:59.999999999'),
+ | (CAST(NULL AS TIMESTAMP_LTZ(9)))
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
}
}
}
@@ -138,38 +138,38 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
test("SPARK-57102: explicit lower-precision read schema truncates sub-precision nanos") {
withNanosEnabled {
- withSQLConf(
- SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
- SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
- // Foreign file with full 9-digit precision: .123456789 after the epoch, and -1ns (which
- // floors to epochMicros = -1, nanosWithinMicro = 999).
- val values = Seq(Some(123456789L), Some(-1L))
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ withAllParquetReaders {
+ // Foreign file with full 9-digit precision: .123456789 after the epoch, and -1ns (which
+ // floors to epochMicros = -1, nanosWithinMicro = 999).
+ val values = Seq(Some(123456789L), Some(-1L))
- withTempPath { dir =>
- val file = new File(dir, "ntz.parquet")
- writeForeignNanosParquet(file, isAdjustedToUTC = false, values)
- // Reading with an explicit TIMESTAMP_NTZ(7) schema must floor the sub-microsecond
- // digits to precision 7, matching DateTimeUtils.localDateTimeToTimestampNanos.
- val read = spark.read.schema("ts TIMESTAMP_NTZ(7)").parquet(file.getCanonicalPath)
- assert(read.schema("ts").dataType === TimestampNTZNanosType(7))
- checkAnswer(read, spark.sql(
- """SELECT * FROM VALUES
- | (TIMESTAMP_NTZ '1970-01-01 00:00:00.123456700'),
- | (TIMESTAMP_NTZ '1969-12-31 23:59:59.999999900')
- | AS t(ts)""".stripMargin).collect().toSeq)
- }
+ withTempPath { dir =>
+ val file = new File(dir, "ntz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, values)
+ // Reading with an explicit TIMESTAMP_NTZ(7) schema must floor the sub-microsecond
+ // digits to precision 7, matching DateTimeUtils.localDateTimeToTimestampNanos.
+ val read = spark.read.schema("ts TIMESTAMP_NTZ(7)").parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampNTZNanosType(7))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:00.123456700'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.999999900')
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
- withTempPath { dir =>
- val file = new File(dir, "ltz.parquet")
- writeForeignNanosParquet(file, isAdjustedToUTC = true, values)
- // precision 8 drops only the last digit.
- val read = spark.read.schema("ts TIMESTAMP_LTZ(8)").parquet(file.getCanonicalPath)
- assert(read.schema("ts").dataType === TimestampLTZNanosType(8))
- checkAnswer(read, spark.sql(
- """SELECT * FROM VALUES
- | (TIMESTAMP_LTZ '1970-01-01 00:00:00.123456780'),
- | (TIMESTAMP_LTZ '1969-12-31 23:59:59.999999990')
- | AS t(ts)""".stripMargin).collect().toSeq)
+ withTempPath { dir =>
+ val file = new File(dir, "ltz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = true, values)
+ // precision 8 drops only the last digit.
+ val read = spark.read.schema("ts TIMESTAMP_LTZ(8)").parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampLTZNanosType(8))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:00.123456780'),
+ | (TIMESTAMP_LTZ '1969-12-31 23:59:59.999999990')
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
}
}
}
@@ -269,22 +269,22 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
test("SPARK-57102: datetime rebase configs do not affect TIMESTAMP(NANOS) reads") {
withNanosEnabled {
- withSQLConf(
- SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
- SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
- withTempPath { dir =>
- val file = new File(dir, "ltz.parquet")
- // 1800-01-01 00:00:00 UTC predates the last Julian-to-Gregorian switch instant of the
- // rebase logic, so applying a timestamp rebase (or the EXCEPTION-mode guard) on this
- // value would change the result or fail the read.
- writeForeignNanosParquet(
- file, isAdjustedToUTC = true, Seq(Some(-5364662400000000000L)))
- Seq("EXCEPTION", "CORRECTED", "LEGACY").foreach { mode =>
- withSQLConf(SQLConf.PARQUET_REBASE_MODE_IN_READ.key -> mode) {
- checkAnswer(
- spark.read.parquet(file.getCanonicalPath),
- spark.sql(
- "SELECT TIMESTAMP_LTZ '1800-01-01 00:00:00.000000000'").collect().toSeq)
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ withAllParquetReaders {
+ withTempPath { dir =>
+ val file = new File(dir, "ltz.parquet")
+ // 1800-01-01 00:00:00 UTC predates the last Julian-to-Gregorian switch instant of the
+ // rebase logic, so applying a timestamp rebase (or the EXCEPTION-mode guard) on this
+ // value would change the result or fail the read.
+ writeForeignNanosParquet(
+ file, isAdjustedToUTC = true, Seq(Some(-5364662400000000000L)))
+ Seq("EXCEPTION", "CORRECTED", "LEGACY").foreach { mode =>
+ withSQLConf(SQLConf.PARQUET_REBASE_MODE_IN_READ.key -> mode) {
+ checkAnswer(
+ spark.read.parquet(file.getCanonicalPath),
+ spark.sql(
+ "SELECT TIMESTAMP_LTZ '1800-01-01 00:00:00.000000000'").collect().toSeq)
+ }
}
}
}
@@ -294,7 +294,7 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
test("SPARK-57102: nanos timestamps round-trip inside a nested (array) column") {
withNanosEnabled {
- withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ withAllParquetReaders {
withTempPath { dir =>
val df = spark.sql(
"SELECT array(TIMESTAMP_NTZ '2020-01-01 00:00:00.123456789', " +
@@ -311,18 +311,63 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
test("SPARK-57102: nanos timestamps round-trip via the V2 file source") {
withNanosEnabled {
- withSQLConf(
- SQLConf.USE_V1_SOURCE_LIST.key -> "",
- SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
- withTempPath { dir =>
- val df = spark.sql(
- "SELECT TIMESTAMP_NTZ '2020-01-01 12:34:56.123456789' AS ntz, " +
- "TIMESTAMP_LTZ '2020-01-01 12:34:56.123456789' AS ltz")
- df.write.parquet(dir.getCanonicalPath)
- val read = spark.read.parquet(dir.getCanonicalPath)
- assert(read.schema("ntz").dataType === TimestampNTZNanosType(9))
- assert(read.schema("ltz").dataType === TimestampLTZNanosType(9))
- checkAnswer(read, df.collect().toSeq)
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
+ withAllParquetReaders {
+ withTempPath { dir =>
+ val df = spark.sql(
+ "SELECT TIMESTAMP_NTZ '2020-01-01 12:34:56.123456789' AS ntz, " +
+ "TIMESTAMP_LTZ '2020-01-01 12:34:56.123456789' AS ltz")
+ df.write.parquet(dir.getCanonicalPath)
+ val read = spark.read.parquet(dir.getCanonicalPath)
+ assert(read.schema("ntz").dataType === TimestampNTZNanosType(9))
+ assert(read.schema("ltz").dataType === TimestampLTZNanosType(9))
+ checkAnswer(read, df.collect().toSeq)
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57828: vectorized reader produces identical results to row-based for nanos") {
+ withNanosEnabled {
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ // Write foreign parquet files with TIMESTAMP(NANOS) values covering edge cases:
+ // positive, negative (pre-epoch), values with precision truncation, and nulls.
+ val values = Seq(
+ Some(123456789L), // 1970-01-01 00:00:00.123456789
+ Some(-1L), // 1969-12-31 23:59:59.999999999 (pre-epoch, floor semantics)
+ Some(1000000000L), // 1970-01-01 00:00:01.000000000 (exact second)
+ Some(-1000000001L), // 1969-12-31 23:59:58.999999999
+ None // null
+ )
+
+ Seq(7, 8, 9).foreach { p =>
+ Seq(true, false).foreach { isAdjustedToUTC =>
+ val typeName = if (isAdjustedToUTC) "TIMESTAMP_LTZ" else "TIMESTAMP_NTZ"
+ withClue(s"$typeName($p)") {
+ withTempPath { dir =>
+ val file = new File(dir, s"ts_nanos_${typeName}_p$p.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC, values)
+
+ val schema = s"ts $typeName($p)"
+
+ // Row-based read
+ val rowBased = withSQLConf(
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ spark.read.schema(schema).parquet(file.getCanonicalPath).collect()
+ }
+
+ // Vectorized read
+ val vectorized = withSQLConf(
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ spark.read.schema(schema).parquet(file.getCanonicalPath).collect()
+ }
+
+ assert(vectorized.toSeq === rowBased.toSeq,
+ s"Vectorized and row-based results differ for $typeName($p)")
+ }
+ }
+ }
}
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
index deade76358fe..febeb7c5fb1e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
+import org.apache.parquet.column.ColumnDescriptor
import org.apache.parquet.io.api.PrimitiveConverter
import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
import org.apache.parquet.schema.LogicalTypeAnnotation.{TimestampLogicalTypeAnnotation, TimeUnit}
@@ -153,6 +154,49 @@ class TimestampNanosParquetOpsSuite extends SparkFunSuite {
captured
}
+ // ---------- vectorized read updater (getVectorUpdater / getVectorUpdaterOrNull) ----------
+
+ private def nanosTimestampColumn(isAdjustedToUTC: Boolean): ColumnDescriptor = {
+ val field = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(isAdjustedToUTC, TimeUnit.NANOS))
+ .named("c")
+ new ColumnDescriptor(Array("c"), field, 0, 0)
+ }
+
+ test("getVectorUpdater returns a framework updater for INT64 TIMESTAMP(NANOS)") {
+ // LTZ: isAdjustedToUTC=true
+ assert(ltz.getVectorUpdater(nanosTimestampColumn(isAdjustedToUTC = true)).isDefined)
+ // NTZ: isAdjustedToUTC=false
+ assert(ntz.getVectorUpdater(nanosTimestampColumn(isAdjustedToUTC = false)).isDefined)
+ // Java-friendly companion entry point used by ParquetVectorUpdaterFactory.
+ assert(ParquetTypeOps.getVectorUpdaterOrNull(
+ TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION),
+ nanosTimestampColumn(isAdjustedToUTC = true)) != null)
+ }
+
+ test("getVectorUpdater returns None for incompatible encodings (clean reject, vectorized)") {
+ // TIMESTAMP(MICROS) - wrong unit
+ val timestampMicros = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(true, TimeUnit.MICROS)).named("c")
+ // raw INT64 - no annotation
+ val rawInt64 = Types.primitive(INT64, REQUIRED).named("c")
+ // TIME(NANOS) - wrong annotation kind
+ val timeNanos = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timeType(false, TimeUnit.NANOS)).named("c")
+
+ // None -> the factory falls through to a clean SchemaColumnConvertNotSupportedException,
+ // matching the row-path reject set (isNanosTimestamp), instead of silently mis-decoding.
+ Seq(timestampMicros, rawInt64, timeNanos).foreach { field =>
+ val descriptor = new ColumnDescriptor(Array("c"), field, 0, 0)
+ assert(ltz.getVectorUpdater(descriptor).isEmpty,
+ s"expected None for ${field.getLogicalTypeAnnotation}")
+ assert(ntz.getVectorUpdater(descriptor).isEmpty,
+ s"expected None for ${field.getLogicalTypeAnnotation}")
+ }
+ }
+
+ // ---------- helpers ----------
+
private def assertNanosTimestampSchema(
parquetType: Type, expectedAdjustedToUTC: Boolean): Unit = {
assert(parquetType.isPrimitive, s"expected a primitive type, got $parquetType")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org