This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch branch-4.0 in repository https://gitbox.apache.org/repos/asf/spark.git
commit a5395f7ec3b05958e80528128d5e8d3bfd453d01 Author: Holden Karau <[email protected]> AuthorDate: Thu Jul 9 14:39:44 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) <[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 | 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 ff899acf4571..10c172d2ddac 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -5166,6 +5166,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 6e13e78c41d6..9adcc9ee481e 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 76449f1704d2..4c9752505a5d 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 @@ -47,6 +47,8 @@ private[sql] trait SqlApiConf { def stackTracesInDataFrameContext: Int def dataFrameQueryContextEnabled: Boolean def legacyAllowUntypedScalaUDFs: Boolean + def allowCreatingUDTFromString: Boolean + def allowedDynamicUDTClasses: Seq[String] } private[sql] object SqlApiConf { @@ -60,6 +62,8 @@ private[sql] object SqlApiConf { val LOCAL_RELATION_CACHE_THRESHOLD_KEY: String = { SqlApiConfHelper.LOCAL_RELATION_CACHE_THRESHOLD_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()() @@ -88,4 +92,6 @@ private[sql] object DefaultSqlApiConf extends SqlApiConf { override def stackTracesInDataFrameContext: Int = 1 override def dataFrameQueryContextEnabled: Boolean = true override def legacyAllowUntypedScalaUDFs: 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 dace1dbaecfa..5655a06ee423 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 @@ -33,6 +33,8 @@ private[sql] object SqlApiConfHelper { val SESSION_LOCAL_TIMEZONE_KEY: String = "spark.sql.session.timeZone" val LOCAL_RELATION_CACHE_THRESHOLD_KEY: String = "spark.sql.session.localRelationCacheThreshold" val ARROW_EXECUTION_USE_LARGE_VAR_TYPES = "spark.sql.execution.arrow.useLargeVarTypes" + 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 9664cc7704e9..8d5c04ab75c2 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 @@ -280,7 +280,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 8427c8320d5c..796ea1087fee 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 @@ -240,6 +240,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.0.4") + .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.0.4") + .stringConf + .toSequence + .createWithDefault(Nil) + val ANALYZER_MAX_ITERATIONS = buildConf("spark.sql.analyzer.maxIterations") .internal() .doc("The max number of iterations the analyzer runs.") @@ -6604,6 +6626,10 @@ class SQLConf extends Serializable with Logging with SqlApiConf { def legacyEvalCurrentTime: Boolean = getConf(SQLConf.LEGACY_EVAL_CURRENT_TIME) + // 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) + /** ********************** SQLConf functionality methods ************ */ /** Set Spark SQL configuration properties. */ 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 d1c83fa43f02..7122adc98a30 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 +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 bba71f1c48de..68410dbedb75 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: [email protected] For additional commands, e-mail: [email protected]
