Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-41015][SQL][PROTOBUF] UnitTest null check for data generator #38515

Closed
wants to merge 5 commits into from
Closed

[SPARK-41015][SQL][PROTOBUF] UnitTest null check for data generator #38515

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
bcc333f
null check for data generator
SandishKumarHN Nov 5, 2022
d4e26b9
bug fix Map.empty to Map with Utest
SandishKumarHN Nov 5, 2022
f11b04d
review changes Utest for cannt construct proto desc
SandishKumarHN Nov 6, 2022
c9abb2d
review changes nit, utests changes
SandishKumarHN Nov 7, 2022
44e9209
review changes utest, empty protobuf
SandishKumarHN Nov 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,7 +189,7 @@ private[sql] object ProtobufUtils extends Logging {
}
}

def parseFileDescriptorSet(descFilePath: String): List[Descriptors.FileDescriptor] = {
private def parseFileDescriptorSet(descFilePath: String): List[Descriptors.FileDescriptor] = {
var fileDescriptorSet: DescriptorProtos.FileDescriptorSet = null
try {
val dscFile = new BufferedInputStream(new FileInputStream(descFilePath))
Expand Down
10 changes: 5 additions & 5 deletions ...uf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufCatalystDataConversionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,14 +130,14 @@ class ProtobufCatalystDataConversionSuite

val rand = new scala.util.Random(seed)
val generator = RandomDataGenerator.forType(dt, rand = rand).get
var data = generator()
var data = generator().asInstanceOf[Row]
// Do not use default values, since from_protobuf() returns null in v3.
while (
data != null &&
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This should just be data == null || data.asInstanceOf[Row].get(0) == defaultValue?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I meant, we don't need the array check.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rangadi

  1. data.asInstanceOf[Row].get(0) == ByteString.empty().toByteArray
  2. data.asInstanceOf[Row].get(0) == Array.emptyByteArray
  3. data.asInstanceOf[Row].get(0) == ByteString.EMPTY.toByteArray
  4. data.asInstanceOf[Row].get(0) == "".getBytes
  5. data.asInstanceOf[Row].get(0).isInstanceOf[Array[Byte]] && data.asInstanceOf[Row].get(0).asInstanceOf[Array[Byte]].isEmpty

Except for (5), none of them worked. I'm printing under the conditions listed above.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see. Equals does not check the content on array.
Optional:
We could recduce data.asInstanceOf calls with val data = generator().asInstanceOf[Row].
Also could replace

data.asInstanceOf[Row].get(0).isInstanceOf[Array[Byte]]
with
dt == BinaryType

(data.asInstanceOf[Row].get(0) == defaultValue ||
(data.asInstanceOf[Row].get(0).isInstanceOf[Array[Byte]] &&
data.asInstanceOf[Row].get(0).asInstanceOf[Array[Byte]].isEmpty)))
data = generator()
(data.get(0) == defaultValue ||
(dt == BinaryType &&
data.get(0).asInstanceOf[Array[Byte]].isEmpty)))
data = generator().asInstanceOf[Row]

val converter = CatalystTypeConverters.createToCatalystConverter(dt)
val input = Literal.create(converter(data), dt)
Expand Down
30 changes: 30 additions & 0 deletions connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufFunctionsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -677,4 +677,34 @@ class ProtobufFunctionsSuite extends QueryTest with SharedSparkSession with Seri
=== inputDf.select("durationMsg.duration").take(1).toSeq(0).get(0))
}
}

test("raise cannot construct protobuf descriptor error") {
val basicMessageDesc = ProtobufUtils.buildDescriptor(testFileDesc, "BasicMessage")

val basicMessage = DynamicMessage
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Btw, you don't need the message. Could use empty byte array while initializing df.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

correct, fixed.

.newBuilder(basicMessageDesc)
.setField(basicMessageDesc.findFieldByName("id"), 1111L)
.setField(basicMessageDesc.findFieldByName("string_value"), "slam")
.setField(basicMessageDesc.findFieldByName("int32_value"), 12345)
.setField(basicMessageDesc.findFieldByName("int64_value"), 0x90000000000L)
.setField(basicMessageDesc.findFieldByName("double_value"), 10000000000.0d)
.setField(basicMessageDesc.findFieldByName("float_value"), 10902.0f)
.setField(basicMessageDesc.findFieldByName("bool_value"), true)
.setField(
basicMessageDesc.findFieldByName("bytes_value"),
ByteString.copyFromUtf8("ProtobufDeserializer"))
.build()

val df = Seq(basicMessage.toByteArray).toDF("value")
val testFileDescriptor = testFile("basicmessage_noimports.desc").replace("file:/", "/")

val e = intercept[AnalysisException] {
df.select(functions.from_protobuf($"value", "BasicMessage", testFileDescriptor) as 'sample)
.where("sample.string_value == \"slam\"").show()
}
checkError(
exception = e,
errorClass = "CANNOT_CONSTRUCT_PROTOBUF_DESCRIPTOR",
parameters = Map("descFilePath" -> testFileDescriptor))
}
}
18 changes: 8 additions & 10 deletions connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/ProtobufSerdeSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,27 +177,25 @@ class ProtobufSerdeSuite extends SharedSparkSession {
withFieldMatchType(Deserializer.create(CATALYST_STRUCT, protoNestedFile, _))
}

test("raise cannot parse protobuf descriptor error") {
test("raise cannot parse and construct protobuf descriptor error") {
// passing serde_suite.proto instead serde_suite.desc
val testFileDesc = testFile("serde_suite.proto").replace("file:/", "/")
val e = intercept[AnalysisException] {
var testFileDesc = testFile("serde_suite.proto").replace("file:/", "/")
val e1 = intercept[AnalysisException] {
ProtobufUtils.buildDescriptor(testFileDesc, "FieldMissingInSQLRoot")
}

checkError(
exception = e,
exception = e1,
errorClass = "CANNOT_PARSE_PROTOBUF_DESCRIPTOR",
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@SandishKumarHN Thank you for the added test.

parameters = Map("descFilePath" -> testFileDesc))
}

test("raise cannot construct protobuf descriptor error") {
val testFileDesc = testFile("basicmessage_noimports.desc").replace("file:/", "/")
val e = intercept[AnalysisException] {
ProtobufUtils.parseFileDescriptorSet(testFileDesc)
testFileDesc = testFile("basicmessage_noimports.desc").replace("file:/", "/")
val e2 = intercept[AnalysisException] {
ProtobufUtils.buildDescriptor(testFileDesc, "FieldMissingInSQLRoot")
}

checkError(
exception = e,
exception = e2,
errorClass = "CANNOT_CONSTRUCT_PROTOBUF_DESCRIPTOR",
parameters = Map("descFilePath" -> testFileDesc))
}
Expand Down