[core] Optimize ordinary manifest merging with raw Avro blocks - #9213
[core] Optimize ordinary manifest merging with raw Avro blocks#9213leaves12138 wants to merge 9 commits into
Conversation
| // retaining planned raw blocks before the single ordered writer consumes them. | ||
| if (hasDeletes | ||
| && deletes.useRowIdFilter() | ||
| && manifestReadParallelism != null |
There was a problem hiding this comment.
[P2] Please preserve the scan.manifest.parallelism contract here. The documented default is the number of CPU processors, but null now disables parallel planning. In addition, add-only and non-RowID merges fall through to the serial manifest loop even when the option is explicitly greater than 1. The legacy merger passes the configured value, including null, to ManifestReadThreadPool, which resolves the default and performs bounded parallel reads. Because this merge runs on the synchronous commit path, serial object-store reads and decompression can materially increase commit/checkpoint latency. Could we keep bounded parallel manifest reads for all optimized paths and restore a concurrency assertion covering both the default and an explicit parallelism?
There was a problem hiding this comment.
Thanks. I fixed the null/default contract for DELETE collection and the existing RowID planning path: null now reaches ManifestReadThreadPool and uses the default processor count, while an explicit value of 1 remains serial. I benchmarked widening planning to add-only manifests on snapshot 4780 (22,614,974 ADDs, 0 DELETEs, about 582 MiB of manifests). Across five minor-compaction runs, streaming raw-block copy had a 448 ms median and 198 MiB peak RSS; 8-thread planning had a 506 ms median and 341 MiB peak RSS. Full compaction showed the same direction (about 270-282 ms / 200 MiB versus 372-380 ms / 323 MiB). Compatible add-only manifests already copy compressed Avro blocks without decompression or filtering, so parallel planning only makes stable block copies, retains multiple plans before the single ordered writer, and adds executor overhead. I therefore kept add-only on the original streaming path. Added concurrency assertions for both default DELETE collection and default RowID planning; 108 targeted Manifest tests pass.
There was a problem hiding this comment.
Thanks, the RowID/default fix looks good, and the add-only benchmark is convincing; keeping compatible add-only manifests on the streaming raw-copy path makes sense. One case still seems uncovered: when DELETEs are present and useRowIdFilter() is false, the code still falls through to the serial loop even when scan.manifest.parallelism is null or explicitly greater than 1. This affects legacy or non-data-evolution manifests and requires decoded identifier filtering, so the add-only benchmark does not cover it. Could we add a separate bounded raw-block prefetch path only for non-RowID + DELETE compaction? Each worker would open one reader, stableCopy() the compressed blocks, and close the reader; the coordinator would then consume plans in input order and perform identifier filtering and all writer, matchedEntries, and emittedDeletes mutations on a single thread. This preserves the ADD/DELETE state machine and keeps memory bounded to one manifest per planning worker, while leaving add-only streaming unchanged. A blocking FileIO test for both default and explicit parallelism would cover the remaining contract.
There was a problem hiding this comment.
Implemented in 07e369e. There is now a separate bounded path for DELETEs with useRowIdFilter() == false when the configured/default manifest parallelism permits parallel reads. Each worker only opens one ManifestAvroReader, stableCopy()s its compressed raw blocks, and closes the reader. The coordinator consumes the prefetched manifests in input order and exclusively owns Avro decoding, identifier filtering, the writer, matchedEntries, and emittedDeletes. Explicit parallelism 1 and single-manifest inputs still stream directly; compatible add-only compaction remains on the original streaming raw-copy path.
I added a blocking FileIO test parameterized for both default (null) and explicit parallelism 2. It verifies two manifest reads overlap and validates the exact surviving entries, including the optional-manifest rewrite/keep behavior. The targeted Manifest suite passes: 110 tests, 0 failures, 1 existing skip.
For a production-scale stress check, I routed the snapshot 8816 payload through the non-RowID identifier path by removing only the meta-level RowID ranges and forcing the selected manifests to rewrite. This is a stress test of the new branch, not the normal RowID-aware 8816 path (which remains about 6.8 s). On dev2 local/page-cached storage, identifier p1 was 43.21 s / 1.22 GiB peak RSS and p8 was 41.48 s / 1.55 GiB; current Legacy p8 was 46.07 s / 16.25 GiB. The modest local-time gain is expected because only compressed-block FileIO is prefetched, while decode/filter/write stays ordered and single-threaded; the main demonstrated gain is bounded memory, with additional latency hiding expected for object storage.
There was a problem hiding this comment.
Thanks. I replaced the prefetch-only implementation with the stronger block-aware design in b728430. All DELETE compactions now use the same bounded parallel manifest planner when the configured/default parallelism permits it, including the non-RowID identifier path. Each worker owns its identifier scratch and block-statistics objects and only reads immutable DELETE state. Unchanged schema-compatible blocks produce EncodedBlockMeta and are copied as compressed Avro blocks; blocks containing a DELETE or a deleted ADD are decoded and filtered by the coordinator in input order. The coordinator remains the sole owner of the writer, matchedEntries, and emittedDeletes. Compatible add-only compaction remains on the faster streaming raw-copy path.
For blocks without RowID, EncodedBlockMeta keeps its original primitive constructor and uses -1/-1 as the no-RowID sentinel, so there is no boxed overload or binary signature change. I added coverage for raw block metadata without RowIDs and retained the blocking FileIO concurrency test for both default and explicit parallelism 2. Formatting/checks pass, and 56 directly related manifest tests pass.
On dev2, the production-scale snapshot 8816 stress path (26,670,786 ADDs and 11,410 DELETEs, forced through non-RowID identifier matching) completed the merge in 7.98 s with 1.66 GiB peak RSS. The earlier prefetch-only p8 result was 41.48 s / 1.55 GiB, while current Legacy p8 was 46.07 s / 16.25 GiB. Full verification read all 26,659,376 output entries and exactly matched the Legacy fingerprints: b7ef9bb1f8347726, 45f036e957cd88ce, f7b77d2fd5a0b1c1, a0f943ec46d9a7ef.
|
+1 |
2 participants
Purpose
Optimize ordinary (non-sorted) manifest full and minor compaction without materializing every manifest entry.
The legacy merger decodes complete
ManifestEntryobjects, builds an identifier-to-entry map, and rewrites every surviving entry. On large data-evolution tables this creates high allocation pressure and makes manifest merging a major part of commit latency.This PR introduces a block-aware ordinary merger which:
ManifestFileLegacyMerger, selectable withmanifest.merge-optimize.enabled=false.This PR only changes ordinary manifest full/minor compaction. The manifest sort-compaction branch and
ManifestFileSorterare unchanged.Correctness validation
Benchmarks used real metadata from a big table. Each mode was warmed up once and then measured without an
-Xmxlimit. Output manifests were scanned and compared with an order-independent content fingerprint containing entry kind, file identifier and first RowID. Every legacy/optimized pair produced the same entry count, ADD/DELETE counts, and content fingerprint.Minor compaction
Full compaction
Peak RSS covers the whole benchmark JVM, including warm-up and output verification, so it is intentionally more conservative than measuring the merge allocation alone. Different output byte sizes are expected because raw-block reuse changes Avro block boundaries/compression layout; logical contents matched in all six comparisons.
Tests
Manifest read parallelism
The documented
scan.manifest.parallelismdefault is preserved: a null value now reachesManifestReadThreadPool, so DELETE collection and the existing RowID planning path use the default number of processors. An explicit value of1remains serial.I also evaluated extending bounded parallel planning to add-only manifests. Snapshot 4780 contains 22,614,974 ADD entries, no DELETE entries, and about 582 MiB of manifests. Five add-only minor-compaction runs produced these medians:
The parallel version was about 13% slower and used about 72% more peak RSS. Full compaction showed the same direction: the streaming path took about 270-282 ms at about 200 MiB RSS, while all-path parallel planning took about 372-380 ms at about 323 MiB RSS. Logical output counts and bytes were identical.
The reason is that a compatible add-only manifest already follows the cheapest path: its compressed Avro blocks are copied directly without decompression or entry filtering. Parallel planning adds no useful CPU work; it only creates stable copies of raw blocks, retains multiple manifest plans until the single ordered writer consumes them, and adds executor scheduling overhead. Therefore add-only compaction intentionally keeps the streaming raw-block-copy path instead of forcing parallel planning.