Skip to content

feat: Add Iceberg Support - #2217

Draft
NoahKusaba wants to merge 7 commits into
apache:mainfrom
NoahKusaba:feature/iceberg-ballista-crate
Draft

feat: Add Iceberg Support#2217
NoahKusaba wants to merge 7 commits into
apache:mainfrom
NoahKusaba:feature/iceberg-ballista-crate

Conversation

@NoahKusaba

@NoahKusaba NoahKusaba commented Aug 1, 2026

Copy link
Copy Markdown

Which issue does this PR close?

Closes #1241

What changes are included in this PR?

(ONLY INCLUDING DISTRIBUTED READS/WRITES integration for now)
(Note that READS are still not properly distributed, as output-partitioning is not supported in iceberg-datafusion yet apache/iceberg-rust#2671)
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

  • Snapshot pinning at encode time on the scheduler for reads.

Tests

  • Distributed reads / writes tested against dockerized minio iceberg catalog ( Standalone + multi-executor cluster). Also tests partitioned files writes + iceberg table registration.
  • Codec serialization / deserialization changes.
  • Snapshot Read tests.
    (will document better later)

Other notes

There is a dependency on the changes to the iceberg-datafusion crate, which are not yet merged in.
Currently referencing my branch found at apache/iceberg-rust#2613

@milenkovicm

Copy link
Copy Markdown
Contributor

Thanks @NoahKusaba will have a look tomorrow

@milenkovicm milenkovicm left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

instead of using docker compose for tests, would it make sense to use testcontainers? we already use it, you could find example at

async fn should_execute_sql_write() -> datafusion::error::Result<()> {

@NoahKusaba

NoahKusaba commented Aug 4, 2026

Copy link
Copy Markdown
Author

instead of using docker compose for tests, would it make sense to use testcontainers? we already use it, you could find example at

async fn should_execute_sql_write() -> datafusion::error::Result<()> {

Good call out, I made the change.

@andygrove

Copy link
Copy Markdown
Member

Thanks @NoahKusaba. I'm just getting up to speed with the contribution. My understanding is that this initial PR only works for in-process standalone mode and cannot be used in a real cluster yet, which is fine. It may be worth noting that in the PR description.

Perhaps you could add some minimal documentation in the PR as well, although I'd be fine with adding those as a follow on.

/// 1. the write commits exactly **one** atomic snapshot (not one per task),
/// 2. the parallel writers contributed multiple data files (one per region), and
/// 3. every input row lands exactly once (no loss, no duplication).
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@andygrove This PR does cover a true multi-executor cluster deployment.
This is the test that covers that, by making a standalone scheduler + executors.
I'll add in validation to ensure both executors did work (not just 1), and add an example for use.

@NoahKusaba
NoahKusaba marked this pull request as draft August 7, 2026 03:24
/// (`override_logical_codec` / `override_physical_codec`).
pub fn register_iceberg_codecs(config: SessionConfig) -> SessionConfig {
config
.with_ballista_logical_extension_codec(Arc::new(IcebergLogicalCodec::default()))

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would this replace existing codecs when used with other ballista extentions? Could this compose instead - read the config's current logical/physical codec and wrap that as inner (fall back to Ballista only
if unset)? cc @milenkovicm @andygrove

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch! Hadn't considered other codec's being used besides Ballista's default.
Will add it in a follow up PR.

I should also note that currently logical_codec's try_encode falls back to the set inner codec if it's not a icebergTableProvider, so this should get us the desired behavior.

@milenkovicm milenkovicm changed the title Iceberg-Ballista Crate feat: Add Iceberg Support Aug 8, 2026
@andygrove

Copy link
Copy Markdown
Member

@NoahKusaba I saw a notification a few days ago on a conversation about also moving the iceberg-datafusion integration out of the iceberg-rust repo, but now I cannot find that conversation. Would you mind sharing a link here?

@NoahKusaba

Copy link
Copy Markdown
Author

@NoahKusaba I saw a notification a few days ago on a conversation about also moving the iceberg-datafusion integration out of the iceberg-rust repo, but now I cannot find that conversation. Would you mind sharing a link here?

apache/iceberg-rust#2613 (comment)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[DISCUSS] Add open table format support.

4 participants