You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2023/02/22 01:34:40 UTC
[spark] branch branch-3.4 updated: [SPARK-42406][SQL] Fix check for missing required fields of to_protobuf
This is an automated email from the ASF dual-hosted git repository.
gengliang pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new 655992841e6 [SPARK-42406][SQL] Fix check for missing required fields of to_protobuf
655992841e6 is described below
commit 655992841e67f91e30bf59bd6cc48c9d4698a0c7
Author: Raghu Angadi <ra...@databricks.com>
AuthorDate: Tue Feb 21 17:34:16 2023 -0800
[SPARK-42406][SQL] Fix check for missing required fields of to_protobuf
### What changes were proposed in this pull request?
Protobuf serializer (used in `to_protobuf()`) should error if non-nullable fields (i.e. protobuf `required` fields) are present in the schema of the catalyst record being converted to a protobuf.
But `isNullable()` method used for this check returns opposite (see PR comment in the diff). As a result, Serializer incorrectly requires the fields that are optional. This PR fixes this check (see PR comment in the diff).
This also requires corresponding fix for couple of unit tests. In order use a Protobuf message with a `required` field, Protobuf version 2 file `proto2_messages.proto` is added.
Two tests are updated to verify missing required fields results in an error.
### Why are the changes needed?
This is need to fix a bug where we were incorrectly enforcing a schema check on optional fields rather than on required fields.
### Does this PR introduce _any_ user-facing change?
It fixes a bug, and gives more flexibility for user queries.
### How was this patch tested?
- Updated unit tests
Closes #40080 from rangadi/fix-required-field-check.
Authored-by: Raghu Angadi <ra...@databricks.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
(cherry picked from commit fb5647732fa2f49838f803f67ea11b20fc14282b)
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../spark/sql/protobuf/utils/ProtobufUtils.scala | 7 +--
.../test/resources/protobuf/proto2_messages.desc | 8 +++
.../protobuf/proto2_messages.proto} | 28 +++++------
.../src/test/resources/protobuf/serde_suite.proto | 11 ----
.../spark/sql/protobuf/ProtobufSerdeSuite.scala | 58 ++++++++++++++++------
.../spark/sql/protobuf/ProtobufTestBase.scala | 4 ++
6 files changed, 69 insertions(+), 47 deletions(-)
diff --git a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala
index 49313a3ce91..bf207d6068f 100644
--- a/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala
+++ b/connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala
@@ -100,7 +100,7 @@ private[sql] object ProtobufUtils extends Logging {
*/
def validateNoExtraRequiredProtoFields(): Unit = {
val extraFields = protoFieldArray.toSet -- matchedFields.map(_.fieldDescriptor)
- extraFields.filterNot(isNullable).foreach { extraField =>
+ extraFields.filter(_.isRequired).foreach { extraField =>
throw QueryCompilationErrors.cannotFindProtobufFieldInCatalystError(
toFieldStr(protoPath :+ extraField.getName()))
}
@@ -283,9 +283,4 @@ private[sql] object ProtobufUtils extends Logging {
case Seq() => "top-level record"
case n => s"field '${n.mkString(".")}'"
}
-
- /** Return true if `fieldDescriptor` is optional. */
- private[protobuf] def isNullable(fieldDescriptor: FieldDescriptor): Boolean =
- !fieldDescriptor.isOptional
-
}
diff --git a/connector/protobuf/src/test/resources/protobuf/proto2_messages.desc b/connector/protobuf/src/test/resources/protobuf/proto2_messages.desc
new file mode 100644
index 00000000000..a9e4099a7f2
--- /dev/null
+++ b/connector/protobuf/src/test/resources/protobuf/proto2_messages.desc
@@ -0,0 +1,8 @@
+
+�
+proto2_messages.proto$org.apache.spark.sql.protobuf.protos"@
+FoobarWithRequiredFieldBar
+foo ( Rfoo
+bar (Rbar"�
+ NestedFoobarWithRequiredFieldBare
+
nested_foobar ( 2@.org.apache.spark.sql.protobuf.protos.FoobarWithRequiredFieldBarR nestedFoobarBBProto2Messages
\ No newline at end of file
diff --git a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufTestBase.scala b/connector/protobuf/src/test/resources/protobuf/proto2_messages.proto
similarity index 57%
copy from connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufTestBase.scala
copy to connector/protobuf/src/test/resources/protobuf/proto2_messages.proto
index 831b4a26c06..a5d09df8514 100644
--- a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufTestBase.scala
+++ b/connector/protobuf/src/test/resources/protobuf/proto2_messages.proto
@@ -15,23 +15,19 @@
* limitations under the License.
*/
-package org.apache.spark.sql.protobuf
+syntax = "proto2";
-import org.apache.spark.sql.test.SQLTestUtils
+package org.apache.spark.sql.protobuf.protos;
+option java_outer_classname = "Proto2Messages";
-trait ProtobufTestBase extends SQLTestUtils {
- /**
- * Returns full path to the given file in the resource folder,
- * if the first choice throw NPE, try to return the full path of alternative.
- * The result path doesn't contain the `file:/` protocol part.
- */
- protected def testFile(fileName: String, alternateFileName: String): String = {
- val ret = try {
- testFile(fileName)
- } catch {
- case _: NullPointerException => testFile(alternateFileName)
- }
- ret.replace("file:/", "/")
- }
+// Used to test missing required field bar in top level schema.
+message FoobarWithRequiredFieldBar {
+ optional string foo = 1;
+ required int32 bar = 2;
+}
+
+// Used to test missing required field bar in nested struct.
+message NestedFoobarWithRequiredFieldBar {
+ optional FoobarWithRequiredFieldBar nested_foobar = 1;
}
diff --git a/connector/protobuf/src/test/resources/protobuf/serde_suite.proto b/connector/protobuf/src/test/resources/protobuf/serde_suite.proto
index a7459213a87..87152b035b0 100644
--- a/connector/protobuf/src/test/resources/protobuf/serde_suite.proto
+++ b/connector/protobuf/src/test/resources/protobuf/serde_suite.proto
@@ -59,17 +59,6 @@ message TypeMiss {
int64 bar = 1;
}
-/* Field boo missing from SQL root, but available in Protobuf root*/
-message FieldMissingInSQLRoot {
- Foo foo = 1;
- int32 boo = 2;
-}
-
-/* Field baz missing from SQL nested and available in Protobuf nested*/
-message FieldMissingInSQLNested {
- Baz foo = 1;
-}
-
message Baz {
int32 bar = 1;
int32 baz = 2;
diff --git a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufSerdeSuite.scala b/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufSerdeSuite.scala
index 87ed5340943..356cd20eb4e 100644
--- a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufSerdeSuite.scala
+++ b/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufSerdeSuite.scala
@@ -21,7 +21,7 @@ import com.google.protobuf.Descriptors.Descriptor
import com.google.protobuf.DynamicMessage
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.NoopFilters
+import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters}
import org.apache.spark.sql.catalyst.expressions.Cast.toSQLType
import org.apache.spark.sql.protobuf.utils.ProtobufUtils
import org.apache.spark.sql.test.SharedSparkSession
@@ -39,6 +39,8 @@ class ProtobufSerdeSuite extends SharedSparkSession with ProtobufTestBase {
val testFileDesc = testFile("serde_suite.desc", "protobuf/serde_suite.desc")
private val javaClassNamePrefix = "org.apache.spark.sql.protobuf.protos.SerdeSuiteProtos$"
+ val proto2Desc = testFile("proto2_messages.desc", "protobuf/proto2_messages.desc")
+
test("Test basic conversion") {
withFieldMatchType { fieldMatch =>
val (top, nest) = fieldMatch match {
@@ -64,6 +66,28 @@ class ProtobufSerdeSuite extends SharedSparkSession with ProtobufTestBase {
}
}
+ test("Optional fields can be dropped from input SQL schema for the serializer") {
+ // This test verifies that optional fields can be missing from input Catalyst schema
+ // while serializing rows to protobuf.
+
+ val desc = ProtobufUtils.buildDescriptor(proto2Desc, "FoobarWithRequiredFieldBar")
+
+ // Confirm desc contains optional field 'foo' and required field bar.
+ assert(desc.getFields.size() == 2)
+ assert(desc.findFieldByName("foo").isOptional)
+
+ // Use catalyst type without optional "foo".
+ val sqlType = structFromDDL("struct<bar int>")
+ val serializer = new ProtobufSerializer(sqlType, desc, nullable = false) // Should work fine.
+
+ // Should be able to deserialize a row.
+ val protoMessage = serializer.serialize(InternalRow(22)).asInstanceOf[DynamicMessage]
+
+ // Verify the descriptor and the value.
+ assert(protoMessage.getDescriptorForType == desc)
+ assert(protoMessage.getField(desc.findFieldByName("bar")).asInstanceOf[Int] == 22)
+ }
+
test("Fail to convert with field type mismatch") {
val protoFile = ProtobufUtils.buildDescriptor(testFileDesc, "MissMatchTypeInRoot")
withFieldMatchType { fieldMatch =>
@@ -144,44 +168,50 @@ class ProtobufSerdeSuite extends SharedSparkSession with ProtobufTestBase {
test("Fail to convert with missing Catalyst fields") {
val protoFile = ProtobufUtils.buildDescriptor(testFileDesc, "FieldMissingInSQLRoot")
- // serializing with extra fails if extra field is missing in SQL Schema
+ val foobarSQLType = structFromDDL("struct<foo string>") // "bar" is missing.
+
assertFailedConversionMessage(
- protoFile,
+ ProtobufUtils.buildDescriptor(proto2Desc, "FoobarWithRequiredFieldBar"),
Serializer,
BY_NAME,
+ catalystSchema = foobarSQLType,
errorClass = "UNABLE_TO_CONVERT_TO_PROTOBUF_MESSAGE_TYPE",
params = Map(
- "protobufType" -> "FieldMissingInSQLRoot",
- "toType" -> toSQLType(CATALYST_STRUCT)))
+ "protobufType" -> "FoobarWithRequiredFieldBar",
+ "toType" -> toSQLType(foobarSQLType)))
/* deserializing should work regardless of whether the extra field is missing
in SQL Schema or not */
withFieldMatchType(Deserializer.create(CATALYST_STRUCT, protoFile, _))
withFieldMatchType(Deserializer.create(CATALYST_STRUCT, protoFile, _))
- val protoNestedFile = ProtobufUtils.buildDescriptor(testFileDesc, "FieldMissingInSQLNested")
+ val protoNestedFile = ProtobufUtils
+ .buildDescriptor(proto2Desc, "NestedFoobarWithRequiredFieldBar")
- // serializing with extra fails if extra field is missing in SQL Schema
+ val nestedFoobarSQLType = structFromDDL(
+ "struct<nested_foobar: struct<foo string>>" // "bar" field is missing.
+ )
+ // serializing with extra fails if required field is missing in inner struct
assertFailedConversionMessage(
- protoNestedFile,
+ ProtobufUtils.buildDescriptor(proto2Desc, "NestedFoobarWithRequiredFieldBar"),
Serializer,
BY_NAME,
+ catalystSchema = nestedFoobarSQLType,
errorClass = "UNABLE_TO_CONVERT_TO_PROTOBUF_MESSAGE_TYPE",
params = Map(
- "protobufType" -> "FieldMissingInSQLNested",
- "toType" -> toSQLType(CATALYST_STRUCT)))
+ "protobufType" -> "NestedFoobarWithRequiredFieldBar",
+ "toType" -> toSQLType(nestedFoobarSQLType)))
/* deserializing should work regardless of whether the extra field is missing
in SQL Schema or not */
- withFieldMatchType(Deserializer.create(CATALYST_STRUCT, protoNestedFile, _))
- withFieldMatchType(Deserializer.create(CATALYST_STRUCT, protoNestedFile, _))
+ withFieldMatchType(Deserializer.create(nestedFoobarSQLType, protoNestedFile, _))
}
test("raise cannot parse and construct protobuf descriptor error") {
// passing serde_suite.proto instead serde_suite.desc
var testFileDesc = testFile("serde_suite.proto", "protobuf/serde_suite.proto")
val e1 = intercept[AnalysisException] {
- ProtobufUtils.buildDescriptor(testFileDesc, "FieldMissingInSQLRoot")
+ ProtobufUtils.buildDescriptor(testFileDesc, "SerdeBasicMessage")
}
checkError(
@@ -191,7 +221,7 @@ class ProtobufSerdeSuite extends SharedSparkSession with ProtobufTestBase {
testFileDesc = testFile("basicmessage_noimports.desc", "protobuf/basicmessage_noimports.desc")
val e2 = intercept[AnalysisException] {
- ProtobufUtils.buildDescriptor(testFileDesc, "FieldMissingInSQLRoot")
+ ProtobufUtils.buildDescriptor(testFileDesc, "SerdeBasicMessage")
}
checkError(
diff --git a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufTestBase.scala b/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufTestBase.scala
index 831b4a26c06..2ead89e4545 100644
--- a/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufTestBase.scala
+++ b/connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufTestBase.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.protobuf
import org.apache.spark.sql.test.SQLTestUtils
+import org.apache.spark.sql.types.{DataType, StructType}
trait ProtobufTestBase extends SQLTestUtils {
@@ -34,4 +35,7 @@ trait ProtobufTestBase extends SQLTestUtils {
}
ret.replace("file:/", "/")
}
+
+ protected def structFromDDL(ddl: String): StructType =
+ DataType.fromDDL(ddl).asInstanceOf[StructType]
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org