Skip to content

Commit 24d8957

Browse files
authored
perf(physical-plan): fold PlanProperties fast-path into with_new_children_if_necessary (PR 1 of #22555) (#23332)
## Which issue does this PR close? Part of #22555. This is **PR 1 of 2** — see the issue body for the full plan. PR 2 will audit direct `with_new_children` callers and add a clippy lint. ## Rationale for this change Today the "skip work when children are unchanged" intent is split across two layers: - **caller-side** — [`with_new_children_if_necessary`](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/execution_plan.rs) short-circuits via `Arc::ptr_eq` on child pointers. - **callee-side** — the `check_if_same_properties!` macro from #19792, invoked inside each impl's `with_new_children`, short-circuits when children's `PlanProperties` Arcs match (allowing the plan to reuse its cached `PlanProperties` Arc instead of recomputing). Having two independent layers means two places to maintain and two places for future changes to drift apart. This PR consolidates the fast-path into the single free-function helper so callers get both short-circuits uniformly. ## What changes are included in this PR? `with_new_children_if_necessary` now applies **three layers**, cheapest first: 1. **Same child pointers** — every `children[i]` is `Arc::ptr_eq` to the corresponding existing child → return the original plan unchanged, no allocation. 2. **Same child properties** — children's `PlanProperties` Arcs match → call the new [`ExecutionPlan::with_new_children_and_same_properties`](#) trait method to reuse the plan's `PlanProperties` cache without recomputing. 3. **Full recompute** — otherwise, delegate to `ExecutionPlan::with_new_children`. To make layer 2 dispatchable via `&dyn ExecutionPlan`, `with_new_children_and_same_properties` is promoted from an ad-hoc inherent method on each impl to a **trait method** with a safe default that falls back to `with_new_children`. All 22 existing impls migrate their inherent method to a trait override (mechanical change — signature `&self → self: Arc<Self>`, return `Self → Result<Arc<dyn ExecutionPlan>>`, body wrapped in `Ok(Arc::new(...))`). The `check_if_same_properties!` macro and its call sites inside impls are **kept**, so direct callers of `with_new_children` (which PR 2 will audit + migrate) do not regress on this PR. ## Are these changes tested? Yes — added `test_with_new_children_if_necessary_layers` in `execution_plan.rs` that constructs test-local `WithChildrenTestLeaf` + `WithChildrenTestParent` plans (the parent tracks recompute vs fast-path calls via `AtomicUsize`) and asserts, for each of the three layers: - **Layer 1**: `Arc::ptr_eq(result, parent)` returns true, `recompute_calls == 0`, `fast_path_calls == 0` - **Layer 2**: `Arc::ptr_eq(result.properties(), orig_props)` returns true, `recompute_calls == 0`, `fast_path_calls == 1` - **Layer 3**: `Arc::ptr_eq(result.properties(), orig_props)` returns false, `recompute_calls == 1`, `fast_path_calls` unchanged All 1523 `datafusion-physical-plan` unit tests pass. Full workspace `cargo check` + `cargo clippy --all-targets --all-features -- -D warnings` pass. ## Are there any user-facing changes? Yes — `ExecutionPlan` gains a new default-implemented trait method `with_new_children_and_same_properties`. Downstream impls that used to override the ad-hoc inherent method with the same name will need to re-implement as a trait override (mechanical signature change). Marking as `api change`. ## Follow-up (PR 2, not in this PR) - Audit the ~47 remaining direct callers of `plan.with_new_children(children)` across the codebase and route them through `with_new_children_if_necessary`. - Add a `disallowed_methods` clippy lint (or custom lint) that forbids direct `ExecutionPlan::with_new_children` outside of a small allow-list. - Once all callers migrate, remove the `check_if_same_properties!` macro and its impl-side invocations, making the helper the single source of truth as described in the issue.
1 parent 7ac784f commit 24d8957

22 files changed

Lines changed: 682 additions & 305 deletions

datafusion/physical-plan/src/aggregates/mod.rs

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1610,17 +1610,6 @@ impl AggregateExec {
16101610
_ => Precision::Absent,
16111611
}
16121612
}
1613-
1614-
fn with_new_children_and_same_properties(
1615-
&self,
1616-
mut children: Vec<Arc<dyn ExecutionPlan>>,
1617-
) -> Self {
1618-
Self {
1619-
input: children.swap_remove(0),
1620-
metrics: ExecutionPlanMetricsSet::new(),
1621-
..Self::clone(self)
1622-
}
1623-
}
16241613
}
16251614

