dongjoon-hyun commented on code in PR #59163:
URL: https://github.com/apache/spark/pull/59163#discussion_r4148636319
##########
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:
Thank you, @viirya. You're right. I missed that the non-merge coalescing
path filters the key sort orders instead of taking a prefix. I applied your
suggestion, using `span` so that nothing extra is built when the whole ordering
is over the output.
##########
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:
Added `prune_order_t3`, which reports `(ts, id)`, and a check of the
non-merge `GroupPartitionsExec(scan).outputOrdering` for all three tables. The
new table fails without the change: `ArraySeq() did not equal List(id#63 ASC
NULLS FIRST)`.
##########
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:
Added a note to the `ordering` pass of `V2ScanPartitioningAndOrdering`.
--
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]
