Apache Iceberg Query Optimization: Production Guide

Every query against an Iceberg table is a negotiation about how much data the engine is allowed to ignore. Read the metadata, throw away everything that cannot match, scan what survives. A well-maintained table lets the planner discard 95–99% of files before a single Parquet footer is opened. A neglected table forces a near-full scan no matter how selective your WHERE clause looks.

That gap is almost never the engine's fault. Trino, Spark, Snowflake, and DuckDB all run the same fundamental pruning logic against the same Iceberg metadata. What differs is whether your physical layout and statistics give that logic anything to work with. A customer_id = 481 filter is worthless if every file in the table contains customer IDs 1 through 10,000,000, and no amount of cluster tuning will fix it.

This guide covers the entire surface: how Iceberg's scan planner actually works and what to measure inside it, how statistics get collected (and silently dropped), the difference between partitioning and clustering, why declaring a sort order often changes nothing, how bloom filters and Puffin statistics fit, what format v3 deletion vectors change for reads, and the engine-specific knobs that matter on Trino and Spark. It ends with a triage runbook and a checklist.

The Short Version

If you only read one section, read this table. It maps the symptom you are staring at to the mechanism that causes it.

Symptom	Most likely cause	Fix
Selective WHERE still scans most of the table	File min/max bounds overlap; data is not clustered on the filter column	Global sort or Z-order during compaction (not just binpack)
Seconds of latency before the first byte is read	Manifest fragmentation, or manifests clustered on the wrong partition field	rewrite_manifests with sort_by on the hot transform
Pruning does nothing and bounds are simply absent	Metrics mode is none, or the column fell past the 100-column inference cap	write.metadata.metrics.column.<col> = truncate(16)
Point lookup on a high-cardinality column is slow	Min/max cannot prune a scattered key	Parquet bloom filter on that column
Table got slower after a CDC backfill	Position or equality delete accumulation	Compact with a low delete-file-threshold; move to v3 deletion vectors
Joins pick catastrophic orders	Missing or stale Puffin NDV, or CBO is switched off	compute_table_stats / ANALYZE, refreshed after every compaction
Sort order is declared but nothing improved	Writers applied a local sort, so file ranges still overlap	Set write.distribution-mode to range, or sort globally during rewrite

That last row is the one that costs teams the most time, and it gets its own section below.

How Iceberg Actually Plans a Query

Iceberg planning is a funnel, and the table spec defines each stage precisely. Each stage is cheaper than the one beneath it, and each stage's usefulness depends entirely on the quality of the metadata written at commit time. Knowing the stages by name matters, because the diagnostics and the fixes are different at each level.

Stage 1: Manifest list filtering

Planning starts at the manifest list, a single Avro file that indexes every manifest in the snapshot. Each entry carries a partitions array of field_summary structs: lower_bound, upper_bound, contains_null, and contains_nan for each partition field across all data files that manifest tracks.

Iceberg converts your query predicate into partition space using inclusive projection, a deliberate widening that guarantees no file which might match is ever excluded. For a table partitioned by days(event_time), the filter event_time >= TIMESTAMP '2026-03-01' projects to a bound on event_time_day. ManifestEvaluator then tests that projected predicate against each manifest's summary. Manifests covering only January and February are dropped whole, along with every file they describe.

This is the cheapest prune available, and it is the one hidden partitioning buys you. Users filter on event_time; the planner does the transform arithmetic. Change the partition spec later and the queries keep working, because the projection is computed per spec at plan time.

Stage 2: Data file filtering

For each surviving manifest, the planner reads individual file entries. Every entry carries per-column metrics: lower_bounds, upper_bounds, null_value_counts, nan_value_counts, value_counts, and column_sizes, all keyed by field ID.

InclusiveMetricsEvaluator tests the predicate against those metrics and answers one question: could this file contain a matching row? It returns false only when the metrics prove no match. One detail dominates everything else in this guide:

If a column has no bounds recorded, InclusiveMetricsEvaluator returns true. Missing statistics do not fail loudly. They silently disable file pruning for that column.

Its counterpart, StrictMetricsEvaluator, answers the opposite question: do all rows in this file match? That powers metadata-only deletes and residual computation rather than read pruning.

A note on value_counts, because it is widely misdescribed: it is the total row count for that column including nulls and NaNs, not the count of non-null values. Null-aware pruning comes from comparing null_value_counts against record_count.

Stage 3: Residual evaluation

This stage is invisible in most articles and is the reason partition-aligned filters are effectively free.

Once a file survives, Iceberg computes a residual: the part of the predicate that still needs row-level evaluation after the partition value is taken into account. Take a table partitioned by days(ts) and a query with ts > TIMESTAMP '2026-03-01 08:00:00'.

• For a file in partition ts_day = 2026-03-05, strict projection proves every row already satisfies the predicate. The residual collapses to alwaysTrue() and the engine skips per-row filter evaluation entirely.
• For the boundary partition ts_day = 2026-03-01, the residual stays ts > '2026-03-01 08:00:00' and must be checked per row.

In Trino's EXPLAIN output you can see this split directly: predicates absorbed into the scan's constraint are enforced by Iceberg and never re-evaluated, while a residual Filter node above the scan is per-row work. If you see your main predicate in the Filter node rather than the constraint, the filter is not partition-aligned.

Stage 4: Inside the Parquet file

Once a file is opened, Iceberg's reader applies three row-group filters in a deliberate order, cheapest first:

1. ParquetMetricsRowGroupFilter — per-row-group min/max and null counts from the footer.
2. ParquetDictionaryRowGroupFilter — if the column chunk is dictionary-encoded, the dictionary is an exact membership test. No value in the dictionary means no matching row, with no false positives.
3. ParquetBloomRowGroupFilter — a probabilistic test, only consulted if the first two could not rule the group out, because it costs an extra read.

Worth knowing precisely: Iceberg's own reader prunes at row-group granularity only. It does not use Parquet's column index / offset index to skip individual pages. Trino uses its own Parquet reader and does apply page-level filtering, which is one real, structural reason the same table can prune differently on Trino than on Spark.

Column projection

Iceberg pushes the projected column set into the reader, fetching only the columns in your SELECT plus those referenced by predicates. Projection matches on stable field IDs, not names, so renames and schema evolution never break it. On a 300-column table, projection alone routinely removes 90%+ of I/O, which is why SELECT * on wide tables is an expensive habit rather than a stylistic one.

Where planning runs, and why it can be the bottleneck

Planning cost scales with the number of manifests opened, not the amount of data returned. Three execution models exist today:

• Trino plans on the coordinator, multithreaded. iceberg.planning-threads (default 2× coordinator cores) controls manifest reads and iceberg.split-manager-threads controls split generation. There is no distributed planning, so a badly fragmented table turns the coordinator into a bottleneck for the whole cluster.
• Spark supports distributed planning via spark.sql.iceberg.data-planning-mode and spark.sql.iceberg.delete-planning-mode, each accepting AUTO, LOCAL, or DISTRIBUTED and defaulting to AUTO. In DISTRIBUTED mode, manifest scanning becomes a Spark job over the entries metadata table. AUTO decides based on estimated manifest volume.
• Server-side scan planning shipped in Iceberg 1.11.0 and is the development most guides still describe as "proposed." The REST catalog spec defines planTableScan, fetchPlanningResult, cancelPlanning, and fetchScanTasks endpoints, with a client-side scan-planning-mode property (client or server) and per-table override via LoadTableResult. Spark can now delegate planning to the catalog. Before you plan around it, check what your catalog actually advertises in GET /v1/config under endpoints; vendor support is uneven and claims run ahead of implementations.

