You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ho...@apache.org on 2026/07/10 06:15:22 UTC
(spark) branch branch-4.1 updated (b36915ae93b4 -> 9ff88c468c21)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
from b36915ae93b4 [SPARK][SQL] LDAP Thriftserver improvement
new 84edbdd7e501 [SPARK-58042][CONNECT] Validate the UDT jvm_class is a UserDefinedType before instantiating it
new 9ff88c468c21 [SQL] Allow more control over UDT creation during input loading Provide users more control over UDT creation during dynamic loading of different inputs.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../src/main/resources/error/error-conditions.json | 17 +++++++
.../apache/spark/sql/errors/DataTypeErrors.scala | 15 ++++++
.../org/apache/spark/sql/internal/SqlApiConf.scala | 6 +++
.../spark/sql/internal/SqlApiConfHelper.scala | 2 +
.../org/apache/spark/sql/types/DataType.scala | 14 +++++-
.../org/apache/spark/sql/internal/SQLConf.scala | 26 +++++++++++
.../org/apache/spark/sql/types/DataTypeSuite.scala | 53 +++++++++++++++++++++-
.../connect/common/DataTypeProtoConverter.scala | 16 +++++--
.../connect/planner/InvalidInputErrorsSuite.scala | 18 ++++++++
.../execution/datasources/SchemaMergeUtils.scala | 51 ++++++++++++---------
.../datasources/parquet/ParquetQuerySuite.scala | 36 +++++++++++++++
11 files changed, 226 insertions(+), 28 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org
(spark) 01/02: [SPARK-58042][CONNECT] Validate the UDT jvm_class is a UserDefinedType before instantiating it
Posted by ho...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
commit 84edbdd7e501af6faf478d9b22338a0cd93b69d7
Author: Peter Toth <pe...@gmail.com>
AuthorDate: Wed Jul 8 17:26:25 2026 -0700
[SPARK-58042][CONNECT] Validate the UDT jvm_class is a UserDefinedType before instantiating it
`DataTypeProtoConverter.toCatalystUDT` deserializes a client-supplied `DataType.UDT` proto. When the UDT carries a `jvm_class`, it loaded and instantiated that class with `classForName(jvmClass).getConstructor().newInstance()` and only then relied on the (erased) cast to `UserDefinedType[_]`. So the class's no-arg constructor - and, because `classForName` defaults to `initialize = true`, its static initializers - ran before any type check, and a `jvm_class` that is not a `UserDefinedT [...]
This change loads the class without initializing it, checks `classOf[UserDefinedType[_]].isAssignableFrom(clazz)` first, and throws a clear `InvalidPlanInput` (`CONNECT_INVALID_PLAN.UDT_JVM_CLASS_NOT_UDT`) if it is not a `UserDefinedType`. A valid UDT is instantiated exactly as before.
Only a class that is actually a `UserDefinedType` should be instantiated when converting a UDT proto. Checking the type first gives a clear error for an invalid `jvm_class` instead of a `ClassCastException`, and avoids running the constructor/static initializer of an unrelated class named in the request.
Yes, a small error change: an invalid non-UDT `jvm_class` in a `DataType.UDT` now raises `CONNECT_INVALID_PLAN.UDT_JVM_CLASS_NOT_UDT` instead of a `ClassCastException`.
New test in `InvalidInputErrorsSuite` asserting a non-UserDefinedType `jvm_class` raises `CONNECT_INVALID_PLAN.UDT_JVM_CLASS_NOT_UDT`.
Generated-by: Claude Code (Opus 4.8)
Closes #57130 from peter-toth/SPARK-58042-validate-udt-jvm-class.
Authored-by: Peter Toth <pe...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 372ff4290faddd05eb1254dd698fd3d04b91bcb4)
---
.../src/main/resources/error/error-conditions.json | 5 +++++
.../sql/connect/common/DataTypeProtoConverter.scala | 16 ++++++++++++----
.../sql/connect/planner/InvalidInputErrorsSuite.scala | 18 ++++++++++++++++++
3 files changed, 35 insertions(+), 4 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 43e293ef8013..4e0f2a397319 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -919,6 +919,11 @@
"This typically occurs when building very complex queries with many operations, large literals, or deeply nested expressions.",
"Consider splitting the query into smaller parts using temporary views for intermediate results or reducing the number of operations."
]
+ },
+ "UDT_JVM_CLASS_NOT_UDT" : {
+ "message" : [
+ "The 'jvm_class' <jvmClass> in the UDT is not a subclass of UserDefinedType."
+ ]
}
},
"sqlState" : "56K00"
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
index ac69f084c307..6c24b2b3861b 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
@@ -151,10 +151,18 @@ object DataTypeProtoConverter {
}
if (t.hasJvmClass) {
- SparkClassUtils
- .classForName[UserDefinedType[_]](t.getJvmClass)
- .getConstructor()
- .newInstance()
+ // Verify the class is a UserDefinedType before constructing it. newInstance() runs the
+ // class's no-arg constructor, so load it without initializing and check the type first,
+ // rather than instantiating an arbitrary client-provided class name and relying on a
+ // later cast (which happens only after the constructor has already run).
+ val clazz =
+ SparkClassUtils.classForName[UserDefinedType[_]](t.getJvmClass, initialize = false)
+ if (!classOf[UserDefinedType[_]].isAssignableFrom(clazz)) {
+ throw InvalidPlanInput(
+ "CONNECT_INVALID_PLAN.UDT_JVM_CLASS_NOT_UDT",
+ Map("jvmClass" -> t.getJvmClass))
+ }
+ clazz.getConstructor().newInstance()
} else {
if (!t.hasPythonClass || !t.hasSerializedPythonClass || !t.hasSqlType) {
throw InvalidPlanInput(
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/InvalidInputErrorsSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/InvalidInputErrorsSuite.scala
index 07c377a77df5..548b8a786c66 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/InvalidInputErrorsSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/InvalidInputErrorsSuite.scala
@@ -135,6 +135,24 @@ class InvalidInputErrorsSuite extends PlanTest with SparkConnectPlanTest {
}
}
+ test("SPARK-58042: UDT jvm_class must be a UserDefinedType") {
+ // A jvm_class that is not a UserDefinedType must be rejected before it is instantiated.
+ val udt = proto.DataType.UDT
+ .newBuilder()
+ .setType("udt")
+ .setJvmClass("java.lang.String")
+ .build()
+ val dataType = proto.DataType.newBuilder().setUdt(udt).build()
+ val exception = intercept[InvalidPlanInput] {
+ DataTypeProtoConverter.toCatalystType(dataType)
+ }
+ checkError(
+ exception = exception,
+ condition = "CONNECT_INVALID_PLAN.UDT_JVM_CLASS_NOT_UDT",
+ parameters = Map("jvmClass" -> "java.lang.String"))
+ assert(exception.getSqlState == "56K00")
+ }
+
// Helper case class to define test cases
case class TestCase(
name: String,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org
(spark) 02/02: [SQL] Allow more control over UDT creation during input loading Provide users more control over UDT creation during dynamic loading of different inputs.
Posted by ho...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
commit 9ff88c468c21c8745046c3f22d0b2f43e622eabd
Author: Holden Karau <ho...@pigscanfly.ca>
AuthorDate: Thu Jul 9 14:31:45 2026 -0700
[SQL] Allow more control over UDT creation during input loading
Provide users more control over UDT creation during dynamic loading of different inputs.
Adds two new SQL configs. Default behavior is unchanged, but if configured a class named
in a schema string that is not a white listed will result in a error (UDT_CLASS_NOT_USER_DEFINED_TYPE).
New unit tests in DataTypeSuite (including a regression that a non-UDT class is
rejected even when loading is enabled) and an end-to-end test in ParquetQuerySuite
that the config gates the Parquet-metadata inference path. Existing Parquet/ORC
schema suites and SparkThrowableSuite pass.
Initial fix written by holden and then given to claude to hack on more and then back to holden to clean up
Backport of 0bd5f70
Co-Authored-By: Claude Opus 4.8 (1M context) <no...@anthropic.com>
Co-Authored-By: Holden Karau <ho...@pigscanfly.ca>
Co-Authored-By: Holden Karau <ho...@snowflake.com>
---
.../src/main/resources/error/error-conditions.json | 12 +++++
.../apache/spark/sql/errors/DataTypeErrors.scala | 15 ++++++
.../org/apache/spark/sql/internal/SqlApiConf.scala | 6 +++
.../spark/sql/internal/SqlApiConfHelper.scala | 2 +
.../org/apache/spark/sql/types/DataType.scala | 14 +++++-
.../org/apache/spark/sql/internal/SQLConf.scala | 26 +++++++++++
.../org/apache/spark/sql/types/DataTypeSuite.scala | 53 +++++++++++++++++++++-
.../execution/datasources/SchemaMergeUtils.scala | 51 ++++++++++++---------
.../datasources/parquet/ParquetQuerySuite.scala | 36 +++++++++++++++
9 files changed, 191 insertions(+), 24 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 4e0f2a397319..9f43957df341 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -6022,6 +6022,18 @@
],
"sqlState" : "42802"
},
+ "UDT_CLASS_LOADING_DISABLED" : {
+ "message" : [
+ "Cannot load the class <udtClass> as a user-defined type. Loading UserDefinedType classes by name is disabled by `spark.sql.udt.allowCreatingUDTFromString`, and <udtClass> is not in the allow list `spark.sql.udt.allowedDynamicUDTClasses` (currently <allowed>). Set `spark.sql.udt.allowCreatingUDTFromString` to true, or add the class to the allow list, only if you trust the source of the data being read."
+ ],
+ "sqlState" : "2203G"
+ },
+ "UDT_CLASS_NOT_USER_DEFINED_TYPE" : {
+ "message" : [
+ "The class <udtClass> cannot be loaded as a user-defined type because it is not a subtype of UserDefinedType."
+ ],
+ "sqlState" : "2203G"
+ },
"UNABLE_TO_ACQUIRE_MEMORY" : {
"message" : [
"Unable to acquire <requestedBytes> bytes of memory, got <receivedBytes>."
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala b/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala
index 1e2b2e691cd3..dcad3567bef8 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala
@@ -83,6 +83,21 @@ private[sql] object DataTypeErrors extends DataTypeErrorsBase {
cause = null)
}
+ def udtClassLoadingDisabledError(udtClass: String, allowed: Seq[String]): Throwable = {
+ new SparkException(
+ errorClass = "UDT_CLASS_LOADING_DISABLED",
+ messageParameters =
+ Map("udtClass" -> udtClass, "allowed" -> allowed.map(toSQLValue).mkString(", ")),
+ cause = null)
+ }
+
+ def udtClassNotUserDefinedTypeError(udtClass: String): Throwable = {
+ new SparkException(
+ errorClass = "UDT_CLASS_NOT_USER_DEFINED_TYPE",
+ messageParameters = Map("udtClass" -> udtClass),
+ cause = null)
+ }
+
def unsupportedArrayTypeError(clazz: Class[_]): SparkRuntimeException = {
new SparkRuntimeException(
errorClass = "_LEGACY_ERROR_TEMP_2120",
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala b/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala
index 1be00d75acbc..1970e6b18bfe 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala
@@ -52,6 +52,8 @@ private[sql] trait SqlApiConf {
def parserDfaCacheFlushRatio: Double
def legacyParameterSubstitutionConstantsOnly: Boolean
def legacyIdentifierClauseOnly: Boolean
+ def allowCreatingUDTFromString: Boolean
+ def allowedDynamicUDTClasses: Seq[String]
}
private[sql] object SqlApiConf {
@@ -75,6 +77,8 @@ private[sql] object SqlApiConf {
val PARSER_DFA_CACHE_FLUSH_RATIO_KEY: String =
SqlApiConfHelper.PARSER_DFA_CACHE_FLUSH_RATIO_KEY
val MANAGE_PARSER_CACHES_KEY: String = SqlApiConfHelper.MANAGE_PARSER_CACHES_KEY
+ val ALLOW_CREATING_UDT_FROM_STRING: String = SqlApiConfHelper.ALLOW_CREATING_UDT_FROM_STRING
+ val ALLOWED_DYNAMIC_UDT_CLASSES: String = SqlApiConfHelper.ALLOWED_DYNAMIC_UDT_CLASSES
def get: SqlApiConf = SqlApiConfHelper.getConfGetter.get()()
@@ -108,4 +112,6 @@ private[sql] object DefaultSqlApiConf extends SqlApiConf {
override def parserDfaCacheFlushRatio: Double = -1.0
override def legacyParameterSubstitutionConstantsOnly: Boolean = false
override def legacyIdentifierClauseOnly: Boolean = false
+ override def allowCreatingUDTFromString: Boolean = true
+ override def allowedDynamicUDTClasses: Seq[String] = Nil
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConfHelper.scala b/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConfHelper.scala
index 4fcc2f4e150d..30fc6f4b9a81 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConfHelper.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConfHelper.scala
@@ -42,6 +42,8 @@ private[sql] object SqlApiConfHelper {
"spark.sql.parser.parserDfaCacheFlushThreshold"
val PARSER_DFA_CACHE_FLUSH_RATIO_KEY: String = "spark.sql.parser.parserDfaCacheFlushRatio"
val MANAGE_PARSER_CACHES_KEY: String = "spark.sql.parser.manageParserCaches"
+ val ALLOW_CREATING_UDT_FROM_STRING: String = "spark.sql.udt.allowCreatingUDTFromString"
+ val ALLOWED_DYNAMIC_UDT_CLASSES: String = "spark.sql.udt.allowedDynamicUDTClasses"
val confGetter: AtomicReference[() => SqlApiConf] = {
new AtomicReference[() => SqlApiConf](() => DefaultSqlApiConf)
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
index 48a6514440dd..4b092e0aa116 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
@@ -295,7 +295,19 @@ object DataType {
("pyClass", _),
("sqlType", _),
("type", JString("udt"))) =>
- SparkClassUtils.classForName[UserDefinedType[_]](udtClass).getConstructor().newInstance()
+ if (!SqlApiConf.get.allowCreatingUDTFromString &&
+ !SqlApiConf.get.allowedDynamicUDTClasses.contains(udtClass)) {
+ throw DataTypeErrors.udtClassLoadingDisabledError(
+ udtClass,
+ SqlApiConf.get.allowedDynamicUDTClasses)
+ }
+ // Defense in depth: resolve the class without initializing it and verify that it really is a
+ // UserDefinedType subclass before constructing it.
+ val clazz = SparkClassUtils.classForName[UserDefinedType[_]](udtClass, initialize = false)
+ if (!classOf[UserDefinedType[_]].isAssignableFrom(clazz)) {
+ throw DataTypeErrors.udtClassNotUserDefinedTypeError(udtClass)
+ }
+ clazz.getConstructor().newInstance()
// Python UDT
case JSortedObject(
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 5a1990f8ec65..708903c72897 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -241,6 +241,28 @@ object SQLConf {
}
}
+ val ALLOW_CREATING_UDT_FROM_STRING =
+ buildConf(SqlApiConfHelper.ALLOW_CREATING_UDT_FROM_STRING)
+ .doc("When true, Spark loads and instantiates the UserDefinedType class named in a schema " +
+ "string (for example the schema stored in Parquet/ORC file metadata) while inferring or " +
+ "parsing a schema. Because the class name is taken from the data being read, a crafted " +
+ "file can make Spark load an arbitrary class from the classpath. Set this to false to " +
+ "block loading UDT classes by name, optionally allowing specific classes via " +
+ s"'${SqlApiConfHelper.ALLOWED_DYNAMIC_UDT_CLASSES}'.")
+ .version("4.1.3")
+ .booleanConf
+ .createWithDefault(true)
+
+ val ALLOWED_DYNAMIC_UDT_CLASSES =
+ buildConf(SqlApiConfHelper.ALLOWED_DYNAMIC_UDT_CLASSES)
+ .doc(s"When '${SqlApiConfHelper.ALLOW_CREATING_UDT_FROM_STRING}' is false, UserDefinedType " +
+ "classes listed here (by fully qualified class name) may still be loaded and " +
+ "instantiated from a schema string. Has no effect when UDT loading is enabled.")
+ .version("4.1.3")
+ .stringConf
+ .toSequence
+ .createWithDefault(Nil)
+
val PREFER_COLUMN_OVER_LCA_IN_ARRAY_INDEX =
buildConf("spark.sql.analyzer.preferColumnOverLcaInArrayIndex")
.internal()
@@ -7780,6 +7802,10 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def legacyOutputSchema: Boolean = getConf(SQLConf.LEGACY_KEEP_COMMAND_OUTPUT_SCHEMA)
+ // UDT loading configuration.
+ override def allowCreatingUDTFromString: Boolean = getConf(SQLConf.ALLOW_CREATING_UDT_FROM_STRING)
+ override def allowedDynamicUDTClasses: Seq[String] = getConf(SQLConf.ALLOWED_DYNAMIC_UDT_CLASSES)
+
override def legacyParameterSubstitutionConstantsOnly: Boolean =
getConf(SQLConf.LEGACY_PARAMETER_SUBSTITUTION_CONSTANTS_ONLY)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeSuite.scala
index c698a03d7f34..266b84c101c1 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeSuite.scala
@@ -23,11 +23,13 @@ import org.json4s.jackson.JsonMethods
import org.apache.spark.{SparkException, SparkFunSuite, SparkIllegalArgumentException}
import org.apache.spark.sql.catalyst.analysis.{caseInsensitiveResolution, caseSensitiveResolution}
import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
+import org.apache.spark.sql.catalyst.plans.SQLHelper
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.{CollationFactory, StringConcat}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.DataTypeTestUtils.{dayTimeIntervalTypes, yearMonthIntervalTypes}
-class DataTypeSuite extends SparkFunSuite {
+class DataTypeSuite extends SparkFunSuite with SQLHelper {
private val UNICODE_COLLATION_ID = CollationFactory.collationNameToId("UNICODE")
private val UTF8_LCASE_COLLATION_ID = CollationFactory.collationNameToId("UTF8_LCASE")
@@ -202,6 +204,55 @@ class DataTypeSuite extends SparkFunSuite {
assert(DataType.fromDDL("ts timestamp_ltz") == expectedStructType)
}
+ test("loading a UDT class from a schema string is enabled by default") {
+ val udt = new ExampleBaseTypeUDT()
+ assert(DataType.fromJson(udt.json).isInstanceOf[ExampleBaseTypeUDT])
+ }
+
+ test("loading a UDT class from a schema string can be disabled") {
+ val udt = new ExampleBaseTypeUDT()
+ withSQLConf(SQLConf.ALLOW_CREATING_UDT_FROM_STRING.key -> "false") {
+ checkError(
+ exception = intercept[SparkException] {
+ DataType.fromJson(udt.json)
+ },
+ condition = "UDT_CLASS_LOADING_DISABLED",
+ parameters = Map("udtClass" -> udt.getClass.getName, "allowed" -> ""))
+ }
+ }
+
+ test("disabled UDT loading still honors the allow list") {
+ val udt = new ExampleBaseTypeUDT()
+ withSQLConf(
+ SQLConf.ALLOW_CREATING_UDT_FROM_STRING.key -> "false",
+ SQLConf.ALLOWED_DYNAMIC_UDT_CLASSES.key -> udt.getClass.getName) {
+ assert(DataType.fromJson(udt.json).isInstanceOf[ExampleBaseTypeUDT])
+ }
+ }
+
+ test("a schema string cannot load an arbitrary non-UserDefinedType class") {
+ // Simulate a crafted schema string (e.g. from Parquet file metadata) whose UDT "class" field
+ // points at an arbitrary class that is not a UserDefinedType. Spark must refuse to load and
+ // instantiate it, both when UDT loading is enabled and when the class is explicitly allowed.
+ val gadget = classOf[java.lang.Object].getName
+ val json = s"""{"type":"udt","class":"$gadget","pyClass":null,"sqlType":"integer"}"""
+ Seq(
+ Map(SQLConf.ALLOW_CREATING_UDT_FROM_STRING.key -> "true"),
+ Map(
+ SQLConf.ALLOW_CREATING_UDT_FROM_STRING.key -> "false",
+ SQLConf.ALLOWED_DYNAMIC_UDT_CLASSES.key -> gadget)
+ ).foreach { conf =>
+ withSQLConf(conf.toSeq: _*) {
+ checkError(
+ exception = intercept[SparkException] {
+ DataType.fromJson(json)
+ },
+ condition = "UDT_CLASS_NOT_USER_DEFINED_TYPE",
+ parameters = Map("udtClass" -> gadget))
+ }
+ }
+ }
+
def checkDataTypeFromJson(dataType: DataType): Unit = {
test(s"from Json - $dataType") {
assert(DataType.fromJson(dataType.json) === dataType)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaMergeUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaMergeUtils.scala
index cf0e67ecc30f..d4a744e0efce 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaMergeUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaMergeUtils.scala
@@ -25,7 +25,9 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.FileSourceOptions
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
+import org.apache.spark.sql.classic.ClassicConversions.castToImpl
import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.SQLExecution
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.SerializableConfiguration
@@ -66,33 +68,38 @@ object SchemaMergeUtils extends Logging {
new FileSourceOptions(CaseInsensitiveMap(parameters)).ignoreCorruptFiles
val caseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
- // Issues a Spark job to read Parquet/ORC schema in parallel.
+ // Issues a Spark job to read Parquet/ORC schema in parallel. Propagate the session's SQL
+ // configs to the executors so that reading the schema stored in file metadata (which calls
+ // `DataType.fromJson`) observes session settings such as
+ // `spark.sql.udt.allowCreatingUDTFromString` instead of executor-side defaults.
val partiallyMergedSchemas =
- sparkSession
- .sparkContext
- .parallelize(partialFileStatusInfo, numParallelism)
- .mapPartitions { iterator =>
- // Resembles fake `FileStatus`es with serialized path and length information.
- val fakeFileStatuses = iterator.map { case (path, length) =>
- new FileStatus(length, false, 0, 0, 0, 0, null, null, null, new Path(path))
- }.toSeq
+ SQLExecution.withSQLConfPropagated(sparkSession) {
+ sparkSession
+ .sparkContext
+ .parallelize(partialFileStatusInfo, numParallelism)
+ .mapPartitions { iterator =>
+ // Resembles fake `FileStatus`es with serialized path and length information.
+ val fakeFileStatuses = iterator.map { case (path, length) =>
+ new FileStatus(length, false, 0, 0, 0, 0, null, null, null, new Path(path))
+ }.toSeq
- val schemas = schemaReader(fakeFileStatuses, serializedConf.value, ignoreCorruptFiles)
+ val schemas = schemaReader(fakeFileStatuses, serializedConf.value, ignoreCorruptFiles)
- if (schemas.isEmpty) {
- Iterator.empty
- } else {
- var mergedSchema = schemas.head
- schemas.tail.foreach { schema =>
- try {
- mergedSchema = mergedSchema.merge(schema, caseSensitive)
- } catch { case cause: SparkException =>
- throw QueryExecutionErrors.failedMergingSchemaError(mergedSchema, schema, cause)
+ if (schemas.isEmpty) {
+ Iterator.empty
+ } else {
+ var mergedSchema = schemas.head
+ schemas.tail.foreach { schema =>
+ try {
+ mergedSchema = mergedSchema.merge(schema, caseSensitive)
+ } catch { case cause: SparkException =>
+ throw QueryExecutionErrors.failedMergingSchemaError(mergedSchema, schema, cause)
+ }
}
+ Iterator.single(mergedSchema)
}
- Iterator.single(mergedSchema)
- }
- }.collect()
+ }.collect()
+ }
if (partiallyMergedSchemas.isEmpty) {
None
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetQuerySuite.scala
index c530dc0d3dfa..97aad5b30eaf 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetQuerySuite.scala
@@ -793,6 +793,42 @@ abstract class ParquetQuerySuite extends QueryTest with ParquetTest with SharedS
}
}
+ test("loading UDT classes named in Parquet metadata respects the UDT allow list") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ val udtClass = classOf[TestNestedStructUDT].getName
+ val schema = new StructType().add("s", new TestNestedStructUDT, nullable = true)
+ val data = Seq(Row(TestNestedStruct(1, 2L, 3.5D)))
+ // Writing a UDT column embeds the UDT class name in the Parquet key-value metadata, which is
+ // read back and passed to DataType.fromJson during schema inference (the vulnerable path).
+ spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+ .coalesce(1)
+ .write
+ .parquet(path)
+
+ def inferredColumnType: DataType = spark.read.parquet(path).schema("s").dataType
+
+ // By default the UDT class named in the file metadata is loaded during schema inference.
+ assert(inferredColumnType.isInstanceOf[TestNestedStructUDT])
+
+ // With UDT loading disabled and the class not on the allow list, Spark must not load the
+ // class named in the file. Schema inference falls back to the underlying physical schema
+ // rather than instantiating the attacker-named class.
+ withSQLConf(SQLConf.ALLOW_CREATING_UDT_FROM_STRING.key -> "false") {
+ assert(!inferredColumnType.isInstanceOf[TestNestedStructUDT])
+ assert(!inferredColumnType.isInstanceOf[UserDefinedType[_]])
+ assert(inferredColumnType.isInstanceOf[StructType])
+ }
+
+ // Explicitly allow-listing the class restores UDT resolution end to end.
+ withSQLConf(
+ SQLConf.ALLOW_CREATING_UDT_FROM_STRING.key -> "false",
+ SQLConf.ALLOWED_DYNAMIC_UDT_CLASSES.key -> udtClass) {
+ assert(inferredColumnType.isInstanceOf[TestNestedStructUDT])
+ }
+ }
+ }
+
testStandardAndLegacyModes("SPARK-39086: UDT read support in vectorized reader") {
withTempPath { dir =>
val path = dir.getCanonicalPath
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org