Skip to content

Minor: Make ProjectionExpr::new easier to use with constants #19343

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alamb merged 3 commits into from
Dec 17, 2025
Merged

Minor: Make ProjectionExpr::new easier to use with constants #19343

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8f599b3
Minor: Make `ProjectionExpr::new` easier to use with constants
alamb Dec 15, 2025
e8d3d7d
Merge remote-tracking branch 'apache/main' into alamb/projection_expr…
alamb Dec 16, 2025
99edf9d
Updat emore
alamb Dec 16, 2025
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
2 changes: 1 addition & 1 deletion datafusion/core/tests/execution/coop.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -653,7 +653,7 @@ async fn join_agg_yields(

let proj_expr = vec![ProjectionExpr::new(
Arc::new(Column::new_with_schema("value", &input_schema)?) as _,
"value".to_string(),
"value",

@alamb alamb Dec 15, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The underlying ProjectionExpr is still the same, this PR just makes it easier to make them

)];

let projection = Arc::new(ProjectionExec::try_new(proj_expr, join)?);
Expand Down
152 changes: 67 additions & 85 deletions datafusion/core/tests/physical_optimizer/projection_pushdown.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -395,7 +395,7 @@ fn create_simple_csv_exec() -> Arc<dyn ExecutionPlan> {
};
Arc::new(CsvSource::new(schema.clone()).with_csv_options(options))
})
.with_file(PartitionedFile::new("x".to_string(), 100))
.with_file(PartitionedFile::new("x", 100))
.with_projection_indices(Some(vec![0, 1, 2, 3, 4]))
.unwrap()
.build();
Expand All @@ -420,7 +420,7 @@ fn create_projecting_csv_exec() -> Arc<dyn ExecutionPlan> {
};
Arc::new(CsvSource::new(schema.clone()).with_csv_options(options))
})
.with_file(PartitionedFile::new("x".to_string(), 100))
.with_file(PartitionedFile::new("x", 100))
.with_projection_indices(Some(vec![3, 2, 1]))
.unwrap()
.build();
Expand All @@ -445,8 +445,8 @@ fn test_csv_after_projection() -> Result<()> {
let csv = create_projecting_csv_exec();
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("b", 2)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("d", 0)), "d".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 2)), "b"),
ProjectionExpr::new(Arc::new(Column::new("d", 0)), "d"),
],
csv.clone(),
)?);
Expand Down Expand Up @@ -484,7 +484,7 @@ fn test_memory_after_projection() -> Result<()> {
vec![
ProjectionExpr::new(Arc::new(Column::new("d", 2)), "d".to_string()),
ProjectionExpr::new(Arc::new(Column::new("e", 3)), "e".to_string()),

@martin-g martin-g Dec 16, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why d and e are not simplified here ?
They look exactly the same as a

@alamb alamb Dec 16, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That is an excellent catch -- done in 99edf9d

ProjectionExpr::new(Arc::new(Column::new("a", 1)), "a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 1)), "a"),
],
memory.clone(),
)?);
Expand Down Expand Up @@ -588,9 +588,9 @@ fn test_streaming_table_after_projection() -> Result<()> {
)?;
let projection = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d".to_string()),
ProjectionExpr::new(Arc::new(Column::new("e", 2)), "e".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d"),
ProjectionExpr::new(Arc::new(Column::new("e", 2)), "e"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a"),
],
Arc::new(streaming_table) as _,
)?) as _;
Expand Down Expand Up @@ -655,28 +655,25 @@ fn test_projection_after_projection() -> Result<()> {
let csv = create_simple_csv_exec();
let child_projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("e", 4)), "new_e".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "new_b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c"),
ProjectionExpr::new(Arc::new(Column::new("e", 4)), "new_e"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "new_b"),
],
csv.clone(),
)?);
let top_projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("new_b", 3)), "new_b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("new_b", 3)), "new_b"),
ProjectionExpr::new(
Arc::new(BinaryExpr::new(
Arc::new(Column::new("c", 0)),
Operator::Plus,
Arc::new(Column::new("new_e", 1)),
)),
"binary".to_string(),
),
ProjectionExpr::new(
Arc::new(Column::new("new_b", 3)),
"newest_b".to_string(),
"binary",
),
ProjectionExpr::new(Arc::new(Column::new("new_b", 3)), "newest_b"),
],
child_projection.clone(),
)?);
Expand Down Expand Up @@ -741,9 +738,9 @@ fn test_output_req_after_projection() -> Result<()> {
));
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "new_a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "new_a"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
],
sort_req.clone(),
)?);
Expand Down Expand Up @@ -832,9 +829,9 @@ fn test_coalesce_partitions_after_projection() -> Result<()> {
Arc::new(CoalescePartitionsExec::new(csv));
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_new".to_string()),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_new"),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d"),
],
coalesce_partitions,
)?);
Expand Down Expand Up @@ -888,9 +885,9 @@ fn test_filter_after_projection() -> Result<()> {
let filter = Arc::new(FilterExec::try_new(predicate, csv)?);
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_new".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_new"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d"),
],
filter.clone(),
)?) as _;
Expand Down Expand Up @@ -982,17 +979,11 @@ fn test_join_after_projection() -> Result<()> {
)?);
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c_from_left".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b_from_left".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_from_left".to_string()),
ProjectionExpr::new(
Arc::new(Column::new("a", 5)),
"a_from_right".to_string(),
),
ProjectionExpr::new(
Arc::new(Column::new("c", 7)),
"c_from_right".to_string(),
),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c_from_left"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b_from_left"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_from_left"),
ProjectionExpr::new(Arc::new(Column::new("a", 5)), "a_from_right"),
ProjectionExpr::new(Arc::new(Column::new("c", 7)), "c_from_right"),
],
join,
)?) as _;
Expand Down Expand Up @@ -1111,16 +1102,16 @@ fn test_join_after_required_projection() -> Result<()> {
)?);
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("a", 5)), "a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 6)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 7)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("d", 8)), "d".to_string()),
ProjectionExpr::new(Arc::new(Column::new("e", 9)), "e".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d".to_string()),
ProjectionExpr::new(Arc::new(Column::new("e", 4)), "e".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 5)), "a"),
ProjectionExpr::new(Arc::new(Column::new("b", 6)), "b"),
ProjectionExpr::new(Arc::new(Column::new("c", 7)), "c"),
ProjectionExpr::new(Arc::new(Column::new("d", 8)), "d"),
ProjectionExpr::new(Arc::new(Column::new("e", 9)), "e"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c"),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d"),
ProjectionExpr::new(Arc::new(Column::new("e", 4)), "e"),
],
join,
)?) as _;
Expand Down Expand Up @@ -1200,7 +1191,7 @@ fn test_nested_loop_join_after_projection() -> Result<()> {
)?) as _;