Measure planning before you tune it

Do not guess at which stage is failing. Iceberg emits a ScanReport through its MetricsReporter interface with exactly the fields you need:

Metric	What it tells you
totalPlanningDuration	Planning latency, isolated from execution
scannedDataManifests / skippedDataManifests	Whether Stage 1 is working
resultDataFiles / skippedDataFiles	Whether Stage 2 is working
resultDeleteFiles / skippedDeleteFiles	Your merge-on-read burden
totalFileSizeInBytes	Bytes the scan committed to reading

The ratio skippedDataManifests / scannedDataManifests is the single best signal for manifest clustering quality, and skippedDataFiles / (skippedDataFiles + resultDataFiles) is the best signal for sort-order quality. In Trino, system.runtime.queries separates analysis_time, planning_time, and execution_time, and EXPLAIN ANALYZE VERBOSE gives you physicalInputBytes per operator. In Spark, the SQL tab's BatchScan node reports files read and files skipped directly.

The Layer Nobody Owns

Here is the structural problem the rest of this guide keeps running into.

Iceberg specifies how pruning works. Engines implement it. Neither decides what your table's physical layout should be. Nothing in the format observes that 80% of production queries filter on region and reorders your files accordingly. Nothing notices that last night's compaction invalidated your Puffin statistics. Nothing tracks that a table's manifest count crossed the threshold where planning latency became user-visible. Those are continuous operational decisions, and the open lakehouse ships them as your problem.

At ten tables you solve it with a Spark job and a cron entry. At a few hundred, across several engines and catalogs, you are running a control loop by hand with no telemetry, and the decisions go stale faster than you can revisit them.

A lakehouse control plane such as LakeOps is the layer that runs that loop. It sits above the stack you already have rather than replacing any of it: your catalogs (AWS Glue, DynamoDB-backed, REST catalogs like Polaris, Nessie, Gravitino, and Lakekeeper, or S3 Tables), your Iceberg tables in object storage, and your query engines. It attaches through standard catalog and Iceberg metadata APIs, so there is no data copy, no pipeline rewrite, and no change to the table format.

The loop it runs is the one the open stack leaves unowned: sense table structure and query telemetry, classify each table's health, plan the maintenance and layout each table needs, execute in dependency order, then measure the result and feed it back into the next decision.

Learn more:

Managed Iceberg Lakehouse | LakeOps

Connect your catalog — every Iceberg table maintained, compacted, and governed across your lakehouse. Trino, Spark, Snowflake, Athena.

lakeops.dev

Iceberg Lakehouse Table Maintenance | LakeOps

Autonomous Iceberg table maintenance for your data lakehouse. Snapshot expiry, orphan cleanup, manifest rewrite, and delete-file resolution — no cron.

lakeops.dev

Later in this guide, once the underlying mechanics are on the table, I will get specific about how that applies to sort-order selection, manifest health, and statistics lifecycle.

Keep reading first. Every fix below is worth understanding whether you automate it or not.

Statistics: The Fuel for Everything

Stage 2 pruning runs entirely on column bounds. If you optimize one thing before touching layout, make it this, because the failure mode is invisible and the fix is a table property.

What gets collected, and what does not

Metrics collection is governed by a small set of table properties:

Property	Default	Effect
write.metadata.metrics.default	truncate(16)	Mode for all columns without an explicit override
write.metadata.metrics.column.<col>	unset	Per-column override; wins over everything
write.metadata.metrics.max-inferred-column-defaults	100	Cap on how many columns get inferred metrics

Modes are none, counts, truncate(L), and full. And then there is the behavior that catches almost everyone:

The 100-column cap applies only to the inferred default. If you explicitly set write.metadata.metrics.default, the cap is bypassed completely. Setting it to 'truncate(16)' on a 2,000-column table looks like a no-op because it is "the default value," but it switches metrics on for all 2,000 columns and can inflate your manifests by an order of magnitude. This is a common self-inflicted wound: someone "tunes" statistics and makes planning dramatically slower.

Two more details from the implementation that are worth internalizing:

• The column count uses projected field IDs, which includes struct fields themselves, not just leaves. A table with 60 leaf columns nested inside 45 structs is already past 100. Selection is pre-order, so it is the columns declared late in a wide schema that silently lose their bounds.
• Sort-order columns are automatically promoted to truncate(16) when the effective default is none or counts. So write.metadata.metrics.default = 'none' still keeps bounds on your declared sort keys. Note the asymmetry: Z-order and Hilbert clustering do not get this promotion, because they do not change the table's declared SortOrder.

The configuration that is actually correct

For wide tables, be explicit in both directions. Turn the default off, then turn metrics on for the columns that matter:

ALTER TABLE db.events SET TBLPROPERTIES (
    'write.metadata.metrics.default' = 'none',
    'write.metadata.metrics.column.event_time' = 'truncate(16)',
    'write.metadata.metrics.column.customer_id' = 'full',
    'write.metadata.metrics.column.region' = 'full',
    'write.metadata.metrics.column.order_id' = 'truncate(16)'
);

This is the one case where setting default explicitly is right, precisely because you want to bypass the inference machinery and take control. Use full for short, high-value keys and truncate(16) for strings. Never use full on a column holding URLs, JSON blobs, or stack traces: an untruncated multi-kilobyte bound gets copied into both the lower and upper bound map for every data file, and the manifests become larger than some of the data.

Verifying which columns actually have bounds

Do not assume. Ask the table. Spark's files metadata table exposes a readable_metrics struct that decodes the binary bounds into their real types:

SELECT
    file_path,
    readable_metrics.customer_id.lower_bound AS cid_min,
    readable_metrics.customer_id.upper_bound AS cid_max,
    readable_metrics.customer_id.null_value_count AS cid_nulls,
    record_count,
    file_size_in_bytes
FROM db.events.files
ORDER BY 2
LIMIT 20;

If cid_min and cid_max come back NULL, that column has no metrics and no query filtering on it will ever prune. If they are populated but span nearly the full domain in every row, metrics exist but the data is not clustered. Those are two different problems with two different fixes, and this query distinguishes them in one shot.

To see the raw picture of which field IDs have bounds at all:

SELECT map_keys(lower_bounds) AS fields_with_bounds
FROM db.events.files
LIMIT 1;

Measuring clustering quality

Overlap is what kills Stage 2. A useful single number is how many distinct lower bounds exist relative to file count:

SELECT
    count(*) AS files,
    count(DISTINCT readable_metrics.customer_id.lower_bound) AS distinct_lower_bounds,
    avg(readable_metrics.customer_id.upper_bound
        - readable_metrics.customer_id.lower_bound) AS avg_range_width,
    max(readable_metrics.customer_id.upper_bound)
        - min(readable_metrics.customer_id.lower_bound) AS global_range
FROM db.events.files;

When avg_range_width approaches global_range, every file covers the whole domain and pruning is mathematically impossible. When distinct_lower_bounds is a small fraction of files, many files start at the same value and their ranges overlap heavily. Either result points at the same fix: cluster the data.

Partitioning: The Coarse Prune

Partitioning decides which groups of files can be ignored wholesale. It is the cheapest prune and the one with the most permanent consequences, because the partition spec shapes how every subsequent write lands.

