[SPARK-59284][SQL] Read INT64 TIMESTAMP(MICROS) columns as nanosecond timestamps

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(

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)

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

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),

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 ||