szehon-ho commented on code in PR #58145:
URL: https://github.com/apache/spark/pull/58145#discussion_r4140133789
##########
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:
The opt-in path now keeps inferred predicates only in
`DataSourceV2ScanRelation.inferredFilters`, so they are not passed through
executable filter propagation. The earlier advisory-specific `PlanMerger`
changes were removed. Constraint integration is deferred; the requested
three-input scan-merge regression under both symmetric modes is still
outstanding.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -126,8 +126,22 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
tableSpec.withNewLocation(newLoc)
}
+ private def removeNotEvaluatedFilters(
+ filters: Seq[Expression],
+ relation: DataSourceV2ScanRelation,
+ otherNotEvaluatedFilters: Seq[Expression] = Nil): Seq[Expression] = {
+ // Fast path: with nothing to exclude, avoid canonicalizing every filter
via ExpressionSet.
+ if (otherNotEvaluatedFilters.isEmpty && relation.advisoryFilters.isEmpty) {
+ filters
+ } else {
+ val notEvaluatedFilterSet =
+ ExpressionSet(otherNotEvaluatedFilters ++ relation.advisoryFilters)
+ filters.filterNot(notEvaluatedFilterSet.contains)
Review Comment:
Implemented the direct statistics path in 3088868b60c. `FilterEstimation`
now accepts a condition and base statistics. Scans opting in through
`SupportsReportStatistics.useInferredFilterEstimation()` keep inferred
predicates as scan metadata, with no inferred logical Filter or physical
expression-subtraction step in this mode.
The `A OR C` / `A OR B` factoring regression checks that inferred-only `B`
stays out of both logical filters and `FilterExec`. Exposing these predicates
through constraints remains deferred. The opt-in defaults to `false`,
preserving the ordinary-filter path for other scans.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -1261,6 +1289,37 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
}
}
+ private def getAdvisoryFilters(sHolder: ScanBuilderHolder): Seq[Expression]
= {
+ sHolder.builder match {
+ case r: SupportsPushDownCatalystFilters =>
+ val advisoryFilters = r.advisoryFilters.filter { filter =>
+ filter.deterministic &&
Review Comment:
Covered by `inferred filters exclude non-deterministic and subquery
expressions` in `DataSourceV2StrategySuite`. It reports predicates containing
`Rand` and `ScalarSubquery`, then checks that the scan's inferred metadata is
empty and neither rejected expression appears in a logical Filter.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -778,54 +849,59 @@ class PlanMerger(
relation: DataSourceV2Relation,
unionAttrs: Seq[Attribute],
strictFilters: Seq[Expression],
+ advisoryFilters: Seq[Expression],
bestEffortFilter: Option[Expression]): Option[DataSourceV2ScanRelation]
= {
val relationOut = relation.outputSet
- // Defensive: strict filters come from `pushedFilters`, which reference
only relation columns,
- // so this holds today. If a future caller offers a filter over
non-relation attributes, decline
- // the merge rather than build an unsound scan (there is no fallback above
the leaf).
- if (!strictFilters.forall(_.references.subsetOf(relationOut))) {
+ // Defensive: both filter groups reference the relation's full output
today. If a future caller
+ // offers one over non-relation attributes, decline rather than build an
unsound scan.
+ if (!(strictFilters ++
advisoryFilters).forall(_.references.subsetOf(relationOut))) {
return None
}
// `unionAttrs` are the relation's own attributes (the caller builds the
union in the relation's
- // space), so they are the projection directly.
- // strictFilters are enforced; the bestEffortFilter is offered too, but
only when it is
+ // space), so they are the projection directly. Strict and advisory
filters must both come back
+ // fully pushed by the rebuilt scan (checked below); advisory filters are
safe to offer because
+ // both inputs guarantee them. The best-effort filter is offered too, but
only when it is
// expressible over the relation -- a condition referencing propagated
boolean filter aliases
// rather than relation columns is dropped. A non-deterministic predicate
needs no handling
- // here: SPARK-58207 keeps non-deterministic filters from being pushed to
a V2 source, so an
- // offered one is simply not pushed (and dropped when the scan is
extracted).
- val conds = strictFilters ++
bestEffortFilter.filter(_.references.subsetOf(relationOut))
+ // here: SPARK-58207 keeps non-deterministic filters from being pushed to
a V2 source.
+ val conds = strictFilters ++ advisoryFilters ++
+ bestEffortFilter.filter(_.references.subsetOf(relationOut))
V2ScanRelationPushDown.rebuildScan(relation, unionAttrs, conds).filter {
scan =>
// The rebuilt scan must itself be mergeable. rebuildScan re-runs the
full pushdown, so any
// non-reproducible pushdown it introduces would make the merged scan
unsound. Today's
// Project-over-Filter input only triggers the plain path (always
mergeable), so this is
// defensive: it re-validates the rebuild's output against the same gate
applied to its
// inputs, rather than trusting the rebuild if its input plan ever
broadens.
+ val freshPushed = ExpressionSet(scan.pushedFilters)
scan.mergeableScan &&
- // Every intended-strict filter must be fully enforced by the rebuilt
scan (nothing above
- // re-checks it), and the scan must produce exactly the requested
union of columns.
- strictFilters.forall(ExpressionSet(scan.pushedFilters).contains) &&
+ // Every intended-strict and advisory filter must be fully enforced by
the rebuilt scan.
+ // Nothing above the leaf re-checks the strict filters, while full
pushdown is what allows
+ // the advisory guarantees to transfer from both input scans to the
fresh merged scan.
+ (strictFilters ++ advisoryFilters).forall(freshPushed.contains) &&
scan.outputSet == AttributeSet(unionAttrs)
}.map { scan =>
+ val confirmedScan = if (advisoryFilters.nonEmpty) {
Review Comment:
The advisory-specific reset/subtraction implementation from this version has
been removed. `DataSourceV2Strategy` no longer subtracts inferred predicates
from post-scan filters, so freshly reported inferred metadata cannot remove a
user residual. Opted-in scans consume that metadata directly for statistics.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -1261,6 +1289,37 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
}
}
+ private def getAdvisoryFilters(sHolder: ScanBuilderHolder): Seq[Expression]
= {
+ sHolder.builder match {
+ case r: SupportsPushDownCatalystFilters =>
+ val advisoryFilters = r.advisoryFilters.filter { filter =>
+ filter.deterministic &&
+ !SubqueryExpression.hasSubquery(filter) &&
+ !filter.containsPattern(EXTERNAL_UDF) &&
+ !filter.exists(_.isInstanceOf[UserDefinedExpression])
+ }
+ rebindFilters(advisoryFilters,
sHolder.output).flatMap(splitConjunctivePredicates)
+ case _ =>
+ Nil
+ }
+ }
+
+ private def rebindFilters(
+ filters: Seq[Expression],
+ output: Seq[AttributeReference]): Seq[Expression] = {
+ val outputPlan = LocalRelation(output)
+ filters.map(_.transformUp {
+ case attr: AttributeReference =>
+ outputPlan.resolveQuoted(attr.name, conf.resolver) match {
+ case Some(Alias(child, _)) => child
+ case Some(resolved) => resolved.toAttribute
+ case None =>
+ throw SparkException.internalError(
+ s"Cannot resolve Catalyst filter attribute '${attr.name}'")
Review Comment:
The current code drops unknown and ambiguous names with warnings, then
validates the rebound expression for resolution, Boolean type, and input types.
`inferred filters ignore unresolvable and ill-typed expressions` covers those
cases, including a non-Boolean expression. The additional field-ordinal and
`Unevaluable` cases from your later review are still open.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -746,25 +774,68 @@ class PlanMerger(
// np.output to cp's relation attributes is consistent with the eventual
built scan.
Some(TryMergeResult(cp.relation, npMapping,
dsv2DeferredScan = Some(DSv2DeferredScan(unionAttrs, cp.pushedFilters,
- expectedKeyGroupedPartitioning, expectedOrdering)), dsv2Merged =
true))
+ expectedAdvisoryFilters, expectedKeyGroupedPartitioning,
expectedOrdering)),
+ dsv2Merged = true))
} else {
// No enclosing Filter: build the merged scan here enforcing the (equal)
strict filters over
// the union of columns, with no best-effort filter (no post-scan Filter
to prune on).
- tryBuildMergedDSv2Scan(cp.relation, unionAttrs, cp.pushedFilters,
bestEffortFilter = None)
+ tryBuildMergedDSv2Scan(
+ cp.relation,
+ unionAttrs,
+ cp.pushedFilters,
+ expectedAdvisoryFilters,
+ bestEffortFilter = None)
.filterNot(mergeDegradesReporting(_, expectedKeyGroupedPartitioning,
expectedOrdering))
.map(TryMergeResult(_, npMapping, dsv2Merged = true))
}
}
+ /** Removes scan advisory predicates from a condition before it becomes
executable projection. */
+ private def removeAdvisoryConjuncts(
+ condition: Expression,
+ advisoryFilters: Seq[Expression]): Expression = {
+ val advisorySet = ExpressionSet(advisoryFilters)
Review Comment:
`removeAdvisoryConjuncts` was removed along with the advisory-specific
`PlanMerger` changes. This PR now has no diff in `PlanMerger`, so this
empty-advisory path no longer rewrites propagated conjunctions.
##########
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,
// The one site that grants mergeability: a plain scan carrying only
reproducible pushdowns
// (column pruning + deterministic filters) may be fused. See
hasBlockingPushdown.
mergeableScan = !hasBlockingPushdown(sHolder))
- val finalFilters = normalizedFilters.map(projectionFunc)
+ val inferredAdjustmentFilters =
Review Comment:
Implemented the minimum-estimate path in 3088868b60c. For opted-in scans,
Spark estimates the original group (fully pushed predicates plus residuals,
including those above deterministic projections) and the inferred group
independently from the same unadjusted scan statistics. It selects the smaller
row-count estimate together with that estimate's size and column statistics.
The correlated-column regression asserts 101 rows for this example, and
separate coverage checks the benefit when the original predicate is opaque to
the estimator. Inferred predicates stay on scan metadata in this mode.
This is controlled by
`SupportsReportStatistics.useInferredFilterEstimation()`, defaulting to `false`
for compatibility. The new estimation requires CBO and
`reflectsFullyPushedDownFilters() == false`; scans that do not opt in retain
the ordinary logical-filter path.
##########
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:
Agreed. The reported expression still needs a check for `GetStructField` and
`GetArrayStructFields` before rebinding, while allowing the extractors Spark
creates when resolving dotted names. The estimation refactoring does not fix
this validation gap; this 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]