let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![ProjectionExpr::new(col_left_c, "c".to_string())],
vec![ProjectionExpr::new(col_left_c, "c")],
Arc::clone(&join),
)?) as _;
let initial = displayable(projection.as_ref()).indent(true).to_string();
Expand Down Expand Up @@ -1290,13 +1281,10 @@ fn test_hash_join_after_projection() -> Result<()> {
)?);
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c_from_left".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b_from_left".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_from_left".to_string()),
ProjectionExpr::new(
Arc::new(Column::new("c", 7)),
"c_from_right".to_string(),
),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c_from_left"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b_from_left"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_from_left"),
ProjectionExpr::new(Arc::new(Column::new("c", 7)), "c_from_right"),
],
join.clone(),
)?) as _;
Expand Down Expand Up @@ -1332,10 +1320,10 @@ fn test_hash_join_after_projection() -> Result<()> {

let projection = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 7)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c"),
ProjectionExpr::new(Arc::new(Column::new("c", 7)), "c"),
],
join.clone(),
)?);
Expand Down Expand Up @@ -1376,9 +1364,9 @@ fn test_repartition_after_projection() -> Result<()> {
)?);
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b_new".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d_new".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b_new"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a"),
ProjectionExpr::new(Arc::new(Column::new("d", 3)), "d_new"),
],
repartition,
)?) as _;
Expand Down Expand Up @@ -1445,9 +1433,9 @@ fn test_sort_after_projection() -> Result<()> {
);
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "new_a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "new_a"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
],
Arc::new(sort_exec),
)?) as _;
Expand Down Expand Up @@ -1498,9 +1486,9 @@ fn test_sort_preserving_after_projection() -> Result<()> {
);
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "new_a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "new_a"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
],
Arc::new(sort_exec),
)?) as _;
Expand Down Expand Up @@ -1540,9 +1528,9 @@ fn test_union_after_projection() -> Result<()> {
let union = UnionExec::try_new(vec![csv.clone(), csv.clone(), csv])?;
let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
vec![
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c".to_string()),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "new_a".to_string()),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b".to_string()),
ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c"),
ProjectionExpr::new(Arc::new(Column::new("a", 0)), "new_a"),
ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
],
union.clone(),
)?) as _;
Expand Down Expand Up @@ -1602,7 +1590,7 @@ fn partitioned_data_source() -> Arc<DataSourceExec> {
ObjectStoreUrl::parse("test:///").unwrap(),
Arc::new(CsvSource::new(table_schema).with_csv_options(options)),
)
.with_file(PartitionedFile::new("x".to_string(), 100))
.with_file(PartitionedFile::new("x", 100))
.with_projection_indices(Some(vec![0, 1, 2]))
.unwrap()
.build();
Expand All @@ -1619,16 +1607,13 @@ fn test_partition_col_projection_pushdown() -> Result<()> {
vec![
ProjectionExpr::new(
col("string_col", partitioned_schema.as_ref())?,
"string_col".to_string(),
"string_col",
),
ProjectionExpr::new(
col("partition_col", partitioned_schema.as_ref())?,
"partition_col".to_string(),
),
ProjectionExpr::new(
col("int_col", partitioned_schema.as_ref())?,
"int_col".to_string(),
"partition_col",
),
ProjectionExpr::new(col("int_col", partitioned_schema.as_ref())?, "int_col"),
],
source,
)?);
Expand Down Expand Up @@ -1657,7 +1642,7 @@ fn test_partition_col_projection_pushdown_expr() -> Result<()> {
vec![
ProjectionExpr::new(
col("string_col", partitioned_schema.as_ref())?,
"string_col".to_string(),
"string_col",
),
ProjectionExpr::new(
// CAST(partition_col, Utf8View)
Expand All @@ -1666,12 +1651,9 @@ fn test_partition_col_projection_pushdown_expr() -> Result<()> {
partitioned_schema.as_ref(),
DataType::Utf8View,
)?,
"partition_col".to_string(),
),
ProjectionExpr::new(
col("int_col", partitioned_schema.as_ref())?,
"int_col".to_string(),
"partition_col",
),
ProjectionExpr::new(col("int_col", partitioned_schema.as_ref())?, "int_col"),
],
source,
)?);
Expand Down
2 changes: 1 addition & 1 deletion datafusion/datasource/src/file_scan_config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1782,7 +1782,7 @@ mod tests {
// the new projection won't include the filtered column.
let exprs = ProjectionExprs::new(vec![ProjectionExpr::new(
col("c1", &file_schema).unwrap(),
"c1".to_string(),
"c1",
)]);
let data_source = config
.try_swapping_with_projection(&exprs)
Expand Down
3 changes: 2 additions & 1 deletion datafusion/physical-expr/src/projection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,8 @@ impl std::fmt::Display for ProjectionExpr {

impl ProjectionExpr {
/// Create a new projection expression
pub fn new(expr: Arc<dyn PhysicalExpr>, alias: String) -> Self {
pub fn new(expr: Arc<dyn PhysicalExpr>, alias: impl Into<String>) -> Self {

@alamb alamb Dec 15, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is the actual code change, the rest of the PR is cleaning up callsites

let alias = alias.into();
Self { expr, alias }
}

Expand Down