Trino iceberg

Trino + Apache Iceberg Engineer

When to Use

Use this skill when:

• Creating or modifying Iceberg tables in Trino (DDL, partitioning, bucketing, table properties)
• Writing or tuning DML: INSERT, UPDATE, DELETE, MERGE, TRUNCATE on Iceberg tables
• Optimizing query performance: reading EXPLAIN plans, configuring join strategies, enabling pushdowns
• Running or scheduling table maintenance: compaction, snapshot expiration, orphan file cleanup
• Diagnosing performance problems using $partitions, $files, $snapshots metadata tables
• Implementing schema evolution, time travel, or snapshot rollback
• Configuring the Iceberg connector for a new Trino cluster

---

Core Workflow

Starting from scratch: new table

1. Choose partition granularity from volume guidelines (see Partitioning section).
2. Decide on bucketing if a high-cardinality join/filter column exists (see Bucketing section).
3. Pick sorted_by columns: the highest-selectivity filter columns and join keys.
4. Create the table with format='PARQUET', format_version=2.
5. Load data, then run ANALYZE on join and filter columns.
6. Verify the plan with EXPLAIN (TYPE DISTRIBUTED).

Diagnosing a slow query

1. Run EXPLAIN (TYPE DISTRIBUTED) — identify join type, exchange type, scan rows.
2. Run SHOW STATS FOR <table> — check if statistics exist.
3. Check "$partitions" — look for partition skew or small-file accumulation.
4. Run ANALYZE if NDV / row counts are missing.
5. Adjust partitioning, bucketing, sorted_by, or session join properties if plan is still suboptimal.

Regular maintenance cycle

Daily   → OPTIMIZE on partitions written in the last 1–2 days
Weekly  → EXPIRE_SNAPSHOTS + REMOVE_ORPHAN_FILES + OPTIMIZE_MANIFESTS
After optimize → ANALYZE to refresh statistics

---

Table Creation

Column data types

Category	Types
Integer	BOOLEAN, TINYINT, SMALLINT, INTEGER / INT, BIGINT
Floating point	REAL / FLOAT, DOUBLE
Fixed precision	DECIMAL(precision, scale) — use for monetary values
String	VARCHAR (unbounded), VARCHAR(n) (max n chars), CHAR(n) (fixed, space-padded)
Binary	VARBINARY
Date / time	DATE, TIME, TIME(p) WITH TIME ZONE, TIMESTAMP, TIMESTAMP(p) WITH TIME ZONE
Complex	ARRAY(element_type), MAP(key_type, value_type), ROW(field_name field_type, ...)
Other	UUID, JSON, IPADDRESS

Prefer TIMESTAMP(6) WITH TIME ZONE over bare TIMESTAMP — Iceberg stores time zone offset and avoids daylight saving ambiguity.

Full CREATE TABLE syntax

CREATE TABLE iceberg_catalog.mydb.events (
    event_id    BIGINT        NOT NULL,
    event_type  VARCHAR       NOT NULL,
    user_id     BIGINT,
    region      VARCHAR,
    amount      DECIMAL(18,2),
    tags        ARRAY(VARCHAR),
    metadata    MAP(VARCHAR, VARCHAR),
    address     ROW(city VARCHAR, country CHAR(2)),
    event_ts    TIMESTAMP(6) WITH TIME ZONE NOT NULL,
    created_at  TIMESTAMP(6) WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
)
COMMENT 'Raw clickstream events'
WITH (
    format                         = 'PARQUET',
    format_version                 = 2,
    partitioning                   = ARRAY['region', 'day(event_ts)'],
    sorted_by                      = ARRAY['user_id', 'event_type'],
    write_target_data_file_size_bytes = 536870912,  -- 512 MB per file
    location                       = 's3://my-bucket/warehouse/mydb/events'
);

All WITH table properties

