This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits 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 fbf25be889d8 [SQL] Allow more control over UDT creation during input
loading
fbf25be889d8 is described below
commit fbf25be889d8dde37bcf82fbb6185c598d4251d6
Author: Holden Karau <[email protected]>
AuthorDate: Thu Jul 9 14:47:09 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 0bd5f7047b4feae13b0259c42cc22b23c05e41fe
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Co-Authored-By: Holden Karau <[email protected]>
Co-Authored-By: Holden Karau <[email protected]>
---
.../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 | 28 ++++++++++++
.../org/apache/spark/sql/types/DataTypeSuite.scala | 49 ++++++++++++++++++++
.../execution/datasources/SchemaMergeUtils.scala | 53 ++++++++++++----------
.../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 eb8a0aa15fb4..fa2712d4b720 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -7696,6 +7696,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 955e242d4ab5..e7c5d329e404 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
@@ -84,6 +84,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 b101ee0563a1..7538580fb234 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
@@ -54,6 +54,8 @@ private[sql] trait SqlApiConf {
def legacyParameterSubstitutionConstantsOnly: Boolean
def legacyIdentifierClauseOnly: Boolean
def timestampNanosTypesEnabled: Boolean
+ def allowCreatingUDTFromString: Boolean
+ def allowedDynamicUDTClasses: Seq[String]
}
private[sql] object SqlApiConf {
@@ -77,6 +79,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()()
@@ -112,4 +116,6 @@ private[sql] object DefaultSqlApiConf extends SqlApiConf {
override def legacyParameterSubstitutionConstantsOnly: Boolean = false
override def legacyIdentifierClauseOnly: Boolean = false
override def timestampNanosTypesEnabled: Boolean = SparkEnvUtils.isTesting
+ 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 319d600788fa..8bce7b50f5d8 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
@@ -337,7 +337,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 a3f0fd8ab43e..22f5c8416e05 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
@@ -251,6 +251,30 @@ 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")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .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")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .stringConf
+ .toSequence
+ .createWithDefault(Nil)
+
val PREFER_COLUMN_OVER_LCA_IN_ARRAY_INDEX =
buildConf("spark.sql.analyzer.preferColumnOverLcaInArrayIndex")
.internal()
@@ -9044,6 +9068,10 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def sessionFunctionResolutionOrder: String =
getConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER)
+ // 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)
+
/**
* Returns true when the system catalog is prioritized for 2-part
builtin/session resolution.
* This is the inverse of [[SQLConf.PERSISTENT_CATALOG_FIRST]].
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 8b90f618dc6d..6282832420b7 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
@@ -206,6 +206,55 @@ class DataTypeSuite extends SparkFunSuite with SQLHelper {
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 a04bfbc67b58..c43ecb199bea 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
@@ -67,34 +69,39 @@ object SchemaMergeUtils extends Logging {
val ignoreMissingFiles = fileSourceOptions.ignoreMissingFiles
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,
ignoreMissingFiles)
+ val schemas = schemaReader(
+ fakeFileStatuses, serializedConf.value, ignoreCorruptFiles,
ignoreMissingFiles)
- 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 5c2bc6829ea5..e36bb50416f3 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
@@ -863,6 +863,42 @@ abstract class ParquetQuerySuite extends ParquetTest
}
}
+ 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: [email protected]
For additional commands, e-mail: [email protected]
