Apache Iceberg

High-frequency streaming commits to Apache Iceberg turn metadata resolution into an $O(N)$ tree-traversal bottleneck, degrading query planning from sub-second execution to minutes of driver stall.

When streaming engines like Apache Flink or Apache Spark Streaming write micro-batches every 10 to 60 seconds to satisfy low-latency data SLAs, they write tiny Parquet files (typically 1 MB to 20 MB) to object storage. While this satisfies real-time freshness, it introduces the small-files trap: an exponential expansion of data and metadata files that exhausts query engine memory, triggers storage rate limits, and multiplies query execution costs.

The Root Cause: Metadata Amplification and S3 Rate Limits

Object storage services like AWS S3 are optimized for throughput on large sequential blocks (128 MB–512 MB), not high-frequency metadata operations. At small file scales, performance collapses across two distinct vectors:

1. Object Storage I/O and Throttling: Every Parquet file read requires an initial GET request to pull the file footer, parse column chunk metadata, and calculate dictionary offsets. Reading a single partition containing 10,000 2 MB files requires 10,000 distinct HTTP GET requests before any actual record data is transferred. AWS S3 enforces a limit of 5,500 GET requests per second per prefix. High-concurrency engines like Trino easily hit 503 Slow Down errors when reading uncompacted streaming partitions.
2. Driver Memory Exhaustion: Apache Iceberg maintains ACIL (Atomicity, Consistency, Isolation, Durability) guarantees using an explicit tree of metadata files. Every micro-batch write creates a new table snapshot containing manifest lists and manifest files.

                        ┌─────────────────────────┐
                        │      Table Metadata     │
                        └────────────┬────────────┘
                                     │
                        ┌────────────▼────────────┐
                        │     Snapshot (v402)     │
                        └────────────┬────────────┘
                                     │
                        ┌────────────▼────────────┐
                        │      Manifest List      │
                        └──────┬────────────┬─────┘
                               │            │
             ┌─────────────────┘            └─────────────────┐
             ▼                                                ▼
┌─────────────────────────┐                      ┌─────────────────────────┐
│      Manifest File      │                      │      Manifest File      │
└──────┬────────────┬─────┘                      └──────┬────────────┬─────┘
       │            │                                   │            │
       ▼            ▼                                   ▼            ▼
┌───────────┐  ┌───────────┐                       ┌───────────┐  ┌───────────┐
│ Data File │  │ Data File │                       │ Data File │  │ Data File │
│  (1.2 MB) │  │  (2.1 MB) │                       │  (1.8 MB) │  │  (0.9 MB) │
└───────────┘  └───────────┘                       └───────────┘  └───────────┘

When query planners (e.g., Spark Driver or Trino Coordinator) evaluate a query, they must load these manifest files into heap memory to perform partition pruning and min/max stats filtering. If a table accumulates 500,000 small data files spread across 50,000 manifests, the coordinator engine can consume 32 GB+ of JVM heap space purely building the query split list, frequently triggering Garbage Collection (GC) pauses or Out-Of-Memory (OOMKilled) crashes.

The Compaction Mechanism

Solving this requires decoupled, asynchronous maintenance routines that rewrite physical files and re-structure the metadata tree without locking active readers or writers.

Compaction in Iceberg consists of two primary operations:

• Data File Compaction (rewrite_data_files): Merges small Parquet files into optimal size boundaries (typically 128 MB to 512 MB) while preserving partition boundaries and ordering constraints.
• Manifest Compaction (rewrite_manifests): Merges thousands of individual manifest files created by streaming micro-batches into larger manifest blocks, flattening the metadata tree.

Compaction Execution Pipeline

The execution engine reads uncompacted data files in parallel bin-packs, writes new merged Parquet files, and executes an atomic commit swapping the old file references for the new references.

