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/03/18 22:24:05 UTC
(spark) branch master updated: [SPARK-56043][SQL] Wrap NullPointerException from Avro 1.12.x ParseContext.resolve() in SchemaParseException
This is an automated email from the ASF dual-hosted git repository.
gurwls223 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 68cbd1c50e44 [SPARK-56043][SQL] Wrap NullPointerException from Avro 1.12.x ParseContext.resolve() in SchemaParseException
68cbd1c50e44 is described below
commit 68cbd1c50e448f862787c7d5360d53efe0952b71
Author: Jerry Peng <je...@databricks.com>
AuthorDate: Thu Mar 19 07:23:52 2026 +0900
[SPARK-56043][SQL] Wrap NullPointerException from Avro 1.12.x ParseContext.resolve() in SchemaParseException
### What changes were proposed in this pull request?
Add AvroUtils.parseAvroSchema() that wraps Schema.Parser().parse() and catches NullPointerException, re-throwing it as SchemaParseException. This lets the existing parseMode-based error handling (FAILFAST → MALFORMED_AVRO_MESSAGE, PERMISSIVE → null row) work correctly. All call sites (AvroDataToCatalyst, CatalystDataToAvro, SchemaOfAvro, AvroOptions, AvroOutputWriterFactory, and the schema registry variants) are updated to use it.
### Why are the changes needed?
After the Avro 1.11.3 → 1.12.0 upgrade ([SPARK-49014](https://issues.apache.org/jira/browse/SPARK-49014)), Schema.Parser().parse() can throw NullPointerException from ParseContext.resolve() for certain invalid user-provided schemas. In Avro 1.11.x, the same schemas threw SchemaParseException, which was caught by the existing NonFatal(e) handler in AvroDataToCatalyst.nullSafeEval and reported as MALFORMED_AVRO_MESSAGE. The NPE bypasses nothing — it's still NonFatal — but it surfaces as [...]
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
tests added
### Was this patch authored or co-authored using generative AI tooling?
yes
Generated-by: claude
Closes #54876 from jerrypeng/SPARK-56043.
Authored-by: Jerry Peng <je...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../apache/spark/sql/avro/AvroDataToCatalyst.scala | 4 +-
.../apache/spark/sql/avro/CatalystDataToAvro.scala | 3 +-
.../org/apache/spark/sql/avro/SchemaOfAvro.scala | 3 +-
.../sql/avro/AvroCatalystDataConversionSuite.scala | 49 ++++++++++++++++++++++
.../apache/spark/sql/avro/AvroFunctionsSuite.scala | 33 +++++++++++++++
.../org/apache/spark/sql/avro/AvroOptions.scala | 2 +-
.../spark/sql/avro/AvroOutputWriterFactory.scala | 3 +-
.../org/apache/spark/sql/avro/AvroUtils.scala | 26 +++++++++++-
8 files changed, 112 insertions(+), 11 deletions(-)
diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala b/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala
index c0b50e5cba48..87bb2f2a17c2 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala
+++ b/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala
@@ -19,7 +19,6 @@ package org.apache.spark.sql.avro
import scala.util.control.NonFatal
-import org.apache.avro.Schema
import org.apache.avro.generic.GenericDatumReader
import org.apache.avro.io.{BinaryDecoder, DecoderFactory}
@@ -57,8 +56,7 @@ case class AvroDataToCatalyst(
private lazy val avroOptions = AvroOptions(options)
- @transient private lazy val actualSchema =
- new Schema.Parser().setValidateDefaults(false).parse(jsonFormatSchema)
+ @transient private lazy val actualSchema = AvroUtils.parseAvroSchema(jsonFormatSchema)
@transient private lazy val expectedSchema = avroOptions.schema.getOrElse(actualSchema)
diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala b/connector/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala
index 01daac1cfa6b..fb7f44e6db0f 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala
+++ b/connector/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala
@@ -19,7 +19,6 @@ package org.apache.spark.sql.avro
import java.io.ByteArrayOutputStream
-import org.apache.avro.Schema
import org.apache.avro.generic.GenericDatumWriter
import org.apache.avro.io.{BinaryEncoder, EncoderFactory}
@@ -35,7 +34,7 @@ case class CatalystDataToAvro(
@transient private lazy val avroType =
jsonFormatSchema
- .map(new Schema.Parser().setValidateDefaults(false).parse)
+ .map(AvroUtils.parseAvroSchema)
.getOrElse(SchemaConverters.toAvroType(child.dataType, child.nullable))
@transient private lazy val serializer =
diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaOfAvro.scala b/connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaOfAvro.scala
index 362fc79a0f1b..d5184139603b 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaOfAvro.scala
+++ b/connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaOfAvro.scala
@@ -36,8 +36,7 @@ case class SchemaOfAvro(
@transient private lazy val avroOptions = AvroOptions(options)
- @transient private lazy val actualSchema =
- new Schema.Parser().setValidateDefaults(false).parse(jsonFormatSchema)
+ @transient private lazy val actualSchema = AvroUtils.parseAvroSchema(jsonFormatSchema)
@transient private lazy val expectedSchema = avroOptions.schema.getOrElse(actualSchema)
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 9371f0a74078..a680634fab30 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
@@ -386,4 +386,53 @@ class AvroCatalystDataConversionSuite extends SparkFunSuite
checkDeserialization(avroSchema, avroRecord, Some(expected))
checkDeserialization(avroSchema, avroRecord, Some(expected))
}
+
+ test("SPARK-56043: AvroDataToCatalyst with unresolvable schema reference") {
+ // Avro 1.12.x throws NPE from ParseContext.resolve() for undefined named types.
+ // Our fix wraps the NPE in SchemaParseException so the existing parseMode error
+ // handling works correctly (FAILFAST -> MALFORMED_AVRO_MESSAGE).
+ val invalidSchema =
+ """
+ |{
+ | "type": "record",
+ | "name": "TestRecord",
+ | "fields": [
+ | {"name": "value", "type": "UndefinedType"}
+ | ]
+ |}
+ """.stripMargin
+
+ val data = Literal(Array[Byte](1, 2, 3))
+ intercept[SparkException] {
+ AvroDataToCatalyst(data, invalidSchema, Map("mode" -> "FAILFAST")).eval()
+ }
+ }
+
+ test("SPARK-56043: AvroDataToCatalyst with malformed JSON schema") {
+ val malformedSchema = "not valid json"
+ val data = Literal(Array[Byte](1, 2, 3))
+ intercept[SparkException] {
+ AvroDataToCatalyst(data, malformedSchema, Map("mode" -> "FAILFAST")).eval()
+ }
+ }
+
+ test("SPARK-56043: bare string schema reference triggers NPE in Avro 1.12.x") {
+ // This is the exact pattern that triggers NPE from ParseContext.resolve() in 1.12.x.
+ // In 1.11.x this threw SchemaParseException. Our fix wraps NPE in SchemaParseException.
+ val bareStringRef = "\"com.test.Missing\""
+
+ // Verify raw Avro 1.12.x parser throws NPE for bare string references
+ val rawException = intercept[Exception] {
+ new Schema.Parser().setValidateDefaults(false).parse(bareStringRef)
+ }
+ assert(rawException.isInstanceOf[NullPointerException],
+ s"Expected NullPointerException from Avro 1.12.x ParseContext.resolve(), " +
+ s"but got ${rawException.getClass.getName}: ${rawException.getMessage}")
+
+ // Verify our fix wraps it in SchemaParseException
+ val wrappedException = intercept[org.apache.avro.SchemaParseException] {
+ AvroUtils.parseAvroSchema(bareStringRef)
+ }
+ assert(wrappedException.getCause.isInstanceOf[NullPointerException])
+ }
}
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroFunctionsSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroFunctionsSuite.scala
index a0b254e3153c..81c9b2f5069f 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroFunctionsSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroFunctionsSuite.scala
@@ -743,4 +743,37 @@ class AvroFunctionsSuite extends QueryTest with SharedSparkSession {
from_avro($"avro", avroStructSchema).as("schedule"))
checkAnswer(readBack, df)
}
+
+ test("SPARK-56043: from_avro with invalid schema should not throw NPE") {
+ // An Avro schema that references an undefined type should produce MALFORMED_AVRO_MESSAGE
+ // rather than a raw NullPointerException from the Avro library's ParseContext.resolve().
+ val invalidSchema =
+ """
+ |{
+ | "type": "record",
+ | "name": "TestRecord",
+ | "fields": [
+ | {"name": "value", "type": "UndefinedType"}
+ | ]
+ |}
+ """.stripMargin
+
+ val df = spark.range(1).select(lit(Array[Byte](1, 2, 3)).as("data"))
+ val ex = intercept[Exception] {
+ df.select(from_avro($"data", invalidSchema)).collect()
+ }
+ assert(!ex.isInstanceOf[NullPointerException],
+ s"Should not throw NPE, but got: ${ex.getClass.getName}: ${ex.getMessage}")
+ }
+
+ test("SPARK-56043: from_avro with completely unparseable schema should not throw NPE") {
+ val garbageSchema = "this is not valid JSON at all"
+
+ val df = spark.range(1).select(lit(Array[Byte](1, 2, 3)).as("data"))
+ val ex = intercept[Exception] {
+ df.select(from_avro($"data", garbageSchema)).collect()
+ }
+ assert(!ex.isInstanceOf[NullPointerException],
+ s"Should not throw NPE, but got: ${ex.getClass.getName}: ${ex.getMessage}")
+ }
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala
index 967a21613791..9903456a031f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala
@@ -73,7 +73,7 @@ private[sql] class AvroOptions(
* instead of "string" type in the default converted schema.
*/
val schema: Option[Schema] = {
- parameters.get(AVRO_SCHEMA).map(new Schema.Parser().setValidateDefaults(false).parse).orElse({
+ parameters.get(AVRO_SCHEMA).map(AvroUtils.parseAvroSchema).orElse({
val avroUrlSchema = parameters.get(AVRO_SCHEMA_URL).map(url => {
log.debug("loading avro schema from url: " + url)
val fs = FileSystem.get(new URI(url), conf)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriterFactory.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriterFactory.scala
index 15c76ec358ed..5a77ba3819c7 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriterFactory.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriterFactory.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.avro
-import org.apache.avro.Schema
import org.apache.avro.mapreduce.AvroJob
import org.apache.hadoop.mapreduce.TaskAttemptContext
@@ -37,7 +36,7 @@ private[sql] class AvroOutputWriterFactory(
avroSchemaAsJsonString: String,
positionalFieldMatching: Boolean) extends OutputWriterFactory {
- private lazy val avroSchema = new Schema.Parser().parse(avroSchemaAsJsonString)
+ private lazy val avroSchema = AvroUtils.parseAvroSchema(avroSchemaAsJsonString)
override def getFileExtension(context: TaskAttemptContext): String = {
val codec = context.getConfiguration.get(AvroJob.CONF_OUTPUT_CODEC)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
index 97e7479732cf..f9e15322c031 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
@@ -21,7 +21,7 @@ import java.util.Locale
import scala.jdk.CollectionConverters._
-import org.apache.avro.{Schema, SchemaFormatter}
+import org.apache.avro.{Schema, SchemaFormatter, SchemaParseException}
import org.apache.avro.file.{DataFileReader, FileReader}
import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
import org.apache.avro.mapred.{AvroOutputFormat, FsInput}
@@ -48,6 +48,30 @@ private[sql] object AvroUtils extends Logging {
val JSON_INLINE_FORMAT: String = "json/inline"
val JSON_PRETTY_FORMAT: String = "json/pretty"
+ /**
+ * Parses an Avro schema from a JSON string with proper error handling.
+ *
+ * The Avro 1.12.x library's Schema.Parser can throw NullPointerException from
+ * ParseContext.resolve() when the schema references named types that cannot be
+ * resolved. This method wraps such NPEs in SchemaParseException so they are
+ * handled by the existing error handling in from_avro/to_avro (which catches
+ * NonFatal exceptions and reports them via the parseMode mechanism).
+ *
+ * @param jsonFormatSchema the Avro schema in JSON string format
+ * @return the parsed Avro Schema
+ * @throws SchemaParseException if the schema cannot be parsed
+ */
+ def parseAvroSchema(jsonFormatSchema: String): Schema = {
+ try {
+ new Schema.Parser().setValidateDefaults(false).parse(jsonFormatSchema)
+ } catch {
+ case e: NullPointerException =>
+ val ex = new SchemaParseException(s"Failed to parse Avro schema: ${e.getMessage}")
+ ex.initCause(e)
+ throw ex
+ }
+ }
+
def inferSchema(
spark: SparkSession,
options: Map[String, String],
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org