[Experiment] Antalya 26.6: Parquet v3 constant column skip

docs/design/parquet-v3-page-level-constant-column.md

@@ -0,0 +1,110 @@
+# Parquet v3: per-subgroup (page-level) constant-column detection
+
+## Motivation
+
+The current constant-column optimization (`detectConstantColumn`) fires only when a whole
+Parquet **column chunk** is single-valued (footer `min == max`, no nulls). Data that is sorted or
+clustered on a column is often constant over long **runs of pages** without the whole row group
+being constant, so chunk-level detection misses it.
+
+Parquet's **Column Index** stores per-page `min_values` / `max_values` / `null_pages` /
+`null_counts`. The v3 reader already loads and parses it (`applyColumnIndex`) for predicate
+push-down, so per-page "is this page constant?" is available at zero extra I/O. This lets us mark a
+column constant for the row range of an individual **row subgroup** (the unit that becomes one
+output `Chunk`) and materialize it as a `ColumnConst`, skipping the covered pages' reads and decode.
+
+## Tiers (both kept)
+
+| Tier | Source | Always present? | Exact flag? | Granularity |
+|------|--------|-----------------|-------------|-------------|
+| 1 (existing) | footer `ColumnMetaData.statistics` | yes | yes (`is_min_value_exact`) | whole chunk |
+| 2 (this doc) | Column Index (per page) | no (optional) | no | per subgroup |
+
+Tier 1 stays the always-on baseline and the **only** safe detector for `BYTE_ARRAY` /
+`FIXED_LEN_BYTE_ARRAY` (Column Index has no per-page exactness flag; 16-byte truncation can make two
+distinct strings compare equal). Tier 2 is opportunistic: only for fixed-width numeric/date/time,
+only when the Column Index is already loaded, only for chunks tier 1 did not already mark constant.
+
+## Key decision: do NOT change subgroup sizing
+
+Aligning subgroups to page boundaries would fragment the block stream into many tiny chunks
+(pages are far smaller than a row group), and a subgroup carries all columns so its size is bounded
+by the non-constant columns anyway. Instead, keep subgroup boundaries exactly as today and do
+**per-subgroup, per-column** detection: a column is constant for a subgroup iff every Column-Index
+page overlapping the subgroup's row range is constant with the *same* value (or every such page is
+`null_pages`). Chunk count is unchanged; we just catch subgroups that sit inside a constant run.
+
+## Phases
+
+- **Phase 0** — retain per-page constant info. `applyColumnIndex` currently discards the parsed
+  `parq::ColumnIndex`. Keep a compact per-page summary on `ColumnChunk` (value + `is_const` +
+  `all_null`), plus the page→`first_row_index` map already in the Offset Index. Populate only for
+  eligible types.
+- **Phase 1** — `detectConstantSubchunk(column, column_info, [start_row, end_row))`: scan the pages
+  overlapping the range; return constant + value when all are `is_const` and share one value;
+  all-null when all are `null_pages`.
+- **Phase 2** — call it in `intersectColumnIndexResultsAndInitSubgroups` for each subgroup /
+  primitive column that tier 1 didn't already mark constant; set `subchunk.is_constant` /
+  `is_all_null` / `constant_value` (the same fields `decodePrimitiveColumn` propagates).
+  `formOutputColumn` needs no change (already emits `ColumnConst`).
+- **Phase 3** — skip work for constant subchunks: `decodePrimitiveColumn` skips decode on
+  `subchunk.is_constant`; `determinePagesToPrefetch` skips fetching a page only when it is constant
+  in **every** subgroup that overlaps it.
+- **Phase 4** — stateless tests (page-run constant, all-null-per-page, byte-array negative/truncation
+  guard, `GROUP BY` correctness), with a new `ParquetConstantColumnSubchunks` ProfileEvent to prove
+  tier 2 fired.
+- **Phase 5** — ProfileEvents comparison on clustered data: expect further `S3GetObject` /
+  `ParquetFetchWaitTimeMicroseconds` drops with chunk count unchanged.
+
+## Optional force-load
+
+By default tier 2 only uses the Column Index when it is already loaded (columns with a predicate
+push-down). `input_format_parquet_use_column_index_for_constant_columns` (default off) extends it:
+the Column Index (+ Offset Index) is force-loaded for eligible read columns that have no predicate,
+so tier 2 can also fire on them. Cost is a small extra read of the (tiny, tail-contiguous, coalesced)
+index; worthwhile mainly for sorted / low-cardinality columns. `applyColumnIndex` records per-page
+constant info but skips predicate pruning when the column has no condition. A future `auto` mode
+could gate this on footer signals (row-group `sorting_columns`, low compressed-bytes-per-value,
+dictionary encoding stats) instead of an all-or-nothing switch.
+
+## Cast safety (future-proofing)
+
+The optimization only fires when the stats decoder needs no value-transforming conversion
+(`SchemaConverter` sets `allow_stats` accordingly), so today `input_type == output_type` for every
+constant column and no cast is applied. To keep the constant path correct if `allow_stats` is ever
+generalized to allow transforming casts, `formOutputColumn` builds the single-value constant in
+`input_type` and runs the same `castColumn` the per-row decode uses when `needs_cast` is set - a
+no-op today, O(1) on the `ColumnConst`, and it preserves const-ness. The all-null constant is
+synthesized directly in the output domain (Null / output default), so it is not cast. The mixed fill
+already goes through the normal `formOutputColumn` cast, so it is future-safe too. This does not
+touch the `allow_stats` decision itself.
+
+## Guardrails
+
+- All types, including `BYTE_ARRAY` / `FIXED_LEN_BYTE_ARRAY` strings. No truncation guard is needed
+  for the `min == max` case: statistics/Column-Index bounds are always valid
+  (`min <= every value <= max`) and truncation only widens them, so `min == max` proves a single
+  exact value (a truncated or multi-valued page yields `min < max`). This holds at both the chunk
+  level (tier 1) and per page (tier 2), so neither needs the `is_*_value_exact` flag.
+- Gate on the existing `input_format_parquet_use_constant_column_optimization` setting; the
+  force-load above is additionally gated by
+  `input_format_parquet_use_column_index_for_constant_columns`.
+- Partial-page subgroup boundaries are fine: a partial overlap of a constant page still yields that
+  value, as long as every overlapping page is constant with the shared value.
+
+## Mixed-topology fill (Approach B) — implemented, default off
+
+`input_format_parquet_fill_constant_pages` handles constant runs *shorter* than / straddling a
+subgroup: `fillConstantPagesAndDecodeRest` fills single-value pages from the Column Index and decodes
+only the varying ones, and `determinePagesToPrefetch` skips prefetching the filled pages (a page
+shared with a subgroup that decodes it normally is still fetched — `willFillConstantPages` is
+deterministic so both paths agree). The fill walks the rows that pass the filter (`row_subgroup.filter`), so it works under PREWHERE /
+row-level filters and page-pruning predicates too - it produces exactly `rows_pass` values for any
+filter, and because the decision no longer depends on `rows_pass` it is identical at prefetch time
+and decode time (so the prefetch-skip can never drop a page the decode needs). All-null pages are
+filled (nulls via the compact values + null map + `expand` path, or the output default under
+`null_as_default`); a subgroup with an all-null page falls back only when the output can represent
+neither null nor a default. The one remaining gate is `needs_cast`: the fill writes the Column Index
+value into the `decoded_type` column, valid only when no post-decode cast applies (resolving whether
+`decodeField` yields the decoded or the output value domain is build-gated). Experimental, off by
+default; needs a build + correctness tests before it can be trusted.