UNCOMPACTED STATE                       COMPACTION PROCESS                       COMPACTED STATE

 Manifest List                           Spark Executor Pool                     Manifest List
  ├── Manifest 1                          ┌─────────────────┐                     └── Manifest Alpha
  │    ├── Data_01.pq (2 MB) ───────────► │ Read & Merge    │                           ├── Compacted_01.pq (512 MB)
  │    └── Data_02.pq (3 MB) ───────────► │ Bin-Packing     │                           └── Compacted_02.pq (512 MB)
  └── Manifest 2                          │ Sort / Z-Order  │
       ├── Data_03.pq (1 MB) ───────────► │ Write New Block │
       └── Data_04.pq (4 MB) ───────────► └────────┬────────┘
                                                   │
                                                   ▼
                                        Atomic Commit (v403)

Production Maintenance Configuration

Compaction should run as a background service using Apache Spark SQL or PySpark jobs scheduled via an orchestrator (such as Airflow or Dagster).

1. Data File Compaction Procedure (`rewrite_data_files`)

Execute this Spark SQL call on an automated schedule (e.g., every 2 to 6 hours depending on write volume):

CALL system.rewrite_data_files(
  table => 'prod_catalog.analytics.user_events',
  strategy => 'binpack',
  options => map(
    'target-file-size-bytes', '536870912',      -- 512 MB target size
    'min-file-size-bytes', '134217728',         -- Treat files < 128 MB as candidates
    'max-file-size-bytes', '671088640',         -- Upper bound 640 MB
    'min-input-files', '5',                     -- Skip groups with fewer than 5 files
    'max-file-group-size-bytes', '10737418240', -- Process max 10 GB per task group to prevent OOM
    'partial-progress.enabled', 'true',         -- Commit incremental batches if total job is huge
    'partial-progress.max-commits', '10'         -- Max intermediate commits
  )
);

2. Manifest Compaction Procedure (`rewrite_manifests`)

Data compaction creates large data files, but streaming writers still leave behind thousands of tiny manifest files. Run manifest rewriting after data compaction:

CALL system.rewrite_manifests(
  table => 'prod_catalog.analytics.user_events',
  use_caching => true
);

3. Historical Cleanup: Snapshot Expiration and Orphan Removal

Compaction does not delete old files; it creates a new snapshot pointing to new files. The old small files remain on object storage to support time-travel queries. If left unchecked, storage costs explode.

Run snapshot expiration to delete data files untracked by active snapshots:

-- Expire snapshots older than 3 days
CALL system.expire_snapshots(
  table => 'prod_catalog.analytics.user_events',
  older_than => TIMESTAMP '2026-03-27 00:00:00.000',
  retain_last => 100
);

-- Delete unreferenced orphan files from S3 created by failed streaming tasks
CALL system.remove_orphan_files(
  table => 'prod_catalog.analytics.user_events',
  older_than => TIMESTAMP '2026-03-29 00:00:00.000'
);

Compaction Strategy Trade-Offs

Choosing the correct strategy depends on your read workload patterns vs. compute budget:

Failure Modes and Operational Edge Cases

1. Optimistic Concurrency Control (OCC) Commit Conflicts

• Mechanism: Iceberg uses OCC for table commits. If a long-running rewrite_data_files job attempts to commit its final manifest update while a Flink streaming job commits a new micro-batch, the compaction job may fail with a CommitFailedException.
• Mitigation: Enable partial progress in the compaction call ('partial-progress.enabled' = 'true'). Set table property commit.retry.num-retries to higher limits (e.g., 20). Iceberg will attempt to rebase the compaction metadata onto the latest snapshot instead of throwing an exception.

2. Driver Memory Collapses During Initialization

• Mechanism: Before Spark can execute rewrite_data_files, the driver reads all file footers from the target partition. If a partition contains $>500,000$ files, the driver will throw java.lang.OutOfMemoryError: Java heap space.
• Mitigation: Scope compactions strictly to explicit partition sub-ranges using SQL predicates, rather than running blanket table-level operations:

  CALL system.rewrite_data_files(
    table => 'prod_catalog.analytics.user_events',
    where => 'event_date = "2026-03-30" AND event_hour = 12'
  );
  

3. Merge-On-Read (MoR) Delete File Accumulation

• Mechanism: If using Iceberg v2 tables with `write.delete.mode