dtenedor commented on code in PR #59640:
URL: https://github.com/apache/spark/pull/59640#discussion_r4353281730
##########
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:
**Medium: schema evolution can add a bare FILE column to a table.**
`INSERT`/`MERGE WITH SCHEMA EVOLUTION` and `withSchemaEvolution()` writes add
each new source column via `TableChange.addColumn(path, f.dataType.asNullable)`
(`ResolveSchemaEvolution.scala` L250 and L279, MERGE in `v2Commands.scala`
L1288) and apply it with `catalog.alterTable` (L74) during analysis, so this
CTAS check never sees it: `FileType.asNullable` returns `this`, the column is
created as a bare FILE, and the write then type-checks (the type-mismatch
argument for INSERT assumes the target column already exists). The suite's
`testcat` advertises automatic schema evolution by default, so on a table
`testcat.t (id INT)`, `INSERT WITH SCHEMA EVOLUTION INTO testcat.t SELECT 1 AS
id, CAST(NULL AS FILE) AS f` should leave `t` with a bare `f FILE` column.
Suggest rejecting bare-FILE `AddColumn` changes in `ResolveSchemaEvolution`
before `evolveSchema`, which covers INSERT, MERGE and the DataFrame writers in
one pla
ce (a `CheckAnalysis` check, including the INSERT handling planned for the
casts part, would run after the catalog has already changed), and testing both
the error and that the table schema is unchanged.
##########
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)
+ case _ =>
+ TypeUtils.checkFileTypeHasQualifier(create.tableSchema)
Review Comment:
**Medium: `CREATE TABLE ... LIKE` can copy a bare FILE column into a new
table.** `CreateTableLike` is a `BinaryCommand`, not a `V2CreateTablePlan`, so
this check never sees it: on the session catalog it becomes
`CreateTableLikeCommand`, which copies the source schema straight into
`catalog.createTable` (`tables.scala` L126-140), and the V2
`CreateTableLikeExec` copies the source columns the same way. The source can be
a view, and views can hold a bare FILE, e.g. `CREATE TEMP VIEW v AS SELECT
CAST(NULL AS FILE) AS f; CREATE TABLE t LIKE v`. Suggest checking the source
schema in `ResolveSessionCatalog` (next to the new V1 `CLUSTER BY` check) or in
`CreateTableLikeCommand.run` for V1, because by the time `CheckAnalysis` runs
V1 has been rewritten into a command that only carries identifiers, and in
`CheckAnalysis` or `CreateTableLikeExec` for V2, with tests for temp and
persistent view sources.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala:
##########
@@ -293,6 +295,8 @@ case class PreprocessTableCreation(catalog: SessionCatalog)
extends Rule[Logical
c.copy(tableDesc = normalizedTable, query = Some(reorderedQuery))
} else {
+ checkFileTypeHasQualifier(tableDesc.schema)
Review Comment:
**Low: Declarative Pipelines create and evolve tables without these
checks.** `DatasetManager` (`sql/pipelines/.../graph/DatasetManager.scala`
L409-425 and L631-645) calls `catalog.createTable` and `alterTable` directly
with the flow's declared or inferred schema and its partition and cluster
transforms, so a bare FILE column, or a FILE partition or clustering column,
gets through. The gap predates this PR, since pipelines skip every
`V2CreateTablePlan` check, so a follow-up JIRA seems reasonable. The check
should run before `materializeTable` truncates an existing table; one shared
helper for the schema a table will store, called from here, `CheckAnalysis` and
the other schema producers, would keep these paths from drifting.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -1236,6 +1295,25 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
}
}
+ /**
+ * Checks that user-defined routines that are not created by SQL DDL, such
as Python UDFs and
+ * registered Scala UDFs, declare only a plain FILE in their return type, as
+ * [[TypeUtils.checkNoQualifierInRegisteredRoutineReturnType]] does when
they are registered.
+ * A non-scalar Python UDF is skipped, as its data type is the schema of the
rows it returns.
+ */
+ private def checkFileTypeInUserDefinedRoutines(plan: LogicalPlan): Unit = {
+ if (SQLConf.get.isFileTypeEnabled) {
+ plan.foreach { operator =>
+ operator.expressions.foreach(_.foreach {
+ case udf: PythonUDF if !PythonUDF.isScalarPythonUDF(udf) =>
Review Comment:
**Low: row-returning Python functions follow two different rules.** This
skip lets `mapInPandas`, `applyInPandas`, `mapInArrow` and the cogrouped and
stateful variants declare `FILE EXTERNAL` output or FILE map keys, while Python
UDTFs, whose data type is also a row schema, are rejected at registration and
here, as is SQL `RETURNS TABLE`. The likely reason for the skip, passing table
rows through with `df.schema`, applies equally to a UDTF that returns its
`TABLE` argument. No user can reach this until PySpark and Spark Connect get a
FILE type, but suggest picking one rule for row-returning functions,
documenting it here, and adding a test that pins it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/OperatorWithUncomparableTypeValidator.scala:
##########
@@ -46,6 +47,10 @@ object OperatorWithUncomparableTypeValidator {
if (hasVariantType(element.dataType)) {
throwUnsupportedSetOperationOnVariantType(element,
unsupportedOperator)
}
+
+ if (hasFileType(element.dataType)) {
Review Comment:
**Low: the two analyzers can report different errors for a set operation
with several uncomparable columns.** This loop checks each column for MAP,
VARIANT and FILE in turn, while `CheckAnalysis` looks for MAP across all
columns first, then VARIANT, then FILE. So `SELECT f, m ... INTERSECT ...`,
with a FILE column before a MAP column, reports `SET_OPERATION_ON_FILE_TYPE`
here and `SET_OPERATION_ON_MAP_TYPE` in the fixed-point analyzer; MAP versus
VARIANT already behaved this way, and dual-run hides it because it rethrows the
fixed-point error when both analyzers fail. Suggest iterating type by type over
all columns here, in the same order as `CheckAnalysis`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -1236,6 +1295,25 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
}
}
+ /**
+ * Checks that user-defined routines that are not created by SQL DDL, such
as Python UDFs and
+ * registered Scala UDFs, declare only a plain FILE in their return type, as
+ * [[TypeUtils.checkNoQualifierInRegisteredRoutineReturnType]] does when
they are registered.
+ * A non-scalar Python UDF is skipped, as its data type is the schema of the
rows it returns.
+ */
+ private def checkFileTypeInUserDefinedRoutines(plan: LogicalPlan): Unit = {
+ if (SQLConf.get.isFileTypeEnabled) {
+ plan.foreach { operator =>
+ operator.expressions.foreach(_.foreach {
+ case udf: PythonUDF if !PythonUDF.isScalarPythonUDF(udf) =>
+ case routine: UserDefinedExpression =>
Review Comment:
**Low: whether this check covers a V2 catalog function depends on how the
function is implemented.** A `ScalarFunction` with only `produceResult`
resolves to `ApplyFunctionExpression`, which is a `UserDefinedExpression` and
is checked, but the same function with an `invoke` method resolves to
`StaticInvoke`/`Invoke`, and an `AggregateFunction` resolves to `V2Aggregator`,
and neither is checked. The scaladoc's 'not created by SQL DDL' is also
inaccurate: a UDAF class registered with `CREATE FUNCTION ... AS '<class>'`
becomes a `ScalaUDAF` that only this check covers, and the existing comment at
L1060-1062 assumes all V2 functions are `UserDefinedExpression`s. Suggest
deciding whether catalog functions are in scope (if so, check `resultType()`
once where V2 functions are bound in `FunctionResolution.resolveV2Function`; if
not, exclude `ApplyFunctionExpression`), then fixing the scaladoc and adding a
test.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/TypeUtils.scala:
##########
@@ -162,4 +165,105 @@ object TypeUtils extends QueryErrorsBase {
messageParameters = scala.collection.immutable.Map.empty)
}
}
+
+ private def containsUnknownFileType(dataType: DataType): Boolean = {
+ dataType.existsRecursively {
+ case ft: FileType => ft.mode == FileTypeMode.UNKNOWN
+ case _ => false
+ }
+ }
+
+ /**
+ * Fails if a FILE in `dataType`, at any nesting depth, has no qualifier, as
a FILE in a table
+ * column definition must be declared `FILE EXTERNAL`.
+ */
+ def checkFileTypeHasQualifier(dataType: DataType): Unit = {
+ if (containsUnknownFileType(dataType)) {
+ throw QueryCompilationErrors.fileTypeMissingQualifierError()
+ }
+ }
+
+ /**
+ * Fails if any of `columns`, which a write such as CREATE TABLE AS SELECT
stores in a table,
+ * holds a FILE without a qualifier, as a table cannot store one.
+ */
+ def checkNoUnknownFileTypeWrite(columns: Seq[Attribute]): Unit = {
+ val unknownCols = columns.filter(c => containsUnknownFileType(c.dataType))
+ if (unknownCols.nonEmpty) {
+ throw
QueryCompilationErrors.fileTypeUnknownWriteError(unknownCols.map(_.name))
+ }
+ }
+
+ /**
+ * Fails if `field` holds a FILE, as FILE has no defined ordering and so
cannot be a partition
+ * column (`usage` is `PARTITION_COLUMN`) or a clustering column
(`CLUSTERING_COLUMN`).
+ */
+ def checkNoFileTypeInColumnUsage(field: StructField, usage: String): Unit = {
+ if (field.dataType.existsRecursively(_.isInstanceOf[FileType])) {
+ throw QueryCompilationErrors.fileTypeUnsupportedColumnUsageError(field,
usage)
+ }
+ }
+
+ /** Fails if a clustering column of `clusterBySpec` holds a FILE. */
+ def checkNoFileTypeClusteringColumn(
+ schema: StructType,
+ clusterBySpec: ClusterBySpec,
+ resolver: Resolver): Unit = {
+ clusterBySpec.columnNames.foreach { ref =>
+ schema.findNestedField(ref.fieldNames.toImmutableArraySeq, resolver =
resolver)
+ .foreach { case (_, field) => checkNoFileTypeInColumnUsage(field,
"CLUSTERING_COLUMN") }
+ }
+ }
+
+ /**
+ * Fails if a FILE in the parameter or return types of a routine has a
qualifier. Routines
+ * declare a plain FILE. `routineName` is quoted already.
+ */
+ def checkNoQualifierInRoutineSignature(
+ dataTypes: Seq[DataType],
+ routineType: String,
+ routineName: String): Unit = {
+ val hasQualified = dataTypes.exists(_.existsRecursively {
+ case ft: FileType => ft.mode != FileTypeMode.UNKNOWN
+ case _ => false
+ })
+ if (hasQualified) {
+ throw
QueryCompilationErrors.fileTypeInRoutineSignatureUnsupportedQualifierError(
+ routineType, routineName)
+ }
+ }
+
+ /**
+ * Fails if the return type of a function holds a FILE in a map key. Map
construction already
+ * rejects such a key through [[checkForMapKeyType]].
+ */
+ def checkNoFileTypeMapKeyInReturnType(returnType: DataType, routineName:
String): Unit = {
+ def fileTypeInMapKey(dt: DataType): Boolean = dt match {
+ case mt: MapType =>
+ mt.keyType.existsRecursively(_.isInstanceOf[FileType]) ||
fileTypeInMapKey(mt.valueType)
+ case at: ArrayType => fileTypeInMapKey(at.elementType)
+ case st: StructType => st.fields.exists(f =>
fileTypeInMapKey(f.dataType))
+ case _ => false
+ }
+ if (fileTypeInMapKey(returnType)) {
+ throw
QueryCompilationErrors.fileTypeInRoutineSignatureUnsupportedMapKeyError(
+ "function", routineName)
+ }
+ }
+
+ /**
+ * Checks the return type of a function that is not created by SQL DDL, for
example a UDF
+ * registered through `spark.udf.register`, as
[[checkNoQualifierInRoutineSignature]] and
+ * [[checkNoFileTypeMapKeyInReturnType]] do for SQL functions. `routineName`
is the unquoted
+ * name of the function. Does nothing when the FILE type is disabled.
+ */
+ def checkNoQualifierInRegisteredRoutineReturnType(
Review Comment:
**Low: a few comments and names don't match the code.** This helper also
checks map keys and also runs for UDFs that were never registered
(`CheckAnalysis` L1310), so a name like `checkFileTypeInRoutineReturnType`
would fit better; `Resolver.scala` L552 (outside the diff) still says only MAP
and VARIANT are unsupported under `Distinct`; the `ExprUtils` comment says the
grouping check is about orderability, but it only covers VARIANT and FILE; and
the `hash.scala` comment refers to the underlying struct, which doesn't exist
yet. Suggest renaming the helper and updating those comments.
##########
sql/core/src/test/scala/org/apache/spark/sql/FileTypeRestrictionsSuite.scala:
##########
@@ -0,0 +1,357 @@
+/*
+ * 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.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("a FILE column cannot be a partition or clustering column") {
+ Seq(
+ ("PARTITIONED BY (f)", "PARTITION_COLUMN"),
+ ("PARTITIONED BY (bucket(4, f))", "PARTITION_COLUMN"),
+ ("CLUSTER BY (f)", "CLUSTERING_COLUMN")).foreach { case (clause, usage)
=>
+ Seq("testcat.t", "t").foreach { table =>
+ // Bucketing is only a V2 transform.
+ if (table.startsWith("testcat") || !clause.contains("bucket")) {
+ val provider = if (table.startsWith("testcat")) "" 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\""))
+ }
+ }
+ }
+ }
+ 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)")
+ Seq("s", "f").foreach { column =>
+ val e = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t ALTER COLUMN $column TYPE FILE")
+ }
+ assert(e.getCondition === "NOT_SUPPORTED_CHANGE_COLUMN")
+ }
+ }
+ }
+
+ 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))
+ 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))
+ }
+
+ test("a FILE cannot be in INTERSECT, EXCEPT or DISTINCT") {
Review Comment:
**Low: several new branches, and cases that should keep working, are
untested.** Nothing exercises `dropDuplicates` with a FILE key (the
`Deduplicate` arm in `CheckAnalysis`), V1 CTAS with `PARTITIONED BY` or
`CLUSTER BY` (`rules.scala` L275), FILE map keys nested in struct or array
return types such as `RETURNS TABLE (m MAP<FILE, INT>)`, the query-time check
for the non-scalar Python UDF exemption, `PythonUDTF`, `register(name, udaf:
UserDefinedAggregateFunction)` or a typed `Aggregator`, or a FILE nested in a
struct or array for `GROUP BY`, `hash`, map keys and `DISTINCT` (every case
uses a top-level `f`). There are also no positive controls showing that `UNION
ALL`, `dropDuplicates` and `DISTRIBUTE BY` on another column, and a CTAS
producing `FILE EXTERNAL` still work when a FILE column is present. Suggest a
case for each, plus tests for the schema-evolution, `CREATE TABLE LIKE` and
VOID `ALTER COLUMN` paths in the other threads.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -5173,6 +5184,37 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map.empty)
}
+ def fileTypeUnknownWriteError(colNames: Seq[String]): Throwable = {
+ new AnalysisException(
+ errorClass = "FILE_TYPE_UNKNOWN_WRITE",
+ messageParameters = Map(
+ "colNames" -> colNames.map(quoteIdentifier).mkString("[", ", ", "]")))
+ }
+
+ def fileTypeUnsupportedColumnUsageError(field: StructField, usage: String):
Throwable = {
+ new AnalysisException(
+ errorClass = s"FILE_TYPE_UNSUPPORTED_COLUMN_USAGE.$usage",
Review Comment:
**Nit: a few simplifications.** `usage: String` is spliced into the error
class name here, so a typo at a call site only shows up at runtime as a missing
error class; `checkNoFileTypeMapKeyInReturnType` reimplements a walk that
`existsRecursively` already provides; `routineType` is always `function`; the
second `checkFileTypeInSignature` call in `CreateSQLFunctionCommand` repeats
the first when `RETURNS` is explicit; and `checkFileTypeInUserDefinedRoutines`
adds a separate full pass over the plan that runs before the final
unresolved-operator check. Suggest two builders or a small enum instead of
`usage`, and folding the routine check into the main loop.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2AlterTableCommands.scala:
##########
@@ -146,6 +147,7 @@ case class ReplaceColumns(
columnsToAdd.foreach { c =>
TypeUtils.failWithIntervalType(c.dataType)
TypeUtils.failUnsupportedDataType(c.dataType, conf)
+ TypeUtils.checkFileTypeHasQualifier(c.dataType)
Review Comment:
**Low: `ALTER COLUMN ... TYPE FILE` on a VOID column still produces a bare
FILE column.** `AlterColumns` (L273) has no equivalent of this check, and
`CheckAnalysis.canAlterColumnType` falls back to `UpCastRule`, which allows
`(NullType, _) => true`. V2 tables can have VOID columns, so `CREATE TABLE
testcat.t (id INT, v VOID)` followed by `ALTER TABLE testcat.t ALTER COLUMN v
TYPE FILE` passes (V1 is safe because `AlterTableChangeColumnCommand` requires
equal types). Suggest adding
`specs.foreach(_.newDataType.foreach(TypeUtils.checkFileTypeHasQualifier))` to
`AlterColumns`, which also keeps a future FILE EXTERNAL to FILE up-cast from
silently loosening `ALTER COLUMN`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RepartitionByExpressionResolver.scala:
##########
@@ -97,6 +97,7 @@ class RepartitionByExpressionResolver(resolver: Resolver,
expressionResolver: Ex
)
validateRepartitionByVariant(resolvedRepartitionByExpression)
+ validateRepartitionByFile(resolvedRepartitionByExpression)
Review Comment:
**Low: this check is unreachable in apache/spark today.** Nothing constructs
`RepartitionByExpressionResolver`: `Resolver.resolve` only dispatches
`Repartition`, and `ResolverGuard` has no `RepartitionByExpression` arm, so
these plans always go to the fixed-point analyzer (the existing VARIANT check
in this file is in the same state). Keeping it for parity is fine, but suggest
saying in the PR description that the single-pass side of the repartition rule
isn't wired up yet, rather than listing it as an enforcement point.
##########
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."
Review Comment:
**Low: error message gaps.** This error has no column name or query context,
so on a wide table the user can't tell which column is wrong, unlike
`FILE_TYPE_UNKNOWN_WRITE` and `FILE_TYPE_UNSUPPORTED_COLUMN_USAGE`, and the
latter names nested columns by their last part only (`CLUSTER BY (s.f)` reports
`f`). Also, `SET_OPERATION_ON_FILE_TYPE` keeps the old 'in a DataFrame that
calls set operations' wording that the MAP and VARIANT versions dropped,
`HASH_FILE_TYPE` is worded differently from `HASH_VARIANT_TYPE`, the routine
errors use SQLSTATE 42601 (syntax error) although they are also raised from
`spark.udf.register`, and `fileTypeMissingQualifierError` lives in `sql/api`
`CompilationErrors` while the other FILE error builders are in
`QueryCompilationErrors`. Suggest adding the full column path to this error,
aligning wording and SQLSTATEs with the sibling errors, and keeping parent
paths in column names.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -160,13 +165,31 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
case _ => None
}
+ protected def fileColumnInSetOperation(plan: LogicalPlan): Option[Attribute]
= plan match {
+ case _: Intersect | _: Except | _: Distinct =>
+ plan.output.find(a => hasFileType(a.dataType))
+ case d: Deduplicate =>
+ d.keys.find(a => hasFileType(a.dataType))
+ case d: DeduplicateWithinWatermark =>
+ d.keys.find(a => hasFileType(a.dataType))
+ case _ => None
+ }
+
protected def variantExprInPartitionExpression(plan: LogicalPlan):
Option[Expression] =
plan match {
case r: RepartitionByExpression =>
r.partitionExpressions.find(e => hasVariantType(e.dataType))
case _ => None
}
+ // Covers both REPARTITION / DISTRIBUTE BY and the REBALANCE hint.
+ protected def fileExprInPartitionExpression(plan: LogicalPlan):
Option[Expression] =
+ plan match {
+ case r: HasPartitionExpressions =>
Review Comment:
**Low: the checks for types without an ordering are per-type copies that
have started to drift.** FILE now covers `REBALANCE` (through
`HasPartitionExpressions` here) and `DeduplicateWithinWatermark`, while the
VARIANT versions still don't; `hasFileType` is defined four times (here,
`OperatorWithUncomparableTypeValidator`, `RepartitionByExpressionResolver` and
`hash.scala`) plus inline copies; and some consumers still accept FILE:
`approx_count_distinct(f)` passes analysis while `xxhash64(f)` is rejected,
because it hashes through `XxHash64Function` directly, and so do `mode(f)` and
`collect_union` over `ARRAY<FILE>`. Suggest one shared helper for VARIANT and
FILE (MAP's rules legitimately differ), adding these aggregates to the
description's list of deferred ones, and a follow-up to extend the VARIANT
checks to `REBALANCE` and `DeduplicateWithinWatermark`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala:
##########
@@ -354,6 +358,17 @@ case class PreprocessTableCreation(catalog:
SessionCatalog) extends Rule[Logical
case _ => cls
}
+ // FILE has no defined ordering, so it cannot be a partition or clustering
column.
+ private def checkNoFileTypePartitionOrClusteringColumn(
+ schema: StructType,
+ table: CatalogTable): Unit = {
+ table.partitionColumnNames.foreach { name =>
+ schema.find(f => conf.resolver(f.name, name))
+ .foreach(checkNoFileTypeInColumnUsage(_, "PARTITION_COLUMN"))
+ }
+ table.clusterBySpec.foreach(checkNoFileTypeClusteringColumn(schema, _,
conf.resolver))
Review Comment:
**Low: V1 bucket columns aren't checked.** This helper covers partition and
clustering columns but not `table.bucketSpec`, and `BucketingUtils.canBucketOn`
only looks at collation, so V1 `CLUSTERED BY (f) INTO 4 BUCKETS` passes
analysis while V2 rejects `bucket(4, f)`, and FILE sort columns hit
`_LEGACY_ERROR_TEMP_3072` instead. It's latent because no V1 source can store
FILE, but the suite's comment that bucketing is only a V2 transform isn't
accurate. Suggest including the bucket and sort columns of `bucketSpec` here,
possibly with a `BUCKET_COLUMN` subclass, and adding the V1 case to the
partition test.
##########
sql/core/src/test/scala/org/apache/spark/sql/FileTypeRestrictionsSuite.scala:
##########
@@ -0,0 +1,357 @@
+/*
+ * 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.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("a FILE column cannot be a partition or clustering column") {
+ Seq(
+ ("PARTITIONED BY (f)", "PARTITION_COLUMN"),
+ ("PARTITIONED BY (bucket(4, f))", "PARTITION_COLUMN"),
+ ("CLUSTER BY (f)", "CLUSTERING_COLUMN")).foreach { case (clause, usage)
=>
+ Seq("testcat.t", "t").foreach { table =>
+ // Bucketing is only a V2 transform.
+ if (table.startsWith("testcat") || !clause.contains("bucket")) {
+ val provider = if (table.startsWith("testcat")) "" 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\""))
+ }
+ }
+ }
+ }
+ 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)")
+ Seq("s", "f").foreach { column =>
+ val e = intercept[AnalysisException] {
+ sql(s"ALTER TABLE testcat.t ALTER COLUMN $column TYPE FILE")
+ }
+ assert(e.getCondition === "NOT_SUPPORTED_CHANGE_COLUMN")
+ }
+ }
+ }
+
+ 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))
+ 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))
+ }
+
+ 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))
+ }
+ }
+
+ 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))
+ }
+
+ 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))
+ }
+
+ test("a FILE cannot be sorted or be a window partition") {
+ Seq(
+ s"SELECT f FROM ($externalFile) ORDER BY f" ->
"DATATYPE_MISMATCH.INVALID_ORDERING_TYPE",
+ s"SELECT row_number() OVER (ORDER BY f) FROM ($externalFile)" ->
+ "DATATYPE_MISMATCH.INVALID_ORDERING_TYPE",
+ s"SELECT count(*) OVER (PARTITION BY f) FROM ($externalFile)" ->
+ "EXPRESSION_TYPE_IS_NOT_ORDERABLE").foreach { case (query, condition)
=>
+ assert(intercept[AnalysisException](sql(query)).getCondition ===
condition, query)
+ }
+ }
+
+ 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>)").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
+ }
+ 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`"))
+ spark.udf.register("file_udf", fileUdf, FileType())
Review Comment:
**Nit: test hygiene.** `file_udf` here, and `py_udf` and `py_udtf` in the
next test, stay registered in the shared session because they aren't wrapped in
`withUserDefinedFunction` the way `fn` is, and the `ALTER COLUMN` and sorting
tests (L163 and L245) assert only `getCondition`. Suggest wrapping the
registrations in `withUserDefinedFunction` and using `checkError` with
parameters.
--
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]
