[AURON #2431] Prune Iceberg changelog tasks by metadata predicates

[goutamadwant]

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

[weiqingy]

Thanks for taking this on. A few questions inline.

[weiqingy]

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:

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 Projects allowed in between? Something like this, though happy to be redirected:

def scanUnder(p: SparkPlan): Option[BatchScanExec] = p match {
  case s: BatchScanExec => Some(s)
  case proj: ProjectExec => scanUnder(proj.child)
  case _ => None
}

[weiqingy]

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?

[weiqingy]

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?

[weiqingy]

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?