Table partitioning optimizations in Parquet and other formats#14
Merged
Conversation
Adds the staged optimization plan (docs/parallel_partitions_opts.md) and a reusable lineitem verification script (scripts/verify_lineitem.py) that emits row count, min/max keys, grouped counts, exact numeric sums, and a stable ordered row hash for correctness gating of every later phase. Baseline (lineitem SF=5, zstd, zero-copy, --max-rows 0), median of 5 sequential runs: 27.33s wall, 29,999,795 rows, 1.366 GB, ~190 MB peak RSS. Output is byte-identical across runs; wall-time variance is WSL2 disk-stall.
Exposes the previously hard-coded RecordBatch size (DBGEN_BATCH_SIZE and the scattered 10000/8192 literals) as a single --batch-rows CLI option and threads it through every generation path. SF=5 lineitem sweep (median of 3 sequential runs): 65536 is 17.3% faster than the old 8192 default (17.35s vs 20.98s), the only size within 3% of the best. Sizes >=131072 regress on WSL2 due to larger write bursts. Batch size is output-invariant (identical ordered hash across batch sizes); full SF=5 output matches the baseline reference exactly. Also fixes a tuple-vs-list false mismatch in verify_lineitem.py's schema comparison and adds it as the reusable correctness gate.
…d Parquet output Phase 2 of docs/parallel_partitions_opts.md. Wrap the Parquet sink in an Arrow BufferedOutputStream (configurable, 0 = disabled). SF=5 lineitem sweep: 64 MiB fastest (16.56s) but 32 MiB (16.98s) captures nearly all the gain with lower memory, ~4% median improvement over unbuffered (17.64s). Output remains byte-identical (1,366,231,370 bytes, hash unchanged, 29,999,795 rows).
… path Reorder generate_all_tables_parallel() table queue to longest-processing-time- first. Measured neutral on WSL2 (disk-bandwidth-bound, -1.5% within noise) but kept as the theoretically correct zero-risk default; single-table path unchanged. See docs/parallel_partitions_opts.md Phase 3.
…capacity 1 Add --pipeline-queue-capacity: a bounded SPSC queue (BatchQueue) plus a PipelinedWriter decorator that runs a consumer thread for Parquet encode/compress/write while the producer thread continues generation and conversion. The zero-copy converter allocates self-owned Arrow buffers, so batches are lifetime-safe across the thread boundary. Preserves FIFO row order, bounds memory to O(capacity*batch), and propagates writer failures without deadlock (close wakes both sides). SF=5 lineitem: 17.53s -> 12.92s median (-26%) at capacity 1; capacity 1 is the smallest within 3% of the best, so it is the new default. Output is byte-identical to the sequential path; aggregates and ordered hash match the baseline reference. Peak RSS unchanged (~270 MB). Adds 6 pipeline unit tests (order, capacity bound, close-wakes-both, error propagation).
Capture exact legacy RNG boundary states so lineitem parent ranges can be generated independently without changing the retained logical dataset. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Record full-scale equivalence evidence and reopen the parallel partition phases now that the retained dataset can be partitioned exactly. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Fork isolated lineitem workers from one canonical partition plan so multi-file generation preserves exact row order while bounding process concurrency and publishing outputs atomically. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Document the exact SF=5 correctness gate and show that the serial snapshot-planning cost keeps parallel files opt-in despite faster worker execution. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Record that fixed partition policies are within measurement variance and that serial planning overhead prevents a defensible automatic policy. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Record the public Arrow metadata API limitation that prevents safely relocating all Parquet page and index offsets into one ordered file. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Record the final SF=5 profile showing dbgen text generation and compression dominate while row materialization and conversion account for only 3.88% of samples. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Record that the final profile does not justify changing the compatible int64 schema into an explicit narrow physical type mode. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Replace the remaining Phase 7 placeholder with its committed evaluation hash so every phase has a terminal documented result. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
SF=1 triage across 8 TPC-H tables x 6 formats and 24 TPC-DS tables x 6 formats, all 144+48 combinations succeeded with matching row counts. Fixed batch_pipeline_test stale-discovery issue by rebuilding the target. Measurement only, no production code changed.
Exposes the previously hard-coded 8192-row batch threshold in run_generation() as a validated CLI flag, covering all 24 TPC-DS tables through the shared dispatch path. Measured SF=5 store_sales and catalog_sales across 8192-131072; neither table's optimum clears the acceptance band by enough to justify a new shared default, so 8192 is kept and the flag is left available for per-invocation tuning.
Extend the Phase 4 bounded generation/write pipeline to the --parallel all-table path, which was missing it (only the single-table CLI path wrapped the writer). See docs/parallel_partitions_opts.md Phase 13.
Generalize the TPC-H Phase 4 BatchQueue/PipelinedWriter decorator to TPC-DS by wrapping the writer at both the single-table and per-table child call sites. Default capacity 1, matching TPC-H. See docs/parallel_partitions_opts.md Phase 13.
…iles Wraps only the internal Parquet FileOutputStream in a BufferedOutputStream when requested; manifest, snapshot, and OPTIONS/schema metadata files are untouched.
…files Wraps only the internal Parquet FileOutputStream in a BufferedOutputStream when requested; Avro manifest, manifest-list, and metadata JSON files are untouched.
Reuses --parquet-output-buffer-bytes to wrap Paimon/Iceberg's internal Parquet data-file streams, same mechanism as ParquetWriter.
New flag wraps each writer's internal Parquet data-file stream in a BufferedOutputStream, mirroring the TPC-H mechanism. Default 0 (disabled).
Paimon/Iceberg buffering added but rejected as default (within noise of no-op); CSV/ORC/Lance investigated and correctly left unchanged.
Small-first ordering tails the heaviest fact tables at the end of a constrained worker window, delaying their start. Largest-estimated-work-first (measured SF=5 single-process elapsed time) starts heavy tables immediately.
Phase 6's own overhead (~7.2s snapshot prepass) already made partitioned lineitem slower than compatibility mode; partsupp and flat TPC-H tables have far smaller baselines and would be dominated by the same fixed cost. TPC-DS has no snapshot/restore primitive at all, and its own jump-ahead facility shares the same divergence risk already rejected for TPC-H.
… writes Same per-child-process ring pattern already used for Parquet's zero-copy path; wired into both executables behind the existing --io-uring flag.
Mirrors lineitem's serial-partition snapshot mechanism for partsupp, whose fixed SUPP_PER_PART child count needs no per-parent counting pass, to re-measure Phase 16's partsupp rejection with real numbers instead of an extrapolation from lineitem's prepass cost. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…ment Phase 16 rejected partsupp partitioning by dividing lineitem's ~7.2s prepass overhead into partsupp's runtime, without ever measuring a partsupp-specific prepass. Built one and measured it directly at SF=10: prepass cost is ~3.7s (not lineitem's 7.2s), confirming the fixed-child-count hypothesis, but the corrected real numbers still show no repeatable win (2/2 regresses +14.5%; 4/4 is within baseline noise). Rejection stands, now on measured grounds.
Vendors Thrift-generated parquet::format types (header-only; libparquet.so already exports the same symbols, avoiding an ODR double-free from duplicating its static globals) to patch row-group byte offsets by a cumulative delta and merge independently-encoded partition mini-files into one ordered Parquet file without decoding or re-encoding payload bytes.
Mechanism works correctly (byte-identical merged output, full SF=1/SF=5 correctness via verify_lineitem.py) but total wall time is 86.5% slower than the single-process baseline, since it layers on top of Phase 6's already-slower partitioned generation. Accepted as a verified opt-in mechanism, rejected as default.
Owner
Author
|
@claude review |
There was a problem hiding this comment.
Pull request overview
This PR extends the project’s “parallel partitions” optimization work by adding bounded generation→write pipelining, configurable batch sizing and output buffering, deterministic dbgen partition support (including serial/parallel partition generation paths), and an opt-in Parquet mini-file footer merge prototype to restore single-file ordered output after partitioned encoding.
Changes:
- Added a bounded SPSC
BatchQueue+PipelinedWriterdecorator and wired it into TPCH/TPCDS generation paths. - Added/propagated CLI controls for batch sizing, pipeline queue capacity, and Parquet data-file output buffering (including Paimon/Iceberg data-file sinks).
- Added deterministic partition planning/generation support (lineitem + partsupp) and a prototype Parquet footer merge implementation with tests and licensing notice.
Reviewed changes
Copilot reviewed 23 out of 26 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| third_party/parquet_format/NOTICE.md | Documents provenance/licensing and regeneration rules for vendored Parquet Thrift types. |
| third_party/parquet_format/parquet_types.h | Vendored Thrift-generated Parquet format types used by the footer merge prototype. |
| third_party/parquet_format/parquet_types.tcc | Thrift read/write templates used by the footer merge prototype. |
| tests/parquet_footer_merge_test.cpp | Adds unit tests for ordered Parquet mini-file merge correctness. |
| tests/dbgen_batch_iterator_test.cpp | Adds partition-sequence correctness tests (including an opt-in full-scale digest gate). |
| tests/CMakeLists.txt | Registers new gtest executables for the pipeline and footer-merge tests. |
| tests/batch_pipeline_test.cpp | Adds unit tests for bounded queue + pipelined writer behavior. |
| src/writers/parquet_writer.cpp | Adds optional BufferedOutputStream wrapping for Parquet streaming writes. |
| include/tpch/parquet_writer.hpp | Exposes set_output_buffer_bytes() API for ParquetWriter. |
| src/writers/parquet_footer_merge.cpp | Implements Parquet mini-file merge by copying payload bytes and patching Thrift footer offsets. |
| include/tpch/parquet_footer_merge.hpp | Declares MergeParquetFiles() API. |
| src/writers/paimon_writer.cpp | Adds optional buffering and optional io_uring-backed Parquet data-file sinks. |
| include/tpch/paimon_writer.hpp | Exposes buffering and io_uring toggles for PaimonWriter’s internal Parquet data files. |
| src/writers/iceberg_writer.cpp | Adds optional buffering and optional io_uring-backed Parquet data-file sinks. |
| include/tpch/iceberg_writer.hpp | Exposes buffering and io_uring toggles for IcebergWriter’s internal Parquet data files. |
| src/tpcds_main.cpp | Adds batch sizing / pipeline capacity / output buffering flags and updates table scheduling order. |
| README.md | Documents new tpcds_benchmark --batch-rows option. |
| src/main.cpp | Adds TPCH flags for batch/pipeline/buffering and adds partitioned parallel-files execution paths. |
| src/dbgen/dbgen_wrapper.cpp | Adds seed snapshot semantics and partition planning/generation + partition batch iterators. |
| include/tpch/dbgen_wrapper.hpp | Declares partition descriptor types and partition batch iterator APIs. |
| src/dbgen/dbgen_stubs.c | Adds seed state copy/set helpers; makes seed snapshot capture idempotent. |
| include/tpch/batch_pipeline.hpp | Introduces BatchQueue and PipelinedWriter implementation. |
| scripts/verify_lineitem.py | Adds an independent correctness gate tool for deterministic aggregates/hashes over lineitem Parquet outputs. |
| docs/parallel_partitions_opts.md | Adds/updates extensive design + measurement notes for the optimization plan and phases. |
| CMakeLists.txt | Links thrift and wires the footer-merge source and include paths into tpch_core. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+91
to
+105
| void ValidateSchemaMatches(const parquet::format::FileMetaData& reference, | ||
| const parquet::format::FileMetaData& candidate, | ||
| const std::string& path) { | ||
| if (reference.schema.size() != candidate.schema.size()) { | ||
| throw std::runtime_error("MergeParquetFiles: schema mismatch (field count) in " + path); | ||
| } | ||
| for (size_t i = 0; i < reference.schema.size(); ++i) { | ||
| const auto& a = reference.schema[i]; | ||
| const auto& b = candidate.schema[i]; | ||
| if (a.name != b.name || a.type != b.type || a.__isset.type != b.__isset.type) { | ||
| throw std::runtime_error("MergeParquetFiles: schema mismatch at field " + | ||
| std::to_string(i) + " in " + path); | ||
| } | ||
| } | ||
| } |
Owner
Author
There was a problem hiding this comment.
Fixed in b3fbdfa: ValidateSchemaMatches now uses Thrift's generated SchemaElement::operator== (covers repetition_type, logicalType, converted_type, field_id, precision/scale, num_children, all respecting __isset) instead of the name/type-only check.
Comment on lines
+183
to
+192
| parquet::format::FileMetaData merged; | ||
| merged.__set_version(files[0].metadata.version); | ||
| merged.__set_schema(files[0].metadata.schema); | ||
| if (files[0].metadata.__isset.column_orders) { | ||
| merged.__set_column_orders(files[0].metadata.column_orders); | ||
| } | ||
| if (files[0].metadata.__isset.created_by) { | ||
| merged.__set_created_by(files[0].metadata.created_by); | ||
| } | ||
|
|
Owner
Author
There was a problem hiding this comment.
Fixed in b3fbdfa: the merged footer now copies key_value_metadata from the first input file when present (preserves Arrow's ARROW:schema entry and any other user metadata). encryption_algorithm/footer_signing_key_metadata weren't added since none of this prototype's writers produce encrypted Parquet files (no encryption support exists anywhere in this codebase); happy to add pass-through for those too if there's a concrete use case, but didn't want to add untested code paths for a feature that isn't otherwise supported.
Comment on lines
+115
to
+116
| std::this_thread::sleep_for(std::chrono::milliseconds(50)); | ||
| EXPECT_EQ(pushed.load(), 2); // filled capacity, blocked on 3rd |
Owner
Author
There was a problem hiding this comment.
Fixed in b3fbdfa: replaced the fixed 50ms sleep with a bounded poll (WaitUntil helper, up to 5s, checking every 2ms) on pushed.load() == 2. Should no longer flake on slow/loaded CI.
Comment on lines
+120
to
+121
| std::this_thread::sleep_for(std::chrono::milliseconds(50)); | ||
| EXPECT_EQ(pushed.load(), 3); |
Owner
Author
There was a problem hiding this comment.
Fixed in b3fbdfa, same WaitUntil polling helper applied to the pushed.load() == 3 assertion after draining one batch.
| fs::path dir_; | ||
|
|
||
| void SetUp() override { | ||
| dir_ = fs::temp_directory_path() / fs::path("parquet_footer_merge_test_" + std::to_string(::testing::UnitTest::GetInstance()->random_seed())); |
Owner
Author
There was a problem hiding this comment.
Fixed in b3fbdfa: temp dir name now includes PID + a static atomic counter + a nanosecond steady_clock timestamp, not just gtest's random_seed(). Should be unique across parallel ctest binaries and repeated runs.
Owner
Author
|
@codex review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1b46b8db10
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| for (size_t i = 0; i < reference.schema.size(); ++i) { | ||
| const auto& a = reference.schema[i]; | ||
| const auto& b = candidate.schema[i]; | ||
| if (a.name != b.name || a.type != b.type || a.__isset.type != b.__isset.type) { |
There was a problem hiding this comment.
When inputs have identical field names and physical types but differ in repetition, logical/converted type, field IDs, or nesting, this check accepts them and the merged footer adopts the first file's schema. The later row groups can then be decoded using incompatible level or logical-type metadata, silently corrupting results; compare the complete SchemaElement representation (or reject all non-identical schemas) before copying payloads.
Useful? React with 👍 / 👎.
Owner
Author
There was a problem hiding this comment.
Fixed in b3fbdfa: same fix as the Copilot comment on this file — ValidateSchemaMatches now compares full SchemaElement equality via Thrift's generated operator== rather than just name/type.
Comment on lines
+184
to
+185
| merged.__set_version(files[0].metadata.version); | ||
| merged.__set_schema(files[0].metadata.schema); |
There was a problem hiding this comment.
Every successful merge drops the first input's key_value_metadata because the new footer only copies version, schema, column orders, and creator. This loses user metadata and, for files written with Arrow schema storage enabled, can discard the ARROW:schema payload that preserves Arrow-specific schema details such as field metadata and dictionary types; copy the reference footer's key-value metadata into merged.
Useful? React with 👍 / 👎.
Owner
Author
There was a problem hiding this comment.
Fixed in b3fbdfa: same issue as the Copilot comment on this file — the merged footer now preserves key_value_metadata from the first input, which covers the ARROW:schema case you flagged.
Phase 9: wire up the dormant Buffer::Wrap "true zero-copy" lineitem converter via --true-zero-copy. Fixes a use-after-free in the dormant code (PipelinedWriter wrap defeats the ParquetWriter* dynamic_cast, dropping the batch's lifetime manager before a background thread reads the wrapped buffers). Measured -0.03% vs standard converter at SF=5 -- confirms the original profile-based rejection. Phase 10: add --narrow-keys, an opt-in int32 schema for lineitem's four key columns (orderkey/partkey/suppkey/linenumber), with an overflow guard for scale factors where keys exceed INT32_MAX. Measured -9.98% at SF=5 across 10 interleaved runs -- a real, previously-undetected Parquet-encoding-size effect the CPU-only profile couldn't see. Kept opt-in since it changes the public schema.
Visualizes the Phase 1/2/4 default-path speedup (27.33s -> 12.96s, -52.6%, SF=5 lineitem) as a waterfall, plus every other measured idea (Phase 3/6/8/9/10/13/15/16/17) as a color-coded delta chart. All numbers pulled directly from docs/parallel_partitions_opts.md. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
- ValidateSchemaMatches now compares full SchemaElement equality (repetition_type, logicalType, field_id, etc.) instead of just name/type, so merges no longer accept mismatched schemas that could silently corrupt row-group decoding (Copilot + Codex P1). - Preserve key_value_metadata in the merged footer, so Arrow's ARROW:schema entry (field metadata, dictionary types) survives the merge instead of being silently dropped (Copilot + Codex P2). - batch_pipeline_test: replace fixed-sleep assertions with a bounded polling helper to remove CI flakiness under scheduler jitter. - parquet_footer_merge_test: make the per-test temp directory unique via pid+counter+timestamp instead of gtest's random_seed, which can collide across parallel ctest binaries.
Owner
Author
|
Addressed all review feedback from Copilot and Codex in b3fbdfa:
Verified: full rebuild, |
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
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.
Summary
Staged, measured optimization plan for TPC-H/TPC-DS generation throughput (
docs/parallel_partitions_opts.md), executed phase by phase with a hard correctness gate (independent PyArrow read, schema, aggregates, ordered row hash) before any performance claim is accepted. Every phase below was actually implemented and benchmarked — nothing is estimated or assumed.Ideas tried, briefly
arrow::io::BufferedOutputStreamto coalesce small writes.--parallelpath and all of TPC-DS.lineitemgeneration into N deterministic partitions (seed-snapshot-per-partition, since dbgen's ownset_state()jump-ahead doesn't reproduce this embedded build's RNG stream).Buffer::Wrap) instead of copying.partsuppand TPC-DS fact tables.AsyncIOContextmechanism.Results
Buffer::Wrap, real code + benchmark)--true-zero-copykept (byte-identical, harmless)--narrow-keys; kept off by default (changes public schema)--batch-rowstuningorders≈ −10%; TPC-DSstore_sales≈ −25 to −30%--parallel-tables 4; neutral at 8partsupp/TPC-DSNet effect on the stated objective (full
lineitemSF=5, single-process default path): Phases 1+2+4 together move the median wall time from 27.33s to 12.96s — roughly a 2.1x speedup (−52.6%), verified byte-identical/hash-identical to the original reference at every step. Multi-table scheduling and pipeline fixes (Phases 3, 13, 15) add smaller, workload-dependent gains (0-30%) on top of that for constrained-concurrency or TPC-DS scenarios. Every idea aimed at intra-table parallelism (partition files, ordered merge,partsupp/TPC-DS partitioning) failed to beat the already-fast single-process path in this environment — a fixed per-run planning/prepass cost (7.2s for lineitem, 3.8s for partsupp) and WSL2's virtio-blk disk bottleneck consistently ate any parallel throughput gain. Phase 10's narrow-keys mode is the one exception among the opt-in, non-default mechanisms: it is a real, measured ~10% win, kept opt-in only because it changes the public Parquet schema, not because it failed to perform.Note on #9/#10: both were originally evaluated by profiling only. Both have since been implemented for real (
--true-zero-copy,--narrow-keys) and benchmarked at SF=5 with full correctness gates (byte-identical/ordered-equal output,ctest65/65). #9's profiling-based rejection held up under real measurement (−0.03%, no effect). #10's profiling-based rejection did not hold up — the real implementation is ~10% faster, because narrowing shrinks Parquet's encoded page bytes (a size effect the CPU-only profile couldn't see), not because it reduces conversion CPU time.Test plan
ctestsuite green after every accepted phasescripts/verify_lineitem.py(schema, aggregates, grouped counts, ordered SHA-256 hash) passes against the SF=5 baseline for every accepted default changeTable.equals()ordered comparison for every partition/merge experimentctest65/65, interleaved-pairs benchmark protocol for Phase 10's ambiguous initial sweep