This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 9de605f012ec [SPARK-59721][SQL] Ignore reported partitioning and
ordering that reference unresolvable columns
9de605f012ec is described below
commit 9de605f012ec8689dda0acb21103cc5744c85981
Author: Anurag Mantripragada <[email protected]>
AuthorDate: Mon Oct 5 10:04:38 2026 -0700
[SPARK-59721][SQL] Ignore reported partitioning and ordering that reference
unresolvable columns
### What changes were proposed in this pull request?
When a V2 scan reports a `KeyGroupedPartitioning` or an `ordering` that
references a column the relation can't resolve, the query used to fail with
`Unable to resolve X given [...]`. This PR makes
`V2ScanPartitioningAndOrdering` check the columns referenced by the reported
partition keys and ordering (recursively, including nested transform arguments)
against the relation before converting them. If any can't be resolved,
including a missing nested field, an ambiguous name, or an empty [...]
This PR adds a `resolveRefOpt`, and makes `resolveRef` a thin wrapper
around it. `resolveRefOpt` still throws for a missing nested field or an
ambiguous name, so the rule treats those errors as unresolvable and includes
the error condition in the warning. The `toCatalyst*` methods are unchanged.
Keys whose columns all resolve take the existing conversion path unchanged,
so an unsupported expression type (e.g. `id + 1`) or a nested transform whose
function can't be loaded still fails the query as before. Because the whole
report is checked before any key is converted, a report that also references an
unresolvable column falls back with a warning instead, e.g. `[id + 1,
identity(missing)]`.
The Javadoc of `KeyGroupedPartitioning#keys()` and
`SupportsReportOrdering#outputOrdering()` now says that Spark resolves the
column references against the table columns, including columns pruned from the
scan output, and ignores the whole report if any of them can't be resolved. The
docs of `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` now
mention that an ignored ordering can be replaced by one derived from the
partition keys.
### Why are the changes needed?
Found while reviewing SPARK-58111 ([PR
#55518](https://github.com/apache/spark/pull/55518)). This is a general,
pre-existing bug in how `V2ScanPartitioningAndOrdering` handles reported
partitioning and ordering, not something introduced by #55518.
### Does this PR introduce _any_ user-facing change?
Yes. A query against a data source whose scan reports a
`KeyGroupedPartitioning` with a key that can't be resolved previously failed
outright. It now succeeds, with partitioning degraded to `UnknownPartitioning`
instead, and Spark logs a warning naming the relation, the scan class, and the
columns that can't be resolved.
The same applies to an ambiguous or empty column name. A report that
combines an unsupported expression with an unresolvable column, e.g. `missing +
1` or `[id + 1, identity(missing)]`, used to fail with
`_LEGACY_ERROR_TEMP_3054` and now falls back with a warning. An unsupported
expression over resolvable columns still fails as before.
If the scan keeps its partitioning but its reported ordering is ignored,
Spark derives the ordering from the partition keys when
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on (the
default).
### How was this patch tested?
New tests in `V2ExpressionUtilsSuite`:
- `resolveRefOpt` returns `None` for an unresolvable reference.
- `resolveRefOpt` throws `FIELD_NOT_FOUND` for a missing nested field and
`AMBIGUOUS_REFERENCE` for an ambiguous reference.
New tests in `KeyGroupedPartitioningSuite`:
- A join against a table whose scan reports unresolvable partition keys
succeeds, plans a shuffle instead of a storage-partitioned join, and logs a
warning naming the relation, the scan class, and only the unresolvable columns.
The cases are:
- a `bucket` key, a nested transform key, and a multi-key partitioning
with one bad key;
- duplicate references, and two distinct unresolvable columns;
- a missing nested field;
- an unsupported expression next to or over an unresolvable column (`[id
+ 1, identity(missing)]`, `missing + 1`);
- a connector-defined transform that hides the reference from
`children()`.
Resolvable keys (`id`, case-insensitive `ID`, and a connector-defined
transform with an extra reference only in `children()`) keep the partitioning
and plan a storage-partitioned join without a shuffle.
- A reported ordering that can't be resolved is ignored as a whole with a
warning, and the scan still derives `id ASC` from its kept partitioning. The
cases are a column that doesn't exist, one hidden from a connector-defined sort
order's `children()`, an empty name, and a two-key ordering with one
unresolvable key (`[id, missing]`). Resolvable orderings (`id`,
case-insensitive `ID`, the metadata column `index`, and a connector-defined
sort order with an extra reference only in `child [...]
- A reported partition key of an unsupported expression type (`id + 1`), or
a nested transform whose function can't be loaded (`f(g(id))`), still fails the
query with `_LEGACY_ERROR_TEMP_3054`.
New test in `DataSourceV2Suite`:
- An unresolvable ordering reported through connector-defined `SortOrder`,
`Transform`, and `NamedReference` classes (`OrderAndPartitionAwareDataSource`
and its Java counterpart) is ignored with a warning, and the scan relation
keeps no ordering.
Also ran `V2ExpressionUtilsSuite`, `KeyGroupedPartitioningSuite`,
`WriteDistributionAndOrderingSuite`, `DataSourceV2Suite`, `MergeSubplansSuite`
(comment update only), `SQLConfSuite`, and
`ProjectedOrderingAndPartitioningSuite` to confirm no regression in the
partitioning/SPJ, ordering, and write paths this change touches.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Sonnet 5, Opus 5.5)
Verified manually by me.
Closes #58979 from anuragmantri/v2expr-opt-resolve-degrade.
Authored-by: Anurag Mantripragada <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
docs/sql-performance-tuning.md | 2 +-
.../sql/connector/read/SupportsReportOrdering.java | 4 +
.../read/partitioning/KeyGroupedPartitioning.java | 4 +
.../catalyst/expressions/V2ExpressionUtils.scala | 23 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 3 +-
.../expressions/V2ExpressionUtilsSuite.scala | 28 ++-
.../datasources/v2/DataSourceV2ScanExecBase.scala | 18 +-
.../v2/V2ScanPartitioningAndOrdering.scala | 80 ++++++-
.../sql/execution/planmerging/PlanMerger.scala | 13 +-
.../spark/sql/connector/DataSourceV2Suite.scala | 29 +++
.../connector/KeyGroupedPartitioningSuite.scala | 239 ++++++++++++++++++++-
.../execution/planmerging/MergeSubplansSuite.scala | 10 +-
12 files changed, 409 insertions(+), 44 deletions(-)
diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md
index 5e9f7aae05f0..232c47d16b13 100644
--- a/docs/sql-performance-tuning.md
+++ b/docs/sql-performance-tuning.md
@@ -708,7 +708,7 @@ The following SQL properties enable Storage Partition Join
in different join que
<td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
<td>true</td>
<td>
- When enabled, Spark derives the output ordering of a V2 scan from its
partition key expressions, if the source reports a keyed partitioning but no
explicit ordering. All rows of such a partition share one key value, so the
partition is trivially sorted by those expressions, and a sort Spark would
otherwise add becomes unnecessary. This config requires
<code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ When enabled, Spark derives the output ordering of a V2 scan from its
partition key expressions, if the source reports a keyed partitioning but no
explicit ordering, or an ordering that Spark ignores because it references a
column that cannot be resolved. All rows of such a partition share one key
value, so the partition is trivially sorted by those expressions, and a sort
Spark would otherwise add becomes unnecessary. This config requires
<code>spark.sql.sources.v2.bucketing.ena [...]
</td>
<td>4.2.0</td>
</tr>
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsReportOrdering.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsReportOrdering.java
index 0dc102d11c21..cf1a0532d3c0 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsReportOrdering.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsReportOrdering.java
@@ -35,6 +35,10 @@ public interface SupportsReportOrdering extends Scan {
/**
* Returns the order in each partition of this data source scan.
+ * <p>
+ * Spark resolves the column references in these sort orders against the
table columns,
+ * including columns pruned from the scan output. If any of them cannot be
resolved, Spark
+ * ignores the whole reported ordering and logs a warning.
*/
SortOrder[] outputOrdering();
}
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/partitioning/KeyGroupedPartitioning.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/partitioning/KeyGroupedPartitioning.java
index 552d92ad0e8b..88ee89af8329 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/partitioning/KeyGroupedPartitioning.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/partitioning/KeyGroupedPartitioning.java
@@ -43,6 +43,10 @@ public class KeyGroupedPartitioning implements Partitioning {
/**
* Returns the partition transform expressions for this partitioning.
+ * <p>
+ * Spark resolves the column references in these expressions against the
table columns,
+ * including columns pruned from the scan output. If any of them cannot be
resolved, Spark
+ * ignores the whole reported partitioning and logs a warning.
*/
public Expression[] keys() {
return keys;
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala
index af84df8fcc06..2ca5a499b84e 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala
@@ -46,15 +46,22 @@ import org.apache.spark.util.ArrayImplicits._
object V2ExpressionUtils extends SQLConfHelper with Logging {
import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+ /**
+ * Variant of `resolveRef` that returns `None` if no attribute matches the
reference. It still
+ * throws for a nested-field extraction error, such as a missing nested
field, or an ambiguous
+ * reference.
+ */
+ private[sql] def resolveRefOpt(
+ ref: NamedReference, plan: LogicalPlan): Option[NamedExpression] = {
+ plan.resolve(ref.fieldNames.toImmutableArraySeq, conf.resolver)
+ }
+
def resolveRef[T <: NamedExpression](ref: NamedReference, plan:
LogicalPlan): T = {
- plan.resolve(ref.fieldNames.toImmutableArraySeq, conf.resolver) match {
- case Some(namedExpr) =>
- namedExpr.asInstanceOf[T]
- case None =>
- val name = ref.fieldNames.toImmutableArraySeq.quoted
- val outputString = plan.output.map(_.name).mkString(",")
- throw QueryCompilationErrors.cannotResolveAttributeError(name,
outputString)
- }
+ resolveRefOpt(ref, plan).getOrElse {
+ val name = ref.fieldNames.toImmutableArraySeq.quoted
+ val outputString = plan.output.map(_.name).mkString(",")
+ throw QueryCompilationErrors.cannotResolveAttributeError(name,
outputString)
+ }.asInstanceOf[T]
}
def resolveRefs[T <: NamedExpression](refs: Seq[NamedReference], plan:
LogicalPlan): Seq[T] = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index b4254747225a..8edf1d7cf899 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -2590,7 +2590,8 @@ object SQLConf {
buildConf("spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled")
.doc("When enabled, Spark derives output ordering from the partition key
expressions of " +
"a V2 data source that reports a KeyedPartitioning but does not report
explicit ordering " +
- "via SupportsReportOrdering. Within a single partition all rows share
the same key " +
+ "via SupportsReportOrdering, or reports one that Spark ignores because
it references a " +
+ "column that cannot be resolved. Within a single partition all rows
share the same key " +
s"value, so the data is trivially sorted by those expressions.
Requires " +
s"${V2_BUCKETING_ENABLED.key} to be enabled.")
.version("4.2.0")
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala
index d1c23d68555a..899ddf5e9d3b 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala
@@ -21,7 +21,7 @@ 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.expressions._
-import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
class V2ExpressionUtilsSuite extends SparkFunSuite {
@@ -37,4 +37,30 @@ class V2ExpressionUtilsSuite extends SparkFunSuite {
}
assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently
supported"))
}
+
+ test("SPARK-59721: resolveRefOpt returns None for an unresolvable
reference") {
+ val plan = LocalRelation(AttributeReference("a", StringType)())
+ assert(V2ExpressionUtils.resolveRefOpt(FieldReference("a"),
plan).isDefined)
+ assert(V2ExpressionUtils.resolveRefOpt(FieldReference("missing"),
plan).isEmpty)
+ }
+
+ test("SPARK-59721: resolveRefOpt throws for a missing nested field or an
ambiguous reference") {
+ val structPlan =
+ LocalRelation(AttributeReference("s", new StructType().add("x",
IntegerType))())
+ checkError(
+ exception = intercept[AnalysisException] {
+ V2ExpressionUtils.resolveRefOpt(FieldReference("s.missing"),
structPlan)
+ },
+ condition = "FIELD_NOT_FOUND",
+ parameters = Map("fieldName" -> "`missing`", "fields" -> "`x`"))
+
+ val ambiguousPlan = LocalRelation(
+ AttributeReference("a", StringType)(), AttributeReference("a",
StringType)())
+ checkError(
+ exception = intercept[AnalysisException] {
+ V2ExpressionUtils.resolveRefOpt(FieldReference("a"), ambiguousPlan)
+ },
+ condition = "AMBIGUOUS_REFERENCE",
+ parameters = Map("name" -> "`a`", "referenceNames" -> "[`a`, `a`]"))
+ }
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala
index 5d1617111885..49bee1e7961c 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala
@@ -135,15 +135,15 @@ trait DataSourceV2ScanExecBase
/**
* Returns the output ordering for this scan. When the source reports
ordering via
- * `SupportsReportOrdering`, that ordering may reference columns pruned out
of the scan output
- * (see V2ScanPartitioningAndOrdering), while a consumer binds it against
the output, e.g. the
- * k-way merge of `GroupPartitionsExec`. Ordering is prefix-based, so the
leading run of sort
- * orders over the output is kept and the rest is dropped, except the sort
orders on a partition
- * key when the output partitioning is a `KeyedPartitioning`: each partition
holds a single key,
- * so those still hold. Otherwise, when the output partitioning is a
`KeyedPartitioning` and
- * `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on, each
partition
- * contains rows where the key expressions evaluate to a single constant
value, so the data
- * is trivially sorted by those expressions within the partition.
+ * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it,
that ordering may
+ * reference columns pruned out of the scan output, while a consumer binds
it against the
+ * output, e.g. the k-way merge of `GroupPartitionsExec`. Ordering is
prefix-based, so the
+ * leading run of sort orders over the output is kept and the rest is
dropped, except the sort
+ * orders on a partition key when the output partitioning is a
`KeyedPartitioning`: each
+ * partition holds a single key, so those still hold. Otherwise, when the
output partitioning
+ * is a `KeyedPartitioning` and
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
+ * is on, each partition contains rows where the key expressions evaluate to
a single constant
+ * value, so the data is trivially sorted by those expressions within the
partition.
*/
override def outputOrdering: Seq[SortOrder] = {
(ordering, outputPartitioning) match {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
index 5cc5d2b494fe..d001b20b5d36 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
@@ -17,11 +17,14 @@
package org.apache.spark.sql.execution.datasources.v2
import org.apache.spark.internal.Logging
-import org.apache.spark.internal.LogKeys.CLASS_NAME
+import org.apache.spark.internal.LogKeys.{CLASS_NAME, COLUMN_NAMES,
RELATION_NAME}
+import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.expressions.V2ExpressionUtils
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
import
org.apache.spark.sql.catalyst.trees.TreePattern.DATA_SOURCE_V2_SCAN_RELATION
+import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+import org.apache.spark.sql.connector.expressions.{Expression => V2Expression,
NamedReference, SortOrder => V2SortOrder, Transform}
import org.apache.spark.sql.connector.read.{SupportsReportOrdering,
SupportsReportPartitioning}
import
org.apache.spark.sql.connector.read.partitioning.{KeyGroupedPartitioning,
UnknownPartitioning}
import org.apache.spark.util.ArrayImplicits._
@@ -47,9 +50,19 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
if d.keyGroupedPartitioning.isEmpty =>
val catalystPartitioning = scan.outputPartitioning() match {
case kgp: KeyGroupedPartitioning =>
- val partitioning = sequenceToOption(
- kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
- .toImmutableArraySeq)
+ val keys = kgp.keys().toImmutableArraySeq
+ val unresolvedColumns = unresolvableColumns(keys, relation)
+ val partitioning = if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the KeyGroupedPartitioning reported by " +
+ log"${MDC(RELATION_NAME, relation.name)} (scan " +
+ log"${MDC(CLASS_NAME, scan.getClass.getName)}) because the
partition key columns " +
+ log"cannot be resolved: ${MDC(COLUMN_NAMES,
unresolvedColumns.mkString(", "))}.")
+ None
+ } else {
+ sequenceToOption(
+ keys.map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog)))
+ }
// Keep the partitioning when at least one of its keys is still in
the scan output: the
// scan projects the pruned key positions away when reporting its
physical output
// partitioning (see DataSourceV2ScanExecBase.outputPartitioning).
When no key survives,
@@ -75,12 +88,57 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
private def ordering(plan: LogicalPlan) = plan.transformDownWithPruning(
_.containsPattern(DATA_SOURCE_V2_SCAN_RELATION)) {
case d @ ExtractV2ScanInfo(relation, scan: SupportsReportOrdering, _) =>
- // The ordering is kept as reported, even where it references columns
pruned out of the scan
- // output: truncating it here would also drop the sort orders on a
partition key past a
- // pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
- // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val reportedOrdering = scan.outputOrdering()
+ val unresolvedColumns =
unresolvableColumns(reportedOrdering.toImmutableArraySeq, relation)
+ if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the ordering reported by ${MDC(RELATION_NAME,
relation.name)} " +
+ log"(scan ${MDC(CLASS_NAME, scan.getClass.getName)}) because the
ordering columns " +
+ log"cannot be resolved: ${MDC(COLUMN_NAMES,
unresolvedColumns.mkString(", "))}.")
+ // Use None, not Some(Nil): only None lets
DataSourceV2ScanExecBase.outputOrdering derive an
+ // ordering from a kept key-grouped partitioning, which does not
depend on the dropped
+ // report.
+ d.copy(ordering = None)
+ } else {
+ // The ordering is kept as reported, even where it references columns
pruned out of the
+ // scan output: truncating it here would also drop the sort orders on
a partition key past a
+ // pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
+ // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ // Walk the accessors the conversion reads, `Transform.arguments()` and
+ // `SortOrder.expression()`, instead of `V2Expression.references()` or
`children()`: a connector
+ // can override those to disagree with the conversion, and
`ApplyTransform.references()` returns
+ // only its top-level arguments, which would miss `f(g(missing))`. Other
expressions, such as
+ // `GeneralScalarExpression`, which the conversion rejects, go through
`children()`.
+ def collectReferences(expr: V2Expression): Seq[NamedReference] = expr
match {
+ case ref: NamedReference => Seq(ref)
+ case t: Transform =>
t.arguments().toImmutableArraySeq.flatMap(collectReferences)
+ case s: V2SortOrder => collectReferences(s.expression())
+ case other =>
other.children().toImmutableArraySeq.flatMap(collectReferences)
+ }
+ exprs.flatMap(collectReferences).flatMap { ref =>
+ val parts = ref.fieldNames.toImmutableArraySeq
+ if (parts.isEmpty) {
+ // An empty name can be neither quoted nor resolved: both throw.
+ Some("<empty>")
+ } else {
+ val name = parts.quoted
+ try {
+ Option.when(V2ExpressionUtils.resolveRefOpt(ref,
relation).isEmpty)(name)
+ } catch {
+ // A nested-field extraction error or an ambiguous reference throws
instead of returning
+ // None.
+ case e: AnalysisException => Some(s"$name (${e.getCondition})")
+ }
+ }
+ }.distinct
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
index 0cf12e8b5e1d..2eb50ce63a4e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
@@ -980,12 +980,13 @@ class PlanMerger(
// Only the reported EXPRESSIONS are compared, which is all a
DataSourceV2ScanRelation carries;
// the merged scan's split count and partition values can still differ from
an input's, since it
// may push a different best-effort filter and so prune differently. And a
report the merged scan
- // GAINS is not a degradation either. For partitioning that is because an
input dropped its own
- // only where none of its keys survived in that scan's output
(V2ScanPartitioningAndOrdering's
- // partitioning pass keeps the report whenever any key survives), not
because the source stopped
- // reporting; the ordering pass has no such guard, so an ordering report is
- // never dropped by pruning and a gained one can only come from the source.
Either way, keeping it
- // is exactly the win this merge is after.
+ // GAINS is not a degradation either. Both V2ScanPartitioningAndOrdering
passes drop a report
+ // that references a column the relation cannot resolve, but that does not
depend on pruning.
+ // Beyond that, an input dropped its partitioning only where none of its
keys survived in that
+ // scan's output (the partitioning pass keeps the report whenever any key
survives), not because
+ // the source stopped reporting; the ordering pass has no such guard, so an
ordering report is
+ // never dropped by pruning and a gained one can only come from the source.
Either way, keeping
+ // it is exactly the win this merge is after.
private def mergeDegradesReporting(
merged: DataSourceV2ScanRelation,
requiredKeyGroupedPartitioning: Seq[Expression],
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
index 55febd40db47..e269e92a2442 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
@@ -24,6 +24,7 @@ import java.util.OptionalLong
import scala.jdk.CollectionConverters._
+import org.apache.logging.log4j.Level
import test.org.apache.spark.sql.connector._
import org.apache.spark.{SparkException, SparkUnsupportedOperationException}
@@ -397,6 +398,34 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
+ test("SPARK-59721: ignore an unresolvable ordering reported with
connector-defined classes") {
+ Seq(
+ classOf[OrderAndPartitionAwareDataSource],
+ classOf[JavaOrderAndPartitionAwareDataSource]
+ ).foreach { cls =>
+ withClue(cls.getName) {
+ val df = spark.read
+ .option("partitionKeys", "i")
+ .option("orderKeys", "missing")
+ .format(cls.getName)
+ .load()
+ val logAppender = new LogAppender("unresolvable reported ordering")
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ checkAnswer(df, Seq(Row(1, 4), Row(1, 5), Row(3, 5), Row(2, 6),
Row(4, 1), Row(4, 2)))
+ }
+ val scanRelation = getScanRelation(df)
+ val expectedWarning =
+ s"Spark ignores the ordering reported by ${scanRelation.name} " +
+ s"(scan ${scanRelation.scan.getClass.getName}) because the
ordering columns " +
+ "cannot be resolved: missing."
+ val warnings =
logAppender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ .filter(_.contains("columns cannot be resolved"))
+ assert(warnings.toSet == Set(expectedWarning), warnings)
+ assert(scanRelation.ordering.isEmpty)
+ }
+ }
+ }
+
test ("statistics report data source") {
Seq(classOf[ReportStatisticsDataSource],
classOf[JavaReportStatisticsDataSource]).foreach {
cls =>
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index ea1fafd94cfe..0b037ed6829b 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -19,20 +19,25 @@ package org.apache.spark.sql.connector
import java.sql.Timestamp
import java.util.Collections
+import org.apache.logging.log4j.Level
+
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.rdd.SortedMergeCoalescedRDD
-import org.apache.spark.sql.{DataFrame, ExplainSuiteHelper, Row}
+import org.apache.spark.sql.{AnalysisException, DataFrame, ExplainSuiteHelper,
Row}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Ascending,
AttributeReference, ExprId, Literal, TransformExpression}
import org.apache.spark.sql.catalyst.expressions.aggregate.Complete
import org.apache.spark.sql.catalyst.plans.{Cross, ExistenceJoin, Inner,
JoinType, LeftAnti, LeftSemi, LeftSingle}
import org.apache.spark.sql.catalyst.plans.physical
import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
-import org.apache.spark.sql.connector.catalog.{Column, Identifier,
InMemoryCatalystRuntimeFilterCatalog, InMemoryTableCatalog}
+import org.apache.spark.sql.connector.catalog.{Column, Identifier,
InMemoryBaseTable, InMemoryCatalystRuntimeFilterCatalog, InMemoryTable,
InMemoryTableCatalog}
+import org.apache.spark.sql.connector.catalog.constraints.Constraint
import org.apache.spark.sql.connector.catalog.functions._
-import org.apache.spark.sql.connector.distributions.Distributions
+import org.apache.spark.sql.connector.distributions.{Distribution,
Distributions}
import org.apache.spark.sql.connector.expressions._
import org.apache.spark.sql.connector.expressions.Expressions._
+import org.apache.spark.sql.connector.read.{Batch, Scan, ScanBuilder,
SupportsReportOrdering, SupportsReportPartitioning}
+import
org.apache.spark.sql.connector.read.partitioning.{KeyGroupedPartitioning,
Partitioning}
import org.apache.spark.sql.execution.{
ExtendedMode,
FormattedMode,
@@ -56,6 +61,7 @@ import org.apache.spark.sql.functions.{col, max}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf._
import org.apache.spark.sql.types._
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.tags.ExtendedSQLTest
abstract class KeyGroupedPartitioningSuiteBase extends
DistributionAndOrderingSuiteBase {
@@ -3932,6 +3938,161 @@ class KeyGroupedPartitioningSuite
Row(1, "bb", "bb"), Row(2, "cc", "cc")))
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+ private val customReportingTableName =
s"$customReportingCatalogName.ns.$table"
+ private def plusOneKey(column: String): Expression =
+ new GeneralScalarExpression("+", Array[Expression](FieldReference(column),
literal(1)))
+ private def nestedKey(column: String): Expression =
+ ApplyTransform("f", Seq(ApplyTransform("g", Seq(FieldReference(column)))))
+ private def ignoredReportWarning(report: String, kind: String, columns:
String): String =
+ s"Spark ignores the $report reported by $customReportingTableName " +
+ s"(scan ${classOf[CustomReportingScan].getName}) because the $kind
columns " +
+ s"cannot be resolved: $columns."
+
+ private def withCustomReportingTable(f: CustomReportingCatalog => Unit):
Unit = {
+ withSQLConf(
+ s"spark.sql.catalog.$customReportingCatalogName" ->
classOf[CustomReportingCatalog].getName,
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val reportingCatalog =
spark.sessionState.catalogManager.catalog(customReportingCatalogName)
+ .asInstanceOf[CustomReportingCatalog]
+ try {
+ val reportingColumns = Array(
+ Column.create("id", IntegerType),
+ Column.create("data", StringType),
+ Column.create("s", new StructType().add("x", IntegerType)))
+ createTable(table, reportingColumns, Array(identity("id")), catalog =
reportingCatalog)
+ sql(s"INSERT INTO $customReportingTableName VALUES " +
+ "(1, 'aa', named_struct('x', 10)), (2, 'bb', named_struct('x',
20))")
+ createTable(table, columns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$table VALUES " +
+ "(1, 'cc', cast('2020-01-03' as timestamp)), " +
+ "(2, 'dd', cast('2020-01-04' as timestamp))")
+ f(reportingCatalog)
+ } finally {
+ reportingCatalog.reportedKeys = Seq.empty
+ reportingCatalog.reportedOrdering = Seq.empty
+ reportingCatalog.clearTables()
+ }
+ }
+ }
+
+ private def customReportingJoinDf(): DataFrame = sql(
+ s"""SELECT t1.data, t2.data FROM $customReportingTableName t1
+ |JOIN testcat.ns.$table t2 ON t1.id = t2.id
+ |""".stripMargin)
+
+ private def customReportingJoin(): (SparkPlan, BatchScanExec, Seq[String]) =
{
+ val df = customReportingJoinDf()
+ val logAppender = new LogAppender("custom reported partitioning or
ordering")
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ checkAnswer(df, Seq(Row("aa", "cc"), Row("bb", "dd")))
+ }
+ val warnings =
logAppender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ .filter(_.contains("columns cannot be resolved")).toSeq
+ val plan = df.queryExecution.executedPlan
+ val reportingScans =
collectScans(plan).filter(_.scan.isInstanceOf[CustomReportingScan])
+ assert(reportingScans.length == 1)
+ (plan, reportingScans.head, warnings)
+ }
+
+ test("SPARK-59721: a reported partitioning with an unresolvable key is
ignored") {
+ // CustomReportingCatalog is not a FunctionCatalog, so a `bucket` key
would be dropped even with
+ // a resolvable column. The warning assertion shows the unresolvable
column is what drops it.
+ val cases = Seq[(Seq[Expression], Option[String])](
+ (Seq(identity("id")), None),
+ (Seq(identity("ID")), None),
+ (Seq(bucket(4, "missing")), Some("missing")),
+ (Seq(nestedKey("missing")), Some("missing")),
+ (Seq(identity("id"), identity("missing")), Some("missing")),
+ (Seq(identity("missing"), bucket(4, "missing")), Some("missing")),
+ (Seq(identity("missing"), bucket(4, "absent"), identity("missing")),
+ Some("missing, absent")),
+ (Seq(plusOneKey("id"), identity("missing")), Some("missing")),
+ (Seq(plusOneKey("missing")), Some("missing")),
+ (Seq(identity("s.missing")), Some("s.missing (FIELD_NOT_FOUND)")),
+ (Seq(ChildrenOverridingTransform("identity",
Seq(FieldReference("missing")), Seq.empty)),
+ Some("missing")),
+ (Seq(ChildrenOverridingTransform("identity", Seq(FieldReference("id")),
+ Seq(FieldReference("missing")))), None))
+ withCustomReportingTable { reportingCatalog =>
+ cases.foreach { case (keys, unresolvedColumns) =>
+ reportingCatalog.reportedKeys = keys
+ val (plan, reportingScan, warnings) = customReportingJoin()
+ val keysString = keys.map(_.describe()).mkString(", ")
+ unresolvedColumns match {
+ case Some(columnsString) =>
+ assert(warnings.toSet ==
+ Set(ignoredReportWarning("KeyGroupedPartitioning", "partition
key", columnsString)),
+ keysString)
+ assert(reportingScan.keyGroupedPartitioning.isEmpty, keysString)
+ assert(collectShuffles(plan).nonEmpty, keysString)
+ case None =>
+ assert(warnings.isEmpty, warnings)
+ assert(reportingScan.keyGroupedPartitioning.isDefined, keysString)
+ assert(collectShuffles(plan).isEmpty, keysString)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59721: a reported ordering with an unresolvable column is
ignored") {
+ withCustomReportingTable { reportingCatalog =>
+ withSQLConf(SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key ->
"true") {
+ reportingCatalog.reportedKeys = Seq(identity("id"))
+ Seq[(SortOrder, String)](
+ (sort(FieldReference("id"), SortDirection.ASCENDING), "id"),
+ (sort(FieldReference("ID"), SortDirection.ASCENDING), "ID"),
+ // `index` is a metadata column of InMemoryBaseTable.
+ (sort(FieldReference("index"), SortDirection.ASCENDING), "index"),
+ (ChildrenOverridingSortOrder(FieldReference("id"),
Seq(FieldReference("missing"))), "id")
+ ).foreach { case (order, column) =>
+ reportingCatalog.reportedOrdering = Seq(order)
+ val (_, keptScan, keptWarnings) = customReportingJoin()
+ assert(keptWarnings.isEmpty, keptWarnings)
+
assert(keptScan.ordering.map(_.map(_.child.references.map(_.name).toSeq)) ==
+ Some(Seq(Seq(column))), order.describe())
+ }
+
+ Seq[(Seq[SortOrder], String)](
+ (Seq(sort(FieldReference("missing"), SortDirection.ASCENDING)),
"missing"),
+ (Seq(ChildrenOverridingSortOrder(FieldReference("missing"),
Seq.empty)), "missing"),
+ (Seq(sort(FieldReference(Seq.empty[String]),
SortDirection.ASCENDING)), "<empty>"),
+ (Seq(
+ sort(FieldReference("id"), SortDirection.ASCENDING),
+ sort(FieldReference("missing"), SortDirection.ASCENDING)),
"missing")
+ ).foreach { case (ordering, columnsString) =>
+ reportingCatalog.reportedOrdering = ordering
+ val (_, reportingScan, warnings) = customReportingJoin()
+ assert(warnings.toSet ==
+ Set(ignoredReportWarning("ordering", "ordering", columnsString)),
columnsString)
+ assert(reportingScan.ordering.isEmpty)
+ assert(reportingScan.keyGroupedPartitioning.isDefined)
+ val derivedOrdering = reportingScan.outputOrdering.map { o =>
+ (o.child.references.map(_.name).toSeq, o.direction)
+ }
+ assert(derivedOrdering == Seq((Seq("id"), Ascending)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-59721: a reported partition key that cannot be converted still
fails") {
+ val cases = Seq(
+ (plusOneKey("id"), "id + 1"),
+ (nestedKey("id"), "g(id)"))
+ withCustomReportingTable { reportingCatalog =>
+ cases.foreach { case (key, expr) =>
+ reportingCatalog.reportedKeys = Seq(key)
+ checkError(
+ exception = intercept[AnalysisException] {
+ customReportingJoinDf().collect()
+ },
+ condition = "_LEGACY_ERROR_TEMP_3054",
+ parameters = Map("expr" -> expr))
+ }
+ }
+ }
+
test("SPARK-47094: SPJ: Support compatible buckets") {
val table1 = "tab1e1"
val table2 = "table2"
@@ -9761,3 +9922,75 @@ class KeyGroupedPartitioningCatalystRuntimeFilterSuite
catalog.clearTables()
}
}
+
+/**
+ * An in-memory catalog whose scans report the partitioning and ordering set
on the catalog instead
+ * of the table's own, so a test can report keys on columns the table does not
have.
+ */
+class CustomReportingCatalog extends InMemoryTableCatalog {
+ var reportedKeys: Seq[Expression] = Seq.empty
+ var reportedOrdering: Seq[SortOrder] = Seq.empty
+
+ // scalastyle:off argcount
+ override protected def newInMemoryTable(
+ name: String,
+ columns: Array[Column],
+ partitioning: Array[Transform],
+ properties: java.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 =
+ CustomReportingScan(super.build(), reportedKeys, reportedOrdering)
+ }
+ }
+ }
+}
+
+case class CustomReportingScan(
+ inner: Scan,
+ keys: Seq[Expression],
+ ordering: Seq[SortOrder]) extends SupportsReportPartitioning with
SupportsReportOrdering {
+ override def readSchema(): StructType = inner.readSchema()
+ override def toBatch: Batch = inner.toBatch
+ override def description(): String = inner.description()
+ override def outputPartitioning(): Partitioning = new KeyGroupedPartitioning(
+ keys.toArray,
+
inner.asInstanceOf[SupportsReportPartitioning].outputPartitioning().numPartitions())
+ override def outputOrdering(): Array[SortOrder] = ordering.toArray
+}
+
+/**
+ * A connector-defined transform whose `children()` differs from its
`arguments()`, which the
+ * default `Transform.children()` returns.
+ */
+case class ChildrenOverridingTransform(
+ name: String,
+ args: Seq[Expression],
+ childExprs: Seq[Expression]) extends Transform {
+ override def arguments(): Array[Expression] = args.toArray
+ override def children(): Array[Expression] = childExprs.toArray
+}
+
+/**
+ * A connector-defined sort order whose `children()` differs from
`{expression()}`, which the
+ * default `SortOrder.children()` returns.
+ */
+case class ChildrenOverridingSortOrder(
+ expression: Expression,
+ childExprs: Seq[Expression]) extends SortOrder {
+ override def direction(): SortDirection = SortDirection.ASCENDING
+ override def nullOrdering(): NullOrdering = NullOrdering.NULLS_FIRST
+ override def children(): Array[Expression] = childExprs.toArray
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
index b8858ce33ce7..af50a1520e44 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
@@ -3027,10 +3027,12 @@ class MergeSubplansSuite extends PlanTest {
}
test("SPARK-40259: merge DSv2 scans that report empty key-grouped
partitioning or ordering") {
- // A source implementing SupportsReportPartitioning/SupportsReportOrdering
but reporting nothing
- // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the
field unconditionally). An
- // empty report carries no partitioning/ordering to drop, so the merge
should still proceed --
- // the gate tests the inner Seq, not the Option. The fused plan is the
plain column union.
+ // A source implementing SupportsReportOrdering but reporting no ordering
yields Some(Nil), not
+ // None (V2ScanPartitioningAndOrdering sets the ordering unless a reported
column cannot be
+ // resolved). The partitioning pass turns an empty report into None
instead, so the rule never
+ // produces the keyGroupedPartitioning = Some(Nil) case below. An empty
report carries no
+ // partitioning/ordering to drop, so the merge should still proceed -- the
gate tests the inner
+ // Seq, not the Option. The fused plan is the plain column union.
def assertMerges(withEmptyField: DataSourceV2ScanRelation =>
DataSourceV2ScanRelation): Unit = {
val sub1 =
ScalarSubquery(withEmptyField(v2ScanReading("a")).groupBy()(sum($"a").as("sum_a")))
val sub2 =
ScalarSubquery(withEmptyField(v2ScanReading("b")).groupBy()(sum($"b").as("sum_b")))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
