dongjoon-hyun commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4146972849
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -135,7 +135,8 @@ trait DataSourceV2ScanExecBase
/**
* Returns the output ordering for this scan. When the source reports
ordering via
- * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise,
when the output
+ * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it,
that ordering is
Review Comment:
nit. The same condition is described in the doc of
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` in `SQLConf` and
in `docs/sql-performance-tuning.md`, e.g., `... reports a KeyedPartitioning but
does not report explicit ordering via SupportsReportOrdering`. After this PR, a
source whose reported ordering is ignored also gets the key-derived ordering,
as the new test asserts. They are incomplete rather than wrong, so this is
optional. If you update them, could you update both? For example, `..., or
reports one that Spark ignores because it references a column that cannot be
resolved`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +89,43 @@ 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, _) =>
Review Comment:
FYI, this is also pre-existing and not for this PR. Unlike the partitioning
pass (`if d.keyGroupedPartitioning.isEmpty`), the ordering pass has no guard
and recomputes the ordering on every application. For a resolvable nested sort
key like `s.x`, `plan.resolve` returns a new `Alias(GetStructField(...))` with
a fresh `ExprId` each time, so the rule is not idempotent for such a report.
It looks like the `Once` batch idempotence check of `RuleExecutor` would
fail in testing mode. In production, the reported `SortOrder(Alias(...))` would
not match a required ordering because the `Alias` is not stripped. No existing
test reports a nested sort key, but this PR adds the struct column `s` and
`reportedOrdering` to the fixture, so it is easy to hit by adding such a case.
This is from reading the code, not from a run.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +89,43 @@ 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, _) =>
- 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(", "))}.")
+ // With no reported ordering, DataSourceV2ScanExecBase.outputOrdering
may still derive one
+ // from a kept key-grouped partitioning, which does not depend on the
dropped report.
+ d.copy(ordering = None)
+ } else {
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ // This is not `V2Expression.references()` because `ApplyTransform` and
`BucketTransform`
+ // override it to return only their top-level arguments, which would miss
`f(g(missing))`.
Review Comment:
nit. The wording which I suggested earlier is imprecise. `BucketTransform`
takes `Seq[NamedReference]`, so its override cannot miss anything. Only
`ApplyTransform` truncates. The more durable reason is that `references()` can
be overridden by connectors (e.g., `MyIdentityTransform.references()` in
`DataSourceV2Suite` returns an empty array), while the conversion resolves
structurally through `arguments()` and `expression()`. Could you revise it? For
example,
```scala
// Walk `children()` instead of calling `V2Expression.references()`: the
conversion resolves
// every reference reachable through `arguments()`/`expression()`, while
`references()` can
// return fewer. `ApplyTransform` returns only its top-level arguments,
which would miss
// `f(g(missing))`, and a connector-defined `Transform` can return
anything.
```
In addition, the comment in the `catch` below and the scaladoc of
`resolveRefOpt` mention only a missing nested field and an ambiguous reference,
but `AMBIGUOUS_REFERENCE_TO_FIELDS` and `INVALID_EXTRACT_BASE_FIELD_TYPE`
(e.g., `id.x` on an INT column) reach this `catch` too.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3486,6 +3492,139 @@ class KeyGroupedPartitioningSuite
}
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+ private val customReportingTableName =
s"$customReportingCatalogName.ns.$table"
+
+ private def withCustomReportingTable(f: CustomReportingCatalog => Unit):
Unit = {
+ spark.conf.set(s"spark.sql.catalog.$customReportingCatalogName",
+ classOf[CustomReportingCatalog].getName)
+ 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 $customReportingCatalogName.ns.$table 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))")
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ f(reportingCatalog)
+ }
+ } finally {
+ reportingCatalog.reportedKeys = Seq.empty
+ reportingCatalog.reportedOrdering = Seq.empty
+ reportingCatalog.clearTables()
+ spark.conf.unset(s"spark.sql.catalog.$customReportingCatalogName")
+ }
+ }
+
+ private def customReportingJoinDf(): DataFrame = sql(
+ s"""SELECT t1.data, t2.data FROM $customReportingCatalogName.ns.$table t1
+ |JOIN testcat.ns.$table t2 ON t1.id = t2.id
+ |""".stripMargin)
+
+ private def customReportingJoin(warningText: String): (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(warningText)).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: an unresolvable reported partition key falls back to
unknown partitioning") {
+ val nestedTransform =
+ ApplyTransform("f", Seq(ApplyTransform("g",
Seq(FieldReference("missing")))))
+ // 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(bucket(4, "missing")), Some("missing")),
+ (Seq(nestedTransform), Some("missing")),
+ (Seq(identity("id"), identity("missing")), Some("missing")),
+ (Seq(identity("missing"), bucket(4, "missing")), Some("missing")),
+ (Seq(identity("s.missing")), Some("s.missing (FIELD_NOT_FOUND)")))
+ withCustomReportingTable { reportingCatalog =>
+ cases.foreach { case (keys, unresolvedColumns) =>
+ reportingCatalog.reportedKeys = keys
+ val (plan, reportingScan, warnings) =
+ customReportingJoin("partition key columns cannot be resolved")
+ val keysString = keys.map(_.describe()).mkString(", ")
+ unresolvedColumns match {
+ case Some(columnsString) =>
+ assert(warnings.nonEmpty, s"no warning for keys [$keysString]")
+ assert(warnings.forall(_.contains(s"reported by
$customReportingTableName ")), warnings)
+
assert(warnings.forall(_.contains(classOf[CustomReportingScan].getName)),
warnings)
+ assert(warnings.forall(_.contains(s"resolved: $columnsString.")),
warnings)
+ 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: an unresolvable reported ordering column is ignored") {
+ withCustomReportingTable { reportingCatalog =>
+ withSQLConf(SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key ->
"true") {
+ reportingCatalog.reportedKeys = Seq(identity("id"))
+ reportingCatalog.reportedOrdering = Seq(
+ sort(FieldReference("id"), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST))
+ val (_, keptScan, keptWarnings) =
+ customReportingJoin("ordering columns cannot be resolved")
+ assert(keptWarnings.isEmpty, keptWarnings)
+
assert(keptScan.ordering.map(_.map(_.child.references.map(_.name).toSeq)) ==
+ Some(Seq(Seq("id"))))
+
+ reportingCatalog.reportedOrdering = Seq(
+ sort(FieldReference("missing"), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST))
+ val (_, reportingScan, warnings) =
+ customReportingJoin("ordering columns cannot be resolved")
+ assert(warnings.nonEmpty)
+ assert(warnings.forall(_.contains(s"reported by
$customReportingTableName ")), warnings)
+
assert(warnings.forall(_.contains(classOf[CustomReportingScan].getName)),
warnings)
+ assert(warnings.forall(_.contains("resolved: missing.")), warnings)
+ 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") {
Review Comment:
This pins `id + 1` and `f(g(id))`, where every column resolves. Note that
the pre-check runs over the whole report before the conversion, so a report
which mixes an unconvertible key with an unresolvable column (e.g., `[id + 1,
identity(missing)]`, or a single key `missing + 1`) does not fail anymore. It
succeeds with a warning which mentions only `missing`, and it starts to fail
with `_LEGACY_ERROR_TEMP_3054` once the column becomes resolvable.
It is consistent with the PR description (`Keys whose columns all resolve
take the existing conversion path unchanged`), so no code change is needed.
Could you add one mixed case to the fallback test above, e.g., `(Seq(<id + 1>,
identity("missing")), Some("missing"))`, so that this boundary is explicit?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3486,6 +3492,139 @@ class KeyGroupedPartitioningSuite
}
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+ private val customReportingTableName =
s"$customReportingCatalogName.ns.$table"
+
+ private def withCustomReportingTable(f: CustomReportingCatalog => Unit):
Unit = {
+ spark.conf.set(s"spark.sql.catalog.$customReportingCatalogName",
+ classOf[CustomReportingCatalog].getName)
+ 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 $customReportingCatalogName.ns.$table 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))")
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ f(reportingCatalog)
+ }
+ } finally {
+ reportingCatalog.reportedKeys = Seq.empty
+ reportingCatalog.reportedOrdering = Seq.empty
+ reportingCatalog.clearTables()
+ spark.conf.unset(s"spark.sql.catalog.$customReportingCatalogName")
+ }
+ }
+
+ private def customReportingJoinDf(): DataFrame = sql(
+ s"""SELECT t1.data, t2.data FROM $customReportingCatalogName.ns.$table t1
+ |JOIN testcat.ns.$table t2 ON t1.id = t2.id
+ |""".stripMargin)
+
+ private def customReportingJoin(warningText: String): (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(warningText)).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: an unresolvable reported partition key falls back to
unknown partitioning") {
+ val nestedTransform =
+ ApplyTransform("f", Seq(ApplyTransform("g",
Seq(FieldReference("missing")))))
+ // 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])](
Review Comment:
All the new cases use the built-in expression classes (`FieldReference`,
`ApplyTransform`, `BucketTransform`, `SortValue`), so the unresolvable path is
not covered for connector-defined `Transform`/`SortOrder`/`NamedReference`
classes. That is where walking `children()` matters most. For example,
`MyIdentityTransform.references()` in `DataSourceV2Suite` returns an empty
array.
Could you add one case which reads the existing
`OrderAndPartitionAwareDataSource` (and its Java twin,
`JavaOrderAndPartitionAwareDataSource`) with `partitionKeys=i` and
`orderKeys=missing`, and verify that the query succeeds, the reported ordering
is dropped, and the warning names `missing`? It also covers my `describe()`
comment in `V2ScanPartitioningAndOrdering`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -46,14 +46,20 @@ 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 missing nested field or an ambiguous reference.
+ */
+ private[sql] def resolveRefOpt[T <: NamedExpression](
Review Comment:
nit. The type parameter looks unnecessary now. All callers other than
`resolveRef` pass `[NamedExpression]`, and none of them uses the typed result.
How about making this non-generic and keeping the cast in `resolveRef` like
before?
```scala
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 = {
resolveRefOpt(ref, plan).getOrElse {
...
}.asInstanceOf[T]
}
```
Then, the explicit `[NamedExpression]` at the call sites and the
`NamedExpression` import in `V2ScanPartitioningAndOrdering` can be removed.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9259,3 +9398,51 @@ 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 {
+ @volatile var reportedKeys: Seq[Expression] = Seq.empty
Review Comment:
nit. `@volatile` looks unnecessary. These fields are written by the test
thread and read only in `build()`, which the optimizer invokes on the same
thread. The other in-memory test catalogs use plain `var` for such knobs.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +89,43 @@ 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, _) =>
- 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(", "))}.")
+ // With no reported ordering, DataSourceV2ScanExecBase.outputOrdering
may still derive one
+ // from a kept key-grouped partitioning, which does not depend on the
dropped report.
+ d.copy(ordering = None)
+ } else {
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ // This is not `V2Expression.references()` because `ApplyTransform` and
`BucketTransform`
+ // override it to return only their top-level arguments, which would miss
`f(g(missing))`.
+ def references(expr: V2Expression): Seq[NamedReference] = expr match {
+ case ref: NamedReference => Seq(ref)
+ case other => other.children().toImmutableArraySeq.flatMap(references)
+ }
+ exprs.flatMap(references).distinct.flatMap { ref =>
+ try {
+ if (V2ExpressionUtils.resolveRefOpt[NamedExpression](ref,
relation).isDefined) {
+ None
+ } else {
+ Some(ref.describe())
Review Comment:
`Expression.describe()` defaults to `toString()`, and `NamedReference`
requires only `fieldNames()`. So, for a connector-defined `NamedReference`
without `toString`/`equals`, this prints `Class@hash` instead of the column
name, and `.distinct` above does not de-duplicate it.
For example, when `OrderAndPartitionAwareDataSource` in `DataSourceV2Suite`
(`MySortOrder` -> `MyIdentityTransform` -> `MyNamedReference`) is read with
`orderKeys=missing`, the query failed with `Unable to resolve missing given
[...]` before this PR. Now, it succeeds, and the only diagnostic is a warning
which ends with `...OrderAndPartitionAwareDataSource$MyNamedReference@<hash>`.
(This is from reading the code, not from a run.)
Could you build the name from `fieldNames` like `resolveRef` does, and
de-duplicate on the string? For example,
```scala
exprs.flatMap(references).flatMap { ref =>
val name = ref.fieldNames.toImmutableArraySeq.quoted
try {
if (V2ExpressionUtils.resolveRefOpt[NamedExpression](ref,
relation).isDefined) {
None
} else {
Some(name)
}
} catch {
// A missing nested field or an ambiguous name throws instead of
returning None.
case e: AnalysisException => Some(s"$name (${e.getCondition})")
}
}.distinct
```
`quoted` comes from `CatalogV2Implicits.MultipartIdentifierHelper`.
`FieldReference.toString` is `parts.quoted`, so the expected strings in the new
tests (`missing` and `s.missing (FIELD_NOT_FOUND)`) stay the same.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -47,9 +49,21 @@ 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()
+ val unresolvedColumns =
unresolvableColumns(keys.toImmutableArraySeq, 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(", "))}. " +
+ log"Storage-partitioned join will not be applied for this
scan.")
Review Comment:
nit. This sentence is narrower than what happens. Without the reported
partitioning, storage-partitioned aggregates and windows and the
partition-key-derived ordering are lost too (cf. `storage-partitioned
operations (joins, aggregates, and windows)` in `SQLConf`). In addition, when
`spark.sql.sources.v2.bucketing.shuffle.enabled` is true and the other side is
keyed, a storage-partitioned join can still be planned by shuffling only this
side. Could you remove this sentence like the ordering warning below, or
generalize it?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -135,7 +135,8 @@ trait DataSourceV2ScanExecBase
/**
* Returns the output ordering for this scan. When the source reports
ordering via
- * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise,
when the output
+ * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it,
that ordering is
+ * returned as-is. Otherwise, when the output
Review Comment:
FYI, this is pre-existing and not for this PR. `returned as-is` also applies
to a sort order on a column which is pruned out of the scan output, because the
ordering pass resolves against the full `relation`.
`DataSourceV2ScanRelation.doCanonicalize` handles it with
`takeWhile(_.references.subsetOf(outputSet))`, but `outputOrdering` does not.
A row-based scan is safe because the `ProjectExec` above it truncates such a
sort order. For a columnar scan without `ProjectExec`, it looks like the k-way
merge of `GroupPartitionsExec` (`new LazyCodeGenOrdering(kWayMergeOrdering,
child.output)`) can fail to bind it when
`spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled` is true
(false by default). This is from reading the code, and I have not reproduced it
yet. I am checking it separately.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2927,9 +2927,10 @@ 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.
+ // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the
field unless a reported
Review Comment:
nit. Since this comment is updated, could you also update the sibling
comment above `mergeDegradesReporting` in `PlanMerger.scala`? It says
`partitioning pass keeps the report whenever any key survives`, which is no
longer literally true. For example, `[identity(id), identity(missing)]` drops
the whole report although `id` survives in the scan output (the new test pins
it). The merge logic and the conclusion of that comment still hold because the
new check depends only on the reported columns and `relation`, so adding
`unless a reported column cannot be resolved` there would be enough.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3486,6 +3492,139 @@ class KeyGroupedPartitioningSuite
}
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+ private val customReportingTableName =
s"$customReportingCatalogName.ns.$table"
+
+ private def withCustomReportingTable(f: CustomReportingCatalog => Unit):
Unit = {
+ spark.conf.set(s"spark.sql.catalog.$customReportingCatalogName",
+ classOf[CustomReportingCatalog].getName)
+ 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 $customReportingCatalogName.ns.$table VALUES " +
Review Comment:
nit. `customReportingTableName` is defined above, but this line and the
`SELECT` in `customReportingJoinDf` assemble the same string again. Could you
use the constant?
--
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]
