viirya commented on code in PR #59163:
URL: https://github.com/apache/spark/pull/59163#discussion_r4148525848
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -135,15 +135,18 @@ 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
- * partitioning is a `KeyedPartitioning` and
+ * `SupportsReportOrdering`, that ordering is returned up to its first sort
order that references
+ * a column pruned out of the scan output (see
V2ScanPartitioningAndOrdering). A consumer binds
+ * the ordering against the output, e.g. the k-way merge of
`GroupPartitionsExec`, and ordering
+ * is prefix-based, so the sort orders after a pruned one do not hold on
their own either.
+ * 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 {
- case (Some(o), _) => o
+ case (Some(o), _) => o.takeWhile(_.references.subsetOf(outputSet))
Review Comment:
This regresses the non-merge coalescing path for columnar scans under
default configs. `GroupPartitionsExec.outputOrdering` there *filters*
`child.outputOrdering` down to key sort orders rather than taking a prefix, so
a reported `(ts, id)` with `ts` pruned used to yield `(id)`. That is valid,
since `id` is constant within each input partition and within a coalesced one.
Now the scan reports `()` and that ordering is gone.
I checked this with two tables partitioned by `identity(id)`, each reporting
`(ts, id)`, running `SELECT id, data` with key 1 split in two. With a
`SortMergeJoinExec` built directly over the two `BatchScanExec`s (standing in
for columnar scans, as the new test does) and `EnsureRequirements` applied, the
plan went from no `SortExec` before this change to one `Sort [id ASC NULLS
FIRST]` on each side after. The crash being fixed needs the non-default
`preserveOrderingOnCoalesce`, while this extra sort happens with defaults.
Row-based scans already lose the ordering through `ProjectExec`, but that
doesn't make the columnar regression acceptable.
When the output partitioning is a `KeyedPartitioning`, each input partition
holds a single key, so a sort order over a surviving key expression still holds
after the truncated prefix:
```scala
case (Some(o), p) =>
val prefix = o.takeWhile(_.references.subsetOf(outputSet))
p match {
case k: KeyedPartitioning =>
val keyExprs = ExpressionSet(k.expressions)
prefix ++ o.drop(prefix.length).filter(so =>
keyExprs.contains(so.child))
case _ => prefix
}
```
With this, the scan above reports `(id)`: the SMJ needs no `SortExec`, and
the k-way merge still binds only output columns and runs.
`KeyGroupedPartitioningSuite` passes with it, including the new test. Could you
take something like this and add a test pinning the non-merge
`GroupPartitionsExec(scan).outputOrdering` for this case?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3486,6 +3486,46 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59899: a scan's reported ordering stops at the first sort order
on a pruned " +
+ "column") {
+ // The source reports its ordering over the full table, so a sort order on
a column pruned out
+ // of the scan output reaches the scan, as a pruned partition key does.
Ordering is
+ // prefix-based, so the scan keeps the leading run of sort orders over its
output: `t1` drops
+ // its last order, `t2` a middle one along with every order after it.
+ val table1 = "prune_order_t1"
+ val table2 = "prune_order_t2"
+ def asc(col: String): SortOrder =
+ sort(FieldReference(col), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST)
+ createTable(table1, columns, Array(identity("id")), Array(asc("id"),
asc("data"), asc("ts")))
+ createTable(table2, columns, Array(identity("id")), Array(asc("id"),
asc("ts"), asc("data")))
+ Seq(table1, table2).foreach { table =>
+ // Key 1 is stored in two splits, so grouping the scan by key merges
them.
+ sql(s"INSERT INTO testcat.ns.$table VALUES " +
+ "(1, 'bb', cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', cast('2020-01-02' as timestamp)), " +
+ "(2, 'cc', cast('2020-01-03' as timestamp))")
+ }
+
+ Seq(table1 -> 2, table2 -> 1).foreach { case (table, numKept) =>
+ val scan = collectScans(sql(s"SELECT id, data FROM testcat.ns.$table")
+ .queryExecution.executedPlan).head
+ val reported = scan.ordering.get
+ assert(reported.exists(!_.references.subsetOf(scan.outputSet)),
+ s"test setup: the ordering of $table is on `ts`, which is pruned out
of the scan")
+ assert(scan.outputOrdering === reported.take(numKept))
+
+ // A k-way merge binds the scan's ordering against the scan output. A
row-based scan like this
+ // one always gets a `ProjectExec`, which truncates the ordering, but a
columnar scan may get
+ // nothing between it and the merge that does. So the merge is built
right over the scan here.
+ val merged = GroupPartitionsExec(scan, enableSortedMerge =
true).execute()
Review Comment:
Building `GroupPartitionsExec` by hand over a row-based scan to stand in for
the columnar path is understandable, since the in-memory catalog has no
columnar keyed source, and the `SortedMergeCoalescedRDD` assertion keeps it
from passing vacuously. This is also a natural place to add the non-merge
`GroupPartitionsExec(scan).outputOrdering` check mentioned on
`DataSourceV2ScanExecBase`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -135,15 +135,18 @@ 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
- * partitioning is a `KeyedPartitioning` and
+ * `SupportsReportOrdering`, that ordering is returned up to its first sort
order that references
+ * a column pruned out of the scan output (see
V2ScanPartitioningAndOrdering). A consumer binds
Review Comment:
Nit: this points at `V2ScanPartitioningAndOrdering`, but its `ordering` pass
says nothing about pruned columns; only the partitioning pass does. A short
note there that the ordering is kept untruncated on purpose and truncated by
its consumers would make the pointer land.
--
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]
