dejankrak-db commented on code in PR #59640: URL: https://github.com/apache/spark/pull/59640#discussion_r4429657933
########## sql/core/src/test/scala/org/apache/spark/sql/FileTypeRestrictionsSuite.scala: ########## @@ -0,0 +1,550 @@ +/* + * 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, s STRUCT<a: INT>)") + val schema = spark.table("testcat.t").schema + Seq( + "SELECT 1 AS id, named_struct('a', 1) AS s, CAST(NULL AS FILE) AS f" -> "[`f`]", + "SELECT 1 AS id, named_struct('a', 1, 'f', CAST(NULL AS FILE)) AS s" -> "[`s`.`f`]" + ).foreach { case (query, colNames) => + checkError( + exception = intercept[AnalysisException] { + sql(s"INSERT WITH SCHEMA EVOLUTION INTO testcat.t $query") + }, + condition = "FILE_TYPE_UNKNOWN_WRITE", + parameters = Map("colNames" -> colNames)) + // The table is not changed. + assert(spark.table("testcat.t").schema === schema) + } + // A FILE EXTERNAL column can be added. + sql("INSERT WITH SCHEMA EVOLUTION INTO testcat.t " + + "SELECT 1 AS id, named_struct('a', 1) AS s, CAST(NULL AS FILE EXTERNAL) AS f") + assert(spark.table("testcat.t").schema === schema.add("f", FileType.external)) + } + } + + 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") + for (source <- Seq("v", "pv"); target <- Seq("t", "testcat.t")) { + 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)) + // As for CREATE TABLE AS SELECT, IF NOT EXISTS does not skip the check when the + // table exists. Review Comment: Fixed all three in 3ac14b5573d: "As with", the test is now named "ALTER COLUMN requires FILE EXTERNAL and cannot change the type of a FILE column", and the Python comment now covers only what that test does (registering the functions and analyzing a call to the UDF). Also added the CTAS `IF NOT EXISTS` case for V1 and V2. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala: ########## @@ -29,7 +29,8 @@ import org.apache.spark.sql.catalyst.optimizer.InlineCTE import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.trees.TreePattern.{LATERAL_COLUMN_ALIAS_REFERENCE, PLAN_EXPRESSION, UNRESOLVED_WINDOW_EXPRESSION} import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, StringUtils, TypeUtils} -import org.apache.spark.sql.connector.catalog.{CatalogManager, LookupCatalog, SupportsPartitionManagement} +import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, LookupCatalog, SupportsPartitionManagement} Review Comment: Done in accfd986550, before the merge: the arm uses `table.columns.asSchema` and the import is gone, so only the import line in `ResolveSessionCatalog.scala` conflicted, which df6ded41203 resolves. -- 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]
