Skip to content

feat: serialize Iceberg file scan work units - #683

Open
alexanderbianchi wants to merge 2 commits into
datafusion-contrib:iceberg-0.10from
alexanderbianchi:iceberg/issue-604
Open

feat: serialize Iceberg file scan work units#683
alexanderbianchi wants to merge 2 commits into
datafusion-contrib:iceberg-0.10from
alexanderbianchi:iceberg/issue-604

Conversation

@alexanderbianchi

@alexanderbianchi alexanderbianchi commented Aug 26, 2026

Copy link
Copy Markdown
Collaborator

Summary

Implements a wire representation for Iceberg FileScanTask work 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:

  1. d08b3b1 — adds a stateless, production-ready encoder where every work unit contains its complete task context.
  2. 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 Arc in memory, serializing an Arc<T> still writes the complete T for 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:

  • repeated complete context: 6,552 bytes
  • shared per-feed context: 4,425 bytes
  • reduction: 2,127 bytes (32.5%)

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-iceberg
  • cargo test --features integration --test work_unit_feed
  • targeted Clippy checks for the Iceberg library and distributed work-unit integration test
  • formatting and diff checks

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

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant