cloud-fan commented on code in PR #58145:
URL: https://github.com/apache/spark/pull/58145#discussion_r3826833700
##########
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:
**Blocking:**
Normalize advisory predicates to the same conjunct representation before
building this set. `PhysicalOperation` splits `And(a, b)` into `a` and `b`,
while the relation retains the original `And`; neither conjunct is removed, so
both reach `FilterExec` despite the never-evaluate contract. Please store or
compare flattened advisory conjuncts and cover this with a compound-filter test.
##########
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:
**Non-blocking:**
This expected result violates the advisory-filter contract. The query allows
rows 0 through 9, but advisory `i > 2` is not implied by `i > -1`; applying it
in the source drops valid rows 0 through 2. Please use an implied advisory
predicate and assert the same answer whether or not the source enforces it.
##########
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:
**Blocking:**
Propagate the advisory metadata through `buildScanWithPushedVariants` too.
That path creates `DataSourceV2ScanRelation` with the empty default and then
rebuilds its filters, so the strategy cannot recognize the advisory expression
and creates `FilterExec` for it. Since variant pushdown is enabled by default,
a builder implementing both capabilities violates the never-evaluate contract
unless this sibling path remaps the metadata as well.
##########
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:
**Blocking:**
Advisory predicates must not block independent source pushdowns. This
ordinary `Filter` is created before join, aggregate, and limit/offset pushdown,
whose matchers require an empty filter list, so opting into `advisoryFilters`
silently disables those capabilities. Please materialize the optimizer-only
filters after those pushdowns, or make every later matcher ignore the advisory
carrier, and add combination coverage.
--
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]