Choosing transforms

Partition on the columns that appear in most query filters. Time nearly always wins, because nearly every analytical query bounds a time range:

CREATE TABLE events (
    event_id    bigint,
    event_time  timestamp,
    region      string,
    customer_id bigint,
    amount      decimal(12,2),
    payload     string
) USING iceberg
PARTITIONED BY (days(event_time));

The available transforms are year, month, day, hour, bucket(N, col), truncate(W, col), and identity. Format v3 adds multi-argument transforms for both partitioning and sorting.

Practical guidance that holds up in production:

• Pick granularity from write volume, not from query habit. The target is partitions large enough to hold files near your target file size. A table ingesting 2 GB/day should be partitioned by day, not hour; hourly partitioning gives you 24 partitions of ~85 MB each and you have manufactured a small-file problem.
• bucket(N, col) is for high-cardinality equality predicates, typically a join or lookup key: PARTITIONED BY (bucket(64, customer_id)). It also enables storage-partitioned joins on Spark and bucket-aware execution on Trino.
• truncate(W, col) is underused for string keys with meaningful prefixes, such as truncate(4, country_code) or a hashed ID prefix.
• identity on anything high-cardinality is a trap. PARTITIONED BY (customer_id) on a million customers produces a million partition directories, a million-plus manifest entries, and planning that is slower than the scan you were trying to avoid.

The over-partitioning failure mode

Over-partitioning is the most common self-inflicted Iceberg performance problem, and it is genuinely counterintuitive because each individual partition looks fine. The costs compound in three places at once: more files means more S3 GET requests and more open-file overhead per query; more partition values means more manifest entries to evaluate during planning; and small files means compression ratios and dictionary encoding both degrade, so the bytes on disk grow too.

Check for it directly:

SELECT
    partition,
    count(*)                    AS files,
    avg(file_size_in_bytes)/1048576 AS avg_mb,
    sum(record_count)           AS rows
FROM db.events.files
GROUP BY partition
ORDER BY files DESC
LIMIT 20;

If avg_mb sits well under 100 across most partitions, either the partition granularity is too fine or compaction is not keeping up. Both are fixable, but they are different fixes and the query above tells you which by showing whether the problem is uniform (granularity) or concentrated in recent partitions (compaction lag).

Partition evolution

Iceberg changes the partition spec without rewriting data. Old files keep their old spec; new writes use the new one. The planner evaluates each spec independently at plan time, so both layouts prune correctly and queries need no changes.

ALTER TABLE db.events REPLACE PARTITION FIELD days(event_time) WITH hours(event_time);
ALTER TABLE db.events ADD PARTITION FIELD region;

Two operational caveats. First, a table carrying several specs makes planning modestly more complex and makes partitions metadata harder to reason about, so evolution is not free forever. Second, if you want old data physically moved into the new spec, that is a rewrite_data_files pass with output-spec-id, and it costs a full rewrite of the affected range. Plan it deliberately rather than discovering it on next month's compute bill.

Sort Order and Clustering: Where Queries Are Actually Won

Partitioning determines which file groups exist. Clustering determines whether the files inside those groups are skippable. They are separate layers and collapsing them is the most common conceptual error in Iceberg performance work.

When data is clustered on a predicate column, each file holds a narrow contiguous slice of the value domain, the manifest bounds are tight, and InclusiveMetricsEvaluator eliminates most files. When it is not, every file holds a random sample of the full domain, bounds overlap completely, and Stage 2 becomes a no-op even though the statistics are present and correct.

This is also why binpack compaction is not clustering. Binpack fixes file count. Merging 900 unsorted 5 MB files into 40 unsorted 512 MB files makes your LIST and GET costs better and your planning faster, and changes file-level skipping by approximately nothing. Large unsorted files still fail file pruning. If you have been running scheduled binpack compaction and wondering why query latency never improved, this is why.

Local sort vs global sort: the distinction that decides everything

You can declare a sort order and get zero pruning improvement. This happens constantly and the reason is that where the sort happens matters more than that it happened.

• A local sort orders rows within each writer task. You get N files, each internally sorted, whose [lower_bound, upper_bound] ranges overlap heavily, because each task saw a random sample of the data. InclusiveMetricsEvaluator cannot skip any of them for a point lookup. Local sort buys better compression and better in-file row-group pruning. It does not buy file-level skipping.
• A global sort shuffles first so that each file receives a disjoint value range. Bounds do not overlap, and file-level skipping works.

The control for this is write.distribution-mode:

Mode	Shuffle	Result
none	None	Whatever the upstream partitioning gave you. Risks a small-file explosion unless you pre-sort or enable fanout writes.
hash	Hash exchange on the partition key	Each partition value lands in one task, so file counts stay sane. Does not order rows within the file.
range	Sample, then range-partition	Global ordering across files. Disjoint bounds. Real file skipping. More expensive to write.

Spark's effective default has been hash since Iceberg 1.2.0, and range when the table has a declared sort order. Per-operation overrides exist as write.delete.distribution-mode, write.update.distribution-mode, and write.merge.distribution-mode.

ALTER TABLE db.events WRITE ORDERED BY customer_id, event_time;

ALTER TABLE db.events SET TBLPROPERTIES (
    'write.distribution-mode' = 'range'
);

One more piece of honesty from the spec: writers should apply the declared sort order but are explicitly not required to, with streaming writes called out as the case where it is legitimately too expensive. A declared sort order is a hint, not a guarantee. Verify it against reality rather than trusting the table property:

SELECT sort_order_id, count(*) AS files, sum(file_size_in_bytes)/1073741824 AS gb
FROM db.events.files
GROUP BY sort_order_id;

sort_order_id = 0 means unsorted. A table that declares a sort order and shows mostly zeros is telling you its writers are ignoring it, and the fix is compaction with an explicit sort strategy rather than a table property change.

Choosing sort keys

Order matters, and it matters more than most teams expect. A linear sort gives the leading column tight, nearly disjoint bounds. The second column is only clustered within runs of equal values in the first, so its bounds are meaningfully wider. By the third or fourth column, the clustering benefit is close to noise.

Query pattern	Recommended layout	Reasoning
One dominant filter column	Linear sort, that column leading	Tightest possible bounds on the column that matters
Entity plus time range (customer_id + event_time)	ORDERED BY customer_id, event_time	Entity clustering first, time ordering inside each entity
Joins on a stable key	Sort by the join key, or bucket() partition it	Enables merge joins and storage-partitioned joins
2–4 filter columns used in varying combinations	Z-order	Balanced pruning across dimensions instead of one good dimension and three bad ones
More than 4 filter columns	Pick the top 2–3	Curves dilute; the rest is wishful thinking
High-cardinality point lookups on a non-sort column	Bloom filter, not sort	You only get one leading column; spend it wisely

Z-order, and the gotcha that silently breaks it

Z-order interleaves the bits of several columns into a single Morton code, producing a space-filling curve that clusters data across all of them at once. No single dimension gets bounds as tight as a dedicated linear sort would give it, but every dimension gets usable bounds instead of one winner and several losers.

CALL catalog.system.rewrite_data_files(
    table       => 'db.events',
    strategy    => 'sort',
    sort_order  => 'zorder(customer_id, region, product_id)',
    options     => map(
        'target-file-size-bytes', '536870912',
        'max-concurrent-file-group-rewrites', '10',
        'partial-progress.enabled', 'true'
    )
);

