This is an automated email from the ASF dual-hosted git repository.

asf-gitbox-commits 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 0bd5f7047b4f [SQL] Allow more control over UDT creation during input 
loading
0bd5f7047b4f is described below

commit 0bd5f7047b4feae13b0259c42cc22b23c05e41fe
Author: Holden Karau <[email protected]>
AuthorDate: Thu Jul 2 13:22:47 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
    
    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 | 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    | 28 ++++++++++++
 .../org/apache/spark/sql/types/DataTypeSuite.scala | 49 ++++++++++++++++++++
 .../connect/common/DataTypeProtoConverter.scala    |  4 +-
 .../connect/planner/InvalidInputErrorsSuite.scala  |  6 +--
 .../execution/datasources/SchemaMergeUtils.scala   | 53 ++++++++++++----------
 .../datasources/parquet/ParquetQuerySuite.scala    | 36 +++++++++++++++
 11 files changed, 196 insertions(+), 34 deletions(-)

diff --git a/common/utils/src/main/resources/error/error-conditions.json 
b/common/utils/src/main/resources/error/error-conditions.json
index 4c5cf5fe5d69..66b96c9b8657 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -1450,11 +1450,6 @@
           "Struct literal is not set."
         ]
       },
-      "UDT_JVM_CLASS_NOT_UDT" : {
-        "message" : [
-          "The 'jvm_class' <jvmClass> in the UDT is not a subclass of 
UserDefinedType."
-        ]
-      },
       "UDT_TYPE_FIELD_INVALID" : {
         "message" : [
           "UserDefinedType requires the 'type' field to be 'udt', but got 
'<udtType>'."
@@ -7701,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/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 62ff4085bc73..15b966ced2ca 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
@@ -159,8 +159,8 @@ object DataTypeProtoConverter {
         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))
+          "UDT_CLASS_NOT_USER_DEFINED_TYPE",
+          Map("udtClass" -> t.getJvmClass))
       }
       clazz.getConstructor().newInstance()
     } else {
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 64866ad6ec66..108fac3ea634 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
@@ -152,9 +152,9 @@ class InvalidInputErrorsSuite extends PlanTest with 
SparkConnectPlanTest {
     }
     checkError(
       exception = exception,
-      condition = "CONNECT_INVALID_PLAN.UDT_JVM_CLASS_NOT_UDT",
-      parameters = Map("jvmClass" -> "java.lang.String"))
-    assert(exception.getSqlState == "56K00")
+      condition = "UDT_CLASS_NOT_USER_DEFINED_TYPE",
+      parameters = Map("udtClass" -> "java.lang.String"))
+    assert(exception.getSqlState == "2203G")
   }
 
   test("noHandlerFoundForExtension") {
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]

Reply via email to