peter-toth commented on PR #58262: URL: https://github.com/apache/spark/pull/58262#issuecomment-5442936141
Thanks @dongjoon-hyun, this was a good catch on 1. All four correctness items are addressed in a new commit, and 6 and 7 with them. **1. Confirmed, and it is a regression this PR introduces.** Your repro plans and returns the right rows though, so it is worth writing down what it takes and where the throw actually comes from, because the mechanism is not the one in the finding. `key.row.get(0, TimestampType)` does not throw: `BaseGenericInternalRow.get` ignores the requested type and hands back the Integer. The projected keys are then wrapped as `TimestampType` over Integer values, and `distinct` hashes them - which tolerates the mismatch too. The throw is in `InternalRowComparableWrapper.equals`, i.e. in the ordering built for the declared types, so it needs two projected keys that **collide**. In your example the two rows are 2020 and 2021, they project to different years, `equals` is never called, and the query plans. Two rows in the same year and different buckets do it: `(1, '2020-01-01')` and `(2, '2020-01-01')` in both tables, reduced keys `(50, 1)` and `(50, 2)`, both projecting to `50`. On this PR's pushed head that is `ClassCastException: class java.lang.Integer cannot be cast to class java.lang.Long` at planning. And the same data shows why the projection has to happen: on master the query plans with no node and returns `s = 10` and `s = 20` for two rows that share a `ts`, where the answer is 30 and 30. So it is the wrong-result shape of this JIRA and the crash in one query, which is now a test. The fix is to read the keys at the types they were built with rather than at the types their expressions declare. `KeyedPartitioning.keyDataTypes` returns `partitionKeys.head.dataTypes` - the schema each key row was written under, and the one it is hashed and compared under - and `projectKeys` uses it. The expressions describe the partitioning; a reducer made that description stale, but the rows never lied, so they are the sound source for reading a row. This is not the general fix for the stale expressions, which is #58335 plus the both-sides-reduce case left after it; it is the part this PR needs, because this PR is what starts projecting those keys. **2. Closed by the same change.** `GroupPartitionsExec` now takes its base types from `keyDataTypes` too, so both sides read the keys the same way. The count the node produces no longer depends on which member `collectFirst` picks: the positions are the same, the keys are shared by the `PartitioningCollection` invariant, and the types now come from those keys rather than from a member's `expressionDataTypes`. That also let the `numPartitionsAfter` memo drop the data types from its key. **3. Agreed, and it is on my list as item 2** of the follow-up comment above - pre-existing, measured identical on master, and I will file it with the repro. You are right that the new cogroup test does not cover it: it uses identical layouts on both sides, so the positions agree and the hole stays invisible. A test for it belongs with the fix, since it fails on master today. **4. Fixed by skipping such a member instead of asserting.** Relaxing the assert would not do: the empty position set flows on as the projection, and a node projecting to no position at all puts every partition into one - an earlier round of this PR measured exactly that (3 partitions to 1, and the result did not satisfy the distribution it was inserted for). So a member covering no position is not a candidate, and the child is shuffled, which is correct for a partitioning grouped on a reference-free expression. `EnsureRequirementsSuite` has a test that threw the `AssertionError` before. **5. Agreed, and it is the same thing you asked for in your first review** (your #6 there, item 5 of my follow-up list): one matcher next to `groupedSatisfies` feeding both admission and projection, so the two cannot drift. That is the follow-up I want to take next in this area; it is more than a bug fix should carry. **6. Done, with `Either` rather than a new type.** `splitKeyedPartitionings` returns `Option[Either[KeyedPartitioning, (KeyedPartitioning, Option[Seq[Int]])]]` behind a private type alias: `Left` satisfies as it is, `Right` needs a node with these positions, `None` needs a shuffle. The `orElse(...).get` and `needsGrouping.get._2` are gone. One wrinkle worth knowing if you touch it: `Left` and `Right` have to be written `scala.Left` / `scala.Right` in this file, because `catalyst.expressions` has its own. **7. Done.** The walk now only sorts the members into keyed and non-keyed, the non-keyed question is asked immediately after it, and the keyed analysis runs in a second method that is only entered when no plain member satisfies. Nothing touches a partition key before that answer. **8. Agreed on the per-plan cache, and it belongs on `KeyedPartitioning`.** The distinct count that `KeyedPartitioning.apply` computes for `isGrouped` and discards is the clearest case. I left it out here because it is a catalyst change in a fix that otherwise only changes `EnsureRequirements` and the node, and because the memo does share across the members of one operator, which is the part that repeats within a call. The stacked-operators case is real and I will carry it into the follow-up above. **9. The second evaluation is the price of asking `satisfies` rather than re-deriving it.** Only a member that fails `satisfies` pays it. Folding the two would mean computing `groupedSatisfies` once and combining it locally with the `requiredNumPartitions` gate - which is what an earlier revision of this PR did, and it dropped the gate for a grouped member: a blind review round measured an Exchange with 5 partitions on master against 3 partitions and no shuffle on the branch. Memoizing `groupedSatisfies` per distribution would fix it without that risk, and again on the partitioning rather than here. `KeyGroupedPartitioningSuite`, `EnsureRequirementsSuite`, `ValidateRequirementsSuite`, `PlannerSuite`, `ProjectedOrderingAndPartitioningSuite`, `UnionSuite`, `DataSourceV2Suite`, `AdaptiveQueryExecSuite`, `DistributionSuite` and `ShuffleSpecSuite` are green - 575 tests. Both new tests fail on `master`, and on this PR's previous head they fail with the `ClassCastException` and the `AssertionError` respectively. -- 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]
