diff --git a/datafusion/physical-plan/src/buffer.rs b/datafusion/physical-plan/src/buffer.rs index 5879b98c348e3..72df0f5345041 100644 --- a/datafusion/physical-plan/src/buffer.rs +++ b/datafusion/physical-plan/src/buffer.rs @@ -332,13 +332,24 @@ impl ExecutionPlan for BufferExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; + // Destructure exhaustively (no `..`) so that adding a field to + // `BufferExec` is a compile error here until it is either serialized or + // explicitly documented as not needing to be. + let Self { + input, + // Derived from the input's properties at construction time. + properties: _, + capacity, + // Runtime metrics, not part of the plan shape. + metrics: _, + } = self; + let input = ctx.encode_child(input)?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Buffer(Box::new( protobuf::BufferExecNode { input: Some(Box::new(input)), - capacity: self.capacity() as u64, + capacity: *capacity as u64, }, )), ), @@ -363,9 +374,11 @@ impl BufferExec { protobuf::physical_plan_node::PhysicalPlanType::Buffer, "BufferExec", ); - let input = - ctx.decode_required_child(buffer.input.as_deref(), "BufferExec", "input")?; - Ok(Arc::new(BufferExec::new(input, buffer.capacity as usize))) + // Destructure exhaustively so that a new field on `BufferExecNode` is a + // compile error here rather than a silently dropped field. + let protobuf::BufferExecNode { input, capacity } = &**buffer; + let input = ctx.decode_required_child(input.as_deref(), "BufferExec", "input")?; + Ok(Arc::new(BufferExec::new(input, *capacity as usize))) } } diff --git a/datafusion/physical-plan/src/coalesce_batches.rs b/datafusion/physical-plan/src/coalesce_batches.rs index cb0f9b2ce4b36..87957ced7b11c 100644 --- a/datafusion/physical-plan/src/coalesce_batches.rs +++ b/datafusion/physical-plan/src/coalesce_batches.rs @@ -323,14 +323,26 @@ impl ExecutionPlan for CoalesceBatchesExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; + // Destructure exhaustively (no `..`) so that adding a field to + // `CoalesceBatchesExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + input, + target_batch_size, + fetch, + // Runtime metrics, not part of the plan shape. + metrics: _, + // Derived plan properties, recomputed on decode. + cache: _, + } = self; + let input = ctx.encode_child(input)?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::CoalesceBatches( Box::new(protobuf::CoalesceBatchesExecNode { input: Some(Box::new(input)), - target_batch_size: self.target_batch_size() as u32, - fetch: self.fetch().map(|n| n as u32), + target_batch_size: *target_batch_size as u32, + fetch: fetch.map(|n| n as u32), }), ), ), @@ -361,14 +373,19 @@ impl CoalesceBatchesExec { protobuf::physical_plan_node::PhysicalPlanType::CoalesceBatches, "CoalesceBatchesExec", ); - let input = ctx.decode_required_child( - coalesce_batches.input.as_deref(), - "CoalesceBatchesExec", - "input", - )?; + // Destructure exhaustively so that a new field on + // `CoalesceBatchesExecNode` is a compile error here rather than a + // silently dropped field. + let protobuf::CoalesceBatchesExecNode { + input, + target_batch_size, + fetch, + } = &**coalesce_batches; + let input = + ctx.decode_required_child(input.as_deref(), "CoalesceBatchesExec", "input")?; Ok(Arc::new( - CoalesceBatchesExec::new(input, coalesce_batches.target_batch_size as usize) - .with_fetch(coalesce_batches.fetch.map(|f| f as usize)), + CoalesceBatchesExec::new(input, *target_batch_size as usize) + .with_fetch(fetch.map(|f| f as usize)), )) } } diff --git a/datafusion/physical-plan/src/coalesce_partitions.rs b/datafusion/physical-plan/src/coalesce_partitions.rs index 6f58eb2f1e6be..7c8bca772e21d 100644 --- a/datafusion/physical-plan/src/coalesce_partitions.rs +++ b/datafusion/physical-plan/src/coalesce_partitions.rs @@ -384,13 +384,24 @@ impl ExecutionPlan for CoalescePartitionsExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; + // Destructure exhaustively (no `..`) so that adding a field to + // `CoalescePartitionsExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + input, + // Runtime metrics, not part of the plan shape. + metrics: _, + // Derived plan properties, recomputed on decode. + cache: _, + fetch, + } = self; + let input = ctx.encode_child(input)?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Merge(Box::new( protobuf::CoalescePartitionsExecNode { input: Some(Box::new(input)), - fetch: self.fetch().map(|f| f as u32), + fetch: fetch.map(|f| f as u32), }, )), ), @@ -417,14 +428,17 @@ impl CoalescePartitionsExec { protobuf::physical_plan_node::PhysicalPlanType::Merge, "CoalescePartitionsExec", ); + // Destructure exhaustively so that a new field on + // `CoalescePartitionsExecNode` is a compile error here rather than a + // silently dropped field. + let protobuf::CoalescePartitionsExecNode { input, fetch } = &**merge; let input = ctx.decode_required_child( - merge.input.as_deref(), + input.as_deref(), "CoalescePartitionsExec", "input", )?; Ok(Arc::new( - CoalescePartitionsExec::new(input) - .with_fetch(merge.fetch.map(|f| f as usize)), + CoalescePartitionsExec::new(input).with_fetch(fetch.map(|f| f as usize)), )) } } diff --git a/datafusion/physical-plan/src/coop.rs b/datafusion/physical-plan/src/coop.rs index 9e27b26d6e7c9..17166e287e6dc 100644 --- a/datafusion/physical-plan/src/coop.rs +++ b/datafusion/physical-plan/src/coop.rs @@ -402,7 +402,15 @@ impl ExecutionPlan for CooperativeExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; + // Destructure exhaustively (no `..`) so that adding a field to + // `CooperativeExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + input, + // Derived from the input's properties at construction time. + properties: _, + } = self; + let input = ctx.encode_child(input)?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Cooperative(Box::new( @@ -432,11 +440,12 @@ impl CooperativeExec { protobuf::physical_plan_node::PhysicalPlanType::Cooperative, "CooperativeExec", ); - let input = ctx.decode_required_child( - cooperative.input.as_deref(), - "CooperativeExec", - "input", - )?; + // Destructure exhaustively so that a new field on + // `CooperativeExecNode` is a compile error here rather than a silently + // dropped field. + let protobuf::CooperativeExecNode { input } = &**cooperative; + let input = + ctx.decode_required_child(input.as_deref(), "CooperativeExec", "input")?; Ok(Arc::new(CooperativeExec::new(input))) } } diff --git a/datafusion/physical-plan/src/empty.rs b/datafusion/physical-plan/src/empty.rs index dd08ff36a9d88..a1d79890b0815 100644 --- a/datafusion/physical-plan/src/empty.rs +++ b/datafusion/physical-plan/src/empty.rs @@ -214,16 +214,22 @@ impl ExecutionPlan for EmptyExec { _ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let schema = self.schema().as_ref().try_into()?; + // Destructure exhaustively (no `..`) so that adding a field to + // `EmptyExec` is a compile error here until it is either serialized or + // explicitly documented as not needing to be. + let Self { + schema, + partitions, + // Derived from `schema` and `partitions`, recomputed on decode. + cache: _, + } = self; + let schema = schema.as_ref().try_into()?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Empty( protobuf::EmptyExecNode { schema: Some(schema), - partitions: self - .properties() - .output_partitioning() - .partition_count() as u32, + partitions: *partitions as u32, }, ), ), @@ -244,7 +250,10 @@ impl EmptyExec { protobuf::physical_plan_node::PhysicalPlanType::Empty, "EmptyExec", ); - let schema = empty.schema.as_ref().ok_or_else(|| { + // Destructure exhaustively so that a new field on `EmptyExecNode` is a + // compile error here rather than a silently dropped field. + let protobuf::EmptyExecNode { schema, partitions } = empty; + let schema = schema.as_ref().ok_or_else(|| { datafusion_common::internal_datafusion_err!( "EmptyExec is missing required field 'schema'" ) @@ -252,7 +261,7 @@ impl EmptyExec { let schema = Arc::new(arrow::datatypes::Schema::try_from(schema)?); // A zero (absent) partition count comes from a plan encoded before the // field existed, which always meant a single partition. - let partitions = empty.partitions.max(1) as usize; + let partitions = (*partitions).max(1) as usize; Ok(Arc::new(EmptyExec::new(schema).with_partitions(partitions))) } } diff --git a/datafusion/physical-plan/src/explain.rs b/datafusion/physical-plan/src/explain.rs index 3b31ee748b736..7ff44c36d8b48 100644 --- a/datafusion/physical-plan/src/explain.rs +++ b/datafusion/physical-plan/src/explain.rs @@ -215,17 +215,26 @@ impl ExecutionPlan for ExplainExec { ) -> Result> { use datafusion_proto_models::protobuf; + // Destructure exhaustively (no `..`) so that adding a field to + // `ExplainExec` is a compile error here until it is either serialized + // or explicitly documented as not needing to be. + let Self { + schema, + stringified_plans, + verbose, + // Derived from `schema`, recomputed on decode. + cache: _, + } = self; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Explain( protobuf::ExplainExecNode { - schema: Some(self.schema().as_ref().try_into()?), - stringified_plans: self - .stringified_plans() + schema: Some(schema.as_ref().try_into()?), + stringified_plans: stringified_plans .iter() .map(stringified_plan_to_proto) .collect(), - verbose: self.verbose(), + verbose: *verbose, }, ), ), @@ -247,19 +256,25 @@ impl ExplainExec { protobuf::physical_plan_node::PhysicalPlanType::Explain, "ExplainExec", ); - let schema = explain.schema.as_ref().ok_or_else(|| { + // Destructure exhaustively so that a new field on `ExplainExecNode` is + // a compile error here rather than a silently dropped field. + let protobuf::ExplainExecNode { + schema, + stringified_plans, + verbose, + } = explain; + let schema = schema.as_ref().ok_or_else(|| { datafusion_common::internal_datafusion_err!( "ExplainExec is missing required field 'schema'" ) })?; Ok(Arc::new(ExplainExec::new( Arc::new(arrow::datatypes::Schema::try_from(schema)?), - explain - .stringified_plans + stringified_plans .iter() .map(stringified_plan_from_proto) .collect(), - explain.verbose, + *verbose, ))) } } diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 414e5a6d8586a..0ca0c2bc73271 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -850,15 +850,30 @@ impl ExecutionPlan for FilterExec { 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() { + // Destructure exhaustively (no `..`) so that adding a field to + // `FilterExec` is a compile error here until it is either serialized or + // explicitly documented as not needing to be. + let Self { + predicate, + input, + // Runtime metrics, not part of the plan shape. + metrics: _, + default_selectivity, + // Derived plan properties, recomputed on decode. + cache: _, + projection, + batch_size, + fetch, + } = self; + let input_node = ctx.encode_child(input)?; + let expr = ctx.encode_expr(predicate)?; + // The identity projection `[0, 1, ..., num_fields - 1]` is the + // canonical wire representation of a full projection, so `None` is + // encoded that way (and decodes back to `None`). + let projection = if let Some(v) = projection { v.iter().map(|x| *x as u32).collect() } else { - (0..self.input().schema().fields().len()) + (0..input.schema().fields().len()) .map(|i| i as u32) .collect() }; @@ -866,12 +881,12 @@ impl ExecutionPlan for FilterExec { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Filter(Box::new( protobuf::FilterExecNode { - input: Some(Box::new(input)), + input: Some(Box::new(input_node)), expr: Some(expr), - default_filter_selectivity: self.default_selectivity() as u32, + default_filter_selectivity: *default_selectivity as u32, projection, - batch_size: self.batch_size() as u32, - fetch: self.fetch().map(|f| f as u32), + batch_size: *batch_size as u32, + fetch: fetch.map(|f| f as u32), }, )), ), @@ -893,28 +908,37 @@ impl FilterExec { ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let filter = crate::expect_plan_variant!( + let filter_node = crate::expect_plan_variant!( node, protobuf::physical_plan_node::PhysicalPlanType::Filter, "FilterExec", ); - let input = - ctx.decode_required_child(filter.input.as_deref(), "FilterExec", "input")?; + // Destructure exhaustively so that a new field on `FilterExecNode` is a + // compile error here rather than a silently dropped field. + let protobuf::FilterExecNode { + input, + expr, + default_filter_selectivity, + projection, + batch_size, + fetch, + } = &**filter_node; + let input = ctx.decode_required_child(input.as_deref(), "FilterExec", "input")?; let predicate = ctx.decode_required_expr( - filter.expr.as_ref(), + expr.as_ref(), input.schema().as_ref(), "FilterExec", "expr", )?; - let filter_selectivity = filter.default_filter_selectivity.try_into(); + let filter_selectivity = (*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 mut is_full_projection = projection.len() == num_fields; + let mut projection_vec: Vec = Vec::with_capacity(projection.len()); + for (i, idx) in projection.iter().enumerate() { let idx = *idx as usize; is_full_projection &= idx == i; projection_vec.push(idx); @@ -926,8 +950,8 @@ impl FilterExec { }; 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)) + .with_batch_size(*batch_size as usize) + .with_fetch(fetch.map(|f| f as usize)) .build()?; match filter_selectivity { Ok(filter_selectivity) => Ok(Arc::new( diff --git a/datafusion/physical-plan/src/limit.rs b/datafusion/physical-plan/src/limit.rs index dd62c93d1cfe0..4e5d9200d9df9 100644 --- a/datafusion/physical-plan/src/limit.rs +++ b/datafusion/physical-plan/src/limit.rs @@ -278,19 +278,30 @@ impl ExecutionPlan for GlobalLimitExec { ) -> Result> { use datafusion_physical_expr_common::sort_expr::optional_ordering_try_to_proto; use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; - let required_ordering = optional_ordering_try_to_proto( - self.required_ordering.as_ref(), - &ctx.expr_ctx(), - )?; + // Destructure exhaustively (no `..`) so that adding a field to + // `GlobalLimitExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + input, + skip, + fetch, + required_ordering, + // Runtime metrics, not part of the plan shape. + metrics: _, + // Derived plan properties, recomputed on decode. + cache: _, + } = self; + let input = ctx.encode_child(input)?; + let required_ordering = + optional_ordering_try_to_proto(required_ordering.as_ref(), &ctx.expr_ctx())?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::GlobalLimit(Box::new( protobuf::GlobalLimitExecNode { input: Some(Box::new(input)), - skip: self.skip() as u32, - fetch: match self.fetch() { - Some(n) => n as i64, + skip: *skip as u32, + fetch: match fetch { + Some(n) => *n as i64, _ => -1, // no limit }, required_ordering, @@ -314,21 +325,27 @@ impl GlobalLimitExec { protobuf::physical_plan_node::PhysicalPlanType::GlobalLimit, "GlobalLimitExec", ); - let input = ctx.decode_required_child( - limit.input.as_deref(), - "GlobalLimitExec", - "input", - )?; - let fetch = if limit.fetch >= 0 { - Some(limit.fetch as usize) + // Destructure exhaustively so that a new field on + // `GlobalLimitExecNode` is a compile error here rather than a silently + // dropped field. + let protobuf::GlobalLimitExecNode { + input, + skip, + fetch, + required_ordering, + } = &**limit; + let input = + ctx.decode_required_child(input.as_deref(), "GlobalLimitExec", "input")?; + let fetch = if *fetch >= 0 { + Some(*fetch as usize) } else { None }; let required_ordering = optional_ordering_try_from_proto( - &limit.required_ordering, + required_ordering, &ctx.expr_ctx(input.schema().as_ref()), )?; - let mut exec = GlobalLimitExec::new(input, limit.skip as usize, fetch); + let mut exec = GlobalLimitExec::new(input, *skip as usize, fetch); exec.set_required_ordering(required_ordering); Ok(Arc::new(exec)) } @@ -538,17 +555,27 @@ impl ExecutionPlan for LocalLimitExec { ) -> Result> { use datafusion_physical_expr_common::sort_expr::optional_ordering_try_to_proto; use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; - let required_ordering = optional_ordering_try_to_proto( - self.required_ordering.as_ref(), - &ctx.expr_ctx(), - )?; + // Destructure exhaustively (no `..`) so that adding a field to + // `LocalLimitExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + input, + fetch, + required_ordering, + // Runtime metrics, not part of the plan shape. + metrics: _, + // Derived plan properties, recomputed on decode. + cache: _, + } = self; + let input = ctx.encode_child(input)?; + let required_ordering = + optional_ordering_try_to_proto(required_ordering.as_ref(), &ctx.expr_ctx())?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::LocalLimit(Box::new( protobuf::LocalLimitExecNode { input: Some(Box::new(input)), - fetch: self.fetch() as u32, + fetch: *fetch as u32, required_ordering, }, )), @@ -570,13 +597,20 @@ impl LocalLimitExec { protobuf::physical_plan_node::PhysicalPlanType::LocalLimit, "LocalLimitExec", ); + // Destructure exhaustively so that a new field on `LocalLimitExecNode` + // is a compile error here rather than a silently dropped field. + let protobuf::LocalLimitExecNode { + input, + fetch, + required_ordering, + } = &**limit; let input = - ctx.decode_required_child(limit.input.as_deref(), "LocalLimitExec", "input")?; + ctx.decode_required_child(input.as_deref(), "LocalLimitExec", "input")?; let required_ordering = optional_ordering_try_from_proto( - &limit.required_ordering, + required_ordering, &ctx.expr_ctx(input.schema().as_ref()), )?; - let mut exec = LocalLimitExec::new(input, limit.fetch as usize); + let mut exec = LocalLimitExec::new(input, *fetch as usize); exec.set_required_ordering(required_ordering); Ok(Arc::new(exec)) } diff --git a/datafusion/physical-plan/src/placeholder_row.rs b/datafusion/physical-plan/src/placeholder_row.rs index 67c063b65cbc6..33833cd1e7811 100644 --- a/datafusion/physical-plan/src/placeholder_row.rs +++ b/datafusion/physical-plan/src/placeholder_row.rs @@ -214,16 +214,22 @@ impl ExecutionPlan for PlaceholderRowExec { _ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let schema = self.schema().as_ref().try_into()?; + // Destructure exhaustively (no `..`) so that adding a field to + // `PlaceholderRowExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + schema, + partitions, + // Derived from `schema` and `partitions`, recomputed on decode. + cache: _, + } = self; + let schema = schema.as_ref().try_into()?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::PlaceholderRow( protobuf::PlaceholderRowExecNode { schema: Some(schema), - partitions: self - .properties() - .output_partitioning() - .partition_count() as u32, + partitions: *partitions as u32, }, ), ), @@ -244,7 +250,11 @@ impl PlaceholderRowExec { protobuf::physical_plan_node::PhysicalPlanType::PlaceholderRow, "PlaceholderRowExec", ); - let schema = placeholder.schema.as_ref().ok_or_else(|| { + // Destructure exhaustively so that a new field on + // `PlaceholderRowExecNode` is a compile error here rather than a + // silently dropped field. + let protobuf::PlaceholderRowExecNode { schema, partitions } = placeholder; + let schema = schema.as_ref().ok_or_else(|| { datafusion_common::internal_datafusion_err!( "PlaceholderRowExec is missing required field 'schema'" ) @@ -252,7 +262,7 @@ impl PlaceholderRowExec { let schema = Arc::new(Schema::try_from(schema)?); // A zero (absent) partition count comes from a plan encoded before the // field existed, which always meant a single partition. - let partitions = placeholder.partitions.max(1) as usize; + let partitions = (*partitions).max(1) as usize; Ok(Arc::new( PlaceholderRowExec::new(schema).with_partitions(partitions), )) diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index cf362cdee55d3..ee6f7dd152f58 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -563,9 +563,23 @@ impl ExecutionPlan for ProjectionExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; - let expr = ctx.encode_expressions(self.expr().iter().map(|p| &p.expr))?; - let expr_name = self.expr().iter().map(|p| p.alias.clone()).collect(); + // Destructure exhaustively (no `..`) so that adding a field to + // `ProjectionExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + // The projector is rebuilt from the projection expressions and the + // input schema on decode; the expressions themselves are serialized. + projector, + input, + // Runtime metrics, not part of the plan shape. + metrics: _, + // Derived plan properties, recomputed on decode. + cache: _, + } = self; + let projection_exprs = projector.projection().as_ref(); + let input = ctx.encode_child(input)?; + let expr = ctx.encode_expressions(projection_exprs.iter().map(|p| &p.expr))?; + let expr_name = projection_exprs.iter().map(|p| p.alias.clone()).collect(); Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Projection(Box::new( @@ -602,16 +616,19 @@ impl ProjectionExec { protobuf::physical_plan_node::PhysicalPlanType::Projection, "ProjectionExec", ); - let input = ctx.decode_required_child( - projection.input.as_deref(), - "ProjectionExec", - "input", - )?; + // Destructure exhaustively so that a new field on `ProjectionExecNode` + // is a compile error here rather than a silently dropped field. + let protobuf::ProjectionExecNode { + input, + expr, + expr_name, + } = &**projection; + let input = + ctx.decode_required_child(input.as_deref(), "ProjectionExec", "input")?; let input_schema = input.schema(); - let exprs = projection - .expr + let exprs = expr .iter() - .zip(projection.expr_name.iter()) + .zip(expr_name.iter()) .map(|(expr, name)| { Ok(ProjectionExpr { expr: ctx.decode_expr(expr, input_schema.as_ref())?, diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs index 033498799449a..447133e44c5a3 100644 --- a/datafusion/physical-plan/src/repartition/mod.rs +++ b/datafusion/physical-plan/src/repartition/mod.rs @@ -1955,9 +1955,25 @@ impl ExecutionPlan for RepartitionExec { &self, ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { - let input = ctx.encode_child(self.input())?; + // Destructure exhaustively (no `..`) so that adding a field to + // `RepartitionExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + input, + // Execution-time channel state, created on `execute()`. + state: _, + // Runtime metrics, not part of the plan shape. + metrics: _, + preserve_order, + // Derived plan properties. The output partitioning lives here (it is + // the plan's own `partitioning`) and *is* serialized below; the rest + // is recomputed on decode. + cache, + } = self; - let partitioning = self.partitioning().try_to_proto(&ctx.expr_ctx())?; + let input = ctx.encode_child(input)?; + + let partitioning = cache.partitioning.try_to_proto(&ctx.expr_ctx())?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( @@ -1965,7 +1981,7 @@ impl ExecutionPlan for RepartitionExec { protobuf::RepartitionExecNode { input: Some(Box::new(input)), partitioning: Some(partitioning), - preserve_order: self.preserve_order(), + preserve_order: *preserve_order, }, )), ), @@ -1985,15 +2001,19 @@ impl RepartitionExec { protobuf::physical_plan_node::PhysicalPlanType::Repartition, "RepartitionExec", ); - let input = ctx.decode_required_child( - repart.input.as_deref(), - "RepartitionExec", - "input", - )?; + // Destructure exhaustively so that a new field on + // `RepartitionExecNode` is a compile error here rather than a silently + // dropped field. + let protobuf::RepartitionExecNode { + input, + partitioning, + preserve_order, + } = &**repart; + let input = + ctx.decode_required_child(input.as_deref(), "RepartitionExec", "input")?; let input_schema = input.schema(); - let partitioning = repart - .partitioning + let partitioning = partitioning .as_ref() .map(|partitioning| { Partitioning::try_from_proto( @@ -2010,7 +2030,7 @@ impl RepartitionExec { })?; let mut repart_exec = RepartitionExec::try_new(input, partitioning)?; - if repart.preserve_order { + if *preserve_order { repart_exec = repart_exec.with_preserve_order(); } Ok(Arc::new(repart_exec)) diff --git a/datafusion/physical-plan/src/scalar_subquery.rs b/datafusion/physical-plan/src/scalar_subquery.rs index f2b7c5e0b53e9..cac8f925039ae 100644 --- a/datafusion/physical-plan/src/scalar_subquery.rs +++ b/datafusion/physical-plan/src/scalar_subquery.rs @@ -285,10 +285,29 @@ impl ExecutionPlan for ScalarSubqueryExec { ) -> Result> { use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; + // Destructure exhaustively (no `..`) so that adding a field to + // `ScalarSubqueryExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + input, + subqueries, + // Execution-time one-shot future, created on `execute()`. + subquery_future: _, + // Runtime results container, rebuilt (empty) on decode and shared + // with the input's `ScalarSubqueryExpr` nodes. + results: _, + // Copied from the input's properties, recomputed on decode. + cache: _, + } = self; + let input = ctx.encode_child(input)?; // Subquery indices are positional and recovered during decoding. - let subqueries = - ctx.encode_children(self.subqueries().iter().map(|subquery| &subquery.plan))?; + let subqueries = ctx.encode_children(subqueries.iter().map( + |ScalarSubqueryLink { + plan, + // Positional: recovered from the element's position on decode. + index: _, + }| plan, + ))?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::ScalarSubquery(Box::new( @@ -316,8 +335,12 @@ impl ScalarSubqueryExec { protobuf::physical_plan_node::PhysicalPlanType::ScalarSubquery, "ScalarSubqueryExec", ); - let results = ScalarSubqueryResults::new(scalar_subquery.subqueries.len()); - let input_node = scalar_subquery.input.as_deref().ok_or_else(|| { + // Destructure exhaustively so that a new field on + // `ScalarSubqueryExecNode` is a compile error here rather than a + // silently dropped field. + let protobuf::ScalarSubqueryExecNode { input, subqueries } = &**scalar_subquery; + let results = ScalarSubqueryResults::new(subqueries.len()); + let input_node = input.as_deref().ok_or_else(|| { datafusion_common::internal_datafusion_err!( "ScalarSubqueryExec is missing required field 'input'" ) @@ -325,8 +348,7 @@ impl ScalarSubqueryExec { // The input's ScalarSubqueryExpr nodes must share this results container. let input = ctx.decode_child_with_scalar_subquery_results(input_node, results.clone())?; - let subqueries = scalar_subquery - .subqueries + let subqueries = subqueries .iter() .enumerate() .map(|(index, plan)| { diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs index bd47668cf38ba..c91df46351e24 100644 --- a/datafusion/physical-plan/src/sorts/sort.rs +++ b/datafusion/physical-plan/src/sorts/sort.rs @@ -1603,9 +1603,24 @@ impl ExecutionPlan for SortExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; - let expr = self - .expr() + // Destructure exhaustively (no `..`) so that adding a field to + // `SortExec` is a compile error here until it is either serialized or + // explicitly documented as not needing to be. + let Self { + input, + expr, + // Runtime metrics, not part of the plan shape. + metrics_set: _, + preserve_partitioning, + fetch, + // Derived from `input` and `expr` at construction time. + common_sort_prefix: _, + // Derived plan properties, recomputed on decode. + cache: _, + filter, + } = self; + let input = ctx.encode_child(input)?; + let expr = expr .iter() .map(|sort_expr| { let sort_node = Box::new(protobuf::PhysicalSortExprNode { @@ -1621,11 +1636,12 @@ impl ExecutionPlan for SortExec { }) }) .collect::>>()?; - let dynamic_filter = self - .dynamic_expressions_produced() - .into_iter() - .next() - .map(|expr| ctx.encode_expr(&expr)) + let dynamic_filter = filter + .as_ref() + .map(|filter| { + let df_expr: Arc = filter.read().expr(); + ctx.encode_expr(&df_expr) + }) .transpose()?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( @@ -1633,11 +1649,11 @@ impl ExecutionPlan for SortExec { protobuf::SortExecNode { input: Some(Box::new(input)), expr, - fetch: match self.fetch() { - Some(n) => n as i64, + fetch: match fetch { + Some(n) => *n as i64, None => -1, }, - preserve_partitioning: self.preserve_partitioning(), + preserve_partitioning: *preserve_partitioning, dynamic_filter, }, )), @@ -1659,11 +1675,18 @@ impl SortExec { protobuf::physical_plan_node::PhysicalPlanType::Sort, "SortExec", ); - let input = - ctx.decode_required_child(sort.input.as_deref(), "SortExec", "input")?; + // Destructure exhaustively so that a new field on `SortExecNode` is a + // compile error here rather than a silently dropped field. + let protobuf::SortExecNode { + input, + expr, + fetch, + preserve_partitioning, + dynamic_filter, + } = &**sort; + let input = ctx.decode_required_child(input.as_deref(), "SortExec", "input")?; let input_schema = input.schema(); - let exprs = sort - .expr + let exprs = expr .iter() .map(|expr| { let Some(ExprType::Sort(sort_expr)) = expr.expr_type.as_ref() else { @@ -1688,12 +1711,12 @@ impl SortExec { let Some(ordering) = LexOrdering::new(exprs) else { return datafusion_common::internal_err!("SortExec requires an ordering"); }; - let fetch = (sort.fetch >= 0).then_some(sort.fetch as usize); + let fetch = (*fetch >= 0).then_some(*fetch as usize); let new_sort = SortExec::new(ordering, input) .with_fetch(fetch) - .with_preserve_partitioning(sort.preserve_partitioning); + .with_preserve_partitioning(*preserve_partitioning); - let new_sort = if let Some(df_proto) = &sort.dynamic_filter { + let new_sort = if let Some(df_proto) = dynamic_filter { let df_expr = ctx.decode_expr(df_proto, new_sort.input().schema().as_ref())?; let df = (df_expr as Arc) diff --git a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs index 09c14890027a0..7091c9785b4b7 100644 --- a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs +++ b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs @@ -473,9 +473,24 @@ impl ExecutionPlan for SortPreservingMergeExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let input = ctx.encode_child(self.input())?; - let expr = self - .expr() + // Destructure exhaustively (no `..`) so that adding a field to + // `SortPreservingMergeExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + input, + expr, + // Runtime metrics, not part of the plan shape. + metrics: _, + fetch, + // Derived plan properties, recomputed on decode. + cache: _, + // Not serialized: `SortPreservingMergeExecNode` has no field for it, + // so decoding always yields the `true` default from + // `SortPreservingMergeExec::new`. + enable_round_robin_repartition: _, + } = self; + let input = ctx.encode_child(input)?; + let expr = expr .iter() .map(|e| { Ok(protobuf::PhysicalExprNode { @@ -496,7 +511,7 @@ impl ExecutionPlan for SortPreservingMergeExec { Box::new(protobuf::SortPreservingMergeExecNode { input: Some(Box::new(input)), expr, - fetch: self.fetch().map(|f| f as i64).unwrap_or(-1), + fetch: fetch.map(|f| f as i64).unwrap_or(-1), }), ), ), @@ -518,14 +533,17 @@ impl SortPreservingMergeExec { protobuf::physical_plan_node::PhysicalPlanType::SortPreservingMerge, "SortPreservingMergeExec", ); + // Destructure exhaustively so that a new field on + // `SortPreservingMergeExecNode` is a compile error here rather than a + // silently dropped field. + let protobuf::SortPreservingMergeExecNode { input, expr, fetch } = &**spm; let input = ctx.decode_required_child( - spm.input.as_deref(), + input.as_deref(), "SortPreservingMergeExec", "input", )?; let input_schema = input.schema(); - let exprs = spm - .expr + let exprs = expr .iter() .map(|e| { let sort = match &e.expr_type { @@ -554,7 +572,7 @@ impl SortPreservingMergeExec { let Some(ordering) = LexOrdering::new(exprs) else { return internal_err!("SortPreservingMergeExec requires an ordering"); }; - let fetch = (spm.fetch >= 0).then_some(spm.fetch as usize); + let fetch = (*fetch >= 0).then_some(*fetch as usize); Ok(Arc::new( SortPreservingMergeExec::new(ordering, input).with_fetch(fetch), )) diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs index 160772dc22314..557d8f21d136b 100644 --- a/datafusion/physical-plan/src/union.rs +++ b/datafusion/physical-plan/src/union.rs @@ -574,7 +574,17 @@ impl ExecutionPlan for UnionExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let inputs = ctx.encode_children(self.inputs())?; + // Destructure exhaustively (no `..`) so that adding a field to + // `UnionExec` is a compile error here until it is either serialized or + // explicitly documented as not needing to be. + let Self { + inputs, + // Runtime metrics, not part of the plan shape. + metrics: _, + // Derived plan properties, recomputed on decode. + cache: _, + } = self; + let inputs = ctx.encode_children(inputs)?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Union( @@ -597,8 +607,10 @@ impl UnionExec { protobuf::physical_plan_node::PhysicalPlanType::Union, "UnionExec", ); - let inputs = union - .inputs + // Destructure exhaustively so that a new field on `UnionExecNode` is a + // compile error here rather than a silently dropped field. + let protobuf::UnionExecNode { inputs } = union; + let inputs = inputs .iter() .map(|input| ctx.decode_child(input)) .collect::>>()?; @@ -854,7 +866,17 @@ impl ExecutionPlan for InterleaveExec { ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, ) -> Result> { use datafusion_proto_models::protobuf; - let inputs = ctx.encode_children(self.inputs())?; + // Destructure exhaustively (no `..`) so that adding a field to + // `InterleaveExec` is a compile error here until it is either + // serialized or explicitly documented as not needing to be. + let Self { + inputs, + // Runtime metrics, not part of the plan shape. + metrics: _, + // Derived plan properties, recomputed on decode. + cache: _, + } = self; + let inputs = ctx.encode_children(inputs)?; Ok(Some(protobuf::PhysicalPlanNode { physical_plan_type: Some( protobuf::physical_plan_node::PhysicalPlanType::Interleave( @@ -877,8 +899,10 @@ impl InterleaveExec { protobuf::physical_plan_node::PhysicalPlanType::Interleave, "InterleaveExec", ); - let inputs = interleave - .inputs + // Destructure exhaustively so that a new field on `InterleaveExecNode` + // is a compile error here rather than a silently dropped field. + let protobuf::InterleaveExecNode { inputs } = interleave; + let inputs = inputs .iter() .map(|input| ctx.decode_child(input)) .collect::>>()?;