This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR 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 d23987f736e9 [SPARK-58300][SQL][PROTOBUF] Add from_protobuf option to
keep Timestamp/Duration as struct
d23987f736e9 is described below
commit d23987f736e996220082484e558dae3ef844a02c
Author: bhollis-dbx <[email protected]>
AuthorDate: Thu Oct 1 13:09:49 2026 +0900
[SPARK-58300][SQL][PROTOBUF] Add from_protobuf option to keep
Timestamp/Duration as struct
### What changes were proposed in this pull request?
`from_protobuf` always maps `google.protobuf.Timestamp` -> `TimestampType`
and `google.protobuf.Duration` -> `DayTimeIntervalType`. This is Spark's
well-known-type conversion and there is currently no way to turn it off.
This adds a `from_protobuf` reader option
**`convert.timestamp.duration.to.native`** (default `true`, preserving today's
behavior). When set to `false`, both well-known types fall through to the
generic message path and deserialize as their raw proto shape `struct<seconds:
bigint, nanos: int>`.
The two well-known-type special-cases in `SchemaConverters.structFieldFor`
are the only places these types are selected. Guarding both with the flag makes
`Timestamp`/`Duration` fall through to the existing generic `case MESSAGE =>`
-> `StructType` path. The deserializer needs no change: it dispatches on the
catalyst type, so the existing `(MESSAGE, StructType)` case already produces
the struct. The option mirrors the existing `unwrap.primitive.wrapper.types`
precedent.
Scope note: the `to_protobuf` serializer is intentionally not touched --
this is a read-path option only.
### Why are the changes needed?
A consumer whose downstream schema was defined against the `{seconds,
nanos}` struct representation (e.g. a table populated by a different decode
path) and must stay stable, or that needs the exact `seconds`/`nanos` rather
than a microsecond-truncated interval/timestamp, currently has no way to keep
these well-known types in their raw proto shape.
### Does this PR introduce _any_ user-facing change?
Yes, additively. A new `from_protobuf` reader option
`convert.timestamp.duration.to.native` is introduced. Its default is `true`,
which preserves the existing behavior (`google.protobuf.Timestamp` ->
`TimestampType`, `google.protobuf.Duration` -> `DayTimeIntervalType`). Setting
it to `false` makes those well-known types deserialize as `struct<seconds:
bigint, nanos: int>` instead. Default behavior is unchanged, so this is purely
opt-in.
### How was this patch tested?
Added `ProtobufFunctionsSuite` test
*"convert.timestamp.duration.to.native=false keeps Timestamp/Duration as
struct"*, which round-trips a `Timestamp` and a `Duration` field through
`to_protobuf`/`from_protobuf` with the option set to `false` and asserts:
- the deserialized field type is `struct<seconds: bigint, nanos: int>` (not
`TimestampType`/`DayTimeIntervalType`), and
- the values round-trip correctly (the `Duration` `4.5s` -> `seconds=4,
nanos=500000000`; the `Timestamp` seconds are derived from the input so the
assertion is time-zone independent).
Default behavior (`true`) is unchanged, so existing well-known-type tests
continue to exercise the native-type path. Ran the new test and the existing
WKT round-trip tests locally via sbt; all pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57474 from bhollis-dbx/bhollis-dbx/protobuf-wkt-native-option.
Authored-by: bhollis-dbx <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../spark/sql/protobuf/utils/ProtobufOptions.scala | 9 +++
.../sql/protobuf/utils/SchemaConverters.scala | 6 +-
.../sql/protobuf/ProtobufFunctionsSuite.scala | 69 ++++++++++++++++++++++
docs/sql-data-sources-protobuf.md | 6 ++
4 files changed, 88 insertions(+), 2 deletions(-)
diff --git
a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufOptions.scala
b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufOptions.scala
index 9a54b06a9d7f..8732c6489d08 100644
---
a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufOptions.scala
+++
b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufOptions.scala
@@ -240,6 +240,15 @@ private[sql] class ProtobufOptions(
val unwrapWellKnownTypes: Boolean =
getBoolean("unwrap.primitive.wrapper.types", defaultValue = false)
+ // Whether google.protobuf.Timestamp and google.protobuf.Duration are
converted to Spark's
+ // native TimestampType / DayTimeIntervalType. This is the default. Setting
this option to
+ // false leaves them as ordinary messages, so they deserialize as
struct<seconds: bigint,
+ // nanos: int> (the raw proto shape) via the generic message path. Useful
when a downstream
+ // schema was defined against the struct representation and must stay
stable, or when a
+ // consumer needs the raw seconds/nanos rather than a micro-truncated
interval/timestamp.
+ val convertTimestampDurationToNative: Boolean =
+ getBoolean("convert.timestamp.duration.to.native", defaultValue = true)
+
// Since Spark doesn't allow writing empty StructType, empty proto message
type will be
// dropped by default. Setting this option to true will insert a dummy
column to empty proto
// message so that the empty message will be retained.
diff --git
a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/SchemaConverters.scala
b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/SchemaConverters.scala
index 5a73d9aaa72b..4b0a69ab979e 100644
---
a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/SchemaConverters.scala
+++
b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/SchemaConverters.scala
@@ -100,13 +100,15 @@ object SchemaConverters extends Logging {
case BYTE_STRING => Some(BinaryType)
case ENUM => if (protobufOptions.enumsAsInts) Some(IntegerType) else
Some(StringType)
case MESSAGE
- if (fd.getMessageType.getName == "Duration" &&
+ if (protobufOptions.convertTimestampDurationToNative &&
+ fd.getMessageType.getName == "Duration" &&
fd.getMessageType.getFields.size() == 2 &&
fd.getMessageType.getFields.get(0).getName.equals("seconds") &&
fd.getMessageType.getFields.get(1).getName.equals("nanos")) =>
Some(DayTimeIntervalType.defaultConcreteType)
case MESSAGE
- if (fd.getMessageType.getName == "Timestamp" &&
+ if (protobufOptions.convertTimestampDurationToNative &&
+ fd.getMessageType.getName == "Timestamp" &&
fd.getMessageType.getFields.size() == 2 &&
fd.getMessageType.getFields.get(0).getName.equals("seconds") &&
fd.getMessageType.getFields.get(1).getName.equals("nanos")) =>
diff --git
a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufFunctionsSuite.scala
b/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufFunctionsSuite.scala
index f31fbc3bc7bf..9541cda406b3 100644
---
a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufFunctionsSuite.scala
+++
b/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufFunctionsSuite.scala
@@ -725,6 +725,75 @@ class ProtobufFunctionsSuite extends SharedSparkSession
with ProtobufTestBase
}
}
+ test("convert.timestamp.duration.to.native=false keeps Timestamp/Duration as
struct") {
+ // Build a proto payload the normal way (native Timestamp/Duration on the
write side),
+ // then read it back with the conversion disabled and assert both fields
deserialize as
+ // struct<seconds: bigint, nanos: int> -- the raw proto shape -- instead
of TimestampType /
+ // DayTimeIntervalType.
+ val tsSchema = StructType(
+ StructField("timeStampMsg",
+ StructType(
+ StructField("key", StringType, nullable = true) ::
+ StructField("stmp", TimestampType, nullable = true) :: Nil
+ ), nullable = true) :: Nil)
+ val ts = Timestamp.valueOf("2016-05-09 10:12:43.999")
+ val tsInput = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(
+ Row(Row("key1", ts)))),
+ tsSchema)
+
+ val expectedStruct = StructType(
+ StructField("seconds", LongType, nullable = true) ::
+ StructField("nanos", IntegerType, nullable = true) :: Nil)
+
+ checkWithFileAndClassName("timeStampMsg") {
+ case (name, descFilePathOpt) =>
+ val toProtoDf = tsInput
+ .select(to_protobuf_wrapper($"timeStampMsg", name, descFilePathOpt)
as Symbol("to_proto"))
+ val fromProtoDf = toProtoDf
+ .select(from_protobuf_wrapper($"to_proto", name, descFilePathOpt,
+ Map("convert.timestamp.duration.to.native" -> "false")) as
Symbol("timeStampMsg"))
+
+ val stmpField = fromProtoDf.schema("timeStampMsg").dataType
+ .asInstanceOf[StructType]("stmp")
+ assert(stmpField.dataType === expectedStruct)
+ // Derive the expected epoch from the input Timestamp so the assertion
is independent
+ // of the JVM default time zone (Timestamp.valueOf parses in the local
zone).
+ val row = fromProtoDf.select("timeStampMsg.stmp.seconds",
"timeStampMsg.stmp.nanos").first()
+ assert(row.getLong(0) === ts.getTime / 1000)
+ assert(row.getInt(1) === 999000000)
+ }
+
+ val durSchema = StructType(
+ StructField("durationMsg",
+ StructType(
+ StructField("key", StringType, nullable = true) ::
+ StructField("duration", DayTimeIntervalType.defaultConcreteType,
nullable = true) ::
+ Nil
+ ), nullable = true) :: Nil)
+ val durInput = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(
+ Row(Row("key1", Duration.ofSeconds(4).plusNanos(500000000))))),
+ durSchema)
+
+ checkWithFileAndClassName("durationMsg") {
+ case (name, descFilePathOpt) =>
+ val toProtoDf = durInput
+ .select(to_protobuf_wrapper($"durationMsg", name, descFilePathOpt)
as Symbol("to_proto"))
+ val fromProtoDf = toProtoDf
+ .select(from_protobuf_wrapper($"to_proto", name, descFilePathOpt,
+ Map("convert.timestamp.duration.to.native" -> "false")) as
Symbol("durationMsg"))
+
+ val durField = fromProtoDf.schema("durationMsg").dataType
+ .asInstanceOf[StructType]("duration")
+ assert(durField.dataType === expectedStruct)
+ val row = fromProtoDf.select("durationMsg.duration.seconds",
"durationMsg.duration.nanos")
+ .first()
+ assert(row.getLong(0) === 4L)
+ assert(row.getInt(1) === 500000000)
+ }
+ }
+
test("SPARK-57573: Handle TimeType between to_protobuf and from_protobuf") {
val schema = StructType(
StructField("timeMsg",
diff --git a/docs/sql-data-sources-protobuf.md
b/docs/sql-data-sources-protobuf.md
index f5b84454f1b9..7ccdea531fdb 100644
--- a/docs/sql-data-sources-protobuf.md
+++ b/docs/sql-data-sources-protobuf.md
@@ -499,6 +499,12 @@ Data source options of Protobuf can be set via:
<td>Whether to unwrap the struct representation for well-known primitive
wrapper types when deserializing. By default, the wrapper types for primitives
(i.e. google.protobuf.Int32Value, google.protobuf.Int64Value, etc.) will get
deserialized as structs. When unwrapped, an absent singular wrapper is
<code>null</code>, while a present wrapper produces its scalar value or the
scalar's type-specific default, independent of
<code>emit.default.values</code>.</td>
<td>read</td>
</tr>
+ <tr>
+ <td><code>convert.timestamp.duration.to.native</code></td>
+ <td><code>true</code></td>
+ <td>Whether to convert Protobuf <code>Timestamp</code> and
<code>Duration</code> fields to Spark's native <code>TimestampType</code> and
<code>DayTimeIntervalType</code>, respectively, when deserializing. When set to
<code>false</code>, these fields are deserialized as structs with
<code>seconds</code> (<code>LongType</code>) and <code>nanos</code>
(<code>IntegerType</code>) fields.</td>
+ <td>read</td>
+ </tr>
<tr>
<td><code>retain.empty.message.types</code></td>
<td><code>false</code></td>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