Z-order has two options of its own, and one of them is a landmine:

• var-length-contribution (default 8) is the number of bytes taken from each String or Binary column when building the curve. Only the first 8 bytes participate. If your string keys share a long common prefix, and prefixed UUIDs, hierarchical paths, and evt_2026_03_... style identifiers all do, then every value contributes identical bits and Z-order degenerates into expensive noise. Raise the value, or truncate the column to its distinguishing portion first.
• max-output-size (default 2147483647) caps total interleaved bytes.

Two more properties of curve clustering worth knowing. It is applied per rewrite and is not persisted as the table's SortOrder, so subsequent ordinary writes will not maintain it. And because the declared sort order is unchanged, Z-ordered columns do not receive the automatic truncate(16) metrics promotion described earlier. If your metrics default is restrictive, set explicit per-column metrics on every column you Z-order, or you will do the clustering work and record no bounds to exploit it.

Hilbert curves: the accurate status

Hilbert curves are a genuine improvement over Z-order for multi-dimensional locality, because consecutive points on a Hilbert curve are always true spatial neighbors, eliminating the long jumps a Morton curve makes across the value space.

Be precise about availability, because the internet is not. As of Iceberg 1.11.0, SparkHilbertUDF exists on the project's main branch under spark/v4.1 only. It is not present in any released version, it is absent from spark/v4.0 and spark/v3.5, and it is documented only in the nightly docs, not the release docs. Backports to Spark 4.0 and 3.5 are tracked as an open issue. An earlier 2022 attempt was closed without merging, so blog posts citing it as evidence of support are wrong.

The nightly documentation records one design difference worth noting for when it does ship: unlike Z-order, every column contributes its full 8-byte primitive width to the Hilbert index, so var-length-contribution and max-output-size do not apply. Hilbert and Z-order also cannot be combined with each other or with plain column sorts in a single sort_order.

For production work today, Z-order is the multi-dimensional option. Plan for Hilbert; do not build on it yet.

One thing no Iceberg property can tell you

Declaring a sort order is a one-line DDL statement. Knowing which columns to declare requires knowing what your queries actually filter on, across every engine, every dbt model, every BI tool, and every notebook that touches the table. Most teams choose a sort order at table creation from intuition and never revisit it, while the query mix drifts underneath them for two years.

That is a telemetry problem, not an Iceberg problem, and it is the first place a control plane earns its keep. Because LakeOps sits above all connected engines rather than inside one, it collects WHERE, JOIN, and GROUP BY column frequency from Trino, Spark, Snowflake, Athena, DuckDB, and Flink together, and scores candidate layouts against the combined mix rather than against whichever engine an individual team happens to use. The mechanics of how that turns into an applied sort order come later in this guide.

File Size and Row Group Size

File sizing is usually discussed as a storage-cost topic. It is also a pruning-resolution topic, and the interaction between file size and row group size sets a hard ceiling on how much data any query can skip inside a file.

The relevant defaults:

Property	Default	Notes
write.target-file-size-bytes	536870912 (512 MB)	Compaction target
write.parquet.row-group-size-bytes	134217728 (128 MB)	Granularity of in-file skipping
write.parquet.page-size-bytes	1048576 (1 MB)
write.parquet.page-row-limit	20000	Matters for engines that use the page index
write.delete.target-file-size-bytes	67108864 (64 MB)
read.split.target-size	134217728 (128 MB)	Scan parallelism unit
read.split.open-file-cost	4194304 (4 MB)	Planner's cost model for opening a file

Do the arithmetic on the first two together. A 512 MB file at the default 128 MB row group size contains exactly four row groups, so the best possible in-file skip is 75%, and a point lookup that matches one row still reads a quarter of the file. Drop to 32 MB row groups and you get sixteen groups, finer skipping, a larger footer, and slightly worse compression.

That trade resolves differently by workload, which is the actual guidance:

• Scan-heavy analytical tables: keep 128 MB row groups. You are reading most of the file anyway and you want the compression and the low footer overhead.
• Point-lookup and selective-filter tables: 32–64 MB row groups genuinely help, and they compound with bloom filters, since both operate at row-group granularity.
• Streaming and CDC tables: smaller target files (128–256 MB) reduce write amplification during compaction, at the cost of more files.

Two more layers of parallelism control sit on top. read.split.target-size determines how a large file is divided across tasks, and it uses the split_offsets recorded in each manifest entry, which are the Parquet row group boundaries, so splitting does not require reading footers at plan time. Iceberg 1.11.0 added adaptive split sizing for Spark 4.1 through spark.sql.iceberg.read.adaptive-split-size.enabled, plus a session-level spark.sql.iceberg.read.split-size override. Adaptive sizing fixes the chronic case where a medium scan produces ten splits on a two-hundred-core cluster.

If you use AQE to control output file sizes during writes, note that spark.sql.adaptive.advisoryPartitionSizeInBytes is measured as in-memory row size, not on-disk compressed size. It must be set larger than write.target-file-size-bytes by a data-dependent ratio, and Spark cannot write a file larger than a single task or spanning an Iceberg partition boundary.

Bloom Filters: The Point-Lookup Escape Hatch

Bloom filters solve the one case that min/max statistics structurally cannot: equality predicates on a high-cardinality column that is not your sort key. A trace_id or user_id column with millions of scattered distinct values has bounds spanning the whole domain in every file. Sorting on it would fix pruning but you only get one leading sort column, and it is usually already spent.

A Parquet bloom filter is a compact probabilistic structure per row group. It answers "definitely not present" with certainty and "possibly present" with a tunable false-positive rate. False negatives cannot occur, so it is always safe.

Configuration

ALTER TABLE db.events SET TBLPROPERTIES (
    'write.parquet.bloom-filter-enabled.column.trace_id' = 'true',
    'write.parquet.bloom-filter-fpp.column.trace_id'     = '0.01',
    'write.parquet.bloom-filter-ndv.column.trace_id'     = '5000000',
    'write.parquet.bloom-filter-max-bytes'               = '1048576'
);

bloom-filter-ndv is the property most guides omit and the one that decides whether the filter works. Without an expected distinct-value count, Parquet has to infer the bitset size, and you routinely end up either with a filter so small its false-positive rate approaches 100% (making it useless) or one clipped by bloom-filter-max-bytes (same outcome, wasted bytes). Set it to a realistic per-row-group cardinality estimate. There is also an undocumented write.parquet.bloom-filter-adaptive-enabled (default false) present in TableProperties but absent from the published configuration table.

These properties affect newly written files only. Existing files need a rewrite pass to gain filters:

CALL catalog.system.rewrite_data_files(
    table    => 'db.events',
    strategy => 'binpack',
    where    => 'event_time >= TIMESTAMP ''2026-08-01''',
    options  => map('rewrite-all', 'true')
);

When they help, and when they are pure overhead

Scenario	Useful?	Why
WHERE trace_id = 'a7f...', high cardinality, unsorted	Yes	Min/max is useless; the bloom filter is the only mechanism left
WHERE status = 'active', low cardinality	No	Dictionary filtering already handles this exactly, with no false positives
WHERE amount > 500	No	Bloom filters only answer equality and IN
WHERE customer_id = 481 on the leading sort column	No	Min/max already prunes; the filter is redundant footer bloat
WHERE user_id IN (...) from a runtime join filter	Yes	Spark's runtime filtering pushes these down as equality sets

