Skip to content

feat: embed eager Iceberg assignments in scan inputs - #799

Closed
alexanderbianchi wants to merge 1 commit into
alex.bianchi/iceberg-selected-column-bytesfrom
codex/iceberg-scan-input
Closed

alexanderbianchi wants to merge 1 commit into
alex.bianchi/iceberg-selected-column-bytesfrom
codex/iceberg-scan-input

Conversation

@alexanderbianchi

Copy link
Copy Markdown
Collaborator

Eager Iceberg scans currently retain files inside a work-unit feed. Before routing, DFD converts that feed into a remote handle, so the existing routing callback cannot inspect the assigned files.

This draft explores representing eager assignments directly in the scan, while retaining feeds for lazy discovery:

pub enum ScanInput {
    AssignedFiles(AssignedFiles),
    WorkUnitFeed(Box<WorkUnitFeed<IcebergWorkUnitFeed>>),
}
Eager: discover files → choose task count → assign file groups
                                              ↓
                                   task-specific Iceberg scan
                                      ↙               ↘
                              routing inspects     worker reads

Lazy: scan definition → discover during execution → feed → worker reads

Mechanics

  • One IcebergDataSource and reader handle both inputs. AssignedFiles exposes borrowed files grouped by local partition.
  • The existing scale-up hook constructs DistributedLeafExec variants after the final task count is known. Each variant owns a concrete assignment; routing and execution no longer independently calculate it.
  • Coordinator variants share retained descriptors and their memory reservation. The codec sends only the selected task's groups and preserves their partition membership on the worker.
  • The existing RouteTaskEvent::task_specialized_plan exposes these eager groups directly. No core DFD routing changes, additional plan view, affinity hints, or placement policy are introduced.
  • Lazy feeds retain their existing execution-time discovery and round-robin delivery. Eager scans retain the current choice between round-robin and greedy assignment; this draft does not make greedy mandatory.
  • Whole-scan statistics remain on the original planning scan. Task variants report their own file/row totals and omit whole-scan column aggregates that would describe the wrong subset.

Stacked on #789. This is an alternative to #797 and #790, intended to compare explicit assigned work with exposing coordinator-local plans to routing. It does not add ordered-read strategies.

Tradeoffs and compatibility

Eager plans carry file descriptors, increasing plan payload size and moving their transport to worker setup. Greedy assignment runs synchronously in the existing scale-up hook. No performance improvement is claimed or benchmarked.

IcebergDataSource::feed() now returns Option; eager callers inspect assigned_files() or input(). Coordinators and workers must be upgraded together before using the new eager wire representation. Lazy feed plans keep their existing representation. The upgrade guide documents both changes.

Validation

  • Full Iceberg unit, integration, and documentation tests passed.
  • Routing tests cover eager round-robin/greedy, lazy feeds, isolated UNION branches, exact serialized file groups, and repeated execution before and after serialization.
  • Existing distribution tests continue to check per-task row counts across the new scan-variant layout.
  • Formatting and all-target Iceberg Clippy checks passed with warnings denied.

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