diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index ad40a3fb5fd8..d367be16eb6e 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -812,6 +812,101 @@ impl ExecutionPlan for FilterExec { .ok() }) } + + #[cfg(feature = "proto")] + fn try_to_proto( + &self, + ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, + ) -> Result> { + use datafusion_proto_models::protobuf; + let input = ctx.encode_child(self.input())?; + let expr = ctx.encode_expr(self.predicate())?; + // Preserve the exact wire format: `None` (full projection) is serialized + // as the identity projection `[0, 1, ..., num_fields - 1]` so that it is + // distinguishable from an explicit projection on decode. + let projection = if let Some(v) = self.projection() { + v.iter().map(|x| *x as u32).collect() + } else { + (0..self.input().schema().fields().len()) + .map(|i| i as u32) + .collect() + }; + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some( + protobuf::physical_plan_node::PhysicalPlanType::Filter(Box::new( + protobuf::FilterExecNode { + input: Some(Box::new(input)), + expr: Some(expr), + default_filter_selectivity: self.default_selectivity() as u32, + projection, + batch_size: self.batch_size() as u32, + fetch: self.fetch().map(|f| f as u32), + }, + )), + ), + })) + } +} + +#[cfg(feature = "proto")] +impl FilterExec { + /// Reconstruct a [`FilterExec`] from its protobuf representation. + /// + /// The exact inverse of [`ExecutionPlan::try_to_proto`]: it takes the whole + /// [`PhysicalPlanNode`] so every plan's `try_from_proto` shares one signature. + /// + /// [`PhysicalPlanNode`]: datafusion_proto_models::protobuf::PhysicalPlanNode + /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto + pub fn try_from_proto( + node: &datafusion_proto_models::protobuf::PhysicalPlanNode, + ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, + ) -> Result> { + use datafusion_proto_models::protobuf; + let filter = crate::expect_plan_variant!( + node, + protobuf::physical_plan_node::PhysicalPlanType::Filter, + "FilterExec", + ); + let input = + ctx.decode_required_child(filter.input.as_deref(), "FilterExec", "input")?; + let predicate = ctx.decode_required_expr( + filter.expr.as_ref(), + input.schema().as_ref(), + "FilterExec", + "expr", + )?; + let filter_selectivity = filter.default_filter_selectivity.try_into(); + + // `None` is encoded as the full identity projection. Reconstruct it only + // when all input columns are present in order, leaving an empty list as + // `Some(vec![])`. + let num_fields = input.schema().fields().len(); + let mut is_full_projection = filter.projection.len() == num_fields; + let mut projection_vec: Vec = Vec::with_capacity(filter.projection.len()); + for (i, idx) in filter.projection.iter().enumerate() { + let idx = *idx as usize; + is_full_projection &= idx == i; + projection_vec.push(idx); + } + let projection = if is_full_projection { + None + } else { + Some(projection_vec) + }; + let filter = FilterExecBuilder::new(predicate, input) + .apply_projection(projection)? + .with_batch_size(filter.batch_size as usize) + .with_fetch(filter.fetch.map(|f| f as usize)) + .build()?; + match filter_selectivity { + Ok(filter_selectivity) => Ok(Arc::new( + filter.with_default_selectivity(filter_selectivity)?, + )), + Err(_) => Err(datafusion_common::internal_datafusion_err!( + "filter_selectivity in PhysicalPlanNode is invalid" + )), + } + } } impl EmbeddedProjection for FilterExec { diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index 1ef82375952c..86496300020c 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -77,7 +77,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, FilterExecBuilder}; +use datafusion_physical_plan::filter::FilterExec; use datafusion_physical_plan::joins::utils::{ColumnIndex, JoinFilter}; use datafusion_physical_plan::joins::{ CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec, @@ -756,8 +756,8 @@ pub trait PhysicalPlanNodeExt: Sized { PhysicalPlanType::Projection(_) => { ProjectionExec::try_from_proto(self.node(), &decode_ctx) } - PhysicalPlanType::Filter(filter) => { - self.try_into_filter_physical_plan(filter, ctx, proto_converter) + PhysicalPlanType::Filter(_) => { + FilterExec::try_from_proto(self.node(), &decode_ctx) } PhysicalPlanType::CsvScan(scan) => { self.try_into_csv_scan_physical_plan(scan, ctx, proto_converter) @@ -913,14 +913,6 @@ pub trait PhysicalPlanNodeExt: Sized { ); } - if let Some(exec) = plan.downcast_ref::() { - return protobuf::PhysicalPlanNode::try_from_filter_exec( - exec, - codec, - proto_converter, - ); - } - if let Some(limit) = plan.downcast_ref::() { return protobuf::PhysicalPlanNode::try_from_global_limit_exec( limit, @@ -1214,59 +1206,25 @@ pub trait PhysicalPlanNodeExt: Sized { Ok(Arc::new(ProjectionExec::try_new(proj_exprs, input)?)) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `FilterExec` deserializes itself via `FilterExec::try_from_proto`" + )] fn try_into_filter_physical_plan( &self, filter: &protobuf::FilterExecNode, ctx: &PhysicalPlanDecodeContext<'_>, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { - let input: Arc = - into_physical_plan(&filter.input, ctx, proto_converter)?; - - let predicate = filter - .expr - .as_ref() - .map(|expr| { - proto_converter.proto_to_physical_expr(expr, input.schema().as_ref(), ctx) - }) - .transpose()? - .ok_or_else(|| { - internal_datafusion_err!( - "filter (FilterExecNode) in PhysicalPlanNode is missing." - ) - })?; - - let filter_selectivity = filter.default_filter_selectivity.try_into(); - // Preserve the `None` state across proto boundaries. Proto cannot distinguish - // between `None` (full projection) and `Some(vec![])` (empty projection) since - // both serialize as an empty list. If all columns are included, we reconstruct - // `None` to avoid losing this semantic distinction on deserialization. - let num_fields = input.schema().fields().len(); - let mut is_full_projection = filter.projection.len() == num_fields; - let mut projection_vec: Vec = Vec::with_capacity(filter.projection.len()); - for (i, idx) in filter.projection.iter().enumerate() { - let idx = *idx as usize; - is_full_projection &= idx == i; - projection_vec.push(idx); - } - let projection = if is_full_projection { - None - } else { - Some(projection_vec) + let node = protobuf::PhysicalPlanNode { + physical_plan_type: Some(PhysicalPlanType::Filter(Box::new(filter.clone()))), }; - let filter = FilterExecBuilder::new(predicate, input) - .apply_projection(projection)? - .with_batch_size(filter.batch_size as usize) - .with_fetch(filter.fetch.map(|f| f as usize)) - .build()?; - match filter_selectivity { - Ok(filter_selectivity) => Ok(Arc::new( - filter.with_default_selectivity(filter_selectivity)?, - )), - Err(_) => Err(internal_datafusion_err!( - "filter_selectivity in PhysicalPlanNode is invalid " - )), - } + let decoder = ConverterPlanDecoder { + ctx, + proto_converter, + }; + let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); + FilterExec::try_from_proto(&node, &decode_ctx) } fn try_into_csv_scan_physical_plan( @@ -2963,36 +2921,22 @@ pub trait PhysicalPlanNodeExt: Sized { }) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `FilterExec` serializes itself via `ExecutionPlan::try_to_proto`" + )] fn try_from_filter_exec( exec: &FilterExec, codec: &dyn PhysicalExtensionCodec, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result { - let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - exec.input().to_owned(), + let encoder = ConverterPlanEncoder { codec, proto_converter, - )?; - Ok(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Filter(Box::new( - protobuf::FilterExecNode { - input: Some(Box::new(input)), - expr: Some( - proto_converter - .physical_expr_to_proto(exec.predicate(), codec)?, - ), - default_filter_selectivity: exec.default_selectivity() as u32, - projection: match exec.projection() { - None => (0..exec.input().schema().fields().len()) - .map(|i| i as u32) - .collect(), - Some(v) => v.iter().map(|x| *x as u32).collect(), - }, - batch_size: exec.batch_size() as u32, - fetch: exec.fetch().map(|f| f as u32), - }, - ))), - }) + }; + let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); + exec.try_to_proto(&encode_ctx)? + .ok_or_else(|| internal_datafusion_err!("FilterExec is not serializable")) } fn try_from_global_limit_exec( diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index 28e70f2ddfc7..8b7223ddd7c7 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs @@ -1028,6 +1028,39 @@ fn roundtrip_filter_with_fetch() -> Result<()> { roundtrip_test(Arc::new(filter)) } +#[test] +fn roundtrip_filter_projection_states() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Boolean, false), + Field::new("b", DataType::Int64, false), + ])); + let ctx = SessionContext::new(); + let codec = DefaultPhysicalExtensionCodec {}; + let proto_converter = DefaultPhysicalProtoConverter {}; + + for projection in [None, Some(vec![]), Some(vec![0])] { + let filter = FilterExecBuilder::new( + col("a", &schema)?, + Arc::new(EmptyExec::new(Arc::clone(&schema))), + ) + .apply_projection(projection.clone())? + .with_default_selectivity(37) + .with_batch_size(1024) + .with_fetch(Some(5)) + .build()?; + + let result = + roundtrip_test_and_return(Arc::new(filter), &ctx, &codec, &proto_converter)?; + let result = result.downcast_ref::().unwrap(); + assert_eq!(result.projection().as_deref(), projection.as_deref()); + assert_eq!(result.default_selectivity(), 37); + assert_eq!(result.batch_size(), 1024); + assert_eq!(result.fetch(), Some(5)); + } + + Ok(()) +} + #[test] fn roundtrip_sort() -> Result<()> { let field_a = Field::new("a", DataType::Boolean, false);