Skip to content

[python][ray] Avoid routing shuffle in self-merge updates - #9335

Draft
XiaoHongbo-Hope wants to merge 1 commit into
apache:masterfrom
XiaoHongbo-Hope:codex/ray-self-merge-no-shuffle
Draft

[python][ray] Avoid routing shuffle in self-merge updates#9335
XiaoHongbo-Hope wants to merge 1 commit into
apache:masterfrom
XiaoHongbo-Hope:codex/ray-self-merge-no-shuffle

Conversation

@XiaoHongbo-Hope

@XiaoHongbo-Hope XiaoHongbo-Hope commented Aug 21, 2026

Copy link
Copy Markdown
Contributor

Purpose

Ray _ROW_ID self-merge already avoids the source-target join, but its update path still sends every transformed row through a _FIRST_ROW_ID hash shuffle before writing. This adds avoidable network, object-store, and scheduling overhead for large column updates.

Changes

  • pin the target scan plan used by self-merge updates
  • unpack planned splits into complete first_row_id file groups and dispatch
    each group directly to a bounded Ray task
  • read, apply matched clauses, and stage row-id updates within that task
  • collect normal CommitMessages and commit once on the driver
  • stop scheduling new work after a task failure and abort files staged by completed tasks

This bounds task materialization to one logical file group rather than a packed split. The public API is unchanged. Callable assignments from #9327 reuse the same path; general source-target merge and self-merge delete paths are unchanged.

Tests

  • self-merge updates succeed when Dataset.groupby() is disabled
  • a packed split is separated into complete logical file groups
  • callable read_columns and VARIANT transformations work without the shuffle
  • a later group failure aborts files staged by earlier groups
  • ray_data_evolution_merge_into_test.py: 115 passed
  • ray_update_by_row_id_test.py: 16 passed
  • flake8, compileall, and git diff --check passed

Local filesystem microbenchmark shaped like the reported workload (Ray 2.56.1, 4 CPUs, append-only data-evolution table with row tracking, 100,000 rows, 20 files, 65 clips per file, and 1–3 KiB VARIANT payloads): master with the routing shuffle took 5.43/5.57 s; this PR took 1.45/1.50 s (about 3.7x). The callable performs variant_get, negates four DOUBLE paths, and writes them with variant_replace. All 60,060 matched rows and 39,940 unmatched rows were verified after every run. This is a local directional result, not a production-table claim.

@XiaoHongbo-Hope
XiaoHongbo-Hope force-pushed the codex/ray-self-merge-no-shuffle branch from 6f57492 to 35ce199 Compare August 21, 2026 06:00
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