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 <[email protected]> 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) <[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 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: [email protected] For additional commands, e-mail: [email protected]
