feat(datafusion-ballista) Add Initial Integration for Ballisa Datafusion

[NoahKusaba]

Which issue does this PR close?

• Closes [DISCUSS] Add open table format support. datafusion-ballista#1241

What changes are included in this PR?

(ONLY INCLUDING DISTRIBUTED READS/WRITES for now)

Adds a new iceberg-ballista crate, which provides a distributed-query driver for Apache Iceberg for a distributed datafusion engine Apache Datafusion-Ballista + the targeted changes to iceberg-datafusion that make Iceberg's existing plan nodes serializable so they can cross node boundaries.

The core problem it solves

Iceberg's DataFusion integration already produces complete physical read and write plans, but every Iceberg plan node holds live, non-serializable state (Arc, an open Table/FileIO). Ballista ships logical and physical plans to remote schedulers/executors, so those nodes couldn't travel. This branch closes that gap with one consistent idea: serialize a minimal self-contained recipe (IcebergCatalogConfig + identifiers), rebuild the live objects on the receiving node.

• IcebergLogicalCodec: serializes the catalog-backed table provider (config + table ident, plus snapshot/metadata variants) so the scheduler can rebuild it and do physical planning, including INSERT.

• IcebergPhysicalCodec: serializes the four Iceberg execution nodes (IcebergTableScan, IcebergWriteExec, IcebergCommitExec, IcebergMetadataScan) and the PartitionExpr physical expression.

• Tagged-envelope wire framing (TAG_ICEBERG / TAG_DELEGATED):every blob carries a leading tag; non-Iceberg nodes are delegated to Ballista's own codec, so shuffles/sorts/etc. keep working and an unknown tag is a hard error--> Based off comments from https://github.com/milenkovicm/ballista_delta

• bridge.rs runtime bridge: Each executor node needs to build an HTTP client with the iceberg catalog which requires an async-call, but PhysicalExtensionCodec from datafusion_proto try_decode is synchronous. The block_on function is a workaround to make this async function call blocking. Each try_decode also performs a load_table catalog round-trip per plan node to resolve the table's current metadata pointer

-Public API: register_iceberg_codecs(SessionConfig) and register_iceberg_table

• Snapshot pinning at encode time

Are these changes tested?

Ballista tests:

• Distributed reads / writes tested against dockerized minio iceberg catalog ( Standalone + multi-executor cluster). Also tests partitioned files writes + iceberg table registration
• roundtrips testing that configurations for nodes are maintained through serialization -> deserialization

Datafusion tests:

• Unit tests for config propagation and snapshot pinning.

[NoahKusaba]

Still need to put more time into this, but it's in a state where it's ready for feedback.

[blackmwk]

Thanks @NoahKusaba for this pr, but I don't think it's the right direction to maintain such a huge integration in this repo. Currently the review resources is quite limited, and most committers are not familiar with ballista. I think it would be better to put it in ballista repo

[milenkovicm]

I'd agree with @blackmwk perhaps we could move this to DataFusion contrib once the design/integration settles down

[NoahKusaba]

@milenkovicm Can you provide a link where you think this should go? Happy to put it wherever

[milenkovicm]

Once you're happy with integration we can discuss appropriate location

[NoahKusaba]

Thanks @NoahKusaba for this pr, but I don't think it's the right direction to maintain such a huge integration in this repo. Currently the review resources is quite limited, and most committers are not familiar with ballista. I think it would be better to put it in ballista repo

@blackmwk This still seems like the logical place to put it. This integration directly builds on the datafusion-iceberg integration that lives in this repo to interact with Iceberg. There already seems precedence for that with the Playground integration that lives in /integrations, also seems like other integrations with iceberg are parked here such as cache-moka.

At the very least I'll look to split up the iceberg-datafusion extensions in separate PR's over the coming weeks.
(still tightening up the full implementation).

[milenkovicm]

Adding ballista dependancy can make problems for maintainers as they have to wait for ballista release before they can release. I believe module with ballista dependancy should be moved somewhere else. If we move this to ballista, which at the end might be an option as well we (ballista) maintainers will get a dependency (iceberg) which we need to wait before release. anyway once we get integration done in good shape we can have discussion where to move it

[andygrove]

I'd really like to see the code stay in an Apache project rather than move to datafusion-contrib, and I am happy to help make that happen. It could potentially be a new repo under the datafusion org.

[andygrove]

I'd really like to see the code stay in an Apache project rather than move to datafusion-contrib, and I am happy to help make that happen. It could potentially be a new repo under the datafusion org.

@milenkovicm Could this be an additional crate (maybe optional) in the datafusion-ballista repo?

[milenkovicm]

that would be simpler than standalone repo.
Issue I have is that we would need to wait for updated iceberg before we can release ballista, but if you that's ok with others I have no problem. You have more experience with iceberg in n comet, if it works for comet should work for ballista I guess

[andygrove]

that would be simpler than standalone repo.
Issue I have is that we would need to wait for updated iceberg before we can release ballista, but if you that's ok with others I have no problem. You have more experience with iceberg in n comet, if it works for comet should work for ballista I guess

Would be good to get @mbutrovich input on this too.

[andygrove]

that would be simpler than standalone repo. Issue I have is that we would need to wait for updated iceberg before we can release ballista, but if you that's ok with others I have no problem. You have more experience with iceberg in n comet, if it works for comet should work for ballista I guess

I created issue in ballista to discuss this apache/datafusion-ballista#2174

[andygrove]

@NoahKusaba What do you think about moving this to the Ballista project? See discussion at apache/datafusion-ballista#2174

[NoahKusaba]

This branch now only holds icebeg-datafusion changes for Ballista to work, until the changes can be merged in.
The iceberg-ballista crate, has been moved to the ballista repo at: apache/datafusion-ballista#2217