Lakehouse Iceberg & Delta skill for Codex

A healthy lakehouse table has files near the target size (128 MB-1 GB), a partition layout that matches the main filters, a bounded number of snapshots, no orphaned files, and one catalog that every engine agrees on. Maintenance is scheduled, not remembered.

Inspect first

1. Read AGENTS.md and find the engine and catalog config: Spark spark-defaults.conf or session builder, Trino etc/catalog/*.properties, .pyiceberg.yaml, dbt profiles.
2. Identify the catalog (REST/Polaris, AWS Glue, Hive Metastore, Nessie, Unity Catalog) and the table format version (Iceberg format-version 2 is the safe default; Delta protocol reader/writer versions decide which engines can read it).
3. Read the table’s metadata before changing anything:

-- Iceberg metadata tables (Spark or Trino; Trino quotes as "events$files")
select count(*), avg(file_size_in_bytes) / 1048576 as avg_mb from db.events.files;
select committed_at, operation, summary from db.events.snapshots order by committed_at desc limit 20;
select partition, record_count, file_count from db.events.partitions order by file_count desc limit 20;

-- Delta
describe detail events;           -- numFiles, sizeInBytes, partitionColumns
describe history events limit 20; -- operations, versions, who wrote

Most of these commands need network access to object storage and catalog credentials. In a Codex sandbox with network off, write the SQL and maintenance jobs and ask the user to run them, or request approval for network access.

Iceberg or Delta

Situation	Choose
Several engines (Spark, Trino, Flink, Snowflake, DuckDB) read and write the same tables	Iceberg with a REST catalog
Platform is Databricks end to end	Delta (Unity Catalog), optionally exposing Iceberg reads via UniForm
Existing Glue-based AWS lake with Athena	Iceberg on Glue
Need branch/tag style data versioning across tables	Iceberg with Nessie, or Iceberg branches and tags
Small single-node analytics on files	Plain Parquet plus DuckDB may be enough; adopt a table format when you need updates, deletes or concurrent writers

Do not switch formats for a single feature; both now cover upserts, schema evolution, time travel and compaction.

Table design

-- Spark SQL, Iceberg
create table lake.analytics.events (
    event_id    string,
    user_id     bigint,
    event_type  string,
    occurred_at timestamp,
    payload     string
)
using iceberg
partitioned by (days(occurred_at), bucket(16, user_id))
tblproperties (
    'format-version' = '2',
    'write.target-file-size-bytes' = '268435456',
    'write.delete.mode' = 'merge-on-read',
    'write.update.mode' = 'merge-on-read',
    'write.merge.mode'  = 'merge-on-read'
);

alter table lake.analytics.events write ordered by event_type, occurred_at;

Partitioning rules:

• Partition by what queries filter on, usually event date. Aim for partitions of at least a few hundred MB; daily partitions holding 5 MB each are over-partitioned, use months.
• Iceberg transforms (days, months, hours, bucket, truncate) are hidden: users filter on occurred_at and pruning still works. Never add a derived event_date column just for partitioning on Iceberg.
• Change layout with partition evolution, which only affects new data: alter table ... replace partition field days(occurred_at) with hours(occurred_at).
• Delta: prefer liquid clustering (cluster by (col)) on recent runtimes over partitioned by for high-cardinality or shifting access patterns. Partition Delta only on low-cardinality columns with more than about 1 GB per partition.

Copy-on-write vs merge-on-read

Workload	Mode
Append-mostly, rare large rewrites	copy-on-write (default): fastest reads
Frequent small updates, deletes, CDC upserts	merge-on-read: cheap writes, needs regular compaction to keep reads fast

Writing

merge into lake.analytics.users t
using staged_users s
on t.user_id = s.user_id
when matched and s.op = 'D' then delete
when matched then update set *
when not matched and s.op <> 'D' then insert *;

• Deduplicate the source to one row per key before MERGE; multiple source matches fail the statement (Spark raises an error rather than picking one).
• Include a partition predicate in the ON clause when the target is large and updates are recent (and t.occurred_at >= current_date - interval 7 days) so only those files are scanned.
• Streaming writers produce many small files and many snapshots. Use a trigger interval of a minute or more and compact downstream.

Maintenance (schedule all of it)

Iceberg via Spark procedures:

call lake.system.rewrite_data_files(table => 'analytics.events',
     where => 'occurred_at >= current_date - 2',
     options => map('min-input-files', '5'));
call lake.system.rewrite_position_delete_files(table => 'analytics.events');
call lake.system.expire_snapshots(table => 'analytics.events',
     older_than => timestamp '2025-10-01 00:00:00', retain_last => 20);
call lake.system.remove_orphan_files(table => 'analytics.events',
     older_than => timestamp '2025-10-25 00:00:00');
call lake.system.rewrite_manifests('analytics.events');

Trino equivalents: alter table events execute optimize, execute expire_snapshots(retention_threshold => '7d'), execute remove_orphan_files(retention_threshold => '7d').

Delta:

optimize events where occurred_at >= current_date() - interval 2 days;
vacuum events retain 168 hours;

Job	Cadence	Notes
Compaction of recent partitions	hourly to daily	Scope with a where filter; full-table rewrites are expensive
Snapshot expiry / VACUUM	daily	Retention must exceed the longest-running query and any time-travel promise
Orphan file removal	weekly	older_than at least 3 days so in-flight writes are not deleted
Manifest rewrite	weekly or when planning is slow	Iceberg only

Expiring snapshots, removing orphans and VACUUM permanently delete files and end time travel to older versions. Confirm retention with the user before the first run and never lower Delta’s 7-day retention check (spark.databricks.delta.retentionDurationCheck.enabled) to make a VACUUM pass.

Reading from other engines

# PyIceberg (pip install "pyiceberg[pyarrow]"); catalog config in ~/.pyiceberg.yaml
from pyiceberg.catalog import load_catalog

catalog = load_catalog("lake")
table = catalog.load_table("analytics.events")
arrow = table.scan(row_filter="event_type == 'purchase'", selected_fields=("user_id", "occurred_at")).to_arrow()

-- DuckDB: install iceberg; load iceberg;  (and the delta extension for Delta)
select count(*) from iceberg_scan('s3://bucket/warehouse/analytics/events');
select count(*) from delta_scan('s3://bucket/delta/events');

Reading by path skips the catalog; it can read a stale or uncommitted state. Prefer catalog-backed reads where the engine supports them.

Time travel and recovery

select * from lake.analytics.events for timestamp as of '2025-10-20 00:00:00';  -- Spark 3.3+
select * from lake.analytics.events version as of 8744736658442914487;
call lake.system.rollback_to_snapshot('analytics.events', 8744736658442914487);

Delta: select * from events version as of 42, restore table events to version as of 42. Rollback and restore change what every reader sees; confirm with the user first.

Migrating Parquet or Hive tables

• Iceberg migrate replaces a Hive table in place; snapshot creates a test copy that references the same files. Run snapshot first, validate counts, then migrate.
• add_files registers existing Parquet files without rewriting them; schemas must match.
• Delta: convert in place, adding the partition schema when the folder is partitioned:

convert to delta parquet.`s3://bucket/raw/events` partitioned by (event_date date);

Gotchas

• Two catalogs pointing at the same table location will corrupt it: each thinks it owns the metadata pointer. One table, one catalog.
• Schema evolution is by column ID in Iceberg; renaming in a raw Parquet reader that maps by name will show nulls. Read through the table format, not the files.
• Timestamp types differ (timestamp vs timestamptz); Iceberg timestamp without zone read in Trino or Snowflake can shift by the session time zone. Store UTC in timestamptz.
• Concurrent writers to the same partitions conflict and retry; heavy contention means the job layout is wrong, not that retries should be raised.
• remove_orphan_files with a short older_than deletes files of commits in progress.

Verification

• After compaction: average file size in files metadata rises toward the target and file count drops; a representative query’s scan bytes and planning time fall.
• After expiry: snapshot count matches the retention policy; storage size trends down.
• After migration: row counts and a column checksum (sum(hash(...)) or sums of numeric columns) match the source per partition.
• Record before/after numbers in the report:

**Table:** lake.analytics.events (Iceberg v2, REST catalog)
**Before:** 182,340 files, avg 3.1 MB, 4,912 snapshots, p50 dashboard query 41 s
**Actions:** compaction last 30 days, expire > 7 days keep 20, orphan cleanup > 3 days
**After:** 2,210 files, avg 256 MB, 20 snapshots, p50 query 6 s
**Scheduled:** compaction daily 02:00, expiry daily 03:00, orphans weekly (Airflow DAG lake_maintenance)