It started with a slow Trino query on Sunday morning and ended four hours later with a rewritten compaction DAG. Here's the timeline and what I changed.
Background
The zoning pipeline writes Iceberg tables in micro-batches: each Airflow task run produces 50–200 small Parquet files. Over three months of daily runs, the documents table had accumulated ~18,000 files averaging 2 MB each — well into small-files territory for Trino.
The fix is standard: run a compaction job that rewrites the small files into larger ones. I had a scheduled Airflow DAG for this, running nightly at 02:00.
The DAG had been failing silently for two weeks.
Timeline
08:14 — Trino query on documents (used by the Superset pipeline health dashboard) takes 4m 12s. Normal is ~18s.
08:22 — Check Trino query profile. 18,000 file reads, average 2 MB. Classic small-files.
08:35 — Check compaction DAG logs. Last successful run: 2026-03-14. Subsequent runs: TaskFailed with no useful error message.
08:50 — Find the actual error: the DAG was calling ALTER TABLE ... EXECUTE optimize via Trino. The Trino coordinator had been restarted after a config change on March 15, and the JDBC connection string in the Airflow connection was using the old internal hostname.
09:10 — Fix the connection string. Trigger the compaction DAG manually. It runs — but takes 2h 40m to compact 18,000 files.
11:55 — Compaction completes. Trino query: 19s. Superset dashboard loads normally.
What went wrong
Three things combined:
1. Silent DAG failure. The TaskFailed state was being swallowed by a bare except Exception: pass in the DAG's error handler. I added it months ago to prevent alert fatigue from transient Trino timeouts. It also swallowed the actual failure.
2. No connection health check. The Airflow connection to Trino had no test_connection health check. A connection string that points at a dead hostname fails only at task execution time.
3. No compaction lag alert. I had uptime monitoring (Uptime Kuma) but no alert for "compaction hasn't run in N days." The problem was invisible until query performance degraded.
What I changed
Remove the bare except. The error handler now catches TrinoQueryError specifically and re-raises everything else.
Add a compaction freshness check. A new DAG task reads the Iceberg table metadata to check the last compaction timestamp. If it's older than 48 hours, the task fails with a clear message before attempting compaction.
Alert on compaction lag. Added a Grafana alert: if last_compaction_age_hours > 36, page via Slack.
Incremental compaction. Instead of rewriting all 18,000 files in one job, the new DAG compacts the oldest 10% of files per run. The first run after a gap is still slow, but subsequent runs are fast and the risk of a 3-hour blocking job is gone.
The rewritten DAG (key parts)
@task
def check_compaction_freshness(table: str, max_age_hours: int = 48) -> None:
last_run = get_last_compaction_timestamp(table)
age = (datetime.utcnow() - last_run).total_seconds() / 3600
if age > max_age_hours:
raise ValueError(f"{table}: last compaction {age:.1f}h ago (limit {max_age_hours}h)")
@task
def compact_oldest_files(table: str, fraction: float = 0.10) -> dict:
# Target only the oldest fraction of snapshots
result = trino_execute(f"""
ALTER TABLE {table}
EXECUTE optimize
WHERE "$snapshot_id" IN (
SELECT snapshot_id FROM "{table}$snapshots"
ORDER BY committed_at ASC
LIMIT (SELECT CAST(COUNT(*) * {fraction} AS INTEGER) FROM "{table}$files")
)
""")
return {"files_compacted": result.rowcount}
Lesson
A one-operator stack has no one to notice when a background job fails quietly. Every maintenance job needs an explicit freshness check and an alert — not just a success/failure state in Airflow's UI that nobody looks at on Sunday morning.