docs/en/engines/database-engines/datalake.md

@@ -59,6 +59,7 @@ The following settings are supported:
 | `region`                | AWS region for the service (e.g., `us-east-1`)                                          |
 | `dlf_access_key_id`     | Access key ID for DLF access                                                            |
 | `dlf_access_key_secret` | Access key Secret for DLF access                                                        |
+| `namespaces`            | Comma-separated list of namespaces, implemented for catalog types: `rest`, `glue` and `unity` |
 
 ## Examples {#examples}
 
@@ -81,4 +82,30 @@ SETTINGS
     onelake_client_secret = client_secret;
 SHOW TABLES IN database_name;
 SELECT count() from database_name.table_name;
-```
+```
+
+## Namespace filter {#namespace}
+
+By default, ClickHouse reads tables from all namespaces available in the catalog. You can limit this behavior using the `namespaces` database setting. The value should be a comma‑separated list of namespaces that are allowed to be read.
+
+Supported catalog types are `rest`, `glue` and `unity`.
+
+For example, if the catalog contains three namespaces - `dev`, `stage`, and `prod` - and you want to read data only from dev and stage, set:
+```
+namespaces='dev,stage'
+```
+
+### Nested namespaces {#namespace-nested}
+
+The Iceberg (`rest`) catalog supports nested namespaces. The `namespaces` filter accepts the following patterns:
+
+- `namespace` - includes tables from the specified namespace, but not from its nested namespaces.
+- `namespace.nested` - includes tables from the nested namespace, but not from the parent.
+- `namespace.*` - includes tables from all nested namespaces, but not from the parent.
+
+If you need to include both a namespace and its nested namespaces, specify both explicitly. For example:
+```
+namespaces='namespace,namespace.*'
+```
+
+The default value is '*', which means all namespaces are included.

