feat: serialize Iceberg file scan work units - #683
Open
alexanderbianchi wants to merge 2 commits into
Open
Conversation
alexanderbianchi
force-pushed
the
iceberg/issue-604
branch
from
August 26, 2026 22:54
e6ef26f to
93a76dd
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements a wire representation for Iceberg
FileScanTaskwork units so they can be serialized by the distributed work-unit transport.The implementation uses a protobuf envelope containing a MessagePack payload. It validates file ranges, projected field IDs, partition metadata, and canonical partition-value encodings while reconstructing the Iceberg task on the consumer.
Closes #604.
Commits
The PR is split into two independently working commits so the context-reuse trade-off is easy to review:
d08b3b1— adds a stateless, production-ready encoder where every work unit contains its complete task context.93a76dd— adds per-feed context reuse and the stateful decoder described below.Shared-context decoder
Iceberg file tasks commonly repeat the same schema, projection, predicate, partition spec, name mapping, and case-sensitivity setting. Although several of those values use
Arcin memory, serializing anArc<T>still writes the completeTfor every work unit.The second commit assigns repeated contexts an ID. The first work unit in each partition feed defines the context, and later work units reference it by ID. A decoder owned by that partition stream retains the definitions while reconstructing tasks. State is isolated per feed and per query; it is not shared across partitions, distributed tasks, or queries.
This reduces repeated serialization, allocation, and network bytes at the cost of an order-dependent per-feed protocol. A distributed transport test verifies that context definitions and references remain ordered through coordinator batching, partition routing, and gRPC transport.
Taxi fixture measurement
Using the seven real file scan tasks from the checked-in taxi table, routed over its four partition feeds:
This is a deterministic encoded-size regression test, not a runtime benchmark. The taxi fixture is conservative because it has only seven files spread over four feeds; larger tables normally reuse each feed context across many more files.
Validation
cargo test -p datafusion-distributed-icebergcargo test --features integration --test work_unit_feed