FilterPushdown do not generate correct column index when merge FilterExec

[haohuaijin]

Describe the bug

i have a plan like below that have two consecutive FilterExec, so i use FilterPushdown two merge it, but after the merge and execute, it report error. if i not call FilterPushDown, the plan can output result.

Physical plan before pushdown filter:
FilterExec: event@0 = Ingestion, projection=[size@1]
 FilterExec: time@0 < 1770020321951000, projection=[event@1, size@2]
 DataSourceExec: partitions=1, partition_sizes=[11]
Get result before pushdown filter, result size: 8

Physical plan after pushdown filter:
ProjectionExec: expr=[size@1 as size]
 FilterExec: event@0 = Ingestion AND time@0 < 1770020321951000
 DataSourceExec: partitions=1, partition_sizes=[11]
Error: ArrowError(InvalidArgumentError("Invalid comparison operation: Int64 == Utf8"), Some(""))

the event@0 is not correct, it should be event@1

To Reproduce

use std::sync::Arc;

use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::common::Result;
use datafusion::datasource::memory::MemorySourceConfig;
use datafusion::physical_expr::expressions::{BinaryExpr, col, lit};
use datafusion::physical_optimizer::PhysicalOptimizerRule;
use datafusion::physical_optimizer::filter_pushdown::FilterPushdown;
use datafusion::physical_plan::filter::FilterExec;
use datafusion::physical_plan::{ExecutionPlan, collect, displayable};
use datafusion::prelude::SessionContext;
use datafusion_expr::Operator;

#[tokio::main]
async fn main() -> Result<()> {
 // 1. Create data source
 let memory_exec = create_data_source();

 // 2. First FilterExec: time < 1770020321951000 with projection=[event@1, size@2]
 let timestamp_col = col("time", &memory_exec.schema())?;
 let upper_bound = lit(1770020321951000i64);
 let lt_expr = Arc::new(BinaryExpr::new(timestamp_col, Operator::Lt, upper_bound));
 let filter1_exec: Arc<dyn ExecutionPlan> = Arc::new(
 FilterExec::try_new(lt_expr, memory_exec)?.with_projection(Some(vec![1, 2]))?, // projection=[event@1, size@2]
 );

 // 3. Second FilterExec: event = 'Ingestion' with projection=[size]
 let event_col = col("event", &filter1_exec.schema())?;
 let ingestion_lit = lit("Ingestion");
 let event_filter = Arc::new(BinaryExpr::new(event_col, Operator::Eq, ingestion_lit));
 let filter2_exec: Arc<dyn ExecutionPlan> = Arc::new(
 FilterExec::try_new(event_filter, filter1_exec)?
 .with_projection(Some(vec![1]))?, // projection=[size@1]
 );

 println!("Physical plan before pushdown filter:");
 print!(
 "{}",
 displayable(filter2_exec.as_ref()).indent(true).to_string()
 );

 // Execute the plan
 let ctx = SessionContext::new();
 let result = collect(filter2_exec.clone(), ctx.task_ctx()).await?;

 println!(
 "Get result before pushdown filter, result size: {:?}\n",
 result.iter().fold(0, |acc, x| acc + x.num_rows())
 );

 // After pushdown filter
 let pushdown_filter = FilterPushdown::new();
 let optimized_plan =
 pushdown_filter.optimize(filter2_exec.clone(), ctx.state().config_options())?;

 println!("Physical plan after pushdown filter:");
 print!(
 "{}",
 displayable(optimized_plan.as_ref())
 .indent(true)
 .to_string(),
 );

 let result = collect(optimized_plan.clone(), ctx.task_ctx()).await?;
 println!("{:?}", result);

 Ok(())
}

fn create_data_source() -> Arc<dyn ExecutionPlan> {
 // Create schema: time, event, size
 let schema = Arc::new(Schema::new(vec![
 Field::new("time", DataType::Int64, false),
 Field::new("event", DataType::Utf8, false),
 Field::new("size", DataType::Int64, false),
 ]));

 // Create 11 batches of sample data to get partition_sizes=[11]
 let timestamps = vec![
 1770020021951000i64,
 1770020021951001,
 1770020021951002,
 1770020021951003,
 1770020021951004,
 1770020021951005,
 1770020021951006,
 1770020021951007,
 1770020021951008,
 1770020021951009,
 1770020021951010,
 ];
 let events = vec![
 "Ingestion",
 "Ingestion",
 "Query",
 "Ingestion",
 "Ingestion",
 "Query",
 "Ingestion",
 "Ingestion",
 "Ingestion",
 "Query",
 "Ingestion",
 ];
 let sizes = vec![100i64, 200, 300, 400, 500, 600, 700, 800, 900, 1000, 1100];

 // Create 11 separate batches (one row each) to get partition_sizes=[11]
 let batches: Vec<RecordBatch> = (0..11)
 .map(|i| {
 RecordBatch::try_new(
 schema.clone(),
 vec![
 Arc::new(Int64Array::from(vec![timestamps[i]])),
 Arc::new(StringArray::from(vec![events[i]])),
 Arc::new(Int64Array::from(vec![sizes[i]])),
 ],
 )
 .unwrap()
 })
 .collect();

 // Create MemorySourceConfig to create the data source
 MemorySourceConfig::try_new_exec(&[batches], schema.clone(), None).unwrap()
}

Expected behavior

success run

Additional context

test all version that have FilterPushdown, all not work

[alamb]

Added as a candidate to the 52.2.0 patch release

• Release DataFusion 52.2.0 (minor/) Release (Feb 2026) #20287