You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by "goutamadwant (via GitHub)" <gi...@apache.org> on 2026/08/08 01:24:17 UTC
[PR] [AURON #2431] Prune Iceberg changelog tasks by metadata predicates [auron]
goutamadwant opened a new pull request, #2456:
URL: https://github.com/apache/auron/pull/2456
# Which issue does this PR close?
Closes #2431
# Rationale for this change
Iceberg changelog metadata values are constant for each changelog task. Auron currently plans every supported task in the changelog range even when a predicate on `_change_type`, `_commit_snapshot_id`, or `_change_ordinal` proves that a task cannot match.
# What changes are included in this PR?
- Associate supported metadata-only filters with an unambiguous Iceberg changelog scan before adaptive query stages are created.
- Prune validated changelog tasks for `=` and `IN` predicates combined with `AND`.
- Keep `OR`, `NOT`, complex expressions, mixed data-column predicates, and ambiguous scan matches on the existing post-scan filter path.
- Preserve existing fallback validation for unsupported changelog tasks and mixed file formats.
- Add regression coverage for matching, nonmatching, conjunction, and unsupported-expression cases.
# Are there any user-facing changes?
Yes. Eligible Iceberg changelog queries plan and read fewer native scan files. There are no public API or configuration changes.
# How was this patch tested?
- `AuronIcebergIntegrationSuite`: 43 tests passed.
- Focused changelog metadata pruning regression: passed.
- `./dev/reformat --check`: passed across the configured Spark 3.0-3.5 and 4.0-4.1 profiles.
- `./build/mvn -B spotless:check test-compile -pl thirdparty/auron-iceberg -am -Ppre -Pscala-2.12 -Pspark-3.5 -Piceberg-1.10.1 -DskipBuildNative -DskipTests`: passed.
# Was this patch authored or co-authored using generative AI tooling?
- [ ] Yes
- [ ] No
Codex was used to Review and understand the codebase
<!-- Select one option. If yes, add the repository-required Generated-by line before submitting. -->
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2431] Prune Iceberg changelog tasks by metadata predicates [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2456:
URL: https://github.com/apache/auron/pull/2456#discussion_r3746764288
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -290,7 +320,10 @@ object IcebergScanSupport extends Logging {
}
val pruningPredicates = collectPruningPredicates(scan.asInstanceOf[AnyRef], readSchema)
- val nativeTasks = nativeChangelogTasks.map(task => toNativeScanTask(task, partitionSchema))
+ val filteredTasks = exec
+ .getTagValue(changelogTaskFilterTag)
Review Comment:
The tag read here goes missing whenever the scan is rebuilt for runtime filters. `withRuntimeFilters` (line 145 in this file) calls `Shims.copyBatchScanExecWithRuntimeFilters`, which builds the new node with the Scala case-class `.copy`. Spark's own doc on `TreeNode.tags` (3.5.8 `TreeNode.scala:72-74`) says tags carry over only "when this node is copied via `makeCopy`, or transformed via `transformUp`/`transformDown`", and a plain `.copy` is none of those. The new node starts with an empty tag map, so `changelogTaskFilterTag` is gone.
So on a changelog scan carrying runtime filters, which is the shape the existing `iceberg native changelog scan remains correct in dynamic pruning join` test sets up, the pruning quietly does nothing. It fails in the safe direction, but nothing logs that it happened.
Was that intentional? If not, would it make sense for `withRuntimeFilters` to carry the tag onto the new node, or at least log when it drops it?
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergConvertProvider.scala:
##########
@@ -28,6 +29,25 @@ import org.apache.auron.util.SemanticVersion
class IcebergConvertProvider extends AuronConvertProvider with Logging {
+ override def prepare(exec: SparkPlan): Unit = {
+ exec.foreach {
+ case filter: FilterExec
+ if IcebergScanSupport.isSupportedChangelogTaskFilter(filter.condition) =>
+ val referencedNames = filter.condition.references.map(_.name).toSet
Review Comment:
`filter.condition.references` is an `AttributeSet`, so it already carries each attribute's `exprId`. Mapping it to `_.name` throws that away, and the check then only asks whether some changelog scan below happens to expose columns with those names, not whether the filter's attributes actually come from that scan.
Two ways that can bite. An alias that reuses the column name passes: in `select max(_change_ordinal) as _change_ordinal from v`, the aggregate's output attribute has a different exprId from the scan's but the same name. And these three names are not reserved by Iceberg. `MetadataColumns.META_COLUMNS` (Iceberg 1.10.1, `MetadataColumns.java:110-117`) does not list them, so a user table can declare its own `_commit_snapshot_id`. A filter on that column above a full outer join, with a changelog scan on the other side, would tag and prune the changelog scan. For the other join types Spark pushes a single-side predicate below the join (`PushPredicateThroughJoin.canPushThrough`), so full outer is the one that reaches `prepare`.
Would `filter.condition.references.subsetOf(scan.outputSet)` work here?
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergConvertProvider.scala:
##########
@@ -28,6 +29,25 @@ import org.apache.auron.util.SemanticVersion
class IcebergConvertProvider extends AuronConvertProvider with Logging {
+ override def prepare(exec: SparkPlan): Unit = {
+ exec.foreach {
+ case filter: FilterExec
+ if IcebergScanSupport.isSupportedChangelogTaskFilter(filter.condition) =>
+ val referencedNames = filter.condition.references.map(_.name).toSet
+ val changelogScans = filter.child.collect {
Review Comment:
`filter.child.collect` searches the whole subtree below the filter, so any operator can sit between the filter and the scan.
That matters because dropping tasks at the scan is only safe if the filter could have been evaluated at the scan in the first place. Spark already decides that, and a filter still sitting above an operator is usually one Spark refused to push down. In Spark 3.5.8 `Optimizer.scala`, `PushPredicateThroughNonJoin.canPushThrough` has no case for `Limit`, and the rule's separate `Aggregate` case only fires when `groupingExpressions.nonEmpty`.
Here is a shape that looks like it would return a wrong answer. On a changelog view `v` with change ordinals 0, 1, 2:
```sql
select * from (select max(_change_ordinal) as _change_ordinal from v) where _change_ordinal = 0
```
There are no grouping keys, so the filter stays above the aggregate. `collect` still reaches the changelog scan and tags it, the scan then reads only the ordinal-0 task, `max(...)` returns 0, the filter passes, and the query returns `Row(0)`. Without pruning `max(...)` is 2 and the result is empty. The new test never puts a filter above a non-adjacent operator, so this would not go red today.
A filter above a `limit`, or above an unpartitioned `count(*) over ()`, looks like the same family. The window case is worth calling out separately: there is no alias involved, so the exprIds line up and tightening the name match on line 36 would not catch it.
I traced this through the optimizer source rather than running it, so I may be missing something that keeps these plans away from `prepare`.
Would it help to match only a filter that sits directly above the scan, with `Project`s allowed in between? Something like this, though happy to be redirected:
```scala
def scanUnder(p: SparkPlan): Option[BatchScanExec] = p match {
case s: BatchScanExec => Some(s)
case proj: ProjectExec => scanUnder(proj.child)
case _ => None
}
```
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala:
##########
@@ -37,6 +37,7 @@ class AuronSparkSessionExtension extends (SparkSessionExtensions => Unit) with L
logInfo(s"${classOf[AuronSparkSessionExtension].getName} enabled")
Shims.get.onApplyingExtension()
+ Shims.get.injectQueryStagePrepRule(extensions)
Review Comment:
There may already be a place to do this. `preColumnarTransitions` in this file receives the whole stage plan, and it already runs a whole-plan pass at line 86 (`AuronConvertStrategy.apply(sparkPlan)`) before converting at line 90. Tags set in a pass like that survive, because for a leaf `BatchScanExec` `withNewChildren(Nil)` returns the same object, so the converter later reads the node that was tagged.
Calling `AuronConverters.prepareExtensionPlans(sparkPlan)` just before line 86 looks like it would do the same job, with no new `Shims` method and no new `SparkSessionExtensions` injection point. It would also run per query stage, which narrows the subtree search I mentioned in `IcebergConvertProvider.prepare`, since each of those shapes has an exchange between the filter and the scan. That is a narrowing though, not a replacement for the adjacency and exprId checks.
Is there something about the pre-stage AQE hook that is needed here?
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2431] Prune Iceberg changelog tasks by metadata predicates [auron]
Posted by "slfan1989 (via GitHub)" <gi...@apache.org>.
slfan1989 commented on PR #2456:
URL: https://github.com/apache/auron/pull/2456#issuecomment-5306417505
@goutamadwant Thanks for the contribution! @weiqingy Thanks for the review!
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2431] Prune Iceberg changelog tasks by metadata predicates [auron]
Posted by "slfan1989 (via GitHub)" <gi...@apache.org>.
slfan1989 merged PR #2456:
URL: https://github.com/apache/auron/pull/2456
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org