Parallelize reads from a single Parquet file in StorageFile, again - #104431
Parallelize reads from a single Parquet file in StorageFile, again#104431alexey-milovidov wants to merge 86 commits into
Conversation
|
Workflow [PR], commit [0b62bd5] Summary: ❌
AI ReviewSummaryThis PR reintroduces parallel reads for a single local Parquet file in Findings
Final VerdictLLVM Coverage Report
Changed lines: Changed C/C++ lines covered: 652/724 (90.06%) · Uncovered code |
clickhouse-gh
Bot
added
the
pr-not-for-changelog
…t-104251-parquet-single-file-parallelism
…t-104251-parquet-single-file-parallelism
|
Good: the newly added performance tests prevent this change: 
|
…t-104251-parquet-single-file-parallelism
…t-104251-parquet-single-file-parallelism
![clickhouse-gh[bot]](https://avatars.githubusercontent.com/in/928874?s=60&v=4){kind=small}
…-file-parallelism
When a single Parquet file is split into multiple bucketed sources by `StorageFile` (the path re-introduced in this PR), the file-level count cache must be bypassed: it is keyed by file path, so consulting or writing it from the bucketed read path would have every source report the file's full row count and multiply the result by the number of buckets. Addresses review feedback on PR #104431 asking for explicit regression coverage of this invariant.
Three gates on the bucketed single-file read path in `StorageFile`, addressing the `clickbench_parquet_short` regression observed on ARM in the CI of #104431 (#104431) and an open review comment. 1. `parallelize_output_from_storages = 0` now disables the split. The per-bucket sources are exactly the kind of read parallelism the setting's contract is about, but the existing check fired only after the sources had been created. Review feedback from #104431 (comment). 2. `need_only_count` queries skip the split. They consult only the file's metadata, so splitting them across N sources just multiplies the metadata-parse cost N-fold without any read-side benefit. This was the largest single contributor to the `Q1` (`SELECT COUNT(*)`) regression in `clickbench_parquet_short`. 3. The Parquet splitter (`ParquetBucketSplitter::splitToBucketsByCount`) now requires each chunk to cover at least 8 row groups. For a file with a small number of row groups, parallelising across all available threads multiplies per-bucket metadata-parse and prefetcher-setup overhead without giving each source enough work to amortise it. Large files (many row groups) still get full parallelism. Updates `02725_parquet_preserve_order.reference`: the 2-row-group test file falls below the new row-group floor so the pipeline goes back to a single `File` source followed by `Resize 1 → 2`, matching pre-#104251 behaviour. CI report: https://s3.amazonaws.com/clickhouse-test-reports/json.html?PR=104431&sha=78ecefef098eedd55ab0a0ce350082364c8c23be&name_0=PR&name_1=Performance%20Comparison%20%28arm_release%2C%20master_head%2C%201%2F6%29
|
However, there is still a chance to speed it up: https://benchmark.clickhouse.com/#system=+curp|H(u,s&type=-&machine=-ca4e|6t|ca2|ae-|g4e|6ax|6ale|3al&cluster_size=-&opensource=+s&hardware=+c&tuned=+n&metric=hot&queries=- |
|
From now on, we will be interested in accelerating those few queries that show the difference between the "single" and "partitioned" variants. If we can make it without degradations of other queries, we can merge this PR. |
… single-file split This test pins the contract requested in #104431 (comment): when `parallelize_output_from_storages = 0` is set, the single-file Parquet split path in `StorageFile` must not fan out into multiple per-bucket sources, even if `max_threads > 1` and the file has enough row groups to otherwise be split. The existing `04230_parquet_single_file_parallel_count.sql` only exercises `parallelize_output_from_storages = 1`. The previous regression was specifically that `parallelize_output_from_storages = 0` was ignored for this branch, so a dedicated test is needed to prevent silent regressions. The test creates a 40-row-group Parquet file (well above the 8-row-groups-per-chunk floor in `ParquetBucketSplitter`) and uses `EXPLAIN PIPELINE` to assert that with `parallelize_output_from_storages = 0` there is no `File ×` multiplier and no `Resize` step, and with `parallelize_output_from_storages = 1` the file IS split (the pipeline contains `File × N`). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…t-104251-parquet-single-file-parallelism
|
@groeneai, three stateless
Same exception across all three: |
|
@groeneai, the The exact same failure mode hit PR #104445 on the same day with a different version pair ( |
`clickbench_parquet_short` on the synthetic 20-row-group file still regressed 18-37% on `arm_release` even after the `8-row-groups-per-chunk` floor in `ParquetBucketSplitter`: 20/8 = 2 chunks, and each extra bucket added ~1-3 ms of per-source setup (metadata parse, prefetcher init, schema conversion, hyperrectangle build, bloom-filter init) on top of the ~5 ms baseline. For "short" queries on small files the parallelism win does not pay for that fixed cost. Bump the floor to 16. The synthetic test now reads with a single source (20/16 = 1 chunk), while `04238_parquet_single_file_parallelize_output_off` still exercises a 2-chunk split (40/16 = 2) and a real `hits.parquet` (hundreds of row groups) still fans out up to `max_threads`. CI report: https://s3.amazonaws.com/clickhouse-test-reports/json.html?PR=104431&sha=5f02d3d3200c9d6572d8848eb639c034a0fc0c62&name_0=PR&name_1=Performance%20Comparison%20%28arm_release%2C%20master_head%2C%201%2F6%29 PR: #104431 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
@alexey-milovidov — confirmed, all three failures ( Status: root cause identified, short-term fix already merged
CIDB verification — WasmEdge MSan SystemLog flush timeouts on master, hourly (UTC):
Master goes to zero immediately after #104971 merges at 08:57Z. PR branches predate the fix and still hit it (expected — branches don't auto-rebase). Failure family count over 14 days: 14 distinct test names, ~720 hits across 200+ PRs — the 3 hits on this PR are part of that. For this PR: the head commit Let me know if you'd like me to file a follow-up issue once Algunenano re-enables v3 in #105019 (to track whether the proper fix holds up), or treat the family as resolved once the master baseline stays clean for 48-72h. — @groeneai (session |
After raising the per-chunk row-group floor to 16 in `ParquetBucketSplitter`, the test file (`numbers(1000)` at row-group size 50 = 20 row groups) stayed below the 2-chunk threshold (`32` row groups), so the bucketed read path was no longer exercised — the test was no longer a regression guard for the count-cache-on-bucketed-source bug it was added to pin. Bump the file to 64 row groups (`numbers(3200)`) so the splitter produces 4 chunks and the bucketed branch is actually hit. The asserted counts are updated to match. This addresses the open review comment on #104431 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…ache The single-file Parquet split path in `StorageFile` parses the file's metadata once in `ParquetBucketSplitter::splitToBucketsByCount` and then the per-bucket `ParquetV3BlockInputFormat` sources each parse it again in `getFileMetadata`. The splitter's parse went through Arrow's `parquet::ReadMetaData` (uncacheable) while the source's parse went through `Parquet::Reader::readFileMetaData` keyed in `ParquetMetadataCache` by `(file_path, etag)` — so the two paths could not share work. For "short" queries on a single Parquet file (`clickbench_parquet_short` on the 828 MB synthetic ARM run), the extra footer parse cost 1-1.5 ms on top of a 5-8 ms baseline — 12-30 % of the runtime — and remained after raising the per-chunk row-group floor to 16 because the splitter still parsed metadata before deciding not to split. Route the splitter through the same cache: * `ParquetBucketSplitter::splitToBucketsByCount` now uses `Parquet::Reader::readFileMetaData` via a `Parquet::Prefetcher` (the same code path `NativeParquetSchemaReader::initializeIfNeeded` uses), producing a `parquet::format::FileMetaData` storable in `ParquetMetadataCache`. * New `splitParquetFileWithCache` helper accepts a cache pointer and the same `(file_path, etag)` key the source builds (sub-second mtime + inode + size), populating the cache on miss and reusing the parse on hit. * `ReadFromFile::initializePipeline` calls the cache-aware helper for Parquet, so the per-bucket sources hit the cache and skip the footer parse. The single-bucket case (no fan-out) benefits too: the surviving single source still hits the cache. Net effect: cold cache → 1 parse total (same as master); warm cache → 0 parses on both paths. CI report: https://s3.amazonaws.com/clickhouse-test-reports/json.html?PR=104431&sha=3c70a3490a9353117f2ecc65ca1458dd7285378c&name_0=PR&name_1=Performance%20Comparison%20%28arm_release%2C%20master_head%2C%201%2F6%29 PR: #104431 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
@groeneai, two CI failures in this PR look unrelated to the Parquet changes — please verify and link any in-progress fix here, otherwise file as separate issues.
|
|
@alexey-milovidov — confirmed, both failures are unrelated to the Parquet changes in this PR. Cross-linking: 1. The assertion shape ( CIDB (30-day): only 4 hits across 4 distinct PRs, 0 on master. No in-progress fix PR. Failures are sparse and re-distributed across unrelated PRs, so the underlying race is still present at very low rate (~0.1%) even after #78474's cleanup. Suggest reopening #78474 (or filing a follow-up) rather than a fresh issue — I can do that next worker run. SELECT toStartOfDay(check_start_time) AS day, count() AS hits, count(DISTINCT pull_request_number) AS prs
FROM default.checks
WHERE check_start_time > now() - INTERVAL 30 DAY
AND test_name LIKE '%test_keeper_force_recovery%cluster_recovery%'
AND test_status IN ('FAIL','ERROR')
GROUP BY day ORDER BY day DESC
-- 2026-05-15: 1 hit (this PR) 2026-05-11: 1 (different shape) 2026-04-19: 2 hits2. The failing shape on
Fixes in flight covering the family:
|
LLVM Coverage Report
Changed lines: Changed C/C++ lines covered: 517/569 (90.86%) · Uncovered code |
…on cache The previous commit extended the `isFileCacheVersionSettled` fail-close rule from the query condition cache to the format metadata cache, on both the per-source path (`object_with_metadata`) and the split-decision path in `ReadFromFile`. That was too broad and regressed master's `04207_parquet_metadata_cache_local_file`, which is reproducibly red on every sequential stateless shard: a file is written and queried right away, so it is never settled, the cache is bypassed, and the expected hit/miss pattern collapses to all zeroes. The two caches draw different conclusions from the same token. The query condition cache skips whole row groups without reading them, so a token that cannot yet prove a rewrite must fail close - that gate stays. The format metadata cache only reuses a parsed footer, and reusing it across an unsettled token is the behaviour master already has and pins with `04207`; a freshly written file is the common case, and bypassing the cache for it reparses the footer on every query. The split decision taken from such a footer is also not silently wrong: the read re-stats the file and throws `FILE_CHANGED_WHILE_READING` when the token moved under it. Also tag `04637_parquet_file_engine_bucketed_query_condition_cache` `no-parallel`. The query condition cache is server-wide and size-bounded, so a concurrent test can evict the entry between the two plain reads and turn the expected hit into a miss - observed once on the `amd_asan_ubsan, distributed plan, parallel` shard, passing on all three reruns. This matches the `no-parallel` tag `04207` carries for the same reason.
|
🕵 Pushed
Verified locally against a server built from this head: Not merged
|
Build profile diff (arm_release)Comparing ✅ No significant changes. Binary sizes
Only the stripped binary is compared: the official master build keeps debug symbols while PR builds strip them, so the other binaries differ by construction. Object file sizes17 object files changed (+177.08 KiB total), 0 added.
737 more object files are built by the master warmup baseline only (it builds every object-file target, a pull request build only Compile time of recompiled translation units572 translation units recompiled, 4014 s compile time in total, 556 of them have a recent master baseline.
|
…t-104251-parquet-single-file-parallelism # Conflicts: # src/Storages/ObjectStorage/StorageObjectStorageSource.cpp
…t read The fail-close check for a parallel single-file split compared `expected_file_cache_version` only against a pre-open `stat`, and `createReadBuffer` then reopened the path, so a concurrent truncating rewrite landing between the `stat` and the `open` could pass the check on the old token and hand the source the new file, applying a stale row-group assignment. The post-open re-stat only flipped `current_file_version_settled`, which merely bypasses the query condition cache, and the stale token was still used as the `ParquetMetadataCache` key via `object_with_metadata`. The opened fd is not reachable through the read buffer (it may be an mmap, io_uring, or compression wrapper), so bracket the open with a second `stat` of the path instead: once the token has settled, any write in the gap changes the token, so an unchanged token proves the opened bytes match it. On a mismatch, a bucketed source now throws `FILE_CHANGED_WHILE_READING` instead of proceeding, and a plain read drops the token so neither the format metadata cache nor the query condition cache keys the read under a version it may not describe. The split decision gets the same bracket: if the token moved while the footer was being read, fall back to a plain unsplit read instead of handing out a bucket assignment that would deterministically fail.
|
🕵 Status update: resolved the conflict with Conflict resolution ( AI-review blocker fixed ( Local verification: incremental CI on the previous head Remaining blockers are human calls: CI on |
…t-104251-parquet-single-file-parallelism # Conflicts: # docs/en/interfaces/specs/NativeProtocol.md # src/Processors/Formats/Impl/Parquet/Reader.cpp
…ened bytes Address the AI review finding on the local `ParquetMetadataCache` key being built from an unsettled file-version token: an in-place rewrite that keeps the inode and the byte size and lands in the same filesystem timestamp tick reuses the token, so a cached footer can describe a previous generation of the file and `checkFileMatchesBucketAssignment` would validate the bucket against the very footer the assignment was computed from. Instead of gating the metadata cache on the settle window (which regressed the master-pinned plain-path behavior of 04207_parquet_metadata_cache_local_file when tried before, and would also disable single-file parallelism for freshly written files), tie the assignment to the file generation actually opened: - `ParquetFileBucketInfo` carries a new local-only `footer_digest` (SipHash of the re-serialized thrift footer the split was computed from). It is never serialized over the cluster protocol and does not raise the minimum protocol version. - A per-bucket `StorageFileSource` no longer builds `object_with_metadata`, so it parses the footer of the bytes it actually opened instead of reusing a cached one; `checkFileMatchesBucketAssignment` now also compares the digest and throws `FILE_CHANGED_WHILE_READING` on a mismatch, failing close instead of silently applying a previous generation's row-group layout. - The split decision keeps warming the cache for later queries; a stale cached footer at decision time now deterministically fails close at read time. The plain (non-bucketed) read path is unchanged and keeps the master-pinned metadata-cache behavior.
|
🕵 Status update (automated pass):
CI is now running on the new head. No self-merge — leaving the merge decision to a human. |
The per-bucket profile-event accounting added for the single-file split treated any row-group assignment as a partition of the file among several readers, so it reported `ParquetPrunedRowGroups` relative to the assignment's own size. But the query condition cache also builds an assignment - for a single reader of the whole file, restricted to the row groups a previous run found matching - and there the omitted row groups were pruned by the cache, not handed to another reader. Accounting relative to the assignment made the event 0 for such a read, so a warm-cache repeat of a query that prunes row groups stopped reporting any pruning at all: `04512_parquet_geo_pruning_iceberg_renamed_column`, `04513_parquet_geo_pruning_iceberg_renamed_bbox_column` and `04514_parquet_geo_pruning_geostats_fallback` all lost their `ParquetPrunedRowGroups` line. A cache-derived assignment is now marked with `FileBucketInfo::omitted_row_groups_are_pruned`, and `Parquet::ReadManager::init` accounts for the whole file in that case (as before this feature) and for the bucket alone for a real split.
`computeParquetFooterDigest` hashed the thrift re-serialization of `FileMetaData`, and the generated `write` loads every enum field as an enumerator. Thrift metadata can legitimately carry an out-of-range value there - `encoding_stats` is advisory input a malformed or future writer may fill with garbage, which is exactly why `Reader::columnChunkCanUseDictionaryFilter` reads those fields through `isValidThriftEnum` - and loading it as an enumerator is undefined behavior. Under `-fsanitize=enum` with `-fno-sanitize-recover=all` the process aborts, so a file that reads perfectly well with a full scan turned into a silently killed query on the ASan + UBSan build: `04546_parquet_v3_dictionary_filter_bad_encoding_stats` produced no output at all for every one of its queries. The digest is now computed directly over the footer's layout - the schema shape and every row group's and column chunk's row counts, byte sizes and offsets, with a presence flag for each optional field. That is what a bucket assignment is derived from, so it still distinguishes file generations, it is still identical for a freshly parsed footer and the same footer served from `ParquetMetadataCache`, and it reads no enum.
`computeBucketsByCount` applied its minimum-row-groups-per-bucket floor unconditionally, so a file with fewer row groups than the floor stayed single-source no matter what `input_format_parquet_min_bytes_to_split` and `input_format_parquet_bytes_per_split_bucket` were set to. That contradicts the compatibility contract recorded in `SettingsChangesHistory.cpp`, which promises that `0` for both reproduces the pre-26.8 fan-out driven by the row-group count alone, and left the new settings unable to opt out of the heuristic for a large-row-group file. The floor is part of the same size heuristic as the byte gates, so it is now skipped exactly when both of them are disabled. With the defaults - and with any non-zero threshold - nothing changes, so the `clickbench_parquet_short` regression the floor was introduced for stays fixed. Adds `04812_parquet_single_file_split_row_group_floor_compat`, which pins both directions on an 8-row-group file.
|
🕵 Status update (automated pass): three fixes pushed, 1. 2. 3. AI review blocker (the hard-coded 16-row-group floor) — fixed in Local verification: No self-merge — leaving the merge decision to a human. CI is running on the new head. |
…_cache` as no-parallel The query condition cache is server-wide and size-bounded, so a concurrent test can evict the entry between the two plain reads and turn the expected hit into a miss - the same reason `04637_parquet_file_engine_bucketed_query_condition_cache` carries the tag. Seen flaky once on `Stateless tests (arm_binary, parallel)` (10/10 reruns passed): https://s3.amazonaws.com/clickhouse-test-reports/json.html?PR=104431&sha=e618aa3a2a0a0d1793bd29ef23effa1042ae7c2c&name_0=PR&name_1=Stateless%20tests%20%28arm_binary%2C%20parallel%29 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
🕵 Status update on head
Remaining: fresh CI on |
…eck fails When the post-open version-token check detects a concurrent rewrite on a plain (non-bucketed) read, the fallback dropped only the cache token, but `current_file_size` and `current_file_last_modified` still came from the pre-open `stat` and were later materialized by `addRequestedFileLikeStorageVirtualsToChunk` as the `_size` / `_time` virtual columns — so the query could read one file generation while exposing virtual columns from another. Reset both optionals in the same branch; the virtual columns are then filled with NULL.
|
🕵 Status update (automated pass, 2026-08-07) Pushed b57d30f, addressing the sole AI Review Major (thread on CI triage on the previous head
No self-merge; waiting on CI for b57d30f, the AI re-verdict, and approval. |
…t-104251-parquet-single-file-parallelism
|
🕵 Status update (automated pass):
Waiting for CI on |
…ves in the split-size estimator The reader renames the Parquet map tuple elements to the `keys` / `values` subcolumn names `DataTypeMap` requires (`SchemaConverter` does this at the output-tuple level), so a direct map-subcolumn read requests `m.keys` / `m.values` — but `collectLogicalPaths` normalized the footer leaves to `m.key` / `m.value`, so such a read matched no chunk, `projectedCompressedBytes` collapsed to 0, and the byte-based gate suppressed the single-file split even for multi-megabyte reads. Now the map branch contributes the fixed `keys` / `values` name components (nested cases included via the recursion). Adds `04821_parquet_single_file_split_map_subcolumns` covering a light `m.keys` control, heavy `m.values`, and a map nested inside a tuple (`t.m.values`). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
🕵 Addressed the sole AI Review Major (thread on While writing the test I found that direct Verified locally: incremental build clean; the new test plus CI triage on the previous head
No self-merge; waiting on CI for |
| /// whole-map requests keep matching. | ||
| ++idx; | ||
| String raw_rep = raw + "." + rep.name; | ||
| String logical_keys = logical.empty() ? "keys" : logical + ".keys"; |
There was a problem hiding this comment.
collectLogicalPaths is now hard-wired to the default Map spelling (keys / values), but the reader still has another supported naming mode here: when a Parquet map is requested as Array(Tuple(...)), SchemaConverter switches to MapTupleAsPlainTuple and exposes the same leaves as m.key / m.value instead of m.keys / m.values (src/Processors/Formats/Impl/Parquet/SchemaConverter.cpp:507-525).
In that mode requested_columns contains m.value, while this branch only emits m.values, so projectedCompressedBytes misses every value chunk and the size gate keeps a heavy m.value read single-source. Please preserve both spellings here (or drive the normalization from the same MapTupleAsPlainTuple decision as the reader) and add a regression test with an explicit Array(Tuple(key String, value String)) schema.
2 participants
Reverts #104359