unikdahal opened a new issue, #6641: URL: https://github.com/apache/datafusion-comet/issues/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: https://github.com/apache/datafusion-comet/issues/6588 - Existing runtime filtering work: - https://github.com/apache/datafusion-comet/issues/4808 - https://github.com/apache/datafusion-comet/issues/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 | Query / layout | Main | Filters OFF | Filters ON | Speedup vs main | | --- | ---: | ---: | ---: | ---: | | Join, sorted | 956 ms | 948 ms | 107 ms | **8.92×** | | Join, position deletes | 988 ms | 973 ms | 113 ms | **8.75×** | | Join, unsorted | 955 ms | 948 ms | 342 ms | **2.79×** | | TopK, sorted | 991 ms | 985 ms | 142 ms | **6.97×** | | TopK, position deletes | 1,029 ms | 1,016 ms | 163 ms | **6.30×** | | TopK, unsorted | 994 ms | 995 ms | 296 ms | **3.36×** | | MIN, sorted | 99 ms | 99 ms | 66 ms | **1.51×** | | MIN, unsorted | 110 ms | 114 ms | 82 ms | **1.34×** | | No runtime producer | 906 ms | 917 ms | 917 ms | **0.99×** | ### Reader bytes | Query / layout | Main | Filters ON | Reduction | | --- | ---: | ---: | ---: | | Join, sorted | 605.8 MiB | 1.7 MiB | **99.7%** | | Join, position deletes | 607.0 MiB | 1.8 MiB | **99.7%** | | Join, unsorted | 659.8 MiB | 253.7 MiB | **61.5%** | | TopK, sorted | 605.8 MiB | 35.4 MiB | **94.2%** | | TopK, position deletes | 607.0 MiB | 39.4 MiB | **93.5%** | | TopK, unsorted | 659.8 MiB | 199.4 MiB | **69.8%** | | MIN, sorted | 50.9 MiB | 4.7 MiB | **90.7%** | | MIN, unsorted | 60.2 MiB | 13.1 MiB | **78.3%** | | No runtime producer | 564.9 MiB | 564.9 MiB | **0%** | 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 apache/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 ### 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
