szehon-ho commented on code in PR #58145:
URL: https://github.com/apache/spark/pull/58145#discussion_r3834853878
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -126,7 +126,17 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
sHolder.pushedPredicates.mkString(", ")
}
- val postScanFilters = postScanFiltersWithoutSubquery ++
normalizedFiltersWithSubquery
+ sHolder.advisoryFilterExpressions = sHolder.builder match {
+ case r: SupportsPushDownCatalystFilters =>
+ val advisoryFilters = r.advisoryFilters.filter { filter =>
+ filter.deterministic && !SubqueryExpression.hasSubquery(filter)
+ }
+ rebindFilters(advisoryFilters, sHolder.output)
+ case _ =>
+ Nil
+ }
+ val postScanFilters = postScanFiltersWithoutSubquery ++
+ sHolder.advisoryFilterExpressions ++ normalizedFiltersWithSubquery
Review Comment:
Thanks. Advisory filters are now captured on the holder in `pushDownFilters`
but kept off the logical `Filter` until `pruneColumns`, so later matchers still
see `PhysicalOperation(..., Nil, holder)`.
They are remapped onto table-shaped scans (plain, sample, limit, offset,
top-N). Join, aggregate, and variant replace the scan output, so the advisory
metadata is dropped there rather than rewritten — same as SPARK-58315 only
re-attaching post-pushdown adjustment filters in `pruneColumns`.
Added combination tests for those operators in `DataSourceV2StrategySuite`,
plus E2E coverage for join, limit, and TABLESAMPLE.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -201,8 +201,12 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
val batchExec = BatchScanExec(relation.output, relation.scan,
runtimeFilters,
relation.ordering, relation.relation.table,
relation.keyGroupedPartitioning)
+ // Advisory filters are kept in the logical Filter for the optimizer
only, and Spark never
+ // evaluates them. See SupportsPushDownCatalystFilters.advisoryFilters.
+ val notEvaluatedFilterSet = ExpressionSet(
+ fullyPushedRuntimeFilters ++ relation.advisoryFilters)
Review Comment:
Thanks, this was a real bug. `getAdvisoryFilters` now flattens with
`splitConjunctivePredicates` before storing, so the advisory set matches the
conjuncts `PhysicalOperation` produces. Covered by a compound-filter test in
`DataSourceV2Suite`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -1019,6 +1035,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
// merge unsound. See DataSourceV2ScanRelation.pushedFilters.
val scanRelation = DataSourceV2ScanRelation(sHolder.relation,
wrappedScan, output,
pushedFilters = sHolder.pushedFilterExpressions,
+ advisoryFilters = remappedAdvisoryFilters,
Review Comment:
Thanks. We do not remap advisory filters through
`buildScanWithPushedVariants` (or the join / aggregate builders). Those paths
replace the scan output, so the original column references would not match, and
Spark would still have to drop them from `FilterExec`.
Variant pushdown itself is unblocked because the advisory `Filter` is no
longer on the plan while that matcher runs. The advisory metadata is discarded.
Covered by "advisory filters do not block variant pushdown".
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala:
##########
@@ -1656,6 +1665,31 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
"non-deterministic filter should be retained as a post-scan Filter")
}
+ test("advisory filters stay in the logical Filter but are not evaluated by
FilterExec") {
+ val advisoryFilter = "i > 2"
+ val df = spark.read
+ .format(classOf[CatalystFilterDataSourceV2].getName)
+ .option("advisoryFilter", advisoryFilter)
+ .load()
+ val query = df.filter($"i" > -1)
+
+ checkAnswer(query, (3 until 10).map(i => Row(i, -i)))
Review Comment:
Thanks. The tests now derive an implied predicate on `j` from the pushed
filter on `i` (rows are `(i, j)` with `j = -i`, so `i > 2` implies `j < -2`).
`checkAnswer` matches the user filter; Spark never evaluates the advisory
predicate. Join and limit E2E tests use the same derivation.
--
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]
