szehon-ho commented on code in PR #58145:
URL: https://github.com/apache/spark/pull/58145#discussion_r4140134786
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -1261,6 +1292,78 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
}
}
+ private def shouldAddSparkPostPushdownAdjustmentFilters(scan: Scan): Boolean
= scan match {
+ case s: SupportsReportStatistics => !s.reflectsFullyPushedDownFilters()
+ case _ => false
+ }
+
+ private def getInferredFilters(sHolder: ScanBuilderHolder): Seq[Expression]
= {
+ sHolder.builder match {
+ case r: SupportsPushDownCatalystFilters =>
+ val validFilters = r.inferredFilters
+ .flatMap(rebindInferredFilter(_, sHolder.output))
+ .filter(validateInferredFilter(_, sHolder.output))
+ .flatMap(splitConjunctivePredicates)
+ ExpressionSet(validFilters).toSeq
+ case _ =>
+ Nil
+ }
+ }
+
+ private def validateInferredFilter(
+ filter: Expression,
+ output: Seq[AttributeReference]): Boolean = {
+ val hasInvalidColumnReference = filter.exists {
+ case _: AttributeReference => false
+ case _: Attribute | _: BoundReference | _: LateralColumnAliasReference |
+ _: OuterReference | _: OuterScopeReference => true
+ case _ => false
+ }
+ val hasValidStructure = filter.deterministic &&
+ !SubqueryExpression.hasSubquery(filter) &&
+ !filter.containsPattern(EXTERNAL_UDF) &&
+ !filter.exists(_.isInstanceOf[UserDefinedExpression]) &&
Review Comment:
This validation gap remains open. One detail to account for in the fix:
`AttributeReference` itself extends `Unevaluable`, so a blanket
`filter.exists(_.isInstanceOf[Unevaluable])` check would also reject valid
column predicates. The guard needs to allow those references while rejecting
unsupported unevaluable expressions, with a regression for the `TimeWindow`
case.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -136,6 +141,18 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
sHolder.pushedFilterExpressions = normalizedFiltersWithoutSubquery
.filterNot(postScanFilterSet.contains)
.filter(_.deterministic)
+ val fullyPushedFilterSet = ExpressionSet(sHolder.pushedFilterExpressions)
+ // Collect inferred filters only after an eligible Catalyst callback;
V1/V2 interfaces take
+ // precedence when a builder implements multiple filter APIs.
+ val catalystFiltersPushed =
normalizedFiltersWithoutSubquery.exists(_.deterministic) &&
+ pushedFilters.isRight &&
+ sHolder.builder.isInstanceOf[SupportsPushDownCatalystFilters] &&
+ !sHolder.builder.isInstanceOf[SupportsPushDownV2Filters]
+ sHolder.inferredFilterExpressions = if (catalystFiltersPushed) {
+ getInferredFilters(sHolder).filterNot(fullyPushedFilterSet.contains)
Review Comment:
The requested fixture that reports a fully pushed predicate again as
inferred is still missing. The opt-in/default-path tests cover estimate
composition and filter placement, but they do not exercise this deduplication
branch. Keeping this coverage request open.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -1019,11 +1049,18 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
// merge unsound. See DataSourceV2ScanRelation.pushedFilters.
val scanRelation = DataSourceV2ScanRelation(sHolder.relation,
wrappedScan, output,
pushedFilters = sHolder.pushedFilterExpressions,
+ inferredFilters = remappedInferredFilters,
Review Comment:
Agreed that this still needs dedicated scan-merge coverage. Opted-in scans
now avoid the extra logical Filter, while scans that do not opt in retain that
path. The focused pushdown tests do not cover the merge/rebuild interaction
described here, so this remains open.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsPushDownCatalystFilters.scala:
##########
@@ -34,6 +34,30 @@ trait SupportsPushDownCatalystFilters extends ScanBuilder {
*/
def pushFilters(filters: Seq[Expression]): Seq[Expression]
+ /**
+ * Returns additional filters inferred from eligible query filters passed to
[[pushFilters]].
+ * Each inferred filter must be implied by those query filters and satisfied
by every row
+ * returned by the scan.
Review Comment:
Agreed: the guarantee must mean SQL `true` for every returned row, including
the default mode that evaluates inferred predicates. The current wording has
not yet been tightened to explicitly exclude `null`, so this documentation item
remains open.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsPushDownCatalystFilters.scala:
##########
@@ -34,6 +34,30 @@ trait SupportsPushDownCatalystFilters extends ScanBuilder {
*/
def pushFilters(filters: Seq[Expression]): Seq[Expression]
+ /**
+ * Returns additional filters inferred from eligible query filters passed to
[[pushFilters]].
Review Comment:
The gate requires a deterministic, subquery-free query predicate to have
been dispatched through the Catalyst filter API. Builders handled through
`SupportsPushDownFilters` or `SupportsPushDownV2Filters` do not have inferred
filters collected through this path. Those behaviors are tested, but the API
documentation still needs to spell them out; this wording change remains open.
--
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]