Property	Values / Default	Description
format	'PARQUET' ★ / 'ORC' / 'AVRO'	Data file format. Use PARQUET for analytics.
format_version	1 / 2 ★	Format v2 required for full ACID (UPDATE/DELETE/MERGE).
partitioning	ARRAY[...]	Partition spec (transforms). See Partitioning section.
sorted_by	ARRAY[...]	Sort columns within each data file.
location	URI string	Storage path. Defaults to catalog warehouse path.
write_target_data_file_size_bytes	integer, default 1073741824 (1 GB)	Max data file size before rolling to a new file.
orc_bloom_filter_columns	ARRAY['col1', 'col2']	ORC-only: enables Bloom filters for these columns.
orc_bloom_filter_fpp	0.0–1.0, default 0.05	ORC Bloom filter false-positive probability.
parquet_bloom_filter_columns	ARRAY['col1', 'col2']	Parquet: Bloom filters for equality / IN predicates.

★ = recommended default.

Connector defaults (etc/catalog/iceberg.properties)

iceberg.file-format=PARQUET
iceberg.compression-codec=ZSTD
iceberg.target-max-file-size=1GB
iceberg.sorted-writing-enabled=true
iceberg.table-statistics-enabled=true
iceberg.file-system-cache.enabled=true
iceberg.file-system-cache.directory=/tmp/trino-cache
iceberg.file-system-cache.max-size=100GB

Set these once globally; omit format and compression from individual table DDL unless overriding.

CREATE TABLE AS SELECT (CTAS)

-- Create and populate in one step; inherits column types from the SELECT
CREATE TABLE iceberg_catalog.mydb.events_daily
WITH (
    format         = 'PARQUET',
    format_version = 2,
    partitioning   = ARRAY['dt']
) AS
SELECT
    date_trunc('day', event_ts) AS dt,
    event_type,
    region,
    count(*)        AS event_count,
    sum(amount)     AS total_amount
FROM iceberg_catalog.mydb.events
GROUP BY 1, 2, 3;

CTAS creates a new snapshot; run ANALYZE afterwards if the table will be used in joins.

CREATE TABLE LIKE

-- Copy column definitions and table properties; does NOT copy data
CREATE TABLE iceberg_catalog.mydb.events_staging
LIKE iceberg_catalog.mydb.events
INCLUDING PROPERTIES;   -- copies format, partitioning, sorted_by, etc.

