Parquet: size reads from the file layout instead of the seek threshold - #2275
Open
UnamedRus wants to merge 6 commits into
Open
Parquet: size reads from the file layout instead of the seek threshold#2275UnamedRus wants to merge 6 commits into
UnamedRus wants to merge 6 commits into
Conversation
…hold On object storage the reader's concurrency came out of how ranges happened to coalesce rather than from anything chosen. Measured on a 271 MB single-file scan (59-column projection, 23 row groups of ~11.8 MB), the default settings produced twelve 22.6 MB reads - fewer than one read per row group - and a 4-thread pool, which left ~3 reads in flight and most of each read's latency on the critical path. Anything that perturbed coalescing swung the result by a factor of two. Changes: * `bytes_per_read_task` now bounds the task. It was compared against the distance from the seed range in each direction independently, so a task could reach `seed + 2 * bytes_per_read_task`; setting it below the natural coalescing width did nothing at all. Compare against the resulting span instead. * A read never spans a row group. `getRangeData` waits for a whole task - there is no partial completion - so a read covering the tail of one row group and the head of the next made the earlier one wait for the later one's bytes, serializing in-order delivery. `Reader` now hands the Prefetcher the row group boundaries; if the metadata is unusable or row groups are not in ascending order, coalescing is left unconstrained. * Read size adapts to how busy the IO pool is. While the pool has spare capacity, smaller reads fill it faster; once it is busy, larger reads amortize the round trip. Hysteresis keeps the size from flapping at the threshold. * The IO pool is sized from the query. `max_download_threads` (default 4) was picked for the URL engine and is usually too small here, so decoding threads end up running reads themselves or waiting for them. * New settings, all defaulting to the previous behaviour except the pool size: `input_format_parquet_bytes_per_read_task`, `input_format_parquet_max_io_threads`, and `input_format_parquet_max_active_files`, which bounds how many files read ahead at once so each active one runs at a useful depth instead of every file crawling. A file without a slot still reads the row group it must deliver next, so a query cannot stall on it. * Profile events to make this visible without a profiler: `ParquetReadTasks`, `ParquetReadTaskBytes` and `ParquetPrefetchStarvation`. Dividing read tasks by `ParquetReadRowGroups` gives reads per row group; below 1 means reads span row groups. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: UnamedRus <dtitmoav@gmail.com>
Fixes the 02995_new_settings_history failure in Fast test: every new setting has to appear in SettingsChangesHistory.cpp. All three default to 0, so the settings themselves change no behavior; the note on max_io_threads records that its derived value is larger than the previous hard-coded max_download_threads. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: UnamedRus <dtitmoav@gmail.com>
…descriptions The cherry-pick that extracted this branch carried over two SettingsChangesHistory entries belonging to the constant-column change: this branch declared input_format_parquet_use_constant_column_optimization and input_format_parquet_constant_column_sparse_ratio without defining either. Also stop naming the reader version in the new descriptions - v3 is the default now, matching 6ba63a1 which already removed it from the other reader settings. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: UnamedRus <dtitmoav@gmail.com>
UnamedRus
changed the title
Keep only what the code cannot say: why the task budget is compared against the span rather than per direction, why a read must not cross a row group, the writer quirk around dictionary_page_offset, why Task::owner exists, and the exactly-once accounting of tasks_in_flight. Drop the restatements. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: UnamedRus <dtitmoav@gmail.com>
Two changes to how data-page reads get issued. Issuance is one unit of work per subgroup instead of one task per column. The work already ran on the scheduling thread (scheduleTask) and the task body was empty, so a task per column bought a thread-pool round trip each and an N-wide barrier before decoding could start. New setting `input_format_parquet_read_ahead_subgroups` (default 0, previous behavior) issues the next subgroup's reads while the current one decodes. Until now a subgroup's reads were issued only after its predecessor was decoded and delivered, so the storage's response time was paid again on every subgroup instead of overlapping with work. Read-ahead skips filtered-out subgroups and claims a subgroup with a CAS, so losing the race to the normal path is harmless. That CAS exposed a latent defect in the sequential-admission loop: it took the expected value from its own load, so it succeeded whatever the current stage was and guarded nothing. It only worked because nothing else moved a subgroup out of NotStarted. It now compares against NotStarted explicitly, which is required for read-ahead not to admit a subgroup twice. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: UnamedRus <dtitmoav@gmail.com>
…instead of admitting the next subgroup early The previous read-ahead admitted subgroup N+1 into the stage machine while N was still decoding. `finishRowSubgroupStage` then issued N+1's `ColumnData` tasks as soon as its `ColumnDataPrefetch` finished, so two subgroups of one row group decoded concurrently. `Reader::ColumnChunk` has a single sequential page cursor (`page`, `next_page_offset`, `data_pages_idx`) and a lazily initialised dictionary shared by all subgroups of the row group, so this corrupted decoding; it also collided on the one-slot-per-(stage, row group) task queue in `setTasksToSchedule` and could regress `read_ptr`, calling `clearColumnChunk` while a subgroup was still being decoded. Now the `ColumnDataPrefetch` task of subgroup N also issues the first-step data-page reads of subgroups N+1..N+k. Nothing about admission or decode order changes: the next subgroup is admitted by the normal path once N's main step is done, and skips its own `ColumnDataPrefetch` stage (`reads_issued_ahead`) because there is nothing left to issue. This is idempotent by construction: `determinePagesToPrefetch` advances `data_pages_prefetch_idx` past the pages it handed out, and `startPrefetch` skips handles that already have a task. Own reads are issued before read-ahead reads so N keeps priority in coalescing. `input_format_parquet_read_ahead_subgroups` values above 1 now work. Read-ahead bytes are charged to a new accounting-only stage `ColumnDataReadAhead` with its own share of the prefetch budget (`input_format_parquet_read_ahead_memory_fraction`, default 0.25 of the prefetch share), so read-ahead cannot eat the `ColumnDataPrefetch` budget that keeps other row groups moving. `flushMemoryUsageDiff` skips scheduling on it. New profile event `ParquetReadAheadSubgroups`. Test `04813_parquet_read_ahead_subgroups` reads a file with many subgroups per row group and several pages per subgroup under a low decode watermark, with read-ahead 0/1/3, filtered and single-threaded, and checks the profile event. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: UnamedRus <dtitmoav@gmail.com> (cherry picked from commit c0ffe68)
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.
Basically, before only driving force behind concurrent get requests to S3 was stealing tasks by decoding threads.
Default concurrency was capped by 4 threads (from URL table engine setting)
But relying on task stealing for concurrency is unreliable and it makes order unstable, which can stall pipeline, because RG can wait for last GET request for last column of it, to actually push chunk down query pipeline.
So, here few attempts to improve this behavior was made.
Measured
Same binary, same query, interleaved, n=6, only the two settings differing between
arms.
forced old=max_io_threads=4, bytes_per_read_task=16Mi.Reads per row group go from 0.52 at 22.6 MB to 1.3 at 8.4 MB. The baseline gains
~9% on its own;
input_format_parquet_use_constant_column_optimizationregressed by55-94% under the old sizing and is now the fastest arm, so it delivers latency as
well as the memory it already saved.
Note on defaults
One behaviour change for users who set nothing: the IO pool is derived rather than
fixed at
max_download_threads. The other new settings default to previousbehaviour.
input_format_parquet_max_io_threadsrestores the old value if needed.Related: #2266
Related: #2235
Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
Parquet reader: size reads from the file layout instead of the storage's seek threshold. A single read no longer spans two row groups, read size adapts to how busy the IO pool is, and the IO pool is sized from the query rather than from
max_download_threads. New settingsinput_format_parquet_bytes_per_read_task,input_format_parquet_max_io_threadsandinput_format_parquet_max_active_files, and new profile eventsParquetReadTasks,ParquetReadTaskBytesandParquetPrefetchStarvation.Documentation entry for user-facing changes
The three new settings are documented in their
DECLAREdoc strings.