This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/spark.git
commit c40f04483a4e0c2d3c9812a133ad7dec3bd99c5c Author: Holden Karau <[email protected]> AuthorDate: Thu Jul 9 14:21:11 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 | 53 +++++++++++++++++++++- .../execution/datasources/SchemaMergeUtils.scala | 53 ++++++++++++---------- .../datasources/parquet/ParquetQuerySuite.scala | 36 +++++++++++++++ 9 files changed, 194 insertions(+), 25 deletions(-) diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index 07f369f1c8c3..527422b946fe 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -7401,6 +7401,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 bedd4afe0ed5..3347c9d3446b 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 typesFrameworkEnabled: 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 typesFrameworkEnabled: 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 60619284c99b..688ca93320bd 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 @@ -250,6 +250,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() @@ -8622,6 +8646,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 ce4f5e89be2b..0154b02da7c4 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, PhysicalDataType, UninitializedPhysicalType} 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 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 b20b6d397fd1..38582236c738 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 @@ -861,6 +861,42 @@ abstract class ParquetQuerySuite extends ParquetTest with SharedSparkSession { } } + 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]
