This is an automated email from the ASF dual-hosted git repository.
comphead pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 3051d1928e Migrate more code from `Expr::to_columns` to
`Expr::column_refs` (#11067)
3051d1928e is described below
commit 3051d1928eae8f684e1049a2cb00a8e721d70182
Author: Andrew Lamb <[email protected]>
AuthorDate: Sun Jun 23 13:13:36 2024 -0400
Migrate more code from `Expr::to_columns` to `Expr::column_refs` (#11067)
* Migrate code from `Expr::to_columns` to `Expr::column_refs`
---
datafusion/core/src/physical_planner.rs | 2 +-
datafusion/expr/src/logical_plan/builder.rs | 6 +++---
datafusion/expr/src/utils.rs | 6 +++---
datafusion/optimizer/src/analyzer/subquery.rs | 2 +-
datafusion/optimizer/src/push_down_filter.rs | 18 +++---------------
datafusion/optimizer/src/utils.rs | 13 ++++++++++++-
6 files changed, 23 insertions(+), 24 deletions(-)
diff --git a/datafusion/core/src/physical_planner.rs
b/datafusion/core/src/physical_planner.rs
index b539544d83..52d976c45c 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -1259,7 +1259,7 @@ impl DefaultPhysicalPlanner {
let join_filter = match filter {
Some(expr) => {
// Extract columns from filter expression and saved in
a HashSet
- let cols = expr.to_columns()?;
+ let cols = expr.column_refs();
// Collect left & right field indices, the field
indices are sorted in ascending order
let left_field_indices = cols
diff --git a/datafusion/expr/src/logical_plan/builder.rs
b/datafusion/expr/src/logical_plan/builder.rs
index 2f1ece32ab..434f4dace1 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -534,11 +534,11 @@ impl LogicalPlanBuilder {
.clone()
.into_iter()
.try_for_each::<_, Result<()>>(|expr| {
- let columns = expr.to_columns()?;
+ let columns = expr.column_refs();
columns.into_iter().for_each(|c| {
- if schema.field_from_column(&c).is_err() {
- missing_cols.push(c);
+ if !schema.has_column(c) {
+ missing_cols.push(c.clone());
}
});
diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs
index 6baabfcc71..9018b6f5ce 100644
--- a/datafusion/expr/src/utils.rs
+++ b/datafusion/expr/src/utils.rs
@@ -872,7 +872,7 @@ pub fn can_hash(data_type: &DataType) -> bool {
/// Check whether all columns are from the schema.
pub fn check_all_columns_from_schema(
- columns: &HashSet<Column>,
+ columns: &HashSet<&Column>,
schema: &DFSchema,
) -> Result<bool> {
for col in columns.iter() {
@@ -900,8 +900,8 @@ pub fn find_valid_equijoin_key_pair(
left_schema: &DFSchema,
right_schema: &DFSchema,
) -> Result<Option<(Expr, Expr)>> {
- let left_using_columns = left_key.to_columns()?;
- let right_using_columns = right_key.to_columns()?;
+ let left_using_columns = left_key.column_refs();
+ let right_using_columns = right_key.column_refs();
// Conditions like a = 10, will be added to non-equijoin.
if left_using_columns.is_empty() || right_using_columns.is_empty() {
diff --git a/datafusion/optimizer/src/analyzer/subquery.rs
b/datafusion/optimizer/src/analyzer/subquery.rs
index 7d1290204e..5725a725e6 100644
--- a/datafusion/optimizer/src/analyzer/subquery.rs
+++ b/datafusion/optimizer/src/analyzer/subquery.rs
@@ -249,7 +249,7 @@ fn check_aggregation_in_scalar_subquery(
let mut group_columns = agg
.group_expr
.iter()
- .map(|group|
Ok(group.to_columns()?.into_iter().collect::<Vec<_>>()))
+ .map(|group|
Ok(group.column_refs().into_iter().cloned().collect::<Vec<_>>()))
.collect::<Result<Vec<_>>>()?
.into_iter()
.flatten();
diff --git a/datafusion/optimizer/src/push_down_filter.rs
b/datafusion/optimizer/src/push_down_filter.rs
index 013e74c8fc..fa432ad76d 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -41,6 +41,7 @@ use datafusion_expr::{
};
use crate::optimizer::ApplyOrder;
+use crate::utils::has_all_column_refs;
use crate::{OptimizerConfig, OptimizerRule};
/// Optimizer rule for pushing (moving) filter expressions down in a plan so
@@ -199,13 +200,7 @@ fn can_pushdown_join_predicate(predicate: &Expr, schema:
&DFSchema) -> Result<bo
]
})
.collect::<HashSet<_>>();
- let columns = predicate.to_columns()?;
-
- Ok(schema_columns
- .intersection(&columns)
- .collect::<HashSet<_>>()
- .len()
- == columns.len())
+ Ok(has_all_column_refs(predicate, &schema_columns))
}
/// Determine whether the predicate can evaluate as the join conditions
@@ -372,14 +367,7 @@ fn extract_or_clause(expr: &Expr, schema_columns:
&HashSet<Column>) -> Option<Ex
}
}
_ => {
- let columns = expr.to_columns().ok().unwrap();
-
- if schema_columns
- .intersection(&columns)
- .collect::<HashSet<_>>()
- .len()
- == columns.len()
- {
+ if has_all_column_refs(expr, schema_columns) {
predicate = Some(expr.clone());
}
}
diff --git a/datafusion/optimizer/src/utils.rs
b/datafusion/optimizer/src/utils.rs
index 00aaff196c..05b1744d90 100644
--- a/datafusion/optimizer/src/utils.rs
+++ b/datafusion/optimizer/src/utils.rs
@@ -17,7 +17,7 @@
//! Utility functions leveraged by the query optimizer rules
-use std::collections::{BTreeSet, HashMap};
+use std::collections::{BTreeSet, HashMap, HashSet};
use crate::{OptimizerConfig, OptimizerRule};
@@ -66,6 +66,17 @@ pub fn optimize_children(
}
}
+/// Returns true if `expr` contains all columns in `schema_cols`
+pub(crate) fn has_all_column_refs(expr: &Expr, schema_cols: &HashSet<Column>)
-> bool {
+ let column_refs = expr.column_refs();
+ // note can't use HashSet::intersect because of different types (owned vs
References)
+ schema_cols
+ .iter()
+ .filter(|c| column_refs.contains(c))
+ .count()
+ == column_refs.len()
+}
+
pub(crate) fn collect_subquery_cols(
exprs: &[Expr],
subquery_schema: DFSchemaRef,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
