szehon-ho commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3731868031
##########
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:
One refinement in the latest revision: the screen on the push side now runs
before the DPP unwrap, with `includeSubquery = true`, so the drop decision and
the push decision apply the same predicate to the same form of the filter
rather than to two forms that happen to agree.
--
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]
