Repository navigation
persist: decode fetched parts in bounded arrow batches - #39587
Conversation
mtabebe
left a comment
mtabebe
left a comment
There was a problem hiding this comment.
Overall this looks good to me. Thank you for the measurement details in the linear ticket.
I had a few nits and a clarification question for my understanding.
| /// Incremental decoder over a parquet-encoded [`BlobTraceBatchPart`]. | ||
| /// | ||
| /// Decodes at most `batch_rows` rows per call to [`Self::next_updates`], so | ||
| /// only the encoded bytes and one decoded batch are resident at a time. Each |
There was a problem hiding this comment.
It might be helpful to explain why we want this property in the comment. That is:
(1) we want to unpack data as we go so that we don't need to unpack the whole part at once and keep all that in memory, thereby reducing OOM when holding multiple parts
(2) we output in order and are contiguous, which matters for merging adjacent tuples. Out of order would mean under merging rows? I believe the invariant comes from this comment here:
// We enforce an invariant that we have a single RowGroup.
But it isn't clear that the caller can rely on that behaviour
There was a problem hiding this comment.
Added a paragraph to the BlobTraceBatchPartReader doc: successive batches cover the part's rows in encoded order without gaps or overlap, so a caller can carry its pending update across a boundary and merge the same updates as on the whole part. The reader walks row groups in order, so this holds independently of the single-row-group invariant. The memory motivation is in the first paragraph. Ordering affects only how much adjacent consolidation happens, not correctness, since FetchedPart consolidation is best-effort.
(Reply by Claude Code.)
| 0, | ||
| "\ | ||
| Maximum number of rows of a fetched blob part that a reader decodes at \ | ||
| once. 0 decodes the whole part at once.", |
There was a problem hiding this comment.
Nit, but 0 as the whole part at once feels weird to me. I know this is a standard way of doing things but it always gives me the ick ... Can we make it an Config<Option> where None defaults to the old behaviour?
There was a problem hiding this comment.
Done, it's now Config<Option<usize>> with None (empty string in SQL) as whole-part decoding. Some(0) behaves like 1, and the test covers it.
(Reply by Claude Code.)
| @@ -1123,21 +1344,31 @@ where | |||
| let mut consolidated = self.peek_stash.take(); | |||
There was a problem hiding this comment.
Just so I make sure I understand: this is what is holding the state of the row being read: either we accumulate into this, or we create a new row. And this is why we need the ordering property
There was a problem hiding this comment.
Yes. consolidated holds the owned, decoded update being accumulated, and peek_stash holds the first update that didn't merge, which seeds the next call. Both are owned values, so they survive the decoded batch being replaced. Consolidation only merges adjacent equal (kv, t) pairs, so it relies on rows arriving in order and contiguously. If they arrived out of order, it would merge less, but the result would still be correct.
(Reply by Claude Code.)
| /// Reads the shard's single hollow part as of `as_of`, with and without | ||
| /// batched decoding, through both `snapshot_and_stream` and the | ||
| /// `BatchFetcher`/`FetchedBlob` path used by `persist_source`. | ||
| async fn assert_batched_reads( |
FetchedPart decoded a hollow part's single parquet row group into one arrow batch and kept it alive until the last row was emitted. With `persist_part_decode_batch_rows` set, readers that go through FetchedPart (persist_source via FetchedBlob::parse, Listen, snapshot_and_stream) keep the encoded bytes and decode, validate, and normalize at most that many rows at a time, so one decoded batch per part is resident. Consolidation via the peek stash carries across batch boundaries. Inline parts and EncodedPart consumers (compaction, consolidating iteration, inspect) still decode whole parts. The flag defaults to 0, which keeps whole-part decoding, and CI runs with 16384. CPU-296 Co-Authored-By: Claude Opus 5.5 <[email protected]>
Decode whole parts through `BlobTraceBatchPartReader`, so the format dispatch exists once. Pick whole or batched decoding of a hollow part in `PartSource::from_hollow_blob`, keep a `FetchedPart`'s current batch and its cursor in one `DecodedRows`, and read all `BatchFetcherConfig` values at fetch time. Add a test for batched decoding of a part with a rewritten description. Co-Authored-By: Claude Opus 5.5 <[email protected]>
Unset keeps whole-part decoding, which replaces the 0 sentinel. Document that reader batches cover a part's rows in order without gaps, which lets callers carry consolidation across batch boundaries. Co-Authored-By: Claude Opus 5.5 <[email protected]>
e23c16c to
99c8d23
Compare
|
Thanks for the review! |
FetchedPartdecodes a hollow part's single parquet row group into one arrow batch and keeps it alive until the last row is emitted. During hydration, each worker therefore holds a whole decoded part (persist targets 128 MiB encoded parts), and more than one when parts queue inpersist_source.With the new dyncfg
persist_part_decode_batch_rowsset, readers that go throughFetchedPartkeep the encoded bytes and decode, validate, and normalize at most that many rows at a time. These readers arepersist_source(viaFetchedBlob::parse), Listen, andsnapshot_and_stream. Consolidation through the peek stash carries across batch boundaries. Inline parts andEncodedPartconsumers (compaction, consolidating iteration, inspect) still decode whole parts. The flag defaults to unset, which keeps today's behavior, and CI runs with 16384 (plus unset and 7 as variants).Whole-part decoding now goes through the same
BlobTraceBatchPartReaderwith a batch size of the part's row count, so both paths share one format dispatch. As a result, a part with no rows decodes to empty updates instead of failing.Two new tests,
fetch::tests::batched_part_decodeandfetch::tests::batched_part_decode_ts_rewrite, read one 10-row hollow part unbatched and with batch sizes 0, 1, 3, 4, 10, and 16384 through both read paths. They cover duplicates that merge across a batch boundary and a pair that cancels across one. The second test reads a part whose description was rewritten, so every batch is truncated and has its timestamps rewritten.Measured on a TPC-H SF100
lineitemindex hydrating on an 8 GiB, 8-worker replica (r8gd.8xlarge, file-backed buffer pool, lgalloc off, fetch semaphore at 0.1): whole-part decode was OOM-killed, and 16384-row batches hydrated in 303 s. In heap profiles, the roughly 1.1 GB of parquet decode output no longer appears. Details are in CPU-296.CPU-296
🤖 Generated with Claude Code