The sweet spot is two or three columns. Enabling bloom filters across a wide schema bloats footers, slows every write, and adds a read cost on every row group that the first two filter stages could not already resolve.

Which engines read them

• Iceberg's own reader (Spark via the Iceberg data source, Flink, generic readers) uses ParquetBloomRowGroupFilter.
• Trino reads them too, which is often assumed otherwise. The catalog property is parquet.use-bloom-filter, default true, with session property parquet_use_bloom_filter. Trino can also write them for Iceberg tables via the parquet_bloom_filter_columns table property.

Verify empirically rather than trusting either statement: run EXPLAIN ANALYZE on Trino and compare physicalInputBytes before and after writing filters. If the number does not move, the filter is not being consulted for your predicate shape.

Puffin, NDV, and Cost-Based Optimization

Everything above optimizes scanning. Puffin statistics optimize planning decisions: join order, join strategy, and aggregation sizing. On a three-way join, the difference between the best and worst join order can be two orders of magnitude in intermediate data volume, and no amount of file skipping recovers from a plan that builds a billion-row hash table first.

What is actually in a Puffin file

Correcting a persistent misconception: Puffin does not store column min/max bounds. Those live in manifest entries, as covered above. The Puffin spec standardizes exactly two blob types:

1. apache-datasketches-theta-v1 — an Apache DataSketches Theta sketch for NDV (number of distinct values). The estimate is carried in the blob's ndv property.
2. deletion-vector-v1 — the format v3 Roaring bitmap deletion vector, discussed in the merge-on-read section below.

Collecting them

-- Spark
CALL catalog.system.compute_table_stats(
    table   => 'db.events',
    columns => array('customer_id', 'region', 'product_id')
);

-- Trino
ANALYZE db.events WITH (columns = ARRAY['customer_id', 'region']);

The three ways this silently fails

1. Spark computes the stats and nothing uses them. spark.sql.iceberg.report-column-stats defaults to true, so Iceberg does report them. But Spark's cost-based optimizer is gated on spark.sql.cbo.enabled, which defaults to false. You compute sketches, Iceberg hands them over, and the optimizer ignores them. This is the single most common "statistics don't work" report, and the fix is one config:

spark.sql.cbo.enabled=true
spark.sql.cbo.joinReorder.enabled=true

2. Trino ignores multi-column sketches. Trino's reader filters for the standard theta blob type and requires fields().size() == 1, so any sketch covering multiple columns is skipped entirely. It is gated by iceberg.table-statistics-enabled (default true) and maintained on write by iceberg.extended-statistics.collect-on-write (default true). Re-analyzing a subset of columns after a full ANALYZE requires calling drop_extended_stats first, or the new run is rejected.

3. There is no invalidation mechanism at all. This is the design weakness and it deserves emphasis. A Puffin blob pins the snapshot-id it was computed from, and engines walk the snapshot ancestry to find the most recent statistics file. Nothing marks it stale. A sketch computed at snapshot N is served unchanged at snapshot N+10,000, so if the table doubled in size the optimizer is planning against fiction. Worse, a statistics file whose snapshot is no longer an ancestor, after a rollback or a branch rewrite, is simply not found, and the table silently drops to no statistics with no error.

Compaction makes this acute: rewriting files changes the cardinality landscape the sketches describe. Treat statistics refresh as a mandatory step in your maintenance sequence, not an occasional chore. In Trino, SHOW STATS FOR db.events tells you what the optimizer currently believes; compare it against reality before you debug a bad plan. The Puffin statistics lifecycle has more on incremental sketch computation, which matters once full recomputation costs more than the compaction that triggered it.

Manifest Health and Planning Latency

Manifest fragmentation is the most overlooked cause of slow Iceberg queries, because the symptom does not look like a data problem. Queries that should plan in milliseconds take seconds, with perfect partitioning and a good sort order, and the time disappears before any data file is touched.

How fragmentation happens

Every commit, whether append, overwrite, delete, or compaction, produces at least one new manifest. Streaming, micro-batch, and CDC tables accumulate thousands of them, each tracking a handful of files. The planner must open and evaluate every manifest in the snapshot. More manifests means more metadata round trips to object storage, more Avro parsing, and a longer serial-ish critical path before execution begins.

SELECT
    partition_spec_id,
    count(*)                                          AS manifests,
    sum(added_files_count + existing_files_count)     AS files,
    avg(length)/1048576                               AS avg_manifest_mb
FROM db.events.manifests
GROUP BY partition_spec_id;

Manifests well under commit.manifest.target-size-bytes (8 MB) mean fragmentation. Hundreds or thousands of manifests on a table with moderate file counts means your planning time is dominated by metadata I/O.

Rewriting manifests, and the sort_by argument almost nobody uses

Manifest rewriting is metadata-only. It does not read or write data files, which makes it the cheapest maintenance operation available with the largest effect on planning latency.

CALL catalog.system.rewrite_manifests(
    table       => 'db.events',
    sort_by     => array('event_time_day'),
    use_caching => false
);

sort_by arrived in Iceberg 1.11.0 and takes an array of partition transform names, as documented in the Spark procedures reference. It is arguably the highest-leverage planning knob in Iceberg and it is barely known, mainly because the default behavior sounds harmless and is not.

The default sorts manifests by all partition transforms in spec order. Consider a table partitioned by (region, day(ts)) where every production query filters on time and no query filters on region. The default clusters manifests by region first, so each manifest ends up containing files spanning the full time range. The manifest-list field_summary bounds on ts_day overlap across every manifest, and ManifestEvaluator prunes nothing at Stage 1. Passing sort_by => array('ts_day') reorders the clustering to match how the table is actually queried, and Stage 1 starts working.

You can see the effect directly in the ScanReport: skippedDataManifests goes from near zero to most of the manifest count.

Trino's equivalent is ALTER TABLE db.events EXECUTE optimize_manifests, which clusters manifests by partitioning columns and reports rewritten_manifests_count, added_manifests_count, kept_manifests_count, and processed_manifest_entries_count.

Sequencing, and why it is not arbitrary

Maintenance operations feed each other, and running them in the wrong order wastes compute or produces metadata describing files that no longer exist. The correct sequence:

1. Expire snapshots beyond your retention window. This shrinks the file set every later step has to consider, and it is why expiry goes first: compacting files that are about to be garbage-collected is pure waste.
2. Remove orphan files left by failed writes and by the snapshots you just expired. Use a grace period, never a zero threshold, or you will race in-flight writers.
3. Compact data files, applying the sort or curve strategy, ideally scoped to the partitions that need it rather than the whole table.
4. Rewrite manifests, so the metadata tree indexes the layout that now exists rather than the one that did.
5. Refresh Puffin statistics, so the optimizer's cardinality estimates match the files it will actually read.

Run step 4 before step 3 and your manifests describe a dead file set. Skip step 5 and your CBO plans against a pre-compaction world, which is arguably worse than having no statistics at all, because a confidently wrong estimate produces a confidently wrong plan.

Merge-on-Read: The Delete Tax

On merge-on-read tables, deletes are recorded rather than applied, and every reader pays to reconcile them. This is a read-performance topic that gets filed under "write strategy" and then surprises people when a table that was fast last quarter is not.

Position deletes and the shape of the cost

