peter-toth commented on code in PR #59257:
URL: https://github.com/apache/spark/pull/59257#discussion_r4294321705
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -155,19 +161,21 @@ trait DataSourceV2ScanExecBase
*/
override def outputOrdering: Seq[SortOrder] = {
def holdsTransform(e: Expression): Boolean =
e.exists(_.isInstanceOf[TransformExpression])
- (ordering, outputPartitioning) match {
- case (Some(o), p) =>
- val (prefix, rest) =
- o.span(order => order.references.subsetOf(outputSet) &&
!holdsTransform(order.child))
- p match {
- case k: KeyedPartitioning if rest.nonEmpty =>
- val keyExprs =
ExpressionSet(k.expressions.filterNot(holdsTransform))
- prefix ++ rest.filter(order => keyExprs.contains(order.child))
- case _ => prefix
- }
- case (_, k: KeyedPartitioning) if
conf.v2BucketingPartitionKeyOrderingEnabled =>
+ val reported = ordering.map { o =>
+ val (prefix, rest) =
+ o.span(order => order.references.subsetOf(outputSet) &&
!holdsTransform(order.child))
Review Comment:
The derivation can't add such a sort order today. A nested partition key
reaches the scan as an `Alias(GetStructField(...))`.
`KeyedPartitioning.supportsExpressions` refuses it, so the scan reports no
`KeyedPartitioning` (SPARK-60041). A reported `[s.y]` is kept, so nothing is
derived for it. I measured that shape during #59199. The `ProjectExec` that
nested pruning leaves above the scan drops the stale sort order, so the merge
never reads it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -95,9 +95,8 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
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.
+ // `None` and `Some(Nil)` mean the same to
DataSourceV2ScanExecBase.outputOrdering, which
Review Comment:
Done in
[b82b42717c7](https://github.com/apache/spark/pull/59257/commits/b82b42717c7e56382c8cd0892e55f03e5667b9e0).
The comment now says that the two restrict the ordering by different rules. A
plain `takeWhile` is enough for `doCanonicalize`, which only feeds plan
equality. Sharing the rule is part of SPARK-60083.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -155,19 +161,21 @@ trait DataSourceV2ScanExecBase
*/
override def outputOrdering: Seq[SortOrder] = {
def holdsTransform(e: Expression): Boolean =
e.exists(_.isInstanceOf[TransformExpression])
- (ordering, outputPartitioning) match {
- case (Some(o), p) =>
- val (prefix, rest) =
- o.span(order => order.references.subsetOf(outputSet) &&
!holdsTransform(order.child))
- p match {
- case k: KeyedPartitioning if rest.nonEmpty =>
- val keyExprs =
ExpressionSet(k.expressions.filterNot(holdsTransform))
- prefix ++ rest.filter(order => keyExprs.contains(order.child))
- case _ => prefix
- }
- case (_, k: KeyedPartitioning) if
conf.v2BucketingPartitionKeyOrderingEnabled =>
+ val reported = ordering.map { o =>
Review Comment:
Done in
[b82b42717c7](https://github.com/apache/spark/pull/59257/commits/b82b42717c7e56382c8cd0892e55f03e5667b9e0).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2654,9 +2654,12 @@ object SQLConf {
.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, 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 " +
- "value, so the data is trivially sorted by those expressions.
Partition transforms such " +
- "as `days(ts)` or `bucket(8, id)` are left out of the ordering.
Requires " +
+ "column that cannot be resolved, or one that starts with a sort order
on a column " +
Review Comment:
Done in
[b82b42717c7](https://github.com/apache/spark/pull/59257/commits/b82b42717c7e56382c8cd0892e55f03e5667b9e0),
in all three docs.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -140,10 +140,16 @@ trait DataSourceV2ScanExecBase
* 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.
+ * partition holds a single key, so those still hold.
+ *
+ * If no sort order is left, the ordering is derived from the partition keys
instead, 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. No sort order is left
when the source:
+ * - reports none, or reports one Spark ignores (see
`V2ScanPartitioningAndOrdering`);
+ * - reports an empty one;
+ * - reports only sort orders that are dropped.
*
* Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
Review Comment:
Done in
[b82b42717c7](https://github.com/apache/spark/pull/59257/commits/b82b42717c7e56382c8cd0892e55f03e5667b9e0).
--
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]
