Uh oh!
There was an error while loading. Please reload this page.
Spark 4.2: Push down opaque partition predicates - #17956
Conversation
4bc76c4 to
29f2660CompareUse Spark 4.2 PartitionPredicate to filter scan tasks by partition values while retaining predicates for Spark post-scan evaluation when specs cannot evaluate them. Generated-by: Codex (GPT-5)
29f2660 to
e685795Compare| this.filters = expressions; | ||
| this.pushedPredicates = pushablePredicates.toArray(new Predicate[0]); | ||
| this.filters.addAll(expressions); |
There was a problem hiding this comment.
Accumulating is correct here, SupportsPushDownV2Filters requires it once supportsIterativePushdown() returns true. But it changes what TestFilteredScan.testUnpartitionedIDFilters exercises: that test builds the SparkScanBuilder once at line 204, outside its loop, so iteration i now plans id = 0 AND id = 1 AND … AND id = i instead of just id = i.
It stays green because Iceberg checks each conjunct separately against the file bounds [0, 9] and never sees that the conjunction is unsatisfiable, and the row-level assertion below it goes through a separate read(...) that doesn't touch the builder. So no coverage is lost outright, but the hasSize(1) assertion is no longer checking the filter it appears to push, and it now quietly depends on Iceberg not detecting contradictions.
Could you move the builder construction inside the loop? That matches testUnpartitionedCaseInsensitiveIDFilters just below it:
for (int i = 0; i < 10; i += 1) {
SparkScanBuilder builder =
new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
pushFilters(builder, EqualTo.apply("id", i));
| try (CloseableIterable<? extends ScanTask> taskIterable = scan.planFiles()) { | ||
| List<T> plannedTasks = Lists.newArrayList(); | ||
| Map<Integer, PartitionPredicateEvaluator> evaluatorsBySpecId = Maps.newHashMap(); | ||
| int numPlannedTasks = 0; |
There was a problem hiding this comment.
nit: numPlannedTasks counts tasks before pruning, but plannedTasks holds the ones that survived. Maybe rename it to numScannedTasks or numCandidateTasks?
| LOG.debug( | ||
| "Planned {} task group(s) with {} grouping key type and {} unique grouping key(s) for table {}", | ||
| "Planned {} task group(s) with {} grouping key type and {} unique grouping key(s) for" | ||
| + " table {}", |
There was a problem hiding this comment.
unnecessary change? There are a few other places.
Description
This adds support for Spark 4.2's new opaque
PartitionPredicateAPI on top of #14984.Iceberg now:
All partition predicates are currently returned to Spark for post-scan evaluation. This provides a correctness backstop for partition evolution. A future optimization can return only predicates that cannot be evaluated against every scanned spec.
The tests cover direct predicate evaluation, iterative pushdown, partition evolution, unsupported V2 filters, UDFs, non-first and multiple partition fields, and nested identity partition fields.
Testing
TestFilteredScan: 85 tests, 9 skipped, 0 failuresTestFilterPushDown: 38 tests, 0 failuresendsWithpartition predicate and partition-evolution tests passAI Disclosure