Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions datafusion/physical-expr/src/utils/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ use std::sync::Arc;

use crate::PhysicalExpr;
use crate::PhysicalSortExpr;
use crate::expressions::UnKnownColumn;
use crate::expressions::{BinaryExpr, Column};
use crate::tree_node::ExprContext;

Expand Down Expand Up @@ -238,6 +239,21 @@ pub fn collect_columns(expr: &Arc<dyn PhysicalExpr>) -> HashSet<Column> {
columns
}

pub fn have_unknown_columns(expr: &Arc<dyn PhysicalExpr>) -> bool {
let mut found = false;
expr.apply(|e| {
if e.as_any().downcast_ref::<UnKnownColumn>().is_some() {
found = true;
Ok(TreeNodeRecursion::Stop)
} else {
Ok(TreeNodeRecursion::Continue)
}
})
.expect("no way to return error during recursion");

found
}

/// Re-assign indices of [`Column`]s within the given [`PhysicalExpr`] according to
/// the provided [`Schema`].
///
Expand Down
8 changes: 6 additions & 2 deletions datafusion/physical-plan/src/filter_pushdown.rs
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,9 @@ use std::collections::HashSet;
use std::sync::Arc;

use datafusion_common::Result;
use datafusion_physical_expr::utils::{collect_columns, reassign_expr_columns};
use datafusion_physical_expr::utils::{
collect_columns, have_unknown_columns, reassign_expr_columns,
};
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
use itertools::Itertools;

Expand Down Expand Up @@ -339,7 +341,9 @@ impl ChildFilterDescription {
.iter()
.all(|col| child_column_names.contains(col.name()));

if all_columns_exist {
let have_unknown_columns = have_unknown_columns(filter);

if all_columns_exist && !have_unknown_columns {
// All columns exist in child - we can push down
// Need to reassign column indices to match child schema
let reassigned_filter =
Expand Down
1 change: 1 addition & 0 deletions datafusion/physical-plan/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ 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 {
Expand Down
Loading