Re: [PR] [SPARK-58892][SQL] Allow DSv2 sources to report advisory Catalyst filters [spark]

["cloud-fan (via GitHub)" <gi...@apache.org>]

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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org