kosiew commented on code in PR #26281:
URL: https://github.com/apache/datafusion/pull/26281#discussion_r4491497850
##########
datafusion/optimizer/src/push_down_filter.rs:
##########
@@ -1593,6 +1521,194 @@ fn rewrite_projection(
Ok((projection, keep_predicates))
}
+/// Pushes the conjuncts of `filter` into the scan that is its input, as far as
+/// the source of the scan accepts them, and keeps the rest in a filter above
+/// the scan. `changed` tells whether the caller already changed the filter, so
+/// that a result without a pushdown still reports the change.
+fn push_down_into_table_scan(
+ mut filter: Filter,
+ changed: bool,
+) -> Result<Transformed<LogicalPlan>> {
+ let LogicalPlan::TableScan(mut scan) = Arc::unwrap_or_clone(filter.input)
else {
+ return internal_err!("the input of the filter is not a table scan");
+ };
+ // The fetch of a scan applies after its filters, so a filter above a scan
+ // with a fetch cannot move into it without changing the result.
+ if scan.fetch.is_some() {
Review Comment:
Please guard positive `scan.skip` values as well as `scan.fetch`. For
`b=[0,10,20]`, `Filter(b > 5) -> Projection -> Scan(skip=1, fetch=None)` should
return `[10,20]`, but pushing even an inexact predicate before the skip can
change the result to `[20]`. Add an offset-only regression using a source
supporting exact filter and skip pushdown, and check the direct-scan path too.
The direct-scan gap predates this PR, but the extraction-projection path newly
exposes it.
##########
datafusion/optimizer/src/push_down_filter.rs:
##########
@@ -1593,6 +1521,194 @@ fn rewrite_projection(
Ok((projection, keep_predicates))
}
+/// Pushes the conjuncts of `filter` into the scan that is its input, as far as
+/// the source of the scan accepts them, and keeps the rest in a filter above
+/// the scan. `changed` tells whether the caller already changed the filter, so
+/// that a result without a pushdown still reports the change.
+fn push_down_into_table_scan(
+ mut filter: Filter,
+ changed: bool,
+) -> Result<Transformed<LogicalPlan>> {
+ let LogicalPlan::TableScan(mut scan) = Arc::unwrap_or_clone(filter.input)
else {
+ return internal_err!("the input of the filter is not a table scan");
+ };
+ // The fetch of a scan applies after its filters, so a filter above a scan
+ // with a fetch cannot move into it without changing the result.
+ if scan.fetch.is_some() {
+ filter.input = Arc::new(LogicalPlan::TableScan(scan));
+ return Ok(Transformed::new(
+ LogicalPlan::Filter(filter),
+ changed,
+ TreeNodeRecursion::Continue,
+ ));
+ }
+ let filter_predicates = split_conjunction(&filter.predicate);
+ // Filters containing scalar subqueries cannot be pushed to
+ // providers because the subquery result is not available
+ // until execution time.
+ let (subquery_filters, pushdown_candidates): (Vec<&Expr>, Vec<&Expr>) =
+ filter_predicates
+ .into_iter()
+ .partition(|pred| pred.contains_scalar_subquery());
+
+ let (volatile_filters, non_volatile_filters): (Vec<&Expr>, Vec<&Expr>) =
+ pushdown_candidates
+ .into_iter()
+ .partition(|pred| pred.is_volatile());
+
+ // Check which non-volatile filters are supported by source
+ let supported_filters = scan
+ .source
+ .supports_filters_pushdown(non_volatile_filters.as_slice())?;
+ assert_eq_or_internal_err!(
+ non_volatile_filters.len(),
+ supported_filters.len(),
+ "Vec returned length: {} from supports_filters_pushdown is not the
same size as the filters passed, which length is: {}",
+ supported_filters.len(),
+ non_volatile_filters.len()
+ );
+
+ if supported_filters
+ .iter()
+ .all(|res| res == &TableProviderFilterPushDown::Unsupported)
+ {
+ filter.input = Arc::new(LogicalPlan::TableScan(scan));
+ return Ok(Transformed::new(
+ LogicalPlan::Filter(filter),
+ changed,
+ TreeNodeRecursion::Continue,
+ ));
+ }
+
+ // Compose scan filters from non-volatile filters of `Exact` or `Inexact`
pushdown type
+ let zip = non_volatile_filters.iter().zip(supported_filters.iter());
+
+ let new_scan_filters = zip
+ .clone()
+ .filter(|(_, res)| *res != &TableProviderFilterPushDown::Unsupported)
+ .map(|(&pred, _)| pred);
+
+ // Add new scan filters
+ let new_scan_filters: Vec<Expr> = scan
+ .filters
+ .iter()
+ .chain(new_scan_filters)
+ .unique()
+ .cloned()
+ .collect();
+
+ // Nothing changes when no predicate leaves the filter and the scan already
+ // records every accepted predicate.
+ if supported_filters
+ .iter()
+ .all(|res| res != &TableProviderFilterPushDown::Exact)
+ && scan.filters == new_scan_filters
+ {
+ filter.input = Arc::new(LogicalPlan::TableScan(scan));
+ return Ok(Transformed::new(
+ LogicalPlan::Filter(filter),
+ changed,
+ TreeNodeRecursion::Continue,
+ ));
+ } else {
+ scan.filters = new_scan_filters;
+ }
+
+ // Compose predicates to be of `Unsupported` or `Inexact` pushdown type,
+ // and also include volatile and subquery-containing filters
+ let new_predicate: Vec<Expr> = zip
+ .filter(|(_, res)| *res != &TableProviderFilterPushDown::Exact)
+ .map(|(&pred, _)| pred)
+ .chain(volatile_filters)
+ .chain(subquery_filters)
+ .cloned()
+ .collect();
+
+ Ok(Transformed::yes(with_filters(
+ new_predicate,
+ LogicalPlan::TableScan(scan),
+ )))
+}
+
+/// Offers the predicates that read only pass-through columns of a pure
+/// extraction projection to the scan directly below it.
+///
+/// The projection and the filter keep their order, so
`PushDownLeafProjections`
+/// has nothing to undo. The scan records each predicate its source accepts. A
+/// predicate the source applies exactly is removed from the returned
+/// predicates, which the caller keeps above the projection. A second call adds
+/// no scan filter that is already present, so the plan is a fixed point.
+fn push_down_through_extraction_projection_into_scan(
+ predicates: Vec<Expr>,
+ mut projection: Projection,
+) -> Result<(Transformed<LogicalPlan>, Vec<Expr>)> {
+ let LogicalPlan::TableScan(_) = projection.input.as_ref() else {
+ return Ok((
+ Transformed::no(LogicalPlan::Projection(projection)),
+ predicates,
+ ));
+ };
+
+ // A pass-through column has the same name above and below the projection,
+ // so a predicate that reads only such columns needs no rewrite.
+ let pass_through: HashSet<&Column> = projection
+ .expr
+ .iter()
+ .filter_map(Expr::try_as_col)
+ .collect();
+ let reads_only_pass_through = |predicate: &Expr| {
+ predicate
+ .column_refs()
+ .iter()
+ .all(|column| pass_through.contains(column))
+ };
+
+ let Some(scan_predicate) = conjunction(
+ predicates
+ .iter()
+ .filter(|predicate| reads_only_pass_through(predicate))
+ .cloned(),
+ ) else {
+ return Ok((
+ Transformed::no(LogicalPlan::Projection(projection)),
+ predicates,
+ ));
+ };
+ let pushed = push_down_into_table_scan(
+ Filter::try_new(scan_predicate, Arc::clone(&projection.input))?,
+ false,
+ )?;
+ if !pushed.transformed {
+ return Ok((
+ Transformed::no(LogicalPlan::Projection(projection)),
+ predicates,
+ ));
+ }
+
+ // The predicates the scan does not apply exactly come back in a filter
+ // above it. They stay above the projection.
+ let (scan, residual) = match pushed.data {
+ LogicalPlan::Filter(residual) => (
+ Arc::unwrap_or_clone(residual.input),
+ split_conjunction_owned(residual.predicate),
+ ),
+ scan => (scan, vec![]),
+ };
+ // Keep the predicates in their original order, which `reorder_predicates`
+ // chose by cost.
+ let keep_predicates = predicates
Review Comment:
Consider adding a mixed-acceptance regression with Exact and Inexact or
Unsupported pass-through conjuncts plus an extracted-alias conjunct. Verify
that only Exact predicates leave the filter, accepted predicates reach the
scan, residual order is preserved, and a second rewrite is unchanged. This is
optional coverage for selective predicate reconciliation.
##########
datafusion/optimizer/src/push_down_filter.rs:
##########
@@ -4829,7 +4945,7 @@ mod tests {
/// optimizer pass. See <https://github.com/apache/datafusion/issues/14540>
/// and the module documentation.
#[test]
- fn filter_not_pushed_through_pure_extraction_projection() -> Result<()> {
+ fn filter_above_pure_extraction_projection_reaches_the_scan() ->
Result<()> {
Review Comment:
Consider an end-to-end regression for optimizing a plan, appending a
partition filter, and optimizing again. With three partition files, verify the
returned rows, preserved extraction projection, and physical scan file groups
containing only the matching partition. This would cover real `get_field`
extraction and dictionary partition literals beyond the existing unit tests.
Optional coverage improvement.
--
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]
