ulysses-you commented on code in PR #58659:
URL: https://github.com/apache/spark/pull/58659#discussion_r4012377678
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -64,70 +72,15 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
*/
case class GroupPartitionsExec(
child: SparkPlan,
- @transient joinKeyPositions: Option[Seq[Int]] = None,
- @transient expectedPartitionKeys:
Option[Seq[(InternalRowComparableWrapper, Int)]] = None,
- @transient reducers: Option[Seq[Option[KeyReducer]]] = None,
- @transient distributePartitions: Boolean = false,
- @transient enableSortedMerge: Boolean = false
+ @transient grouping: PartitionGrouping,
Review Comment:
grouping and outputPartitioning are carried over a child rewrite (:302-303),
and the factory doc (:404-415) states the invariant that nothing in the tree
hands this node a child reporting a different partitioning. I verified it for
the codegen and columnar wrappers (ColumnarToRowExec, RowToColumnarExec,
InputAdapter, WholeStageCodegenExec all report child.outputPartitioning), but
not for AQE, and I could not settle it by reading.
A probe at this head shows what the carried fields do once the invariant is
broken:
- withNewChildren(Seq(childReportingOnePartition)) still reports
numPartitions 2 and groups the two old input partitions, silently, while
child.outputPartitioning reports 1.
- A Java serialization round trip leaves grouping == null (NPE on
groupedPartitions) and an outputPartitioning whose KeyLayout.partitionKeys is
null (KeyLayout:493 is `@transient`) while numPartitions still reports the old
count, so toString and equals change across the wire.
The base held all three of these as `@transient` ... lazy val (base :77,
:226, :309), so a rewritten or deserialized node recomputed them from its new
child and could not be stale.
The rewriter I could not rule out is an AQEShuffleReadExec landing between
this node and a keyed shuffle child: CoalesceShufflePartitions.isSupported
accepts a keyed ENSURE_REQUIREMENTS shuffle (CoalesceShufflePartitions:38-44),
and AQEShuffleReadExec reports UnknownPartitioning for a keyed shuffle
(AQEShuffleReadExec:100-105). With the stale keyed claim the join above still
satisfies its distribution, so ValidateRequirements.validate
(AdaptiveSparkPlanExec:205-213) would accept the coalesced plan, where the base
rejects it and reverts the rewrite.
Since I could not build a plan with a GroupPartitionsExec directly over a
keyed ShuffleExchangeExec, I am not calling this blocking. Could we either
re-derive in withNewChildInternal when newChild.outputPartitioning ne
child.outputPartitioning, or name in the doc the rules that guarantee the
invariant? Either way a future child-rewriting rule fails loudly instead of
reading a layout its child no longer has.
--
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]
