[python] Add range_join: shuffle-free Ray join for range-clustered tables#8738
Draft
XiaoHongbo-Hope wants to merge 5 commits into
Draft
[python] Add range_join: shuffle-free Ray join for range-clustered tables#8738XiaoHongbo-Hope wants to merge 5 commits into
XiaoHongbo-Hope wants to merge 5 commits into
Conversation
…bles Cut the join-key space into ranges from per-file min/max stats in the manifest, then read and join each range in its own Ray task -- no global shuffle. Complements bucket_join for tables sorted/clustered by the join key rather than co-bucketed. Half-open ranges plus an in-memory range filter keep every matching pair produced exactly once regardless of range count; splits with no usable stats fall back to joining every range. Shared driver/worker helpers (table cache, snapshot pin, split read) are factored into join_common and reused by bucket_join.
XiaoHongbo-Hope
force-pushed
the
ray_range_join
branch
from
July 19, 2026 14:05
981697f to
daad7ae
Compare
…join output - Blocker: scan.plan() drops value stats, so every split looked unknown and the join always collapsed to one task. Read per-file min/max straight from the manifest (drop_stats=False), keyed by file name; stats never reach the workers. - Reject FLOAT/DOUBLE range keys (NaN falls out of every range while the hash join still matches NaN==NaN -> dropped matches) and TIMESTAMP_LTZ (naive manifest stats vs tz-aware column can't be compared). - Fix the collision check to catch a left key colliding with a right non-key column; document that the output keeps the left key names. - Cap the range count by an unknown-stats read budget so the every-range fallback re-reads at most one extra full scan.
The value_stats column names come from value_stats_cols when set, else the full schema -- the write_cols/schema_id detour resolved wrong names, so every split looked unknown and the join collapsed to one range. Read min/max via the same path files_table uses. Update the collision-message assertion.
value_stats.min/max_values are self-describing BinaryRows: they expose get_field(idx) and their own fields, not a .values list. The previous code read a non-existent .values attribute, so every split looked unknown and the join collapsed to one range. Index by the BinaryRow's own field names.
…fest pypaimon-written tables leave the manifest value stats empty (min/max live only in the parquet footer), so the manifest approach always saw unknown ranges and collapsed to one task. Read each split file's footer min/max via FileIO on the driver; the splits sent to workers still carry no stats. Verified locally on Python 3.11 (pyarrow+ray): ray_range_join_test 14 passed, ray_bucket_join_test 18 passed.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
ray.data.joinshuffles both sides. When two tables are sorted/clustered by the join key, that shuffle is avoidable: cut the key space into ranges from per-file min/max stats in the manifest, then read and join each range in its own Ray task — no global shuffle. Complementsbucket_join(which needs co-bucketed tables) for the sorted-but-not-co-bucketed case.Half-open ranges plus an in-memory clip keep every matching pair produced exactly once regardless of range count; splits with no usable stats fall back to joining every range. Supports
on=orleft_on/right_on, inner join. Shared driver/worker helpers are factored intojoin_commonand reused bybucket_join.Tests
ray_range_join_test.py; flake8 clean. The exactly-once planning invariant was additionally verified over randomized range/skew/null trials.