src/Common/ErrorCodes.cpp

@@ -658,6 +658,7 @@
     M(776, RESOURCE_LIMIT_EXCEEDED) \
     M(777, MEMORY_RESERVATION_KILLED) \
     M(778, MEMORY_RESERVATION_FAILED) \
+    M(779, CATALOG_NAMESPACE_DISABLED) \
 \
     M(900, DISTRIBUTED_CACHE_ERROR) \
     M(901, CANNOT_USE_DISTRIBUTED_CACHE) \

src/Common/ProfileEvents.cpp

@@ -754,6 +754,9 @@ The server successfully detected this situation and will download merged part fr
     M(S3CopyObject, "Number of S3 API CopyObject calls.", ValueType::Number) \
     M(S3ListObjects, "Number of S3 API ListObjects calls.", ValueType::Number) \
     M(S3HeadObject,  "Number of S3 API HeadObject calls.", ValueType::Number) \
+    M(S3GetObjectAttributes, "Number of S3 API GetObjectAttributes calls.", ValueType::Number) \
+    M(ObjectStorageIdentityCacheHits, "Number of object-storage identity (size/etag/part-offsets) cache hits, avoiding a HEAD/GetObjectAttributes.", ValueType::Number) \
+    M(ObjectStorageIdentityCacheMisses, "Number of object-storage identity cache misses that triggered a metadata request.", ValueType::Number) \
     M(S3GetObjectTagging, "Number of S3 API GetObjectTagging calls.", ValueType::Number) \
     M(S3CreateMultipartUpload, "Number of S3 API CreateMultipartUpload calls.", ValueType::Number) \
     M(S3UploadPartCopy, "Number of S3 API UploadPartCopy calls.", ValueType::Number) \
@@ -767,6 +770,7 @@ The server successfully detected this situation and will download merged part fr
     M(DiskS3CopyObject, "Number of DiskS3 API CopyObject calls.", ValueType::Number) \
     M(DiskS3ListObjects, "Number of DiskS3 API ListObjects calls.", ValueType::Number) \
     M(DiskS3HeadObject,  "Number of DiskS3 API HeadObject calls.", ValueType::Number) \
+    M(DiskS3GetObjectAttributes, "Number of DiskS3 API GetObjectAttributes calls.", ValueType::Number) \
     M(DiskS3GetObjectTagging, "Number of DiskS3 API GetObjectTagging calls.", ValueType::Number) \
     M(DiskS3CreateMultipartUpload, "Number of DiskS3 API CreateMultipartUpload calls.", ValueType::Number) \
     M(DiskS3UploadPartCopy, "Number of DiskS3 API UploadPartCopy calls.", ValueType::Number) \
