[Experiment] Antalya 26.6: Parquet v3 constant column skip

[UnamedRus]

Changelog category (leave one):

• Performance Improvement

Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):

...

Documentation entry for user-facing changes

...

CI/CD Options

Exclude tests:

• Fast test
• Integration Tests
• Stateless tests
• Stateful tests
• Performance tests
• Aarch64 tests
• All with ASAN
• All with TSAN
• All with MSAN
• All with UBSAN
• All with Coverage
• All Regression
• Disable CI Cache

Regression jobs to run:

• Fast suites (mostly <1h)
• Aggregate Functions (2h)
• Alter (1.5h)
• Benchmark (30m)
• ClickHouse Keeper (1h)
• Iceberg (2h)
• LDAP (1h)
• OAuth (5m)
• Parquet (1.5h)
• RBAC (1.5h)
• SSL Server (1h)
• S3 (2h)
• S3 Export (2h)
• Swarms (30m)
• Tiered Storage (2h)

[github-actions]

Workflow [PR], commit [83da1a8]

[UnamedRus]

RowGroup level + push ColumnConst to transforms

┌───────────────────────────────────┬──────────┬─────────────┬───────────────┐
│              Metric               │ baseline │ ColumnConst │       Δ       │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ DefaultImplementationForNullsRows │ 3.367B   │ 0.718B      │ −78.7%        │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ RealTimeMicroseconds (wall)       │ 333.5M   │ 243.1M      │ −27.1%        │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ UserTimeMicroseconds              │ 21.17M   │ 18.22M      │ −13.9%        │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ OSCPUVirtualTimeMicroseconds      │ 23.93M   │ 21.04M      │ −12.1%        │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ S3GetObject                       │ 663      │ 467         │ −196 (−29.6%) │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ ParquetPrefetcherReadRandomRead   │ 662      │ 466         │ −196 (−29.6%) │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ ParquetFetchWaitTimeMicroseconds  │ 104.5M   │ 69.0M       │ −34%          │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ ReadBufferFromS3InitMicroseconds  │ 63.1M    │ 37.0M       │ −41%          │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ MemoryAllocatedWithoutCheckBytes  │ 486.8M   │ 359.3M      │ −26.2%        │
├───────────────────────────────────┼──────────┼─────────────┼───────────────┤
│ SelectedBytes                     │ 2214.1M  │ 2115.7M     │ −4.4%         │
└───────────────────────────────────┴──────────┴─────────────┴───────────────┘

[UnamedRus]

Object identity cache

ClickHouse have multiple caches for Parquet/Iceberg/Data, but some of them are cached on: object_path, etag

This does make sense for certain use cases with s3, (not to use expired cache if object is changed)
But it lead to creation of 2 head requests per object before hitting cache:

