dejankrak-db commented on code in PR #59640:
URL: https://github.com/apache/spark/pull/59640#discussion_r4429657316
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -1014,6 +1037,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)
+ case _ =>
+ TypeUtils.checkFileTypeHasQualifier(create.tableSchema)
+ }
+ // FILE has no defined ordering, so it cannot be a partition or
clustering column.
+ create.partitioning.foreach { transform =>
Review Comment:
Thanks, fixed in 3ac14b5573d, after merging master in df6ded41203. After the
partition loop, the columns referenced by `create.writeOrdering` are checked
with `checkNoFileTypeInColumnUsage` under a new
`FILE_TYPE_UNSUPPORTED_COLUMN_USAGE.WRITE_ORDERING_COLUMN` subclass. The new
test runs at analysis on `testcat` and covers `ORDERED BY f`, `LOCALLY ORDERED
BY bucket(4, f)`, a struct key that holds a FILE, REPLACE TABLE, CTAS and RTAS,
plus a positive control that orders by `id`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -1044,6 +1086,20 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
}
}
+ // A view can hold a FILE without a qualifier, which a table cannot
store. A V1 target
+ // is checked by `CreateTableLikeCommand`, as the V1 command carries
no resolved source.
Review Comment:
Reworded in 3ac14b5573d: "When the target and source are both in the session
catalog, `ResolveSessionCatalog` has already rewritten this into
`CreateTableLikeCommand`, which checks the source when it runs." The
description now uses the same condition. I left the V1 check in
`CreateTableLikeCommand` to keep this change small.
##########
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")
Review Comment:
Added in 3ac14b5573d: an analysis-only assertion for `CREATE TABLE testcat.t
LIKE v`, a `testcat.src` table holding a bare FILE, created through the catalog
API and copied to both `testcat.t` and `t` (also at analysis, which covers the
`ResolvedTable` branch and a V2 source with a session-catalog target), and a
positive control that copies a view holding `FILE EXTERNAL`.
##########
sql/api/src/main/scala/org/apache/spark/sql/types/FileType.scala:
##########
@@ -25,30 +25,30 @@ import org.apache.spark.annotation.Unstable
/**
* The data type representing a reference to a file, with metadata describing
the file.
*
- * In SQL the type is named `FILE`, and the `EXTERNAL` keyword is planned to
be required when
- * declaring a table column, as in `CREATE TABLE t (f FILE EXTERNAL)`; until
then a bare `FILE` is
- * accepted wherever a type is. Its JSON representation is `"file external"`
for
- * `FileType.external` and `"file unknown"` for `FileType()`, except that an
external file in a
- * struct field is the plain `"file"`. A bare `"file"` is read as external
inside a struct field
- * and as unknown elsewhere.
+ * In SQL the type is named `FILE`. A table column must be declared with the
`EXTERNAL` keyword,
+ * as in `CREATE TABLE t (f FILE EXTERNAL)`, while the signature of a function
declares a bare
+ * `FILE`. Its JSON representation is `"file external"` for
`FileType.external` and
+ * `"file unknown"` for `FileType()`, except that an external file in a struct
field is the plain
+ * `"file"`. A bare `"file"` is read as external inside a struct field and as
unknown elsewhere.
*
* FileType is planned to be physically stored as a struct with the following
fields (see
* `FileType.FIELDS`), but it deliberately does not extend [[StructType]]:
- * - uri (STRING): the opaque URI to the file
+ * - uri (STRING): the URI of the referenced file, which may be absolute or
relative
* - offset (BIGINT): the start of the referenced byte range
- * - size (BIGINT): the size of the file in bytes
- * - content_type (STRING): the MIME/content type of the file
- * - checksum (STRING): an algorithm-tagged integrity token for the file,
e.g. "ETAG:{etag}"
+ * - size (BIGINT): the length of the referenced byte range
+ * - content_type (STRING): the MIME/content type of the referenced bytes
+ * - checksum (STRING): an algorithm-tagged integrity token for the
referenced bytes, e.g.
+ * "MD5:{digest}", or "ETAG:{etag}" for the whole file
* - inline (BINARY): the referenced bytes stored inline in the value
*
* FileType may be EXTERNAL (simple references, lifecycle independent of the
table) or UNKNOWN
* (mode not yet determined, such as the result of a cast or of a UDF). Mode
is part of type
* identity - files with different modes are distinct under `equals`.
*
- * FileType is planned to be not comparable -- it cannot be used in join
expressions, grouping
- * expressions, or as a partition/clustering column. Users can extract the uri
field for such
- * operations. This type currently only guarantees the schema-level pieces
(DDL and JSON); the
- * physical representation and most of the restrictions are added by later
parts.
+ * FileType is not comparable -- it cannot be the key of a map built in a
query or of a map in the
+ * return type of a function, and it cannot be used in join expressions,
grouping expressions, set
+ * operations, or as a partition/clustering column. Users can extract the uri
field for such
Review Comment:
Rewrote it in 3ac14b5573d to state the rule: an operation that compares,
sorts, groups or hashes a FILE fails at analysis, with examples, followed by
the map-key rule and the partition, bucket, clustering and write-ordering
columns.
--
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]
