Parquet: read-path redesign — serve bytes as they arrive, bounded coalescing, lifetime memory pools, read-ahead controller, filesystem-cache fixes

src/Common/ProfileEvents.cpp

@@ -78,6 +78,9 @@
     M(PageCacheResized, "Number of times the userspace page cache was auto-resized (typically happens a few times per second, controlled by memory_worker_period_ms).", ValueType::Number) \
     M(PageCacheOvercommitResize, "Number of times the userspace page cache was auto-resized to free memory during a memory allocation.", ValueType::Number) \
     M(PageCacheReadBytes, "Number of bytes read from userspace page cache.", ValueType::Bytes) \
+    M(PageCacheReadThroughBudgetSamples, "Times the userspace page cache recomputed its read-through budget from its own fitted bandwidth and round-trip time (once per source read)", ValueType::Number) \
+    M(PageCacheReadThroughBudgetBytesSum, "Sum of the read-through budgets the page cache computed; divide by `PageCacheReadThroughBudgetSamples` for the mean budget in bytes", ValueType::Bytes) \
+    M(PageCacheReadThroughBytes, "Number of bytes fetched from the source but thrown away because they cover cache blocks that were already present: the page cache read through a short island of cached blocks in order to serve two runs of missing blocks with one request", ValueType::Bytes) \
     M(MMappedFileCacheHits, "Number of times a file has been found in the MMap cache (for the 'mmap' read_method), so we didn't have to mmap it again.", ValueType::Number) \
     M(MMappedFileCacheMisses, "Number of times a file has not been found in the MMap cache (for the 'mmap' read_method), so we had to mmap it again.", ValueType::Number) \
     M(OpenedFileCacheHits, "Number of times a file has been found in the opened file cache, so we didn't have to open it again.", ValueType::Number) \
@@ -1646,6 +1649,13 @@ The server successfully detected this situation and will download merged part fr
     M(ParquetPrefetcherReadRandomRead, "The total number of reads with ReadMode::RandomRead by DB::Parquet::Prefetcher", ValueType::Number) \
     M(ParquetPrefetcherReadSeekAndRead, "The total number of reads with ReadMode::SeekAndRead by DB::Parquet::Prefetcher", ValueType::Number) \
     M(ParquetPrefetcherReadEntireFile, "The total number of read with ReadMode::EntireFileIsInMemory by DB::Parquet::Prefetcher", ValueType::Number) \
+    M(ParquetPartialReadsServed, "Times the Parquet reader started decoding from a coalesced read before that read had finished, because the requested bytes had already arrived", ValueType::Number) \
+    M(ParquetReadTasks, "Coalesced read tasks created by the Parquet reader", ValueType::Number) \
+    M(ParquetReadTaskBytes, "Bytes covered by `ParquetReadTasks`, including bytes read to close short gaps between requested ranges", ValueType::Bytes) \
+    M(ParquetReadFirstByteMicroseconds, "Sum of the time from starting a `DB::Parquet::Prefetcher` source read to its first progress callback (or to completion, if the transport never calls back), i.e. round-trip time to the first byte", ValueType::Microseconds) \
+    M(ParquetReadTransferMicroseconds, "Sum of the time spent transferring bytes in a `DB::Parquet::Prefetcher` source read after the first byte arrived", ValueType::Microseconds) \
+    M(ParquetPlannedReads, "Groups of Parquet reads (the index reads of one row group, or the data pages one row subgroup needs) issued by the reader's issue controller ahead of the stage that consumes them. Counts only the groups the controller itself issued: a group that the stage needing it took over first is not counted", ValueType::Number) \
+    M(ParquetIssueQueueStalls, "Times the Parquet reader's issue controller had to stop reading ahead because the bytes-in-flight target or the memory pool of the next planned read was full. Counted once per attempt, not per planned read left waiting; counted even if the attempt still issued a privileged read (one the reader cannot make progress without), since read-ahead was blocked either way", ValueType::Number) \
     M(ParquetRowsFilterExpression, "The total number of rows that were passed through filter", ValueType::Number) \
     M(ParquetColumnsFilterExpression, "The total number of columns that were passed through filter", ValueType::Number) \
     M(FilterTransformPassedRows, "Number of rows that passed the filter in the query", ValueType::Number) \

src/Core/FormatFactorySettings.h

