szehon-ho commented on code in PR #58145:
URL: https://github.com/apache/spark/pull/58145#discussion_r4151021756
##########
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 |
Review Comment:
Fixed in
[fc995075e62](https://github.com/apache/spark/commit/fc995075e62b76fa7073dd753673344d4d95b193).
Source-reported `GetStructField` and `GetArrayStructFields` are rejected
before name rebinding, so their ordinals cannot be applied to the relation
schema. The regression uses a reversed source struct layout and covers both
scan opt-in settings. The existing dotted-name and quoted-name cases still
cover field accessors created by Spark during resolution.
##########
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:
Fixed in
[fc995075e62](https://github.com/apache/spark/commit/fc995075e62b76fa7073dd753673344d4d95b193)
with the corrected guard: `AttributeReference` is allowed, while other
`Unevaluable` expressions and `RuntimeReplaceable` expressions are rejected.
Added regressions for `TimeWindow`, `SortOrder`, `Nvl`, and `Between` under
both opt-in settings. `TimeWindow` already reports itself as unresolved on this
head; `SortOrder` exercises a resolved-but-unevaluable expression, and
`Nvl`/`Between` exercise the missed replacement step. Existing valid
column-predicate tests cover the `AttributeReference` exception.
##########
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:
Added the requested fixture in
[fc995075e62](https://github.com/apache/spark/commit/fc995075e62b76fa7073dd753673344d4d95b193):
it returns the fully pushed predicates again as inferred predicates. Both
opt-in settings verify empty inferred metadata, correct query results, and
exactly one copy of the pushed comparison in both the logical and physical
filters.
##########
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:
Added `MergeSubplansSuite` coverage in
[fc995075e62](https://github.com/apache/spark/commit/fc995075e62b76fa7073dd753673344d4d95b193)
using a Catalyst-filter source with `SCAN_MERGING`. It covers rebuilding three
scans with inferred predicates for both default and opted-in scans, checks the
rebound pushed/inferred metadata and output references, and covers three
distinct residual filters under the general and DSv2-specific symmetric modes.
With both symmetric modes disabled, the scans remain separate.
##########
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:
Updated in
[fc995075e62](https://github.com/apache/spark/commit/fc995075e62b76fa7073dd753673344d4d95b193).
The contract now explicitly requires SQL `true` for every returned row and
excludes both `false` and `null`.
##########
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:
Updated in
[fc995075e62](https://github.com/apache/spark/commit/fc995075e62b76fa7073dd753673344d4d95b193).
The documentation now states that Spark collects inferred filters only after
passing at least one deterministic, subquery-free predicate through the
Catalyst callback. It also documents that `SupportsPushDownFilters` and
`SupportsPushDownV2Filters` take precedence when a builder implements multiple
filter APIs.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -361,8 +419,10 @@ case class DataSourceV2ScanRelation(
.map(o => o.copy(child = QueryPlan.normalizeExpressions(o.child,
output)))
),
// pushedFilters may reference columns pruned out of `output` (see the
field doc), so they are
- // normalized against the relation's full output rather than `output`.
- pushedFilters = pushedFilters.map(QueryPlan.normalizeExpressions(_,
relation.output))
+ // normalized against the relation's full output rather than `output`.
The same holds for
+ // inferredFilters.
Review Comment:
Corrected in
[fc995075e62](https://github.com/apache/spark/commit/fc995075e62b76fa7073dd753673344d4d95b193):
only `pushedFilters` may reference pruned columns. The comment now states that
`inferredFilters` are restricted to surviving output columns and can safely use
the same normalization.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -500,31 +509,43 @@ class PlanMerger(
// Note: the new Project always uses mergedChild as its
child (rather than
// flattening into an existing Project below) because
mergedChild.output may
// contain previously-propagated filter attributes that
cp.condition references.
+ val advisoryFilters =
deferred.toSeq.flatMap(_.advisoryFilters)
+ val residualNPCondition =
+ removeAdvisoryConjuncts(mappedNPCondition, advisoryFilters)
+ val residualCPCondition =
+ removeAdvisoryConjuncts(cp.condition, advisoryFilters)
val newNPCondition =
- npFilter.fold(mappedNPCondition) { case (f, _) => And(f,
mappedNPCondition) }
- val newCPCondition = cpFilter.fold(cp.condition)(And(_,
cp.condition))
+ npFilter.fold(residualNPCondition) {
+ case (f, _) => And(f, residualNPCondition)
+ }
+ val newCPCondition = cpFilter.fold(residualCPCondition) {
+ And(_, residualCPCondition)
+ }
// The OR-widen moves both conditions into a boolean Project
above the merged
// scan, so the scan itself would read the full table. Build
the deferred scan
// here with OR(np condition, cp condition) as the
best-effort filter (the Filter
// above still enforces exactness). The best-effort filter
is derived from the
// scan-level conditions, not the propagated filter
attributes in newNP/newCP.
tryBuildFilterDSv2ScanChild(
mergedChild, deferred, Some(Or(mappedNPCondition,
cp.condition)))
- .map { prunedChild =>
- val newNPFilterAlias =
- Alias(newNPCondition,
s"propagatedFilter_${PlanMerger.newId}")()
- val newCPFilterAlias =
- Alias(newCPCondition,
s"propagatedFilter_${PlanMerger.newId}")()
- val newNPFilter = newNPFilterAlias.toAttribute
- val newCPFilter = newCPFilterAlias.toAttribute
- val project = Project(
- prunedChild.output.toList ++ Seq(newNPFilterAlias,
newCPFilterAlias),
- prunedChild)
- val newFilter = Filter(Or(newNPFilter, newCPFilter),
project)
- newFilter.copyTagsFrom(cp)
- newFilter.setTagValue(PlanMerger.MERGED_FILTER_TAG, ())
- TryMergeResult(newFilter, npMapping, Some((newNPFilter,
true)),
- Some(newCPFilter), dsv2Merged = dsv2Merged)
+ .flatMap { prunedChild =>
+ addAdvisoryFilter(prunedChild, advisoryFilters).map {
filteredChild =>
Review Comment:
Added the three-input regression in
[fc995075e62](https://github.com/apache/spark/commit/fc995075e62b76fa7073dd753673344d4d95b193)
under both general and DSv2-specific symmetric propagation, covering both scan
opt-in settings. For opted-in scans it verifies that the inferred-only column
stays out of propagated executable aliases. When
`reflectsFullyPushedDownFilters()` is `false`, the default mode retains
ordinary executable `Filter` nodes and their standard constraints. Exposing
scan-metadata predicates directly as constraints remains deferred.
--
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]