The v2 position delete file records (file_path, position) pairs, sorted by both so readers can push down by file and stream the merge without holding deletes in memory. The critical property is that cost scales with the number of delete files, not the number of deleted rows. A table with 50 uncompacted small delete files means a full scan opens 51 files where it should open one.

Published benchmarks on position-delete overhead show the cost curve jumping roughly 4× between 0% and 1% deleted rows, then rising only gradually from 1% to 100%. The fixed cost of having any delete files dominates the marginal cost of having more deleted rows. That shape is the argument for compacting on file count aggressively rather than waiting for a deletion-ratio threshold to trip.

Equality deletes, and why they are much worse

An equality delete file records the values of identifier fields rather than positions. That makes writes cheap, which is exactly why Flink's streaming upsert mode defaults to them: locating (file_path, pos) would require a read-before-write at CDC rates.

The cost lands entirely on readers, and two scoping rules make it severe. First, an equality delete applies when the data file's sequence number is strictly less than the delete's, regardless of locality. Second, and this is the one that bites, if the delete file's partition spec is unpartitioned, it is a global delete applied against every data file in the table.

Equality delete files do carry their own metrics, so InclusiveMetricsEvaluator can prune them by identifier bounds. Most streaming writers do not optimize for tight bounds, which means most real deployments do not get that benefit.

Check where you stand:

SELECT
    content,
    file_format,
    count(*)                          AS files,
    sum(record_count)                 AS deleted_rows,
    sum(file_size_in_bytes)/1048576   AS mb
FROM db.events.delete_files
GROUP BY content, file_format;
-- content = 1 -> position deletes or deletion vectors
-- content = 2 -> equality deletes
-- file_format = 'puffin' -> v3 deletion vectors

Any content = 2 rows on a table that is not fed by Flink is a red flag worth chasing down.

Controls

ALTER TABLE db.events SET TBLPROPERTIES (
    'write.delete.mode'         = 'merge-on-read',
    'write.update.mode'         = 'merge-on-read',
    'write.merge.mode'          = 'copy-on-write',
    'write.delete.granularity'  = 'file'
);

All three mode properties default to copy-on-write. write.delete.granularity defaults to partition at the table level; setting it to file produces one delete file per data file, which is more files but each is precisely scoped and converts to a deletion vector without a merge step.

For compaction, the two relevant rewrite_data_files options are delete-file-threshold (default 2147483647, effectively off) and delete-ratio-threshold (default 0.3). Given the cost curve above, delete-file-threshold is the more useful lever. Set it low, around 2 or 3, so any data file accumulating delete files gets rewritten rather than waiting for 30% of its rows to be deleted:

CALL catalog.system.rewrite_data_files(
    table   => 'db.events',
    options => map(
        'delete-file-threshold', '2',
        'remove-dangling-deletes', 'true',
        'partial-progress.enabled', 'true',
        'partial-progress.max-commits', '20'
    )
);

Note that remove-dangling-deletes costs an extra commit and, per the documentation, works solely from data sequence numbers. It does not cover global equality deletes, invalid equality deletes, or multi-file position delete files.

What Format v3 Changes for Reads

Format v3 is complete and adopted. Version 4 is under active development and has not been formally adopted. For query performance specifically, v3 matters in four places.

Deletion vectors replace position delete files. The spec allows at most one DV per data file per snapshot, enforced by writers merging on commit. That converts delete application from "open N delete files and sort-merge by (file_path, pos)" into "load one Roaring bitmap, test a bit." Manifests carry referenced_data_file, content_offset, and content_size_in_bytes, so the reader issues a byte-range request for exactly the DV blob with no Puffin footer scan. Position delete files are deprecated in v3 and must be folded into a DV when one is created.

Variant shredding turns semi-structured data into something prunable. With write.parquet.shred-variants (default false), variant sub-fields become real Parquet columns with real statistics, and manifest bounds for variant columns are keyed by normalized JSON path such as $['location']['latitude']. That is file skipping on JSON payloads, which was previously impossible. Bounds are not written for mixed-type fields.

Geospatial bounding boxes give geometry and geography columns real lower and upper bounds as X/Y/Z/M points, with intersects checking added in 1.11.0. geography permits xmin > xmax to express antimeridian crossing.

Row lineage (_row_id, _last_updated_sequence_number) is always on for v3, not optional. It is inheritance-based, so readers materialize the values from first_row_id + _pos and the manifest sequence number. Cheap, but it is work engines must implement, and coverage arrived incrementally across 1.9 through 1.11.

The engine support matrix, and why it is a one-way door

Upgrading to v3 is a single atomic metadata change with no data rewrite. Rolling back, once deletion vectors have been written, is not practical. Check your entire read path first.

Engine	v3	Deletion vectors	Notes
Spark (Iceberg 1.9+)	Yes	Yes	Can rewrite v2 deletes into v3 DVs
Flink	Yes	Yes	IcebergSink writes DVs as of 1.11.0
Trino 483	Experimental	Read only	Row-level updates, deletes, and OPTIMIZE are not supported on v3
Snowflake	Yes (GA May 2026)	Read and write	Managed and externally managed tables
Amazon Athena	No	No	Creates and operates on v2 only; v3 errors out
DuckDB 1.5.3+	Yes	Read and write	Selects DV vs positional automatically from format-version
StarRocks 4.1	Partial	No	Supports defaults and row lineage; fails fast on DVs, encryption, and geo types
ClickHouse	No	No	v1 and v2 only

The practical reading: if Athena, StarRocks, or ClickHouse are anywhere in your read path, v3 breaks them. StarRocks' fail-fast behavior is deliberate and was introduced after cached DVs produced silently wrong results, which is a good reminder that "unsupported" is safer than "partially supported."

One forward-looking note worth tracking. Today's manifest metrics are stored as map<int, binary>, so every bound is length-prefixed binary that must be deserialized per value and the untyped map prevents Avro from projecting individual columns out of it. Version 4 moves these into a typed content_stats struct with deterministic field IDs, which would make column projection over manifest statistics possible: a planner filtering on one column could read only that column's stats instead of deserializing every bound in the file. It also adds tight_bounds (a boolean asserting the bounds are exact, enabling metadata-only MIN/MAX) and avg_value_size_in_bytes for memory estimation. Foundational types for this landed in 1.11.0.

Related and immediately relevant: Iceberg 1.11.0 disabled min/max aggregation pushdown for string and binary columns. It was a correctness fix, because truncated bounds produced wrong answers. If you documented metadata-only MIN(varchar_col) as a trick, it now reads data.

Engine-Specific Tuning

Table-level optimization benefits every engine that reads the table. Engine tuning is the layer on top, and the defaults are frequently wrong for lakehouse workloads.

Trino

These are the properties from the Iceberg connector docs that actually move query time.

Property	Default	Why it matters
iceberg.dynamic-filtering.wait-timeout	1s	How long split generation waits for join-side filters to arrive. Too short and dynamic filtering silently does nothing on large builds.
iceberg.max-split-size	unset → table's read.split.target-size	Note the name; it is not iceberg.split-size. Session override is experimental_split_size.
iceberg.parquet-footer-cache.type	none	Off by default. Setting it to memory is a free latency win on high-file-count tables.
iceberg.metadata-cache.enabled	true	Coordinator-side. Deactivated when fs.cache.enabled = true.
iceberg.query-partition-filter-required	false	Only enforced for schemas listed in iceberg.query-partition-filter-required-schemas. Setting the boolean alone does nothing.
iceberg.planning-threads	2× coordinator cores	Manifest read parallelism
iceberg.bucket-execution	true	Trino's answer to storage-partitioned joins on bucketed tables
iceberg.minimum-assigned-split-weight	0.05	Raise for skewed aggregations; lower for many-small-file tables
parquet.use-bloom-filter	true	Bloom filters are consulted by default
parquet.ignore-statistics	false	Escape hatch when a writer produced corrupt stats

