Repository navigation
feat: reconstruct eager Iceberg column statistics - #789
Draft
alexanderbianchi wants to merge 1 commit into
Draft
alexanderbianchi wants to merge 1 commit into
alexanderbianchi wants to merge 1 commit into
Conversation
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-byte-balanced-assignment
branch
from
October 20, 2026 23:22
fa1fd87 to
a72c891
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-selected-column-bytes
branch
from
October 20, 2026 23:22
4164009 to
37b1a5c
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-byte-balanced-assignment
branch
from
October 21, 2026 00:17
a72c891 to
839fece
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-selected-column-bytes
branch
from
October 21, 2026 00:17
37b1a5c to
4876898
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-byte-balanced-assignment
branch
from
October 21, 2026 00:20
839fece to
8b3e3e9
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-selected-column-bytes
branch
from
October 21, 2026 00:20
4876898 to
e29cb4e
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-byte-balanced-assignment
branch
from
October 21, 2026 05:49
8b3e3e9 to
07abef7
Compare
gabotechs
force-pushed
the
alex.bianchi/iceberg-byte-balanced-assignment
branch
from
October 21, 2026 10:53
a513d32 to
8c32533
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-selected-column-bytes
branch
from
October 21, 2026 21:20
e29cb4e to
66643ac
Compare
alexanderbianchi
changed the title
alexanderbianchi
changed the title
| let planned = self.feed.inner().and_then(|feed| feed.planned.clone()); | ||
| let snapshot = self.table_snapshot.clone(); | ||
| let (_cancel_on_drop, canceled) = oneshot::channel::<()>(); | ||
| let task = self.iceberg_runtime.io().spawn(async move { |
Collaborator
Author
There was a problem hiding this comment.
This is bad. It means we have to read manifests twice just because iceberg-rust doesn't expose it properly.
This was referenced Oct 22, 2026
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-byte-balanced-assignment
branch
from
October 26, 2026 23:52
2783b8f to
5b4cce9
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-byte-balanced-assignment
branch
4 times, most recently
from
October 27, 2026 16:31
316783c to
caaf79f
Compare
alexanderbianchi
added a commit
that referenced
this pull request
Oct 27, 2026
## What changes
Add opt-in eager Iceberg file discovery with
`iceberg.eager_file_planning=true`. Physical planning reads manifests,
applies Iceberg's pruning and delete matching, and retains the selected
`FileScanTask`s on an `EagerIcebergDataSource`. Distributed planning
assigns those files to task-local partitions before worker placement.
Each worker receives only its assigned descriptors in its execution plan
and reads them through the existing Iceberg reader.
Lazy discovery remains the default. Both modes use `DataSourceExec`,
shared reader configuration, and Iceberg's `FileIO`; eager planning does
not read Parquet data.
### Lazy
Planning uses snapshot statistics. During execution, the coordinator
reads manifests and progressively supplies file tasks to workers through
a remote work-unit feed.
This overlaps discovery with reading and can avoid some discovery when
execution stops early. Initial planning and worker placement happen
without the complete selected-file set.
### Eager
Planning completes discovery and retains the selected tasks before
execution starts. The source owns the assignment that routing inspects
and workers execute.
This adds startup latency, retained metadata, and larger worker-plan
payloads. In exchange, selected files are available for placement and
future task-sizing, balancing, and statistics improvements. This matters
especially when partition pruning reduces a large table to a handful of
files. Fully consumed scans generally require the same discovery I/O in
either mode.
```text
Lazy: plan → remote feed reference → coordinator discovers and dispatches → reader
Eager: discover → source owns files → assign task-local groups → route
→ serialize each task's groups → worker restores groups → reader
```
## Why replace #773
#773 moves discovery earlier but retains the files behind the work-unit
feed. Before routing, DFD converts native feeds into remote references,
hiding the retained descriptors from the router. Keeping that design
would require another way to preserve access to the files across the
conversion.
This PR instead lands source-owned eager execution directly against
`main`, following [Gabriel's suggestion to retain routing-relevant
information on the data
source](#797 (comment)).
It replaces #773 as the proposed foundation rather than merging eager
feeds and immediately refactoring them.
- **One authoritative assignment:** task-local sources share descriptor
storage on the coordinator. Routing borrows the source's assigned files;
serialization sends those same groups; workers restore them without
reassignment. There is no separate routing metadata copy.
- **Existing routing API:** like DFD's `FileScanConfig` path, the
scale-up handler creates task-specific `DistributedLeafExec` variants.
`RouteTaskEvent::task_specialized_plan` exposes the selected variant
before placement. No second plan view or work-unit-feed extension is
needed.
- **Shared reading, separate delivery:** lazy scans keep their feed
lifecycle; eager scans carry assigned files. Both produce file-task
streams for the same reader implementation. Independent codecs preserve
the existing lazy payload. Coordinator and worker codec registrations
must match.
- **Iceberg-native I/O:** the structure resembles file scans without
forcing Iceberg into `FileScanConfig` or supplying a placeholder
object-store URL.
- **Bounded retention:** descriptors are charged to the query memory
pool, with no arbitrary file-count cap. Abandoned eager planning cancels
its discovery future.
Lazy remains available for progressive startup and early termination.
Keeping it means maintaining two discovery/delivery lifecycles; this PR
isolates that difference rather than making eager execution depend on a
feed.
## Relationship to upstream datafusion-iceberg #67
Like [#67](apache/datafusion-iceberg#67), this
retains native `FileScanTask`s with a private memory-reservation owner,
exposes borrowed accessors on the source, and feeds local partition
streams into the Iceberg reader. Moving ownership out of DFD's feed
brings the eager representation closer to upstream.
DFD adds distributed assignment, placement, and reconstruction of
assigned groups from worker plans. The implementations are not
interchangeable yet: #67 uses one source with optional eager planning
and local lazy discovery; this PR uses separate eager/lazy source types
and retains DFD's remote feeds for lazy scans. Upstream would need to
support reconstructing an already assigned scan without rediscovery or
repartitioning before it could replace this implementation directly.
## Source and scheduling boundary
The eager source represents local file groups; it does not choose
distributed task assignments. `with_file_groups` derives a source using
positions in `planned_files()` (retained-file order), preserving
supplied group order and sharing descriptors and their memory
reservation. It rejects invalid or repeated positions and resets
metrics/statistics for the resulting scan.
The scale-up handler chooses task assignments through this public API.
The codec uses public schema, file, and partition accessors and
reconstructs received native descriptors with `from_file_groups`,
without rediscovery. `PlannedFiles` remains private. Reader
configuration and shared delete-loader behavior still need
reconciliation with upstream #67; this change separates scheduling
without changing reader behavior.
## Scope and follow-ups
This includes eager discovery, descriptor retention, task-first
assignment, routing access, and transport. It preserves current snapshot
statistics and task-size estimates; task-local eager variants and
decoded plans report unknown statistics instead of applying
whole-snapshot aggregates to a subset.
Selected-file sizing (#787), byte balancing (#788), and selected-file
statistics (#789) remain separate. Affinity policy can use the existing
routing hook outside this crate. Better selected-file metrics still need
iceberg-rust support to avoid rereading manifests. This PR makes no
performance claim.
## How to review
Start with [the eager-planning feature
commit](d6c8b6a).
The first four commits extract existing statistics, reader
configuration, lazy scan construction, and task-first partition
calculation.
The next two commits address review feedback separately: [module moves
and source
names](bd68327),
then [readable routing
tests](bad9344).
In the final tree, follow `table_provider/scan.rs` →
`eager/data_source.rs` → the scale-up handler → `eager/codec.rs`. Lazy
sources and codecs live under `lazy/`, with shared reading in
`common/scan_config.rs`. The old `IcebergDataSource` and `IcebergCodec`
exports are removed; consumers must use `LazyIcebergDataSource` and
`LazyIcebergCodec`, as documented in the upgrade guide.
The latest [source/scheduling separation
commit](96f55ed)
adds these public construction/regrouping operations and moves task
assignment into the scale-up handler.
## Validation
Coverage includes eager/lazy result parity, per-task distribution
snapshots, placement after pruning, isolated UNION branches, repeated
scan execution, and query-memory limits. Routing snapshots show the
assigned files, selected worker, and query results; normal distributed
execution exercises transport. Codec tests reject missing configuration
and malformed groups/descriptors.
The full Iceberg suite previously passed (61 tests and one doctest).
After the grouping-test cleanup, all library tests, grouping tests, and
routing tests passed with snapshot updates disabled. All-target Clippy
with warnings denied and formatting also passed.
Public grouping tests snapshot partition groups and actual dates/counts
read after regrouping and reconstruction. Invalid cases are labeled
explicitly; reservation lifetime is tested separately. The codec
validation test uses `encode_config` through public source accessors
without a synthetic plan roundtrip. Existing routing/distribution
snapshots remain unchanged.
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-byte-balanced-assignment
branch
2 times, most recently
from
October 27, 2026 19:27
d71fcd0 to
c540294
Compare
alexanderbianchi
force-pushed
the
alex.bianchi/iceberg-selected-column-bytes
branch
from
October 27, 2026 21:21
d268d6e to
692e362
Compare
This branch has not been deployed
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.
What changes
With
iceberg.column_stats_enabled=true, eager scans reconstruct column statistics from the files retained after pruning. Both eager and lazy paths share per-file extraction and aggregation helpers.The current Iceberg API drops column metrics from file tasks, so reconstruction uses another manifest pass on the configured I/O runtime. Empty selections skip that pass. Projected bytes inform relation costs; whole-file bytes still determine scan parallelism. No generic optimizer statistics provider is introduced.
Validation
Iceberg integration and unit tests, doctests, formatting, and strict Clippy. Metadata-only harness coverage includes selected-file bounds and counts, missing metrics, NDV capping, empty selections, nested fields, and unchanged task counts.
Stack
Based on #788, following #787 and the eager-source foundation in #803. Addresses the eager statistics reconstruction discussed in #795; selected-file NDV remains a snapshot-derived estimate.
The previous feed-based version is preserved at d268d6e.
No runtime speedup is claimed. Developed with AI assistance.