Parquet: skip reading constant column chunks via statistics

src/Common/ProfileEvents.cpp

@@ -1476,8 +1476,14 @@ The server successfully detected this situation and will download merged part fr
     \
     M(ParquetReadRowGroups, "The total number of row groups read from parquet data", ValueType::Number) \
     M(ParquetPrunedRowGroups, "The total number of row groups pruned from parquet data", ValueType::Number) \
+    M(ParquetConstantColumnChunks, "The total number of parquet column chunks materialized from a single value in their min/max statistics, without reading their data pages", ValueType::Number) \
+    M(ParquetConstantColumnChunksWithNulls, "The total number of parquet column chunks holding a single value plus nulls (per their statistics), for which only the definition levels were decoded and the value was taken from the statistics", ValueType::Number) \
     M(ParquetDecodingTasks, "Tasks issued by parquet reader", ValueType::Number) \
     M(ParquetDecodingTaskBatches, "Task groups sent to a thread pool by parquet reader", ValueType::Number) \
+    M(ParquetReadTasks, "Coalesced read tasks created by the Parquet reader. Divided by `ParquetReadRowGroups`, values below 1 mean one read spans several row groups, which serializes their delivery", ValueType::Number) \
+    M(ParquetReadTaskBytes, "Bytes covered by `ParquetReadTasks`, including bytes read to close short gaps between requested ranges", ValueType::Bytes) \
+    M(ParquetReadAheadSubgroups, "Row subgroups whose data-page reads the Parquet reader issued ahead, while the previous subgroup of the row group was still decoding (see `input_format_parquet_read_ahead_subgroups`)", ValueType::Number) \
+    M(ParquetPrefetchStarvation, "Times a decoding thread asked for a range whose read had not finished. High relative to `ParquetReadTasks` means read-ahead is too shallow", ValueType::Number) \
     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) \

src/Core/FormatFactorySettings.h

@@ -212,6 +212,12 @@ Skip pages using min/max values from column index.
 )", 0) \
     DECLARE(Bool, input_format_parquet_use_offset_index, true, R"(
 Minor tweak to how pages are read from parquet file when no page filtering is used.
+)", 0) \
+    DECLARE(Bool, input_format_parquet_use_constant_column_optimization, true, R"(
+When a Parquet column chunk provably holds a single value in every row (according to its min/max statistics), materialize that value directly instead of reading and decoding the column's data pages. Also covers chunks that are all null (materialized as a sparse column) and chunks holding a single value plus nulls (only the definition levels are decoded).
+)", 0) \
+    DECLARE(Float, input_format_parquet_constant_column_sparse_ratio, 0.9375, R"(
+For `input_format_parquet_use_constant_column_optimization`: a Parquet column chunk holding a single value plus nulls is materialized as a sparse column (memory proportional to the non-null rows) when the fraction of nulls is at least this ratio, and as a dense `Nullable` column otherwise. `1` disables sparse materialization for such chunks.
 )", 0) \
     DECLARE(Bool, input_format_parquet_verify_checksums, true, R"(
 Verify page checksums when reading parquet files.
@@ -250,6 +256,42 @@ Min bytes required for local read (file) to do seek, instead of read with ignore
 )", 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_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 from `max_download_threads` and `max_parsing_threads`.
