You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ru...@apache.org on 2026/09/18 09:14:20 UTC
(spark) branch master updated: [SPARK-58972][SQL][PROTOBUF] from_protobuf unwrapped primitive wrappers follow wrapper presence, not emit.default.values
This is an automated email from the ASF dual-hosted git repository.
zhengruifeng 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 93697d5b4df6 [SPARK-58972][SQL][PROTOBUF] from_protobuf unwrapped primitive wrappers follow wrapper presence, not emit.default.values
93697d5b4df6 is described below
commit 93697d5b4df6d18a0d5c377109d4f5259cbff70f
Author: bhollis-dbx <be...@databricks.com>
AuthorDate: Fri Sep 18 17:13:55 2026 +0800
[SPARK-58972][SQL][PROTOBUF] from_protobuf unwrapped primitive wrappers follow wrapper presence, not emit.default.values
### What changes were proposed in this pull request?
When `from_protobuf` is used with `unwrap.primitive.wrapper.types=true`, a present well-known primitive wrapper (`google.protobuf.{Bool,Int32,UInt32,Int64,UInt64,Float,Double,String,Bytes}Value`) now unwraps to its inner scalar's default (0/""/false/empty-bytes) when the inner value is unset, rather than to null. Null results only when the wrapper field is absent. This is independent of `emit.default.values`.
The deserializer previously read the inner scalar through the same path used for bare proto3 scalars (`getFieldValue`), which returns null for an unset/default scalar unless `emit.default.values=true`. That conflated wrapper *message presence* with the `emit.default.values` option (which governs bare proto3 scalar defaults). The unwrap converters now read the inner value directly via `DynamicMessage.getField`, which returns the scalar's default and never null, matching proto3 wrapper [...]
### Why are the changes needed?
- It matches the canonical proto3 JSON / wrapper mapping: wrapper presence carries the value; absent -> null.
- It fixes an untested crash. For `repeated`/`map` fields of unwrapped wrappers the container is non-nullable (`containsNull=false` / `valueContainsNull=false`). A present-but-empty wrapper element used to deserialize to null inside that non-null container, causing a downstream `UnsafeWriter` NullPointerException. With the fix such elements become non-null defaults, so the crash and the schema mismatch both disappear with no schema change.
### Does this PR introduce _any_ user-facing change?
Yes. With `from_protobuf` and `unwrap.primitive.wrapper.types=true`, a present wrapper whose inner value is the default now deserializes to that default instead of null (and repeated/map wrappers with empty elements no longer error). Absent wrappers still deserialize to null.
### How was this patch tested?
Updated the "test well known wrappers with emit defaults" expectations and added a new test covering empty elements in a repeated wrapper (-> 0) and a map wrapper value (-> ""). Ran the protobuf module suite: 94 tests pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #58253 from bhollis-dbx/bhollis-from-protobuf-wrapper-presence.
Authored-by: bhollis-dbx <be...@databricks.com>
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
.../spark/sql/protobuf/ProtobufDeserializer.scala | 93 +++++++---------------
.../sql/protobuf/ProtobufFunctionsSuite.scala | 38 +++++++--
docs/sql-data-sources-protobuf.md | 4 +-
3 files changed, 62 insertions(+), 73 deletions(-)
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 3ff810b70182..7a088479d746 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
@@ -364,101 +364,51 @@ private[sql] class ProtobufDeserializer(
updater.set(ordinal, UTF8String.fromString(jsonStr))
// Handle well known wrapper types. We unpack the value field when the desired
- // output type is a primitive (determined by the option in [[ProtobufOptions]])
+ // output type is a primitive (determined by the option in [[ProtobufOptions]]).
+ // A wrapper is unwrapped only when present: an absent singular wrapper is nulled by the
+ // caller before reaching here, and container elements are always present. So a present
+ // wrapper -- even an empty one -- carries a value, and unwrapWktValue reads the inner
+ // scalar (its default when unset), never null, independent of `emit.default.values`.
case (MESSAGE, BooleanType)
if protoType.getMessageType.getFullName == BoolValue.getDescriptor.getFullName =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- updater.setBoolean(ordinal, unwrapped.asInstanceOf[Boolean])
- }
+ updater.setBoolean(ordinal, unwrapWktValue(value).asInstanceOf[Boolean])
case (MESSAGE, IntegerType)
if (protoType.getMessageType.getFullName == Int32Value.getDescriptor.getFullName
|| protoType.getMessageType.getFullName == UInt32Value.getDescriptor.getFullName) =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- updater.setInt(ordinal, unwrapped.asInstanceOf[Int])
- }
+ updater.setInt(ordinal, unwrapWktValue(value).asInstanceOf[Int])
case (MESSAGE, LongType)
if (protoType.getMessageType.getFullName == UInt32Value.getDescriptor.getFullName) =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- updater.setLong(ordinal, Integer.toUnsignedLong(unwrapped.asInstanceOf[Int]))
- }
+ updater.setLong(ordinal, Integer.toUnsignedLong(unwrapWktValue(value).asInstanceOf[Int]))
case (MESSAGE, LongType)
if (protoType.getMessageType.getFullName == Int64Value.getDescriptor.getFullName
|| protoType.getMessageType.getFullName == UInt64Value.getDescriptor.getFullName) =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- updater.setLong(ordinal, unwrapped.asInstanceOf[Long])
- }
+ updater.setLong(ordinal, unwrapWktValue(value).asInstanceOf[Long])
case (MESSAGE, DecimalType.LongDecimal)
if (protoType.getMessageType.getFullName == UInt64Value.getDescriptor.getFullName) =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- val dec = Decimal.fromString(
- UTF8String.fromString(java.lang.Long.toUnsignedString(unwrapped.asInstanceOf[Long])))
- updater.setDecimal(ordinal, dec)
- }
+ val dec = Decimal.fromString(UTF8String.fromString(
+ java.lang.Long.toUnsignedString(unwrapWktValue(value).asInstanceOf[Long])))
+ updater.setDecimal(ordinal, dec)
case (MESSAGE, StringType)
if protoType.getMessageType.getFullName == StringValue.getDescriptor.getFullName =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- updater.set(ordinal, UTF8String.fromString(unwrapped.asInstanceOf[String]))
- }
+ updater.set(ordinal, UTF8String.fromString(unwrapWktValue(value).asInstanceOf[String]))
case (MESSAGE, BinaryType)
if protoType.getMessageType.getFullName == BytesValue.getDescriptor.getFullName =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- updater.set(ordinal, unwrapped.asInstanceOf[ByteString].toByteArray)
- }
+ updater.set(ordinal, unwrapWktValue(value).asInstanceOf[ByteString].toByteArray)
case (MESSAGE, FloatType)
if protoType.getMessageType.getFullName == FloatValue.getDescriptor.getFullName =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- updater.setFloat(ordinal, unwrapped.asInstanceOf[Float])
- }
+ updater.setFloat(ordinal, unwrapWktValue(value).asInstanceOf[Float])
case (MESSAGE, DoubleType)
if protoType.getMessageType.getFullName == DoubleValue.getDescriptor.getFullName =>
(updater, ordinal, value) =>
- val dm = value.asInstanceOf[DynamicMessage]
- val unwrapped = getFieldValue(dm, dm.getDescriptorForType.getFields.get(0))
- if (unwrapped == null) {
- updater.setNullAt(ordinal)
- } else {
- updater.setDouble(ordinal, unwrapped.asInstanceOf[Double])
- }
+ updater.setDouble(ordinal, unwrapWktValue(value).asInstanceOf[Double])
case (MESSAGE, st: StructType) =>
val writeRecord = getRecordWriter(
@@ -577,6 +527,17 @@ private[sql] class ProtobufDeserializer(
}
}
+ // Reads the inner `value` scalar of a present well-known wrapper message
+ // (google.protobuf.{Bool,Int32,...}Value). Unlike getFieldValue, this returns the scalar's
+ // default (0/""/false/empty-bytes) rather than null when the inner value is unset, matching the
+ // proto3 wrapper semantics where a present wrapper always carries a value. Absent singular
+ // wrappers never reach here (they are nulled by the caller), so this is only invoked for present
+ // wrappers, and is independent of `emit.default.values`.
+ private def unwrapWktValue(value: Any): AnyRef = {
+ val dm = value.asInstanceOf[DynamicMessage]
+ dm.getField(dm.getDescriptorForType.getFields.get(0))
+ }
+
// TODO: All of the code below this line is same between protobuf and avro, it can be shared.
private def createArrayData(elementType: DataType, length: Int): ArrayData = elementType match {
case BooleanType => UnsafeArrayData.fromPrimitiveArray(new Array[Boolean](length))
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 8ecc983c4603..f31fbc3bc7bf 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
@@ -2071,11 +2071,10 @@ class ProtobufFunctionsSuite extends SharedSparkSession with ProtobufTestBase
)
}
} else {
- if (defaults == "false") {
- checkAnswer(parsedExplicitZero, expectedEmpty)
- } else {
- checkAnswer(parsedExplicitZero, Seq((0)).toDF("int32_val"))
- }
+ // When unwrapping, a present wrapper carries a value even if it is the inner scalar's
+ // default, so an explicit zero unwraps to 0 regardless of emit.default.values (which
+ // only governs bare proto3 scalars, not a wrapper message's presence).
+ checkAnswer(parsedExplicitZero, Seq((0)).toDF("int32_val"))
}
// For nonzero, we should get back the number or wrapped version regardless
@@ -2104,6 +2103,35 @@ class ProtobufFunctionsSuite extends SharedSparkSession with ProtobufTestBase
}
}
+ test("well known wrappers with empty container elements unwrap to defaults") {
+ // A repeated/map field of unwrapped wrappers uses a non-null container
+ // (containsNull = false / valueContainsNull = false). A present-but-empty wrapper element
+ // must unwrap to the inner scalar's default, not null -- a null in a non-null container
+ // crashes downstream. Other container-wrapper tests only use non-empty elements, so this
+ // covers the empty-element case for both a repeated and a map field.
+ val message = spark.range(1).select(
+ lit(
+ WellKnownWrapperTypes.newBuilder()
+ .addInt32List(Int32Value.getDefaultInstance) // empty element -> 0
+ .addInt32List(Int32Value.of(7))
+ .putWktMap(1, StringValue.getDefaultInstance) // empty value -> ""
+ .build().toByteArray
+ ).as("raw_proto"))
+
+ val opt = Map("unwrap.primitive.wrapper.types" -> "true")
+ checkWithFileAndClassName("WellKnownWrapperTypes") { case (name, descFilePathOpt) =>
+ val parsed = message.select(
+ from_protobuf_wrapper($"raw_proto", name, descFilePathOpt, opt).as("proto"))
+
+ checkAnswer(
+ parsed.select("proto.int32_list"),
+ spark.range(1).select(typedLit(List(0, 7)).as("int32_list")))
+ checkAnswer(
+ parsed.select("proto.wkt_map"),
+ spark.range(1).select(typedLit(Map(1 -> "")).as("wkt_map")))
+ }
+ }
+
test("test well known wrappers with upcast ints") {
// Test that the unwrap primitives behavior and upcast uint64 work correctly together.
// We'll check the deserialization behavior under every combination of the
diff --git a/docs/sql-data-sources-protobuf.md b/docs/sql-data-sources-protobuf.md
index b9deaee1e06d..f5b84454f1b9 100644
--- a/docs/sql-data-sources-protobuf.md
+++ b/docs/sql-data-sources-protobuf.md
@@ -478,7 +478,7 @@ Data source options of Protobuf can be set via:
<tr>
<td><code>emit.default.values</code></td>
<td><code>false</code></td>
- <td>Whether to render fields with zero values when deserializing Protobuf to a Spark struct. When a field is empty in the serialized Protobuf, this library will deserialize them as <code>null</code> by default, this option can control whether to render the type-specific zero values.</td>
+ <td>Whether to render fields with zero values when deserializing Protobuf to a Spark struct. When a field is empty in the serialized Protobuf, this library will deserialize it as <code>null</code> by default; this option controls whether to render the type-specific zero value instead. This option does not affect unwrapped primitive wrapper types.</td>
<td>read</td>
</tr>
<tr>
@@ -496,7 +496,7 @@ Data source options of Protobuf can be set via:
<tr>
<td><code>unwrap.primitive.wrapper.types</code></td>
<td><code>false</code></td>
- <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.</td>
+ <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>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org