Dynamic filtering is Trino's highest-value Iceberg feature and the one most affected by that first timeout. Trino collects values from the build side of a join at runtime and pushes them into the Iceberg scan, pruning files by join key rather than only by WHERE predicate. On a star schema, a small dimension filter can eliminate most of the fact table. If the build side takes longer than the wait timeout to materialize, splits are generated without the filter and the benefit vanishes.

File caching is separate from metadata caching and requires more setup than a single flag: fs.cache.enabled (default false), fs.cache.directories (must exist on the coordinator and every worker), and one of fs.cache.max-sizes or fs.cache.max-disk-usage-percentages. Node affinity comes from node-scheduler.cache-preferred-hosts-count (default 2) in the coordinator config; raising it spreads load and adds resilience at the cost of effective cache size. Each catalog needs its own cache directory.

Spark

Runtime filtering is Iceberg's implementation of dynamic partition pruning, via SupportsRuntimeV2Filtering. Spark re-plans the scan after the build side materializes, and Iceberg re-runs InclusiveMetricsEvaluator with the runtime value set. The underappreciated part: this works on any column with bounds, not just partition columns, which makes it strictly more capable than Hive-style DPP.

Storage-partitioned joins eliminate the shuffle entirely when both sides are bucketed on the join key. Getting them to actually engage requires more than the one flag most guides cite:

spark.sql.sources.v2.bucketing.enabled=true
spark.sql.iceberg.planning.preserve-data-grouping=true
spark.sql.sources.v2.bucketing.pushPartValues.enabled=true
spark.sql.requireAllClusterKeysForCoPartition=false
spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled=true
spark.sql.autoBroadcastJoinThreshold=-1

Three failure modes from real deployments. preserve-data-grouping defaults to false and is the one people forget, so the join silently falls back to a shuffle. Partition columns must appear in the join condition or you get an UnsupportedOperationException about retrieving values from an empty struct. And MERGE INTO with a WHEN NOT MATCHED clause still requires a full shuffle and can OOM under skew; the documented workaround is splitting matched and not-matched into two statements.

Other Spark settings with real read impact:

Config	Default	Effect
spark.sql.iceberg.vectorization.enabled	table default (Parquet: true)	Vectorized Parquet reads
spark.sql.iceberg.data-planning-mode	AUTO	LOCAL / DISTRIBUTED manifest scanning
spark.sql.iceberg.executor-cache.enabled	true	Caches delete files on executors so the same DV is not re-read by every task
spark.sql.iceberg.executor-cache.max-entry-size	64 MB	Silently excludes large equality delete files
spark.sql.iceberg.aggregate-push-down.enabled	true	Metadata-only aggregates where bounds permit
spark.sql.iceberg.report-column-stats	true	Reports Puffin stats; needs spark.sql.cbo.enabled to matter

The executor cache is the quietly valuable one on merge-on-read tables with concentrated deletes, and the 64 MB per-entry cap is the reason it sometimes appears not to work.

Everything else

• Snowflake reached v3 GA in May 2026 with DV read and write on both managed and externally managed tables. Its internal pruning and caching behavior is proprietary and unpublished, so treat claims about how it layers on Iceberg statistics with suspicion, including this one.
• Amazon Athena is v2-only. Its documentation cites an Iceberg library version, which is frequently misread as a format version; do not conflate them.
• DuckDB 1.5.3+ has full v3 read and write support including DVs, variant, and row lineage, and it is genuinely excellent for single-node work on well-compacted tables. iceberg_metadata() is a fast way to inspect layout without a cluster.
• StarRocks 4.1 adds native Iceberg DELETE, global late materialization for v3, and a metadata cache covering tables, partition names, data files, and delete files.
• ClickHouse does not enable use_iceberg_partition_pruning by default. Setting it to 1 is an easy and commonly missed win.

The point that survives all of this: table-level layout, statistics, and manifest health benefit every engine simultaneously. Engine tuning benefits one. Fix the table first.

A Runbook for One Slow Query

When a specific query is slower than it should be, work down the funnel in order. Each step either finds the problem or rules out a stage.

1. Split planning from execution

In Trino, check planning_time against execution_time in system.runtime.queries, or read the query detail page. In Spark, look at the gap between job submission and first task launch. If planning dominates, skip to step 4. If execution dominates, the problem is how much data you are reading, so continue.

2. Confirm the predicate is reaching the scan

Run EXPLAIN and look at where your predicate landed. In Trino, a predicate absorbed into the scan's constraint is enforced by Iceberg; one sitting in a Filter node above the scan is evaluated per row after the data is read. A filter wrapped in a function the engine cannot push down, an implicit cast between timestamp precisions, or a comparison against a non-deterministic expression will all quietly demote your predicate to a post-scan filter.

3. Check whether the filter column has usable bounds

SELECT
    count(*) AS files,
    count(readable_metrics.customer_id.lower_bound) AS files_with_bounds,
    count(DISTINCT readable_metrics.customer_id.lower_bound) AS distinct_lowers
FROM db.events.files;

files_with_bounds well below files means a metrics configuration problem: fix it with write.metadata.metrics.column.<col> and a rewrite. distinct_lowers well below files means a clustering problem: fix it with a sorted compaction.

4. Check manifest count and clustering

SELECT count(*) AS manifests, avg(length)/1048576 AS avg_mb
FROM db.events.manifests;

Hundreds of manifests averaging well under 8 MB means fragmentation. Run rewrite_manifests, and pass sort_by naming the transform your queries actually filter on rather than accepting the spec-order default.

5. Check the delete burden

SELECT content, count(*) AS files, sum(record_count) AS rows
FROM db.events.delete_files
GROUP BY content;

High delete file counts, or any equality deletes outside a Flink pipeline, mean every scan is paying reconciliation cost. Compact with a low delete-file-threshold.

6. Check partition and file size distribution

SELECT
    partition,
    count(*) AS files,
    avg(file_size_in_bytes)/1048576 AS avg_mb
FROM db.events.files
GROUP BY partition
ORDER BY files DESC
LIMIT 20;

Thousands of tiny files in a few partitions means compaction lag. Tiny files uniformly across all partitions means the partition spec is too fine.

7. Only now, tune the engine

Dynamic filtering timeouts, footer caches, split sizes, and join configs are real wins, but they are multipliers on a healthy table. Applying them first means tuning the constant factor on an exponential problem.

How a Control Plane Runs This Loop

Everything above is tractable by hand on ten tables. The structure of the problem changes as the lake grows, and it changes in a specific way worth naming: each individual decision stays easy, but the number of decisions grows with tables × engines × time, and every one of them expires.

Consider what "keep sort orders correct" actually requires at a few hundred tables. You need query telemetry from every engine, because the optimal layout for a table depends on how Trino queries it, how Spark ETL writes to it, and how a Snowflake dashboard reads it, simultaneously, and no single engine has that view. You need to score candidate layouts against the combined mix. You need to validate a proposed layout before rewriting terabytes, because a sort order change is expensive to reverse. You need to sequence expiry, cleanup, compaction, manifest rewrite, and statistics refresh correctly per table. You need to avoid compacting a partition with an active writer, and handle optimistic-concurrency conflicts when you lose the race anyway. And you need to redo all of it when the query mix shifts.

