[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 are readied with SchedulerConfigExt::with_iceberg and ExecutorProcessConfigExt::with_iceberg (the scheduler and executor features). 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. Only the location is sent, not the metadata, so a plan's size doesn't grow with the table's history; each node reads a version's metadata once and caches it.
• 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. A view keeps the version from when it was created; feat(iceberg): re-pin a view's Iceberg tables each time a query names it #2784 re-pins its tables on each query. 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. SchedulerConfig::with_iceberg(catalogs) wraps the scheduler's session builder to mount Iceberg catalogs on its sessions, pins their tables there the same way, and adds the Iceberg codecs. 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.
• Composing codecs: register_iceberg and with_iceberg add the Iceberg codecs after the session's existing codecs, with Ballista's ComposedLogicalExtensionCodec (feat: add ComposedLogicalExtensionCodec #2752) and DataFusion's ComposedPhysicalExtensionCodec. Each payload records which codec wrote it, so other Ballista extensions keep working, and a driver for another table format, such as Delta, maintained outside this repo, can be composed alongside, so one query can join tables from both. The Iceberg codecs handle only Iceberg tables and scans, and fail on everything else.
• 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, keyed by metadata location and FileIO, so a catalog that vends new credentials on every load makes each query rebuild the table. 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. The physical codec refuses a scan reporting more than one partition, since Ballista can't restrict a custom DataSource to each task's partitions and every task would read all of them. A one-partition scan in a stage the adaptive planner splits between tasks, such as under UNION ALL, has the same problem; the fix needs feat: let custom ExecutionPlan nodes restrict themselves to a task's partitions #2721 (see the review thread).
• A catalog-backed table's version is fixed when the query is planned, so a view keeps the version from when it was created, and a DataFrame or Flight SQL prepared statement run twice reads one version.
• ctx.catalog(name) returns the pinning wrapper, so it doesn't downcast to the registered catalog's type; keep the Arc you registered.
• No metadata tables (t$snapshots).
• No encrypted tables (encryption.key-id): the scheduler and executors have no key management client.
• Call with_iceberg once, passing the scheduler every catalog; a second call adds the codecs again.
• Storage credentials travel in plain text, so the links between nodes must be trusted.

Dependencies: the Ballista crates are pinned to a commit on main that includes #2752, and move to crates.io once a release includes it. 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; Ballista itself still builds on 1.94.

Release boundary: as suggested in #2174, the crate is its own workspace, excluded from Ballista's like python/, with its own Cargo.lock and CI workflow (.github/workflows/iceberg.yml). Ballista's root Cargo.toml changes only by that exclusion and its Cargo.lock doesn't change. Because the crate depends on a pinned Ballista commit rather than the crates beside it, and the workflow doesn't run on changes under ballista/, a DataFusion upgrade iceberg-rust doesn't support yet doesn't block Ballista's build or release.

Are these changes tested?

Yes:

• Codec tests:
  • a scan keeps its snapshot, projection, filter and fetch through encode and decode with the session's composed codecs;
  • a self-join on a catalog-backed table is pinned to one version when planned, and an INSERT into it is refused;
  • an unpinned catalog-backed table is refused with an error naming register_iceberg;
  • the session's existing codecs stay first in the composed codecs, and the Iceberg codecs refuse non-Iceberg tables and plans;
  • a malformed Iceberg payload is 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 readied with SchedulerConfigExt::with_iceberg and an executor readied with ExecutorProcessConfigExt::with_iceberg, 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 iceberg.yml enables with --all-features.

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 --example standalone-iceberg-scan from integrations/iceberg.

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.

[NoahKusaba]

Hey @wirybeaver, we need to establish a convention that lets different table-providers interop with each other, so for example, a single query could read Hudi and Iceberg tables together.

This is the standard I'm trying to establish: have HudiLogicalExtensionCodec::new(inner) wrap config.ballista_logical_extension_codec() instead of hard-coding BallistaLogicalExtensionCodec, and forward every method to inner, including the UDF ones.

Let me know if this works for you.

[wirybeaver]

@NoahKusaba, the codec composition convention makes sense, including forwarding the UDF methods. I think we should establish the build and release boundary first, before landing individual format integrations, including the codec composition.

This was the motivation behind my proposal in #1241 and the split between #2388 and #2389: a minimal, SDK-free, opt-in build gate on the Ballista side, with the actual format integration owning its dependencies separately. The same convention should apply to Iceberg, Hudi, Lance, and Delta.

Your putting Iceberg in a separate crate is a useful step, but keeping it in Ballista’s root workspace and lockfile still couples dependency resolution. For example, Ballista may upgrade DataFusion before a format’s adapter supports that version. We should be able to release Ballista without waiting for every format integration to catch up.

I’d suggest establishing the minimal gate first, with format SDKs and implementations outside Ballista’s main workspace, root lockfile, and release graph, then adding Iceberg against that boundary. Each integration could maintain its own compatibility requirements and release cadence.

The gate itself would not make incompatible DataFusion versions interoperable; it would keep a format adapter’s compatibility constraints from becoming a prerequisite for Ballista’s build and release. Thus, the Codec composition comes to the rescue. Codec composition would then provide interoperability among integrations built against compatible versions.

[wirybeaver]

@andygrove, could you also take another look at whether it would make sense to reopen #2388 and #2389, with revisions as needed?

Since the earlier discussion, hudi-rs has matured further, including the merged DataFusion 55 / Arrow 59 compatibility work in apache/hudi-rs#698. I also now have a standalone Ballista–Hudi integration exercising the existing extension points.

With Iceberg integration progressing too, I think this is a useful time to revisit the shared build and release convention. #2388 could serve as the starting point for the minimal SDK-free gate, and #2389 could demonstrate the separately maintained format implementation. Establishing that boundary first would give Iceberg, Hudi, Lance, and Delta a consistent integration model while keeping their dependency and release schedules separate from Ballista’s.

I understand the earlier maintainer-bandwidth concern still applies. Would you be open to revisiting those PRs around this build and release boundary?

[milenkovicm]

I still see no reason to bring Hudi as dependency

[milenkovicm]

Would standard composable codec work with multiple codecs?

[wirybeaver]

@milenkovicm, I would separate this into two priorities:

P0: Establish the build and release boundary for Iceberg

Composable codecs address interoperability between integrations built against compatible dependencies. They do not establish an independent build or release boundary.

Suppose Ballista upgrades its DataFusion/Arrow stack while the Iceberg adapter still exposes types from the previous versions. Cargo can resolve both versions, but the integration may fail to compile where those types cross the adapter boundary.

This PR includes Iceberg in the root workspace and inherits Ballista’s dependencies, so that failure would affect workspace-wide CI. Keeping the build green would then require updating the adapter alongside Ballista or explicitly excluding it. That coordination burden is my concern; it does not mean the core crates are inherently unpublishable.

I suggest establishing an explicit opt-in build gate and independent release policy first: the integration should have its own workspace, lockfile, and CI, and remain on a compatible Ballista release until it is ready to upgrade. A feature flag alone is insufficient. Whether this requires an additional core gate or can use the existing extension points is something we can settle around that requirement.

We can focus entirely on Iceberg for this step.

P2: Consider other formats later

Hudi’s record-location indexes support efficient key-based upserts, while Lance’s scalar indexes can accelerate merge-insert operations. These offer useful capabilities for CDC and frequent-update workloads, and users may want Ballista to query those datasets.

We can defer those integrations. The immediate question is how to integrate Iceberg without making its dependency upgrade schedule part of Ballista’s core release obligations.

[wirybeaver]

Following up on the P0/P2 discussion, I have made the proposed boundary concrete in two PRs:

• Upstream merge proposal: feat: add an SDK-free Iceberg integration build gate #2731. Adds an opt-in, SDK-free Iceberg codec gate and documents the independent build/release boundary. It introduces no Iceberg SDK dependencies or root lockfile changes.
• Personal-fork demo: wirybeaver/datafusion-ballista#12. Adapts @NoahKusaba's implementation from feat(iceberg): scan Iceberg tables in Ballista #2638 to the gate in an excluded workspace with its own SDK dependencies, lockfile and CI. It is a draft demonstration, not intended for upstream merge and not a replacement for feat(iceberg): scan Iceberg tables in Ballista #2638.

The demo is stacked on the gate branch, so its diff isolates the integration changes. Its path dependencies exercise the unpublished gate; independently released adapters would select compatible released Ballista versions. This separates dependency resolution and release ownership, without claiming compatibility between different DataFusion/Arrow types.

[milenkovicm]

Thanks @NoahKusaba will try to review it tomorrow

[milenkovicm]

Thanks @NoahKusaba this is going to be a great addition. Few comments to start with

[milenkovicm]

I still see no reason to bring Hudi as dependency

[milenkovicm]

Would standard composable codec work with multiple codecs?

[milenkovicm]

What's the benefits of this approach over https://docs.rs/datafusion-proto/latest/datafusion_proto/physical_plan/struct.ComposedPhysicalExtensionCodec.html ?

[NoahKusaba]

ComposedPhysicalExtensionCodec only covers the physical codec. DataFusion has no logical counterpart, and Iceberg tables travel from the client to the scheduler through the logical codec, as table providers in the logical plan. So it can't combine Iceberg with another table format on that hop. The physical side alone wouldn't be enough.

Each codec wrapping the one inside it works the same way for both the logical and the physical codecs. It also fits Ballista's one-codec-per-session config: register_iceberg wraps whatever codec the session already has. And because each codec recognises its own prefix, nodes don't need to list codecs in the same order, whereas the composed codec labels payloads by list position.

[milenkovicm]

I'm not sure if just wrapping codecs is correct as there might be a cases where wrong codec can decode message not intended for it, please have a look at datafusion python there is a lot of work around codec composition there

[NoahKusaba]

You're right that a malicious/incorrect codec that is registered can decode messages not intended for it.
The alternative you're bringing up to copy datafusion-python's identity envelope design, would require a core-ballista/datafusion*** change (which I'm trying to avoid as it isn't entirely necessary).

I'll get you a draft up for you to consider.
Going to tinker with the design more.

[milenkovicm]

why can't we just copy ComposedPhysicalExtensionCodec and make ComposedLogicalExtensionCodec the whole point is to remember position of encoder ?

[NoahKusaba]

I promise to dig into that (my schedule is hectic recently).
You're thinking we implement ComposedLogicalExtensionCodec into datafusion-proto correct? Not just keeping it in ballista-core?

[milenkovicm]

we can start from here and move it to datafusion, lets see.
no rush i understand you

[NoahKusaba]

Now uses ComposedExtensionCodec stuff. Should be good for you to review again.

[andygrove]

This has come a long way, nice work. Since most of the diff is the lockfile, I don't think it needs a big split. One thing I'd consider pulling into a follow-up is the view re-pinning from 9025298. It came in after the last round and it's the trickiest bit of plan rewriting here, so I'd like to give it a focused review on its own. Landing views with the pinned-at-creation behavior documented as a limitation first works for me. Up to you though, happy to review it here if you'd rather keep it together.

Small nit on the description: it says the testcontainers tests run in the existing linux-test job, but since the move to its own workspace they run in iceberg.yml.

[NoahKusaba]

Thanks! Split the view re-pinning into #2784 (draft, stacked on this), and fixed the description: the testcontainers tests run in iceberg.yml.

[andygrove]

Thanks for the quick turnaround on the last round. The git pin, the trigger change, the partition guard and the note on the catalog wrapper all look good, and moving the view work into #2784 made this much easier to review. I built the crate and ran some extra queries against it. I found one case that returns wrong results, which I think needs sorting out before this lands, plus a few smaller things inline.

[andygrove]

I think the new guard covers one shape of this, but not all of them. A one-partition Iceberg scan under a UNION ALL still gets read by every task when the adaptive planner splits the union's stage. The task that owns the other branch asks the Iceberg leaf for zero partitions. restrict_plan_to_partitions doesn't know IcebergDataSource, so the leaf keeps its one partition and that task reads the whole table again.

I reproduced it on standalone Ballista with SELECT id, name FROM t UNION ALL SELECT id, name FROM t ORDER BY id. With ballista.scheduler.max_partitions_per_task = 1, every row comes back 4 times instead of 2. Default settings do the same when the executor has fewer free vcores than the union has partitions (I used ballista.standalone.parallelism = 1), which is normal on a busy cluster. A union of an Iceberg table and a Parquet table counted 9 rows instead of 6. The static planner gets it right, because can_stay_inline gives the branch its own stage. None of the current tests run a union, and an idle executor with enough vcores runs the whole stage as one task anyway.

The root cause is in Ballista core and is tracked in #2679. #2721 turns this into an error and adds a resolver on SchedulerConfig, so SchedulerConfigExt::with_iceberg could restrict the Iceberg leaf itself. It would keep the scan for [0] and swap in an empty zero-partition MemorySourceConfig for []. Would you be OK holding this until #2721 lands, then bumping the pin and adding a UNION ALL test? The standalone scheduler can't take a resolver, so the test would need to go through a scheduler readied with with_iceberg, like the one in tests/flight_sql.rs, with an executor that has fewer vcores than the union has branches. The docs line saying a multi-partition scan fails "rather than being read in full by every task" is also worth updating, since it reads as if a one-partition scan is always safe. Happy to share my repro test.