[alexanderbianchi]

Summary

Plan pruned Iceberg file work before physical optimization, then optionally balance whole-file work across tasks and route those tasks toward stable data homes.

Base: main-upstream-dd-13-08-2026 (includes upstream merged dynamic filtering, including datafusion-contrib#639). The fork's default main is not the appropriate integration base.

Pruning-aware planning

• Both Iceberg table providers discover pruned files in TableProvider::scan(), when predicates and projection are available.
• Retain native FileScanTasks in query-local PlannedFiles, shared across plan clones and reused during execution.
• Expose planned_files / planned_bytes. Filtered/deleted row counts remain inexact; an empty file plan reports exact zero rows.
• Estimate desired scan task count from pruned bytes, capped by file work. This is a soft hint reconciled with the rest of the stage, not a hard stage-wide cap.
• Default assignment visits distributed tasks before filling local partitions.
• Account for estimated retained metadata in the query memory pool; fail rather than truncate at iceberg.planning_max_files (default 100,000).
• Use the configured Iceberg IO runtime and cancel collection when planning is abandoned. iceberg.plan_files=false retains execution-time discovery.

Optional whole-file balancing and affinity routing

• Add generic greedy_work_unit_assignment() in DFD: largest-first placement onto the least-loaded task, with deterministic ties and exactly-once input-index coverage. It does not split work or choose workers.
• iceberg.greedy_file_assignment=true uses candidate-file bytes for that allocator. Stable file path/range ordering avoids manifest-enumeration tie instability. Assignments are retained per scan/task count, with memory reservations. A separate scoped reservation covers temporary order/cost arrays, allocator ordering/load-heap storage and vector-growth headroom; it is released after construction.
• Add the default-empty WorkUnitFeedProvider::task_affinity() hook. iceberg.file_task_affinity=true supplies this task's assigned immutable data-file paths and byte weights, using the same assignment as delivery.
• Capture hints with scan-local task contexts before serialization converts native feeds to remote handles. Expose them through RouteTaskEvent::work_unit_affinity() without adding required fields to that event or changing the worker protocol.
• Add opt-in AffinityRouteTaskHandler: each distinct key elects a worker by rendezvous hashing; the whole task prefers the worker owning the greatest total hinted weight. Repeated keys count once using their maximum weight. Query IDs, task IDs and resolver ordering do not influence file homes. Equal affinity scores use a separate hash of the canonical file-key set and worker URL, rather than systematically preferring low worker URLs.
• Retain normal classified connection retry/failover. Unknown work defers to existing routing handlers. Custom routers can inspect hints and reuse rank_workers() while applying their own admission policy.
• Expose affinity attempts, successful preferred/fallback placements, failures, elapsed routing time, and per-task deduplicated hinted-object/byte totals. Placements use the returned worker URL; these are not cache-hit metrics. Iceberg scan metrics combine work-feed delivery and reader metrics.

Both Iceberg routing options default to false and require iceberg.plan_files=true. Enable them on the coordinating session and register AffinityRouteTaskHandler; the README shows the configuration. Neither the connector nor DFD depends on a cache implementation. Use stable worker URLs identifying individual cache owners, not a load-balanced service URL.

The reader, serialized work-unit fields and codec order are unchanged. No iceberg-rust dependency change or protocol bump is included. UNION branch grouping is unchanged.

Validation

cargo test -p datafusion-distributed -p datafusion-distributed-iceberg --features integration --lib --tests
cargo test -p datafusion-distributed -p datafusion-distributed-iceberg --features integration --doc -- --test-threads=2
cargo clippy -p datafusion-distributed -p datafusion-distributed-iceberg --features integration --all-targets -- -D warnings
cargo fmt --all --check

• 513 unit/integration tests passed; one existing test ignored.
• 19 doctests passed; two existing doctests ignored.
• Clippy with warnings denied, formatting and diff checks passed.
• Coverage includes routing hints matching actual delivered files, preserved query results, repeated-query assignment/worker stability, two isolated UNION scans of the same file selecting the same worker, disabled-option fallback, required-planning validation, skew/zero/large cost assignment, duplicate-key handling and worker membership changes.

Both review regressions were demonstrated failing before the fixes and passing afterward: a 20,000-task equal-weight two-file ranking test across ten workers, and a constrained-pool test that completes file discovery, rejects insufficient assignment scratch headroom, then verifies retry, scratch release and cached-assignment reuse. The ranking test is not a distributed-query benchmark.

Routing metric tests cover preferred placement, same-worker retries, failover, exhausted retries, terminal/resolver errors, deferred routing, duplicate/zero-weight hints and byte-total saturation. The Iceberg integration test verifies feed-delivery and reader metrics coexist on scan nodes.

Routing integration tests use the fixture and in-memory worker transport: they verify requested worker identities and results, not separate physical caches.

Limits / deployment validation still required

• Candidate-file bytes are not decoded memory, exact requested ranges, CPU cost or filtered output size.
• Affinity is best effort: mixed-file tasks, membership changes and failover may send the same file to different workers. The hash implementation can remap homes on upgrade.
• This policy does not inspect cache contents or live load and does not impose worker memory/concurrency admission. Keep admission controls and memory headroom when testing popular files.
• Cold/warm cache hit rate, coalesced misses, backend reads, per-worker peak memory, skew and latency still need measurement on the target service. No cache-hit or isolated end-to-end speedup is claimed.
• Sub-file splitting, query-result reuse, cache implementation changes, UNION regrouping and distributed dynamic filtering are out of scope.

Documentation: The life of an iceberg file, Iceberg README, and work-feed/worker-routing guides.