@@ -1442,11 +1446,18 @@ 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(ParquetConstantColumnSubchunks, "The total number of parquet column subchunks (per row subgroup) materialized from a single value in their per-page column-index statistics, without decoding their data pages", 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(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(ParquetPrefetcherServedFromRetainedTail, "The number of ranges (e.g. Column/Offset Index) served from the retained footer tail by DB::Parquet::Prefetcher without issuing a read", ValueType::Number) \
+    M(ParquetPrefetcherPartAlignedTasks, "The number of read tasks whose coalescing was constrained to a single S3 multipart-upload part boundary by DB::Parquet::Prefetcher", ValueType::Number) \
+    M(ParquetPrefetcherAlignmentSkippedSmall, "The number of times DB::Parquet::Prefetcher skipped read alignment because the aligned segment would be smaller than the configured minimum (anti-fragmentation)", ValueType::Number) \
+    M(ParquetPrefetcherHedgedReads, "The number of hedged (duplicate) reads issued by DB::Parquet::Prefetcher to cut read tail latency", ValueType::Number) \
+    M(ParquetPrefetcherHedgedWins, "The number of hedged reads that produced the result used (beat or replaced the primary read) in DB::Parquet::Prefetcher", 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

@@ -206,6 +206,36 @@ 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.
+)", 0) \
+    DECLARE(Bool, input_format_parquet_use_column_index_for_constant_columns, false, R"(
+Load the Parquet Column Index for read columns that have no predicate of their own, so the constant-column optimization can also skip data pages that are single-valued over a row subgroup (not just over a whole column chunk). Costs a small extra read of the (tiny) Column Index; only worthwhile when columns are sorted or low-cardinality. Applies only when `input_format_parquet_use_constant_column_optimization` is enabled.
+)", 0) \
+    DECLARE(Bool, input_format_parquet_fill_constant_pages, false, R"(
+Experimental. When a Parquet column is single-valued over some data pages but not the whole row subgroup, fill those pages' rows from the per-page Column Index statistics instead of reading and decoding them (mixed-topology subgroups). Extends the constant-column optimization below the subgroup granularity. Requires `input_format_parquet_use_constant_column_optimization`; disabled by default.
+)", 0) \
+    DECLARE(Bool, input_format_parquet_align_reads_to_multipart_boundaries, false, R"(
+Experimental. Align coalesced Parquet read requests to the boundaries of the object's S3 multipart-upload parts, so a single read never straddles two parts (an AWS best practice). Requires the per-file multipart layout, learned via GetObjectAttributes and cached (see the object-storage identity cache); has no effect for single-part objects or stores that don't expose part info. Disabled by default.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_read_alignment_bytes, 0, R"(
+Experimental. Align coalesced Parquet read requests to a fixed byte grid of this size, so no read straddles a multiple of it. Set to the writer's multipart part size (e.g. 10Mi for delta-rs, 64Mi for Spark/S3A) to keep reads within single parts without probing per-file layout. 0 disables. When `input_format_parquet_align_reads_to_multipart_boundaries` is enabled and the real per-file part layout is known, that takes precedence over this fixed grid.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_read_alignment_min_bytes, 1048576, R"(
+Experimental. Anti-fragmentation guard for Parquet read alignment (`input_format_parquet_read_alignment_bytes` / `input_format_parquet_align_reads_to_multipart_boundaries`): do not cut a read at a boundary when the resulting aligned segment would be smaller than this many bytes; allow the straddle instead of emitting a tiny extra request. Default 1 MiB.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_hedged_read_threshold_ms, 0, R"(
+Experimental. Tail-latency mitigation for the Parquet v3 reader on remote object storage: if a read a query is blocked on has not completed within this many milliseconds, issue a duplicate (hedged) request and use whichever returns first. Cuts the S3 GET p99 tail at the cost of a few extra requests. 0 disables. Only reads no larger than `input_format_parquet_hedged_read_max_bytes` are hedged, and at most `input_format_parquet_hedged_read_max_inflight` hedges run at once.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_hedged_read_max_bytes, 4194304, R"(
+Experimental. Only hedge Parquet reads (see `input_format_parquet_hedged_read_threshold_ms`) no larger than this - hedging targets latency of small/critical reads, not throughput of large coalesced reads. 0 = no size limit. Default 4 MiB.
+)", 0) \
+    DECLARE(UInt64, input_format_parquet_hedged_read_max_inflight, 4, R"(
+Experimental. Cap on concurrent hedged Parquet reads (see `input_format_parquet_hedged_read_threshold_ms`), so a slow region cannot double all traffic. Default 4.
+)", 0) \
+    DECLARE(Double, input_format_parquet_prefetch_bandwidth_hide_seconds, 0, R"(
+Read back-pressure for the Parquet v3 reader. When greater than zero, stop prefetching more compressed data pages ahead of decoding once the in-flight compressed bytes exceed this many seconds' worth of the measured read throughput (i.e. once the storage link is kept busy). Prevents buffering compressed data far beyond what bandwidth can consume. 0 disables the back-pressure (compressed prefetch is then bounded only by its memory budget).
 )", 0) \
     DECLARE(Bool, input_format_parquet_verify_checksums, true, R"(
 Verify page checksums when reading parquet files.