peter-toth commented on code in PR #59257:
URL: https://github.com/apache/spark/pull/59257#discussion_r4294320299
##########
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))
+ outputPartitioning match {
+ case k: KeyedPartitioning if rest.nonEmpty =>
+ val keyExprs = ExpressionSet(k.expressions.filterNot(holdsTransform))
+ prefix ++ rest.filter(order => keyExprs.contains(order.child))
+ case _ => prefix
+ }
+ }.getOrElse(Seq.empty)
+ outputPartitioning match {
+ case k: KeyedPartitioning
+ if reported.isEmpty && conf.v2BucketingPartitionKeyOrderingEnabled =>
k.expressions.filterNot(holdsTransform).map(SortOrder(_, Ascending))
Review Comment:
Done in
[b82b42717c7](https://github.com/apache/spark/pull/59257/commits/b82b42717c7e56382c8cd0892e55f03e5667b9e0).
The 4.4 note now says that the derivation relies on each partition holding a
single partition key value, as `KeyGroupedPartitioning` requires. It already
says how to restore the previous behavior.
##########
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))
+ outputPartitioning match {
+ case k: KeyedPartitioning if rest.nonEmpty =>
+ val keyExprs = ExpressionSet(k.expressions.filterNot(holdsTransform))
+ prefix ++ rest.filter(order => keyExprs.contains(order.child))
+ case _ => prefix
+ }
+ }.getOrElse(Seq.empty)
+ outputPartitioning match {
+ case k: KeyedPartitioning
+ if reported.isEmpty && conf.v2BucketingPartitionKeyOrderingEnabled =>
Review Comment:
Right. A kept sort order wins over the derived one, since ordering is
prefix-based. So the result depends on the columns a query reads. The Scaladoc
now says so, in
[b82b42717c7](https://github.com/apache/spark/pull/59257/commits/b82b42717c7e56382c8cd0892e55f03e5667b9e0).
`mergeDegradesReporting` compares the reported orderings, so it misses the
`PlanMerger` case. On master the merge loses nothing there, since neither scan
gets `[id]`.
Putting the keys in front would break a kept `[data]` satisfying a required
`[data]`, as you note. The real fix is to let `orderingSatisfies` know that the
keys are constant within a partition. A constant sort order was tried in #52474
and reverted, see [this
comment](https://github.com/apache/spark/pull/52474#issuecomment-3383971418).
That direction was turned down then, so I wouldn't file a JIRA for it now.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala:
##########
@@ -315,7 +315,11 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
test("ordering and partitioning reporting") {
- withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "true") {
+ // The source breaks the `KeyGroupedPartitioning` contract, e.g. [1, 1, 3]
under the key 1.
+ // So the ordering derived from the keys is turned off. This test checks
the reported one.
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key -> "false") {
Review Comment:
Done in
[b82b42717c7](https://github.com/apache/spark/pull/59257/commits/b82b42717c7e56382c8cd0892e55f03e5667b9e0).
The derivation is now off only for the case with partitioning on `i` and an
empty reported ordering. The other cases run with it on.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -4078,6 +4078,41 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59981: a reported ordering that keeps no sort order falls back
to the keys") {
+ withCustomReportingTable { reportingCatalog =>
+ reportingCatalog.reportedKeys = Seq(identity("id"))
+ // An empty report, one on `s`, which the join's projection prunes from
the scan output, and
+ // one that starts with `s` and goes on with `data`, which is not a
partition key.
+ val sOrder = sort(FieldReference("s"), SortDirection.ASCENDING)
+ val dataOrder = sort(FieldReference("data"), SortDirection.ASCENDING)
+ Seq(Seq.empty, Seq(sOrder), Seq(sOrder, dataOrder)).foreach { ordering =>
Review Comment:
Done in
[b82b42717c7](https://github.com/apache/spark/pull/59257/commits/b82b42717c7e56382c8cd0892e55f03e5667b9e0).
A new test covers `[data]`, `[s, id]` and `[s, id DESC]`. The first test now
also checks that a coalescing `GroupPartitionsExec` keeps the derived `[id]`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -377,8 +377,8 @@ case class GroupPartitionsExec(
override def outputOrdering: Seq[SortOrder] = {
if (!hasCoalescing) {
// No coalescing: each output partition is exactly one input partition.
The child's
- // within-partition ordering is fully preserved (including any
key-derived ordering that
- // `DataSourceV2ScanExecBase` already prepended).
+ // within-partition ordering is fully preserved, including an ordering
that
+ // `DataSourceV2ScanExecBase` derives from the partition keys.
Review Comment:
I'd leave this:
- When the node groups on every key, `preserveKeyOrderingOnCoalesce` (on by
default) already reports the key ordering. So `EnsureRequirements` doesn't pick
the merge.
- When it groups on fewer keys, the merge sorts by the keys it drops. I
measured it with `allowKeysSubsetOfPartitionKeys` on and a window ordered by
the second key. The merge replaced the sort.
A gate on a sort order beyond the partition keys would lose the second case.
##########
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).
`outputOrdering` now reads two lazy vals and the config.
--
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]
