[EPIC] Adaptive runtime pruning for native Iceberg scans #6641

What is the problem the feature request solves?

Connect Comet's runtime predicate producers to native Iceberg scans so predicates discovered during execution can reduce storage reads.

Initial producers:

• Hash join build-side bounds
• MIN/MAX aggregate bounds
• TopK thresholds

The underlying adaptive reader support is implemented in iceberg-rust. This epic tracks Comet integration, dependency alignment, producer attachment, Spark lifecycle correctness, observability, and validation.

Describe the potential solution

Related work

• Exploration: Explore runtime Iceberg pruning from join, MIN/MAX and TopK #6588
• Existing runtime filtering work:
  • feat: IO-level dynamic filter pushdown into IcebergScanExec (iceberg-rust predicate conversion) #4808
  • feat: native dynamic filter pushdown for hash join into Parquet scans #5699

Initial supported scope

• INNER hash join with one direct signed-integer key
• Ungrouped single-partition integer MIN/MAX
• TopK with one direct signed-integer ordering key

Unsupported shapes retain the existing execution path.

Environment

• AMD EPYC, 4 vCPUs, 15 GiB memory
• Spark 3.5.9
• Iceberg 1.8.1
• Java 17
• 16M fact rows
• 16 data files per layout
• 18 warm runs per query/variant across 6 JVM rounds
• Variant order rotated between rounds
• All variants returned identical results

Execution time

Reader bytes

The filters-OFF variant read the same bytes as main for every query.

Additional pruning evidence:

• Sorted join: 18 file tasks rejected before open and 4 additional row groups pruned
• Sorted TopK: 15 file tasks rejected, 12 row groups pruned live, 4 predicate refreshes
• Unsorted TopK: 18 row groups pruned live, 47 predicate refreshes
• Unsorted MIN: 4 file tasks rejected before open and 20 row groups pruned live

These are warm local-disk measurements. Cold-cache and object-storage validation remain follow-up work.

Work items

Dependencies

• Upgrade Comet to an Arrow 60-compatible DataFusion release
• Complete Iceberg-rust side work [EPIC] Execution-time adaptive pruning for Iceberg readers iceberg-rust#3343
• Pin compatible upstream iceberg-rust revisions

Iceberg integration

• Add the conservative Iceberg predicate adapter
• Attach generic runtime predicate providers to IcebergScanExec
• Connect completed hash-join bounds
• Transport retained file statistics to native execution
• Collect statistics only for eligible Iceberg scans
• Connect MIN/MAX runtime producers
• Connect TopK runtime producers

Productionization

• Complete Spark metrics, configuration, and documentation
• Complete end-to-end correctness and lifecycle coverage
• Preserve normal-Parquet runtime-filter behavior
• Keep native scans for data tables named like metadata tables — Iceberg data tables whose names end with a metadata-table name fall back to Spark #6719

Additional context

Metrics

The final integration should expose:

• iceberg_runtime_predicate_tasks
• iceberg_runtime_file_tasks_pruned
• iceberg_runtime_row_groups_pruned
• iceberg_runtime_row_groups_pruned_live
• iceberg_runtime_predicate_refreshes
• bytes_scanned

Correctness requirements

• Runtime predicates are conservative restrictions.
• Exact query semantics remain enforced by existing operators.
• Runtime state is fresh for every execution.
• Unsupported predicates fall back safely.
• Missing statistics never cause unsafe pruning.
• Position and equality delete semantics remain unchanged.
• Normal-Parquet execution remains regression-tested.

Definition of done

• Comet uses compatible upstream DataFusion, Arrow/Parquet, and iceberg-rust dependencies.
• Eligible join, MIN/MAX, and TopK producers reach native Iceberg scans.
• No-delete, position-delete, equality-delete, and combined-delete cases are covered.
• Runtime metrics are visible through Spark SQL metrics.
• Matched benchmarks are reproducible.
• Nonselective overhead and object-storage behavior are measured.
• Supported shapes, configuration, and fallback behavior are documented.
• Each work item is linked to its corresponding issue or PR.

Follow-up scope

Future work can expand support to:

• Additional data types
• Multi-column joins
• Projection traversal
• Grouped/distributed aggregates
• More complex TopK shapes
• Small exact IN-set predicates

Selective copy-on-write rewrites remain separate work.