Use EXCLUDING PROPERTIES to copy only columns without the partition spec (useful for staging tables that shouldn't be partitioned the same way).

External / registered table

Point to existing Iceberg metadata already on object storage:

CREATE TABLE iceberg_catalog.mydb.events_external (
    event_id   BIGINT,
    event_ts   TIMESTAMP(6) WITH TIME ZONE
)
WITH (
    location = 's3://existing-bucket/path/to/table'
);

Trino will read the existing metadata/ directory at that location and register it.

---

Partitioning

Partition transforms

Transform	Syntax	Use case
Identity	region	Low-cardinality categorical columns
Day	day(ts)	Daily pipelines — most common temporal choice
Hour	hour(ts)	High-frequency event streams (> 5 GB/day)
Month	month(ts)	Low-volume data (< 100 MB/day)
Year	year(ts)	Archival or very sparse data
Bucket	bucket(N, col)	High-cardinality IDs — see Bucketing section
Truncate	truncate(N, col)	String prefix or integer range partitioning

Hidden partitioning: users query the original column (event_ts, user_id). Trino rewrites predicates into the corresponding partition transform internally. The partition column never leaks into the schema.

Partition sizing guideline

Daily data volume	Recommended granularity
< 100 MB	month(ts)
100 MB – 5 GB	day(ts)
5 GB – 50 GB	hour(ts)
> 50 GB	hour(ts) + bucket(N, high_card_col)

Target 256 MB – 1 GB per data file after compaction.

Combined partition specs

-- Temporal + categorical
partitioning = ARRAY['day(event_ts)', 'region']

-- Temporal + bucket for even distribution per day
partitioning = ARRAY['day(event_ts)', 'bucket(32, user_id)']

-- Truncate for string prefix partitioning
partitioning = ARRAY['truncate(2, country_code)', 'month(event_ts)']

-- Identity on enum
partitioning = ARRAY['status', 'day(created_at)']

How partition pruning works

1. The optimizer extracts predicates on partition columns from the WHERE clause.
2. Each predicate is translated to the partition transform and matched against manifest file metadata.
3. Manifest files whose partition range doesn't match are skipped without reading any data.
4. Within surviving manifests, per-file min/max statistics skip individual data files.

-- This predicate prunes day(event_ts) partitions — only Jan files are read
SELECT * FROM events
WHERE event_ts >= TIMESTAMP '2024-01-01 00:00:00 UTC'
  AND event_ts <  TIMESTAMP '2024-02-01 00:00:00 UTC';

-- Combined pruning: day + region both applied
SELECT * FROM events
WHERE event_ts >= TIMESTAMP '2024-01-15 00:00:00 UTC'
  AND region = 'us-east-1';

Pruning works for: equality (=), ranges (>, <, BETWEEN), IN lists, IS NULL / IS NOT NULL. Pruning does NOT work for: expressions on columns (date_trunc('day', event_ts) = DATE '2024-01-15' — use the raw column), OR across partition columns.

Verify pruning in EXPLAIN

EXPLAIN (TYPE IO, FORMAT JSON)
SELECT count(*) FROM events
WHERE event_ts >= TIMESTAMP '2024-01-15 00:00:00 UTC'
  AND region = 'us-east-1';
-- Look for: "partitionsCount": N  (low N means pruning worked)
-- and "estimatedSizeInBytes" matching only the selected partitions

Inspect current partition spec

-- File distribution per partition
SELECT partition, file_count, record_count,
       total_size / 1024 / 1024 AS total_mb,
       total_size / NULLIF(file_count, 0) / 1024 / 1024 AS avg_file_mb
FROM iceberg_catalog.mydb."events$partitions"
ORDER BY total_size DESC;

-- Table properties including current partition spec
SELECT * FROM iceberg_catalog.mydb."events$properties";

Evolving partitioning (no rewrite)

-- Finer granularity going forward; existing files keep old layout
ALTER TABLE events
SET PROPERTIES partitioning = ARRAY['region', 'hour(event_ts)'];

-- Add a bucket dimension to an existing temporal partition
ALTER TABLE events
SET PROPERTIES partitioning = ARRAY['day(event_ts)', 'bucket(32, user_id)'];

-- Remove partitioning entirely (unpartitioned going forward)
ALTER TABLE events
SET PROPERTIES partitioning = ARRAY[];

Old files are not rewritten and remain queryable under both layouts.

---

Bucketing

What bucketing does

bucket(N, col) hashes the column value and assigns it to one of N bucket numbers. Bucket number is stored in partition metadata. At query time, equality predicates on the bucketed column are translated to specific bucket numbers — only manifest files for those buckets are read.

row with user_id=12345
  → hash(12345) mod 32 = 17
  → written to bucket 17 file
  → query WHERE user_id = 12345 → reads only bucket 17 file

When to use bucketing

Use bucket(N, col) when:

• The column has high cardinality (millions+ distinct values: user_id, order_id, session_id).
• Queries frequently filter by equality or IN-list on that column.
• A temporal partition alone would create too many tiny files per partition, or without bucketing each partition file would be too large.
• The table participates in joins on that column — co-located bucketed tables joined on the same column avoid network shuffle.

Do not use bucketing on low-cardinality columns (< 1000 distinct values) — use identity partitioning or truncate instead.

Choosing bucket count N

Start with: N = ceil(expected_partition_size_GB / 0.5) — aiming for ~512 MB per bucket file per partition.

Partition size after load	Suggested N
< 4 GB/day	8
4–16 GB/day	32
16–64 GB/day	64
64–256 GB/day	128
> 256 GB/day	256+

N must be a power of 2 — Iceberg's hash function relies on this for consistent distribution. Changing N later requires a full table rewrite, so choose carefully upfront.

Create a bucketed table

-- Pure bucket partitioning (no temporal — suited for lookup tables)
CREATE TABLE iceberg_catalog.mydb.user_profiles (
    user_id     BIGINT NOT NULL,
    email       VARCHAR,
    country     CHAR(2),
    created_at  TIMESTAMP(6) WITH TIME ZONE
)
WITH (
    format         = 'PARQUET',
    format_version = 2,
    partitioning   = ARRAY['bucket(64, user_id)'],
    sorted_by      = ARRAY['user_id']
);

-- Combined temporal + bucket (most common production pattern)
CREATE TABLE iceberg_catalog.mydb.events (
    event_id    BIGINT,
    user_id     BIGINT,
    event_ts    TIMESTAMP(6) WITH TIME ZONE,
    event_type  VARCHAR,
    payload     VARCHAR
)
WITH (
    format         = 'PARQUET',
    format_version = 2,
    partitioning   = ARRAY['day(event_ts)', 'bucket(32, user_id)'],
    sorted_by      = ARRAY['user_id', 'event_type']
);

Bucket pruning mechanics

Trino translates equality and IN predicates into bucket numbers and skips all other bucket files:

-- Equality → looks up exactly bucket hash(42) mod 32
SELECT * FROM events WHERE user_id = 42;

-- IN list → reads only the buckets for each value in the list
SELECT * FROM events WHERE user_id IN (42, 1001, 9999);

-- Range predicate → cannot compute bucket set → reads ALL buckets (no pruning)
SELECT * FROM events WHERE user_id BETWEEN 1 AND 1000;

Verify bucket pruning via EXPLAIN IO:

EXPLAIN (TYPE IO, FORMAT JSON)
SELECT * FROM events WHERE user_id = 42 AND event_ts >= TIMESTAMP '2024-01-01 00:00:00 UTC';
-- Expected: partitionsCount = 1 (only the bucket containing user_id=42 within the day partition)

Bucket pruning through joins (dynamic partition pruning)

When a bucketed table is joined to a small filtered dimension, Trino builds a Bloom filter from the dimension side and pushes it to the bucket scan — further reducing files read at runtime:

-- Trino builds a Bloom filter from the WHERE on dim_users and pushes it
-- to the events scan, skipping bucket files not containing matching user_ids
SELECT e.event_type, count(*)
FROM events e
JOIN dim_users u ON e.user_id = u.user_id
WHERE u.country = 'US'
GROUP BY 1;

EXPLAIN shows DynamicFilterSource on the build side and DynamicFilter applied to the scan — this is working as expected.

Adding bucketing to an existing table

-- Affects future writes and OPTIMIZE only; existing files are not changed
ALTER TABLE events
SET PROPERTIES partitioning = ARRAY['day(event_ts)', 'bucket(32, user_id)'];

After changing the partition spec: run OPTIMIZE to compact and re-bucket existing files, then ANALYZE.

Changing bucket count

There is no in-place bucket count change — existing files are tied to the original bucket mapping. To change N:

-- 1. Create new table with desired bucket count
CREATE TABLE events_v2 AS SELECT * FROM events;  -- CTAS with new WITH properties

-- 2. Swap alias / rename (coordinate with consumers)
ALTER TABLE events RENAME TO events_old;
ALTER TABLE events_v2 RENAME TO events;

---

ALTER TABLE: Full Reference

Column operations

-- Add a nullable column (visible as NULL for existing rows)
ALTER TABLE events ADD COLUMN session_id VARCHAR;

-- Add with NOT NULL constraint (only safe if table is empty or you provide a DEFAULT)
ALTER TABLE events ADD COLUMN is_bot BOOLEAN NOT NULL DEFAULT false;

-- Add a column with a comment
ALTER TABLE events ADD COLUMN campaign_id BIGINT COMMENT 'Ad campaign identifier';

-- Add a column and position it explicitly
ALTER TABLE events ADD COLUMN channel VARCHAR FIRST;
ALTER TABLE events ADD COLUMN channel VARCHAR AFTER event_type;

-- Drop a column (data in existing files is preserved but invisible to queries)
ALTER TABLE events DROP COLUMN legacy_field;

-- Rename a column (tracked by column ID — safe for time travel reads)
ALTER TABLE events RENAME COLUMN user_id TO account_id;

-- Widen a column type (only safe promotions allowed)
ALTER TABLE events ALTER COLUMN click_count SET DATA TYPE BIGINT;   -- INT → BIGINT
ALTER TABLE events ALTER COLUMN score      SET DATA TYPE DOUBLE;    -- FLOAT → DOUBLE

-- Change nullability
ALTER TABLE events ALTER COLUMN session_id SET NOT NULL;
ALTER TABLE events ALTER COLUMN session_id DROP NOT NULL;

-- Set or change default value
ALTER TABLE events ALTER COLUMN status SET DEFAULT 'active';
ALTER TABLE events ALTER COLUMN status DROP DEFAULT;

-- Reorder an existing column
ALTER TABLE events ALTER COLUMN session_id FIRST;
ALTER TABLE events ALTER COLUMN session_id AFTER event_id;

-- Add or change column comment
ALTER TABLE events ALTER COLUMN user_id COMMENT 'Internal account ID';

Nested column operations (ROW fields)

-- Add a subfield to an existing ROW column
ALTER TABLE events ADD COLUMN address.zip_code VARCHAR;

-- Drop a subfield
ALTER TABLE events DROP COLUMN address.zip_code;

-- Rename a subfield
ALTER TABLE events RENAME COLUMN address.city TO address.city_name;

Allowed type promotions

From	To	Safe
TINYINT	SMALLINT, INTEGER, BIGINT	Yes
SMALLINT	INTEGER, BIGINT	Yes
INTEGER	BIGINT	Yes
FLOAT / REAL	DOUBLE	Yes
DECIMAL(p, s)	DECIMAL(p', s) where p' > p	Yes
VARCHAR(n)	VARCHAR(m) where m > n or unbounded VARCHAR	Yes
Any type	Incompatible type	No — requires table rebuild

Table property changes

-- Change partition spec (in-place; old files keep old layout)
ALTER TABLE events SET PROPERTIES partitioning = ARRAY['day(event_ts)', 'bucket(32, user_id)'];

-- Remove partitioning
ALTER TABLE events SET PROPERTIES partitioning = ARRAY[];

-- Change or add sorted_by (affects future writes and OPTIMIZE)
ALTER TABLE events SET PROPERTIES sorted_by = ARRAY['user_id', 'event_type'];

-- Remove sorting
ALTER TABLE events SET PROPERTIES sorted_by = ARRAY[];

-- Upgrade format version (v1 → v2 enables full ACID)
ALTER TABLE events SET PROPERTIES format_version = 2;

-- Change target file size for writes (bytes)
ALTER TABLE events SET PROPERTIES write_target_data_file_size_bytes = 268435456;  -- 256 MB

-- Enable Parquet Bloom filters on specific columns
ALTER TABLE events SET PROPERTIES parquet_bloom_filter_columns = ARRAY['user_id', 'session_id'];

-- Enable ORC Bloom filters
ALTER TABLE events SET PROPERTIES orc_bloom_filter_columns = ARRAY['user_id'];
ALTER TABLE events SET PROPERTIES orc_bloom_filter_fpp     = 0.01;

Table rename

ALTER TABLE iceberg_catalog.mydb.events RENAME TO iceberg_catalog.mydb.events_v2;

Table comment

COMMENT ON TABLE events IS 'Raw clickstream events — partitioned by day and user bucket';
COMMENT ON COLUMN events.user_id IS 'Internal account identifier';

Change table owner

ALTER TABLE events SET AUTHORIZATION USER 'new_owner';
ALTER TABLE events SET AUTHORIZATION ROLE 'data_engineering';

Table procedures (via ALTER TABLE EXECUTE)

These are covered in detail in Table Maintenance, but listed here for completeness:

ALTER TABLE events EXECUTE optimize(file_size_threshold => '100MB');
ALTER TABLE events EXECUTE expire_snapshots(retention_threshold => '14d', retain_last => 10);
ALTER TABLE events EXECUTE remove_orphan_files(retention_threshold => '14d');
ALTER TABLE events EXECUTE optimize_manifests;

Drop table

-- Drops table metadata AND data files
DROP TABLE iceberg_catalog.mydb.events;

-- Safe drop (no error if table doesn't exist)
DROP TABLE IF EXISTS iceberg_catalog.mydb.events;

---

Sorted Tables

sorted_by pre-sorts rows by the given columns within each data file. Benefits:

• Tighter min/max bounds in manifests → more file skipping for selective predicates.
• Higher compression ratios (similar values are adjacent).

A 1.4B-row event table sorted on (user_id, event_type) shrank from 8.09 GB / 3,619 files to 2.40 GB / 1,074 files (73% reduction).

Rule: set sorted_by on columns most used in WHERE equality filters and join predicates. Primary sort column should be the one with the highest cardinality in those filters.

-- Set at creation
CREATE TABLE events (...) WITH (
    partitioning = ARRAY['day(event_ts)'],
    sorted_by    = ARRAY['user_id', 'event_type']
);

-- Add to existing table (affects future writes and OPTIMIZE)
ALTER TABLE events SET PROPERTIES sorted_by = ARRAY['user_id', 'event_type'];

After adding sorted_by, run OPTIMIZE to compact and re-sort existing files, then ANALYZE.

---

EXPLAIN: Reading the Plan

-- Distributed plan: stages, join types, exchanges
EXPLAIN (TYPE DISTRIBUTED) SELECT ...;

-- IO plan: scan row estimates used by optimizer (JSON for tooling)
EXPLAIN (TYPE IO, FORMAT JSON) SELECT ...;

-- Actual runtime stats (runs the query)
EXPLAIN ANALYZE SELECT ...;

Key signals

Signal	Meaning	Action
InnerJoin[BROADCAST]	Small table replicated to all workers	Expected for small dims; verify size estimate
InnerJoin[PARTITIONED]	Both sides hash-shuffled	Normal for large–large joins
LocalExchange[GATHER]	Single-threaded aggregation	Check partition key or add parallelism
ScanFilterProject with high estimated rows	Poor pruning or missing stats	Add partition predicate; run ANALYZE
DynamicFilterSource	Build-side Bloom filter pushed to scan	Good — reduces probe scan at runtime
low partitionsCount in IO plan	Partition pruning working	Confirm expected count
high partitionsCount despite filter	Predicate not pruning	Check column is a partition transform; avoid expressions

---

Statistics (ANALYZE)

Manifest files automatically track min/max per file — no ANALYZE needed for file-level skipping. ANALYZE additionally computes table-level row counts, null fractions, and NDV (required for join cardinality estimation and join reordering).

-- Full table
ANALYZE iceberg_catalog.mydb.events;

-- Specific columns only (faster, target join and filter columns)
ANALYZE iceberg_catalog.mydb.events
  WITH (columns = ARRAY['user_id', 'event_type', 'region']);

-- Single partition (incremental refresh)
ANALYZE iceberg_catalog.mydb.events
  WITH (partitioning = ARRAY['dt=2024-01-15']);

-- Inspect current stats
SHOW STATS FOR iceberg_catalog.mydb.events;

Run ANALYZE after: initial load, bulk DML > 10% of table, OPTIMIZE, or when EXPLAIN shows wrong row estimates.

---

Join Optimization

SET SESSION join_reordering_strategy       = 'AUTOMATIC';   -- CBO reorders with stats
SET SESSION join_distribution_type         = 'AUTOMATIC';   -- CBO chooses broadcast vs partitioned
SET SESSION join_max_broadcast_table_size  = '200MB';       -- default 100 MB

Force broadcast when CBO underestimates the build side:

SELECT /*+ BROADCAST(small_dim) */ f.*, d.name
FROM large_fact f
JOIN small_dim d ON f.dim_id = d.id;

Enable adaptive join reordering (corrects estimates mid-execution):

fault-tolerant-execution-enabled=true
fault-tolerant-execution-adaptive-query-planning-enabled=true

---

DML

INSERT

INSERT INTO events SELECT * FROM staging_events;
INSERT INTO events (event_id, event_type, user_id, event_ts) VALUES (1, 'click', 42, CURRENT_TIMESTAMP);

UPDATE (copy-on-write)

Trino rewrites affected data files entirely. Align the WHERE predicate with partition columns to minimize rewritten files:

UPDATE users
SET status = 'inactive'
WHERE region = 'us-east-1'
  AND last_login_ts < TIMESTAMP '2023-01-01 00:00:00 UTC';

DELETE

DELETE FROM events WHERE dt = DATE '2024-01-14';                    -- partition-aligned: rewrites minimal files
DELETE FROM events WHERE event_type = 'test' AND dt < DATE '2024-01-01';

MERGE — canonical patterns

Conditional upsert:

MERGE INTO users t
USING user_updates s ON t.user_id = s.user_id
WHEN MATCHED AND s.updated_at > t.updated_at
  THEN UPDATE SET email = s.email, status = s.status, updated_at = s.updated_at
WHEN NOT MATCHED
  THEN INSERT (user_id, email, status, updated_at)
       VALUES (s.user_id, s.email, s.status, s.updated_at);

SCD Type 2:

MERGE INTO customers t
USING customer_updates s
  ON t.customer_id = s.customer_id AND t.is_current = true
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET is_current = false, expiry_date = s.updated_at
WHEN NOT MATCHED THEN
  INSERT (customer_id, name, email, effective_date, expiry_date, is_current)
  VALUES (s.customer_id, s.name, s.email, s.updated_at, null, true);

Insert-only (dedup load):

MERGE INTO target t
USING staging s ON t.id = s.id
WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val, s.updated_at);

TRUNCATE

TRUNCATE TABLE events;  -- instant; creates empty snapshot without file rewrites

---

Table Maintenance

Run in this order. Never skip REMOVE_ORPHAN_FILES after EXPIRE_SNAPSHOTS.

1. OPTIMIZE (compaction)

ALTER TABLE events EXECUTE optimize(file_size_threshold => '100MB');

-- Scope to recent partitions
ALTER TABLE events EXECUTE optimize
WHERE event_ts >= CURRENT_TIMESTAMP - INTERVAL '2' DAY;

-- Scope by file modification time
ALTER TABLE events EXECUTE optimize
WHERE "$file_modified_time" >= CURRENT_TIMESTAMP - INTERVAL '1' DAY;

2. EXPIRE_SNAPSHOTS

ALTER TABLE events EXECUTE expire_snapshots(
    retention_threshold          => '14d',
    retain_last                  => 10,
    clean_expired_metadata_files => true
);

3. REMOVE_ORPHAN_FILES

retention_threshold must be ≥ your cluster's query.max-run-time:

ALTER TABLE events EXECUTE remove_orphan_files(retention_threshold => '14d');

4. OPTIMIZE_MANIFESTS

ALTER TABLE events EXECUTE optimize_manifests;

5. ANALYZE (after any optimize)

ANALYZE iceberg_catalog.mydb.events;

Recommended schedule

-- Daily: compact recent partitions only
ALTER TABLE events EXECUTE optimize
WHERE event_ts >= CURRENT_TIMESTAMP - INTERVAL '2' DAY;

-- Weekly:
ALTER TABLE events EXECUTE expire_snapshots(
    retention_threshold => '14d', retain_last => 10,
    clean_expired_metadata_files => true
);
ALTER TABLE events EXECUTE remove_orphan_files(retention_threshold => '14d');
ALTER TABLE events EXECUTE optimize_manifests;
ANALYZE iceberg_catalog.mydb.events;

---

Time Travel and Rollback

-- Query a past snapshot by ID
SELECT * FROM events FOR VERSION AS OF 1234567890;

-- Query the state at a timestamp
SELECT * FROM events FOR TIMESTAMP AS OF TIMESTAMP '2024-01-14 12:00:00 UTC';

-- Daily delta
SELECT new.cnt - old.cnt AS delta
FROM (SELECT count(*) AS cnt FROM events) new,
     (SELECT count(*) AS cnt FROM events
      FOR TIMESTAMP AS OF CURRENT_TIMESTAMP - INTERVAL '1' DAY) old;

-- Roll back to a snapshot (creates a new snapshot)
CALL iceberg_catalog.system.rollback_to_snapshot(
    schema_name => 'mydb', table_name => 'events', snapshot_id => 1234567890
);
CALL iceberg_catalog.system.rollback_to_timestamp(
    schema_name => 'mydb', table_name => 'events',
    timestamp   => TIMESTAMP '2024-01-14 12:00:00.000 UTC'
);

List snapshots:

SELECT snapshot_id, committed_at, operation, summary
FROM iceberg_catalog.mydb."events$snapshots"
ORDER BY committed_at DESC LIMIT 10;

---

Metadata Tables

Metadata table	Contents
"t$snapshots"	Snapshot log: IDs, timestamps, operations
"t$history"	Schema/partition/property change log
"t$files"	All data files in current snapshot with sizes and column stats
"t$manifests"	Manifest files: paths, lengths, partition summaries
"t$partitions"	Per-partition: file count, record count, total size
"t$refs"	Named branches and tags
"t$properties"	Table properties key/value

Diagnostic queries

-- Partitions with too many small files → candidates for OPTIMIZE
SELECT partition, file_count,
       total_size / 1024 / 1024 AS total_mb,
       total_size / NULLIF(file_count, 0) / 1024 / 1024 AS avg_file_mb
FROM iceberg_catalog.mydb."events$partitions"
WHERE total_size / NULLIF(file_count, 0) < 64 * 1024 * 1024
ORDER BY file_count DESC;

-- Largest partitions
SELECT partition, record_count,
       total_size / 1024.0 / 1024 / 1024 AS total_gb
FROM iceberg_catalog.mydb."events$partitions"
ORDER BY total_size DESC LIMIT 20;

-- Snapshot growth rate
SELECT committed_at, operation, added_files_count, deleted_files_count
FROM iceberg_catalog.mydb."events$snapshots"
ORDER BY committed_at DESC LIMIT 20;

---

Anti-Patterns

Anti-Pattern	Problem	Fix
SELECT * on wide tables	Reads all columns; defeats columnar pushdown	Enumerate required columns
No ANALYZE after large loads	Default row estimates → bad join order	Run ANALYZE after every bulk DML
OPTIMIZE without ANALYZE after	Statistics reflect old file layout	Always run ANALYZE after OPTIMIZE
EXPIRE_SNAPSHOTS without REMOVE_ORPHAN_FILES	Orphan files accumulate	Always pair the two procedures
retention_threshold < longest query	Running queries read deleted files	Set threshold ≥ query.max-run-time
hour(ts) on low-volume tables	Millions of tiny files	Use day(ts) or month(ts)
Bucket count N not a power of 2	Uneven distribution	Always use powers of 2: 8, 16, 32, 64, 128...
Bucket count too high	Too many small files per partition	Aim for ~512 MB per bucket file
Range predicate on bucketed column	No bucket pruning — reads all buckets	Use equality or IN; for ranges, add a temporal partition
MERGE/UPDATE without partition predicate	Full-table COW rewrite	Include partition column in WHERE
AVRO for analytics tables	Row-oriented; no column skipping	Use PARQUET for all analytics tables
Skipping OPTIMIZE indefinitely	Small files accumulate → slow manifest scans	Run daily on recent partitions
Changing bucket count in-place	Not possible — old files use old bucket mapping	CTAS into new table with desired N, then rename
Using expressions on partition columns in WHERE	Prevents predicate pushdown to partition pruning	Use raw column: event_ts >= ... not date_trunc('day', event_ts) = ...

---

Output Expectations

When working on Trino + Iceberg tasks:

• Show the EXPLAIN (TYPE DISTRIBUTED) excerpt highlighting the problem (join type, estimated rows, exchange type).
• Propose specific CREATE TABLE DDL or ALTER TABLE SET PROPERTIES with explicit partitioning, sorted_by, and format_version.
• For bucketing questions: state the chosen N, explain why, show how to verify pruning with EXPLAIN (TYPE IO).
• Name maintenance commands in order and explain why order matters.
• State when ANALYZE is needed and which columns to include.
• Provide the $partitions diagnostic query to validate the fix after implementation.

---

References

• Local deep-dive spec: docs/specs/trino_iceberg_performance_optimization.md
• Trino optimizer: https://trino.io/docs/current/optimizer.html
• Trino Iceberg connector: https://trino.io/docs/current/connector/iceberg.html