feat: support pushdown alias on dynamic filter with ProjectionExec

[discord9]

Which issue does this PR close?

• Closes Dynamic filter pushdown through projection should support aliases #17246

Rationale for this change

For dynamic filter to work properly, table scan must get correct column even if it's passing through alias(by ProjectionExec) hence need to modify parent filter when gather_filters_for_pushdown

What changes are included in this PR?

as title, add support for handling simple alias in pushdown filter, which expand aliased column(in pushdown filter) to it's original expressions(or not pushdown if can't found aliased column in pushdown filter) so alias in projection is supported, also added unit tests.
AI Content disclaim: the core logic is hand written and thoroughly understood, but unit test are largely generated with some human guidance

Are these changes tested?

Unit tests&slt are added, please comment if more tests are needed

Are there any user-facing changes?

Yes, dynamic filter will work properly with alias now, I'm not sure if that count as breaking change though?

[discord9]

@jackkleeman @adriangb hi, I added the projection alias support in #17246, since you have the most context on this, could you please take a look when you have a chance?

[adriangb]

Looking good! Just needs some tweaks and more tests

[discord9]

Added tests in datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs and slt

[adriangb]

@discord9 I can't make a PR to your fork as far as far as I can tell, but can you check the diff in 7ee2bfb and see what you think? Sorry some of it is in 22981aa as well.

[alamb]

This appears to already be used elsewhere (this is not a net new depednecy), so I think it is ok to add

[discord9]

@discord9 I can't make a PR to your fork as far as far as I can tell, but can you check the diff in 7ee2bfb and see what you think? Sorry some of it is in 22981aa as well.

I have some doubt about leave it unchanged, say if in this case:

Sort filter=a@0>800
  <Some strange Extension node that went unhandled>(say something cause a unknown column(a new a@0 in this case) to appear>
    Projection some other columns, old a@0 not included(filter should be unknown column at this stage)
      <do something with old a@0>
      Projection b@0 as a@0
        TableScan filter= DynamicFilter [ b@0 > 800](should be unknown column, but left unchange confuse it with existing columns)

I guess my point is that replace not found column with UnknownColumn will allow downstream user to better debug their problem when DynamicFilter is involved(Since evaluate UnknownColumn return errors, and downstream custom TableScan could use this to easily determine whether he can use this dynamic filter, and even downstream user miss to handle Extension node's gather pushdown, it's easier to spot due to UnknownColumn), and for datafusion itself, in normal case without external extension logical plan, projection shouldn't cause unknown columns in filter, so it's ok?
But if we left unknown columns unchanged, certain query(with extension plan) might confuse those unchanged columns with something else and cause hidden bug which is much hard to discover or debug, so having UnknownColumn is more of a defensive guard thing?
And it seems better to have a explict way to see if it went wrong does no harm and is beneficial? Altough if you insist I can cherry pick 7ee2bfb
Edit: Maybe a middle ground is just refuse to pushdown filters when encounter unknown columns?
Edit: done

[adriangb]

@discord9 I can't make a PR to your fork as far as far as I can tell, but can you check the diff in 7ee2bfb and see what you think? Sorry some of it is in 22981aa as well.

I have some doubt about leave it unchanged, say if in this case:

Sort filter=a@0>800
  <Some strange Extension node that went unhandled>(say something cause a unknown column(a new a@0 in this case) to appear>
    Projection some other columns, old a@0 not included(filter should be unknown column at this stage)
      <do something with old a@0>
      Projection b@0 as a@0
        TableScan filter= DynamicFilter [ b@0 > 800](should be unknown column, but left unchange confuse it with existing columns)

I guess my point is that replace not found column with UnknownColumn will allow downstream user to better debug their problem when DynamicFilter is involved(Since evaluate UnknownColumn return errors, and downstream custom TableScan could use this to easily determine whether he can use this dynamic filter, and even downstream user miss to handle Extension node's gather pushdown, it's easier to spot due to UnknownColumn), and for datafusion itself, in normal case without external extension logical plan, projection shouldn't cause unknown columns in filter, so it's ok? But if we left unknown columns unchanged, certain query(with extension plan) might confuse those unchanged columns with something else and cause hidden bug which is much hard to discover or debug, so having UnknownColumn is more of a defensive guard thing? And it seems better to have a explict way to see if it went wrong does no harm and is beneficial? Altough if you insist I can cherry pick 7ee2bfb Edit: Maybe a middle ground is just refuse to pushdown filters when encounter unknown columns? Edit: done

I just don't think this will happen. If a plan is introducing an UnknownColumn indeed we can't support projection pushdown.

[discord9]

@adriangb done, now PhysicalColumnRewriter will simply return error, also had to rebase to resolve conflict

[adriangb]

Sorry to be annoying - one last request - could we rename this to column_rewriter.rs to avoid a util.rs module that as of today only holds one thing?

I would push to your repo / open a PR but I think your repo settings don't allow it.

Here's the diff:

diff --git a/datafusion/physical-plan/src/util.rs b/datafusion/physical-plan/src/column_rewriter.rs
similarity index 100%
rename from datafusion/physical-plan/src/util.rs
rename to datafusion/physical-plan/src/column_rewriter.rs
diff --git a/datafusion/physical-plan/src/lib.rs b/datafusion/physical-plan/src/lib.rs
index 79c0e9ef5..9352a143c 100644
--- a/datafusion/physical-plan/src/lib.rs
+++ b/datafusion/physical-plan/src/lib.rs
@@ -68,6 +68,7 @@ pub mod async_func;
 pub mod coalesce;
 pub mod coalesce_batches;
 pub mod coalesce_partitions;
+pub mod column_rewriter;
 pub mod common;
 pub mod coop;
 pub mod display;
@@ -92,7 +93,6 @@ pub mod streaming;
 pub mod tree_node;
 pub mod union;
 pub mod unnest;
-pub mod util;
 pub mod windows;
 pub mod work_table;
 pub mod udaf {
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index 1e9671900..8d4c775f8 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -26,13 +26,13 @@ use super::{
     DisplayAs, ExecutionPlanProperties, PlanProperties, RecordBatchStream,
     SendableRecordBatchStream, SortOrderPushdownResult, Statistics,
 };
+use crate::column_rewriter::PhysicalColumnRewriter;
 use crate::execution_plan::CardinalityEffect;
 use crate::filter_pushdown::{
     ChildFilterDescription, ChildPushdownResult, FilterColumnChecker, FilterDescription,
     FilterPushdownPhase, FilterPushdownPropagation, PushedDownPredicate,
 };
 use crate::joins::utils::{ColumnIndex, JoinFilter, JoinOn, JoinOnRef};
-use crate::util::PhysicalColumnRewriter;
 use crate::{DisplayFormatType, ExecutionPlan, PhysicalExpr};
 use std::any::Any;
 use std::collections::HashMap;

[adriangb]

thank you @discord9 !!