uros-b commented on code in PR #59640:
URL: https://github.com/apache/spark/pull/59640#discussion_r4346412893
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionCommand.scala:
##########
@@ -76,6 +77,15 @@ case class CreateSQLFunctionCommand(
val inputParam =
inputParamText.map(UserDefinedFunction.parseRoutineParam(_, parser, collation))
val returnType = parseReturnTypeText(returnTypeText, isTableFunc, parser,
collation)
+ // A FILE in the signature of a function is a plain FILE, without a
qualifier, and the return
+ // type cannot hold a FILE in a map key.
+ val routineName = QueryCompilationErrors.toSQLId(name.nameParts)
+ TypeUtils.checkNoQualifierInRoutineSignature(
+ inputParam.toSeq ++ returnType.toSeq.map(_.fold(identity, identity)),
+ routineType = "function",
+ routineName = routineName)
+ returnType.map(_.fold(identity, identity)).foreach(
Review Comment:
Without a RETURNS clause, or with a bare RETURNS TABLE, the inferred type is
stored unchecked. So RETURN CAST(NULL AS FILE EXTERNAL) or a MAP<FILE, INT>
body still creates the function. The query-time check doesn't catch it either,
because SQL function expressions aren't UserDefinedExpressions.
##########
sql/api/src/main/scala/org/apache/spark/sql/types/FileType.scala:
##########
@@ -45,10 +45,9 @@ import org.apache.spark.annotation.Unstable
* (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 used in a map key, join
expressions, grouping
Review Comment:
This says "cannot be used in a map key", but table DDL still accepts one.
CREATE TABLE t (m MAP<FILE EXTERNAL, INT>) passes, and so does the CTAS form.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/TypeUtils.scala:
##########
@@ -162,4 +165,104 @@ 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; a FILE of any mode can still be passed to them.
Review Comment:
"A FILE of any mode can still be passed to them" isn't true for SQL
functions. There is no FILE EXTERNAL → FILE cast, so fn(x FILE) rejects a
table's FILE column. The FILE_TYPE_UNKNOWN_WRITE advice is unusable for the
same reason.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/hash.scala:
##########
@@ -392,6 +396,12 @@ abstract class HashExpression[E] extends Expression {
DataTypeMismatch(
errorSubClass = "HASH_VARIANT_TYPE",
messageParameters = Map("functionName" -> toSQLId(prettyName)))
+ } else if (children.exists(child => hasFileType(child.dataType))) {
+ // FILE is not comparable, so hashing it, for example by a hash
aggregate, hash join or hash
+ // partitioning the planner inserts, is rejected rather than hashing the
underlying struct.
Review Comment:
The claim that hashes "the planner inserts" are rejected is wrong. SELECT
/*+ REBALANCE(f) */ ... gets through analysis, because the check matches only
RepartitionByExpression. DeduplicateWithinWatermark keys aren't covered either.
Fix: match HasPartitionExpressions.
--
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]