+
+With too few reads in flight to cover the storage's response time, decoding threads end up running
+the reads themselves or waiting for them.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_max_active_files, 0, R"(
+How many Parquet files may read ahead at the same time when a query reads many files. `0` means no
+limit, which is the previous behaviour.
+
+All files share one IO pool, so with many files each gets too few reads in flight and none finish
+early. Files that do not hold a slot still read the row group they must deliver next.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_read_ahead_subgroups, 0, R"(
+How many row subgroups ahead the Parquet reader may issue data-page reads for within a row group. `0`
+keeps the previous behaviour, where a subgroup's reads are issued only after its predecessor has been
+decoded, so the reader waits out the storage's response time on every subgroup. `1` issues the next
+subgroup's reads while the current one decodes. Subgroups are still decoded in order.
+
+Read-ahead is opportunistic: it stops when its memory budget
+(`input_format_parquet_read_ahead_memory_fraction`) is used up. Only helps for files whose row groups
+are split into several subgroups (see `input_format_parquet_max_block_size`) and that have an offset
+index; otherwise a row group's data is already read as one range.
+)", 0) \
+    DECLARE(Double, input_format_parquet_read_ahead_memory_fraction, 0.25, R"(
+Share of the Parquet reader's prefetch memory budget (`input_format_parquet_prefetch_memory_fraction`)
+reserved for data pages read ahead within a row group (`input_format_parquet_read_ahead_subgroups`).
+Range `[0, 1]`. The rest of the prefetch budget keeps other row groups reading ahead.
+)", 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 are coalesced up to
+this size. `0` derives it from the min-bytes-for-seek of the underlying storage.
+
+A read never spans two row groups regardless of this setting.
 )", 0) \
     DECLARE(Bool, input_format_arrow_allow_missing_columns, true, R"(
 Allow missing columns while reading Arrow input formats

src/Core/SettingsChangesHistory.cpp

@@ -42,6 +42,13 @@ 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_use_constant_column_optimization", false, true, "New setting: when a Parquet column chunk provably holds a single value in every row (per its min/max statistics), materialize that value directly instead of reading and decoding the column's data pages (reader v3)."},
+            {"input_format_parquet_constant_column_sparse_ratio", 1.0, 0.9375, "New setting: a Parquet column chunk holding a single value plus nulls is materialized as a sparse column when the fraction of nulls is at least this ratio (reader v3)."},
+            {"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 from `max_download_threads` and `max_parsing_threads`; the derived value is larger than the previous hard-coded `max_download_threads`, which defaults to 4 and was chosen for the URL engine."},
+            {"input_format_parquet_bytes_per_read_task", 0, 0, "New setting: target size of a single read issued by the Parquet reader. 0 derives it from the min-bytes-for-seek of the underlying storage."},
+            {"input_format_parquet_read_ahead_subgroups", 0, 0, "New setting: how many row subgroups ahead the Parquet reader may issue data-page reads for within a row group. 0 keeps the previous behavior of issuing a subgroup's reads only after its predecessor was decoded."},
+            {"input_format_parquet_read_ahead_memory_fraction", 0.25, 0.25, "New setting: share of the Parquet prefetch memory budget reserved for data pages read ahead within a row group."},
+            {"input_format_parquet_max_active_files", 0, 0, "New setting: how many Parquet files may read ahead at the same time when a query reads many of them. 0 means no limit, which is the previous behavior."},
         });
 
         addSettingsChanges(settings_changes_history, "26.6",

src/Formats/FormatFactory.cpp

@@ -219,6 +219,8 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se
     format_settings.parquet.bloom_filter_push_down = settings[Setting::input_format_parquet_bloom_filter_push_down];
     format_settings.parquet.page_filter_push_down = settings[Setting::input_format_parquet_page_filter_push_down];
     format_settings.parquet.use_offset_index = settings[Setting::input_format_parquet_use_offset_index];
+    format_settings.parquet.use_constant_column_optimization = settings[Setting::input_format_parquet_use_constant_column_optimization];
+    format_settings.parquet.constant_column_sparse_ratio = settings[Setting::input_format_parquet_constant_column_sparse_ratio];
 
     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];
@@ -248,6 +250,11 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se
     format_settings.parquet.bloom_filter_bits_per_value = settings[Setting::output_format_parquet_bloom_filter_bits_per_value];
     format_settings.parquet.bloom_filter_flush_threshold_bytes = settings[Setting::output_format_parquet_bloom_filter_flush_threshold_bytes];
     format_settings.parquet.local_read_min_bytes_for_seek = settings[Setting::input_format_parquet_local_file_min_bytes_for_seek];
+    format_settings.parquet.max_io_threads = settings[Setting::input_format_parquet_max_io_threads];
+    format_settings.parquet.bytes_per_read_task = settings[Setting::input_format_parquet_bytes_per_read_task];
+    format_settings.parquet.max_active_files = settings[Setting::input_format_parquet_max_active_files];
+    format_settings.parquet.read_ahead_subgroups = settings[Setting::input_format_parquet_read_ahead_subgroups];
+    format_settings.parquet.read_ahead_memory_fraction = settings[Setting::input_format_parquet_read_ahead_memory_fraction];
     format_settings.parquet.enable_row_group_prefetch = settings[Setting::input_format_parquet_enable_row_group_prefetch];
     format_settings.parquet.verify_checksums = settings[Setting::input_format_parquet_verify_checksums];
     format_settings.parquet.local_time_as_utc = settings[Setting::input_format_parquet_local_time_as_utc];

src/Formats/FormatParserSharedResources.cpp

@@ -27,6 +27,24 @@ FormatParserSharedResourcesPtr FormatParserSharedResources::singleThreaded(const
 }
 
 
