[NoahKusaba]

Which issue does this PR close?

Part of #2174: the first step of its integration foundation, Iceberg scans and codec support, for Rust clients and Flight SQL clients. Split out of #2217 as its smallest reviewable piece: reads only, built on merged upstream code. Related: #1241.

Rationale for this change

datafusion-iceberg plans Iceberg scans, but its tables and scans hold live state (a loaded Table and its FileIO), so Ballista can't send them to the scheduler and executors. This PR adds codecs that can.

What changes are included in this PR?

A new crate, ballista-iceberg (integrations/iceberg), with a logical and a physical extension codec. A client readies its session with register_iceberg(state); a separately deployed scheduler and executors set IcebergLogicalCodec and IcebergPhysicalCodec as their codec overrides. No changes to Ballista core.

• Each table and scan travels as one version of the table: the location of its metadata file plus its serialized FileIO. Metadata files are immutable, so every node rebuilds exactly that version without a catalog. Tables sent from the client also carry the metadata the client already loaded, so the scheduler doesn't read the file. Scans sent to executors carry only the location, keeping each task's plan small.
• Catalog-backed tables (IcebergCatalogTableProvider, including the ones a mounted IcebergCatalogProvider serves) are pinned on the client while the query is planned. register_iceberg wraps the session's catalogs so that DataFusion's async table lookup, which runs once per table a query names, loads the table and returns it at that version. Every scan of a table in one query, such as both sides of a self-join, reads the same version; each new query reads the latest one; and the scheduler and executors need no catalog configuration. Encoding does no I/O, so the client can run on any tokio runtime.
• Static tables (IcebergStaticTableProvider) send the version they hold, for time travel.
• SQL clients: Flight SQL clients (JDBC, ADBC, BI tools) send SQL for the scheduler to plan, so they have no catalog. iceberg_session_builder(base, catalogs) wraps the scheduler's session builder, passed to SchedulerConfig::with_override_session_builder, to mount Iceberg catalogs on its sessions and pins their tables there the same way. This also gives the scheduler the catalogs that scheduler-side commits ([Iceberg] Run a job's final commit step on the scheduler, not an executor #2622) will need.
• Framing: each payload starts with a tag byte, Iceberg (1) or delegated (0), which also versions the payload format. Everything else goes to the codec the session already had, so other Ballista extensions keep working. A driver for another table format, such as Delta, maintained outside this repo, can wrap these codecs the same way, so one query can join tables from both. It only has to be stacked in the same order on every node.
• Rebuilt tables run on a runtime that lives as long as the process, so they outlive whatever runtime calls the codecs. Each process caches up to 16 table versions. The cap bounds memory, since a table with a long history can hold megabytes of metadata. It isn't configurable yet: a session setting can't govern a per-process cache, and a constructor option can be added later without breaking anything.

Known limitations, also listed in the docs:

• Reads only. INSERT INTO fails with "isn't supported on Ballista yet".
• One task per scan, until Iceberg scans split into partitions.
• A catalog-backed table's version is fixed when the query is planned. A view keeps the version from when it was created, because DataFusion stores a view as a planned query; a DataFrame or Flight SQL prepared statement run twice likewise reads one version. A test pins this down. Follow-up: re-plan views from their SQL on each query, as Spark does.
• No metadata tables (t$snapshots).
• Storage credentials travel in plain text, so the links between nodes must be trusted.

Dependencies: iceberg-rust and datafusion-iceberg are pinned to git revisions until they release, with iceberg at the revision datafusion-iceberg pins so the build resolves a single iceberg crate. The crate is publish = false and needs Rust 1.95; the rest of the workspace still builds on 1.94. Almost all of the Cargo.lock diff (about 90 new packages) is iceberg's own dependency tree, plus opendal for S3 access.

Are these changes tested?

Yes:

• Codec tests:
  • a scan keeps its snapshot, projection, filter and fetch through encode and decode;
  • a self-join on a catalog-backed table is pinned to one version when planned, decodes without its metadata file (the test deletes it), and an INSERT into it is refused;
  • an unpinned catalog-backed table is refused with an error naming register_iceberg;
  • non-Iceberg nodes are handed to the inner codec;
  • malformed buffers (empty, an unknown tag, an invalid payload) are rejected.
• Standalone Ballista over a local memory catalog: a mounted catalog reads commits made after it was registered, a static table keeps its version, and a view keeps the version from when it was created.
• Flight SQL: a client with no catalog queries a table through a scheduler built with iceberg_session_builder, and sees a commit made between two queries.
• The same checks against S3, through an Iceberg REST catalog backed by RustFS. These run with testcontainers behind the testcontainers feature, which the existing linux-test job already enables.

All end-to-end tests run on a single-threaded tokio runtime.

Are there any user-facing changes?

Yes, a new crate and a docs page:

• docs/source/user-guide/iceberg.md, covering usage, time travel, SQL clients, how it works and the limitations;
• a runnable example: cargo run -p ballista-iceberg --example standalone-iceberg-scan.

No existing APIs change.

[NoahKusaba]

PR is ready for those interested, as the description says, we will need to wait for new releases for datafusion-iceberg and iceberg-rust crate.