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 155d706859505..75af98a2affd1 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 ParquetVectorUpdater getUpdater(ColumnDescriptor descriptor, DataType spa 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 @@ boolean isTimestampTypeMatched(LogicalTypeAnnotation.TimeUnit unit) { 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 void decodeSingleDictionaryId( } } + // 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 0adece577c58b..93f4000fa2760 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 04f6633afd125..746d8b46fb462 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 f023e8390f155..9b851134fc0b2 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 2ce1566027258..2e2a7e18316cd 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 ac33bd7ba3364..7a72d485b429b 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 f0063f86d1117..598a84f3dbe35 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 = {