dejankrak-db commented on code in PR #59640:
URL: https://github.com/apache/spark/pull/59640#discussion_r4348277689
##########
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:
Fixed in 40df55fce0d: the return type inferred from the body is now checked
as well, so `RETURN CAST(NULL AS FILE EXTERNAL)`, `RETURN CAST(NULL AS
MAP<FILE, INT>)` and a bare `RETURNS TABLE` whose query returns `FILE EXTERNAL`
are rejected at CREATE FUNCTION. Added these cases to
`FileTypeRestrictionsSuite`.
##########
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:
Reworded in 40df55fce0d: the restriction is on maps that a query builds
(`map(f, 1)` fails with `INVALID_MAP_KEY_TYPE`) and on the return type of a
function, so the doc now says that. A `MAP<FILE EXTERNAL, INT>` table column is
still accepted; I kept that as it is, since rejecting map keys in table DDL
would need a new error class and is better decided together with how DBR treats
them. Happy to add it as a follow-up if you think it should be blocked.
##########
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:
Good catch. In 40df55fce0d the partitioning check now matches
`HasPartitionExpressions`, so the `REBALANCE` hint is rejected with
`UNSUPPORTED_FEATURE.PARTITION_BY_FILE`, and `DeduplicateWithinWatermark` keys
are rejected like `Deduplicate` keys. Also corrected the comment here: the
planner-inserted hashes do not go through `checkInputDataTypes`, so it now only
claims the hash functions. Tests added for both.
##########
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:
Right, there is no cast between FILE modes yet; that comes with the casts
part of the SPIP. I removed the claim from the comment in 40df55fce0d. For
`FILE_TYPE_UNKNOWN_WRITE`, see the reply on error-conditions.json.
##########
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",
+ messageParameters = Map(
+ "colName" -> toSQLId(field.name),
+ "dataType" -> toSQLType(field.dataType)))
+ }
+
+ def fileTypeInRoutineSignatureUnsupportedQualifierError(
+ routineType: String,
+ routineName: String): Throwable = {
+ new AnalysisException(
+ errorClass = "FILE_TYPE_IN_ROUTINE_SIGNATURE_UNSUPPORTED_QUALIFIER",
+ messageParameters = Map("routineType" -> routineType, "routineName" ->
routineName))
+ }
+
Review Comment:
Done in 40df55fce0d: the registered-routine path (`spark.udf.register`,
`registerPython`, `spark.udtf.registerPython` and the analyzer check for inline
UDFs) now quotes the name with `toSQLId`, so all callers report a quoted id.
The tests now expect `` `file_udf` ``, `` `py_udf` `` and so on.
--
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]