1. CH know object path, but don't know etag -> HEAD request
2. CH know object path + etag -> cache.get (MISS, because 1. didn't put result in cache) -> HEAD request

But, this protection (against expired object) doesn't make sense for iceberg tables model, because in their consistency model, they do not store etag value in manifest lists. So, even if object is replaced we will not know/detect that.

So, idea was to introduce very small cache just for those engines, which does not fix in contract value of etag.

 path → {exists, size, etag/version, mtime} 

And use it, in order to probe those caches, which use object_path + etag as key.

[UnamedRus]

Parquet footer

When reading Parquet file:

1. ClickHouse first was reading last 64KB of object (GET 1)
2. Extract footer len from it
3. Read rest of footer (GET 2)

So, total latency was (GET 1) + (GET 2)
What we can do better.
There is little to no latency difference for relatively small requests.

Single GET — TTFB-bound, size barely matters

  ┌───────┬──────┬──────┬──────┐
  │ size  │ p50  │ p90  │ p99  │
  ├───────┼──────┼──────┼──────┤
  │  64 K │ 25.3 │ 31.5 │ 59.3 │
  ├───────┼──────┼──────┼──────┤
  │ 256 K │ 26.7 │ 33.0 │ 60.2 │
  ├───────┼──────┼──────┼──────┤
  │ 512 K │ 28.2 │ 42.2 │ 75.8 │
  ├───────┼──────┼──────┼──────┤
  │  1 MB │ 29.9 │ 39.9 │  178 │
  ├───────┼──────┼──────┼──────┤
  │  2 MB │ 32.9 │ 46.8 │  160 │
  └───────┴──────┴──────┴──────┘

So, we either can be bit more generous and just read more data for GET 1 request (in hope that whole footer will land in it)
Or/And we can try to estimate parquet footer size from column num, parquet file size, row count. (what is available for us from iceberg metadata).

Another option, in case of big estimates, do 2 requests in parallel:

1. GET 1 is last 2 MB of file
2. GET 2 is -2MB to rest of our estimate

┌────────┬─────┬────────────┬─────────────────┬──────────────┐
│ footer │ RTs │ oracle p50 │ spec_serial p50 │ spec_par p50 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│  256 K │   1 │       26.7 │            33.6 │         34.3 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│   1 MB │   1 │       30.1 │            33.9 │         33.1 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│   2 MB │   1 │       33.7 │            33.0 │         33.9 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│   3 MB │   2 │       36.7 │            65.0 │         36.1 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│   4 MB │   2 │       46.3 │            68.0 │         37.6 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│   6 MB │   2 │       68.3 │            81.2 │         47.5 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│   8 MB │   2 │       91.3 │           103.9 │         68.6 │
└────────┴─────┴────────────┴─────────────────┴──────────────┘

Below 2 MB (the common case) — 1 GET, essentially free

spec ≈ oracle ≈ ~33 ms. Over-reading to 2 MB when the footer is 256 K costs only +7 ms (TTFB-bound, transfer negligible). So a 2 MB estimate handles the typical footer in one round-trip.

Above 2 MB — the serial 2nd GET hurts, parallel fixes it

┌────────┬──────────────────┬──────────────────┬────────────────┐
│ footer │ serial vs oracle │  par vs oracle   │ parallel saves │
├────────┼──────────────────┼──────────────────┼────────────────┤
│   3 MB │ +28.3 ms (1.77×) │  −0.6 ms (0.98×) │        28.9 ms │
├────────┼──────────────────┼──────────────────┼────────────────┤
│   4 MB │ +21.7 ms (1.47×) │  −8.7 ms (0.81×) │        30.4 ms │
├────────┼──────────────────┼──────────────────┼────────────────┤
│   6 MB │ +12.9 ms (1.19×) │ −20.8 ms (0.70×) │        33.7 ms │
├────────┼──────────────────┼──────────────────┼────────────────┤
│   8 MB │ +12.6 ms (1.14×) │ −22.6 ms (0.75×) │        35.3 ms │
└────────┴──────────────────┴──────────────────┴────────────────┘

[UnamedRus]

Guard for *_min_bytes_for_seek

ClickHouse glue reads in one GET request, if there is less than 4MB gap between them
Basically, when we need to read nicely compressed columns, due to shape of Parquet files, it's possible that we would end up with:
(need) 64kb....4MB(of unneeded columns)...(need) 128kb....4MB(of unneeded columns)...(need) 64k...

Which in general, was tradeoff we were willing to accept, but as always there should be some limit to that.
Try to bridge blocks in very few GET requests, shouldn't lead to reading, which return 90%-95% of bridge and only 10% of what we actually need.

So, idea was to allow creation of more GET requests, for some pathological cases.

[UnamedRus]

Hedged requests

Send duplicate GET request, if original is not answering (TTFB? total latency?) for longer than X
Alone, did show some usefulness(but may be with latency aware host selector will be less usable/trigger less frequently), but not tested well, and actually should use TTFB as metric.

Parallelize iceberg manifest calculation

Done in #1753

Serial: one producer thread Avro-decodes + min/max-evaluates all ~17.5k manifest entries before any reader starts. Parallel (=16): fan that over N threads → cut the startup stall. Measured via IcebergMetadataReadWaitTimeMicroseconds (summed across reader threads) and wall.

Measured — production binary qbench

Serial workload (max_threads=16, uncontended):

┌─────────────────────────┬───────────────────────┬──────────┐
│                         │ metawait (µs, summed) │ wall p50 │
├─────────────────────────┼───────────────────────┼──────────┤
│ baseline (threads=1)    │                11.3 M │  1973 ms │
├─────────────────────────┼───────────────────────┼──────────┤
│ icebergpar (threads=16) │                 3.7 M │  1846 ms │
├─────────────────────────┼───────────────────────┼──────────┤
│ Δ                       │    −67% (3× realized) │    −6.4% │
└─────────────────────────┴───────────────────────┴──────────┘

Small wall gain — when uncontended the metadata phase is a minor fraction of wall, and only ~3× of the 16 threads is realized (producer setup + shared-manifest serialization overhead).

Concurrent workload (queries overlapping):

┌─────────────────────────┬───────────────────────┬──────────┐
│                         │ metawait (µs, summed) │ wall p50 │
├─────────────────────────┼───────────────────────┼──────────┤
│ baseline (threads=1)    │              ~46–65 M │  4882 ms │
├─────────────────────────┼───────────────────────┼──────────┤
│ icebergpar (threads=16) │              ~2.5–8 M │  3065 ms │
├─────────────────────────┼───────────────────────┼──────────┤
│ Δ                       │ ~−90% (≈11× realized) │     −37% │
└─────────────────────────┴───────────────────────┴──────────┘

[UnamedRus]

Constant Column optimization

Iceberg

Iceberg store metadata/statistics for some columns.
So, we can understand that some column in whole parquet file have the same value, so:

1. We can avoid reading ranges associated with it.
2. Produce it to next ClickHouse Transforms in pipeline as const column in chunk, which allow to process it more efficiently for some operations.

Done in #1069

Parquet

Parquet in footer themself store statistics for all columns in file for RG.
There are certain variations of it(depends on writers), but in general
(min_value == max_value AND (null_count == 0 OR (definition level = required)) or null_count == num_values
For non numerics is_min_value_exact/is_max_value_exact is true

is safe to use., then again we can avoid read this column and push it as const in chunk.

Another case,
If column is not constant in RG, but constant in some of Pages in chunk, and Parquet have Column Index
It store

  Per column chunk, arrays indexed by page:
  - min_values[], max_values[] — one min/max per data page (sort-order-aware, same semantics as min_value).
  - null_pages[] — bool per page: page is entirely null.
  - null_counts[] (optional) — per-page null count.

For non numerics, if min_values/max_values are truncated, max_value should be truncated-and-increment.
SO, if min_values[i] = max_values[i], we know for sure that there is only one value in page.

And again, we can do the same optimization (not read page/range of pages), create const column in chunk

[ianton-ru]

I count at least 5 different independent features in this PR

1. Cache for S3 object metadata
2. Host resolving with weights based on latency
3. Hint for parquet footer size
4. Parallel parquet file loading
5. Constant columns
   Plus namespace filter from #1019

This is not PR for merging, only as a proof of concept.

[ianton-ru]

Does this make something faster?
getObjectInfo makes HeadObject request, getObjectIdentity makes GetObjectAttributes with fallback to getObjectInfo.
As I understand, GetObjectAttributes give profit if replaced several different requests, but in current case it replaces only one.
Fallback makes only worse for some old S3-compatible storages without GetObjectAttributes support.

[UnamedRus]

Does this make something faster?

No, it's not faster.
It was attempt to gather/cache real multiPart boundaries from S3 via GetObjectAttributes.
But, it need special conditions (multiPart upload with extra checksums) AND have access to GetObjectAttributes.

So, probably if it will be implemented using Head is more reliable.

[ianton-ru]

As I understand this caches also metadata during direct reading with S3 table function. Cache is good for immutable objects, "owned" by ClickHouse, but not for all objects on S3.

[UnamedRus]

As I understand this caches also metadata during direct reading with S3 table function.

It might, and if so, need to be fixed, but doesn't affect bench for now.

Cache is good for immutable objects, "owned" by ClickHouse, but not for all objects on S3.

I think, we should do it for "any" immutable kind of object. (either ClickHouse owned or Iceberg catalog).
We potentially could do it for any object of any source, if we use conditional GET later (GET object/range if etag match X, where X is cached value, so we can get cache invalidation & data request in one go, but it's unnecessary for Iceberg)

[ianton-ru]

Does this number have grounding? From my point of view, guessing "set initial footer size based on random constant number of rows per group" is not worse and not better than "set initial footer size as 64 kB"

[UnamedRus]

is not worse and not better than "set initial footer size as 64 kB"

Requesting larger footer, up to 2MB is basically free.
And if we can save 1 GET request almost for free...

┌────────┬─────┬────────────┬─────────────────┬──────────────┐
│ footer │ RTs │ oracle p50 │ spec_serial p50 │ spec_par p50 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│  256 K │   1 │       26.7 │            33.6 │         34.3 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│   1 MB │   1 │       30.1 │            33.9 │         33.1 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤
│   2 MB │   1 │       33.7 │            33.0 │         33.9 │
├────────┼─────┼────────────┼─────────────────┼──────────────┤

But, logic for footer estimation can be done better, i think.
We should have list of columns (with types?), file size, number of rows from iceberg metadata.
Spark for example split by default using 128MB of uncompressed data, we can probably estimate that from types x col_num from one side and using compressed parquet size X ("average/good" parquet compression) from another.

[ianton-ru]

So may be just increase default const footer size to 2Mb instead of complex logic?

[UnamedRus]

Yeah, it's one option i keep in mind.

But, for very small files, it probably nice to somewhat scale it down.
Or, for big files, for >2MB estimated footer, we can (would like to) fire 2 requests in parallel, which allow to compress latency as well.

So, i think in the end we can decide on:

1. complex estimator, which overestimate with good enough margin footer size.
2. very simple estimator based on file size (just to reduce size of estimation for small one)
3. just plain request 1-2 MB last bytes

[ianton-ru]

AI found a lot of issues for most features, but I think make sense split PR on pieces, one feature in one PR, and recheck these PRs.

[UnamedRus]

This is not PR for merging, only as a proof of concept.

Yes, and it was intendent as such.
I've need builds to test it on benchmark, and if we deem some ideas useful, we need to understand which one are more useful than others to find best value order of implementation.

but I think make sense split PR on pieces, one feature in one PR, and recheck these PRs.

I think, it bit early to do that. First we need to find (from bench on real data) which one we really (if any) need.