That is a control loop, and LakeOps exists to run it as one. Here is what that means concretely for the specific problems in this guide.

Apache Iceberg Lakehouse Control Plane | LakeOps

Manage your Iceberg lakehouse from one control plane — health-driven compaction, table maintenance, multi-engine routing, and observability.

lakeops.dev

Sort order selected from telemetry, not from intuition

The hardest question in this guide is which columns to cluster on. LakeOps answers it from observation: it collects SELECT, WHERE, JOIN, and GROUP BY column frequency across the connected engines, builds a field-access profile per table, and evaluates single-column sort, multi-column sort, and Z-order strategies independently against that profile rather than applying one policy lake-wide. When a new dashboard ships that filters on a different column, the profile shifts and the next compaction cycle applies the updated layout. The query performance results published for this are 51% less data scanned per query, 12× faster queries, and 76% less CPU.

Iceberg Lakehouse Governance | LakeOps

Iceberg lakehouse governance — compliance automation, declarative policies, rollback safety, and audit trails across every catalog and engine.

lakeops.dev

Layout changes validated on a branch before production files move

This is the part that makes automated layout changes defensible rather than reckless. Rewriting every data file in scope is expensive and awkward to undo, so the decision needs evidence before it is applied, not after.

LakeOps runs layout simulations on Iceberg branches: it branches from the latest snapshot, applies a candidate layout, replays observed production query patterns against it, compares scan reduction and file distribution to the baseline, and discards the branch. Only the strategy that measurably wins gets applied to production. Iceberg's branching model is what makes this possible at all, and it is a good example of a format capability that most teams never operationalize because wiring it up is more work than the layout question seemed to be worth.

Iceberg Lakehouse Observability | LakeOps

Iceberg table health scores, cross-engine monitoring, and proactive alerts for your data lakehouse. Track file layout, snapshot depth, and partition skew.

lakeops.dev

Compaction on a dedicated engine rather than a Spark cluster

Sorted compaction is the expensive half of everything above, and running it on Spark means cluster startup, idle capacity, GC pressure, and OOM risk on large rewrites. LakeOps runs compaction on a Rust and DataFusion engine instead, with conflict-aware Iceberg commits, bounded memory with disk spill, and partitions that have active writers excluded from the rewrite set. Its published benchmark compacts 200 GB in 221 seconds against 1,612 seconds for Spark on the same hardware, at a peak throughput of 2,522 MB/s. The compaction strategy deep dive covers how binpack, sort, and Z-order selection differ per table.

Apache Iceberg Lakehouse Compaction | LakeOps

Query-aware Iceberg compaction on a Rust engine. Sorts Parquet files by production query patterns. 95% faster than Spark. Lakehouse ready.

lakeops.dev

The maintenance sequence, executed in dependency order

The five-step sequence earlier in this guide is not hard to understand and is hard to hold correct across hundreds of tables with different retention windows, write patterns, and SLAs. LakeOps sequences expiry, orphan cleanup, sorted compaction, manifest rewrite, and statistics refresh automatically, per table, with each step's output feeding the next. Puffin NDV sketches are recomputed after the file layout changes rather than on a fixed schedule, which is the only way the staleness problem described earlier stays solved rather than solved once.

Health classification, so you know which tables to care about

Rather than a schedule, maintenance triggers on structural signals: file count and size distribution, manifest depth, snapshot accumulation, delete-file ratio, partition skew, and sort-order alignment with observed query patterns. That last signal is the one that has no equivalent in schedule-driven maintenance, because it is not a property of the table at all. It is a property of the relationship between the table and its workload, and it can degrade without a single byte being written. Lake-wide observability surfaces it alongside the structural signals so a table whose layout has drifted out of alignment shows up before users file a ticket.

Routing, which closes the loop

Physical layout determines which engines are viable for a query. A well-clustered table with tight bounds makes a DuckDB point lookup faster and cheaper than the same query on a distributed engine; a table needing a full scan does not. LakeOps routes queries from a single SQL endpoint to the engine best suited to each query shape, and those routing decisions feed back into the same telemetry that drives layout, so agent and application query patterns inform the next sort-order decision. The platform overview covers how the observe, maintain, compact, govern, and route capabilities compose.

Routing Multiple Query Engines with Iceberg - LakeOps Blog

Route queries across Trino, Spark, DuckDB, Athena, and more on shared Iceberg tables. Routing architecture, strategies, dialect translation, and patterns.

lakeops.dev

Iceberg Lakehouse Multi-Engine Query Routing | LakeOps

One SQL endpoint for your lakehouse. Route Iceberg queries to Trino, Spark, DuckDB, Snowflake, or Athena by cost, latency, or workload.

lakeops.dev

None of this is a substitute for understanding the mechanics. It is a substitute for running the loop by hand, on a schedule you set once and never revisited, across a lake that changed underneath you.

Optimization Checklist

Lever	Typical impact	When to apply
Verify metrics exist on filter columns	Enables Stage 2 at all	First. Before any layout work.
Partition on time, plus one high-value dimension	90–99% coarse elimination	Table creation; evolve as query patterns shift
Global sort on the dominant filter column	Large improvement in file-level skipping	Once you know the dominant predicate
write.distribution-mode = range	Turns a declared sort into real disjoint bounds	Whenever a sort order is declared
Z-order on 2–4 columns	Balanced pruning across dimensions	Varied multi-column filters; check var-length-contribution on string keys
Bloom filters on 2–3 high-cardinality keys	Decisive for point lookups	Equality predicates on non-sort-key columns; always set bloom-filter-ndv
Row group size 32–64 MB	Finer in-file skipping	Selective and point-lookup workloads only
rewrite_manifests with sort_by	Seconds to milliseconds of planning	Manifest count in the hundreds, or skippedDataManifests near zero
Compact with low delete-file-threshold	Removes merge-on-read reconciliation cost	Any MoR table; urgently if equality deletes are present
Refresh Puffin stats after compaction	Correct join orders	Every time the file layout changes
Enable spark.sql.cbo.enabled	Makes collected stats actually count	If you compute stats and see no plan change
Project only needed columns	Linear I/O reduction	Always on wide tables

The Takeaway

Iceberg query performance is not an engine property. It is a function of how much data the planner is able to discard, and every layer of that decision is something you control at write and maintenance time: whether the statistics exist, whether the bounds are tight enough to be useful, whether the manifests are clustered the way your queries filter, whether the deletes have been applied, and whether the optimizer's cardinality estimates describe the table that exists today.

Those layers compound rather than add. A table with sensible partitioning, globally sorted data, bloom filters on its lookup keys, consolidated manifests sorted on the hot transform, resolved deletes, and current statistics hands the engine everything it needs to eliminate 99% of the work before execution starts. The same table missing any one of them can lose most of that benefit, which is why the diagnostic order in the runbook matters more than any individual knob.

The engineering is the same whether you run it with carefully sequenced Spark procedures or hand the loop to a control plane like LakeOps that senses query patterns, plans layout per table, validates on a branch, and measures the result. Make data skipping work at every level, keep the metadata honest, and make sure the statistics still describe reality.