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 584c87335982 [SPARK-57573][CONNECT] Support the TIME data type in the
Protobuf connector
584c87335982 is described below
commit 584c87335982bffb05a95fa7a566c04064d353f5
Author: Maxim Gekk <[email protected]>
AuthorDate: Fri Jun 26 14:41:01 2026 +0200
[SPARK-57573][CONNECT] Support the TIME data type in the Protobuf connector
### What changes were proposed in this pull request?
This PR adds support for the Spark `TIME` (`TimeType`) data type in the
Protobuf connector (`to_protobuf` / `from_protobuf`), matching the existing
Avro support added in SPARK-54473.
Spark `TimeType` is mapped to a plain Protobuf `int64` field holding the
number of nanoseconds since midnight (nanoseconds-of-day). Since `TimeType` is
stored internally as nanoseconds since midnight, the conversion is a direct
copy of the `Long` value (no unit math).
- `ProtobufSerializer`: added a `(TimeType, LONG)` case that writes a
`TimeType` column into an `int64` field as nanos-of-day.
- `ProtobufDeserializer`: added a `(LONG, TimeType)` case that reads an
`int64` field back into a `TimeType` value.
- `docs/sql-data-sources-protobuf.md`: documented the new `TimeType` ->
`int64 (nanoseconds-of-day)` mapping.
Note: a bare Protobuf `int64` carries no logical-type marker, and
`from_protobuf` always derives its output schema from the descriptor via
`SchemaConverters.toSqlType`. Therefore `from_protobuf` infers such a field as
`LongType` (nanoseconds-of-day), which can be cast back to `TIME` with
`CAST(... AS TIME)`. The `int64 -> TimeType` branch is currently reachable only
by constructing a `ProtobufDeserializer` with an explicit `TimeType` schema
(exercised by the round-trip test); it is th [...]
### Why are the changes needed?
To extend support for the TIME data type to the Protobuf connector, closing
the gap noted in SPARK-57573 where `SchemaConverters`/serializer/deserializer
had no `TimeType` mapping. This is a sub-task of SPARK-57550 (Extend support
for the TIME data type).
### Does this PR introduce _any_ user-facing change?
Yes. `to_protobuf` now accepts `TIME` columns, writing them to a Protobuf
`int64` field as nanoseconds-of-day, and the Protobuf data source documentation
lists the new mapping.
### How was this patch tested?
Added tests in `ProtobufFunctionsSuite`:
- `SPARK-57573: Handle TimeType between to_protobuf and from_protobuf` -
verifies a `TIME` column serializes to the `int64` field and reads back as
`LongType` equal to `LocalTime.toNanoOfDay`.
- `SPARK-57573: TimeType roundtrip through ProtobufSerializer and
ProtobufDeserializer` - direct serializer/deserializer round trip with an
explicit `TimeType` schema, covering precisions (`TimeType(0)`, `TimeType(6)`)
and boundary values (`00:00:00`, `23:59:59.999999`).
All existing `ProtobufFunctionsSuite`, `ProtobufSerdeSuite`, and
`ProtobufCatalystDataConversionSuite` tests pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor
Closes #56807 from MaxGekk/time-protobuf.
Authored-by: Maxim Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
---
.../spark/sql/protobuf/ProtobufDeserializer.scala | 5 ++
.../spark/sql/protobuf/ProtobufSerializer.scala | 5 ++
.../test/resources/protobuf/functions_suite.proto | 5 ++
.../sql/protobuf/ProtobufFunctionsSuite.scala | 65 +++++++++++++++++++++-
docs/sql-data-sources-protobuf.md | 6 ++
5 files changed, 85 insertions(+), 1 deletion(-)
diff --git
a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufDeserializer.scala
b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufDeserializer.scala
index 9f2bafb8a180..baee05a29948 100644
---
a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufDeserializer.scala
+++
b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufDeserializer.scala
@@ -235,6 +235,11 @@ private[sql] class ProtobufDeserializer(
case (LONG, LongType) =>
(updater, ordinal, value) => updater.setLong(ordinal,
value.asInstanceOf[Long])
+ case (LONG, _: TimeType) =>
+ // The int64 field stores nanoseconds-of-day, matching TimeType's
internal
+ // nanos-since-midnight representation, so copy the value directly.
+ (updater, ordinal, value) => updater.setLong(ordinal,
value.asInstanceOf[Long])
+
case (LONG, DecimalType.LongDecimal) =>
(updater, ordinal, value) =>
updater.setDecimal(
diff --git
a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufSerializer.scala
b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufSerializer.scala
index cd638f3ac12a..4ad954ed3c09 100644
---
a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufSerializer.scala
+++
b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufSerializer.scala
@@ -149,6 +149,11 @@ private[sql] class ProtobufSerializer(
case (DateType, INT) =>
(getter, ordinal) => getter.getInt(ordinal)
+ case (_: TimeType, LONG) =>
+ // TimeType is stored internally as nanoseconds-since-midnight; write
it
+ // directly into the int64 field as nanos-of-day.
+ (getter, ordinal) => getter.getLong(ordinal)
+
case (TimestampType, MESSAGE) =>
(getter, ordinal) =>
val millis = DateTimeUtils.microsToMillis(getter.getLong(ordinal))
diff --git
a/connector/protobuf/src/test/resources/protobuf/functions_suite.proto
b/connector/protobuf/src/test/resources/protobuf/functions_suite.proto
index 4cae7f9abf28..8a873682ec3d 100644
--- a/connector/protobuf/src/test/resources/protobuf/functions_suite.proto
+++ b/connector/protobuf/src/test/resources/protobuf/functions_suite.proto
@@ -175,6 +175,11 @@ message durationMsg {
Duration duration = 2;
}
+message timeMsg {
+ string key = 1;
+ int64 time_val = 2;
+}
+
message OneOfEvent {
string key = 1;
oneof payload {
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 3f9b77d63d67..5c875989e714 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
@@ -17,7 +17,7 @@
package org.apache.spark.sql.protobuf
import java.sql.Timestamp
-import java.time.Duration
+import java.time.{Duration, LocalTime}
import scala.jdk.CollectionConverters._
@@ -25,6 +25,7 @@ import com.google.protobuf.{Any => AnyProto, BoolValue,
ByteString, BytesValue,
import org.json4s.jackson.JsonMethods
import org.apache.spark.sql.{AnalysisException, Column, DataFrame, Row}
+import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.functions.{array, lit, map, struct, typedLit}
import org.apache.spark.sql.protobuf.protos.Proto2Messages.Proto2AllTypes
import org.apache.spark.sql.protobuf.protos.SimpleMessageProtos._
@@ -34,6 +35,7 @@ import org.apache.spark.sql.protobuf.utils.ProtobufUtils
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.{ProtobufUtils => CommonProtobufUtils}
+import org.apache.spark.unsafe.types.UTF8String
class ProtobufFunctionsSuite extends SharedSparkSession with ProtobufTestBase
with Serializable {
@@ -722,6 +724,67 @@ class ProtobufFunctionsSuite extends SharedSparkSession
with ProtobufTestBase
}
}
+ test("SPARK-57573: Handle TimeType between to_protobuf and from_protobuf") {
+ val schema = StructType(
+ StructField("timeMsg",
+ StructType(
+ StructField("key", StringType, nullable = true) ::
+ StructField("time_val", TimeType(), nullable = true) :: Nil
+ ),
+ nullable = true
+ ) :: Nil
+ )
+
+ val localTime = LocalTime.of(12, 34, 56, 123456000)
+ val inputDf = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(
+ Row(Row("key1", localTime))
+ )),
+ schema
+ )
+
+ checkWithFileAndClassName("timeMsg") {
+ case (name, descFilePathOpt) =>
+ val toProtoDf = inputDf
+ .select(to_protobuf_wrapper($"timeMsg", name,
+ descFilePathOpt) as Symbol("to_proto"))
+
+ val fromProtoDf = toProtoDf
+ .select(from_protobuf_wrapper($"to_proto", name,
+ descFilePathOpt) as Symbol("timeMsg"))
+
+ // The int64 field carries no logical-type marker, so from_protobuf
infers it as
+ // LongType holding nanoseconds-of-day rather than TimeType.
+
assert(fromProtoDf.schema("timeMsg").dataType.asInstanceOf[StructType]("time_val")
+ .dataType === LongType)
+ assert(fromProtoDf.select("timeMsg.key").first().get(0) === "key1")
+ assert(fromProtoDf.select("timeMsg.time_val").first().get(0) ===
localTime.toNanoOfDay)
+ }
+ }
+
+ test("SPARK-57573: TimeType roundtrip through ProtobufSerializer and
ProtobufDeserializer") {
+ val descriptor = ProtobufUtils.buildDescriptor("timeMsg",
Some(testFileDesc)).descriptor
+ Seq(
+ (TimeType(TimeType.MIN_PRECISION), LocalTime.of(0, 0, 0)),
+ (TimeType(TimeType.MICROS_PRECISION), LocalTime.of(23, 59, 59,
999999000)),
+ (TimeType(TimeType.MICROS_PRECISION), LocalTime.of(12, 34, 56,
123456000))
+ ).foreach { case (timeType, localTime) =>
+ val catalyst = StructType(
+ StructField("key", StringType) ::
+ StructField("time_val", timeType) :: Nil)
+ val serializer = new ProtobufSerializer(catalyst, descriptor, nullable =
false)
+ val deserializer = new ProtobufDeserializer(descriptor, catalyst)
+ val nanos = localTime.toNanoOfDay
+ val input = InternalRow(UTF8String.fromString("key1"), nanos)
+
+ val message = serializer.serialize(input).asInstanceOf[DynamicMessage]
+ val result = deserializer.deserialize(message).get
+
+ assert(result.getUTF8String(0).toString === "key1")
+ assert(result.getLong(1) === nanos)
+ }
+ }
+
test("raise protobuf descriptor error") {
val df = Seq(ByteString.empty().toByteArray).toDF("value")
val descWithoutImports = descriptorSetWithoutImports(testFileDesc,
"BasicMessage")
diff --git a/docs/sql-data-sources-protobuf.md
b/docs/sql-data-sources-protobuf.md
index 4dd6579f92cd..b9deaee1e06d 100644
--- a/docs/sql-data-sources-protobuf.md
+++ b/docs/sql-data-sources-protobuf.md
@@ -397,8 +397,14 @@ Spark supports the writing of all Spark SQL types into
Protobuf. For most types,
<td>MapType</td>
<td>map</td>
</tr>
+ <tr>
+ <td>TimeType</td>
+ <td>int64 (nanoseconds-of-day)</td>
+ </tr>
</table>
+`TimeType` is written as a plain Protobuf `int64` holding the number of
nanoseconds since midnight. Because a bare `int64` carries no logical-type
marker, `from_protobuf` reads such a field back as `LongType` (the
nanoseconds-of-day value), which can be cast back to a TIME value with
`cast(... AS TIME)` if needed.
+
## Handling circular references protobuf fields
One common issue that can arise when working with Protobuf data is the
presence of circular references. In Protobuf, a circular reference occurs when
a field refers back to itself or to another field that refers back to the
original field. This can cause issues when parsing the data, as it can result
in infinite loops or other unexpected behavior.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
