feat(physical-plan): add FilterExec constructor with built-in projection

Enables FilterExec creation with projection in a single call

Avoids duplicate compute_properties execution

Improves performance for windowed queries and pipelines

Author: GaneshPatil7517

datafusion/physical-plan/src/filter.rs

@@ -126,6 +126,64 @@ impl FilterExec {
         }
     }
 
+    /// Create a FilterExec with projection applied during construction.
+    ///
+    /// This method is more efficient than calling `try_new().with_projection()`
+    /// because it computes properties only once, avoiding redundant work.
+    ///
+    /// # Arguments
+    ///
+    /// * `predicate` - Boolean expression to filter rows
+    /// * `input` - Input execution plan
+    /// * `projection` - Optional column indices to project after filtering
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// // Create a filter that selects rows where column 0 > 5 and projects columns [1, 3]
+    /// let filter = FilterExec::try_new_with_projection(
+    ///     predicate,
+    ///     input_plan,
+    ///     Some(vec![1, 3])
+    /// )?;
+    /// ```
+    #[expect(clippy::needless_pass_by_value)]
+    pub fn try_new_with_projection(
+        predicate: Arc<dyn PhysicalExpr>,
+        input: Arc<dyn ExecutionPlan>,
+        projection: Option<Vec<usize>>,
+    ) -> Result<Self> {
+        match predicate.data_type(input.schema().as_ref())? {
+            DataType::Boolean => {
+                // Validate projection if provided
+                if let Some(ref proj) = projection {
+                    can_project(&input.schema(), Some(proj))?;
+                }
+
+                let default_selectivity = FILTER_EXEC_DEFAULT_SELECTIVITY;
+                let cache = Self::compute_properties(
+                    &input,
+                    &predicate,
+                    default_selectivity,
+                    projection.as_ref(),
+                )?;
+                Ok(Self {
+                    predicate,
+                    input: Arc::clone(&input),
+                    metrics: ExecutionPlanMetricsSet::new(),
+                    default_selectivity,
+                    cache,
+                    projection,
+                    batch_size: FILTER_EXEC_DEFAULT_BATCH_SIZE,
+                    fetch: None,
+                })
+            }
+            other => {
+                plan_err!("Filter predicate must return BOOLEAN values, got {other:?}")
+            }
+        }
+    }
+
     pub fn with_default_selectivity(
         mut self,
         default_selectivity: u8,
@@ -1586,4 +1644,251 @@ mod tests {
 
         Ok(())
     }
-}
+
+    #[tokio::test]
+    async fn test_try_new_with_projection_basic() -> Result<()> {
+        // Create a schema with multiple columns
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int32, false),
+            Field::new("c", DataType::Int32, false),
+        ]));
+
+        let input = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+
+        // Create a filter predicate: a > 10
+        let predicate = Arc::new(BinaryExpr::new(
+            Arc::new(Column::new("a", 0)),
+            Operator::Gt,
+            Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
+        ));
+
+        // Create filter with projection [0, 2] (columns a and c)
+        let projection = Some(vec![0, 2]);
+        let filter = FilterExec::try_new_with_projection(
+            Arc::clone(&predicate),
+            Arc::clone(&input),
+            projection.clone(),
+        )?;
+
+        // Verify projection is set correctly
+        assert_eq!(filter.projection(), Some(&vec![0, 2]));
+
+        // Verify schema contains only projected columns
+        let output_schema = filter.schema();
+        assert_eq!(output_schema.fields().len(), 2);
+        assert_eq!(output_schema.field(0).name(), "a");
+        assert_eq!(output_schema.field(1).name(), "c");
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_try_new_with_projection_none() -> Result<()> {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int32, false),
+        ]));
+
+        let input = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+
+        let predicate = Arc::new(BinaryExpr::new(
+            Arc::new(Column::new("a", 0)),
+            Operator::Gt,
+            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
+        ));
+
+        // Create filter without projection
+        let filter = FilterExec::try_new_with_projection(
+            Arc::clone(&predicate),
+            Arc::clone(&input),
+            None,
+        )?;
+
+        // Verify no projection is set
+        assert_eq!(filter.projection(), None);
+
+        // Verify schema contains all columns
+        let output_schema = filter.schema();
+        assert_eq!(output_schema.fields().len(), 2);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_try_new_with_projection_invalid() -> Result<()> {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int32, false),
+        ]));
+
+        let input = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+
+        let predicate = Arc::new(BinaryExpr::new(
+            Arc::new(Column::new("a", 0)),
+            Operator::Gt,
+            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
+        ));
+
+        // Try to create filter with invalid projection (index out of bounds)
+        let result = FilterExec::try_new_with_projection(
+            Arc::clone(&predicate),
+            Arc::clone(&input),
+            Some(vec![0, 5]), // 5 is out of bounds
+        );
+
+        // Should return an error
+        assert!(result.is_err());
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_try_new_with_projection_vs_with_projection() -> Result<()> {
+        // This test verifies that try_new_with_projection produces the same result
+        // as try_new().with_projection(), but more efficiently
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int32, false),
+            Field::new("c", DataType::Int32, false),
+            Field::new("d", DataType::Int32, false),
+        ]));
+
+        let input = Arc::new(StatisticsExec::new(
+            Statistics {
+                num_rows: Precision::Inexact(1000),
+                total_byte_size: Precision::Inexact(4000),
+                column_statistics: vec![
+                    ColumnStatistics {
+                        min_value: Precision::Inexact(ScalarValue::Int32(Some(1))),
+                        max_value: Precision::Inexact(ScalarValue::Int32(Some(100))),
+                        ..Default::default()
+                    },
+                    ColumnStatistics {
+                        ..Default::default()
+                    },
+                    ColumnStatistics {
+                        ..Default::default()
+                    },
+                    ColumnStatistics {
+                        ..Default::default()
+                    },
+                ],
+            },
+            Arc::clone(&schema),
+        ));
+
+        let predicate = Arc::new(BinaryExpr::new(
+            Arc::new(Column::new("a", 0)),
+            Operator::Lt,
+            Arc::new(Literal::new(ScalarValue::Int32(Some(50)))),
+        ));
+
+        let projection = Some(vec![0, 2]);
+
+        // Method 1: try_new_with_projection (one call to compute_properties)
+        let filter1 = FilterExec::try_new_with_projection(
+            Arc::clone(&predicate),
+            Arc::clone(&input),
+            projection.clone(),
+        )?;
+
+        // Method 2: try_new().with_projection() (two calls to compute_properties)
+        let filter2 = FilterExec::try_new(Arc::clone(&predicate), Arc::clone(&input))?
+            .with_projection(projection)?;
+
+        // Both methods should produce equivalent results
+        assert_eq!(filter1.schema(), filter2.schema());
+        assert_eq!(filter1.projection(), filter2.projection());
+
+        // Verify statistics are the same
+        let stats1 = filter1.partition_statistics(None)?;
+        let stats2 = filter2.partition_statistics(None)?;
+        assert_eq!(stats1.num_rows, stats2.num_rows);
+        assert_eq!(stats1.total_byte_size, stats2.total_byte_size);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_try_new_with_projection_statistics() -> Result<()> {
+        // Test that statistics are correctly computed with projection
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int32, false),
+            Field::new("c", DataType::Int32, false),
+        ]));
+
+        let input = Arc::new(StatisticsExec::new(
+            Statistics {
+                num_rows: Precision::Inexact(1000),
+                total_byte_size: Precision::Inexact(12000),
+                column_statistics: vec![
+                    ColumnStatistics {
+                        min_value: Precision::Inexact(ScalarValue::Int32(Some(1))),
+                        max_value: Precision::Inexact(ScalarValue::Int32(Some(100))),
+                        ..Default::default()
+                    },
+                    ColumnStatistics {
+                        min_value: Precision::Inexact(ScalarValue::Int32(Some(10))),
+                        max_value: Precision::Inexact(ScalarValue::Int32(Some(200))),
+                        ..Default::default()
+                    },
+                    ColumnStatistics {
+                        min_value: Precision::Inexact(ScalarValue::Int32(Some(5))),
+                        max_value: Precision::Inexact(ScalarValue::Int32(Some(50))),
+                        ..Default::default()
+                    },
+                ],
+            },
+            Arc::clone(&schema),
+        ));
+
+        // Filter: a < 50, Project: [0, 2]
+        let predicate = Arc::new(BinaryExpr::new(
+            Arc::new(Column::new("a", 0)),
+            Operator::Lt,
+            Arc::new(Literal::new(ScalarValue::Int32(Some(50)))),
+        ));
+
+        let filter = FilterExec::try_new_with_projection(
+            Arc::clone(&predicate),
+            Arc::clone(&input),
+            Some(vec![0, 2]),
+        )?;
+
+        let statistics = filter.partition_statistics(None)?;
+
+        // Verify statistics reflect both filtering and projection
+        assert!(matches!(statistics.num_rows, Precision::Inexact(_)));
+
+        // Schema should only have 2 columns after projection
+        assert_eq!(filter.schema().fields().len(), 2);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_try_new_with_projection_predicate_validation() -> Result<()> {
+        // Test that predicate type validation works correctly
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int32, false),
+        ]));
+
+        let input = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+
+        // Create a predicate that doesn't return boolean (returns Int32)
+        let invalid_predicate = Arc::new(Column::new("a", 0));
+
+        // Should fail because predicate doesn't return boolean
+        let result = FilterExec::try_new_with_projection(
+            invalid_predicate,
+            Arc::clone(&input),
+            Some(vec![0]),
+        );
+
+        assert!(result.is_err());
+
+        Ok(())
+    }