You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/21 13:34:03 UTC
(datafusion) branch gh-readonly-queue/main/pr-24167-e2bdc2c21889ee4706071708ab6fe9b6b6f9fe5e created (now a4e071d17a)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24167-e2bdc2c21889ee4706071708ab6fe9b6b6f9fe5e
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at a4e071d17a refactor(proto): destructure plan and proto structs in core plan serde hooks (#24167)
This branch includes the following new commits:
new a4e071d17a refactor(proto): destructure plan and proto structs in core plan serde hooks (#24167)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: refactor(proto): destructure plan and proto structs in core plan serde hooks (#24167)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24167-e2bdc2c21889ee4706071708ab6fe9b6b6f9fe5e
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit a4e071d17a4d691fe4764a2a5180b94412667240
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Fri Aug 21 13:33:39 2026 +0000
refactor(proto): destructure plan and proto structs in core plan serde hooks (#24167)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Follow-up cleanup for EPIC #23494; does not close an issue on its own.
## Rationale for this change
EPIC #23494 moved every built-in `ExecutionPlan` off the central
`downcast_ref`
chain in `datafusion-proto` onto per-plan hooks: a `try_to_proto`
override on
the `ExecutionPlan` trait and an inherent `FooExec::try_from_proto`,
both living
in the plan's own module.
Those hooks currently read plan state through **getters**
(`self.expr()`,
`self.fetch()`, ...). That means adding a field to a plan struct is
invisible to
serialization: nothing breaks, nothing warns, the field is just silently
dropped on round-trip. This is not hypothetical — it has already cost us
a real
bug: `HashJoinExec.fetch` is not serialized and is lost on round-trip
(being
fixed separately).
Starting each hook with an *exhaustive* destructure turns that class of
mistake
into a compile error:
* **Encode side** — destructure `self` naming every field, no `..`. Add
a field
to `SortExec` and `SortExec::try_to_proto` stops compiling until you
either
serialize it or bind it to `_` with a comment saying why it does not
need to
be (derived at construction, runtime state, recomputed on decode).
* **Decode side** — destructure the prost node struct the same way. The
generated structs are plain, all-`pub` and not `#[non_exhaustive]`, so
adding
a field to the `.proto` becomes a compile error in every decoder rather
than a
silent omission.
## What changes are included in this PR?
A mechanical, behavior-preserving refactor of the serde hooks for the
core
(non-join, non-aggregate, non-window) plans in
`datafusion/physical-plan/src/`:
`SortExec`, `SortPreservingMergeExec`, `GlobalLimitExec`,
`LocalLimitExec`,
`FilterExec`, `ProjectionExec`, `RepartitionExec`, `UnionExec`,
`InterleaveExec`, `CoalesceBatchesExec`, `CoalescePartitionsExec`,
`CooperativeExec`, `BufferExec`, `EmptyExec`, `PlaceholderRowExec`,
`ExplainExec`, `ScalarSubqueryExec`.
**The wire format is unchanged — byte for byte — and no behavior
changes.** The
same values are written to and read from the same proto fields; only the
way
they are reached in Rust changes.
Split into four commits, one per group of plans, each of which builds
green.
### Fields the refactor documented as not serialized
Writing the destructures surfaced and pinned down, in comments at the
point of
use, which fields do not survive a round-trip today. All of these are
pre-existing; **none of them are changed here**, because fixing any of
them
means changing the wire format, which is a separate decision:
* `SortPreservingMergeExec::enable_round_robin_repartition` — no field
on
`SortPreservingMergeExecNode`; decoding always restores the `true`
default
from `SortPreservingMergeExec::new`.
* `GlobalLimitExec::required_ordering` and
`LocalLimitExec::required_ordering` —
no field on the limit nodes; they are set by the `enforce_sorting`
optimizer
rule, so a decoded plan starts with `None`.
Everything else that binds to `_` is genuinely not plan shape: runtime
metrics
(`metrics` / `metrics_set`), execution-time state
(`RepartitionExec::state`,
`ScalarSubqueryExec::subquery_future` and `results`), values derived at
construction from other serialized fields
(`SortExec::common_sort_prefix`), the
positional `ScalarSubqueryLink::index`, and the `cache` / `properties`
plan
properties that are recomputed on decode.
Two small notes for reviewers:
* `RepartitionExec` keeps its output partitioning *inside* `cache`
(`partitioning()` is `&self.cache.partitioning`), so `cache` is bound
rather
than `_`-ed there and is read for the serialized `partitioning` field.
* `EmptyExec` / `PlaceholderRowExec` now read the encoded partition
count from
the `partitions` field instead of going through
`properties().output_partitioning().partition_count()`. The two are kept
in
sync by `with_partitions`, the only writer besides `new`, so the encoded
value
is identical.
## Are these changes tested?
Covered by the existing round-trip tests, which are the real proof that
the wire
format did not move. Locally on this branch:
* `cargo test -p datafusion-proto --test proto_integration` — 214
passed, 0 failed
* `cargo test -p datafusion-physical-plan` — 1640 passed, 0 failed
* `cargo clippy -p datafusion-physical-plan --all-targets --all-features
-- -D warnings` — clean
* `cargo fmt --all`
No new tests are added here: the change adds no new behavior to test,
and the
compile-error property it introduces is enforced by rustc rather than by
a test.
**On the comprehensiveness of that existing coverage** (thanks
@andygrove for
pushing on this — the original wording oversold it):
* `SortPreservingMergeExec`, updated by this PR, has **no** round-trip
test at
all — `SortPreservingMerge` does not appear anywhere in
`roundtrip_physical_plan.rs`. The other 16 plans touched here each have
at
least one dedicated `roundtrip_*` test.
* Separately, `roundtrip_test` asserts on `format!("{plan:?}")`, so it
cannot
catch a dropped field that a plan's `Debug` impl doesn't print. That is
precisely how the `HashJoinExec::fetch` bug (#24165) went unnoticed.
Both are now tracked in #24171, with the test work split out into its
own PR so
this mechanical refactor and the coverage improvements can be reviewed
independently. Sequencing is up to reviewers — happy to land the tests
first if
that's preferred.
## Are there any user-facing changes?
No. No public API changes, no wire format changes, no behavior changes.
---------
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
---
datafusion/physical-plan/src/buffer.rs | 23 ++++--
datafusion/physical-plan/src/coalesce_batches.rs | 37 +++++++---
.../physical-plan/src/coalesce_partitions.rs | 24 ++++--
datafusion/physical-plan/src/coop.rs | 21 ++++--
datafusion/physical-plan/src/empty.rs | 23 ++++--
datafusion/physical-plan/src/explain.rs | 31 ++++++--
datafusion/physical-plan/src/filter.rs | 66 +++++++++++------
datafusion/physical-plan/src/limit.rs | 86 +++++++++++++++-------
datafusion/physical-plan/src/placeholder_row.rs | 24 ++++--
datafusion/physical-plan/src/projection.rs | 39 +++++++---
datafusion/physical-plan/src/repartition/mod.rs | 42 ++++++++---
datafusion/physical-plan/src/scalar_subquery.rs | 36 +++++++--
datafusion/physical-plan/src/sorts/sort.rs | 59 ++++++++++-----
.../src/sorts/sort_preserving_merge.rs | 34 +++++++--
datafusion/physical-plan/src/union.rs | 36 +++++++--
15 files changed, 425 insertions(+), 156 deletions(-)
diff --git a/datafusion/physical-plan/src/buffer.rs b/datafusion/physical-plan/src/buffer.rs
index 5879b98c34..72df0f5345 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 cb0f9b2ce4..87957ced7b 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 6f58eb2f1e..7c8bca772e 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 9e27b26d6e..17166e287e 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 dd08ff36a9..a1d79890b0 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 3b31ee748b..7ff44c36d8 100644
--- a/datafusion/physical-plan/src/explain.rs
+++ b/datafusion/physical-plan/src/explain.rs
@@ -215,17 +215,26 @@ impl ExecutionPlan for ExplainExec {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 a1be69849b..186c49562b 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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<Arc<dyn ExecutionPlan>> {
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<usize> = 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<usize> = 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 dd62c93d1c..4e5d9200d9 100644
--- a/datafusion/physical-plan/src/limit.rs
+++ b/datafusion/physical-plan/src/limit.rs
@@ -278,19 +278,30 @@ impl ExecutionPlan for GlobalLimitExec {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 67c063b65c..33833cd1e7 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 1ec278eb9c..2a24eb60e6 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -619,9 +619,23 @@ impl ExecutionPlan for ProjectionExec {
ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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(
@@ -658,16 +672,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 11ddcec24d..8f4b8558a5 100644
--- a/datafusion/physical-plan/src/repartition/mod.rs
+++ b/datafusion/physical-plan/src/repartition/mod.rs
@@ -1960,9 +1960,25 @@ impl ExecutionPlan for RepartitionExec {
&self,
ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
) -> Result<Option<protobuf::PhysicalPlanNode>> {
- 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(
@@ -1970,7 +1986,7 @@ impl ExecutionPlan for RepartitionExec {
protobuf::RepartitionExecNode {
input: Some(Box::new(input)),
partitioning: Some(partitioning),
- preserve_order: self.preserve_order(),
+ preserve_order: *preserve_order,
},
)),
),
@@ -1990,15 +2006,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(
@@ -2015,7 +2035,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 f2b7c5e0b5..cac8f92503 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 bd47668cf3..c91df46351 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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::<Result<Vec<_>>>()?;
- 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<dyn PhysicalExpr> = 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<dyn std::any::Any + Send + Sync>)
diff --git a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs
index e088d92d65..9fadeb972d 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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 Some(protobuf::physical_expr_node::ExprType::Sort(sort)) =
@@ -553,7 +571,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 160772dc22..557d8f21d1 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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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::<Result<Vec<_>>>()?;
@@ -854,7 +866,17 @@ impl ExecutionPlan for InterleaveExec {
ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
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::<Result<Vec<_>>>()?;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org