This is an automated email from the ASF dual-hosted git repository.
MaxGekk 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 7c81eb6cb17c [SPARK-57661][SQL] Preserve TIME precision in the Spark
<-> Arrow type mapping
7c81eb6cb17c is described below
commit 7c81eb6cb17cd267fe481e5733ea13b27264b34e
Author: Maxim Gekk <[email protected]>
AuthorDate: Fri Jun 26 08:10:50 2026 +0200
[SPARK-57661][SQL] Preserve TIME precision in the Spark <-> Arrow type
mapping
### What changes were proposed in this pull request?
This PR carries the `TimeType(p)` fractional-second precision `p` (in `[0,
9]`) across the Spark <-> Arrow type mapping so that a `TIME(p)` column
round-trips back to the same `TIME(p)`, instead of collapsing to the canonical
`TIME(6)`.
Arrow's `Time` logical type encodes only `(unit, bitWidth)` and has no
fractional-precision field, so the precision cannot live in the `ArrowType`
itself. It is instead carried in the Arrow field metadata under a dedicated key
`SPARK::time::precision`, reusing the precision-in-field-metadata pattern
introduced for the nanosecond timestamp types (SPARK-57159).
- `ArrowUtils.toArrowField`: tag `TimeType(p)` fields with the precision
metadata key, merged with the column metadata. The Arrow type stays
`Time(NANOSECOND, 64)`.
- `ArrowUtils.fromArrowField`: read that key to reconstruct `TimeType(p)`;
when the key is absent (foreign Arrow data) or out of `[0, 9]`, fall back to
the canonical `TimeType(MICROS_PRECISION)` (= 6) via `fromArrowType`,
preserving today's behavior for non-Spark producers.
- The shared precision-stashing helper `toTimestampNanosArrowField` is
generalized to `toPrecisionTaggedArrowField`, parameterized by the metadata
key, so the nanosecond timestamp types and `TIME` share it.
`TimeTypeApiOps.toArrowType` and `TypeApiOps.fromArrowType` are unchanged:
`toArrowType` keeps producing `Time(NANOSECOND, 64)`, and the metadata-less
`fromArrowType` remains the canonical `TIME(6)` fallback.
### Why are the changes needed?
`ArrowUtils` / the Types Framework currently map every `TimeType(p)` to
`ArrowType.Time(NANOSECOND, 64)` (no precision field), and
`TypeApiOps.fromArrowType` maps it back to a fixed `TimeType(6)`. As a result
the declared precision is lost on any Arrow round-trip (`TIME(0)`, `TIME(3)`,
`TIME(9)`, ... all read back as `TIME(6)`), so Arrow-based schema transfer
(Connect schema/results, `createDataFrame` from Arrow, `mapInArrow`, etc.)
silently widens or narrows the type label. The store [...]
### Does this PR introduce _any_ user-facing change?
No. The TIME data type is gated behind the internal flag
`spark.sql.timeType.enabled`, which defaults to `Utils.isTesting` and so is off
by default in production. With the flag enabled, the behavior improves: a
`TIME(p)` column transferred over Arrow retains its declared precision instead
of always reading back as `TIME(6)`. No change to stored values.
### How was this patch tested?
Added `test("time")` to `ArrowUtilsSuite`: round-trip `TIME(p)` for `p` in
`{0, 3, 6, 9}` preserves `p` (and the Arrow field stays `Time(NANOSECOND,
64)`); a `Time(NANOSECOND)` field with no precision metadata, or with a
present-but-invalid precision (out of `[0, 9]` or non-numeric), falls back to
`TIME(6)`; and the precision key does not leak into the reconstructed column
`Metadata`. Run with `build/sbt 'catalyst/testOnly *ArrowUtilsSuite'` (8 tests
pass).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor (Claude Opus 4.8)
Closes #56778 from MaxGekk/time-arrow-precision.
Authored-by: Maxim Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit ca0629f61e219a2550950d0b44368d826e5a717a)
Signed-off-by: Max Gekk <[email protected]>
---
.../org/apache/spark/sql/util/ArrowUtils.scala | 44 ++++++++++++++++++----
.../apache/spark/sql/util/ArrowUtilsSuite.scala | 42 +++++++++++++++++++++
.../sql/execution/arrow/ArrowConvertersSuite.scala | 6 ++-
3 files changed, 84 insertions(+), 8 deletions(-)
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
b/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
index 15cf5b23e4ac..a06a77d9d113 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
@@ -115,6 +115,11 @@ private[sql] object ArrowUtils {
// metadata under this dedicated key (namespaced like `metadataKey`,
separate from the user
// metadata blob so user metadata is untouched) and recovered on read in
`fromArrowField`.
private val timestampNanosPrecisionKey = "SPARK::timestampNanos::precision"
+ // Arrow's Time type carries only (unit, bitWidth) and has no
fractional-second precision field,
+ // so the precision of TimeType is stored in the Arrow field metadata under
this dedicated key
+ // (namespaced like `metadataKey`, separate from the user metadata blob so
user metadata is
+ // untouched) and recovered on read in `fromArrowField`.
+ private val timePrecisionKey = "SPARK::time::precision"
private def toArrowMetaData(metadata: Metadata) = {
if (metadata != null && !metadata.isEmpty) {
Map(metadataKey -> metadata.json).asJava
@@ -131,19 +136,22 @@ private[sql] object ArrowUtils {
}
/**
- * Builds an Arrow field for a nanosecond timestamp type, stashing the
column precision in the
- * field metadata (alongside the user metadata) so it can be recovered in
`fromArrowField`.
+ * Builds an Arrow field for a type whose Arrow representation cannot encode
its
+ * fractional-second precision (nanosecond timestamps, TIME), stashing the
column precision in
+ * the field metadata under `precisionKey` (alongside the user metadata) so
it can be recovered
+ * in `fromArrowField`.
*/
- private def toTimestampNanosArrowField(
+ private def toPrecisionTaggedArrowField(
name: String,
dt: DataType,
precision: Int,
+ precisionKey: String,
nullable: Boolean,
timeZoneId: String,
largeVarTypes: Boolean,
metadata: Metadata): Field = {
val base =
Option(toArrowMetaData(metadata)).map(_.asScala.toMap).getOrElse(Map.empty)
- val md = (base + (timestampNanosPrecisionKey -> precision.toString)).asJava
+ val md = (base + (precisionKey -> precision.toString)).asJava
val fieldType = new FieldType(nullable, toArrowType(dt, timeZoneId,
largeVarTypes), null, md)
new Field(name, fieldType, Seq.empty[Field].asJava)
}
@@ -255,19 +263,31 @@ private[sql] object ArrowUtils {
toArrowField("value", BinaryType, false, timeZoneId,
largeVarTypes),
new Field("metadata", metadataFieldType,
Seq.empty[Field].asJava)).asJava)
case t: TimestampNTZNanosType =>
- toTimestampNanosArrowField(
+ toPrecisionTaggedArrowField(
name,
t,
t.precision,
+ timestampNanosPrecisionKey,
nullable,
timeZoneId,
largeVarTypes,
metadata)
case t: TimestampLTZNanosType =>
- toTimestampNanosArrowField(
+ toPrecisionTaggedArrowField(
name,
t,
t.precision,
+ timestampNanosPrecisionKey,
+ nullable,
+ timeZoneId,
+ largeVarTypes,
+ metadata)
+ case t: TimeType =>
+ toPrecisionTaggedArrowField(
+ name,
+ t,
+ t.precision,
+ timePrecisionKey,
nullable,
timeZoneId,
largeVarTypes,
@@ -352,7 +372,7 @@ private[sql] object ArrowUtils {
}
StructType(fields.toArray)
// Recover the exact precision of nanosecond timestamps from the field
metadata written by
- // `toTimestampNanosArrowField`. Foreign Arrow data (or an out-of-range
value) has no usable
+ // `toPrecisionTaggedArrowField`. Foreign Arrow data (or an out-of-range
value) has no usable
// key, so fall back to the canonical maximum precision via
`fromArrowType`.
case ts: ArrowType.Timestamp if ts.getUnit == TimeUnit.NANOSECOND =>
val precision =
Option(field.getMetadata.get(timestampNanosPrecisionKey))
@@ -365,6 +385,16 @@ private[sql] object ArrowUtils {
case Some(p) => TimestampLTZNanosType(p)
case None => fromArrowType(ts)
}
+ // Recover the exact precision of TIME from the field metadata written
by `toArrowField`.
+ // Foreign Arrow data has no precision key, and a present-but-invalid
value (out of [0, 9] or
+ // non-numeric) is unusable, so either way fall back to the canonical
microsecond precision
+ // via `fromArrowType`.
+ case t: ArrowType.Time if t.getUnit == TimeUnit.NANOSECOND =>
+ Option(field.getMetadata.get(timePrecisionKey))
+ .flatMap(s => scala.util.Try(s.toInt).toOption)
+ .filter(p => p >= TimeType.MIN_PRECISION && p <=
TimeType.MAX_PRECISION)
+ .map(TimeType(_))
+ .getOrElse(fromArrowType(t))
case arrowType => fromArrowType(arrowType)
}
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/util/ArrowUtilsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/util/ArrowUtilsSuite.scala
index 16682f981633..2d2186aed85c 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/util/ArrowUtilsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/util/ArrowUtilsSuite.scala
@@ -154,6 +154,48 @@ class ArrowUtilsSuite extends SparkFunSuite {
ArrowUtils.toArrowSchema(schemaWithMeta, null, true, false)) ===
schemaWithMeta)
}
+ test("time") {
+ // Arrow's Time type has no precision field, so TIME(p) precision is
preserved via field
+ // metadata; the Arrow type itself stays Time(NANOSECOND, 64).
+ Seq(0, 3, 6, 7, 9).foreach { p =>
+ val schema = new StructType().add("value", TimeType(p))
+ val arrowSchema = ArrowUtils.toArrowSchema(schema, null, true, false)
+ val fieldType =
arrowSchema.findField("value").getType.asInstanceOf[ArrowType.Time]
+ assert(fieldType.getUnit === TimeUnit.NANOSECOND)
+ assert(fieldType.getBitWidth === 8 * 8)
+ assert(ArrowUtils.fromArrowSchema(arrowSchema) === schema)
+ }
+
+ // Fallback: a nanosecond Arrow time without precision metadata maps to
canonical TIME(6).
+ def timeField: Field = new Field(
+ "value",
+ new FieldType(true, new ArrowType.Time(TimeUnit.NANOSECOND, 8 * 8),
null, null),
+ java.util.Collections.emptyList[Field]())
+ assert(ArrowUtils.fromArrowField(timeField) ===
TimeType(TimeType.MICROS_PRECISION))
+
+ // Fallback also covers a present-but-invalid precision key (out of [0, 9]
or non-numeric):
+ // the value is unusable, so the type maps to the canonical TIME(6) just
like the no-metadata
+ // case.
+ def timeFieldWithPrecision(precision: String): Field = new Field(
+ "value",
+ new FieldType(
+ true,
+ new ArrowType.Time(TimeUnit.NANOSECOND, 8 * 8),
+ null,
+ java.util.Collections.singletonMap("SPARK::time::precision",
precision)),
+ java.util.Collections.emptyList[Field]())
+ val micros = TimeType(TimeType.MICROS_PRECISION)
+ assert(ArrowUtils.fromArrowField(timeFieldWithPrecision("-1")) === micros)
+ assert(ArrowUtils.fromArrowField(timeFieldWithPrecision("10")) === micros)
+ assert(ArrowUtils.fromArrowField(timeFieldWithPrecision("x")) === micros)
+
+ // The precision metadata key does not leak into the reconstructed column
Metadata.
+ val md = new MetadataBuilder().putString("city", "beijing").build()
+ val schemaWithMeta = new StructType().add("value", TimeType(3), nullable =
true, md)
+ assert(ArrowUtils.fromArrowSchema(
+ ArrowUtils.toArrowSchema(schemaWithMeta, null, true, false)) ===
schemaWithMeta)
+ }
+
test("array") {
roundtrip(ArrayType(IntegerType, containsNull = true))
roundtrip(ArrayType(IntegerType, containsNull = false))
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowConvertersSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowConvertersSuite.scala
index e3d5e2e4c5a2..a8fdf210ef95 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowConvertersSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowConvertersSuite.scala
@@ -744,7 +744,11 @@ class ArrowConvertersSuite extends SharedSparkSession {
| "bitWidth" : 64
| },
| "nullable" : true,
- | "children" : [ ]
+ | "children" : [ ],
+ | "metadata" : [ {
+ | "key" : "SPARK::time::precision",
+ | "value" : "6"
+ | } ]
| } ]
| },
| "batches" : [ {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
