szehon-ho commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3723876473
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -213,6 +217,26 @@ 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.
+ // Screen with the same pushability guard as the V2 PartitionPredicate
path
+ // (deterministic, no subquery, no Python UDF). Keeps
non-deterministic filters
+ // from being the sole evaluator when fullyPushedFilterAttributes
drops FilterExec.
+ val catalystFilters = runtimeFilters
+ .flatMap(unwrapRuntimeFilterExpression)
+ .filterNot(_ == Literal.TrueLiteral)
+ .filter(isPushablePartitionFilter)
Review Comment:
This is already fixed on the current revision.
The analysis: `part = (SELECT max(val) FROM dim) OR rand() < 0.5` did reach
`fullyPushedRuntimeFilters`, since that set was computed from `f.references`
alone, so it was dropped from `postScanFilters` and then rejected by this
guard, leaving nothing to evaluate it. The rebase onto master picks up the
determinism screen from #57760: `scalarSubqueryFilters` now requires
`f.deterministic`, and `fullyPushedRuntimeFilters` is a subset of it, so a
non-deterministic filter never becomes a fully pushed candidate. It is not
routed into `runtimeFilters` and keeps its post-scan `FilterExec`. I added a
test for that exact query against a table declaring `part` fully pushed, in
`DataSourceV2CatalystRuntimeFilterSuite`: "non-deterministic predicate on fully
pushed attributes -> evaluated after the scan".
I applied the suggestion anyway as strengthening, since two independent
decisions are what let these drift apart in the first place.
`DataSourceV2Strategy` now runs the same `isPushablePartitionFilter` before
dropping the post-scan filter, with an `includeSubquery` flag because at
planning time a runtime filter still holds its scalar subquery, which
`unwrapRuntimeFilterExpression` literalizes on the way to the scan. That gate
is load-bearing on its own: with the determinism screen removed locally, the
filter still keeps its post-scan `FilterExec`.
--
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]
