dtenedor commented on code in PR #59640:
URL: https://github.com/apache/spark/pull/59640#discussion_r4382011865
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSchemaEvolution.scala:
##########
@@ -49,6 +49,14 @@ object ResolveSchemaEvolution extends Rule[LogicalPlan] {
case write: WriteWithSchemaEvolution if
write.pendingSchemaChanges.nonEmpty =>
write.table match {
case relation @ ExtractV2CatalogAndIdentifier(catalog, ident) =>
+ // A table cannot store a FILE without a qualifier, so schema
evolution cannot add one.
+ // This is checked before the catalog changes the table.
+
TypeUtils.checkNoUnknownFileTypeInColumns(write.pendingSchemaChanges.collect {
Review Comment:
**Low: writes to tables with `ACCEPT_ANY_SCHEMA` can still carry a bare
FILE.** `V2WriteCommand.outputResolved` short-circuits when the table skips
schema resolution (`v2Commands.scala` L126-131), and no FILE check covers V2
writes, so a bare FILE column reaches the connector, and a connector that
evolves its schema from incoming data (Delta-style `mergeSchema`) would store
it. It's latent upstream, where only `noop` declares that capability, so
tracking it with the pipelines follow-up seems fine.
##########
sql/core/src/test/scala/org/apache/spark/sql/FileTypeRestrictionsSuite.scala:
##########
@@ -0,0 +1,501 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.SparkConf
+import org.apache.spark.api.python.{PythonBroadcast, PythonEvalType,
SimplePythonFunction}
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.sql.api.java.UDF1
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
+import org.apache.spark.sql.connector.catalog.InMemoryCatalog
+import org.apache.spark.sql.execution.python.{UserDefinedPythonFunction,
UserDefinedPythonTableFunction}
+import org.apache.spark.sql.expressions.{MutableAggregationBuffer,
UserDefinedAggregateFunction}
+import org.apache.spark.sql.functions.{col, udf}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DataType, FileType, IntegerType, MapType,
StructType}
+
+/**
+ * Tests the rules on where the FILE type can be used: table columns must be
FILE EXTERNAL,
+ * routines declare a plain FILE, and FILE has no defined ordering.
+ */
+class FileTypeRestrictionsSuite extends QueryTest with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf =
+ super.sparkConf.set("spark.sql.catalog.testcat",
classOf[InMemoryCatalog].getName)
+
+ // A FILE without a qualifier, at the top level and nested.
+ private val unqualifiedFileTypes = Seq(
+ "FILE",
+ "ARRAY<FILE>",
+ "MAP<STRING, FILE>",
+ "STRUCT<f1: FILE>",
+ "STRUCT<f1: INT, f2: STRUCT<f3: ARRAY<FILE>>>")
+
+ private val externalFile = "SELECT CAST(NULL AS FILE EXTERNAL) AS f"
+
+ test("a table column definition requires FILE EXTERNAL") {
+ Seq("testcat.t", "t").foreach { table =>
+ val provider = if (table.startsWith("testcat")) "" else "USING parquet"
+ unqualifiedFileTypes.foreach { dataType =>
+ withTable(table) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $table (id INT, c $dataType) $provider")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ }
+ }
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT, c STRUCT<f1: ARRAY<FILE
EXTERNAL>>)")
+ }
+ }
+
+ test("REPLACE TABLE requires FILE EXTERNAL") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT)")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("REPLACE TABLE testcat.t (id INT, c FILE)")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ }
+
+ test("ALTER TABLE ADD COLUMNS and REPLACE COLUMNS require FILE EXTERNAL") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT)")
+ unqualifiedFileTypes.foreach { dataType =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t ADD COLUMN c $dataType")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t REPLACE COLUMNS (id INT, c $dataType)")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ sql("ALTER TABLE testcat.t ADD COLUMN c FILE EXTERNAL")
+ }
+ }
+
+ test("CREATE TABLE AS SELECT cannot store a FILE without a qualifier") {
+ Seq("testcat.t", "t").foreach { table =>
+ val provider = if (table.startsWith("testcat")) "" else "USING parquet"
+ withTable(table) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $table $provider AS SELECT 1 AS id, CAST(NULL
AS FILE) AS f")
+ },
+ condition = "FILE_TYPE_UNKNOWN_WRITE",
+ parameters = Map("colNames" -> "[`f`]"))
+ }
+ }
+ }
+
+ test("schema evolution cannot add a FILE without a qualifier") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT)")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("INSERT WITH SCHEMA EVOLUTION INTO testcat.t " +
+ "SELECT 1 AS id, CAST(NULL AS FILE) AS f")
+ },
+ condition = "FILE_TYPE_UNKNOWN_WRITE",
+ parameters = Map("colNames" -> "[`f`]"))
+ // The table is not changed.
+ assert(spark.table("testcat.t").schema === new StructType().add("id",
IntegerType))
+ }
+ }
+
+ test("CREATE TABLE LIKE cannot copy a FILE without a qualifier from a view")
{
+ withTempView("v") {
+ withView("pv") {
+ sql("CREATE TEMPORARY VIEW v AS SELECT 1 AS id, CAST(NULL AS FILE) AS
f")
+ sql("CREATE VIEW pv AS SELECT 1 AS id, CAST(NULL AS FILE) AS f")
+ Seq("v" -> "t", "pv" -> "t", "v" -> "testcat.t").foreach { case
(source, target) =>
+ withTable(target) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $target LIKE $source")
+ },
+ condition = "FILE_TYPE_UNKNOWN_WRITE",
+ parameters = Map("colNames" -> "[`f`]"))
+ assert(!spark.catalog.tableExists(target))
+ }
+ }
+ }
+ }
+ }
+
+ test("a FILE column cannot be a partition or clustering column") {
+ Seq(
+ ("PARTITIONED BY (f)", "PARTITION_COLUMN"),
+ ("PARTITIONED BY (bucket(4, f))", "PARTITION_COLUMN"),
+ ("CLUSTERED BY (f) INTO 4 BUCKETS", "PARTITION_COLUMN"),
+ ("CLUSTER BY (f)", "CLUSTERING_COLUMN")).foreach { case (clause, usage)
=>
+ Seq("testcat.t", "t").foreach { table =>
+ // `bucket` is a V2 transform, and CLUSTERED BY ... INTO ... BUCKETS
is V1 syntax.
+ val isV2 = table.startsWith("testcat")
+ if (isV2 != clause.startsWith("CLUSTERED BY")) {
+ val provider = if (isV2) "" else "USING parquet"
+ withTable(table) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $table (id INT, f FILE EXTERNAL) $provider
$clause")
+ },
+ condition = s"FILE_TYPE_UNSUPPORTED_COLUMN_USAGE.$usage",
+ parameters = Map("colName" -> "`f`", "dataType" -> "\"FILE
EXTERNAL\""))
+ if (!clause.contains("bucket")) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $table $provider $clause AS
$externalFile")
+ },
+ condition = s"FILE_TYPE_UNSUPPORTED_COLUMN_USAGE.$usage",
+ parameters = Map("colName" -> "`f`", "dataType" -> "\"FILE
EXTERNAL\""))
+ }
+ }
+ }
+ }
+ }
+ Seq("testcat.t", "t").foreach { table =>
+ withTable(table) {
+ if (table.startsWith("testcat")) {
+ sql(s"CREATE TABLE $table (id INT, f FILE EXTERNAL) CLUSTER BY (id)")
+ } else {
+ // No V1 data source can store a FILE yet, so the table is created
in the catalog.
+ spark.sessionState.catalog.createTable(CatalogTable(
+ identifier = TableIdentifier(table),
+ tableType = CatalogTableType.MANAGED,
+ storage = CatalogStorageFormat.empty,
+ schema = new StructType().add("id", IntegerType).add("f",
FileType.external),
+ provider = Some("parquet")), ignoreIfExists = false)
+ }
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE $table CLUSTER BY (f)")
+ },
+ condition = "FILE_TYPE_UNSUPPORTED_COLUMN_USAGE.CLUSTERING_COLUMN",
+ parameters = Map("colName" -> "`f`", "dataType" -> "\"FILE
EXTERNAL\""))
+ }
+ }
+ }
+
+ test("a column cannot be altered to a FILE") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT, s STRING, f FILE EXTERNAL, v VOID)")
+ Seq("s", "f", "v").foreach { column =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t ALTER COLUMN $column TYPE FILE")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("ALTER TABLE testcat.t ALTER COLUMN s TYPE FILE EXTERNAL")
+ },
+ condition = "NOT_SUPPORTED_CHANGE_COLUMN",
+ parameters = Map(
+ "table" -> "`testcat`.`t`",
+ "originName" -> "`s`",
+ "originType" -> "\"STRING\"",
+ "newName" -> "`s`",
+ "newType" -> "\"FILE EXTERNAL\""),
+ context = ExpectedContext(
+ fragment = "ALTER TABLE testcat.t ALTER COLUMN s TYPE FILE EXTERNAL",
+ start = 0,
+ stop = 54))
+ }
+ }
+
+ test("a FILE cannot be a map key") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"SELECT map(f, 1) FROM ($externalFile)")
+ },
+ condition = "DATATYPE_MISMATCH.INVALID_MAP_KEY_TYPE",
+ parameters = Map("sqlExpr" -> "\"map(f, 1)\"", "keyType" -> "\"FILE
EXTERNAL\""),
+ context = ExpectedContext(fragment = "map(f, 1)", start = 7, stop = 15))
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"SELECT map(array(f), 1) FROM ($externalFile)")
+ },
+ condition = "DATATYPE_MISMATCH.INVALID_MAP_KEY_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"map(array(f), 1)\"",
+ "keyType" -> "\"ARRAY<FILE EXTERNAL>\""),
+ context = ExpectedContext(fragment = "map(array(f), 1)", start = 7, stop
= 22))
+ sql(s"SELECT map(1, f) FROM ($externalFile)")
+ }
+
+ test("a FILE cannot be a grouping expression") {
+ val query = s"SELECT count(*) FROM ($externalFile) GROUP BY f"
+ checkError(
+ exception = intercept[AnalysisException](sql(query)),
+ condition = "GROUP_EXPRESSION_TYPE_IS_NOT_ORDERABLE",
+ parameters = Map("sqlExpr" -> "\"f\"", "dataType" -> "\"FILE
EXTERNAL\""),
+ context = ExpectedContext(
+ fragment = "f", start = query.length - 1, stop = query.length - 1))
+ val nested = s"SELECT count(*) FROM ($externalFile) GROUP BY struct(f)"
+ checkError(
+ exception = intercept[AnalysisException](sql(nested)),
+ condition = "GROUP_EXPRESSION_TYPE_IS_NOT_ORDERABLE",
+ parameters = Map(
+ "sqlExpr" -> "\"struct(f)\"",
+ "dataType" -> "\"STRUCT<f: FILE EXTERNAL>\""),
+ context = ExpectedContext(
+ fragment = "struct(f)", start = nested.length - 9, stop =
nested.length - 1))
+ }
+
+ test("a FILE cannot be in INTERSECT, EXCEPT or DISTINCT") {
+ Seq(
+ s"$externalFile INTERSECT $externalFile",
+ s"$externalFile EXCEPT $externalFile",
+ s"$externalFile UNION $externalFile",
+ s"SELECT DISTINCT f FROM ($externalFile)").foreach { query =>
+ checkError(
+ exception = intercept[AnalysisException](sql(query)),
+ condition = "UNSUPPORTED_FEATURE.SET_OPERATION_ON_FILE_TYPE",
+ parameters = Map("colName" -> "`f`", "dataType" -> "\"FILE
EXTERNAL\""),
+ context = ExpectedContext(fragment = query, start = 0, stop =
query.length - 1))
+ }
+ val nested = s"SELECT DISTINCT struct(f) AS s FROM ($externalFile)"
+ checkError(
+ exception = intercept[AnalysisException](sql(nested)),
+ condition = "UNSUPPORTED_FEATURE.SET_OPERATION_ON_FILE_TYPE",
+ parameters = Map("colName" -> "`s`", "dataType" -> "\"STRUCT<f: FILE
EXTERNAL>\""),
+ context = ExpectedContext(fragment = nested, start = 0, stop =
nested.length - 1))
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(externalFile).dropDuplicates("f")
+ },
+ condition = "UNSUPPORTED_FEATURE.SET_OPERATION_ON_FILE_TYPE",
+ parameters = Map("colName" -> "`f`", "dataType" -> "\"FILE EXTERNAL\""))
+ // UNION ALL does not compare rows.
+ sql(s"$externalFile UNION ALL $externalFile")
+ }
+
+ test("a FILE cannot be a repartition expression") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(externalFile).repartition(5, col("f"))
+ },
+ condition = "UNSUPPORTED_FEATURE.PARTITION_BY_FILE",
+ parameters = Map("expr" -> "\"f\"", "dataType" -> "\"FILE EXTERNAL\""))
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"SELECT /*+ REBALANCE(f) */ * FROM ($externalFile)")
+ },
+ condition = "UNSUPPORTED_FEATURE.PARTITION_BY_FILE",
+ parameters = Map("expr" -> "\"f\"", "dataType" -> "\"FILE EXTERNAL\""),
+ context = ExpectedContext(fragment = "/*+ REBALANCE(f) */", start = 7,
stop = 25))
+ // Partitioning by a column that is not a FILE is accepted.
+ sql("SELECT * FROM (SELECT 1 AS id, CAST(NULL AS FILE EXTERNAL) AS f)
DISTRIBUTE BY id")
+ }
+
+ test("a FILE cannot be a key of dropDuplicatesWithinWatermark") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(externalFile).dropDuplicatesWithinWatermark("f")
+ },
+ condition = "UNSUPPORTED_FEATURE.SET_OPERATION_ON_FILE_TYPE",
+ parameters = Map("colName" -> "`f`", "dataType" -> "\"FILE EXTERNAL\""))
+ }
+
+ test("a FILE cannot be hashed") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"SELECT hash(f) FROM ($externalFile)")
+ },
+ condition = "DATATYPE_MISMATCH.HASH_FILE_TYPE",
+ parameters = Map("sqlExpr" -> "\"hash(f)\"", "functionName" -> "`hash`"),
+ context = ExpectedContext(fragment = "hash(f)", start = 7, stop = 13))
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"SELECT hash(array(f)) FROM ($externalFile)")
+ },
+ condition = "DATATYPE_MISMATCH.HASH_FILE_TYPE",
+ parameters = Map("sqlExpr" -> "\"hash(array(f))\"", "functionName" ->
"`hash`"),
+ context = ExpectedContext(fragment = "hash(array(f))", start = 7, stop =
20))
+ }
+
+ test("a FILE cannot be sorted or be a window partition") {
+ val orderBy = s"SELECT f FROM ($externalFile) ORDER BY f"
+ checkError(
+ exception = intercept[AnalysisException](sql(orderBy)),
+ condition = "DATATYPE_MISMATCH.INVALID_ORDERING_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"f ASC NULLS FIRST\"",
+ "functionName" -> "`sortorder`",
+ "dataType" -> "\"FILE EXTERNAL\""),
+ context = ExpectedContext(
+ fragment = "f", start = orderBy.length - 1, stop = orderBy.length - 1))
+ val windowOrder = s"SELECT row_number() OVER (ORDER BY f) FROM
($externalFile)"
+ checkError(
+ exception = intercept[AnalysisException](sql(windowOrder)),
+ condition = "DATATYPE_MISMATCH.INVALID_ORDERING_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"f ASC NULLS FIRST\"",
+ "functionName" -> "`sortorder`",
+ "dataType" -> "\"FILE EXTERNAL\""),
+ context = ExpectedContext(fragment = "f", start = 35, stop = 35))
+ val windowPartition = s"SELECT count(*) OVER (PARTITION BY f) FROM
($externalFile)"
+ checkError(
+ exception = intercept[AnalysisException](sql(windowPartition)),
+ condition = "EXPRESSION_TYPE_IS_NOT_ORDERABLE",
+ parameters = Map("expr" -> "\"f\"", "exprType" -> "\"FILE EXTERNAL\""),
+ context = ExpectedContext(fragment = "f", start = 35, stop = 35))
+ }
+
+ test("a FILE cannot be compared, so it cannot be a join key") {
+ val query = s"SELECT 1 FROM ($externalFile) l JOIN ($externalFile) r ON
l.f = r.f"
+ checkError(
+ exception = intercept[AnalysisException](sql(query)),
+ condition = "DATATYPE_MISMATCH.INVALID_ORDERING_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"(f = f)\"",
+ "functionName" -> "`=`",
+ "dataType" -> "\"FILE EXTERNAL\""),
+ context = ExpectedContext(
+ fragment = "l.f = r.f", start = query.length - 9, stop = query.length
- 1))
+ }
+
+ test("a SQL function declares a plain FILE") {
+ Seq(
+ "(x FILE EXTERNAL) RETURNS INT RETURN 1",
+ "(x INT) RETURNS FILE EXTERNAL RETURN NULL",
+ "(x INT) RETURNS ARRAY<FILE EXTERNAL> RETURN NULL",
+ "(x INT) RETURNS TABLE (f FILE EXTERNAL) RETURN SELECT NULL",
+ // Without a RETURNS clause, or with a bare RETURNS TABLE, the return
type is inferred.
+ "(x INT) RETURN CAST(NULL AS FILE EXTERNAL)",
+ "(x INT) RETURNS TABLE RETURN SELECT CAST(NULL AS FILE EXTERNAL) AS
f").foreach { signature =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TEMPORARY FUNCTION fn$signature")
+ },
+ condition = "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_QUALIFIER",
+ parameters = Map("routineType" -> "function", "routineName" -> "`fn`"))
+ }
+ Seq(
+ "RETURNS MAP<FILE, INT> RETURN NULL",
+ "RETURN CAST(NULL AS MAP<FILE, INT>)",
+ "RETURNS TABLE (s STRUCT<m: MAP<FILE, INT>>) RETURN SELECT
NULL").foreach { body =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TEMPORARY FUNCTION fn(x INT) $body")
+ },
+ condition = "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_MAP_KEY",
+ parameters = Map("routineType" -> "function", "routineName" -> "`fn`"))
+ }
+ withUserDefinedFunction("fn" -> true) {
+ sql("CREATE TEMPORARY FUNCTION fn(x FILE) RETURNS FILE RETURN x")
+ }
+ }
+
+ test("a UDF declares a plain FILE in its return type") {
+ val fileUdf = new UDF1[Integer, AnyRef] {
+ override def call(x: Integer): AnyRef = null
+ }
+ withUserDefinedFunction("file_udf" -> true, "file_udaf" -> true) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.udf.register("file_udf", fileUdf, FileType.external)
+ },
+ condition = "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_QUALIFIER",
+ parameters = Map("routineType" -> "function", "routineName" ->
"`file_udf`"))
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.udf.register("file_udf", fileUdf, MapType(FileType(),
IntegerType))
+ },
+ condition = "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_MAP_KEY",
+ parameters = Map("routineType" -> "function", "routineName" ->
"`file_udf`"))
+ // A UDF that is not registered is checked when it is used in a query.
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.range(1).select(udf(fileUdf,
FileType.external).withName("inline_udf")(col("id")))
+ },
+ condition = "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_QUALIFIER",
+ parameters = Map("routineType" -> "function", "routineName" ->
"`inline_udf`"))
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.udf.register("file_udaf", new FileTypeUDAF(FileType.external))
+ },
+ condition = "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_QUALIFIER",
+ parameters = Map("routineType" -> "function", "routineName" ->
"`file_udaf`"))
+ spark.udf.register("file_udf", fileUdf, FileType())
+ spark.udf.register("file_udaf", new FileTypeUDAF(FileType()))
+ }
+ }
+
+ test("a Python UDF and UDTF declare a plain FILE in their return type") {
Review Comment:
**Low: the query-time Python checks don't need a Python worker.** Classic
DataFrame analysis is eager and a `PythonUDF` is only executed after planning,
so `spark.range(1).select(pythonUdf(FileType.external)(col("id")))` fails in
`CheckAnalysis` without starting Python. The same goes for a UDAF class
registered with `CREATE TEMPORARY FUNCTION f AS '<class>'`, for which this
query-time check is the only enforcement. Suggest adding both cases (the
encoder reason does hold for a typed `Aggregator`).
##########
sql/core/src/test/scala/org/apache/spark/sql/FileTypeRestrictionsSuite.scala:
##########
@@ -0,0 +1,501 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.SparkConf
+import org.apache.spark.api.python.{PythonBroadcast, PythonEvalType,
SimplePythonFunction}
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.sql.api.java.UDF1
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
+import org.apache.spark.sql.connector.catalog.InMemoryCatalog
+import org.apache.spark.sql.execution.python.{UserDefinedPythonFunction,
UserDefinedPythonTableFunction}
+import org.apache.spark.sql.expressions.{MutableAggregationBuffer,
UserDefinedAggregateFunction}
+import org.apache.spark.sql.functions.{col, udf}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DataType, FileType, IntegerType, MapType,
StructType}
+
+/**
+ * Tests the rules on where the FILE type can be used: table columns must be
FILE EXTERNAL,
+ * routines declare a plain FILE, and FILE has no defined ordering.
+ */
+class FileTypeRestrictionsSuite extends QueryTest with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf =
+ super.sparkConf.set("spark.sql.catalog.testcat",
classOf[InMemoryCatalog].getName)
+
+ // A FILE without a qualifier, at the top level and nested.
+ private val unqualifiedFileTypes = Seq(
+ "FILE",
+ "ARRAY<FILE>",
+ "MAP<STRING, FILE>",
+ "STRUCT<f1: FILE>",
+ "STRUCT<f1: INT, f2: STRUCT<f3: ARRAY<FILE>>>")
+
+ private val externalFile = "SELECT CAST(NULL AS FILE EXTERNAL) AS f"
+
+ test("a table column definition requires FILE EXTERNAL") {
+ Seq("testcat.t", "t").foreach { table =>
+ val provider = if (table.startsWith("testcat")) "" else "USING parquet"
+ unqualifiedFileTypes.foreach { dataType =>
+ withTable(table) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $table (id INT, c $dataType) $provider")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ }
+ }
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT, c STRUCT<f1: ARRAY<FILE
EXTERNAL>>)")
+ }
+ }
+
+ test("REPLACE TABLE requires FILE EXTERNAL") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT)")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("REPLACE TABLE testcat.t (id INT, c FILE)")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ }
+
+ test("ALTER TABLE ADD COLUMNS and REPLACE COLUMNS require FILE EXTERNAL") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT)")
+ unqualifiedFileTypes.foreach { dataType =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t ADD COLUMN c $dataType")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t REPLACE COLUMNS (id INT, c $dataType)")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ sql("ALTER TABLE testcat.t ADD COLUMN c FILE EXTERNAL")
+ }
+ }
+
+ test("CREATE TABLE AS SELECT cannot store a FILE without a qualifier") {
+ Seq("testcat.t", "t").foreach { table =>
+ val provider = if (table.startsWith("testcat")) "" else "USING parquet"
+ withTable(table) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $table $provider AS SELECT 1 AS id, CAST(NULL
AS FILE) AS f")
+ },
+ condition = "FILE_TYPE_UNKNOWN_WRITE",
+ parameters = Map("colNames" -> "[`f`]"))
+ }
+ }
+ }
+
+ test("schema evolution cannot add a FILE without a qualifier") {
Review Comment:
**Low: the new write-path tests miss a few cases.** The schema-evolution
test only covers a top-level INSERT, so nothing pins the nested `` `s`.`f` ``
path format this commit introduced, the MERGE and `withSchemaEvolution()` paths
mentioned in the reply, or that a `FILE EXTERNAL` column can still be added;
the `CREATE TABLE LIKE` test has no persistent view copied to a V2 target and
no `IF NOT EXISTS` against an existing table, which now errors, consistent with
CTAS. Suggest adding a nested add (for example `named_struct('a', 1, 'f',
CAST(NULL AS FILE))` into a `STRUCT<a INT>` column), a `FILE EXTERNAL` positive
control, the `pv` to `testcat.t` case and an `IF NOT EXISTS` case; MERGE needs
a row-level test catalog, so it's optional.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3008,6 +3013,48 @@
],
"sqlState" : "42704"
},
+ "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_MAP_KEY" : {
+ "message" : [
+ "The FILE type cannot be a map key in the return type of <routineType>
<routineName>. Use FILE only as a map value, array element, or struct field."
+ ],
+ "sqlState" : "42601"
+ },
+ "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_QUALIFIER" : {
+ "message" : [
+ "The FILE type in <routineType> <routineName> does not support the
EXTERNAL qualifier. Use FILE without a qualifier."
+ ],
+ "sqlState" : "42601"
+ },
+ "FILE_TYPE_MISSING_QUALIFIER" : {
+ "message" : [
+ "FILE type requires the EXTERNAL qualifier in a column definition. Use
FILE EXTERNAL."
+ ],
+ "sqlState" : "42601"
+ },
+ "FILE_TYPE_UNKNOWN_WRITE" : {
+ "message" : [
+ "Column(s) <colNames> have an ambiguous FILE type, so they cannot be
stored in a table. A table column must be FILE EXTERNAL, so the values written
to these columns must be FILE EXTERNAL."
Review Comment:
**Nit: error wording and helper follow-ups.** The reworded message says 'the
values written to these columns', but `CREATE TABLE LIKE` writes no values, and
`colNames` is a nested path only for schema-evolution adds (CTAS and `LIKE`
report the top-level column). Three public `TypeUtils` helpers now throw this
error and differ only by input type, and FILE bucket sort columns still get
`_LEGACY_ERROR_TEMP_3072` on V1 but `PARTITION_COLUMN` on V2. Suggest folding
these into the error-message follow-up, with wording such as `Column(s)
<colNames> contain a FILE without the EXTERNAL qualifier, which a table cannot
store.`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala:
##########
@@ -94,6 +94,8 @@ case class CreateTableLikeCommand(
override def run(sparkSession: SparkSession): Seq[Row] = {
val catalog = sparkSession.sessionState.catalog
val sourceTableDesc =
catalog.getTempViewOrPermanentTableMetadata(sourceTable)
+ // A view can hold a FILE without a qualifier, which a table cannot store.
+ TypeUtils.checkNoUnknownFileTypeInTableSchema(sourceTableDesc.schema)
Review Comment:
**Low: `CREATE TABLE LIKE` is checked at execution rather than analysis.**
This check and the one in `CreateTableLikeExec` run inside `run()`, so `EXPLAIN
CREATE TABLE t LIKE v` doesn't surface the error while the equivalent CTAS
does, and the description's 'enforces both at analysis time' no longer quite
holds. V1 has no choice, since `ResolveSessionCatalog` has already rewritten
the statement into a command that carries only identifiers, but the V2
`CreateTableLike` node reaches `CheckAnalysis` with a resolved source. Suggest
moving the V2 check there, or qualifying the description.
##########
sql/core/src/test/scala/org/apache/spark/sql/FileTypeRestrictionsSuite.scala:
##########
@@ -0,0 +1,501 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.SparkConf
+import org.apache.spark.api.python.{PythonBroadcast, PythonEvalType,
SimplePythonFunction}
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.sql.api.java.UDF1
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
+import org.apache.spark.sql.connector.catalog.InMemoryCatalog
+import org.apache.spark.sql.execution.python.{UserDefinedPythonFunction,
UserDefinedPythonTableFunction}
+import org.apache.spark.sql.expressions.{MutableAggregationBuffer,
UserDefinedAggregateFunction}
+import org.apache.spark.sql.functions.{col, udf}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DataType, FileType, IntegerType, MapType,
StructType}
+
+/**
+ * Tests the rules on where the FILE type can be used: table columns must be
FILE EXTERNAL,
+ * routines declare a plain FILE, and FILE has no defined ordering.
+ */
+class FileTypeRestrictionsSuite extends QueryTest with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf =
+ super.sparkConf.set("spark.sql.catalog.testcat",
classOf[InMemoryCatalog].getName)
+
+ // A FILE without a qualifier, at the top level and nested.
+ private val unqualifiedFileTypes = Seq(
+ "FILE",
+ "ARRAY<FILE>",
+ "MAP<STRING, FILE>",
+ "STRUCT<f1: FILE>",
+ "STRUCT<f1: INT, f2: STRUCT<f3: ARRAY<FILE>>>")
+
+ private val externalFile = "SELECT CAST(NULL AS FILE EXTERNAL) AS f"
+
+ test("a table column definition requires FILE EXTERNAL") {
+ Seq("testcat.t", "t").foreach { table =>
+ val provider = if (table.startsWith("testcat")) "" else "USING parquet"
+ unqualifiedFileTypes.foreach { dataType =>
+ withTable(table) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $table (id INT, c $dataType) $provider")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ }
+ }
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT, c STRUCT<f1: ARRAY<FILE
EXTERNAL>>)")
+ }
+ }
+
+ test("REPLACE TABLE requires FILE EXTERNAL") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT)")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("REPLACE TABLE testcat.t (id INT, c FILE)")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ }
+
+ test("ALTER TABLE ADD COLUMNS and REPLACE COLUMNS require FILE EXTERNAL") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT)")
+ unqualifiedFileTypes.foreach { dataType =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t ADD COLUMN c $dataType")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t REPLACE COLUMNS (id INT, c $dataType)")
+ },
+ condition = "FILE_TYPE_MISSING_QUALIFIER")
+ }
+ sql("ALTER TABLE testcat.t ADD COLUMN c FILE EXTERNAL")
+ }
+ }
+
+ test("CREATE TABLE AS SELECT cannot store a FILE without a qualifier") {
+ Seq("testcat.t", "t").foreach { table =>
+ val provider = if (table.startsWith("testcat")) "" else "USING parquet"
+ withTable(table) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $table $provider AS SELECT 1 AS id, CAST(NULL
AS FILE) AS f")
+ },
+ condition = "FILE_TYPE_UNKNOWN_WRITE",
+ parameters = Map("colNames" -> "[`f`]"))
+ }
+ }
+ }
+
+ test("schema evolution cannot add a FILE without a qualifier") {
+ withTable("testcat.t") {
+ sql("CREATE TABLE testcat.t (id INT)")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("INSERT WITH SCHEMA EVOLUTION INTO testcat.t " +
+ "SELECT 1 AS id, CAST(NULL AS FILE) AS f")
+ },
+ condition = "FILE_TYPE_UNKNOWN_WRITE",
+ parameters = Map("colNames" -> "[`f`]"))
+ // The table is not changed.
+ assert(spark.table("testcat.t").schema === new StructType().add("id",
IntegerType))
+ }
+ }
+
+ test("CREATE TABLE LIKE cannot copy a FILE without a qualifier from a view")
{
+ withTempView("v") {
+ withView("pv") {
+ sql("CREATE TEMPORARY VIEW v AS SELECT 1 AS id, CAST(NULL AS FILE) AS
f")
+ sql("CREATE VIEW pv AS SELECT 1 AS id, CAST(NULL AS FILE) AS f")
+ Seq("v" -> "t", "pv" -> "t", "v" -> "testcat.t").foreach { case
(source, target) =>
+ withTable(target) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"CREATE TABLE $target LIKE $source")
+ },
+ condition = "FILE_TYPE_UNKNOWN_WRITE",
+ parameters = Map("colNames" -> "[`f`]"))
+ assert(!spark.catalog.tableExists(target))
+ }
+ }
+ }
+ }
+ }
+
+ test("a FILE column cannot be a partition or clustering column") {
+ Seq(
+ ("PARTITIONED BY (f)", "PARTITION_COLUMN"),
+ ("PARTITIONED BY (bucket(4, f))", "PARTITION_COLUMN"),
+ ("CLUSTERED BY (f) INTO 4 BUCKETS", "PARTITION_COLUMN"),
+ ("CLUSTER BY (f)", "CLUSTERING_COLUMN")).foreach { case (clause, usage)
=>
+ Seq("testcat.t", "t").foreach { table =>
+ // `bucket` is a V2 transform, and CLUSTERED BY ... INTO ... BUCKETS
is V1 syntax.
+ val isV2 = table.startsWith("testcat")
+ if (isV2 != clause.startsWith("CLUSTERED BY")) {
Review Comment:
**High: this guard skips the V1 `PARTITIONED BY` and `CLUSTER BY` cases.**
`isV2 != clause.startsWith("CLUSTERED BY")` is an exclusive-or, so V1 now runs
only `CLUSTERED BY (f) INTO 4 BUCKETS`: the V1 `CREATE TABLE ... PARTITIONED BY
(f)` and `CLUSTER BY (f)` cases that ran at 40df55f no longer run, and the V1
CTAS variants added in this commit never run, so the partition and
`clusterBySpec` parts of the V1 check in `PreprocessTableCreation` have no test
(skipped cases can't fail CI). The comment is also inaccurate both ways:
`PARTITIONED BY (bucket(4, f))` and `CLUSTERED BY (f) INTO 4 BUCKETS` produce
the same `bucket` transform, which V1 (through `convertTransforms`) and V2 both
reject. Suggest deleting both guards and the comment and running all four
clauses against both catalogs for `CREATE TABLE` and CTAS, ideally with a CTAS
query that also has an `id` column, so V1 `PARTITIONED BY (f)` doesn't rely on
the FILE check running before the all-partition-columns check.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -998,6 +1021,25 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
create.tableSchema.foreach(f =>
TypeUtils.failWithIntervalType(f.dataType))
TypeUtils.failUnsupportedDataType(create.tableSchema, SQLConf.get)
SchemaUtils.checkIndeterminateCollationInSchema(create.tableSchema)
+ create match {
+ // The columns of CREATE TABLE AS SELECT come from its query,
not from DDL.
+ case ctas: V2CreateTableAsSelectPlan =>
+ TypeUtils.checkNoUnknownFileTypeWrite(ctas.query.output)
Review Comment:
Verified, thanks: the check runs before `alterTable` and covers INSERT,
MERGE and the DataFrame writers, including nested adds. The remaining test gaps
for this path (a nested add, MERGE, a `FILE EXTERNAL` positive control) are in
a new thread on the test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
