[AURON #2378] Support runtime filters in native Iceberg scan#2379
[AURON #2378] Support runtime filters in native Iceberg scan#2379lyne7-sc wants to merge 11 commits into
Conversation
| def withRuntimeFilters( | ||
| exec: BatchScanExec, | ||
| runtimeFilters: Seq[SparkExpression]): BatchScanExec = { | ||
| if (exec.runtimeFilters == runtimeFilters) { |
There was a problem hiding this comment.
This guard exec.runtimeFilters == runtimeFilters looks like it's always true, so the Shims.get.copyBatchScanExecWithRuntimeFilters(...) else-branch is never taken. NativeIcebergTableScanExec is only ever constructed at IcebergConvertProvider.scala:59 as NativeIcebergTableScanExec(e, plan, e.runtimeFilters), so its runtimeFilters field is always the same object as basedScan.runtimeFilters (same e). Both call sites of withRuntimeFilters — NativeIcebergTableScanExec.scala:68 and :250 — pass withRuntimeFilters(basedScan, runtimeFilters) with runtimeFilters eq basedScan.runtimeFilters, and the node (a LeafExecNode) is never rebuilt with different filters anywhere in the tree.
Two things follow from that. First, all five @sparkver overloads of the new shim in ShimsImpl.scala (plus the abstract method in Shims.scala) are never invoked at runtime, so the new integration tests can't exercise them — a wrong version-specific copy(...) argument list would surface only as a compile error on that profile, never as a test failure. Second, doCanonicalize at NativeIcebergTableScanExec.scala:249-250 reduces to the previous basedScan.canonicalized (the wrapper returns basedScan unchanged), so the new comment there — "first make sure it sees the top-level runtime filters" — describes a transformation that doesn't currently happen.
Is this intentional groundwork for a future path that builds the node with filters different from basedScan (in which case a comment saying so, plus a test that takes the copy branch, would make the shim's ~40 version-specific lines defensible), or could withRuntimeFilters and the shim be dropped in favor of using basedScan directly? Since the field is always basedScan.runtimeFilters, I'm curious which direction you had in mind.
There was a problem hiding this comment.
Good catch. You are right that in the current built-in Spark path, NativeIcebergTableScanExec is initialized from the same BatchScanExec, so runtimeFilters normally matches basedScan.runtimeFilters and the copy branch is not exercised by this PR.
I would still prefer to keep this boundary. I do not want to bake in the assumption that the original basedScan.runtimeFilters is always the final filter sequence the native scan should use. In the future, if Auron rebuilds native scans with an updated runtime filter sequence (for example, from a native physical/AQE-stage rule), the native scan's runtimeFilters may differ from basedScan.runtimeFilters. In that case, withRuntimeFilters lets us plan runtime-filtered partitions using the native scan's filters instead of silently falling back to the original basedScan.runtimeFilters.
I added a short comment to make this intention explicit and adjusted the doCanonicalize comment so it no longer implies that the copy branch is always taken.
I do not think we can add a meaningful integration test for the copy branch in this PR because the built-in Spark path does not currently produce filters different from basedScan.runtimeFilters. Covering that branch would require an artificial/native rewrite rule that rebuilds the native scan with a different filter sequence, which feels broader than this PR?
For the version-specific copy signatures, I verified the Spark 4.1 shims profile compiles successfully.
There was a problem hiding this comment.
Makes sense — keeping the native scan's runtime filters independent from basedScan.runtimeFilters is a reasonable boundary to hold, and the added comment makes that intent explicit so the copy branch won't read as dead code to the next person. Agreed that exercising the differing-filters path would need a native rewrite rule that's broader than this PR. Thanks for checking the 4.1 shims profile compiles too.
| exec.applyPartialClustering, | ||
| exec.replicatePartitions) | ||
|
|
||
| @sparkver("3.5 / 4.0 / 4.1") |
There was a problem hiding this comment.
This groups 4.1 with 3.5/4.0 on the assumption that Spark 4.1's BatchScanExec constructor is still (output, scan, runtimeFilters, ordering, table, spjParams). The shims module compiles for 4.1 even though iceberg doesn't build there, so if 4.1 changed that constructor the 4.1 profile would fail to compile rather than fail a test. Was the 4.1 branch actually built against a 4.1 profile, or is this optimistic grouping ahead of 4.1 GA? If it hasn't been compiled against 4.1 yet, would it be safer to split 4.1 into its own branch (or drop it from the group) until the signature is confirmed?
| private lazy val plan: IcebergScanPlan = { | ||
| if (runtimeFilters.nonEmpty) { | ||
| val filteredScan = IcebergScanSupport.withRuntimeFilters(basedScan, runtimeFilters) | ||
| IcebergScanSupport.plan(filteredScan, useRuntimeFilters = true).getOrElse(staticPlan) |
There was a problem hiding this comment.
When runtime-filtered planning returns None, this falls back to staticPlan — the unfiltered plan over all partitions — with no log line. Correctness is safe here (the enclosing join re-applies the predicate, and prune-to-empty returns Some(Seq.empty) rather than None, so it doesn't hit this branch), so this is really an observability question: someone debugging "did DPP actually apply?" gets no signal that the query quietly scanned everything. Would a logWarning on the getOrElse(staticPlan) branch help — something noting that runtime-filtered planning was unavailable and all partitions are being scanned?
| None | ||
| } | ||
| } catch { | ||
| case NonFatal(t) => |
There was a problem hiding this comment.
This NonFatal catch wraps prepare / waitForSubqueries / filteredPartitions and returns None, which flows up to inputPartitions and full-scans all partitions. That also swallows a genuine DPP subquery or broadcast execution failure and turns it into a silent full-table scan, where vanilla Spark's BatchScanExec would have surfaced the failure. Partitioning errors here isn't wrong for correctness (the join re-filters), but it changes error semantics — a real failure in waitForSubqueries gets masked. Should a waitForSubqueries failure be swallowed and fall through to a full scan at all, or would you rather let it propagate so a real subquery/broadcast failure stays visible? (It does log a warning, so this is about whether full-scanning past a genuine failure is the right default, not about missing logs.)
| } | ||
|
|
||
| try { | ||
| MethodUtils.invokeMethod(exec, true, "prepare") |
There was a problem hiding this comment.
prepare() is a public SparkPlan method and exec is a typed BatchScanExec, so exec.prepare() compiles directly — only waitForSubqueries (protected) and filteredPartitions genuinely need reflective access. Calling exec.prepare() here would drop one reflection call and leave a single reflection helper (invokeDeclaredMethod) for the two methods that actually need it. Minor.
|
Thanks @weiqingy for the review. I addressed the concrete issues and kept the shim boundary intentionally. PTAL |
|
Thanks for the quick turnaround. The concrete fixes read well: letting |
There was a problem hiding this comment.
Pull request overview
Adds support for Spark runtime filters (used by dynamic partition pruning) to Auron’s native Iceberg scan path by carrying BatchScanExec.runtimeFilters into native execution and re-planning Iceberg partitions with those filters applied at runtime, plus integration coverage for common formats and changelog scans.
Changes:
- Thread runtime filters into
NativeIcebergTableScanExecand re-plan Iceberg scan tasks/partitions with runtime filters before native execution. - Cache runtime-filtered Iceberg scan plans separately from static scan plans and add a Spark-version shim for copying
BatchScanExecwith runtime filters. - Add Iceberg integration tests covering dynamic partition pruning for Parquet/ORC, empty-pruning, and changelog scans.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala | Adds integration tests and helpers to assert native Iceberg scan sees runtime filters and DPP reduces files/partitions (including empty-pruning and changelog join). |
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeIcebergTableScanExec.scala | Carries runtime filters into the native scan node, re-plans Iceberg scan with runtime-filtered partitions, and canonicalizes including runtime filters. |
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala | Adds runtime-filter-aware plan caching and reflective extraction of runtime-filtered partitions (filteredPartitions) when available. |
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergConvertProvider.scala | Passes BatchScanExec.runtimeFilters into NativeIcebergTableScanExec. |
| spark-extension/src/main/scala/org/apache/spark/sql/auron/Shims.scala | Introduces shim API to copy BatchScanExec while preserving runtime filters across Spark versions. |
| spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/auron/ShimsImpl.scala | Implements copyBatchScanExecWithRuntimeFilters for Spark 3.0–4.1 constructor signature differences. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| private def runtimeFilteredPartitions(exec: BatchScanExec): Option[Seq[InputPartition]] = { | ||
| if (exec.runtimeFilters.isEmpty) { | ||
| return None | ||
| } | ||
|
|
||
| exec.prepare() | ||
| MethodUtils.invokeMethod(exec, true, "waitForSubqueries") | ||
| invokeDeclaredMethod(exec, "filteredPartitions") match { | ||
| case Some(seq: scala.collection.Seq[_]) => | ||
| Some(flattenPartitions(seq)) | ||
| case _ => | ||
| None | ||
| } | ||
| } |
Which issue does this PR close?
Closes #2378
Rationale for this change
Spark's
BatchScanExeccan carry runtime filters for dynamic partition pruning. These filters are resolved after planning and can reduce the input partitions/files that a scan should read.Auron's native Iceberg scan should preserve and apply these runtime filters so dynamic partition pruning can reduce Iceberg scan work on the native path.
What changes are included in this PR?
BatchScanExec.runtimeFiltersintoNativeIcebergTableScanExec.BatchScanExecwith runtime filters across supported Spark versions.Are there any user-facing changes?
Queries using native Iceberg scan can read fewer partitions/files when Spark dynamic partition pruning applies. No API change.
How was this patch tested?
Added
AuronIcebergIntegrationSuitecases.