[EPIC] Production-grade Apache Iceberg scans #2174

Goal

Make Apache Iceberg scans a first-class Ballista workload: correct across the client, scheduler, and executors; parallel on a real cluster; efficient for large and skewed tables; and observable and testable enough to prevent performance regressions.

This supersedes the narrower request in #890 for tracking purposes. The broader policy discussion for open table formats remains in #1241.

Current status

Draft PR #2217 adds an iceberg-ballista integration with logical and physical extension codecs, table/catalog reconstruction, and distributed integration tests. It is currently blocked on upstream iceberg-datafusion work and still notes that reads are not split across executors because the scan exposes a single output partition.

Upstream work:

• feat(datafusion): Add opt-in eager file scan planning with output partitioning iceberg-rust#2671 / Enable parallel file-level scanning for IcebergTableScan Datafusion Integration iceberg-rust#2220: opt-in eager file planning and multiple DataFusion output partitions
• [iceberg-datafusion] Balance eager scan task groups by estimated read size iceberg-rust#2962: size-aware grouping of scan tasks
• [iceberg-datafusion] Cache eager scan planning across equivalent scans iceberg-rust#2963: reuse planning results across equivalent scans
• EPIC: Support parallel scan in iceberg-datafusion iceberg-rust#1604: prior row-group/sub-file parallelism design (closed as not planned, retained as design context)

Workstreams

1. Integration foundation

• Land the standalone Iceberg integration and codec support in feat: Add Iceberg Support #2217
• Support a real scheduler with multiple executors, not only in-process execution
• Document how to register a table/catalog and run distributed Iceberg queries
• Keep the integration isolated from Ballista's core release/dependency graph as agreed in this issue's discussion

2. Distributed scan planning

• Consume upstream multi-partition scan support from feat(datafusion): Add opt-in eager file scan planning with output partitioning iceberg-rust#2671
• Propagate the Iceberg scan's output partitioning through Ballista without collapsing it to one task
• Verify that work is distributed across multiple executors
• Balance partitions by estimated bytes rather than file count ([iceberg-datafusion] Balance eager scan task groups by estimated read size iceberg-rust#2962)
• Define the path for sub-file/row-group parallelism so tables with a few large files can still use the cluster effectively

3. Pruning and planning efficiency

• Preserve projection and filter pushdown across Ballista plan serialization
• Verify snapshot, manifest, partition, file, and row-group pruning with scan metrics
• Preserve delete-file association and Iceberg correctness semantics in every partition
• Avoid repeated manifest/file planning for equivalent scans ([iceberg-datafusion] Cache eager scan planning across equivalent scans iceberg-rust#2963)
• Ensure metadata planning does not become a scheduler bottleneck for tables with many manifests/files

4. Correctness and resilience

• Test partitioned and unpartitioned tables
• Test schema evolution and time travel
• Test position and equality deletes where supported upstream
• Test empty scans, limits, selective filters, and scans with fewer files than target partitions
• Test executor loss/retry without duplicate, missing, or inconsistent results
• Keep Iceberg extension codecs composable with other Ballista extensions

5. Production object-store access

• Support the credential flows needed by REST catalogs and vended, short-lived, or path-scoped S3 credentials; coordinate with Proposal: pluggable S3 credential providers #2532
• Ensure credentials and catalog properties are redacted from logs and debug output
• Exercise a REST catalog and S3-compatible object store in end-to-end tests

6. Performance and observability

• Add reproducible multi-executor scan benchmarks covering many small files, balanced large files, skewed files, and highly selective filters
• Compare direct DataFusion/Iceberg execution with Ballista to isolate distribution overhead
• Report scan partitions, planned files/bytes, pruned files/bytes, manifest-planning time, and per-partition read time
• Add a regression baseline and document expected scaling as executors are added

Acceptance criteria

This epic is complete when:

1. A multi-file Iceberg scan fans out across multiple Ballista executors and returns the same results as direct DataFusion/Iceberg execution.
2. Projection, filter, snapshot/time-travel, partition pruning, and delete semantics survive client-to-scheduler-to-executor plan transport.
3. Scan task assignment is size-aware, and no single output partition becomes a major straggler solely because file sizes differ.
4. Equivalent scans do not repeatedly perform avoidable manifest/file planning.
5. End-to-end tests cover a REST catalog plus an S3-compatible store, including executor retry.
6. Benchmarks and scan metrics make correctness, pruning effectiveness, planning cost, distribution, and scaling visible.
7. User documentation provides a supported setup and a complete distributed-read example.

Non-goals

• Making Iceberg a dependency of Ballista's core crates
• Tracking distributed Iceberg writes in this epic, except where shared codec or correctness work is required for scans
• Reimplementing Iceberg scan planning inside Ballista when the capability belongs in iceberg-rust / iceberg-datafusion

Background

The original request was to bring the Ballista integration proposed in apache/iceberg-rust#2613 into the Ballista ecosystem with a separate crate and release cadence. That remains the intended integration boundary.