16261615
impl DisplayAs for AggregateExec {
@@ -1838,6 +1827,17 @@ impl ExecutionPlan for AggregateExec {
18381827
Ok(Arc::new(me))
18391828
}
18401829

1830+
fn with_new_children_and_same_properties(
1831+
self: Arc<Self>,
1832+
mut children: Vec<Arc<dyn ExecutionPlan>>,
1833+
) -> Result<Arc<dyn ExecutionPlan>> {
1834+
Ok(Arc::new(Self {
1835+
input: children.swap_remove(0),
1836+
metrics: ExecutionPlanMetricsSet::new(),
1837+
..Self::clone(&*self)
1838+
}))
1839+
}
1840+
18411841
fn execute(
18421842
&self,
18431843
partition: usize,

datafusion/physical-plan/src/async_func.rs

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -113,17 +113,6 @@ impl AsyncFuncExec {
113113
pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
114114
&self.input
115115
}
116-
117-
fn with_new_children_and_same_properties(
118-
&self,
119-
mut children: Vec<Arc<dyn ExecutionPlan>>,
120-
) -> Self {
121-
Self {
122-
input: children.swap_remove(0),
123-
metrics: ExecutionPlanMetricsSet::new(),
124-
..Self::clone(self)
125-
}
126-
}
127116
}
128117

129118
impl DisplayAs for AsyncFuncExec {
@@ -180,6 +169,17 @@ impl ExecutionPlan for AsyncFuncExec {
180169
)?))
181170
}
182171

172+
fn with_new_children_and_same_properties(
173+
self: Arc<Self>,
174+
mut children: Vec<Arc<dyn ExecutionPlan>>,
175+
) -> Result<Arc<dyn ExecutionPlan>> {
176+
Ok(Arc::new(Self {
177+
input: children.swap_remove(0),
178+
metrics: ExecutionPlanMetricsSet::new(),
179+
..Self::clone(&*self)
180+
}))
181+
}
182+
183183
fn execute(
184184
&self,
185185
partition: usize,

datafusion/physical-plan/src/buffer.rs

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -123,17 +123,6 @@ impl BufferExec {
123123
pub fn capacity(&self) -> usize {
124124
self.capacity
125125
}
126-
127-
fn with_new_children_and_same_properties(
128-
&self,
129-
mut children: Vec<Arc<dyn ExecutionPlan>>,
130-
) -> Self {
131-
Self {
132-
input: children.swap_remove(0),
133-
metrics: ExecutionPlanMetricsSet::new(),
134-
..Self::clone(self)
135-
}
136-
}
137126
}
138127

139128
impl DisplayAs for BufferExec {
@@ -181,6 +170,17 @@ impl ExecutionPlan for BufferExec {
181170
Ok(Arc::new(Self::new(children.swap_remove(0), self.capacity)))
182171
}
183172

173+
fn with_new_children_and_same_properties(
174+
self: Arc<Self>,
175+
mut children: Vec<Arc<dyn ExecutionPlan>>,
176+
) -> Result<Arc<dyn ExecutionPlan>> {
177+
Ok(Arc::new(Self {
178+
input: children.swap_remove(0),
179+
metrics: ExecutionPlanMetricsSet::new(),
180+
..Self::clone(&*self)
181+
}))
182+
}
183+
184184
fn execute(
185185
&self,
186186
partition: usize,

datafusion/physical-plan/src/coalesce_batches.rs

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -117,17 +117,6 @@ impl CoalesceBatchesExec {
117117
input.boundedness(),
118118
)
119119
}
120-
121-
fn with_new_children_and_same_properties(
122-
&self,
123-
mut children: Vec<Arc<dyn ExecutionPlan>>,
124-
) -> Self {
125-
Self {
126-
input: children.swap_remove(0),
127-
metrics: ExecutionPlanMetricsSet::new(),
128-
..Self::clone(self)
129-
}
130-
}
131120
}
132121

133122
#[expect(deprecated)]
@@ -195,6 +184,17 @@ impl ExecutionPlan for CoalesceBatchesExec {
195184
))
196185
}
197186

