Skip to content
Draft
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
35 changes: 33 additions & 2 deletions datafusion/physical-plan/src/projection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ use crate::filter_pushdown::{
use crate::joins::utils::{ColumnIndex, JoinFilter, JoinOn, JoinOnRef};
use crate::statistics::{ChildStats, StatisticsArgs};
use crate::{DisplayFormatType, ExecutionPlan, PhysicalExpr, check_if_same_properties};
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
Expand Down Expand Up @@ -138,6 +138,15 @@ impl ProjectionExec {
{
let input_schema = input.schema();
let expr_arc = expr.into_iter().map(Into::into).collect::<Arc<_>>();
let mut output_names = HashSet::with_capacity(expr_arc.len());
for expr in expr_arc.iter() {
if !output_names.insert(expr.alias.as_str()) {
return plan_err!(
"ProjectionExec requires unique output column names, but found duplicate name '{}'",
expr.alias
);
}
}
let projection = ProjectionExprs::from_expressions(expr_arc);
let projector = projection.make_projector(&input_schema)?;
Self::try_from_projector(projector, input)
Expand Down Expand Up @@ -1270,7 +1279,7 @@ fn collect_column_indices(exprs: &[ProjectionExpr]) -> Vec<usize> {
// expression tree to collect column references in traversal order.
// This allows the embedded projection to match the desired output
// column order, avoiding a residual ProjectionExec.
let mut seen = std::collections::HashSet::new();
let mut seen = HashSet::new();
let mut indices = Vec::new();
for proj_expr in exprs {
if let Some(col) = proj_expr.expr.downcast_ref::<Column>() {
Expand Down Expand Up @@ -1609,6 +1618,28 @@ mod tests {
.unwrap();
}

#[test]
fn projection_rejects_duplicate_output_names() -> Result<()> {
let input = test::scan_partitioned(1);
let schema = input.schema();
let expr = col("i", &schema)?;

let error = ProjectionExec::try_new(
[
(Arc::clone(&expr), "duplicate".to_string()),
(expr, "duplicate".to_string()),
],
input,
)
.unwrap_err();

assert_eq!(
error.strip_backtrace(),
"Error during planning: ProjectionExec requires unique output column names, but found duplicate name 'duplicate'"
);
Ok(())
}

#[test]
fn test_projection_statistics_uses_input_schema() {
let input_schema = Schema::new(vec![
Expand Down