You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/08/06 05:35:18 UTC
(spark) branch master updated: [SPARK-58218][SQL] Replace generic RuntimeException with typed exceptions in AvroDeserializer
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new dc58c809fb6a [SPARK-58218][SQL] Replace generic RuntimeException with typed exceptions in AvroDeserializer
dc58c809fb6a is described below
commit dc58c809fb6aab9c2de2a0c0df2318207dece3ed
Author: Ganesha Shreedhara <ga...@gmail.com>
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 <ga...@gmail.com>
Co-authored-by: Ganesha S <ga...@databricks.com>
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
.../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 2ade38757be3..7953595e7b5e 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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org