dongjoon-hyun commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4117400458
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -50,6 +50,18 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
val partitioning = sequenceToOption(
kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
.toImmutableArraySeq)
+ if (partitioning.isEmpty) {
+ val unresolvedRefs = kgp.keys().flatMap(_.references()).filter {
ref =>
Review Comment:
`Transform.references()` returns only the top-level `NamedReference`
arguments for `ApplyTransform` (`arguments.collect { case named: NamedReference
=> named }`). Since this PR routes `NamedTransform` args through the
non-throwing path, a key like `f(g(missing))` now degrades to `None` while
`unresolvedRefs` is empty, so no warning is logged at all. Before this PR, the
query failed with `_LEGACY_ERROR_TEMP_1137`. Could you collect the references
recursively so that this case is not silently ignored?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -39,24 +39,38 @@ import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.connector.PartitionPredicateImpl
import org.apache.spark.sql.types._
import org.apache.spark.util.ArrayImplicits._
+import org.apache.spark.util.collection.Utils.sequenceToOption
/**
* A utility class that converts public connector expressions into Catalyst
expressions.
*/
object V2ExpressionUtils extends SQLConfHelper with Logging {
import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+ /**
+ * Non-throwing variant of `resolveRef`. Returns `None` if the reference
cannot be resolved.
Review Comment:
BTW, the scaladoc wording that I suggested earlier turns out to be
imprecise. `resolveRefOpt` is not strictly non-throwing: for a multi-part name,
`LogicalPlan.resolve` goes through `ExtractValue.applyOrNull`, so a key like
`s.missing` (where `s` is an existing struct column) still throws
`FIELD_NOT_FOUND`, and an ambiguous reference throws as well. So the fallback
is still partial for nested partition keys. Could you mention this limitation
in the scaladoc and the PR description, or add a test that pins the current
behavior?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -142,27 +166,34 @@ object V2ExpressionUtils extends SQLConfHelper with
Logging {
def toCatalystTransformOpt(
trans: Transform,
query: LogicalPlan,
- funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] =
trans match {
+ funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] =
+ toCatalystTransformCore(trans, query, funCatalogOpt, failOnUnresolved =
false)
+
+ private def toCatalystTransformCore(
+ trans: Transform,
+ query: LogicalPlan,
+ funCatalogOpt: Option[FunctionCatalog],
+ failOnUnresolved: Boolean): Option[Expression] = trans match {
case IdentityTransform(ref) =>
- Some(resolveRef[NamedExpression](ref, query))
+ resolveRefOrDegrade[NamedExpression](ref, query, failOnUnresolved)
case BucketTransform(numBuckets, refs, sorted)
if sorted.isEmpty && refs.length == 1 &&
refs.forall(_.isInstanceOf[NamedReference]) =>
- val resolvedRefs = refs.map(r => resolveRef[NamedExpression](r, query))
// Create a dummy reference for `numBuckets` here and use that, together
with `refs`, to
// look up the V2 function.
val numBucketsRef = AttributeReference("numBuckets", IntegerType,
nullable = false)()
- funCatalogOpt.flatMap { catalog =>
- loadV2FunctionOpt(catalog, "bucket", Seq(numBucketsRef) ++
resolvedRefs).map { bound =>
- TransformExpression(bound, resolvedRefs, Some(numBuckets))
- }
- }
+ for {
+ resolvedRefs <- sequenceToOption(
+ refs.map(r => resolveRefOrDegrade[NamedExpression](r, query,
failOnUnresolved)))
+ catalog <- funCatalogOpt
+ bound <- loadV2FunctionOpt(catalog, "bucket", Seq(numBucketsRef) ++
resolvedRefs)
+ } yield TransformExpression(bound, resolvedRefs, Some(numBuckets))
case NamedTransform(name, args) =>
- val catalystArgs = args.map(toCatalyst(_, query, funCatalogOpt))
- funCatalogOpt.flatMap { catalog =>
- loadV2FunctionOpt(catalog, name, catalystArgs).map { bound =>
- TransformExpression(bound, catalystArgs)
- }
- }
+ for {
+ catalystArgs <- sequenceToOption(
+ args.map(toCatalystCore(_, query, funCatalogOpt, failOnUnresolved)))
Review Comment:
As mentioned in the PR description, a nested transform whose function cannot
be loaded now returns `None` here instead of failing with
`_LEGACY_ERROR_TEMP_3054`. Note that this drop is completely silent when
`funCatalogOpt` is `None` or `loadFunction` throws
`UnsupportedOperationException`: `loadV2FunctionOpt` warns only on
`NoSuchFunctionException`, and the new warning in
`V2ScanPartitioningAndOrdering` is not triggered because all references are
resolvable. Since this turns a query failure into a silent loss of SPJ, could
you at least add a test for this case so that the new behavior is explicit?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -50,6 +50,18 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
val partitioning = sequenceToOption(
kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
.toImmutableArraySeq)
+ if (partitioning.isEmpty) {
Review Comment:
Not a blocker, but since `toCatalystOpt` does not tell why it returned
`None`, this block re-resolves every reference to reconstruct the reason, and
the two can diverge (e.g., the nested case in my other comment on
`unresolvedRefs`). Have you considered checking the key references
(recursively) against `relation` up front in this rule, and logging the warning
and returning `None` before calling `toCatalystOpt`? Then the fallback decision
and the warning would come from the same place.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -50,6 +50,18 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
val partitioning = sequenceToOption(
kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
.toImmutableArraySeq)
+ if (partitioning.isEmpty) {
+ val unresolvedRefs = kgp.keys().flatMap(_.references()).filter {
ref =>
+ V2ExpressionUtils.resolveRefOpt[NamedExpression](ref,
relation).isEmpty
+ }
+ if (unresolvedRefs.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ${MDC(CLASS_NAME,
kgp.getClass.getSimpleName)} " +
+ log"because the partition key columns cannot be resolved: " +
+ log"${MDC(COLUMN_NAMES,
unresolvedRefs.map(_.describe()).mkString(", "))}. " +
+ log"Storage-partitioned join will not be applied for this
scan.")
Review Comment:
This message implies that the query proceeds without SPJ, but `ordering()`
below still uses the throwing `toCatalystOrdering`. If a scan also reports
`outputOrdering()` on the same unresolvable column, the query still fails with
`_LEGACY_ERROR_TEMP_1137`. Should we degrade the reported ordering in the same
way (log a warning and drop it), or at least clarify this scope in the PR
description?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala:
##########
@@ -20,21 +20,84 @@ package org.apache.spark.sql.catalyst.expressions
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
+import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryCatalog}
+import org.apache.spark.sql.connector.catalog.functions.{BoundFunction,
ScalarFunction, UnboundFunction}
import org.apache.spark.sql.connector.expressions._
-import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.types.{DataType, IntegerType, StringType,
StructType}
class V2ExpressionUtilsSuite extends SparkFunSuite {
+ private val plan = LocalRelation.apply(AttributeReference("a", StringType)())
+
+ private object AnyInputFunction extends UnboundFunction {
+ override def name(): String = "any_input"
+ override def description(): String = name()
+ override def bind(inputType: StructType): BoundFunction = new
ScalarFunction[Int] {
+ override def inputTypes(): Array[DataType] =
inputType.fields.map(_.dataType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "any_input"
+ }
+ }
+
+ private val funCatalog = {
+ val catalog = new InMemoryCatalog
+ Seq("bucket", "years").foreach { name =>
+ catalog.createFunction(Identifier.of(Array.empty, name),
AnyInputFunction)
+ }
+ Some(catalog)
+ }
+
test("SPARK-39313: toCatalystOrdering should fail if V2Expression can not be
translated") {
val supportedV2Sort = SortValue(
FieldReference("a"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
val unsupportedV2Sort = supportedV2Sort.copy(
expression = ApplyTransform("v2Fun", FieldReference("a") :: Nil))
val exc = intercept[AnalysisException] {
- V2ExpressionUtils.toCatalystOrdering(
- Array(supportedV2Sort, unsupportedV2Sort),
- LocalRelation.apply(AttributeReference("a", StringType)()))
+ V2ExpressionUtils.toCatalystOrdering(Array(supportedV2Sort,
unsupportedV2Sort), plan)
}
assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently
supported"))
}
+
+ test("SPARK-59721: toCatalystOpt returns None, not throw, for an
unresolvable FieldReference") {
+ assert(V2ExpressionUtils.toCatalystOpt(FieldReference("missing"),
plan).isEmpty)
+ }
+
+ test("SPARK-59721: toCatalyst throws the specific resolution error for an
unresolvable " +
Review Comment:
`failOnUnresolved` is also threaded through the `BucketTransform` refs and
the `NamedTransform` args, but only `FieldReference` and `SortValue` are
covered for the throwing path. Could you add cases like
`toCatalyst(Expressions.bucket(4, "missing"), plan, funCatalog)` and
`toCatalyst(Expressions.years("missing"), plan, funCatalog)` expecting
`_LEGACY_ERROR_TEMP_1137`? This is the path that `DistributionAndOrderingUtils`
uses for `ClusteredDistribution` on the write side.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9259,3 +9304,43 @@ class KeyGroupedPartitioningCatalystRuntimeFilterSuite
catalog.clearTables()
}
}
+
+/**
+ * An in-memory catalog whose tables report a `KeyGroupedPartitioning` keyed
on a column that
+ * does not exist in the table.
+ */
+class UnresolvablePartitionKeyCatalog extends InMemoryTableCatalog {
+ // scalastyle:off argcount
+ override protected def newInMemoryTable(
+ name: String,
+ columns: Array[Column],
+ partitioning: Array[Transform],
+ properties: util.Map[String, String],
+ constraints: Array[Constraint],
+ distribution: Distribution,
+ ordering: Array[SortOrder],
+ requiredNumPartitions: Option[Int],
+ advisoryPartitionSize: Option[Long],
+ distributionStrictlyRequired: Boolean,
+ numRowsPerSplit: Int,
+ id: String): InMemoryBaseTable = {
+ // scalastyle:on argcount
+ new InMemoryTable(name, columns, partitioning, properties, constraints,
distribution,
+ ordering, requiredNumPartitions, advisoryPartitionSize,
distributionStrictlyRequired,
+ numRowsPerSplit, id) {
+ override def newScanBuilder(options: CaseInsensitiveStringMap):
ScanBuilder =
+ new InMemoryScanBuilder(schema(), options) {
+ override def build(): Scan =
UnresolvablePartitionKeyScan(super.build())
+ }
+ }
+ }
+}
+
+case class UnresolvablePartitionKeyScan(inner: Scan) extends
SupportsReportPartitioning {
+ override def readSchema(): StructType = inner.readSchema()
+ override def toBatch: Batch = inner.toBatch
+ override def description(): String = inner.description()
+ override def outputPartitioning(): Partitioning = new KeyGroupedPartitioning(
+ Array[Expression](FieldReference("missing")),
Review Comment:
This end-to-end test covers only a single identity key. Could you also cover
a transform key (e.g., `bucket(4, missing)`), a nested transform key, and a
multi-key case where only one key is unresolvable (e.g., `[id, missing]`) to
verify that the whole partitioning is dropped? The nested case would catch the
missing warning which I mentioned in `V2ScanPartitioningAndOrdering`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -50,6 +50,18 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
val partitioning = sequenceToOption(
kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
.toImmutableArraySeq)
+ if (partitioning.isEmpty) {
+ val unresolvedRefs = kgp.keys().flatMap(_.references()).filter {
ref =>
+ V2ExpressionUtils.resolveRefOpt[NamedExpression](ref,
relation).isEmpty
+ }
+ if (unresolvedRefs.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ${MDC(CLASS_NAME,
kgp.getClass.getSimpleName)} " +
+ log"because the partition key columns cannot be resolved: " +
+ log"${MDC(COLUMN_NAMES,
unresolvedRefs.map(_.describe()).mkString(", "))}. " +
Review Comment:
nit. Could you add `.distinct` to `unresolvedRefs`? Otherwise, the same
column is listed multiple times when several keys reference it, e.g.,
`identity(missing)` and `bucket(4, missing)` give `missing, missing`. In
addition, the same warning can be logged more than once for a query (e.g., when
`PlanMerger` re-applies this rule to a merged scan), but that's the same as the
existing warning below.
--
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]