@@ -201,8 +201,11 @@ Schedule prefetches more aggressively if memory usage is below than threshold. P
     DECLARE(UInt64, input_format_parquet_memory_high_watermark, 4ul << 30, R"(
 Approximate memory limit for the Parquet reader. Limits how many row groups or columns can be read in parallel. When reading multiple files in one query, the limit is on total memory usage across those files.
 )", 0) \
-    DECLARE(Double, input_format_parquet_prefetch_memory_fraction, 0.6, R"(
-Advanced tuning knob for the Parquet reader scheduler. Of the memory budget reserved for column data, the fraction given to compressed read-ahead (the `ColumnDataPrefetch` stage) versus decoded output (the `ColumnData` stage); the rest goes to decode. A higher value keeps more compressed pages in flight to hide read latency (useful on high-latency storage such as S3); a lower value caps read-ahead and leaves more budget for decoded columns. Must be in [0, 1]. The index and bloom-filter stages keep a fixed share of the memory budget regardless of this setting.
+    DECLARE(Double, input_format_parquet_compressed_memory_fraction, 0.35, R"(
+Share of `input_format_parquet_memory_high_watermark` the Parquet reader may hold as compressed data
+pages that are in flight or waiting to be decoded. This bounds how far ahead of decoding the reader
+reads. The rest of the budget (minus 5% for metadata) holds decoded columns, including chunks already
+handed to the query pipeline. Range `(0, 0.95)`.
 )", 0) \
     DECLARE(Double, input_format_parquet_decode_thread_fraction, 0.375, R"(
 Advanced tuning knob for the Parquet reader scheduler. The fraction of the Parquet parsing thread pool dedicated to column decoding (the `ColumnData` stage); the remaining stages, which only issue asynchronous reads, share the rest. Raise it to give decoding (the only CPU-bound stage) more parallelism on fast/local storage; the default suits latency-bound remote reads where memory, not threads, limits concurrency. Must be in [0, 1].
@@ -247,9 +250,68 @@ Allow missing columns while reading Parquet input formats
 )", 0) \
     DECLARE(UInt64, input_format_parquet_local_file_min_bytes_for_seek, 8192, R"(
 Min bytes required for local read (file) to do seek, instead of read with ignore in Parquet input format
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_max_io_threads, 0, R"(
+Size of the thread pool that issues reads for the Parquet reader, shared by all files read by the
+query. `0` derives it as `max(max_download_threads, min(max_parsing_threads, 16))`.
+
+With too few reads in flight to cover the storage's response time, decoding threads end up waiting
+for reads.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_bytes_per_read_task, 0, R"(
+Target size of a single read issued by the Parquet reader; nearby column chunks and pages are
+coalesced up to this size. `0` derives it as four times the min-bytes-for-seek of the underlying
+storage. Bytes of a coalesced read become available to decoding as they arrive, so a large value
+does not delay the first row group of the read.
 )", 0) \
     DECLARE(Bool, input_format_parquet_enable_row_group_prefetch, true, R"(
 Enable row group prefetching during parquet parsing. Currently, only single-threaded parsing can prefetch.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_coalesce_gap_bytes, 2097152, R"(
+Largest gap between two needed byte ranges of a Parquet file that the reader reads through in order to
+serve both with one request. Applied on top of the storage's min-bytes-for-seek (the smaller wins);
+`0` uses the storage value only. On object storage the useful gap is about one round trip's worth of
+bandwidth, ~2 MiB; reading through larger gaps costs bytes without saving time.
+)", 0) \
+    DECLARE(Double, input_format_parquet_max_read_amplification, 8, R"(
+Upper bound on `bytes read / bytes needed` for one coalesced Parquet read. Coalescing stops extending a
+read when the span would exceed this multiple of the useful bytes it covers, so a few small column chunks
+cannot drag megabytes of unrelated data through the cache or the network. `0` disables the bound. Any other
+value must be `>= 1` (a read always spans at least the bytes it serves); values in `(0, 1)` are rejected.
+
+The bound is checked while a read is being grown one neighbouring range at a time, so it also rejects merges
+whose *intermediate* ratio is too high even when the finished read would be well within the bound. Tight
+values therefore cost round trips: on a query reading six narrow columns of a wide table, `4` split the reads
+of a row group into 1.6x as many requests as no bound at all and made the decoding threads wait 3.6x longer,
+while `8` and up reached the same bytes-read as no bound with the same request count. Values from 6 to 16
+measured the same on that dataset; below 6 the request count climbs.
+
+See also `input_format_parquet_read_amplification_floor_bytes`, which exempts reads that waste little in
+absolute terms from this bound.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_read_amplification_floor_bytes, 262144, R"(
+A coalesced Parquet read that wastes no more than this many bytes -- reads at most this much beyond the bytes
+it was asked for -- is never split by `input_format_parquet_max_read_amplification`, whatever its ratio.
+
+A ratio alone says nothing about how much is actually wasted, and on small files it reads as alarming when the
+waste is trivial: five columns of a 25 KiB file lie a few KB apart, so reading the file in one request wastes
+about 20 KB, which the ratio scores as a bad read. Measured on a 17,554-file Iceberg table, the ratio bound
+alone split each file's read three to four ways, buying 13% fewer bytes for 65% more requests and 45% more
+wall time; with this floor the same query matched the reader without any bound at all, while a query whose
+reads waste megabytes per useful range kept the bound's full effect (330 MiB read down to 11 MiB).
+
+`0` applies the amplification bound to every read regardless of how little it wastes.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_min_bytes_in_flight, 67108864, R"(
+Lower bound for the Parquet reader's bytes-in-flight target: the reader issues the index and data-page
+reads it has planned ahead of time until this many bytes (or more, if the fitted bandwidth times
+round-trip time of the storage asks for more) are being read at once. Higher values give the storage
+more concurrent requests to work on, at the cost of holding more compressed bytes in memory; the index
+reads of the row group that is next to be delivered, and the pages of the subgroup it reads next, are
+always issued regardless of this bound.
+
+`0` disables read-ahead planning; the reader issues reads on demand as before. Values below the fitted
+floor of 4 x `input_format_parquet_bytes_per_read_task` have no additional effect.
 )", 0) \
     DECLARE(Bool, input_format_arrow_allow_missing_columns, true, R"(
 Allow missing columns while reading Arrow input formats
@@ -1655,6 +1717,7 @@ Supported modes:
     MAKE_OBSOLETE(M, ParquetVersion, output_format_parquet_version, "2.latest") \
     MAKE_OBSOLETE(M, Bool, output_format_parquet_compliant_nested_types, true) \
     MAKE_OBSOLETE(M, Bool, output_format_parquet_unsupported_types_as_binary, false) \
+    MAKE_OBSOLETE(M, Double, input_format_parquet_prefetch_memory_fraction, 0.6) \
 
 #endif // __CLION_IDE__
 

src/Core/SettingsChangesHistory.cpp

@@ -42,6 +42,14 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya",
         {
             {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."},
+            {"input_format_parquet_max_io_threads", 0, 0, "New setting: size of the thread pool that issues reads for the Parquet reader. 0 derives it as `max(max_download_threads, min(max_parsing_threads, 16))`, which is larger than the previous hard-coded `max_download_threads` (default 4). `compatibility` keeps the derived value, since no single number expresses \"whatever `max_download_threads` is\"; to restore the old pool size, set `input_format_parquet_max_io_threads` to the value of `max_download_threads`."},
+            {"input_format_parquet_bytes_per_read_task", 0, 0, "New setting: target size of a single coalesced read issued by the Parquet reader. 0 derives it from the min-bytes-for-seek of the underlying storage, as before."},
+            {"input_format_parquet_coalesce_gap_bytes", 0, 2097152, "New setting: cap on the gap the Parquet reader reads through when coalescing nearby ranges; previously the storage's min-bytes-for-seek (4 MiB on object storage) applied unconditionally."},
+            {"input_format_parquet_max_read_amplification", 0, 8, "New setting: bound on bytes read / bytes needed per coalesced Parquet read. 8 rather than a tighter value because the bound is applied to each intermediate state of a read as it grows, so tight values reject merges that would have finished within the bound and pay round trips for it (measured: 4 cost 1.6x the requests and 3.6x the read wait of no bound on a six-narrow-column query, while 8 matched no bound's request count and still read 37% fewer bytes than the unbounded reader over a 23-query benchmark)."},
+            {"input_format_parquet_read_amplification_floor_bytes", 0, 262144, "New setting: a coalesced Parquet read wasting no more than this many bytes is exempt from `input_format_parquet_max_read_amplification`. Without it the ratio bound splits reads of small files whose waste is trivial in absolute terms (measured on a 17,554-file Iceberg table: 65% more requests and 45% more wall time for 13% fewer bytes). previous_value=0 so `compatibility` with older versions applies the ratio bound unconditionally, as before this setting existed."},
+            {"input_format_parquet_compressed_memory_fraction", 0.45, 0.35, "New setting: share of the Parquet reader memory budget held as compressed pages in flight; replaces the previous per-stage split (`data_memory_fraction`=0.75 x default `prefetch_memory_fraction`=0.6), which gave the compressed read-ahead 45% of the budget. previous_value=0.45 so `compatibility` with older versions restores that proportion."},
+            {"input_format_parquet_prefetch_memory_fraction", 0.6, 0.6, "Obsolete setting, the Parquet reader memory budget is now split by lifetime class (metadata / compressed / decoded) and sized by `input_format_parquet_compressed_memory_fraction`."},
+            {"input_format_parquet_min_bytes_in_flight", 0, 67108864, "New setting: lower bound for the Parquet reader's bytes-in-flight target, which bounds how far ahead the reader pre-issues planned index and data-page reads. The previous behavior -- reads issued on demand, one row group's stage at a time -- is the new setting's `0` value, which `compatibility` restores."},
         });
 
         addSettingsChanges(settings_changes_history, "26.6",