feat: serialize Iceberg data source plans by alexanderbianchi · Pull Request #684 · datafusion-contrib/datafusion-distributed

alexanderbianchi · 2026-08-26T23:33:50Z

Summary

Implements IcebergCodec so physical plans containing IcebergDataSource can be serialized by the coordinator and reconstructed on workers.

The generated protobuf preserves the source schema, work-unit feed, DataFusion partitioning, fetch limit, and storage properties. Decoding rebuilds FileIO using the worker-local storage factory and Iceberg runtime. Local statistics retain the existing current-snapshot behavior; remote sources report unknown statistics rather than serializing planning-only information.

The Iceberg session integration registers both the plan codec and work-unit feed getter. A harness-backed codec round-trip test covers schema, partitioning, fetch, feed identity, storage properties, and remote unknown statistics.

Closes #603.

Scope

Based on iceberg-0.10, which now includes #683's file scan task serialization and #690's unknown-statistics fix.

Per review, distributed harness setup and end-to-end distributed tests are deferred until #605 supplies the real task-count handler. This PR does not add a temporary fixed-count handler or a separate distributed harness mode.

Explicit non-empty storage-property round-trip coverage is tracked in draft #703, based on #700's harness builder. This avoids a hidden test dependency on default fixture options.

Validation

cargo test -p datafusion-distributed-iceberg
cargo clippy -p datafusion-distributed-iceberg --lib -- -D warnings
cargo fmt --all -- --check
git diff --check origin/iceberg-0.10...HEAD

gabotechs

Nice! this is going in the right direction, left some comments that should make things a bit easier.

gabotechs · 2026-08-31T08:51:47Z

+    #[prost(uint64, tag = "3")]
+    partitions: u64,
+    #[prost(enumeration = "PartitioningKind", tag = "4")]
+    partitioning: i32,
+    #[prost(uint64, optional, tag = "5")]


This proto message design is likely not going to age well. As soon as Range or Hash partitioning are supported, new fields for holding the SplitPoints and the hashing expressions will need to be added, and consistency across fields will not be granted at compile time.

Take NetworkShuffleExec as an example:

https://github.com/datafusion-contrib/datafusion-distributed/blob/main/src/codec/distributed_codec.rs#L427

You can just rely on upstream functions for [de]serializaing Partitioning

gabotechs · 2026-08-31T08:52:35Z

+    #[prost(uint64, tag = "7")]
+    num_rows: u64,
+    #[prost(uint64, tag = "8")]
+    total_byte_size: u64,
+}


As mentioned in the other message, we should be good without this.

gabotechs · 2026-08-31T08:54:56Z

+        let codec = IcebergCodec::new(
+            Arc::new(FixtureStorageFactory::default()),
+            iceberg::Runtime::current(),
+        );
+        let mut bytes = Vec::new();


Ideally, the codec should be injected by the testing harness, not manually in each test.

gabotechs · 2026-08-31T08:57:26Z

+        let (mut ctx, _guard, _) = start_localhost_context(2, build_worker_state).await;
+        ctx.set_iceberg_integration(integration_options());
+        ctx.set_distributed_desired_task_count_handler(iceberg_test_task_count);
+        ctx.sql(&format!(
+            "CREATE EXTERNAL TABLE taxi STORED AS ICEBERG \
+             LOCATION '{FIXTURE_URI}/metadata/v1.metadata.json'"
+        ))
+        .await?
+        .collect()
+        .await?;
+


Here, we should be using the IcebergTestHarness rather than constructing the context manually.

gabotechs · 2026-08-31T09:01:28Z


 #[derive(Debug, Clone, Serialize, Deserialize)]
