This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new f223a168a0b4 [SPARK-57581][4.2][SQL] Encode the TIME data type in Avro
with a unit-correct logical type
f223a168a0b4 is described below
commit f223a168a0b46dad9af117a5b01f99457abb2357
Author: Maxim Gekk <[email protected]>
AuthorDate: Mon Jun 22 14:40:02 2026 +0200
[SPARK-57581][4.2][SQL] Encode the TIME data type in Avro with a
unit-correct logical type
### What changes were proposed in this pull request?
This is a backport of #56633 (master commit
24e0663709335f5a1d8d5d3e0d45658bec3259f2) to `branch-4.2`.
It makes Spark's Avro encoding of the `TIME` data type unit-correct.
`TimeType` is represented internally as nanoseconds-since-midnight, but the
Avro path annotated the column with the `time-micros` logical type while
writing the raw nanosecond value, so the declared unit (microseconds) did not
match the stored unit (nanoseconds).
The fix converts the value to match the logical type:
- Write path (`AvroSerializer`): `nanos -> micros`
(`DateTimeUtils.nanosToMicros`) before writing under `time-micros`.
- Read path (`AvroDeserializer`): `micros -> nanos`
(`DateTimeUtils.microsToNanos`) when reading a `time-micros` value into
`TimeType`.
`SchemaConverters` is unchanged: `time-micros` is the correct unit-matching
logical type for precision 0-6, and the `spark.sql.catalyst.type` property
continues to carry precision fidelity for Spark-to-Spark round-trips.
Backport note: the production change applied cleanly. The only cherry-pick
conflict was in `AvroSuite.scala`, because the master change places the TIME
tests next to the `SPARK-57166: nanosecond timestamp types are not supported in
Avro` test, which does not exist on `branch-4.2` (nanosecond timestamp types
are a master/`branch-4.x` feature). Resolved by placing the five TIME tests in
the base `AvroSuite` (so they run under both `AvroV1Suite` and `AvroV2Suite`)
and omitting the unrela [...]
### Why are the changes needed?
Any external Avro reader (Hive, Trino, Flink, fastavro, etc.) that honors
the `time-micros` logical type would decode a Spark-written `TIME` column as
microseconds-since-midnight while it actually held nanoseconds-since-midnight -
a 1000x error that also falls outside the valid micros-of-day range. The
TIME-in-Avro support is present in the 4.2 line (SPARK-54473), so the bug needs
to be fixed before 4.2.0 GA.
### Does this PR introduce _any_ user-facing change?
Yes, within the unreleased 4.2.0 line. The on-disk encoding of a `TIME`
column written via Avro changes from raw nanoseconds (mislabeled as
`time-micros`) to actual microseconds under `time-micros`. Avro files written
by earlier unreleased 4.2 builds are intentionally not migrated (accepted
break). Spark-to-Spark read/write of `TIME` over Avro continues to round-trip
correctly.
### How was this patch tested?
Ran the TIME Avro tests on this `branch-4.2` backport:
`AvroV1Suite`/`AvroV2Suite` ("TIME type read/write with Avro format", "TIME
type in nested structures in Avro", "TIME type precision metadata is preserved
in Avro", "SPARK-57581: TIME is written as unit-correct time-micros for
external readers", "SPARK-57581: TIME read from a plain time-micros Avro file
(no catalyst prop)") - 10 tests (5 x V1/V2), all pass. `dev/scalastyle` is
clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor (Claude Opus 4.8)
Closes #56651 from MaxGekk/fix-time-avro-4.2.
Authored-by: Maxim Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
---
.../org/apache/spark/sql/avro/AvroSuite.scala | 265 ++++++++++++++-------
.../apache/spark/sql/avro/AvroDeserializer.scala | 4 +-
.../org/apache/spark/sql/avro/AvroSerializer.scala | 6 +-
3 files changed, 183 insertions(+), 92 deletions(-)
diff --git
a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
index 56b107c14f57..376d12a8a492 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
@@ -3191,6 +3191,181 @@ abstract class AvroSuite
}
}
+ test("TIME type read/write with Avro format") {
+ withTempPath { dir =>
+ // Test boundary values and NULL handling
+ val df = spark.sql("""
+ SELECT
+ TIME'00:00:00.123456' as midnight,
+ TIME'12:34:56.789012' as noon,
+ TIME'23:59:59.999999' as max_time,
+ CAST(NULL AS TIME) as null_time
+ """)
+
+ df.write.format("avro").save(dir.toString)
+ val readDf = spark.read.format("avro").load(dir.toString)
+
+ checkAnswer(readDf, df)
+
+ // Verify schema - all should be default TimeType(6)
+ readDf.schema.fields.foreach { field =>
+ assert(field.dataType == TimeType(), s"Field ${field.name} should be
TimeType")
+ }
+
+ // Verify boundary values
+ val row = readDf.collect()(0)
+ assert(row.getAs[java.time.LocalTime]("midnight") ==
+ java.time.LocalTime.of(0, 0, 0, 123456000))
+ assert(row.getAs[java.time.LocalTime]("noon") ==
+ java.time.LocalTime.of(12, 34, 56, 789012000))
+ assert(row.getAs[java.time.LocalTime]("max_time") ==
+ java.time.LocalTime.of(23, 59, 59, 999999000))
+ assert(row.get(3) == null, "NULL time should be preserved")
+ }
+ }
+
+ test("TIME type in nested structures in Avro") {
+ withTempPath { dir =>
+ // Test TIME type in arrays and structs with different precisions
+ val df = spark.sql("""
+ SELECT
+ named_struct('start', CAST(TIME'09:00:00.123' AS TIME(3)),
+ 'end', CAST(TIME'17:30:45.654321' AS TIME(6))) as
schedule,
+ array(TIME'08:15:30.111222', TIME'12:45:15.333444',
TIME'16:20:50.555666') as checkpoints
+ """)
+
+ df.write.format("avro").save(dir.toString)
+ val readDf = spark.read.format("avro").load(dir.toString)
+
+ checkAnswer(readDf, df)
+ }
+ }
+
+ test("TIME type precision metadata is preserved in Avro") {
+ withTempPath { dir =>
+ // Test all TIME precisions (0-6) with multiple columns
+ val df = spark.sql("""
+ SELECT
+ id,
+ CAST(TIME '12:34:56' AS TIME(0)) as time_p0,
+ CAST(TIME '12:34:56.1' AS TIME(1)) as time_p1,
+ CAST(TIME '12:34:56.12' AS TIME(2)) as time_p2,
+ CAST(TIME '12:34:56.123' AS TIME(3)) as time_p3,
+ CAST(TIME '12:34:56.1234' AS TIME(4)) as time_p4,
+ CAST(TIME '12:34:56.12345' AS TIME(5)) as time_p5,
+ CAST(TIME '12:34:56.123456' AS TIME(6)) as time_p6,
+ description
+ FROM VALUES
+ (1, 'Morning'),
+ (2, 'Evening')
+ AS t(id, description)
+ """)
+
+ // Verify original schema has all precisions
+ (0 to 6).foreach { p =>
+ assert(df.schema(s"time_p$p").dataType == TimeType(p))
+ }
+
+ // Write to Avro and read back
+ df.write.format("avro").save(dir.toString)
+ val readDf = spark.read.format("avro").load(dir.toString)
+
+ // Verify ALL precisions are preserved after round-trip
+ (0 to 6).foreach { p =>
+ assert(readDf.schema(s"time_p$p").dataType == TimeType(p),
+ s"Precision $p should be preserved")
+ }
+
+ // Verify data integrity
+ checkAnswer(readDf, df)
+ }
+ }
+
+ test("SPARK-57581: TIME is written as unit-correct time-micros for external
readers") {
+ // Expected microseconds-since-midnight for TIME'12:34:56.123456'
truncated to each precision.
+ val baseSeconds = (12 * 3600 + 34 * 60 + 56).toLong
+ val expectedMicros = Map(
+ 0 -> (baseSeconds * 1000000L + 0L),
+ 1 -> (baseSeconds * 1000000L + 100000L),
+ 2 -> (baseSeconds * 1000000L + 120000L),
+ 3 -> (baseSeconds * 1000000L + 123000L),
+ 4 -> (baseSeconds * 1000000L + 123400L),
+ 5 -> (baseSeconds * 1000000L + 123450L),
+ 6 -> (baseSeconds * 1000000L + 123456L))
+ // Valid micros-of-day range; values mislabeled as micros but holding
nanos would exceed this.
+ val microsPerDay = 24L * 3600L * 1000000L
+
+ (0 to 6).foreach { p =>
+ withTempPath { dir =>
+ spark.sql(s"SELECT CAST(TIME'12:34:56.123456' AS TIME($p)) as t")
+ .write.format("avro").save(dir.toString)
+
+ val avroFile = dir.listFiles()
+ .filter(f => f.isFile && f.getName.endsWith("avro"))
+ .head
+ val reader = new DataFileReader[GenericRecord](
+ avroFile, new GenericDatumReader[GenericRecord]())
+ try {
+ // The Avro field must be annotated with the time-micros logical
type.
+ val fieldSchema = reader.getSchema.getField("t").schema()
+ val timeSchema = if (fieldSchema.getType == Type.UNION) {
+ fieldSchema.getTypes.asScala.find(_.getType == Type.LONG).get
+ } else {
+ fieldSchema
+ }
+ assert(timeSchema.getLogicalType.getName == "time-micros",
+ s"precision $p should be written as time-micros")
+
+ assert(reader.hasNext)
+ val record = reader.next()
+ val stored = record.get("t").asInstanceOf[Long]
+ assert(stored == expectedMicros(p),
+ s"precision $p should store micros-of-day ${expectedMicros(p)},
but was $stored")
+ assert(stored >= 0 && stored < microsPerDay,
+ s"precision $p stored value $stored is outside the valid
micros-of-day range")
+ } finally {
+ reader.close()
+ }
+ }
+ }
+ }
+
+ test("SPARK-57581: TIME read from a plain time-micros Avro file (no catalyst
prop)") {
+ withTempDir { dir =>
+ // Build an Avro file the way an external tool (Hive/Trino/fastavro)
would: a `time-micros`
+ // long with no `spark.sql.catalyst.type` property. Spark must read it
back as TIME,
+ // converting the stored microseconds-since-midnight to its internal
nanoseconds and
+ // defaulting to the micros precision TIME(6). This pins the
deserializer's micros -> nanos
+ // conversion independently of the write path.
+ val micros = (12L * 3600 + 34 * 60 + 56) * 1000000L + 123456L
+ val avroSchema = new Schema.Parser().parse(
+ """
+ |{
+ | "type": "record",
+ | "name": "top",
+ | "fields": [
+ | {"name": "t", "type": {"type": "long", "logicalType":
"time-micros"}}
+ | ]
+ |}
+ """.stripMargin)
+ val avroFile = new File(dir, "external.avro")
+ val datumWriter = new GenericDatumWriter[GenericRecord](avroSchema)
+ val dataFileWriter = new DataFileWriter[GenericRecord](datumWriter)
+ dataFileWriter.create(avroSchema, avroFile)
+ try {
+ val record = new GenericData.Record(avroSchema)
+ record.put("t", micros)
+ dataFileWriter.append(record)
+ } finally {
+ dataFileWriter.close()
+ }
+
+ val readDf = spark.read.format("avro").load(dir.toString)
+ assert(readDf.schema("t").dataType ==
TimeType(TimeType.MICROS_PRECISION))
+ checkAnswer(readDf, Row(java.time.LocalTime.of(12, 34, 56, 123456000)))
+ }
+ }
+
}
class AvroV1Suite extends AvroSuite {
@@ -3392,96 +3567,6 @@ class AvroV2Suite extends AvroSuite with
ExplainSuiteHelper {
}
}
- test("TIME type read/write with Avro format") {
- withTempPath { dir =>
- // Test boundary values and NULL handling
- val df = spark.sql("""
- SELECT
- TIME'00:00:00.123456' as midnight,
- TIME'12:34:56.789012' as noon,
- TIME'23:59:59.999999' as max_time,
- CAST(NULL AS TIME) as null_time
- """)
-
- df.write.format("avro").save(dir.toString)
- val readDf = spark.read.format("avro").load(dir.toString)
-
- checkAnswer(readDf, df)
-
- // Verify schema - all should be default TimeType(6)
- readDf.schema.fields.foreach { field =>
- assert(field.dataType == TimeType(), s"Field ${field.name} should be
TimeType")
- }
-
- // Verify boundary values
- val row = readDf.collect()(0)
- assert(row.getAs[java.time.LocalTime]("midnight") ==
- java.time.LocalTime.of(0, 0, 0, 123456000))
- assert(row.getAs[java.time.LocalTime]("noon") ==
- java.time.LocalTime.of(12, 34, 56, 789012000))
- assert(row.getAs[java.time.LocalTime]("max_time") ==
- java.time.LocalTime.of(23, 59, 59, 999999000))
- assert(row.get(3) == null, "NULL time should be preserved")
- }
- }
-
- test("TIME type in nested structures in Avro") {
- withTempPath { dir =>
- // Test TIME type in arrays and structs with different precisions
- val df = spark.sql("""
- SELECT
- named_struct('start', CAST(TIME'09:00:00.123' AS TIME(3)),
- 'end', CAST(TIME'17:30:45.654321' AS TIME(6))) as
schedule,
- array(TIME'08:15:30.111222', TIME'12:45:15.333444',
TIME'16:20:50.555666') as checkpoints
- """)
-
- df.write.format("avro").save(dir.toString)
- val readDf = spark.read.format("avro").load(dir.toString)
-
- checkAnswer(readDf, df)
- }
- }
-
- test("TIME type precision metadata is preserved in Avro") {
- withTempPath { dir =>
- // Test all TIME precisions (0-6) with multiple columns
- val df = spark.sql("""
- SELECT
- id,
- CAST(TIME '12:34:56' AS TIME(0)) as time_p0,
- CAST(TIME '12:34:56.1' AS TIME(1)) as time_p1,
- CAST(TIME '12:34:56.12' AS TIME(2)) as time_p2,
- CAST(TIME '12:34:56.123' AS TIME(3)) as time_p3,
- CAST(TIME '12:34:56.1234' AS TIME(4)) as time_p4,
- CAST(TIME '12:34:56.12345' AS TIME(5)) as time_p5,
- CAST(TIME '12:34:56.123456' AS TIME(6)) as time_p6,
- description
- FROM VALUES
- (1, 'Morning'),
- (2, 'Evening')
- AS t(id, description)
- """)
-
- // Verify original schema has all precisions
- (0 to 6).foreach { p =>
- assert(df.schema(s"time_p$p").dataType == TimeType(p))
- }
-
- // Write to Avro and read back
- df.write.format("avro").save(dir.toString)
- val readDf = spark.read.format("avro").load(dir.toString)
-
- // Verify ALL precisions are preserved after round-trip
- (0 to 6).foreach { p =>
- assert(readDf.schema(s"time_p$p").dataType == TimeType(p),
- s"Precision $p should be preserved")
- }
-
- // Verify data integrity
- checkAnswer(readDf, df)
- }
- }
-
test("SPARK-56457: Avro V2 formatName matches V1 FileFormat.toString") {
val v2Provider = DataSource.lookupDataSourceV2("avro",
spark.sessionState.conf)
assert(v2Provider.isDefined)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
index f61e6da8583e..ba574c091dae 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
@@ -204,9 +204,11 @@ private[sql] class AvroDeserializer(
}
case (LONG, _: TimeType) => avroType.getLogicalType match {
+ // The time-micros logical type stores microseconds-since-midnight,
while TimeType
+ // is represented internally as nanoseconds-since-midnight, so convert
micros to nanos.
case _: LogicalTypes.TimeMicros => (updater, ordinal, value) =>
val micros = value.asInstanceOf[Long]
- updater.setLong(ordinal, micros)
+ updater.setLong(ordinal, DateTimeUtils.microsToNanos(micros))
case other => throw new IncompatibleSchemaException(errorPrefix +
s"Avro logical type $other cannot be converted to SQL type
${TimeType().sql}.")
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
index 533aa6ee09af..aacf8dc9f347 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
@@ -192,7 +192,11 @@ private[sql] class AvroSerializer(
}
case (_: TimeType, LONG) => avroType.getLogicalType match {
- case _: LogicalTypes.TimeMicros => (getter, ordinal) =>
getter.getLong(ordinal)
+ // TimeType is stored internally as nanoseconds-since-midnight. The
time-micros
+ // logical type stores microseconds-since-midnight, so convert nanos
to micros
+ // to keep the on-disk value unit-correct for external Avro readers.
+ case _: LogicalTypes.TimeMicros => (getter, ordinal) =>
+ DateTimeUtils.nanosToMicros(getter.getLong(ordinal))
case other => throw new IncompatibleSchemaException(errorPrefix +
s"SQL type ${TimeType().sql} cannot be converted to Avro logical
type $other")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
