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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
