From 35604af090d457449b57fe139eb5350a3ab35eaf Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Mon, 20 Jul 2026 11:52:18 +0800 Subject: [PATCH 1/2] refactor(proto): migrate FilterExec serde The centralized protobuf dispatcher couples each execution plan to datafusion-proto. Adding plan serialization therefore requires changes outside the plan implementation. Move FilterExec encoding and decoding behind its ExecutionPlan hooks. Preserve the existing wire format and retain the old helper methods as deprecated compatibility APIs. CLOSES #23499 Signed-off-by: Jiawei Zhao --- datafusion/physical-plan/src/filter.rs | 95 +++++++++++++++++++ datafusion/proto/src/physical_plan/mod.rs | 20 ++-- .../tests/cases/roundtrip_physical_plan.rs | 33 +++++++ 3 files changed, 138 insertions(+), 10 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index ad40a3fb5fd83..d367be16eb6ed 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 1ef82375952c0..72d66e0fb1c0f 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -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,6 +1206,10 @@ 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, @@ -2963,6 +2959,10 @@ 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, diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index 28e70f2ddfc7e..8b7223ddd7c72 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); From 7172823888e69d79cad789b0625233db5d91737e Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Tue, 21 Jul 2026 13:13:44 +0800 Subject: [PATCH 2/2] refactor(proto): delegate deprecated Filter serde helpers FilterExec serde moved to per-plan hooks, but the deprecated PhysicalPlanNodeExt helpers retained separate implementations. Keeping the wire-format logic in both paths allows the compatibility API to drift from normal dispatch. Build the existing encode and decode adapter contexts in the deprecated helpers and forward to FilterExec's canonical hooks, preserving the old API without duplicating serialization logic. Refs #23499 Signed-off-by: Jiawei Zhao --- datafusion/proto/src/physical_plan/mod.rs | 84 ++++------------------- 1 file changed, 14 insertions(+), 70 deletions(-) diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index 72d66e0fb1c0f..86496300020cb 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, @@ -1216,53 +1216,15 @@ pub trait PhysicalPlanNodeExt: Sized { 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( @@ -2968,31 +2930,13 @@ pub trait PhysicalPlanNodeExt: Sized { 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(