dongjoon-hyun commented on code in PR #59257:
URL: https://github.com/apache/spark/pull/59257#discussion_r4294729920
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -139,38 +139,58 @@ trait DataSourceV2ScanExecBase
* reference columns pruned out of the scan output, while a consumer binds
it against the
* 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.
+ * orders on a partition key when the output partitioning is a
`KeyedPartitioning`. Each
+ * partition holds a single key, so those still hold.
*
- * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
- * In a reported ordering it also ends the leading run, like a sort order
over a pruned column.
- * Dropping it loses nothing today when Spark can call the transform's
function, since no
- * operator then requires an ordering over the transform. The write path
sorts by the function
- * call instead. Keeping such a sort order would add comparisons nobody
uses. Dropping it also
- * handles a transform Spark cannot evaluate. Revisit this if an ordering
over a transform
- * becomes a real requirement.
+ * If nothing of the reported ordering is kept, 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. Nothing is kept, for
example, when the
+ * source reports no ordering, one Spark ignores, or an empty one.
Review Comment:
Minor: This is the rule a connector author has to know, but the public
Javadoc of `SupportsReportOrdering.outputOrdering()`
(SupportsReportOrdering.java:36-47) does not mention it. That Javadoc already
describes how Spark treats an unresolvable report, a transform, and a sort
order on a partition key, so it reads as the full contract. After this PR, an
empty array, or an ordering Spark keeps nothing of, makes a scan with a
`KeyGroupedPartitioning` report `[key ASC]` and drop a sort above it. A
connector that returns an empty array to mean "unsorted" learns this only from
the 4.4 migration note. Could we add a sentence there, for example that when
Spark keeps none of the reported sort orders, including for an empty array, it
may treat each partition as sorted by the partition keys that are not
transforms (see `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`)?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -139,38 +139,58 @@ trait DataSourceV2ScanExecBase
* reference columns pruned out of the scan output, while a consumer binds
it against the
* 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.
+ * orders on a partition key when the output partitioning is a
`KeyedPartitioning`. Each
+ * partition holds a single key, so those still hold.
*
- * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
- * In a reported ordering it also ends the leading run, like a sort order
over a pruned column.
- * Dropping it loses nothing today when Spark can call the transform's
function, since no
- * operator then requires an ordering over the transform. The write path
sorts by the function
- * call instead. Keeping such a sort order would add comparisons nobody
uses. Dropping it also
- * handles a transform Spark cannot evaluate. Revisit this if an ordering
over a transform
- * becomes a real requirement.
+ * If nothing of the reported ordering is kept, 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. Nothing is kept, for
example, when the
+ * source reports no ordering, one Spark ignores, or an empty one.
+ *
+ * A kept sort order wins over the derived ordering, since ordering is
prefix-based. Putting the
+ * keys in front would stop a kept `[data]` from satisfying a required
`[data]`. So the result
+ * can depend on the columns a query reads. A report `[s]` derives the key
ordering only when `s`
+ * is pruned.
+ *
+ * In both the reported and the derived ordering, a sort order that holds a
partition transform
+ * is dropped, even on a partition key. In a reported ordering it also ends
the leading run, like
+ * a sort order over a pruned column. Dropping it loses nothing today when
Spark can call the
+ * transform's function, since no operator then requires an ordering over
the transform. The
+ * write path sorts by the function call instead. Keeping such a sort order
would add
+ * comparisons nobody uses. Dropping it also handles a transform Spark
cannot evaluate. Revisit
+ * this if an ordering over a transform becomes a real requirement.
*/
- 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 =>
- k.expressions.filterNot(holdsTransform).map(SortOrder(_, Ascending))
- case _ => Seq.empty
+ override def outputOrdering: Seq[SortOrder] =
+ if (keptOrdering.isEmpty && derivedOrdering.nonEmpty &&
Review Comment:
Nit: `derivedOrdering.nonEmpty` does not change the result. When
`keptOrdering` and `derivedOrdering` are both empty, both branches return
`Nil`. Could we drop it, e.g. `if (keptOrdering.isEmpty &&
conf.v2BucketingPartitionKeyOrderingEnabled) derivedOrdering else keptOrdering`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2680,11 +2680,13 @@ object SQLConf {
val V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED =
buildConf("spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled")
.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 " +
+ "a V2 data source that reports a KeyedPartitioning and no ordering
that Spark keeps for " +
+ "the scan output. For example, the source reports none, an empty one,
or one that " +
+ "starts with a sort order on a column pruned from the scan output or
on a partition " +
+ "transform and has no later sort order on a partition key. Within a
single partition " +
Review Comment:
Minor: "has no later sort order on a partition key" is broader than the
code. `keptOrdering` keeps a later sort order only on a key that is not a
transform. So with keys `[years(ts), identity(id)]`, `s` pruned and a report
`[s, years(ts)]`, the scan still derives `[id]`, although `years(ts)` is a
later sort order on a partition key. The PR description says "a partition key
other than a transform". The same clause is in `sql-performance-tuning.md` and
the 4.4 migration note. Could the three docs say "no later sort order on a
partition key other than a transform"?
--
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]
