This is an automated email from the ASF dual-hosted git repository.
viirya 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 606116b08f9d [MINOR][4.X][SQL] Bound nesting depth when converting
protobuf Any to JSON
606116b08f9d is described below
commit 606116b08f9df205fc9b5654d5c95cf33663d8c9
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Tue Oct 27 21:46:19 2026 -0700
[MINOR][4.X][SQL] Bound nesting depth when converting protobuf Any to JSON
### What changes were proposed in this pull request?
Backport of ef6f78bc8930157890a0238ac462ad214ead57d3 ("[SQL][MINOR] Bound
nesting depth when converting protobuf Any to JSON") to branch-4.x. The commit
is a clean cherry-pick.
When `convert.any.fields.to.json` is enabled, `ProtobufDeserializer` now
checks the message under each `google.protobuf.Any` field before calling
`JsonFormat.print`. The check walks the nested messages iteratively under a
single depth budget of 100. A record that goes over the budget fails with
`InvalidProtocolBufferException`, which the existing parse-mode handling treats
as a malformed record.
This change is already in master, branch-4.3, branch-4.2 and branch-4.1,
but not in branch-4.x.
### Why are the changes needed?
`JsonFormat.print` re-parses each nested `Any` with a fresh protobuf
recursion limit while it recurses on the JVM stack. A deeply nested Any-in-Any
record can therefore overflow the executor stack. `StackOverflowError` is
fatal, so it escapes the `NonFatal` handling behind PERMISSIVE/FAILFAST and
fails the task instead of producing a malformed record.
Without this backport, 4.4.0 would lack a fix that 4.3 has.
### Does this PR introduce _any_ user-facing change?
Yes. With `convert.any.fields.to.json=true`, a record whose message nesting
under an `Any` field is deeper than 100 is now treated as malformed: null in
PERMISSIVE mode, an error in FAILFAST mode. Before, it could crash the task
with `StackOverflowError`. This is the same behavior that master, 4.3, 4.2 and
4.1 already have.
### How was this patch tested?
Includes the original test from ProtobufFunctionsSuite, "deeply nested
Any-in-Any is rejected as malformed instead of overflowing the stack".
`build/sbt 'protobuf/testOnly *Protobuf*Suite'` on JDK 17: 98/98 passed (5
suites).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor (original commit). The backport was prepared with
Claude Code.
Closes #59846 from viirya/backport-ef6f78-branch-4.x.
Lead-authored-by: Liang-Chi Hsieh <[email protected]>
Co-authored-by: Holden Karau <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../spark/sql/protobuf/ProtobufDeserializer.scala | 61 +++++++++++++++++++++-
.../sql/protobuf/ProtobufFunctionsSuite.scala | 24 +++++++++
2 files changed, 83 insertions(+), 2 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 61ae3c98d2eb..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
@@ -18,7 +18,10 @@ package org.apache.spark.sql.protobuf
import java.util.concurrent.TimeUnit
-import com.google.protobuf.{BoolValue, ByteString, BytesValue, DoubleValue,
DynamicMessage, FloatValue, Int32Value, Int64Value, Message, StringValue,
TypeRegistry, UInt32Value, UInt64Value}
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+import com.google.protobuf.{BoolValue, ByteString, BytesValue, DoubleValue,
DynamicMessage, FloatValue, Int32Value, Int64Value,
InvalidProtocolBufferException, Message, StringValue, TypeRegistry,
UInt32Value, UInt64Value}
import com.google.protobuf.Descriptors._
import com.google.protobuf.Descriptors.FieldDescriptor.JavaType._
import com.google.protobuf.util.JsonFormat
@@ -94,6 +97,58 @@ private[sql] class ProtobufDeserializer(
.usingTypeRegistry(typeRegistry)
}
+ // Matches protobuf-java's default recursion limit for wire-format parsing.
A nested
+ // google.protobuf.Any is stored as opaque bytes, so JsonFormat.print
re-parses each Any
+ // level with a fresh recursion limit while recursing on the JVM stack;
without a shared
+ // budget across levels, an Any-in-Any chain a few hundred KB long overflows
the stack, and
+ // StackOverflowError is fatal: it escapes the NonFatal-based
malformed-record handling
+ // that implements PERMISSIVE/FAILFAST parse modes and kills the task
instead.
+ private val anyJsonRecursionLimit = 100
+
+ /**
+ * Performs the same Any unpacking that JsonFormat will perform (against the
same type
+ * registry), but iteratively and under a single depth budget, so an
over-deep message
+ * fails as an ordinary malformed record (InvalidProtocolBufferException).
+ */
+ private def checkAnyJsonRecursion(root: DynamicMessage): Unit = {
+ val stack = mutable.Stack[(Message, Int)]((root, 0))
+ while (stack.nonEmpty) {
+ val (message, depth) = stack.pop()
+ if (depth > anyJsonRecursionLimit) {
+ throw new InvalidProtocolBufferException(
+ "Message nesting in google.protobuf.Any exceeds the maximum
supported depth " +
+ s"($anyJsonRecursionLimit) for JSON conversion")
+ }
+ val descriptor = message.getDescriptorForType
+ if (descriptor.getFullName == "google.protobuf.Any") {
+ val typeUrl =
+
message.getField(descriptor.findFieldByName("type_url")).asInstanceOf[String]
+ // Leave empty or unregistered type URLs for jsonPrinter.print to
handle: an empty
+ // Any prints as {}, and an unresolvable type fails there with a
descriptive error.
+ if (typeUrl.nonEmpty) {
+ val nestedDescriptor = typeRegistry.getDescriptorForTypeUrl(typeUrl)
+ if (nestedDescriptor != null) {
+ val bytes =
+
message.getField(descriptor.findFieldByName("value")).asInstanceOf[ByteString]
+ stack.push((DynamicMessage.parseFrom(nestedDescriptor, bytes),
depth + 1))
+ }
+ }
+ } else {
+ message.getAllFields.asScala.foreach { case (field, fieldValue) =>
+ if (field.getJavaType == MESSAGE) {
+ if (field.isRepeated) {
+ fieldValue.asInstanceOf[java.util.List[Message]].asScala.foreach
{ item =>
+ stack.push((item, depth + 1))
+ }
+ } else {
+ stack.push((fieldValue.asInstanceOf[Message], depth + 1))
+ }
+ }
+ }
+ }
+ }
+ }
+
private def newArrayWriter(
protoField: FieldDescriptor,
protoPath: Seq[String],
@@ -302,8 +357,10 @@ private[sql] class ProtobufDeserializer(
case (MESSAGE, StringType)
if protoType.getMessageType.getFullName == "google.protobuf.Any" =>
(updater, ordinal, value) =>
+ val anyMessage = value.asInstanceOf[DynamicMessage]
+ checkAnyJsonRecursion(anyMessage)
// Convert 'Any' protobuf message to JSON string.
- val jsonStr = jsonPrinter.print(value.asInstanceOf[DynamicMessage])
+ val jsonStr = jsonPrinter.print(anyMessage)
updater.set(ordinal, UTF8String.fromString(jsonStr))
// Handle well known wrapper types. We unpack the value field when the
desired
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 3d8368435bfe..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
@@ -1575,6 +1575,30 @@ class ProtobufFunctionsSuite extends SharedSparkSession
with ProtobufTestBase
}
}
+ test("deeply nested Any-in-Any is rejected as malformed instead of
overflowing the stack") {
+ checkWithFileAndClassName("ProtoWithAny") { case (name, descFilePathOpt) =>
+ // Each nested Any level is re-parsed by JsonFormat with a fresh
protobuf recursion
+ // limit; without a cross-level budget a deeply nested record overflows
the executor
+ // stack with StackOverflowError, which PERMISSIVE mode cannot recover
from.
+ var nested = AnyProto.pack(SimpleMessage.newBuilder().setId(1).build())
+ (0 until 2000).foreach(_ => nested = AnyProto.pack(nested))
+ val deeplyNested = ProtoWithAny.newBuilder()
+ .setEventName("deeply-nested")
+ .setDetails(nested)
+ .build()
+ .toByteArray
+ val inputDF = Seq(deeplyNested).toDF("binary")
+
+ val options = Map(
+ ProtobufOptions.CONVERT_ANY_FIELDS_TO_JSON_CONFIG -> "true",
+ "mode" -> "PERMISSIVE")
+ val dfJson = inputDF.select(
+ from_protobuf_wrapper($"binary", name, descFilePathOpt,
options).as("proto"))
+ // The record is treated as malformed (null in PERMISSIVE mode) and the
task survives.
+ assert(dfJson.collect()(0).getStruct(0) == null)
+ }
+ }
+
test("test explicitly set zero values - proto3") {
// All fields explicitly zero. Message, map, repeated, and oneof fields
// are left unset, as null is their zero value.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
