This is an automated email from the ASF dual-hosted git repository.
MaxGekk 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 75260f828fdd [SPARK-57416][SQL] Align the Types Framework Parquet read
guard for TimeType with the legacy reader
75260f828fdd is described below
commit 75260f828fdd97ff571d2b027669a6afd67198a2
Author: Stevo Mitric <[email protected]>
AuthorDate: Sun Jun 21 07:19:41 2026 +0200
[SPARK-57416][SQL] Align the Types Framework Parquet read guard for
TimeType with the legacy reader
### What changes were proposed in this pull request?
Follow-up to the Types Framework Phase 3a Parquet work (SPARK-55444).
`TimeTypeParquetOps.requireCompatibleParquetType` (the row-based read
guard) is relaxed to accept `INT64 TIME(MICROS)` regardless of the
`isAdjustedToUTC` flag, by dropping the `&& !t.isAdjustedToUTC` condition.
This mirrors the legacy `ParquetRowConverter` guard, which only checked the
TIME annotation and the MICROS unit. All other encodings (raw `INT64`,
`TIME(NANOS)`, `INT32 TIME(MILLIS)`, `TIMESTAMP(_)`, `DECIMAL`, group) are
still rejected.
### Why are the changes needed?
Phase 3a's guard was stricter than the guard it replaced, so reading an
`INT64 TIME(MICROS, isAdjustedToUTC=true)` column as `TimeType` failed on the
row-based reader (`FAILED_READ_FILE`) while the default vectorized reader still
accepted it — an inconsistency between the two readers and a behavior
regression versus pre-framework Spark. Since `TimeType` is zone-less,
`isAdjustedToUTC` carries no information on read (the raw micros-of-day decodes
identically), so the value is unchanged [...]
### Does this PR introduce _any_ user-facing change?
Yes (within unreleased master). Reading an `INT64 TIME(MICROS,
isAdjustedToUTC=true)` Parquet column as `TimeType` via the row-based reader
(vectorized reader disabled, or the column nested under struct/array/map) now
succeeds instead of throwing `FAILED_READ_FILE`. The vectorized read path is
unchanged.
### How was this patch tested?
- `TimeTypeParquetOpsSuite`: flipped the `isAdjustedToUTC=true` case from
reject to accept; the genuine mis-decode rejections are retained
(8/8 pass).
- `ParquetIOSuite`: new end-to-end test reading an `INT64 TIME(MICROS,
isAdjustedToUTC=true)` column as `TimeType`, asserting correct values
on both the vectorized and row-based readers (via `withAllParquetReaders`).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56610 from stevomitric/stevomitric/fix-time.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
---
.../parquet/types/ops/TimeTypeParquetOps.scala | 22 ++++++----
.../datasources/parquet/ParquetIOSuite.scala | 39 ++++++++++++++++++
.../types/ops/TimeTypeParquetOpsSuite.scala | 48 +++++++++++-----------
3 files changed, 79 insertions(+), 30 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
index 7f05361d8f6c..96c7bc30a1da 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOps.scala
@@ -93,12 +93,14 @@ case class TimeTypeParquetOps(t: TimeType) extends
ParquetTypeOps {
private[ops] object TimeTypeParquetOps {
/**
- * Validates that a Parquet field can be decoded as TimeType. TimeType is
stored
- * as INT64 with TIME(MICROS, isAdjustedToUTC=false). Any other encoding (raw
- * INT64, INT64 TIME(NANOS), INT32 TIME(MILLIS), INT64 TIMESTAMP(_), decimal-
- * annotated, etc.) cannot be decoded as TimeType - throw the same error as
- * the legacy ParquetRowConverter path so reads fail loudly instead of
- * silently misinterpreting bytes.
+ * Validates that a Parquet field can be decoded as TimeType. TimeType is
written
+ * as INT64 with TIME(MICROS, isAdjustedToUTC=false). On read, any INT64
TIME(MICROS)
+ * column is accepted regardless of the isAdjustedToUTC flag: Spark's
zone-less TimeType
+ * decodes the raw micros-of-day identically either way, matching the legacy
+ * ParquetRowConverter guard (see SPARK-57416). Any other encoding (raw
INT64, INT64
+ * TIME(NANOS), INT32 TIME(MILLIS), INT64 TIMESTAMP(_), decimal-annotated,
etc.) cannot
+ * be decoded as TimeType - throw the same error as the legacy
ParquetRowConverter path
+ * so reads fail loudly instead of silently misinterpreting bytes.
*/
private[ops] def requireCompatibleParquetType(
sparkType: TimeType, parquetType: Type): Unit = {
@@ -106,7 +108,13 @@ private[ops] object TimeTypeParquetOps {
parquetType.asPrimitiveType.getPrimitiveTypeName == INT64 &&
(parquetType.getLogicalTypeAnnotation match {
case t: LogicalTypeAnnotation.TimeLogicalTypeAnnotation =>
- t.getUnit == TimeUnit.MICROS && !t.isAdjustedToUTC
+ // Accept both isAdjustedToUTC=false and =true. Spark's TimeType is
zone-less
+ // local time, so the UTC-adjustment flag carries no extra
information on read:
+ // the raw micros-of-day value decodes identically either way.
Mirroring the
+ // legacy ParquetRowConverter guard (which only checked the TIME
annotation and
+ // the MICROS unit) keeps the framework row-based read path
consistent with both
+ // the legacy row-based reader and the still-lenient vectorized
reader. SPARK-57416.
+ t.getUnit == TimeUnit.MICROS
case _ => false
})
if (!ok) {
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 0fc32b15d833..3a06b59b6a18 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
@@ -1888,6 +1888,45 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
}
}
+ test("SPARK-57416: read INT64 TIME(MICROS, isAdjustedToUTC=true) as TimeType
" +
+ "on both readers") {
+ // The Parquet TIME logical type may carry isAdjustedToUTC=true. Spark's
TimeType is
+ // zone-less local time, so such a column is decoded as the raw
micros-of-day, exactly
+ // as isAdjustedToUTC=false. Before SPARK-57416 the Types Framework
row-based read guard
+ // rejected this encoding (FAILED_READ_FILE) while the vectorized path
accepted it,
+ // leaving the two readers inconsistent. This pins that both readers now
accept it.
+ // The column must be supplied via an explicit read schema because schema
inference
+ // maps isAdjustedToUTC=true to an unsupported type.
+ val schema = MessageTypeParser.parseMessageType(
+ """message root {
+ | required int64 time_micros(TIME(MICROS,true));
+ |}""".stripMargin)
+ val readSchema = new StructType().add("time_micros", TimeType())
+
+ for (dictEnabled <- Seq(true, false)) {
+ withTempDir { dir =>
+ val tablePath = new Path(s"${dir.getCanonicalPath}/times_utc.parquet")
+ val numRecords = 100
+
+ val writer = createParquetWriter(schema, tablePath, dictionaryEnabled
= dictEnabled)
+ (0 until numRecords).foreach { _ =>
+ val record = new SimpleGroup(schema)
+ record.add(0, localTime(23, 59, 59, 123456) /
DateTimeConstants.NANOS_PER_MICROS)
+ writer.write(record)
+ }
+ writer.close
+
+ withAllParquetReaders {
+ val df = spark.read.schema(readSchema).parquet(tablePath.toString)
+ assertResult(df.schema)(readSchema)
+ val lt = LocalTime.of(23, 59, 59, 123456000)
+ val expected = (0 until numRecords).map { _ => lt }.toDF()
+ checkAnswer(df, expected)
+ }
+ }
+ }
+ }
+
// Deterministic INT32 sample shared by the INT32 widening tests below.
Mixes sign,
// zero, and MIN/MAX boundaries to catch sign-extension and precision
regressions.
private def widenSampleAt(i: Int): Int = i % 5 match {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
index 9a21b5e3f4bb..36221cec8b2c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimeTypeParquetOpsSuite.scala
@@ -28,17 +28,18 @@ import org.apache.spark.sql.types.TimeType
/**
* Unit tests for [[TimeTypeParquetOps.requireCompatibleParquetType]].
*
- * TimeType is stored in Parquet as INT64 TIME(MICROS, isAdjustedToUTC=false).
- * The read-path guard accepts only that canonical encoding and rejects every
- * other primitive/annotation combination so that reading fails loudly rather
- * than silently mis-decoding (e.g. interpreting NANOS as MICROS, which would
- * be off by 1000x).
+ * TimeType is written to Parquet as INT64 TIME(MICROS,
isAdjustedToUTC=false). The
+ * read-path guard accepts any INT64 TIME(MICROS) column - both
isAdjustedToUTC values -
+ * and rejects every other primitive/annotation combination so that reading
fails loudly
+ * rather than silently mis-decoding (e.g. interpreting NANOS as MICROS, which
would be
+ * off by 1000x).
*
- * Note: rejecting isAdjustedToUTC=true is stricter than the legacy
- * ParquetRowConverter guard, which accepts that encoding. This is a known,
- * intentional divergence between the framework and legacy paths for this
- * single case; reconciling it (either by relaxing the framework guard or
- * tightening the legacy one) is tracked by SPARK-57416.
+ * SPARK-57416: the guard accepts isAdjustedToUTC=true to mirror the legacy
+ * ParquetRowConverter guard (which only checks the TIME annotation and the
MICROS unit).
+ * Spark's TimeType is zone-less local time, so the flag carries no extra
information on
+ * read and the raw micros-of-day value decodes identically either way. This
keeps the
+ * framework read path consistent with both the legacy row-based reader and
the vectorized
+ * reader.
*/
class TimeTypeParquetOpsSuite extends SparkFunSuite {
@@ -54,7 +55,20 @@ class TimeTypeParquetOpsSuite extends SparkFunSuite {
TimeTypeParquetOps.requireCompatibleParquetType(timeMicros, field)
}
- // ---------- the four primary reject paths ----------
+ test("accepts INT64 TIME(MICROS, isAdjustedToUTC=true) - matches legacy
lenient read") {
+ // SPARK-57416: the framework read guard mirrors the legacy
ParquetRowConverter guard,
+ // which accepts INT64 TIME(MICROS) regardless of isAdjustedToUTC. Spark's
TimeType is
+ // zone-less, so the raw micros-of-day value decodes identically either
way; rejecting
+ // this encoding would diverge from both the legacy row-based reader and
the (lenient)
+ // vectorized reader.
+ val field = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timeType(true, TimeUnit.MICROS))
+ .named("c")
+ // Must not throw.
+ TimeTypeParquetOps.requireCompatibleParquetType(timeMicros, field)
+ }
+
+ // ---------- the primary reject paths ----------
test("rejects raw INT64 with no logical type annotation") {
val field = Types.primitive(INT64, REQUIRED).named("c")
@@ -76,18 +90,6 @@ class TimeTypeParquetOpsSuite extends SparkFunSuite {
assertRejects(timeMicros, field)
}
- test("rejects INT64 TIME(MICROS, isAdjustedToUTC=true)") {
- // The intended framework behavior is to reject this encoding: the
canonical
- // TimeType representation is local-time (isAdjustedToUTC=false). The
legacy
- // ParquetRowConverter guard accepts the encoding, so this is a known,
- // intentional framework-vs-legacy divergence; reconciliation is tracked by
- // SPARK-57416.
- val field = Types.primitive(INT64, REQUIRED)
- .as(LogicalTypeAnnotation.timeType(true, TimeUnit.MICROS))
- .named("c")
- assertRejects(timeMicros, field)
- }
-
// ---------- additional rejects for full reject-set coverage ----------
test("rejects INT64 TIMESTAMP(MICROS) - wrong annotation kind") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
