fix: fall back to known chunks when S3 continuity check fails#38
Open
NikhilSharmaWe wants to merge 2 commits into
Open
fix: fall back to known chunks when S3 continuity check fails#38NikhilSharmaWe wants to merge 2 commits into
NikhilSharmaWe wants to merge 2 commits into
Conversation
Signed-off-by: Nikhil Sharma <nikhilsharma230303@gmail.com>
define-null
pushed a commit
that referenced
this pull request
Jun 24, 2026
* perf(pg-storage): batch the N+1 loops into set-based statements
Replace every per-row .execute() loop the instrumentation flagged with one
set-based statement, eliminating the O(chunks) round-trips per cycle:
- insert_new_chunks: one bulk INSERT…SELECT via jsonb_to_recordset. A unique
violation aborts the batch and is reported generically as ChunkAlreadyExists
(no per-row culprit lookup); an unknown dataset is detected by RETURNING-count
mismatch.
- update_worker_set: one batch upsert over UNNEST(peer_id[], version[])
RETURNING id,(xmax=0); added = freshly-inserted ids.
- record_routing_diffs / commit_new_ideal: one jsonb_to_recordset INSERT /
upsert (commit keeps its existing sweep-delete).
- mint_superseded_stale_mappings / resolve_flip_flops: flatten to parallel
bigint[] pairs, one INSERT…ON CONFLICT DO NOTHING / one DELETE…USING UNNEST.
- replay_confirmed_diffs: fully server-side — DISTINCT ON (chunk_pk) … ORDER BY
worker_assignment_id DESC for last-write-wins, one data-modifying CTE
(upsert non-empty + delete empty); no rows round-trip to Rust.
PhaseTimer instrumentation retained; statement counts now reflect the batched
queries. Measured on the reshuffle-sim multistep path at ~95K chunks: the five
N+1 phases drop from ~95K statements/~85s combined to 1-2 statements/~5s.
pg correction tests pass; clippy clean.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* perf(pg-storage): drop jsonb_to_recordset for native UNNEST+array_agg
commit_new_ideal and record_routing_diffs now flatten the placement to two
parallel bigint[] params and re-aggregate server-side (array_agg ORDER BY
worker_id, GROUP BY) via a LEFT JOIN over all touched chunk_pks with COALESCE
to '{}' — instead of serializing rows to a JSON string and parsing it with
jsonb_to_recordset. One/two statements, no JSON build/parse, and it scales to
millions of elements in two array params (a single bigint[] is far under the
65535 bind-parameter cap). The LEFT-JOIN-over-all-pks form preserves the old
behavior for empty/removed holder sets. Dropped the now-unused DiffRow struct.
insert_new_chunks keeps jsonb (now commented): its 5 scalar columns plus an
order-sensitive ragged files[] make a native reconstruction high-complexity for
a one-time load path; COPY is the escalation if the 10M-baseline JSON
allocation ever bites.
Addresses review feedback on #38.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* perf(pg-storage): pre-allocate flat-array binds, skip no-op ideal updates
- commit_new_ideal / resolve_flip_flops: pre-size the flattened (chunk_pk,
worker_id) bind vectors with the exact pair count, avoiding repeated
reallocation of the per-cycle placement arrays.
- record_routing_diffs: drop the redundant Rust-side sort; the stored array is
already ordered by array_agg(... ORDER BY worker_id).
- commit_new_ideal upsert: add WHERE worker_ids IS DISTINCT FROM EXCLUDED so an
unchanged chunk produces no dead tuple / WAL. Correct because both sides are
canonically ordered, so an unchanged holder set compares equal.
Also trim the branch's comments (drop scale figures and restated-code lines).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* fix(pg-storage): batch insert_new_chunks; surface underlying DB errors
A single all-rows INSERT…SELECT over jsonb_to_recordset fails at scale — the
jsonb payload can exceed Postgres' 1 GB value limit, and one statement can trip
a statement_timeout (observed: ~30s then a generic "database error"). Insert in
10k-row batches inside one transaction so neither the payload nor a single
statement's runtime scales with the input; still atomic (any batch error rolls
the load back). ChunkAlreadyExists / dataset-not-found semantics unchanged.
Also render StorageError::Database with `{0:#}` so the underlying cause (timeout,
value-size, …) is shown instead of just the outermost context.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* fix(pg-storage): batch the full-ideal write phases to bound bind-array size
record_routing_diffs, commit_new_ideal, resolve_flip_flops and
mint_superseded_stale_mappings flattened the whole ideal into one statement's
bind arrays, which at large placements exceeds Postgres' wire/value limit and
drops the connection ("broken pipe"). Write in 10k-chunk batches instead.
record_routing_diffs / commit_new_ideal window by whole chunks so a chunk's
holder set never splits across batches (their array_agg / PK depend on it);
commit keeps a single trailing sweep-delete. All batches run inside the cycle's
transaction, so the phase stays atomic.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* perf(pg-storage): intern dataset names; drop the full chunks map
Two per-cycle allocation cuts from the storage review:
- chunk_from_row took a fresh Arc::new(dataset_name.clone()) per row — one
String allocation for each chunk despite only a handful of distinct dataset
names. It now takes a pre-interned Arc; a DatasetInterner shares one Arc per
name across a ...---------
Co-authored-by: Defnull <879658+define-null@users.noreply.github.com>
Stacked PR **2 of 2** — **base this on #37** (the instrumentation), so the diff here is only the SQL rewrites. Merge #37 first.
## What
Replace every per-row `.execute()` loop the instrumentation (#37) flagged with one set-based statement, eliminating the O(chunks) round-trips per cycle:
- **insert_new_chunks**: one bulk `INSERT…SELECT` via `jsonb_to_recordset`. A unique violation aborts the batch and is reported **generically** as `ChunkAlreadyExists` (no per-row culprit lookup — deliberate, to keep it one statement); an unknown dataset is detected by RETURNING-count mismatch.
- **update_worker_set**: one batch upsert over `UNNEST(peer_id[], version[]) … RETURNING id,(xmax=0)`.
- **record_routing_diffs / commit_new_ideal**: one `jsonb_to_recordset` INSERT / upsert (commit keeps its existing sweep-delete).
- **mint_superseded_stale_mappings / resolve_flip_flops**: flatten to parallel `bigint[]` pairs; one `INSERT…ON CONFLICT DO NOTHING` / one `DELETE…USING UNNEST`.
- **replay_confirmed_diffs**: fully server-side — `DISTINCT ON (chunk_pk) … ORDER BY worker_assignment_id DESC` for last-write-wins, one data-modifying CTE (upsert non-empty + delete empty); no rows round-trip to Rust.
`PhaseTimer` instrumentation retained; statement counts now reflect the batched queries.
## Measured (reshuffle-sim multistep, ~95K chunks)
Per-cycle, the five N+1 phases drop from ~95K statements/~85s combined to **1–2 statements/~5s**. End-to-end wall clock for `--chunks-per-step 1000 --steps 5`: **241s → 53s (~4.5×)** — and the gap widens at 10M scale and on a real networked DB (round-trip latency is exactly what batching removes; the bench used localhost `fsync=off`).
## Verification
Full feature-gated suite **220 passed / 0 failed** (incl. the #36 overcommit regression test, which drives the batched `insert_new_chunks`/`update_worker_set`); clippy clean; sim runs end-to-end.
Note: the sim's step-4/5 report differs run-to-run on **both** branches (pre-existing nondeterminism from unordered `SELECT`s feeding order-sensitive tie-breaks) — not introduced here.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: claude <claude@example.com>
Co-committed-by: claude <claude@example.com>
Resolve storage.rs conflicts: keep continuity fallback logic and upstream summary download retry constants. Drop ensure_chain_continuity in favor of advance_continuity. Co-authored-by: Cursor <cursoragent@cursor.com>
Author
|
@kalabukdima, please take a look. Thanks! |
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.
Closes #12
Right now a continuity gap in one dataset kills the whole scheduler run. This changes that so only the affected dataset falls back to its last known good chunks from ClickHouse.
When
ChunkStreamhits a gap while listing new S3 chunks, it logs a warning, sets a metric, stops loading more chunks for that dataset, and returns whatever contiguous prefix it already accepted. The run continues normally for everything else.Non-continuity errors (missing
blocks.parquet, S3 failures, summary download issues) still fail hard.Left
strict_continuity_checkalone for now — if fallback works, we shouldn't be feeding gappy chunks into assignment encoding anyway.Test plan
cargo testpassesadvance_continuitynext_batchreturningNoneafter fallback