-struct FixtureStorageFactory {
+pub struct FixtureStorageFactory {


I see this was made public so that the integration tests can use it, but the integration tests should be using the IcebergTestHarness itself instead.

The main purpose of the IcebergTestHarness is to reduce all the boilerplate and preparation code from each individual test. If you see yourself needing to prepare a DataFusion SessionContext manually inside a test, it means that the harness should instead be used.

This means that maybe you need to extend a bit the harness to hold more functionality, like spawning a fleet of localhost workers while enriching the context with distributed extensions for example. This is fine, it's part of the expected natural evolution of IcebergTestHarness

gabotechs

I think it's worth deferring changes to the test harness until we ship #605. I see the new functions in the harness are only used by one small test, so I think we can just remove it from this PR, and add a bunch of more tests in an follow up once we get the DesiredTaskCountHandler in place.

The rest looks pretty good! so +1

gabotechs · 2026-09-04T07:12:13Z

+    #[cfg(feature = "integration")]
+    _worker_guard: Option<datafusion::common::runtime::JoinSet<()>>,
 }


🤔 all the feature checks in this file are getting a bit hard to manage. Probably the whole harness should be hidden behind the integration flag, not just pieces of it.

gabotechs · 2026-09-04T07:18:42Z

+#[cfg(feature = "integration")]
+fn iceberg_test_task_count(
+    event: DesiredTaskCountEvent,
+) -> Option<Result<DesiredTaskCountEventResponse>> {
+    event


I can imagine how this will disappear in the near future:

After #605, we'd have one of these handlers that has a configured "bytes_per_partition"-like setting that we can set in the IcebergConfig to something unreasonably low for forcing distribution.

gabotechs · 2026-09-04T07:19:37Z

+    }
+
+    #[cfg(feature = "integration")]
+    pub async fn new_distributed() -> Result<Self> {


🤔 I can imagine how we'd want the distributed planner to be enabled in every test, whether the test ends up distributing or not.

alexanderbianchi · 2026-09-05T17:51:35Z

+            source.iceberg_file_io.config().props(),
+            decoded.iceberg_file_io.config().props()
+        );
+        assert!(!source.iceberg_file_io.config().props().is_empty());


This line depends on the update to the harness which is a really tricky link, I don't like that and would rather remove it for now and wait for #700

alexanderbianchi · 2026-09-05T18:05:55Z

Originally there were some conflicts inbound with #687 - i've stripped this PR to be more focused (good practice anyways) and @sandugood should have an easier time now. I do want to make sure a couple things don't fall through the cracks so I opened
#702
#703
Which build off #700 to add test coverage for this PR and 687

Extends `roundtrips_data_source_plan`, the existing codec test from #684, with two explicit storage properties supplied through #700's harness builder. Checks their decoded values, including a quote-containing value that exercises SQL literal escaping, while retaining the existing schema, partitioning, fetch, feed, property-map, and statistics assertions. The diff against #700 remains 15 changed lines in `iceberg/src/codec.rs`, with no new harness methods or test functions. Stacked on #700, now rebased onto `iceberg-0.10` at `6507bf8` after #687 merged. Only #700 is a prerequisite. Validation: - `cargo test -p datafusion-distributed-iceberg --locked` — 90 tests passed, including the codec roundtrip test and doctest - `cargo clippy -p datafusion-distributed-iceberg --tests --locked -- -D warnings` - `cargo fmt --all -- --check` - `git diff --check`

alexanderbianchi force-pushed the iceberg/issue-603 branch from 0ed3a5a to f87a81d Compare August 27, 2026 22:40

alexanderbianchi changed the base branch from iceberg-0.10 to iceberg/issue-604 August 27, 2026 22:40

gabotechs reviewed Aug 31, 2026

View reviewed changes

alexanderbianchi force-pushed the iceberg/issue-603 branch from f87a81d to 6f50e29 Compare September 2, 2026 14:39

alexanderbianchi force-pushed the iceberg/issue-604 branch from 93a76dd to b9b3316 Compare September 2, 2026 14:41

alexanderbianchi force-pushed the iceberg/issue-603 branch from 6f50e29 to ae2e0a6 Compare September 3, 2026 19:05

gabotechs approved these changes Sep 4, 2026

View reviewed changes

alexanderbianchi added 2 commits September 5, 2026 12:41

feat: serialize Iceberg data source plans

7366f2b

refactor: defer distributed Iceberg test harness until task sizing

d2533c9

alexanderbianchi force-pushed the iceberg/issue-603 branch from ae2e0a6 to d2533c9 Compare September 5, 2026 16:44

alexanderbianchi changed the base branch from iceberg/issue-604 to iceberg-0.10 September 5, 2026 16:44

refactor: limit Iceberg statistics changes to remote plan handling

0dcb698

alexanderbianchi commented Sep 5, 2026

View reviewed changes

test: defer explicit storage properties coverage to harness builder

9148a23

alexanderbianchi mentioned this pull request Sep 5, 2026

refactor: make Iceberg codec fixture options explicit #703

Merged

alexanderbianchi merged commit fab6425 into datafusion-contrib:iceberg-0.10 Sep 5, 2026
31 of 32 checks passed

This was referenced Sep 5, 2026

refactor: build Iceberg test metadata at runtime #700

Merged

refactor: construct historical Iceberg fixtures in tests #702

Open

Uh oh!

Conversation

alexanderbianchi commented Aug 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Scope

Validation

Uh oh!

gabotechs left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gabotechs left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexanderbianchi commented Sep 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

alexanderbianchi commented Aug 26, 2026 •

edited

Loading

alexanderbianchi commented Sep 5, 2026 •

edited

Loading