peter-toth commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3711888964
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -213,6 +217,22 @@ object PushDownUtils extends Logging {
}
translatedFiltersPushed || partPredicatesPushed
+
+ case catalystScan: SupportsRuntimeCatalystFiltering if
runtimeFilters.nonEmpty =>
+ // A DPP filter degrades to TrueLiteral when its subquery is pruned
away; it carries no
+ // information for the source. The V2 path above drops these
implicitly because
+ // translateRuntimeFilterV2 returns None; here we push Catalyst
expressions directly,
+ // so filter them out explicitly.
+ val catalystFilters = runtimeFilters
+ .flatMap(unwrapRuntimeFilterExpression)
+ .filterNot(_ == Literal.TrueLiteral)
Review Comment:
Follow-up on this: the non-determinism gap isn't specific to your branch, so
I've fixed it at the root under the same ticket rather than asking you to carry
it here — apache/spark#57760 ([SPARK-58207][SQL][FOLLOWUP]).
The gate goes in `DataSourceV2Strategy` rather than in `pushRuntimeFilters`:
`scalarSubqueryFilters` now also requires `f.deterministic`, so a
non-deterministic filter never enters `runtimeFilters` in the first place —
which also keeps it out of `EXPLAIN`'s `RuntimeFilters: [...]`, out of
`BatchScanExec.equals`/`doCanonicalize`, and skips a pointless `filter()` +
`planInputPartitions()` round. On master that stops `(part = 3) OR (RAND() <
0.5)` reaching a `SupportsRuntimeV2Filtering` source; on your branch the same
predicate then never reaches the new Catalyst branch either, and — since it
stays in `postScanFilters` — `fullyPushedRuntimeFilters` can no longer drop the
only evaluator of a non-deterministic predicate. So once that lands and you
rebase, I'd consider this finding resolved and the
`.filter(isPushablePartitionFilter)` I suggested above optional.
For completeness on the other half of `runtimeFilters`: DPP filters bypass
that gate, but they can't carry non-determinism here either. A
`DynamicPruningSubquery` whose filtering plan is non-deterministic makes the
whole expression non-deterministic (`PlanExpression.deterministic` folds in
`plan.deterministic`), so the `Filter` above the scan fails
`NodeWithOnlyDeterministicProjectAndFilter` and `CleanupDynamicPruningFilters`
rewrites the `DynamicPruning` to `TrueLiteral` before planning. Adding the
screen on your side would be harmless, just redundant.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
