Visitar URL original
IcebergLocal: aggregates on the partition column scan all data files (no metadata shortcut) · Issue #2731 · Altinity/ClickHouse · GitHub
Skip to content

IcebergLocal: aggregates on the partition column scan all data files (no metadata shortcut) #2731

Open

Description

@minguyen9988

Describe the bug

For an Iceberg table partitioned by a date column d, aggregate queries that only need the partition column read every row of every matching data file. The partition values are already in the Iceberg manifests, and SELECT count() is already answered without a scan, but no other aggregate shape uses that metadata.

Reproduced on upstream v26.9.12.8-stable and v26.10.3.7-stable.

On a 10,000-row table with 10 daily partitions (PARTITION BY (d)), one row per shape:

Query shape Read rows Should read
SELECT max(d) FROM t 10000 ~1 (manifest upper bound)
SELECT d FROM t GROUP BY d 10000 ~10 (one per partition)
SELECT d, count() FROM t GROUP BY d 10000 ~10
SELECT d, count() FROM t WHERE d >= '2026-01-09' GROUP BY d 2000 (the 2 matching partitions, all rows of each) ~2
SELECT max(d) FROM (SELECT d, count() FROM t GROUP BY d) 10000 ~10
SELECT count() FROM t 1 1 (already optimized)

Partition pruning itself works: the ranged query touches only the 2 matching partitions. The gap is that, once inside a partition, every row is read to produce a value the manifest already records.

To Reproduce

One server on loopback, no object storage, no catalog. Replace the placeholder ABS with an absolute path.

Write this config to ABS/config.xml:

<clickhouse>
  <path>ABS/store/</path>
  <tmp_path>ABS/tmp/</tmp_path>
  <user_files_path>ABS/user_files/</user_files_path>
  <listen_host>127.0.0.1</listen_host>
  <tcp_port>19000</tcp_port>
  <logger><level>warning</level><console>1</console></logger>
  <query_log><database>system</database><table>query_log</table>
    <flush_interval_milliseconds>200</flush_interval_milliseconds></query_log>
</clickhouse>

Create the directories ABS/store, ABS/tmp and ABS/user_files, start the server with clickhouse server --config-file=ABS/config.xml, then run:

SET allow_insert_into_iceberg = 1;
CREATE TABLE t (id Int64, d Date32, v String)
ENGINE = IcebergLocal('ABS/user_files/lake/t/')
PARTITION BY (d) SETTINGS iceberg_format_version = 2;
INSERT INTO t
SELECT toInt64(number),
       toDate32('2026-01-01') + toInt32(intDiv(number, 1000)),
       concat('v', toString(number))
FROM numbers(10000);

Then run each shape with a distinguishing log_comment and read what the server actually read:

SELECT max(d) FROM t
  SETTINGS log_comment = 'max_d';
SELECT d, count() FROM t GROUP BY d ORDER BY d
  SETTINGS log_comment = 'groupby_count';
SELECT d, count() FROM t WHERE d >= '2026-01-09' GROUP BY d ORDER BY d
  SETTINGS log_comment = 'groupby_ranged';
SELECT d FROM t GROUP BY d ORDER BY d
  SETTINGS log_comment = 'groupby_nocount';
SELECT max(d) FROM (SELECT d, count() FROM t GROUP BY d)
  SETTINGS log_comment = 'max_subquery';
SELECT count() FROM t
  SETTINGS log_comment = 'count_star';
SYSTEM FLUSH LOGS;
SELECT log_comment, sum(read_rows) AS read_rows, sum(result_rows) AS result_rows
FROM system.query_log
WHERE type = 'QueryFinish' AND log_comment != ''
GROUP BY log_comment ORDER BY log_comment;

Observed on v26.9.12.8-stable and again on v26.10.3.7-stable:

┌─log_comment─────┬─read_rows─┬─result_rows─┐
  1. │ count_star │ 1 │ 1 │
  2. │ groupby_count │ 10000 │ 10 │
  3. │ groupby_nocount │ 10000 │ 10 │
  4. │ groupby_ranged │ 2000 │ 2 │
  5. │ max_d │ 10000 │ 1 │
  6. │ max_subquery │ 10000 │ 1 │
    └─────────────────┴───────────┴─────────────┘

EXPLAIN confirms no metadata path is chosen for any of them:

SELECT count() FROM t                  -> ReadFromPreparedSource (Optimized trivial count)
SELECT d, count() FROM t GROUP BY d    -> ReadFromObjectStorage
SELECT max(d) FROM t                   -> ReadFromObjectStorage
SELECT d FROM t GROUP BY d             -> ReadFromObjectStorage

Expected behavior

The partition column value for a row is fully determined by the partition spec, and min/max of it inside a partition are recorded as the partition field upper and lower bound in the manifest. So:

  • max(d) and min(d) on a partition column can be answered from the manifest bounds: one value per manifest entry, no data file read.
  • GROUP BY d with or without count() can be answered from the partition values plus the per-file record_count in the manifest.
  • The same holds when either shape appears as a subquery, and with a range predicate on d (which already prunes correctly; only the row read inside each surviving partition is wasted).

Root cause

IcebergMetadata::totalRows is the only metadata-driven row-count entry point, and it is reached only through the trivial count() optimization (it increments ProfileEvents::IcebergTrivialCountOptimizationApplied). Its own guard already handles the hard cases correctly: it bails out to a real scan when the snapshot carries equality or position delete files, and when a manifest reports a negative record_count. That guard is the model the other shapes need.

There is no equivalent path that answers min/max of a partition column from the manifest bounds, or GROUP BY of the partition column with count() from partition values plus record_count.

Proposed fix

Direction, not a tested patch:

  • For min/max of a partition column: use the partition field lower and upper bound from the manifest entry, with the same bail-out conditions totalRows already applies (delete files present; missing or inconsistent metadata).
  • For GROUP BY of the partition column with count(): aggregate record_count per partition value from the manifest entries, keeping the same guard.
  • Both must fail back to a real scan whenever the metadata cannot answer exactly, rather than returning an approximate result. The existing totalRows behaviour, which verifies the snapshot summary against the manifest sum and warns on a mismatch, is the right precedent.

Key information

  • Upstream v26.9.12.8-stable and v26.10.3.7-stable: the same numbers on both, shown above. Also reproduces on the 26.6 lineage.
  • Data is on the local filesystem with the IcebergLocal engine. No object storage and no catalog are involved, so the cost is purely the data-file scan.
  • Both shapes return correct results today; this is a read-efficiency defect, not a wrong-answer defect.
  • The same table partitioning in MergeTree answers max(d) and the ranged GROUP BY d from the _minmax_count_projection, so the two engines behave differently for the same logical query.

How this affects us

We read Iceberg tables of about 186 billion rows in about 200 daily partitions. Dashboard queries that ask for the newest partition or the row count per day for a recent window cost a scan of tens of billions of rows even though the answer is in the table metadata:

Query Time Rows read
SELECT Date, count() ... GROUP BY Date, single node 14 ms 200
SELECT max(Date) FROM (SELECT Date, count(*) ... GROUP BY Date), distributed 11.6 s most of the table (the initiator alone reported 90 billion, Date column only)
SELECT Date, count() ... WHERE Date >= a date about 2.5 months back GROUP BY Date, single node 6.3 s 46.7 billion
SELECT max(Date) ... WHERE Date >= a date about 2.5 months back 2.4 s distributed / 5.4 s single node 46.7 billion

Related

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions