You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "cloud-fan (via GitHub)" <gi...@apache.org> on 2026/05/14 09:02:09 UTC
[PR] [SPARK-46367][SQL][FOLLOWUP] Filter mixed-arity KPs in projectKeyedPartitionings [spark]
cloud-fan opened a new pull request, #55876:
URL: https://github.com/apache/spark/pull/55876
### What changes were proposed in this pull request?
Follow-up to https://github.com/apache/spark/pull/55519.
`PartitioningPreservingUnaryExecNode.projectKeyedPartitionings` uses
`kps.head.expressions.length` as the arity for the per-position projection and
then indexes `kp.expressions(i)` for every input KP. If a child exposes a
`PartitioningCollection` containing KPs of different arities — which can happen
when upstream nodes have already narrowed some KPs but not others — indexing
`kp.expressions(i)` for `i >= kp.expressions.length` throws
`IndexOutOfBoundsException`.
Filter the input KPs to those whose arity matches the head before the
per-position loop. This restores the function's "shared partitionKeys"
invariant on the surviving KPs.
### Why are the changes needed?
Without the filter, plans that put a `ProjectExec` (or any other
`PartitioningPreservingUnaryExecNode`) above a `PartitioningCollection` with
mixed-arity KPs fail with `IndexOutOfBoundsException` during planning.
### Does this PR introduce _any_ user-facing change?
No (bug fix; planner only).
### How was this patch tested?
New unit test `SPARK-46367: projectKeyedPartitionings tolerates mixed-arity
KPs in child output` in `ProjectedOrderingAndPartitioningSuite` that constructs
a mixed-arity `PartitioningCollection` child and verifies
`ProjectExec.outputPartitioning` does not throw.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 4.6
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #55876:
URL: https://github.com/apache/spark/pull/55876#issuecomment-4456682275
The parquet test failure is unrelated, thanks for review, merging to master/4.x/4.2!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55876:
URL: https://github.com/apache/spark/pull/55876#discussion_r3242430521
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -94,6 +94,16 @@ trait PartitioningPreservingUnaryExecNode extends UnaryExecNode
kps: Seq[KeyedPartitioning]): LazyList[KeyedPartitioning] = {
if (kps.isEmpty) return LazyList.empty
val numPositions = kps.head.expressions.length
+ // The function assumes all input KPs share the same `partitionKeys`, which implies matching
+ // expression arity. This invariant is asserted by [[GroupPartitionsExec]] and is established
+ // by the constructors of [[PartitioningCollection]] feeding this method (a join's
+ // `PartitioningCollection(left.outputPartitioning, right.outputPartitioning)` combines KPs
+ // that have been aligned by [[EnsureRequirements]] to the same join keys). If the invariant
+ // is ever violated upstream, fail early with a clear message instead of throwing an opaque
+ // `IndexOutOfBoundsException` from `kp.expressions(i)` below.
+ assert(kps.forall(_.expressions.length == numPositions),
Review Comment:
I wonder if we should move these invariant checks into `PartitioningCollection`.
Maybe `expressions.length =` + `partitionKeys eq` checks on KPs for maximum efficiency.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55876:
URL: https://github.com/apache/spark/pull/55876#discussion_r3242430521
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -94,6 +94,16 @@ trait PartitioningPreservingUnaryExecNode extends UnaryExecNode
kps: Seq[KeyedPartitioning]): LazyList[KeyedPartitioning] = {
if (kps.isEmpty) return LazyList.empty
val numPositions = kps.head.expressions.length
+ // The function assumes all input KPs share the same `partitionKeys`, which implies matching
+ // expression arity. This invariant is asserted by [[GroupPartitionsExec]] and is established
+ // by the constructors of [[PartitioningCollection]] feeding this method (a join's
+ // `PartitioningCollection(left.outputPartitioning, right.outputPartitioning)` combines KPs
+ // that have been aligned by [[EnsureRequirements]] to the same join keys). If the invariant
+ // is ever violated upstream, fail early with a clear message instead of throwing an opaque
+ // `IndexOutOfBoundsException` from `kp.expressions(i)` below.
+ assert(kps.forall(_.expressions.length == numPositions),
Review Comment:
I wonder if we should move these invariant checks into `PartitioningCollection`.
Maybe `expressions.length =` + `partitionKeys eq` checks for maximum efficiency.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55876:
URL: https://github.com/apache/spark/pull/55876#discussion_r3242180952
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -94,6 +94,16 @@ trait PartitioningPreservingUnaryExecNode extends UnaryExecNode
kps: Seq[KeyedPartitioning]): LazyList[KeyedPartitioning] = {
if (kps.isEmpty) return LazyList.empty
val numPositions = kps.head.expressions.length
+ // The function assumes all input KPs share the same `partitionKeys`, which implies matching
+ // expression arity. This invariant is asserted by [[GroupPartitionsExec]] and is established
+ // by the constructors of [[PartitioningCollection]] feeding this method (a join's
+ // `PartitioningCollection(left.outputPartitioning, right.outputPartitioning)` combines KPs
+ // that have been aligned by [[EnsureRequirements]] to the same join keys). If the invariant
+ // is ever violated upstream, fail early with a clear message instead of throwing an opaque
+ // `IndexOutOfBoundsException` from `kp.expressions(i)` below.
+ assert(kps.forall(_.expressions.length == numPositions),
Review Comment:
Maybe you could use `kps.tail.forall(...)`.
Please note that in `GroupPartitionsExec` we use `.partitionKeys` comparison, which is even more stricter, but more costly.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55876:
URL: https://github.com/apache/spark/pull/55876#discussion_r3249332921
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -94,6 +94,16 @@ trait PartitioningPreservingUnaryExecNode extends UnaryExecNode
kps: Seq[KeyedPartitioning]): LazyList[KeyedPartitioning] = {
if (kps.isEmpty) return LazyList.empty
val numPositions = kps.head.expressions.length
+ // The function assumes all input KPs share the same `partitionKeys`, which implies matching
+ // expression arity. This invariant is asserted by [[GroupPartitionsExec]] and is established
+ // by the constructors of [[PartitioningCollection]] feeding this method (a join's
+ // `PartitioningCollection(left.outputPartitioning, right.outputPartitioning)` combines KPs
+ // that have been aligned by [[EnsureRequirements]] to the same join keys). If the invariant
+ // is ever violated upstream, fail early with a clear message instead of throwing an opaque
+ // `IndexOutOfBoundsException` from `kp.expressions(i)` below.
+ assert(kps.forall(_.expressions.length == numPositions),
Review Comment:
@cloud-fan I wonder if we coulnd enforce this invariant in `PartitioningCollection`: https://github.com/apache/spark/pull/55901
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55876:
URL: https://github.com/apache/spark/pull/55876#discussion_r3249332921
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -94,6 +94,16 @@ trait PartitioningPreservingUnaryExecNode extends UnaryExecNode
kps: Seq[KeyedPartitioning]): LazyList[KeyedPartitioning] = {
if (kps.isEmpty) return LazyList.empty
val numPositions = kps.head.expressions.length
+ // The function assumes all input KPs share the same `partitionKeys`, which implies matching
+ // expression arity. This invariant is asserted by [[GroupPartitionsExec]] and is established
+ // by the constructors of [[PartitioningCollection]] feeding this method (a join's
+ // `PartitioningCollection(left.outputPartitioning, right.outputPartitioning)` combines KPs
+ // that have been aligned by [[EnsureRequirements]] to the same join keys). If the invariant
+ // is ever violated upstream, fail early with a clear message instead of throwing an opaque
+ // `IndexOutOfBoundsException` from `kp.expressions(i)` below.
+ assert(kps.forall(_.expressions.length == numPositions),
Review Comment:
@cloud-fan, I wonder if we could enforce this invariant in `PartitioningCollection`: https://github.com/apache/spark/pull/55901
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Group mixed-arity KPs by arity in projectKeyedPartitionings [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55876:
URL: https://github.com/apache/spark/pull/55876#issuecomment-4449692202
@cloud-fan , I don't think this is valid.
There simply can't be mixed-arity `KeyedPartitioning`s in a `PartitioningCollection`. This is an invariant required by `GroupPartitions` as well: https://github.com/apache/spark/blob/5949ab30b41860574ab57b94a8848464b5e127a7/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala#L70-L72
and actualy, it doesn't make sense to have/keep lower arity KPs in a collection if there are higher arity ones in it because the expressions of the lower must be the subset of the expressions of the higher.
If we know more details of the keys of partitions, like `KP([a, b], [(1, 1), (2, 2)])` so we know that `a = 1` and `b = 1` in the first; and `a = 2` and `b = 2` in the second partition, why would we have/keep the less granular `KP([a], [(1), (2)])` or `KP([b], [(1), (2)])` in the collection?
Maybe the wording of the comment is not the best but the "`partitionKeys` must match" means that arity of KPs in a collection must be the same.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #55876:
URL: https://github.com/apache/spark/pull/55876#issuecomment-4451669035
@peter-toth you're right — the invariant should stay strict. I've reworked this PR to just add an `assert` so an upstream violation surfaces with a clear message instead of an opaque `IndexOutOfBoundsException`. The producer-side bug I was working around lives in our internal fork; I'll fix it there. PTAL.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #55876: [SPARK-46367][SQL][FOLLOWUP] Assert same arity in projectKeyedPartitionings
URL: https://github.com/apache/spark/pull/55876
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46367][SQL][FOLLOWUP] Group mixed-arity KPs by arity in projectKeyedPartitionings [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #55876:
URL: https://github.com/apache/spark/pull/55876#issuecomment-4449360719
cc @peter-toth
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org