peter-toth commented on code in PR #59189:
URL: https://github.com/apache/spark/pull/59189#discussion_r4158880681
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5461,6 +5461,41 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59905: ORDER BY the join-key expression of a one-side shuffle") {
Review Comment:
Thanks, confirmed and done in
[8f061e0](https://github.com/apache/spark/pull/59189/commits/8f061e0e8384854e2d87e041471523f56db367c2).
On master with only `sorting.enabled`, the broadcast form returns `b` as 0 to
6, and the `p.s.a` and `p.b + p.c` forms fail as in the shuffle case. The test
now runs every query through both a one-side shuffle and a broadcast join, and
the description no longer says the shuffle conf is needed.
The same expansion also reaches #59165's wrong result with every conf at its
default
([comment](https://github.com/apache/spark/pull/59165#issuecomment-5937582338)).
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5461,6 +5461,41 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59905: ORDER BY the join-key expression of a one-side shuffle") {
+ createTable("ident", Array(Column.create("id", LongType)),
Array(identity("id")))
+ sql("INSERT INTO testcat.ns.ident VALUES " + (94 to 100).map(i =>
s"($i)").mkString(", "))
+ createTable("plain", Array(Column.create("b", LongType),
Column.create("c", LongType),
+ Column.create("s", new StructType().add("a", LongType))), Array.empty)
+ sql("INSERT INTO testcat.ns.plain VALUES " +
+ (0 to 6).map(b => s"($b, 94, named_struct('a', ${94 +
b}L))").mkString(", "))
+
+ // The join shuffles `plain` onto `ident`'s keys and reports the join key,
e.g. `100 - b`, as
+ // its partition expression. The ORDER BY over it is served by the key
order, so the one-side
+ // shuffle is the only one. The keys come out ascending, so only a
descending order needs a
+ // `GroupPartitionsExec` to sort the partitions. AQE stays off, since its
plan validation fails
+ // on `b + c` until SPARK-59901 is fixed.
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
Review Comment:
Done in
[8f061e0](https://github.com/apache/spark/pull/59189/commits/8f061e0e8384854e2d87e041471523f56db367c2).
The single-column queries run with AQE on and off. Only `p.b + p.c` stays off
until #59165 lands.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5461,6 +5461,41 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59905: ORDER BY the join-key expression of a one-side shuffle") {
+ createTable("ident", Array(Column.create("id", LongType)),
Array(identity("id")))
+ sql("INSERT INTO testcat.ns.ident VALUES " + (94 to 100).map(i =>
s"($i)").mkString(", "))
+ createTable("plain", Array(Column.create("b", LongType),
Column.create("c", LongType),
+ Column.create("s", new StructType().add("a", LongType))), Array.empty)
+ sql("INSERT INTO testcat.ns.plain VALUES " +
+ (0 to 6).map(b => s"($b, 94, named_struct('a', ${94 +
b}L))").mkString(", "))
+
+ // The join shuffles `plain` onto `ident`'s keys and reports the join key,
e.g. `100 - b`, as
+ // its partition expression. The ORDER BY over it is served by the key
order, so the one-side
+ // shuffle is the only one. The keys come out ascending, so only a
descending order needs a
+ // `GroupPartitionsExec` to sort the partitions. AQE stays off, since its
plan validation fails
+ // on `b + c` until SPARK-59901 is fixed.
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_SORTING_ENABLED.key -> "true") {
+ val ids = (94 to 100).map(i => Row(i.toLong))
+ Seq(
Review Comment:
Added in
[8f061e0](https://github.com/apache/spark/pull/59189/commits/8f061e0e8384854e2d87e041471523f56db367c2),
with your `ident2(x, y)` shape ordered by `100 - p.b, p.c DESC`. It needs the
mixed-direction `GroupPartitionsExec`, and master returns its rows in the wrong
order on both routes.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -139,10 +139,16 @@ case class EnsureRequirements(
// OrderedDistribution requires grouped KeyedPartitioning with
sorted keys
// according to the distribution's ordering.
val satisfyingKeyedPartitioning = resolution.fold(identity, _._1)
- // The single-column invariant in
KeyedPartitioning.supportsExpressions guarantees
- // one attribute per partition expression.
- val attrs =
satisfyingKeyedPartitioning.expressions.flatMap(_.references)
- val keyRowOrdering = RowOrdering.create(o.ordering, attrs)
+ // A key row holds the values of the partition expressions.
`keysSatisfy` admits a
+ // partitioning here only when those expressions are the
ordering's, position by
+ // position, so each sort order reads its position of the key
row. The expression
+ // need not be a bare column, e.g. a shuffled side's join key
`100 - b`.
+ val keyRowOrdering = RowOrdering.create(
Review Comment:
Done in
[8f061e0](https://github.com/apache/spark/pull/59189/commits/8f061e0e8384854e2d87e041471523f56db367c2),
with the assertion.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -139,10 +139,16 @@ case class EnsureRequirements(
// OrderedDistribution requires grouped KeyedPartitioning with
sorted keys
// according to the distribution's ordering.
val satisfyingKeyedPartitioning = resolution.fold(identity, _._1)
- // The single-column invariant in
KeyedPartitioning.supportsExpressions guarantees
- // one attribute per partition expression.
- val attrs =
satisfyingKeyedPartitioning.expressions.flatMap(_.references)
- val keyRowOrdering = RowOrdering.create(o.ordering, attrs)
+ // A key row holds the values of the partition expressions.
`keysSatisfy` admits a
+ // partitioning here only when those expressions are the
ordering's, position by
+ // position, so each sort order reads its position of the key
row. The expression
+ // need not be a bare column, e.g. a shuffled side's join key
`100 - b`.
Review Comment:
Done in
[8f061e0](https://github.com/apache/spark/pull/59189/commits/8f061e0e8384854e2d87e041471523f56db367c2).
Both now say that a scan reports one column per expression, and that `reorder`
gives up on a join key like `b + c`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -139,10 +139,16 @@ case class EnsureRequirements(
// OrderedDistribution requires grouped KeyedPartitioning with
sorted keys
// according to the distribution's ordering.
val satisfyingKeyedPartitioning = resolution.fold(identity, _._1)
- // The single-column invariant in
KeyedPartitioning.supportsExpressions guarantees
- // one attribute per partition expression.
- val attrs =
satisfyingKeyedPartitioning.expressions.flatMap(_.references)
- val keyRowOrdering = RowOrdering.create(o.ordering, attrs)
+ // A key row holds the values of the partition expressions.
`keysSatisfy` admits a
+ // partitioning here only when those expressions are the
ordering's, position by
+ // position, so each sort order reads its position of the key
row. The expression
+ // need not be a bare column, e.g. a shuffled side's join key
`100 - b`.
+ val keyRowOrdering = RowOrdering.create(
+ o.ordering.zipWithIndex.map { case (order, i) =>
+ SortOrder(BoundReference(i, order.child.dataType, nullable =
true),
Review Comment:
Done in
[8f061e0](https://github.com/apache/spark/pull/59189/commits/8f061e0e8384854e2d87e041471523f56db367c2).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -139,10 +139,16 @@ case class EnsureRequirements(
// OrderedDistribution requires grouped KeyedPartitioning with
sorted keys
// according to the distribution's ordering.
val satisfyingKeyedPartitioning = resolution.fold(identity, _._1)
- // The single-column invariant in
KeyedPartitioning.supportsExpressions guarantees
- // one attribute per partition expression.
- val attrs =
satisfyingKeyedPartitioning.expressions.flatMap(_.references)
- val keyRowOrdering = RowOrdering.create(o.ordering, attrs)
+ // A key row holds the values of the partition expressions.
`keysSatisfy` admits a
+ // partitioning here only when those expressions are the
ordering's, position by
+ // position, so each sort order reads its position of the key
row. The expression
+ // need not be a bare column, e.g. a shuffled side's join key
`100 - b`.
+ val keyRowOrdering = RowOrdering.create(
+ o.ordering.zipWithIndex.map { case (order, i) =>
+ SortOrder(BoundReference(i, order.child.dataType, nullable =
true),
+ order.direction, order.nullOrdering, Seq.empty)
Review Comment:
Done in
[8f061e0](https://github.com/apache/spark/pull/59189/commits/8f061e0e8384854e2d87e041471523f56db367c2).
--
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]