187+
fn with_new_children_and_same_properties(
188+
self: Arc<Self>,
189+
mut children: Vec<Arc<dyn ExecutionPlan>>,
190+
) -> Result<Arc<dyn ExecutionPlan>> {
191+
Ok(Arc::new(Self {
192+
input: children.swap_remove(0),
193+
metrics: ExecutionPlanMetricsSet::new(),
194+
..Self::clone(&*self)
195+
}))
196+
}
197+
198198
fn execute(
199199
&self,
200200
partition: usize,

datafusion/physical-plan/src/coalesce_partitions.rs

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -100,17 +100,6 @@ impl CoalescePartitionsExec {
100100
.with_evaluation_type(drive)
101101
.with_scheduling_type(scheduling)
102102
}
103-
104-
fn with_new_children_and_same_properties(
105-
&self,
106-
mut children: Vec<Arc<dyn ExecutionPlan>>,
107-
) -> Self {
108-
Self {
109-
input: children.swap_remove(0),
110-
metrics: ExecutionPlanMetricsSet::new(),
111-
..Self::clone(self)
112-
}
113-
}
114103
}
115104

116105
impl DisplayAs for CoalescePartitionsExec {
@@ -164,6 +153,17 @@ impl ExecutionPlan for CoalescePartitionsExec {
164153
Ok(Arc::new(plan))
165154
}
166155

156+
fn with_new_children_and_same_properties(
157+
self: Arc<Self>,
158+
mut children: Vec<Arc<dyn ExecutionPlan>>,
159+
) -> Result<Arc<dyn ExecutionPlan>> {
160+
Ok(Arc::new(Self {
161+
input: children.swap_remove(0),
162+
metrics: ExecutionPlanMetricsSet::new(),
163+
..Self::clone(&*self)
164+
}))
165+
}
166+
167167
fn execute(
168168
&self,
169169
partition: usize,

datafusion/physical-plan/src/coop.rs

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -234,16 +234,6 @@ impl CooperativeExec {
234234
pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
235235
&self.input
236236
}
237-
238-
fn with_new_children_and_same_properties(
239-
&self,
240-
mut children: Vec<Arc<dyn ExecutionPlan>>,
241-
) -> Self {
242-
Self {
243-
input: children.swap_remove(0),
244-
..Self::clone(self)
245-
}
246-
}
247237
}
248238

249239
impl DisplayAs for CooperativeExec {
@@ -290,6 +280,16 @@ impl ExecutionPlan for CooperativeExec {
290280
Ok(Arc::new(CooperativeExec::new(children.swap_remove(0))))
291281
}
292282

283+
fn with_new_children_and_same_properties(
284+
self: Arc<Self>,
285+
mut children: Vec<Arc<dyn ExecutionPlan>>,
286+
) -> Result<Arc<dyn ExecutionPlan>> {
287+
Ok(Arc::new(Self {
288+
input: children.swap_remove(0),
289+
..Self::clone(&*self)
290+
}))
291+
}
292+
293293
fn execute(
294294
&self,
295295
partition: usize,

0 commit comments

Comments
 (0)