weiqingy commented on code in PR #2391:
URL: https://github.com/apache/auron/pull/2391#discussion_r3601155536
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -391,6 +425,57 @@ object IcebergScanSupport extends Logging {
private def deletesEmpty(deletes: java.util.List[_]): Boolean =
deletes == null || deletes.isEmpty
+ private def plannedInputPartitions(exec: BatchScanExec, useRuntimeFilters:
Boolean)
+ : Option[(Seq[InputPartition], Option[Seq[Seq[InputPartition]]])] = {
+ exec.outputPartitioning match {
+ case partitioning: KeyGroupedPartitioning =>
+ // Runtime filtering can change the final groups after static
planning. Keep this
+ // combination on Spark until native execution can preserve those
dynamic groups.
+ if (exec.runtimeFilters.nonEmpty) {
Review Comment:
`runtimeFilters.nonEmpty` and "runtime filtering actually happens" come
apart in one case. With
`spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly` on (the
default) and no reusable broadcast exchange to match,
`PlanDynamicPruningFilters` substitutes
`DynamicPruningExpression(Literal.TrueLiteral)` (v3.5.8,
`PlanDynamicPruningFilters.scala:78-80`) — a filter that filters nothing. The
AQE path does the same at `PlanAdaptiveDynamicPruningFilters.scala:66-67`, so
this isn't limited to non-adaptive execution. Spark treats that value as absent
in two places: `doCanonicalize` strips it explicitly
(`BatchScanExec.scala:237-239`), and `filteredPartitions` translates runtime
filters through `DataSourceV2Strategy.translateRuntimeFilterV2`, which only
matches `InSubqueryExec` (`DataSourceV2Strategy.scala:644-655`) — so
`TrueLiteral` translates to nothing, `filteredPartitions` returns `partitions`
unfiltered, and the groups can't change.
Where that lands: `select ... from fact f join dim d on f.p = d.p where d.x
= 1`, both tables Iceberg-partitioned on `p`, gets SPJ precisely *because* it's
a sort-merge join (dim too large to broadcast). No BHJ ⇒ no reusable exchange ⇒
the DPP filter degrades to `TrueLiteral` ⇒ `runtimeFilters.nonEmpty` is true ⇒
the whole scan drops back to Spark, with zero runtime filtering having been
performed.
That costs native coverage rather than correctness — on master this query
went native and returned wrong rows, so this is better either way. Is the wider
guard deliberate, or would mirroring Spark's own `doCanonicalize` predicate be
closer to the intent? One option, in case it's useful:
```scala
val hasEffectiveRuntimeFilters = exec.runtimeFilters
.exists(_ != DynamicPruningExpression(Literal.TrueLiteral))
if (hasEffectiveRuntimeFilters) None else keyGroupedInputPartitions(exec,
partitioning)…
```
(`Literal` is already imported at `:30`; only `DynamicPruningExpression`
would need adding.)
Related: would a test pinning this branch help? Nothing in the suite reaches
it today — `spark.sql.sources.v2.bucketing.enabled` appears only in the two new
tests, and every DPP test carries an explicit `/*+ BROADCAST(d) */` hint
(`:281-340`, and also `:357` / `:675`), which forces the reusable exchange and
a real `InSubqueryExec`. An SPJ query carrying runtime filters, asserting a
clean fallback (right answer, no `NativeIcebergTableScanExec`, no exception),
would pin the intended behavior — and if it turns out awkward to write because
`TrueLiteral` makes the fallback fire where you'd expect native execution,
that's the answer to the question above.
--
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]
