dongjoon-hyun commented on code in PR #59257:
URL: https://github.com/apache/spark/pull/59257#discussion_r4283290756
##########
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:
`order.references.subsetOf(outputSet)` checks only the top-level attribute.
Nested schema pruning keeps the exprId of a struct column but narrows its type.
The reported ordering, and the partition keys, are converted against the
relation's full schema.
For example, with `s struct<x, y>`, a report `[s.y]` (or a key on `s.y`),
and a query that reads only `s.x`:
- the scan output has `s: struct<x>`;
- `s.y` is `GetStructField(s, 1)`, and its references `{s}` are a subset of
the output, so it is kept, or derived as the key ordering;
- a consumer that binds it against the scan output, such as the k-way merge
of `GroupPartitionsExec`, then reads ordinal 1 of a one-field struct.
I haven't reproduced this, and the check predates this PR (SPARK-59899). But
this PR derives the key ordering in more cases, so more scans can reach it.
Could we check whether a sort order on a pruned nested field survives here? If
it does, it could be treated like one on a pruned column.
##########
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:
nit: This clause now lists report shapes: none, an ignored one, and one that
starts with a pruned column or a transform and has no later sort order on a
partition key. The code's rule is shorter: derive when no sort order of the
report is kept once it is restricted to the scan output.
The same clause is copied into `sql-performance-tuning.md` and the 4.4
migration note, and SPARK-59899 and SPARK-59995 already needed edits to it.
Could the three docs state the rule instead? For example:
```
... a V2 data source that reports a KeyedPartitioning and no ordering that
Spark keeps, for
example because it reports none via SupportsReportOrdering, or reports one
that references
columns pruned from the scan output or partition transforms.
```
##########
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:
`None` and `Some(Nil)` mean the same to `outputOrdering`, but
`DataSourceV2ScanRelation.doCanonicalize` keeps them apart. It also restricts
`ordering` with a plain `takeWhile(_.references.subsetOf(outputSet))`. So,
unlike `outputOrdering`, it does not drop the sort orders over a transform, and
it does not keep the sort orders on a partition key past a pruned column.
As a result, the canonical form and the physical ordering follow different
rules. With keys `[identity(id), identity(data)]` and `s` pruned, the reports
`[s]` and `[s, id]` both canonicalize to `Some(Nil)`, while `outputOrdering`
gives `[id, data]` and `[id]`.
Could the comment say this holds for `outputOrdering` only? Or could
`doCanonicalize` share the restriction with `outputOrdering`?
##########
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:
nit: `outputOrdering` is a `def`, so every call re-runs the `span`, builds
an `ExpressionSet`, and matches `outputPartitioning` twice. It is called often:
by `EnsureRequirements`, by each ancestor's `outputOrdering`, and by
`GroupPartitionsExec` (`kWayMergeIsFeasible`, `kWayMergeOrdering` and the
coalescing filter).
`reported` depends only on `ordering`, `outputSet` and `outputPartitioning`,
which do not change for the instance. Could it be a `@transient lazy val`, like
`outputPartitioning` above? Then only the conf check runs on each call.
##########
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:
nit: After the new paragraph, "Either way" follows the bullet list on the
derivation, so it reads as if it refers to those three cases. It means both the
restricted report and the derived key ordering (`filterNot(holdsTransform)`).
Could it say so, e.g. "In both the reported and the derived ordering, a sort
order that holds a partition transform is dropped, ..."?
--
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]