+bool FormatParserSharedResources::tryAcquirePrefetchSlot(size_t max_active)
+{
+    if (max_active == 0)
+        return true; // unlimited
+    size_t cur = active_prefetch_readers.load(std::memory_order_relaxed);
+    while (cur < max_active)
+    {
+        if (active_prefetch_readers.compare_exchange_weak(cur, cur + 1, std::memory_order_acq_rel, std::memory_order_relaxed))
+            return true;
+    }
+    return false;
+}
+
+void FormatParserSharedResources::releasePrefetchSlot()
+{
+    active_prefetch_readers.fetch_sub(1, std::memory_order_release);
+}
+
 void FormatParserSharedResources::finishStream()
 {
     num_streams.fetch_sub(1, std::memory_order_relaxed);

src/Formats/FormatParserSharedResources.h

@@ -23,6 +23,7 @@ struct FormatParserSharedResources
     const size_t max_io_threads = 0;
 
     std::atomic<size_t> num_streams{0};
+    std::atomic<size_t> active_prefetch_readers{0};
     ThreadPoolCallbackRunnerFast parsing_runner;
     ThreadPoolCallbackRunnerFast io_runner;
 
@@ -35,6 +36,12 @@ struct FormatParserSharedResources
 
     void finishStream();
 
+    /// See input_format_parquet_max_active_files. Spreading one IO pool across many files leaves
+    /// each with too few reads in flight to cover the storage's response time and none finishing
+    /// early. A reader without a slot still reads what it must deliver next, so this cannot stall.
+    bool tryAcquirePrefetchSlot(size_t max_active);
+    void releasePrefetchSlot();
+
     size_t getParsingThreadsPerReader() const;
     size_t getIOThreadsPerReader() const;
 

src/Formats/FormatSettings.h

@@ -349,6 +349,8 @@ struct FormatSettings
         bool bloom_filter_push_down = true;
         bool page_filter_push_down = true;
         bool use_offset_index = true;
+        bool use_constant_column_optimization = true;
+        float constant_column_sparse_ratio = 0.9375f;
 
         bool enable_json_parsing = true;
         bool preserve_order = false;
@@ -359,6 +361,16 @@ struct FormatSettings
         UInt64 max_block_size = DEFAULT_BLOCK_SIZE;
         size_t prefer_block_bytes = DEFAULT_BLOCK_SIZE * 256;
         size_t local_read_min_bytes_for_seek = 8192;
+        /// 0 = derive from max_download_threads / max_parsing_threads.
+        size_t max_io_threads = 0;
+        /// 0 = derive from the storage's min-bytes-for-seek.
+        size_t bytes_per_read_task = 0;
+        /// 0 = no limit on how many files prefetch ahead concurrently.
+        size_t max_active_files = 0;
+        /// 0 = issue a subgroup's reads only after its predecessor finished.
+        size_t read_ahead_subgroups = 0;
+        /// Share of the prefetch memory budget reserved for read-ahead within a row group.
+        double read_ahead_memory_fraction = 0.25;
         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

src/Formats/insertNullAsDefaultIfNeeded.cpp

@@ -16,6 +16,12 @@ namespace DB
 
 bool insertNullAsDefaultIfNeeded(ColumnWithTypeAndName & input_column, const ColumnWithTypeAndName & header_column, size_t column_i, BlockMissingValues * block_missing_values)
 {
+    /// Input formats may produce sparse columns (e.g. the Parquet reader for chunks that are all
+    /// null or mostly null), and they survive the Native protocol from the client. The casts below
+    /// expect the concrete Nullable / Array / Tuple / Map columns, so expand first.
+    if (input_column.column->isSparse())
+        input_column.column = input_column.column->convertToFullColumnIfSparse();
+
     if (isArray(input_column.type) && isArray(header_column.type))
     {
         ColumnWithTypeAndName nested_input_column;

src/Interpreters/AsynchronousInsertQueue.cpp

@@ -1414,7 +1414,12 @@ Chunk AsynchronousInsertQueue::processPreprocessedEntries(
 
         auto columns = block_to_insert.getColumns();
         for (size_t i = 0, s = columns.size(); i < s; ++i)
-            result_columns[i]->insertRangeFrom(*columns[i], 0, columns[i]->size());
+        {
+            /// Blocks may carry ColumnConst / ColumnSparse (e.g. from an input format that materializes
+            /// provably-constant column chunks); insertRangeFrom needs the concrete column type.
+            auto full_column = columns[i]->convertToFullColumnIfConst()->convertToFullColumnIfSparse();
+            result_columns[i]->insertRangeFrom(*full_column, 0, full_column->size());
+        }
 
         total_rows += block_to_insert.rows();
 

src/Processors/Executors/StreamingFormatExecutor.cpp

@@ -149,7 +149,13 @@ size_t StreamingFormatExecutor::insertChunk(Chunk chunk, size_t num_bytes)
 
     auto columns = chunk.detachColumns();
     for (size_t i = 0, s = columns.size(); i < s; ++i)
-        result_columns[i]->insertRangeFrom(*columns[i], 0, columns[i]->size());
+    {
+        /// Input formats may produce ColumnConst / ColumnSparse (e.g. the Parquet reader for column
+        /// chunks whose statistics prove a single value or all nulls); insertRangeFrom into the full
+        /// result column requires the concrete column type.
+        auto full_column = columns[i]->convertToFullColumnIfConst()->convertToFullColumnIfSparse();
+        result_columns[i]->insertRangeFrom(*full_column, 0, full_column->size());
+    }
 
     return chunk_rows;
 }