Fix Distinct expression lowering - #22970
Conversation
rjzamora
added
2 - In Progress
github-actions
Bot
added
Python
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Enterprise Run ID: 📒 Files selected for processing (2)
📝 WalkthroughSummary by CodeRabbit
WalkthroughStreaming distinct lowering now separates shared key selection and static reduction from dynamic-planning dispatch. unique() builds the distinct IR explicitly, join filter task setup is split into routing and channel wiring, and groupby strategy selection now clamps the non-tree output count. ChangesStreaming distinct lowering and unique regression
Streaming actor-graph wiring
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~30 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@python/cudf_polars/tests/streaming/test_unique.py`:
- Around line 57-73: The dynamic-planning unique test only checks the physical
plan and does not verify correctness of `unique()` results. In
`test_unique_select_dynamic_planning_uses_dynamic_distinct`, add a
result-equivalence assertion for `q` against the CPU Polars output, using the
same `GPUEngine` setup and allowing `check_row_order=False` if needed. Keep the
existing `explain_query` assertions for `DISTINCT` and `REPARTITION`, but make
sure the test also validates GPU output matches Polars semantics.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 66bb2b07-731d-4dae-bc1d-b468fba53d7e
📒 Files selected for processing (3)
python/cudf_polars/cudf_polars/streaming/distinct.pypython/cudf_polars/cudf_polars/streaming/expressions.pypython/cudf_polars/tests/streaming/test_unique.py
…jzamora/cudf into fix-unique-expression-dynamic-distinct
…on-dynamic-distinct
…on-dynamic-distinct
pentschev
left a comment
pentschev
left a comment
There was a problem hiding this comment.
I've tested this, and alone it exposes a bug that @rjzamora discovered is luckily being resolved by #22995 . The bug causes a hang in NDSH Q22, in Rick's words about the issue:
When we go down the shuffle path, we are already partitioned on the right side of a left join. The current bloom-filter logic will create filtering channels, but then bail on the filtering because we are already partitioned on the keys. Then we end up waiting on the wrong channels.
Combining this change with #22995 resolves both the original issue (being fixed by this bug) and the hang. The results for SF30K on 8xNVL72 is 4.50s lukewarm, 3.13s hot.
Since, originally, Q22 didn't work anyway, I don't think it's necessary to wait for #22995 and we can merge it anytime now.
| # query DAG. | ||
| return ch_left, ch_right, [], [] | ||
|
|
||
| bloom_build_output: Channel[BloomFilterChunk] = context.create_channel() |
There was a problem hiding this comment.
@pentschev - Thanks for the review! Just a note that these changes "fix" the hang when this branch is used in the absence of #22995 - Hopefully you don't mind the conflicts this creates in that PR.
There was a problem hiding this comment.
My apologies, I had indeed forgotten to pull the latest changes. However, the current state with latest 042b9b4 is actually back to the original problem, q22 fails:
OverflowError: CUDF failure at: /tmp/conda-bld-output/bld/rattler-build_libcudf/work/cpp/src/copying/concatenate.cu:476: Total number of concatenated rows exceeds the column size limit
The previous state I was trying was the latest commit I had locally, which was 19ed80e, and merging that on top of #22997 , that had worked as expected, and all queries passed (without the need for changes to the queries themselves).
With the above being said I want to ask whether the changes coming after 19ed80e are really necessary, or are they fixed by a combination of 19ed80e + #22997 (which includes changes also from #22995 and #22996), WDYT? Once again, I have already verified original changes from this PR + #22997 has everything in a good state, but the same is not true with the current in 042b9b4, which brings back the original issue to Q22. For the sake of simplicity (rerunning everything at scale is time-consuming) I would propose instead merging the changes here only up to and including 19ed80e and then #22995, #22996 and #22997, which I have already confirmed to get us to the state we want to be in.
Let me know what you think. For now I'm changing my approval to block the PR from an accidental merge until we are sure of next steps.
There was a problem hiding this comment.
the current state with latest 042b9b4 is actually back to the original problem, q22 fails
Okay - I don't quite understand why that might be the case yet, but that's good to know.
With the above being said I want to ask whether the changes coming after 19ed80e are really necessary
The changes were meant to avoid a hang between 22970 and 22995 being merged. However, I was assuming you would just ignore/replace any changes made to this file.
I definitely don't understand why we would be back to the int-overflow issue with this change.
There was a problem hiding this comment.
It seems like something messed up with my image build and I probably ran an incorrect version. Indeed, after rebuilding I can confirm everything works with this PR now. I'm very sorry for the confusion and added work on verifications.
| # query DAG. | ||
| return ch_left, ch_right, [], [] | ||
|
|
||
| bloom_build_output: Channel[BloomFilterChunk] = context.create_channel() |
There was a problem hiding this comment.
My apologies, I had indeed forgotten to pull the latest changes. However, the current state with latest 042b9b4 is actually back to the original problem, q22 fails:
OverflowError: CUDF failure at: /tmp/conda-bld-output/bld/rattler-build_libcudf/work/cpp/src/copying/concatenate.cu:476: Total number of concatenated rows exceeds the column size limit
The previous state I was trying was the latest commit I had locally, which was 19ed80e, and merging that on top of #22997 , that had worked as expected, and all queries passed (without the need for changes to the queries themselves).
With the above being said I want to ask whether the changes coming after 19ed80e are really necessary, or are they fixed by a combination of 19ed80e + #22997 (which includes changes also from #22995 and #22996), WDYT? Once again, I have already verified original changes from this PR + #22997 has everything in a good state, but the same is not true with the current in 042b9b4, which brings back the original issue to Q22. For the sake of simplicity (rerunning everything at scale is time-consuming) I would propose instead merging the changes here only up to and including 19ed80e and then #22995, #22996 and #22997, which I have already confirmed to get us to the state we want to be in.
Let me know what you think. For now I'm changing my approval to block the PR from an accidental merge until we are sure of next steps.
pentschev
left a comment
pentschev
left a comment
There was a problem hiding this comment.
After another review, it was a mistake with the build of my image, I don't know what happened that caused it to build with the wrong change. I can now confirm again this is indeed fixing the original issue and there are no blockers, we are good to merge.
Once again, thanks @rjzamora for working on this fix!
| # query DAG. | ||
| return ch_left, ch_right, [], [] | ||
|
|
||
| bloom_build_output: Channel[BloomFilterChunk] = context.create_channel() |
There was a problem hiding this comment.
It seems like something messed up with my image build and I probably ran an incorrect version. Indeed, after rebuilding I can confirm everything works with this PR now. I'm very sorry for the confusion and added work on verifications.
|
/merge |
4 participants
Expression-level
unique()was bypassing the dynamicDistinctlowering path by calling the staticlower_distincttree reduction directly. This updates the sharedDistinctlowering helper soselect(pl.col(...).unique())can use dynamic planning and avoid static repartition/allgather paths that may overflow for large cardinalities.Adds a small regression test covering dynamic expression
unique()lowering.