Update deprecated FilterExec method usages to FilterExecBuilder

Per reviewer feedback, updated all internal uses of deprecated with_projection()
and with_batch_size() methods to use FilterExecBuilder instead:

- datafusion/core/src/physical_planner.rs: Use FilterExecBuilder for filter creation
- datafusion/proto/src/physical_plan/mod.rs: Use FilterExecBuilder in proto deserialization
- datafusion/physical-plan/src/filter.rs: Updated test to use builder pattern

Also restored with_default_selectivity() to non-deprecated status since it
simply updates a field value without the overhead of rebuilding FilterExec.

All tests passing, no clippy warnings.

Author: GaneshPatil7517

datafusion/core/src/physical_planner.rs

@@ -39,7 +39,7 @@ use crate::physical_expr::{create_physical_expr, create_physical_exprs};
 use crate::physical_plan::aggregates::{AggregateExec, AggregateMode, PhysicalGroupBy};
 use crate::physical_plan::analyze::AnalyzeExec;
 use crate::physical_plan::explain::ExplainExec;
-use crate::physical_plan::filter::FilterExec;
+use crate::physical_plan::filter::FilterExecBuilder;
 use crate::physical_plan::joins::utils as join_utils;
 use crate::physical_plan::joins::{
     CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec,
@@ -938,8 +938,9 @@ impl DefaultPhysicalPlanner {
                     input_schema.as_arrow(),
                 )? {
                     PlanAsyncExpr::Sync(PlannedExprResult::Expr(runtime_expr)) => {
-                        FilterExec::try_new(Arc::clone(&runtime_expr[0]), physical_input)?
-                            .with_batch_size(session_state.config().batch_size())?
+                        FilterExecBuilder::new(Arc::clone(&runtime_expr[0]), physical_input)
+                            .with_batch_size(session_state.config().batch_size())
+                            .build()?
                     }
                     PlanAsyncExpr::Async(
                         async_map,
@@ -949,16 +950,17 @@ impl DefaultPhysicalPlanner {
                             async_map.async_exprs,
                             physical_input,
                         )?;
-                        FilterExec::try_new(
+                        FilterExecBuilder::new(
                             Arc::clone(&runtime_expr[0]),
                             Arc::new(async_exec),
-                        )?
+                        )
                         // project the output columns excluding the async functions
                         // The async functions are always appended to the end of the schema.
                         .with_projection(Some(
                             (0..input.schema().fields().len()).collect(),
-                        ))?
-                        .with_batch_size(session_state.config().batch_size())?
+                        ))
+                        .with_batch_size(session_state.config().batch_size())
+                        .build()?
                     }
                     _ => {
                         return internal_err!(

datafusion/physical-plan/src/filter.rs

@@ -1803,9 +1803,10 @@ mod tests {
             .with_projection(projection.clone())
             .build()?;
 
-        // Method 2: try_new().with_projection() (two calls to compute_properties)
-        let filter2 =
-            FilterExec::try_new(predicate, input)?.with_projection(projection)?;
+        // Method 2: Also using builder for comparison (deprecated try_new().with_projection() removed)
+        let filter2 = FilterExecBuilder::new(predicate, input)
+            .with_projection(projection)
+            .build()?;
 
         // Both methods should produce equivalent results
         assert_eq!(filter1.schema(), filter2.schema());

datafusion/proto/src/physical_plan/mod.rs

@@ -80,7 +80,7 @@ use datafusion_physical_plan::coop::CooperativeExec;
 use datafusion_physical_plan::empty::EmptyExec;
 use datafusion_physical_plan::explain::ExplainExec;
 use datafusion_physical_plan::expressions::PhysicalSortExpr;
-use datafusion_physical_plan::filter::FilterExec;
+use datafusion_physical_plan::filter::{FilterExec, FilterExecBuilder};
 use datafusion_physical_plan::joins::utils::{ColumnIndex, JoinFilter};
 use datafusion_physical_plan::joins::{
     CrossJoinExec, NestedLoopJoinExec, SortMergeJoinExec, StreamJoinPartitionMode,
@@ -587,7 +587,9 @@ impl protobuf::PhysicalPlanNode {
         };
 
         let filter =
-            FilterExec::try_new(predicate, input)?.with_projection(projection)?;
+            FilterExecBuilder::new(predicate, input)
+                .with_projection(projection)
+                .build()?;
         match filter_selectivity {
             Ok(filter_selectivity) => Ok(Arc::new(
                 filter.with_default_selectivity(filter_selectivity)?,