You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/09/07 08:06:09 UTC
(spark) branch branch-4.x updated: [SPARK-59284][SQL] Read INT64 TIMESTAMP(MICROS) columns as nanosecond timestamps
This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 1f57956abcd6 [SPARK-59284][SQL] Read INT64 TIMESTAMP(MICROS) columns as nanosecond timestamps
1f57956abcd6 is described below
commit 1f57956abcd628b82f8707b292dc4415b93fe097
Author: Stevo Mitric <st...@gmail.com>
AuthorDate: Mon Sep 7 08:05:48 2026 +0000
[SPARK-59284][SQL] Read INT64 TIMESTAMP(MICROS) columns as nanosecond timestamps
### What changes were proposed in this pull request?
Let a nanosecond timestamp type read a Parquet INT64 TIMESTAMP(MICROS) column, promoting each micros value to the internal (epochMicros, nanosWithinMicro = 0) representation. Three touch points:
- Row-based: the extended newConverter now accepts a TIMESTAMP(MICROS) column (LTZ applies the Julian→Gregorian rebase; NTZ needs none).
- Vectorized: new MicrosAsTimestampNanos(Rebase)Updater updaters, dispatched for a nanos type over a MICROS file.
- Filter pushdown: only push down a micros/millis timestamp bound that lies exactly on the microsecond grid, so a sub-µs bound never over-prunes row groups.
Scope: INT64 TIMESTAMP(MICROS) only; INT96 and TIMESTAMP_MILLIS sources remain unsupported.
### Why are the changes needed?
It's the read side of widening a microsecond timestamp column (TIMESTAMP(6)) to nanosecond precision without rewriting existing files. Today the nanos reader fails loudly on any non-TIMESTAMP(NANOS) column, so old micros files would be unreadable after such a widening. The promotion is exact and range-complete - micros to (micros, 0) involves no ×1000 encode, so it avoids the INT64 epoch-nanos overflow range (~1677–2262).
### Does this PR introduce _any_ user-facing change?
Yes, behind the preview flag spark.sql.timestampNanosTypes.enabled. Previously, reading a TIMESTAMP(MICROS) Parquet column under a nanosecond timestamp type threw (SchemaColumnConvertNotSupportedException / PARQUET_CONVERSION_FAILURE); now it reads the values correctly.
### How was this patch tested?
Added TimestampNanosParquetOpsSuite and updated ParquetTypeWideningSuite ParquetFilterSuite.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored-by: Claude Opus 4.8
Closes #58550 from stevomitric/stevomitric/nanos-read-micros-parquet.
Authored-by: Stevo Mitric <st...@gmail.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
(cherry picked from commit 3c2495097f4ad23e1424bab423eb4803bd8debb9)
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../parquet/ParquetVectorUpdaterFactory.java | 124 +++++++++++++++++++++
.../datasources/parquet/ParquetFilters.scala | 11 +-
.../types/ops/TimestampNanosParquetOps.scala | 85 ++++++++++++--
.../datasources/parquet/ParquetFilterSuite.scala | 28 +++++
.../parquet/ParquetTimestampNanosSuite.scala | 19 ++--
.../parquet/ParquetTypeWideningSuite.scala | 67 ++++++++++-
.../types/ops/TimestampNanosParquetOpsSuite.scala | 69 ++++++++++++
7 files changed, 382 insertions(+), 21 deletions(-)
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
index 155d70685950..75af98a2affd 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
@@ -168,6 +168,25 @@ public class ParquetVectorUpdaterFactory {
isTimestampTypeMatched(LogicalTypeAnnotation.TimeUnit.MILLIS)) {
// TIMESTAMP_NTZ is a new data type and has no legacy files that need to do rebase.
return new LongAsMicrosUpdater();
+ } else if (sparkType instanceof TimestampLTZNanosType &&
+ isTimestampTypeMatched(LogicalTypeAnnotation.TimeUnit.MICROS, true)) {
+ // Read side of widening a microsecond LTZ timestamp (TIMESTAMP(6)) to nanosecond
+ // precision: old files stay INT64 TIMESTAMP(MICROS); promote each micros value to
+ // (epochMicros = value, nanosWithinMicro = 0). The isAdjustedToUTC=true match keeps this
+ // to the LTZ family. LTZ files can be legacy Julian, so rebase exactly like the
+ // TimestampType read path above.
+ if ("CORRECTED".equals(datetimeRebaseMode)) {
+ return new MicrosAsTimestampNanosUpdater();
+ } else {
+ boolean failIfRebase = "EXCEPTION".equals(datetimeRebaseMode);
+ return new MicrosAsTimestampNanosRebaseUpdater(failIfRebase, datetimeRebaseTz);
+ }
+ } else if (sparkType instanceof TimestampNTZNanosType &&
+ isTimestampTypeMatched(LogicalTypeAnnotation.TimeUnit.MICROS, false)) {
+ // TIMESTAMP_NTZ(nanos) postdates the proleptic Gregorian switch: no legacy files, no
+ // rebase (mirrors the TimestampNTZType read path). The isAdjustedToUTC=false match keeps
+ // this to the NTZ family.
+ return new MicrosAsTimestampNanosUpdater();
} else if (sparkType instanceof DayTimeIntervalType) {
return new LongUpdater();
} else if (canReadAsDecimal(descriptor, sparkType)) {
@@ -249,6 +268,13 @@ public class ParquetVectorUpdaterFactory {
annotation.getUnit() == unit;
}
+ // Also matches the time-zone family (isAdjustedToUTC). Used when reading a micros column as a
+ // nanosecond type, so a cross-family file (e.g. an NTZ file requested as LTZ) is not mis-decoded.
+ boolean isTimestampTypeMatched(LogicalTypeAnnotation.TimeUnit unit, boolean isAdjustedToUTC) {
+ return logicalTypeAnnotation instanceof TimestampLogicalTypeAnnotation annotation &&
+ annotation.getUnit() == unit && annotation.isAdjustedToUTC() == isAdjustedToUTC;
+ }
+
boolean isUnsignedIntTypeMatched(int bitWidth) {
return logicalTypeAnnotation instanceof IntLogicalTypeAnnotation annotation &&
!annotation.isSigned() && annotation.getBitWidth() == bitWidth;
@@ -913,6 +939,104 @@ public class ParquetVectorUpdaterFactory {
}
}
+ // Reads an INT64 TIMESTAMP(MICROS) column as a nanosecond timestamp, promoting each micros value
+ // to the two-child (epochMicros, nanosWithinMicro) vector as (value, 0) -- the vectorized read
+ // side of widening TIMESTAMP(6) to nanosecond precision.
+ private static class MicrosAsTimestampNanosUpdater implements ParquetVectorUpdater {
+ @Override
+ public void readValues(
+ int total,
+ int offset,
+ WritableColumnVector values,
+ VectorizedValuesReader valuesReader) {
+ for (int i = 0; i < total; i++) {
+ putMicrosAsNanos(offset + i, values, valuesReader.readLong());
+ }
+ }
+
+ @Override
+ public void skipValues(int total, VectorizedValuesReader valuesReader) {
+ valuesReader.skipLongs(total);
+ }
+
+ @Override
+ public void readValue(
+ int offset,
+ WritableColumnVector values,
+ VectorizedValuesReader valuesReader) {
+ putMicrosAsNanos(offset, values, valuesReader.readLong());
+ }
+
+ @Override
+ public void decodeSingleDictionaryId(
+ int offset,
+ WritableColumnVector values,
+ WritableColumnVector dictionaryIds,
+ Dictionary dictionary) {
+ putMicrosAsNanos(offset, values, dictionary.decodeToLong(dictionaryIds.getDictId(offset)));
+ }
+
+ private static void putMicrosAsNanos(
+ int offset, WritableColumnVector values, long epochMicros) {
+ values.getChild(0).putLong(offset, epochMicros);
+ values.getChild(1).putShort(offset, (short) 0);
+ }
+ }
+
+ // LTZ variant of MicrosAsTimestampNanosUpdater: legacy (Julian) micros files are rebased to
+ // proleptic Gregorian before promotion, mirroring LongWithRebaseUpdater for the microsecond read
+ // path. Rebase is applied per value (not via the bulk readLongsWithRebase primitive) because the
+ // target is a two-child struct vector rather than a flat long vector.
+ private static class MicrosAsTimestampNanosRebaseUpdater implements ParquetVectorUpdater {
+ private final boolean failIfRebase;
+ private final String timeZone;
+
+ MicrosAsTimestampNanosRebaseUpdater(boolean failIfRebase, String timeZone) {
+ this.failIfRebase = failIfRebase;
+ this.timeZone = timeZone;
+ }
+
+ @Override
+ public void readValues(
+ int total,
+ int offset,
+ WritableColumnVector values,
+ VectorizedValuesReader valuesReader) {
+ for (int i = 0; i < total; i++) {
+ putRebasedMicrosAsNanos(offset + i, values, valuesReader.readLong());
+ }
+ }
+
+ @Override
+ public void skipValues(int total, VectorizedValuesReader valuesReader) {
+ valuesReader.skipLongs(total);
+ }
+
+ @Override
+ public void readValue(
+ int offset,
+ WritableColumnVector values,
+ VectorizedValuesReader valuesReader) {
+ putRebasedMicrosAsNanos(offset, values, valuesReader.readLong());
+ }
+
+ @Override
+ public void decodeSingleDictionaryId(
+ int offset,
+ WritableColumnVector values,
+ WritableColumnVector dictionaryIds,
+ Dictionary dictionary) {
+ putRebasedMicrosAsNanos(
+ offset, values, dictionary.decodeToLong(dictionaryIds.getDictId(offset)));
+ }
+
+ private void putRebasedMicrosAsNanos(
+ int offset, WritableColumnVector values, long julianMicros) {
+ values.getChild(0).putLong(offset, rebaseMicros(julianMicros, failIfRebase, timeZone));
+ values.getChild(1).putShort(offset, (short) 0);
+ }
+ }
+
private static class FloatUpdater implements ParquetVectorUpdater {
@Override
public void readValues(
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
index 0adece577c58..93f4000fa276 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
@@ -39,6 +39,7 @@ import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.sql.catalyst.expressions.variant.{ObjectExtraction, VariantPathParser}
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils, IntervalUtils}
+import org.apache.spark.sql.catalyst.util.DateTimeConstants.NANOS_PER_MICROS
import org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays, rebaseGregorianToJulianMicros, RebaseSpec}
import org.apache.spark.sql.execution.datasources.VariantMetadata
import org.apache.spark.sql.execution.datasources.parquet.types.ops.{ParquetFilterOps, ParquetTypeOps}
@@ -942,7 +943,15 @@ class ParquetFilters(
case ParquetDateType =>
value.isInstanceOf[Date] || value.isInstanceOf[LocalDate]
case ParquetTimestampMicrosType | ParquetTimestampMillisType =>
- value.isInstanceOf[Timestamp] || value.isInstanceOf[Instant]
+ // A micros/millis column may be read under a nanosecond type (widening TIMESTAMP(6) to
+ // nanos). Pushing a sub-microsecond bound through the micros/millis converter truncates it
+ // and can over-prune row groups, so only push down a bound on the microsecond grid; a
+ // sub-microsecond one falls back to a full scan. Mirrors the decimal scale-match guard.
+ value match {
+ case i: Instant => i.getNano % NANOS_PER_MICROS == 0
+ case t: Timestamp => t.getNanos % NANOS_PER_MICROS == 0
+ case _ => false
+ }
case FrameworkFilterOps(ops) => ops.acceptsValue(value)
case ParquetSchemaType(decimalType: DecimalLogicalTypeAnnotation, INT32, _) =>
isDecimalMatched(value, decimalType)
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 04f6633afd12..746d8b46fb46 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
@@ -18,7 +18,7 @@
package org.apache.spark.sql.execution.datasources.parquet.types.ops
import java.lang.{Long => JLong}
-import java.time.{Instant, LocalDateTime}
+import java.time.{Instant, LocalDateTime, ZoneId}
import org.apache.parquet.column.{ColumnDescriptor, Dictionary}
import org.apache.parquet.io.api.{Converter, RecordConsumer}
@@ -29,8 +29,10 @@ 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.catalyst.util.RebaseDateTime.RebaseSpec
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater, ParentContainerUpdater, ParquetPrimitiveConverter, ParquetVectorUpdater, VectorizedValuesReader}
+import org.apache.spark.sql.execution.datasources.DataSourceUtils
+import org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater, ParentContainerUpdater, ParquetPrimitiveConverter, ParquetToSparkSchemaConverter, 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}
@@ -87,9 +89,9 @@ private[parquet] trait TimestampNanosParquetOps extends ParquetTypeOps {
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.
+ // Vectorized-decode only the canonical INT64 TIMESTAMP(NANOS) encoding here; return None for
+ // anything else. The factory then handles an INT64 TIMESTAMP(MICROS) column (promoting it to
+ // nanos) and raises SchemaColumnConvertNotSupportedException for the rest.
override def getVectorUpdater(descriptor: ColumnDescriptor): Option[ParquetVectorUpdater] = {
val parquetType = descriptor.getPrimitiveType
if (TimestampNanosParquetOps.isNanosTimestamp(parquetType)) {
@@ -123,23 +125,62 @@ private[parquet] trait TimestampNanosParquetOps extends ParquetTypeOps {
// ==================== Row-Based Read ====================
+ // Simple (no-context) path: only the canonical INT64 TIMESTAMP(NANOS) encoding can be decoded
+ // without the datetime rebase spec. A microsecond source is handled only by the extended overload
+ // below (which receives the rebase spec); here anything but NANOS fails loudly, matching the
+ // legacy ParquetRowConverter behavior where the guarded nanos arms fell through to the
+ // cannot-create-converter error.
override def newConverter(
parquetType: Type,
updater: ParentContainerUpdater): Converter with HasParentContainerUpdater = {
- // Framework-first dispatch in ParquetRowConverter routes here for any nanos catalyst type,
- // regardless of the actual Parquet encoding. Only an INT64 TIMESTAMP(NANOS) column can be
- // decoded as a nanos timestamp; anything else (a non-NANOS timestamp, a foreign annotation,
- // etc.) must fail loudly, matching the legacy ParquetRowConverter behavior where the guarded
- // nanos arms fell through to the cannot-create-converter error.
if (!TimestampNanosParquetOps.isNanosTimestamp(parquetType)) {
throw QueryExecutionErrors.cannotCreateParquetConverterForDataTypeError(
sparkType, parquetType.toString)
}
+ nanosConverter(updater)
+ }
+
+ // Extended path (the one ParquetRowConverter actually calls): besides the canonical
+ // TIMESTAMP(NANOS) encoding, reads an INT64 TIMESTAMP(MICROS) column as a nanos value (widening
+ // TIMESTAMP(6) to nanos), mapping each micros value to (value, 0) with no *1000 encode.
+ override def newConverter(
+ parquetType: Type,
+ updater: ParentContainerUpdater,
+ schemaConverter: ParquetToSparkSchemaConverter,
+ convertTz: Option[ZoneId],
+ datetimeRebaseSpec: RebaseSpec,
+ int96RebaseSpec: RebaseSpec): Converter with HasParentContainerUpdater = {
+ if (TimestampNanosParquetOps.isNanosTimestamp(parquetType)) {
+ nanosConverter(updater)
+ } else if (TimestampNanosParquetOps.isMicrosTimestamp(parquetType, isAdjustedToUTC)) {
+ microsAsNanosConverter(updater, datetimeRebaseSpec)
+ } else {
+ throw QueryExecutionErrors.cannotCreateParquetConverterForDataTypeError(
+ sparkType, parquetType.toString)
+ }
+ }
+
+ private def nanosConverter(
+ updater: ParentContainerUpdater): Converter with HasParentContainerUpdater = {
val p = precision
new ParquetPrimitiveConverter(updater) {
- override def addLong(value: Long): Unit = {
+ override def addLong(value: Long): Unit =
this.updater.set(DateTimeUtils.epochNanosToTimestampNanos(value, p))
- }
+ }
+ }
+
+ private def microsAsNanosConverter(
+ updater: ParentContainerUpdater,
+ datetimeRebaseSpec: RebaseSpec): Converter with HasParentContainerUpdater = {
+ // LTZ micros files may be legacy Julian and need rebasing; NTZ never does (it postdates the
+ // proleptic Gregorian switch). Mirrors ParquetVectorUpdaterFactory's TimestampType (rebase) vs
+ // TimestampNTZType (no rebase) split for the microsecond read path.
+ val rebase: Long => Long =
+ if (isNtz) identity
+ else DataSourceUtils.createTimestampRebaseFuncInRead(datetimeRebaseSpec, "Parquet")
+ new ParquetPrimitiveConverter(updater) {
+ override def addLong(value: Long): Unit =
+ this.updater.set(TimestampNanosVal.fromParts(rebase(value), 0.toShort))
}
}
}
@@ -189,6 +230,26 @@ private[ops] object TimestampNanosParquetOps {
case _ => false
})
+ /**
+ * Whether the Parquet field is an INT64 TIMESTAMP(MICROS) column of the given time-zone family,
+ * i.e. the on-disk encoding of a microsecond-precision timestamp (TIMESTAMP(6)). Such a column
+ * can be read as a nanosecond timestamp -- the read side of widening TIMESTAMP(6) to nanosecond
+ * precision -- by promoting each micros value to (epochMicros = value, nanosWithinMicro = 0).
+ *
+ * `expectedAdjustedToUTC` must equal the requested type's family (true for LTZ, false for NTZ) so
+ * an explicit read schema cannot silently reinterpret a cross-family micros file (e.g. reading an
+ * NTZ file as an LTZ instant). Widening only ever pairs same-family types, so this guard just
+ * fails a deliberately mismatched `.schema(...)` loudly instead of mis-decoding.
+ */
+ private[ops] def isMicrosTimestamp(parquetType: Type, expectedAdjustedToUTC: Boolean): Boolean =
+ parquetType.isPrimitive &&
+ parquetType.asPrimitiveType.getPrimitiveTypeName == INT64 &&
+ (parquetType.getLogicalTypeAnnotation match {
+ case ts: TimestampLogicalTypeAnnotation =>
+ ts.getUnit == TimeUnit.MICROS && ts.isAdjustedToUTC == expectedAdjustedToUTC
+ case _ => false
+ })
+
// Repacks an externalized nanos filter value into the signed INT64 epoch-nanoseconds the write
// path produces. Conversion is at precision 9 (a lossless repack): the literal has already been
// floored to the column precision upstream, so no sub-microsecond digits are dropped here. The
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
index f023e8390f15..9b851134fc0b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
@@ -910,6 +910,34 @@ abstract class ParquetFilterSuite extends ParquetTest with SharedSparkSession {
}
}
+ test("don't push down sub-microsecond timestamp bounds on a micros column read as nanos") {
+ // A micros timestamp column may be read under a nanosecond timestamp type (the read side of
+ // widening TIMESTAMP(6) to nanosecond precision). Pushing a sub-microsecond bound through the
+ // micros converter would truncate it and could over-prune row groups, so such a bound must not
+ // be pushed down; a bound already on the microsecond grid still is.
+ withSQLConf(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> TIMESTAMP_MICROS.toString) {
+ val schema = StructType(Seq(StructField("cts", TimestampType)))
+ val parquetSchema = new SparkToParquetSchemaConverter(conf).convert(schema)
+ val parquetFilters = createParquetFilters(parquetSchema)
+
+ val onGrid = Instant.parse("2020-01-01T00:00:00.000001Z") // on the microsecond grid
+ val subMicro = Instant.parse("2020-01-01T00:00:00.000000500Z") // 500 ns past the grid
+
+ // A bound on the microsecond grid still pushes down.
+ assert(parquetFilters.createFilter(sources.LessThan("cts", onGrid)).isDefined)
+ // Sub-microsecond bounds must fall back to a full scan (no pushdown) for every predicate.
+ Seq[sources.Filter](
+ sources.LessThan("cts", subMicro),
+ sources.LessThanOrEqual("cts", subMicro),
+ sources.GreaterThan("cts", subMicro),
+ sources.GreaterThanOrEqual("cts", subMicro),
+ sources.EqualTo("cts", subMicro),
+ sources.In("cts", Array[Any](subMicro))).foreach { f =>
+ assert(parquetFilters.createFilter(f).isEmpty, s"should not push down sub-micro bound: $f")
+ }
+ }
+ }
+
test("don't push down filters that would result in overflows") {
val schema = StructType(Seq(
StructField("cbyte", ByteType),
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 2ce156602725..2e2a7e18316c 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
@@ -177,16 +177,21 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
test("SPARK-57102: requesting a nanos type over a non-NANOS Parquet column fails clearly") {
withNanosEnabled {
- withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ withSQLConf(
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key ->
+ SQLConf.ParquetOutputTimestampType.TIMESTAMP_MILLIS.toString) {
withTempPath { dir =>
- // Write a microsecond column: the Parquet annotation is TIMESTAMP(MICROS), not NANOS.
- spark.sql("SELECT TIMESTAMP_NTZ '2020-01-01 12:34:56.123456' AS ts")
+ // Write a TIMESTAMP(MILLIS) column: a coarser non-NANOS encoding, not the INT64
+ // TIMESTAMP(MICROS) the nanos reader now widens to nanos.
+ spark.sql("SELECT TIMESTAMP '2020-01-01 12:34:56.123' AS ts")
.write.parquet(dir.getCanonicalPath)
- // Forcing a nanosecond read schema leaves no matching converter case (the guard requires
- // a NANOS annotation), so it falls through to the generic PARQUET_CONVERSION_FAILURE
- // error - the same path every other type uses, not a confusing one.
+ // Forcing a nanosecond read schema leaves no matching converter case (only NANOS and
+ // same-family MICROS are accepted), so it falls through to the generic
+ // PARQUET_CONVERSION_FAILURE error - the same path every other type uses, not a confusing
+ // one.
val e = intercept[SparkException] {
- spark.read.schema("ts TIMESTAMP_NTZ(7)").parquet(dir.getCanonicalPath).collect()
+ spark.read.schema("ts TIMESTAMP_LTZ(7)").parquet(dir.getCanonicalPath).collect()
}
var cause: Throwable = e
while (cause != null && (cause.getMessage == null ||
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTypeWideningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTypeWideningSuite.scala
index ac33bd7ba336..7a72d485b429 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTypeWideningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTypeWideningSuite.scala
@@ -55,7 +55,8 @@ class ParquetTypeWideningSuite
toType: DataType,
expectError: => Boolean): Unit = {
val timestampRebaseModes = toType match {
- case _: TimestampNTZType | _: DateType =>
+ case _: TimestampNTZType | _: DateType |
+ _: TimestampLTZNanosType | _: TimestampNTZNanosType =>
Seq(LegacyBehaviorPolicy.CORRECTED, LegacyBehaviorPolicy.LEGACY)
case _ =>
Seq(LegacyBehaviorPolicy.CORRECTED)
@@ -199,6 +200,70 @@ class ParquetTypeWideningSuite
checkAllParquetReaders(values, fromType, toType, expectError = false)
}
+ // Widening TIMESTAMP(6) to nanosecond precision: INT64 TIMESTAMP(MICROS) files are read as nanos
+ // by promoting each micros value to (epochMicros, 0). Source must be TIMESTAMP(MICROS) (what
+ // Delta writes), hence the explicit output type; INT96/MILLIS aren't supported (see below).
+ // Values stay on the micros grid and include a pre-1582 date (LTZ Julian rebase, LEGACY mode) and
+ // a far-future date past the int64 epoch-nanos range (~2262). Requires the nanos preview flag.
+ for {
+ (fromType: DataType, toType: DataType) <- Seq(
+ TimestampType -> TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION),
+ TimestampType -> TimestampLTZNanosType(7),
+ TimestampNTZType -> TimestampNTZNanosType(TimestampNTZNanosType.NANOS_PRECISION))
+ }
+ test(s"parquet widening conversion $fromType (micros) -> $toType") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key ->
+ ParquetOutputTimestampType.TIMESTAMP_MICROS.toString) {
+ checkAllParquetReaders(
+ values = Seq(
+ "2020-01-01 12:34:56.123456", "1312-02-27 01:02:03.654321", "5138-11-16 09:46:40"),
+ fromType = fromType,
+ toType = toType,
+ expectError = false)
+ }
+ }
+
+ for {
+ outputTimestampType <-
+ Seq(ParquetOutputTimestampType.INT96, ParquetOutputTimestampType.TIMESTAMP_MILLIS)
+ }
+ test(s"unsupported parquet conversion TimestampType ($outputTimestampType) -> nanos") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> outputTimestampType.toString,
+ SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key -> LegacyBehaviorPolicy.CORRECTED.toString) {
+ checkAllParquetReaders(
+ values = Seq("2020-01-01 12:34:56.123456"),
+ fromType = TimestampType,
+ toType = TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION),
+ expectError = true)
+ }
+ }
+
+ // Cross-family reads must fail loudly, not reinterpret the values: the micros->nanos read matches
+ // the file's isAdjustedToUTC to the requested LTZ/NTZ family. Widening is same-family, so this
+ // only guards a deliberately mismatched explicit read schema.
+ for {
+ (fromType: DataType, toType: DataType) <- Seq(
+ TimestampNTZType -> TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION),
+ TimestampType -> TimestampNTZNanosType(TimestampNTZNanosType.NANOS_PRECISION))
+ }
+ test(s"unsupported cross-family parquet conversion $fromType (micros) -> $toType") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key ->
+ ParquetOutputTimestampType.TIMESTAMP_MICROS.toString) {
+ checkAllParquetReaders(
+ values = Seq("2020-01-01 12:34:56.123456"),
+ fromType = fromType,
+ toType = toType,
+ expectError = true)
+ }
+ }
+
+
for {
(values: Seq[String], fromType: DataType, toType: DataType) <- Seq(
(Seq("1", Byte.MaxValue.toString), ByteType, IntDecimal),
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 f0063f86d111..598a84f3dbe3 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
@@ -30,7 +30,9 @@ import org.apache.parquet.schema.Type.Repetition.REQUIRED
import org.apache.spark.{SparkArithmeticException, SparkFunSuite, SparkRuntimeException}
import org.apache.spark.sql.catalyst.util.{DateTimeConstants, DateTimeUtils}
+import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
import org.apache.spark.sql.execution.datasources.parquet.ParentContainerUpdater
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy
import org.apache.spark.sql.types.{TimestampLTZNanosType, TimestampNTZNanosType}
import org.apache.spark.unsafe.types.TimestampNanosVal
@@ -119,6 +121,52 @@ class TimestampNanosParquetOpsSuite extends SparkFunSuite {
assert(!TimestampNanosParquetOps.isNanosTimestamp(raw))
}
+ test("isMicrosTimestamp matches INT64 TIMESTAMP(MICROS) of the expected time-zone family") {
+ def field(adjustedToUTC: Boolean, unit: TimeUnit): Type =
+ Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(adjustedToUTC, unit)).named("c")
+ val ltzMicros = field(adjustedToUTC = true, TimeUnit.MICROS)
+ val ntzMicros = field(adjustedToUTC = false, TimeUnit.MICROS)
+
+ // Unit and time-zone family must both match.
+ assert(TimestampNanosParquetOps.isMicrosTimestamp(ltzMicros, expectedAdjustedToUTC = true))
+ assert(TimestampNanosParquetOps.isMicrosTimestamp(ntzMicros, expectedAdjustedToUTC = false))
+ // Cross-family (adjustment mismatch) is rejected so it cannot be mis-decoded.
+ assert(!TimestampNanosParquetOps.isMicrosTimestamp(ltzMicros, expectedAdjustedToUTC = false))
+ assert(!TimestampNanosParquetOps.isMicrosTimestamp(ntzMicros, expectedAdjustedToUTC = true))
+ // Non-micros encodings never match.
+ assert(!TimestampNanosParquetOps.isMicrosTimestamp(
+ field(adjustedToUTC = false, TimeUnit.NANOS), expectedAdjustedToUTC = false))
+ assert(!TimestampNanosParquetOps.isMicrosTimestamp(
+ field(adjustedToUTC = false, TimeUnit.MILLIS), expectedAdjustedToUTC = false))
+ assert(!TimestampNanosParquetOps.isMicrosTimestamp(
+ Types.primitive(INT64, REQUIRED).named("c"), expectedAdjustedToUTC = false))
+ }
+
+ test("extended newConverter reads INT64 TIMESTAMP(MICROS) as (epochMicros, 0)") {
+ assert(decodeMicros(ltz, isAdjustedToUTC = true, 1234567L) ===
+ TimestampNanosVal.fromParts(1234567L, 0.toShort))
+ assert(decodeMicros(ntz, isAdjustedToUTC = false, -1L) ===
+ TimestampNanosVal.fromParts(-1L, 0.toShort))
+ }
+
+ test("extended newConverter rejects a cross-family TIMESTAMP(MICROS) column") {
+ // A micros file whose time-zone family differs from the requested nanos type must fail loudly
+ // rather than reinterpret the values (e.g. an NTZ file requested as an LTZ instant).
+ intercept[SparkRuntimeException](decodeMicros(ltz, isAdjustedToUTC = false, 1L))
+ intercept[SparkRuntimeException](decodeMicros(ntz, isAdjustedToUTC = true, 1L))
+ }
+
+ test("extended newConverter promotes micros beyond the INT64 epoch-nanos range (no *1000)") {
+ // 1e17 micros (~year 5138) would overflow int64 if multiplied by 1000 to epoch-nanos; the
+ // micros->nanos read is range-complete because it sets (epochMicros = value, 0) directly.
+ val farFutureMicros = 100000000000000000L
+ assert(decodeMicros(ltz, isAdjustedToUTC = true, farFutureMicros) ===
+ TimestampNanosVal.fromParts(farFutureMicros, 0.toShort))
+ assert(decodeMicros(ntz, isAdjustedToUTC = false, farFutureMicros) ===
+ TimestampNanosVal.fromParts(farFutureMicros, 0.toShort))
+ }
+
// ---------- (epochMicros, nanosWithinMicro) -> INT64 epoch-nanos packing ----------
test("timestampNanosToEpochNanos combines micros and sub-micro nanos") {
@@ -281,6 +329,27 @@ class TimestampNanosParquetOpsSuite extends SparkFunSuite {
captured
}
+ private def microsField(isAdjustedToUTC: Boolean): Type =
+ Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(isAdjustedToUTC, TimeUnit.MICROS))
+ .named("c")
+
+ // Builds the extended converter (the one ParquetRowConverter calls) over a TIMESTAMP(MICROS)
+ // field with a CORRECTED (no-op) rebase spec, feeds one micros value through addLong, and returns
+ // the decoded TimestampNanosVal the converter set into its updater.
+ private def decodeMicros(
+ ops: TimestampNanosParquetOps, isAdjustedToUTC: Boolean, micros: Long): Any = {
+ var captured: Any = null
+ val updater = new ParentContainerUpdater {
+ override def set(value: Any): Unit = captured = value
+ }
+ val correctedSpec = RebaseSpec(LegacyBehaviorPolicy.CORRECTED)
+ val converter = ops.newConverter(
+ microsField(isAdjustedToUTC), updater, null, None, correctedSpec, correctedSpec)
+ converter.asInstanceOf[PrimitiveConverter].addLong(micros)
+ captured
+ }
+
// ---------- vectorized read updater (getVectorUpdater / getVectorUpdaterOrNull) ----------
private def nanosTimestampColumn(isAdjustedToUTC: Boolean): ColumnDescriptor = {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org