Skip to content

Commit 53de994

Browse files
authored
Add Expr::try_as_col, deprecate Expr::try_into_col (#10448)
1 parent 5fac581 commit 53de994

File tree

7 files changed

+55
-12
lines changed

7 files changed

+55
-12
lines changed

datafusion/expr/src/expr.rs

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1264,13 +1264,36 @@ impl Expr {
12641264
})
12651265
}
12661266

1267+
#[deprecated(since = "39.0.0", note = "use try_as_col instead")]
12671268
pub fn try_into_col(&self) -> Result<Column> {
12681269
match self {
12691270
Expr::Column(it) => Ok(it.clone()),
12701271
_ => plan_err!("Could not coerce '{self}' into Column!"),
12711272
}
12721273
}
12731274

1275+
/// Return a reference to the inner `Column` if any
1276+
///
1277+
/// returns `None` if the expression is not a `Column`
1278+
///
1279+
/// Example
1280+
/// ```
1281+
/// # use datafusion_common::Column;
1282+
/// use datafusion_expr::{col, Expr};
1283+
/// let expr = col("foo");
1284+
/// assert_eq!(expr.try_as_col(), Some(&Column::from("foo")));
1285+
///
1286+
/// let expr = col("foo").alias("bar");
1287+
/// assert_eq!(expr.try_as_col(), None);
1288+
/// ```
1289+
pub fn try_as_col(&self) -> Option<&Column> {
1290+
if let Expr::Column(it) = self {
1291+
Some(it)
1292+
} else {
1293+
None
1294+
}
1295+
}
1296+
12741297
/// Return all referenced columns of this expression.
12751298
pub fn to_columns(&self) -> Result<HashSet<Column>> {
12761299
let mut using_columns = HashSet::new();

datafusion/expr/src/expr_rewriter/mod.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -221,7 +221,7 @@ pub fn coerce_plan_expr_for_schema(
221221
let exprs: Vec<Expr> = plan.schema().iter().map(Expr::from).collect();
222222

223223
let new_exprs = coerce_exprs_for_schema(exprs, plan.schema(), schema)?;
224-
let add_project = new_exprs.iter().any(|expr| expr.try_into_col().is_err());
224+
let add_project = new_exprs.iter().any(|expr| expr.try_as_col().is_none());
225225
if add_project {
226226
let projection = Projection::try_new(new_exprs, Arc::new(plan.clone()))?;
227227
Ok(LogicalPlan::Projection(projection))

datafusion/expr/src/logical_plan/builder.rs

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1489,7 +1489,7 @@ pub fn wrap_projection_for_join_if_necessary(
14891489
let mut projection = expand_wildcard(input_schema, &input, None)?;
14901490
let join_key_items = alias_join_keys
14911491
.iter()
1492-
.flat_map(|expr| expr.try_into_col().is_err().then_some(expr))
1492+
.flat_map(|expr| expr.try_as_col().is_none().then_some(expr))
14931493
.cloned()
14941494
.collect::<HashSet<Expr>>();
14951495
projection.extend(join_key_items);
@@ -1504,8 +1504,12 @@ pub fn wrap_projection_for_join_if_necessary(
15041504
let join_on = alias_join_keys
15051505
.into_iter()
15061506
.map(|key| {
1507-
key.try_into_col()
1508-
.or_else(|_| Ok(Column::from_name(key.display_name()?)))
1507+
if let Some(col) = key.try_as_col() {
1508+
Ok(col.clone())
1509+
} else {
1510+
let name = key.display_name()?;
1511+
Ok(Column::from_name(name))
1512+
}
15091513
})
15101514
.collect::<Result<Vec<_>>>()?;
15111515

datafusion/expr/src/logical_plan/plan.rs

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -369,8 +369,18 @@ impl LogicalPlan {
369369
// The join keys in using-join must be columns.
370370
let columns =
371371
on.iter().try_fold(HashSet::new(), |mut accumu, (l, r)| {
372-
accumu.insert(l.try_into_col()?);
373-
accumu.insert(r.try_into_col()?);
372+
let Some(l) = l.try_as_col().cloned() else {
373+
return internal_err!(
374+
"Invalid join key. Expected column, found {l:?}"
375+
);
376+
};
377+
let Some(r) = r.try_as_col().cloned() else {
378+
return internal_err!(
379+
"Invalid join key. Expected column, found {r:?}"
380+
);
381+
};
382+
accumu.insert(l);
383+
accumu.insert(r);
374384
Result::<_, DataFusionError>::Ok(accumu)
375385
})?;
376386
using_columns.push(columns);

datafusion/optimizer/src/push_down_filter.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -535,8 +535,8 @@ fn push_down_join(
535535
.on
536536
.iter()
537537
.filter_map(|(l, r)| {
538-
let left_col = l.try_into_col().ok()?;
539-
let right_col = r.try_into_col().ok()?;
538+
let left_col = l.try_as_col().cloned()?;
539+
let right_col = r.try_as_col().cloned()?;
540540
Some((left_col, right_col))
541541
})
542542
.collect::<Vec<_>>();

datafusion/optimizer/src/simplify_expressions/inlist_simplifier.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ impl TreeNodeRewriter for ShortenInListSimplifier {
5252
// expressions
5353
list.len() == 1
5454
|| list.len() <= THRESHOLD_INLINE_INLIST
55-
&& expr.try_into_col().is_ok()
55+
&& expr.try_as_col().is_some()
5656
)
5757
{
5858
let first_val = list[0].clone();

datafusion/proto/src/logical_plan/mod.rs

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,8 +45,9 @@ use datafusion::{
4545
prelude::SessionContext,
4646
};
4747
use datafusion_common::{
48-
context, internal_err, not_impl_err, parsers::CompressionTypeVariant,
49-
plan_datafusion_err, DataFusionError, Result, TableReference,
48+
context, internal_datafusion_err, internal_err, not_impl_err,
49+
parsers::CompressionTypeVariant, plan_datafusion_err, DataFusionError, Result,
50+
TableReference,
5051
};
5152
use datafusion_expr::{
5253
dml,
@@ -695,7 +696,12 @@ impl AsLogicalPlan for LogicalPlanNode {
695696
// The equijoin keys in using-join must be column.
696697
let using_keys = left_keys
697698
.into_iter()
698-
.map(|key| key.try_into_col())
699+
.map(|key| {
700+
key.try_as_col().cloned()
701+
.ok_or_else(|| internal_datafusion_err!(
702+
"Using join keys must be column references, got: {key:?}"
703+
))
704+
})
699705
.collect::<Result<Vec<_>, _>>()?;
700706
builder.join_using(
701707
into_logical_plan!(join.right, ctx, extension_codec)?,

0 commit comments

Comments
 (0)