timsaucer commented on code in PR #24670:
URL: https://github.com/apache/datafusion/pull/24670#discussion_r3913901198
##########
datafusion/physical-plan/src/projection.rs:
##########
@@ -270,6 +270,22 @@ impl ProjectionExec {
))
}
+ /// Returns whether this projection's output metadata differs from the
+ /// metadata derived from its expressions and input schema.
+ fn overrides_metadata(&self) -> Result<bool> {
+ let derived_schema = self
+ .projector
+ .projection()
+ .project_schema(self.input.schema().as_ref())?;
+ let output_schema = self.schema();
+ Ok(derived_schema.metadata() != output_schema.metadata()
+ || derived_schema
+ .fields()
+ .iter()
+ .zip(output_schema.fields())
+ .any(|(derived, output)| derived.metadata() !=
output.metadata()))
+ }
Review Comment:
From an agent, but it looks like a good idea to me. Specifically I've seen
things like this bite when you have very wide schemas.
- Perf: overrides_metadata() recomputes full schema derivation every pass.
`project_schema` walks every expr 3× (`return_field`, `data_type`,
`nullable`) and allocates a Schema. Called via transform_down for every
ProjectionExec node, plus once per inner projection in collapse loop, plus
recursively at projection.rs:1430. Wide projections pay repeatedly.
Cache bool once in try_from_projector as a struct field. try_new path always
false; only try_new_with_schema_metadata needs the compare.
##########
datafusion/physical-plan/src/projection.rs:
##########
@@ -1331,12 +1357,20 @@ pub fn update_join_filter(
fn try_collapse_projection_chain(
outer: &ProjectionExec,
) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ if outer.overrides_metadata()? {
+ return Ok(None);
+ }
Review Comment:
From an agent's review. I didn't track this one through myself so feel free
to push back if you disagree.
- Outer guard in try_collapse_projection_chain looks unnecessary.
Line 1425 already forces outer.schema() metadata onto the unified
projection. The inner guard (line 1370) is the one carrying the correctness
weight — substituting through an inner projection loses metadata the outer expr
observes. Dropping the outer bail keeps collapse working for extension-type
casts. Intentional extra caution, or leftover from an earlier revision?
##########
datafusion/physical-plan/src/projection.rs:
##########
@@ -1014,6 +1030,10 @@ pub fn remove_unnecessary_projections(
plan: Arc<dyn ExecutionPlan>,
) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
let maybe_modified = if let Some(projection) =
plan.downcast_ref::<ProjectionExec>() {
+ // Removing a projection with observable metadata can change query
results.
+ if projection.overrides_metadata()? {
+ return Ok(Transformed::no(plan));
+ }
Review Comment:
After some back and forth with an agent, this now seems like a reasonable
issue about this check:
This is a precondition, so it exits before `try_swapping_with_projection` —
it blocks relocation as well as removal. The 20 `try_swapping_with_projection`
impls split into two groups:
- **7 relocate** the projection via `make_with_child` — which this PR
already made metadata-safe
- **3 absorb** it (`source.rs:525` covering all `DataSource` impls,
`streaming.rs:337`) and **1 embeds** it (`try_embed_projection`, covering the
`FilterExec` fallback and all 5 joins) — these re-derive their schema and do
drop the metadata
The precondition can't distinguish them, so it pays for the lossy 4 by
disabling the safe 7. The catch is who that lands on: `overrides_metadata()`
only fires for projections built through `try_new_with_schema_metadata`, which
after #23169 means embedder-constructed plans, not SQL. So the users who need
this fix are exactly the ones who lose the pushdown.
Concretely:
```
ProjectionExec: a@0 AS a declared output: a -> {"unit": "ms"}
FilterExec: a@0 > 5
DataSourceExec: a, b, c
```
| | metadata | filter reads |
| ---------------- | -------- | ------------ |
| pre-PR | **lost** | 1 col |
| this PR | kept | **3 cols** |
| suggestion below | kept | 1 col |
Would it work to make it a postcondition instead — check the result rather
than deciding up front?
```rust
projection
.input()
.try_swapping_with_projection(projection)?
.filter(|swapped| swapped.schema() == projection.schema())
```
Deleting the precondition should be free: the removal path is already
covered by the `projection.schema() == projection.input().schema()` you added
to `is_projection_removable`, so the precondition was only ever guarding the
swap. This also means a future `try_swapping_with_projection` impl is covered
without anyone remembering to add a guard — the failure mode becomes a lost
optimization rather than lost metadata. And it adds no public API, which I
think keeps this patch-eligible.
I tried it against `d112760`:
| Check | Result
|
| ----------------------------------- |
----------------------------------------------- |
| Filter relocation (safe path) | swap happens, metadata preserved
|
| `try_embed_projection` (lossy path) | swap rejected, projection kept,
metadata intact |
| Full sqllogictest suite, 510 files | zero false rejections
|
| slt failures vs. unmodified PR head | identical (same 35 files)
|
| `projection::tests` | 24/24 pass, including all 4 you
added |
The rejection trace on the lossy path shows the metadata that would have
been dropped:
```
SWAP_REJECTED child=FilterExec
proj_schema= [Field { name: "a", metadata: {"unit": "ms"} }]
swapped_schema= [Field { name: "a" }]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
