weiqingy commented on code in PR #2379:
URL: https://github.com/apache/auron/pull/2379#discussion_r3522037320
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -382,6 +416,40 @@ object IcebergScanSupport extends Logging {
}
}
+ private def runtimeFilteredPartitions(exec: BatchScanExec):
Option[Seq[InputPartition]] = {
+ if (exec.runtimeFilters.isEmpty) {
+ return None
+ }
+
+ try {
+ MethodUtils.invokeMethod(exec, true, "prepare")
Review 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.
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -82,35 +85,54 @@ object IcebergScanSupport extends Logging {
}
}
- def plan(exec: BatchScanExec): Option[IcebergScanPlan] = {
- exec.getTagValue(scanPlanTag) match {
+ def plan(exec: BatchScanExec, useRuntimeFilters: Boolean = false):
Option[IcebergScanPlan] = {
+ val tag =
+ if (useRuntimeFilters && exec.runtimeFilters.nonEmpty) {
+ runtimeFilteredScanPlanTag
+ } else {
+ scanPlanTag
+ }
+ exec.getTagValue(tag) match {
case Some(cached) => cached
case None =>
- val planned = planUncached(exec)
- exec.setTagValue(scanPlanTag, planned)
+ val planned = planUncached(exec, useRuntimeFilters)
+ exec.setTagValue(tag, planned)
planned
}
}
- private def planUncached(exec: BatchScanExec): Option[IcebergScanPlan] = {
+ def withRuntimeFilters(
+ exec: BatchScanExec,
+ runtimeFilters: Seq[SparkExpression]): BatchScanExec = {
+ if (exec.runtimeFilters == runtimeFilters) {
Review 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.
##########
spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/auron/ShimsImpl.scala:
##########
@@ -301,6 +302,45 @@ class ShimsImpl extends Shims with Logging {
child: SparkPlan): NativeGenerateBase =
NativeGenerateExec(generator, requiredChildOutput, outer, generatorOutput,
child)
+ @sparkver("3.0 / 3.1")
+ override def copyBatchScanExecWithRuntimeFilters(
+ exec: BatchScanExec,
+ runtimeFilters: Seq[Expression]): BatchScanExec =
+ exec.copy(exec.output, exec.scan)
+
+ @sparkver("3.2")
+ override def copyBatchScanExecWithRuntimeFilters(
+ exec: BatchScanExec,
+ runtimeFilters: Seq[Expression]): BatchScanExec =
+ exec.copy(exec.output, exec.scan, runtimeFilters)
+
+ @sparkver("3.3")
+ override def copyBatchScanExecWithRuntimeFilters(
+ exec: BatchScanExec,
+ runtimeFilters: Seq[Expression]): BatchScanExec =
+ exec.copy(exec.output, exec.scan, runtimeFilters,
exec.keyGroupedPartitioning)
+
+ @sparkver("3.4")
+ override def copyBatchScanExecWithRuntimeFilters(
+ exec: BatchScanExec,
+ runtimeFilters: Seq[Expression]): BatchScanExec =
+ exec.copy(
+ exec.output,
+ exec.scan,
+ runtimeFilters,
+ exec.keyGroupedPartitioning,
+ exec.ordering,
+ exec.table,
+ exec.commonPartitionValues,
+ exec.applyPartialClustering,
+ exec.replicatePartitions)
+
+ @sparkver("3.5 / 4.0 / 4.1")
Review 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?
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeIcebergTableScanExec.scala:
##########
@@ -60,6 +63,15 @@ case class NativeIcebergTableScanExec(basedScan:
BatchScanExec, plan: IcebergSca
override val output = basedScan.output
override val outputPartitioning = basedScan.outputPartitioning
+ private lazy val plan: IcebergScanPlan = {
+ if (runtimeFilters.nonEmpty) {
+ val filteredScan = IcebergScanSupport.withRuntimeFilters(basedScan,
runtimeFilters)
+ IcebergScanSupport.plan(filteredScan, useRuntimeFilters =
true).getOrElse(staticPlan)
Review 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?
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -382,6 +416,40 @@ object IcebergScanSupport extends Logging {
}
}
+ private def runtimeFilteredPartitions(exec: BatchScanExec):
Option[Seq[InputPartition]] = {
+ if (exec.runtimeFilters.isEmpty) {
+ return None
+ }
+
+ try {
+ MethodUtils.invokeMethod(exec, true, "prepare")
+ MethodUtils.invokeMethod(exec, true, "waitForSubqueries")
+ invokeDeclaredMethod(exec, "filteredPartitions") match {
+ case Some(seq: scala.collection.Seq[_]) =>
+ Some(flattenPartitions(seq))
+ case _ =>
+ None
+ }
+ } catch {
+ case NonFatal(t) =>
Review 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.)
--
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]
