Parquet v3 read concurrency

src/Core/FormatFactorySettings.h

@@ -199,7 +199,13 @@ When reading Parquet files, parse JSON columns as ClickHouse JSON Column.
 Schedule prefetches more aggressively if memory usage is below than threshold. Potentially useful e.g. if there are many small bloom filters to read over network.
 )", 0) \
     DECLARE(UInt64, input_format_parquet_memory_high_watermark, 4ul << 30, R"(
-Approximate memory limit for Parquet reader v3. 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.
+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.
+)", 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].
 )", 0) \
     DECLARE(Bool, input_format_parquet_page_filter_push_down, true, R"(
 Skip pages using min/max values from column index.

src/Core/SettingsChangesHistory.cpp

@@ -49,6 +49,8 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"analyzer_compatibility_apply_final_to_all_joined_tables", true, false, "Fixed a bug in the analyzer where FINAL on the left-most table of a JOIN was incorrectly applied to the other joined tables as well. previous_value=true so `compatibility` with versions before 26.6 restores the old behavior."},
             {"analyzer_compatibility_allow_non_aggregate_in_having", false, false, "New compatibility setting. When enabled, the new analyzer mimics the legacy `HAVING`-to-`WHERE` rewrite for non-aggregate AND-conjuncts instead of raising `NOT_AN_AGGREGATE`."},
             {"reserve_memory", 0, 0, "New setting to reserve memory for specific workload before starting a query."},
+            {"input_format_parquet_prefetch_memory_fraction", 0.6, 0.6, "New setting to tune the Parquet reader split of the column-data memory budget between compressed read-ahead and decode."},
+            {"input_format_parquet_decode_thread_fraction", 0.375, 0.375, "New setting to tune the Parquet reader share of the parsing thread pool given to column decoding."},
             {"output_format_image_width", 1024, 1024, "New setting controlling the width of the output image for image output formats such as PNG."},
             {"output_format_image_height", 1024, 1024, "New setting controlling the height of the output image for image output formats such as PNG."},
             {"output_format_image_terminal_mode", "", "", "New setting controlling whether image output formats such as PNG are rendered directly to the terminal using an inline image protocol."},

src/Formats/FormatFactory.cpp

@@ -223,6 +223,8 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se
     format_settings.parquet.enable_json_parsing = settings[Setting::input_format_parquet_enable_json_parsing];
     format_settings.parquet.memory_low_watermark = settings[Setting::input_format_parquet_memory_low_watermark];
     format_settings.parquet.memory_high_watermark = settings[Setting::input_format_parquet_memory_high_watermark];
+    format_settings.parquet.prefetch_memory_fraction = settings[Setting::input_format_parquet_prefetch_memory_fraction];
+    format_settings.parquet.decode_thread_fraction = settings[Setting::input_format_parquet_decode_thread_fraction];
     format_settings.parquet.allow_missing_columns = settings[Setting::input_format_parquet_allow_missing_columns];
     format_settings.parquet.skip_columns_with_unsupported_types_in_schema_inference = settings[Setting::input_format_parquet_skip_columns_with_unsupported_types_in_schema_inference];
     format_settings.parquet.output_string_as_string = settings[Setting::output_format_parquet_string_as_string];

src/Formats/FormatSettings.h

@@ -361,6 +361,10 @@ struct FormatSettings
         size_t local_read_min_bytes_for_seek = 8192;
         size_t memory_low_watermark = 2ul << 20;
         size_t memory_high_watermark = 4ul << 30;
+        /// Reader scheduler knobs: share of the column-data memory budget given to compressed
+        /// read-ahead vs decode, and ColumnData's share of the parsing thread pool.
+        double prefetch_memory_fraction = 0.6;
+        double decode_thread_fraction = 0.375;
 
         /// Write.
         UInt64 row_group_rows = 1000000;

src/Processors/Formats/Impl/Parquet/ReadCommon.cpp

@@ -6,15 +6,17 @@
 namespace DB::Parquet
 {
 
-SharedResourcesExt::Limits SharedResourcesExt::getLimitsPerReader(const FormatParserSharedResources & parser_shared_resources, double fraction)
+SharedResourcesExt::Limits SharedResourcesExt::getLimitsPerReader(const FormatParserSharedResources & parser_shared_resources, double memory_fraction, double thread_fraction)
 {
     const SharedResourcesExt & ext = *static_cast<const SharedResourcesExt *>(parser_shared_resources.opaque.get());
     size_t n = parser_shared_resources.num_streams.load(std::memory_order_relaxed);
-    fraction /= static_cast<double>(std::max(n, size_t(1)));
+    /// Split each budget across the files read in parallel.
+    memory_fraction /= static_cast<double>(std::max(n, size_t(1)));
+    thread_fraction /= static_cast<double>(std::max(n, size_t(1)));
     return Limits {
-        .memory_low_watermark = size_t(ext.total_memory_low_watermark * fraction),
-        .memory_high_watermark = size_t(ext.total_memory_high_watermark * fraction),
-        .parsing_threads = size_t(std::max(std::lround(parser_shared_resources.parsing_runner.getMaxThreads() * fraction + .5), 1l))};
+        .memory_low_watermark = size_t(ext.total_memory_low_watermark * memory_fraction),
+        .memory_high_watermark = size_t(ext.total_memory_high_watermark * memory_fraction),
+        .parsing_threads = size_t(std::max(std::lround(parser_shared_resources.parsing_runner.getMaxThreads() * thread_fraction + .5), 1l))};
 }
 
 #ifdef OS_LINUX

src/Processors/Formats/Impl/Parquet/ReadCommon.h

@@ -50,7 +50,7 @@ struct SharedResourcesExt
         size_t parsing_threads;
     };
 
-    static Limits getLimitsPerReader(const FormatParserSharedResources & parser_shared_resources, double fraction);
+    static Limits getLimitsPerReader(const FormatParserSharedResources & parser_shared_resources, double memory_fraction, double thread_fraction);
 };
 
 
@@ -88,6 +88,9 @@ enum class ReadStage
     ColumnIndexAndOffsetIndex,
 
     OffsetIndex,
+    /// Issues the compressed data-page reads (startPrefetch), no decode. Own memory budget, so many
+    /// row groups prefetch ahead while only a few decode at once. Decouples fetch from decode depth.
+    ColumnDataPrefetch,
     ColumnData,
 
     Deliver,
@@ -186,6 +189,9 @@ class MemoryUsageToken
         val += amount;
     }
 
+    /// How much memory this token currently charges.
+    size_t charged() const { return val; }
+
 private:
     ReadStage alloc_stage = ReadStage::Deallocated;
     size_t val = 0;