[SPARK-44337][PROTOBUF] Any fields set to 'Any.getDefaultInstance' cause parse errors

### What changes were proposed in this pull request?

While building the Protobuf descriptor from FileDescriptorSet, this PR uses bundled Java descriptor for 'google/protobuf/any.proto'. This works around a bug in Protobuf's JsonFormat while parsing Any fields initialized with `Any.getDefaultInstance()`.

An existing test for Any fields is updated to test this case. Without the fix, the test passes with Java classes, but fails with FileDescriptorSet bytes. The latter is the common use case.

The bug in JsonFormat is related to how it checks for Any values at runtime that were originally initialized with `Any.getDefaultInstance()`. Specifically the conditional [here](https://github.com/protocolbuffers/protobuf/blob/51f79d4415338cfe068a39d7ce466870df223245/java/util/src/main/java/com/google/protobuf/util/JsonFormat.java#L860)  in JsonFormat.java fails since it tries to compare with actual default instance in Java from Java descriptor. If we build the same descriptor separately, JsonFormat does not recognize it.

### Why are the changes needed?

Without this fix, input that includes fields initialized to `Any.getDefaultInstance` can not be parsed.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

An existing unit test is updated to test this case.

Closes #41897 from rangadi/any-default.

Authored-by: Raghu Angadi <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: Raghu Angadi

connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala

@@ -266,7 +266,16 @@ private[sql] object ProtobufUtils extends Logging {
     val fileDescriptorList = fileDescriptorProto.getDependencyList().asScala.map { dependency =>
       fileDescriptorProtoMap.get(dependency) match {
         case Some(dependencyProto) =>
-          buildFileDescriptor(dependencyProto, fileDescriptorProtoMap)
+          if (dependencyProto.getName == "google/protobuf/any.proto"
+            && dependencyProto.getPackage == "google.protobuf") {
+            // For Any, use the descriptor already included as part of the Java dependency.
+            // Without this, JsonFormat used for converting Any fields fails when
+            // an Any field in input is set to `Any.getDefaultInstance()`.
+            com.google.protobuf.AnyProto.getDescriptor
+            // Should we do the same for timestamp.proto and empty.proto?
+          } else {
+            buildFileDescriptor(dependencyProto, fileDescriptorProtoMap)
+          }
         case None =>
           throw QueryCompilationErrors.protobufDescriptorDependencyError(dependency)
       }

connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufFunctionsSuite.scala

@@ -1214,9 +1214,10 @@ class ProtobufFunctionsSuite extends QueryTest with SharedSparkSession with Prot
 
     checkWithFileAndClassName("ProtoWithAnyArray") { case (name, descFilePathOpt) =>
 
-      // proto: message { string description = 1; repeated google.protobuf.Any items = 2;
+      // proto: message { string description = 1; repeated google.protobuf.Any items = 2 };
 
-      // Use two different types of protos for 'items'. One with an Any field, and one without.
+      // Use 3 different types of protos for 'items'. One with an Any field, and one without,
+      // and one with default instance of Any. The last one triggers JsonFormat bug.
 
       val simpleProto = SimpleMessage.newBuilder() // Json: {"id":10,"string_value":"galaxy"}
         .setId(10)
@@ -1232,6 +1233,7 @@ class ProtobufFunctionsSuite extends QueryTest with SharedSparkSession with Prot
         .setDescription("nested any demo")
         .addItems(AnyProto.pack(simpleProto)) // A simple proto
         .addItems(AnyProto.pack(protoWithAny)) // A proto with any field inside it.
+        .addItems(AnyProto.getDefaultInstance) // An Any field initialized to default instance.
         .build()
         .toByteArray
 
@@ -1246,6 +1248,9 @@ class ProtobufFunctionsSuite extends QueryTest with SharedSparkSession with Prot
         "items: ARRAY<STRUCT<type_url: STRING, value: BINARY>>>"
       )
 
+      // Print df to see how the Any fields look like without json conversion.
+      log.info(s"Input row without json conversion: ${df.collect()(0)}")
+
       // String for items with 'convert.to.json' option enabled.
       val options = Map(ProtobufOptions.CONVERT_ANY_FIELDS_TO_JSON_CONFIG -> "true")
       val dfJson = inputDF.select(from_protobuf_wrapper(
@@ -1276,6 +1281,7 @@ class ProtobufFunctionsSuite extends QueryTest with SharedSparkSession with Prot
           |     "string_value":"galaxy"
           |   }
           | }""".stripMargin))
+      assert(items.get(2) == "{}") // 3rd field is empty (Any.getDefaultInstance)
     }
   }