This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon 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 57c35dbcb504 [SPARK-58218][SQL] Replace generic RuntimeException with
typed exceptions in AvroDeserializer
57c35dbcb504 is described below
commit 57c35dbcb50408a996b7a880c8cf4ef5b6ad9ec8
Author: Ganesha Shreedhara <[email protected]>
AuthorDate: Thu Aug 6 14:35:02 2026 +0900
[SPARK-58218][SQL] Replace generic RuntimeException with typed exceptions
in AvroDeserializer
### What changes were proposed in this pull request?
`AvroDeserializer` threw three generic `java.lang.RuntimeException`s. This
replaces them with typed exceptions:
- The `BYTES -> BinaryType` type-mismatch case now throws
`IncompatibleSchemaException` (via `errorPrefix`), matching the five sibling
type-mismatch cases in the same `newWriter` method that already do so.
- The non-nullable array-element and map-value null-read cases now throw a
`SparkRuntimeException` with a new error condition
`AVRO_CANNOT_READ_NULL_FIELD` (SQLSTATE 22004), the read-side analog of the
existing `AVRO_CANNOT_WRITE_NULL_FIELD`.
### Why are the changes needed?
Generic `RuntimeException`s are not catchable by error class, carry no
SQLSTATE, and (for the binary case) were inconsistent with the sibling cases in
the same method. Spark has largely migrated runtime failures to the error-class
framework; these sites were left behind. The write path already had a typed,
catalogued error for the analogous non-null-field situation
(`AVRO_CANNOT_WRITE_NULL_FIELD`); the read path now has its counterpart.
### Does this PR introduce _any_ user-facing change?
Yes. When reading Avro data where a null appears in a non-nullable array
element or map value, or where a `BYTES` value is not a valid Avro binary, the
exception type and message change from a generic `RuntimeException` to a typed
`SparkRuntimeException` / `IncompatibleSchemaException` carrying an error
condition and SQLSTATE.
### How was this patch tested?
Added unit tests in `AvroCatalystDataConversionSuite` that deserialize Avro
data containing a null into a non-nullable Catalyst array element and map
value, asserting the `AVRO_CANNOT_READ_NULL_FIELD` error condition and
field-path parameter via `checkError`. Also ran `SparkThrowableSuite`
(error-conditions.json validation), `AvroCatalystDataConversionSuite`, and
`AvroSerdeSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57376 from ganeshashree/SPARK-58218.
Lead-authored-by: Ganesha Shreedhara <[email protected]>
Co-authored-by: Ganesha S <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit dc58c809fb6aab9c2de2a0c0df2318207dece3ed)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 7 +++
.../sql/avro/AvroCatalystDataConversionSuite.scala | 68 +++++++++++++++++++++-
.../apache/spark/sql/avro/AvroDeserializer.scala | 14 +++--
3 files changed, 83 insertions(+), 6 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index dd43b76ca403..9ff51c66fc6b 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -365,6 +365,13 @@
],
"sqlState" : "42613"
},
+ "AVRO_CANNOT_READ_NULL_FIELD" : {
+ "message" : [
+ "Cannot read null value for <name> into a non-nullable SQL type.",
+ "To allow null values, declare the corresponding type as nullable in the
read schema."
+ ],
+ "sqlState" : "22004"
+ },
"AVRO_CANNOT_WRITE_NULL_FIELD" : {
"message" : [
"Cannot write null value for field <name> defined as non-null Avro data
type <dataType>.",
diff --git
a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
index 49d597bfc8a7..7bffcc7da9ba 100644
---
a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
+++
b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
@@ -24,7 +24,7 @@ import org.apache.avro.Schema
import org.apache.avro.generic.{GenericData, GenericRecordBuilder}
import org.apache.avro.message.{BinaryMessageDecoder, BinaryMessageEncoder}
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkException, SparkRuntimeException}
import org.apache.spark.sql.{RandomDataGenerator, Row}
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow,
NoopFilters, OrderedFilters, StructFilters}
import org.apache.spark.sql.catalyst.expressions.{ExpressionEvalHelper,
GenericInternalRow, Literal}
@@ -300,6 +300,72 @@ class AvroCatalystDataConversionSuite extends
SharedSparkSession
}
}
+ private def deserializerFor(schema: Schema, dataType: DataType):
AvroDeserializer = {
+ new AvroDeserializer(
+ schema,
+ dataType,
+ false,
+ RebaseSpec(LegacyBehaviorPolicy.CORRECTED),
+ new NoopFilters,
+ false,
+ "",
+ -1)
+ }
+
+ test("SPARK-58218: null array element for non-null Catalyst type reports a
typed error") {
+ val avroSchema = new Schema.Parser().parse(
+ """
+ |{ "type": "record",
+ | "name": "record",
+ | "fields": [{
+ | "name": "array",
+ | "type": { "type": "array", "items": ["null", "int"] }
+ | }]
+ |}
+ """.stripMargin)
+ // The Avro element is nullable, but the target Catalyst element type is
not, so a null
+ // element must surface as a typed error rather than a generic
RuntimeException.
+ val catalystType = new StructType()
+ .add("array", ArrayType(IntegerType, containsNull = false), nullable =
false)
+ val data = new GenericRecordBuilder(avroSchema)
+ .set("array", util.Arrays.asList(1, null, 3))
+ .build()
+
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ deserializerFor(avroSchema, catalystType).deserialize(data)
+ },
+ condition = "AVRO_CANNOT_READ_NULL_FIELD",
+ parameters = Map("name" -> "field 'array.element'"))
+ }
+
+ test("SPARK-58218: null map value for non-null Catalyst type reports a typed
error") {
+ val avroSchema = new Schema.Parser().parse(
+ """
+ |{ "type": "record",
+ | "name": "record",
+ | "fields": [{
+ | "name": "map",
+ | "type": { "type": "map", "values": ["null", "int"] }
+ | }]
+ |}
+ """.stripMargin)
+ val catalystType = new StructType()
+ .add("map", MapType(StringType, IntegerType, valueContainsNull = false),
nullable = false)
+ val values = new util.HashMap[String, Integer]()
+ values.put("k", null)
+ val data = new GenericRecordBuilder(avroSchema)
+ .set("map", values)
+ .build()
+
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ deserializerFor(avroSchema, catalystType).deserialize(data)
+ },
+ condition = "AVRO_CANNOT_READ_NULL_FIELD",
+ parameters = Map("name" -> "field 'map.value'"))
+ }
+
test("avro array can be generic java collection") {
val jsonFormatSchema =
"""
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
index ce16c4a2cc3a..078d80350f4d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
@@ -30,6 +30,7 @@ import org.apache.avro.Schema.Type._
import org.apache.avro.generic._
import org.apache.avro.util.Utf8
+import org.apache.spark.SparkRuntimeException
import org.apache.spark.sql.avro.AvroUtils.{nonNullUnionBranches, toFieldStr,
AvroMatchedField}
import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters, StructFilters}
import org.apache.spark.sql.catalyst.expressions.{SpecificInternalRow,
UnsafeArrayData}
@@ -275,7 +276,8 @@ private[sql] class AvroDeserializer(
bytes
case b: Array[Byte] => b
case other =>
- throw new RuntimeException(errorPrefix + s"$other is not a valid
avro binary.")
+ throw new IncompatibleSchemaException(
+ errorPrefix + s"$other is not a valid Avro binary.")
}
updater.set(ordinal, bytes)
@@ -329,8 +331,9 @@ private[sql] class AvroDeserializer(
val element = iter.next()
if (element == null) {
if (!containsNull) {
- throw new RuntimeException(
- s"Array value at path ${toFieldStr(avroElementPath)} is not
allowed to be null")
+ throw new SparkRuntimeException(
+ errorClass = "AVRO_CANNOT_READ_NULL_FIELD",
+ messageParameters = Map("name" ->
toFieldStr(avroElementPath)))
} else {
elementUpdater.setNullAt(i)
}
@@ -361,8 +364,9 @@ private[sql] class AvroDeserializer(
keyWriter(keyUpdater, i, entry.getKey)
if (entry.getValue == null) {
if (!valueContainsNull) {
- throw new RuntimeException(
- s"Map value at path ${toFieldStr(avroPath :+ "value")} is
not allowed to be null")
+ throw new SparkRuntimeException(
+ errorClass = "AVRO_CANNOT_READ_NULL_FIELD",
+ messageParameters = Map("name" -> toFieldStr(avroPath :+
"value")))
} else {
valueUpdater.setNullAt(i)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
