Skip to content

[python] Add range_join: shuffle-free Ray join for range-clustered tables#8738

Draft
XiaoHongbo-Hope wants to merge 12 commits into
apache:masterfrom
XiaoHongbo-Hope:ray_range_join
Draft

[python] Add range_join: shuffle-free Ray join for range-clustered tables#8738
XiaoHongbo-Hope wants to merge 12 commits into
apache:masterfrom
XiaoHongbo-Hope:ray_range_join

Conversation

@XiaoHongbo-Hope

Copy link
Copy Markdown
Contributor

Purpose

ray.data.join shuffles 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. Complements bucket_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= or left_on/right_on, inner join. Shared driver/worker helpers are factored into join_common and reused by bucket_join.

Tests

ray_range_join_test.py; flake8 clean. The exactly-once planning invariant was additionally verified over randomized range/skew/null trials.

…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.
…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.
Under DATE->TIMESTAMP schema evolution, old files' footer bounds are dates
and new files' are datetimes; comparing/sorting them raw raised TypeError.
Coerce every footer min/max to the current join-key arrow type before use;
a bound that can't be cast makes the split unknown (joins every range).

Verified locally on Python 3.11: real DATE->TIMESTAMP evolution join now
returns the correct result; ray_range_join_test 15 passed.
Coercing an old file's footer bounds to the current key type is unsafe when
the cast doesn't preserve order: INT->STRING makes 5/100 into '5'/'100', but
'42' > '100' as a string, so a coerced range silently drops the '42' row.

Use a file's footer stats only when it stored the key in the current type;
any other type (schema evolution) marks the split unknown (joins every range).
Also drop the range predicate pushdown -- the reader can't compare a new-type
bound against an evolved file's physical column; the in-memory clip does the
exact, evolution-safe filtering.

Verified locally on Python 3.11: INT->STRING and DATE->TIMESTAMP evolutions
now return the full correct result; ray_range_join_test 16 passed.
@XiaoHongbo-Hope
XiaoHongbo-Hope marked this pull request as ready for review July 22, 2026 12:06
@XiaoHongbo-Hope
XiaoHongbo-Hope marked this pull request as draft July 22, 2026 12:20
- left_partitions/right_partitions ({column: value} on partition columns) prune
  each side to those partitions before planning, so a join can target specific
  partitions instead of whole tables.
- Read each split's parquet footer in a thread pool instead of serially, cutting
  driver-side planning latency on many-file tables.

Verified locally on Python 3.11: partition-filtered and partitioned-table joins,
many-to-many correctness; ray_range_join_test 18 passed.
…nown ones

The old budget only counted unknown-stats splits, but a known split whose
[lo,hi] spans many ranges is also read once per range (no pushdown; read the
whole split, clip in memory). On poorly clustered input that re-read could
exceed a shuffle. Cap ranges by the actual total re-read -- rows times ranges
each split overlaps -- halving num_ranges until it stays within a couple of
full scans. Document that the feature only benefits clustered inputs.

Verified locally: ray_range_join_test 18 passed (incl. the budget contract for
wide/unknown splits), ray_bucket_join_test 18 passed.
…tes, validate inputs

Review fixes:
- Resolve the range key's footer column by field id, so a renamed/swapped
  (same-type) column reads its own stats instead of another column's.
- Treat a PK table's non-PK range key as unknown: merge (aggregation/partial
  update) can move it past the file min/max, which would silently drop matches.
- Bound re-read by file bytes, not row count, so a wide-row split can't be
  re-read hundreds of times within budget.
- Validate partition keys (reject non-partition columns; None -> is_null) and
  num_ranges (int >= 1, capped at _MAX_RANGES even when passed explicitly).
- Warn instead of silently degrading when a parquet footer read fails.
…e driver

Fail fast for range keys that can't be range-partitioned instead of raising deep
inside a Ray worker: nested/complex types (ARRAY/MAP/ROW/...) and both string
forms of TIMESTAMP WITH LOCAL TIME ZONE (TIMESTAMP_LTZ(p) as well as ... WITH
LOCAL TIME ZONE, only the latter was caught before).
…te all keys

- P0: a range key under auth column-masking is rewritten at read time, so raw
  parquet footer min/max no longer bounds it -> treat as unknown (join every
  range, clip in memory), same fallback as a merge-rewritten non-PK key.
- Validate every join key (not just the range key) up front and reject VARIANT
  as well as nested types, instead of failing as ArrowInvalid inside a worker.
- Fix a stale test comment (footer, not manifest, stats).
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant