FilterPushdown physical optimizer incorrectly remaps filters through ProjectionExec with duplicate column names #21246

// column_with_name always returns the FIRST match
let (aliased_index, _) = self.projector.output_schema().column_with_name(&projection.alias)?;      // ← always returns 0 for duplicate names
let aliased_col = Column::new(&projection.alias, aliased_index);
alias_map.insert(aliased_col, Arc::clone( & projection.expr)); // ← overwrites on collision

// index_of always returns the FIRST match
let Ok(new_index) = self.child_schema.index_of(col.name())
// col was id@2 → new_index is 0 → rewrites to id@0

/// Reproduces a bug where FilterPushdown through ProjectionExec with duplicate
/// column names remaps filter predicates to the wrong source column.
///
/// Plan structure:
///
///   FilterExec: id@0 IS NULL          ← checks output col 0 (right side's id)
///     ProjectionExec: [right_id@2 as id,   ← output col 0 (from RIGHT)
///                      name@1,
///                      left_id@0 as id]    ← output col 2 (from LEFT)
///       HashJoinExec: Left
///         left:  [left_id, name]       (columns 0-1)
///         right: [right_id]            (column 2)
///
/// Bug 1 overwrites Column("id", 0) → left_id@0 instead of right_id@2.
/// The filter `id@0 IS NULL` gets remapped to `left_id@0 IS NULL` (wrong side).
#[tokio::test]
async fn test_filter_pushdown_projection_duplicate_column_names() {
    use datafusion_common::JoinType;
    use datafusion_physical_expr::expressions::is_null;
    use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode};

    // Left table: orders (5 rows, all non-NULL left_id)
    let left_batches = vec![record_batch!(
        ("left_id", Int32, [1, 2, 3, 4, 5]),
        ("name", Utf8, ["Alice", "Bob", "Charlie", "Diana", "Eve"])
    )
    .unwrap()];
    let left_schema = Arc::new(Schema::new(vec![
        Field::new("left_id", DataType::Int32, false),
        Field::new("name", DataType::Utf8, false),
    ]));
    let left_scan = TestScanBuilder::new(Arc::clone(&left_schema))
        .with_batches(left_batches)
        .build();

    // Right table: returns (2 rows matching orders 1 and 3)
    let right_batches = vec![
        record_batch!(("right_id", Int32, [1, 3])).unwrap(),
    ];
    let right_schema = Arc::new(Schema::new(vec![Field::new(
        "right_id",
        DataType::Int32,
        false,
    )]));
    let right_scan = TestScanBuilder::new(Arc::clone(&right_schema))
        .with_batches(right_batches)
        .build();

    // HashJoinExec: LEFT JOIN on left_id = right_id
    // Join output schema: [left_id(0), name(1), right_id(2)]
    let join = Arc::new(
        HashJoinExec::try_new(
            left_scan,
            right_scan,
            vec![(
                col("left_id", &left_schema).unwrap(),
                col("right_id", &right_schema).unwrap(),
            )],
            None,
            &JoinType::Left,
            None,
            PartitionMode::CollectLeft,
            datafusion_common::NullEquality::NullEqualsNothing,
            false,
        )
        .unwrap(),
    );
    let join_schema = join.schema();

    // ProjectionExec: creates duplicate "id" columns
    //   output col 0: right_id@2 AS id  (from RIGHT side, NULL for unmatched)
    //   output col 1: name@1
    //   output col 2: left_id@0 AS id   (from LEFT side, never NULL)
    let projection = Arc::new(
        ProjectionExec::try_new(
            vec![
                (col("right_id", &join_schema).unwrap(), "id".to_string()),
                (col("name", &join_schema).unwrap(), "name".to_string()),
                (col("left_id", &join_schema).unwrap(), "id".to_string()),
            ],
            join,
        )
        .unwrap(),
    );

    // FilterExec: id@0 IS NULL
    // This should check the RIGHT side's id (output col 0 = right_id).
    let filter_expr = is_null(Arc::new(Column::new("id", 0))).unwrap();
    let plan = Arc::new(
        FilterExec::try_new(filter_expr, projection).unwrap(),
    ) as Arc<dyn ExecutionPlan>;

    // Apply the physical FilterPushdown optimizer
    let config = ConfigOptions::default();
    let optimized = FilterPushdown::new()
        .optimize(Arc::clone(&plan), &config)
        .unwrap();

    // Execute both plans and compare
    let session_ctx = SessionContext::new_with_config(SessionConfig::new());
    session_ctx.register_object_store(
        ObjectStoreUrl::parse("test://").unwrap().as_ref(),
        Arc::new(InMemory::new()),
    );
    let task_ctx = session_ctx.state().task_ctx();

    let original_batches = collect(Arc::clone(&plan), Arc::clone(&task_ctx))
        .await
        .unwrap();
    let original_count: usize =
        original_batches.iter().map(|b| b.num_rows()).sum();

    let optimized_batches = collect(optimized, Arc::clone(&task_ctx))
        .await
        .unwrap();
    let optimized_count: usize =
        optimized_batches.iter().map(|b| b.num_rows()).sum();

    // Original plan correctly returns 3 rows (orders 2, 4, 5 — unmatched)
    assert_eq!(original_count, 3, "original plan should return 3 rows");

    // BUG: optimized plan returns 0 rows — filter was remapped to wrong column
    // After fix, change to: assert_eq!(optimized_count, 3);
    assert_eq!(optimized_count, 0,
        "BUG: filter pushed to wrong column, expected 3 rows but got 0");
}