dejankrak-db commented on code in PR #59640:
URL: https://github.com/apache/spark/pull/59640#discussion_r4359287951
##########
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:
Agreed this predates the PR, since pipelines skip all the
`V2CreateTablePlan` checks. I will file a follow-up JIRA for running the table
schema checks in `DatasetManager` before `materializeTable`, with a shared
helper for the stored schema.
##########
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:
For this PR I kept the scope as is and fixed the scaladoc in 85cc86b395f: it
now says the check covers the `UserDefinedExpression`s in a query, including
UDAF classes registered with `CREATE FUNCTION ... AS`, that SQL functions are
checked by `CREATE FUNCTION`, and that V2 catalog functions are covered only
when bound as an `ApplyFunctionExpression`. Checking `resultType()` in
`resolveV2Function` would cover all V2 functions; I will propose it as a
follow-up.
##########
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:
Agreed it should be one rule. As you note, nothing can reach it until
PySpark and Spark Connect have a FILE type, so I would settle the rule for
row-returning functions, with a test, in the PySpark part of the SPIP rather
than here.
##########
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:
Added `approx_count_distinct`, `mode` and `collect_union` to the deferred
aggregates in the description. A shared helper for VARIANT and FILE, and
extending the VARIANT checks to `REBALANCE` and `DeduplicateWithinWatermark`,
touch the existing VARIANT rules, so I will propose them as a follow-up rather
than change them here.
##########
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:
The per-column order is the same as for MAP and VARIANT today, so this PR
keeps it. Checking one type at a time over all columns would be better changed
for all the types at once, so I will propose it as a follow-up.
##########
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:
Thanks, the description now says the single-pass side of the repartition
rule is not wired up yet, and that the check there is kept for parity with
VARIANT.
##########
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:
Fixed in 85cc86b395f: V1 bucket columns are rejected as `PARTITION_COLUMN`,
as for the V2 `bucket` transform, and the test covers `CLUSTERED BY (f) INTO 4
BUCKETS`. Bucket sort columns are already rejected by the existing ordering
check, so I left them.
--
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]
