apache / apache/datafusion

FFI serialization uses DefaultPhysicalProtoConverter, severing shared dynamic filter references

Open
#25,154 0 comments 0 reactions 0 assignees View on GitHub
bug ffi
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### Describe the bug

`datafusion-proto` ships two proto converters. `DeduplicatingProtoConverter` exists specifically to preserve referential integrity across a round trip:

```rust
// datafusion/proto/src/physical_plan/mod.rs:1972-1976
/// A proto converter that deduplicates [`PhysicalExpr`] by [`PhysicalExpr::expression_id`].
/// This helps preserve referential integrity when deserializing [`ExecutionPlan`]s
/// which may contain multiple occurrences of the same [`PhysicalExpr`] (ex. when
/// [`DynamicFilterPhysicalExpr`] are pushed down, it is important to preserve
/// referential integrity).
```

Every FFI serialization path uses the other one.

- `datafusion/ffi/src/query_planner.rs:168` and `:314` call `physical_plan_{to,from}_bytes_with_extension_codec`, which hardcode `DefaultPhysicalProtoConverter` (`datafusion/proto/src/bytes/mod.rs:245`, and the same in the `from_bytes` counterpart).
- `datafusion/ffi/src/proto/physical_extension_codec.rs:155` and `:173` pass `&DefaultPhysicalProtoConverter {}` explicitly.

The result is that shared `Arc` references are silently split into independent objects whenever a plan crosses the FFI planner boundary.

This matters because the **last** rule in the default physical optimizer list is `FilterPushdown::new_post_optimization()` (`datafusion/physical-optimizer/src/optimizer.rs:181`), commented:

```rust
// This FilterPushdown handles dynamic filters that may have references to the source ExecutionPlan.
```

What that rule produces is object identity, not structure. `HashJoinExec` holds `dynamic_filter.filter: Arc` (`datafusion/physical-plan/src/joins/hash_join/exec.rs:892`) and fills it in once the build side completes; the `DataSourceExec` on the probe side reads that same object to prune. `AggregateExec` has the equivalent arrangement (`datafusion/physical-plan/src/aggregates/mod.rs:776`).

After a non-deduplicating round trip the join updates its copy and the scan reads its own, which stays at the `lit(true)` placeholder. No error is raised; the optimization simply stops applying.

### To Reproduce

Any query through an FFI `QueryPlanner` that produces a dynamic filter — a hash join with `enable_join_dynamic_filter_pushdown` on, or a TopK/aggregate with a dynamic filter — loses the pushdown. Because the failure is silent, it shows up as a performance regression rather than a wrong answer.

A direct test: build a plan holding one `DynamicFilterPhysicalExpr` referenced from two nodes, round-trip it through the FFI planner boundary, and check whether the two references are still `Arc::ptr_eq` on the far side. They will not be.

### Expected behavior

A plan crossing the FFI boundary should preserve shared `PhysicalExpr` identity, as it already does for callers who opt into `DeduplicatingProtoConverter`.

### Additional context

The fix appears to be a call-site swap. Public helpers taking an explicit converter already exist — `physical_plan_to_bytes_with_proto_converter` (`datafusion/proto/src/bytes/mod.rs:252`) and `physical_plan_from_bytes_with_proto_converter` (`:310`) — so the four FFI call sites above can pass `DeduplicatingProtoConverter` instead. No ABI change.

This is independent of the FFI query planner issue and predates it: it affects any FFI planner today, whether or not that planner ever trips over a foreign-wrapped node. It is filed separately for that reason. See umbrella #25152 for the surrounding analysis.

**Relationship to prior work.** The deduplication machinery landed via #20416 (`proto: serialize and dedupe dynamic filters`) and #20418 (`Serialize dynamic filters across network boundaries`), both closed, and the design context is in #21207 (`[DISCUSSION] Future of Dynamic Filters Sync`). This issue is not a gap in that work — it is that `datafusion-ffi` never opted into it. The boundary it affects is the same class of "cross-boundary propagation" those PRs targeted, so the FFI path arguably should have been included.

Worth checking as part of a fix whether any other in-tree caller of the `_with_extension_codec` helpers is silently relying on the non-deduplicating behaviour.

Contributor guide

Open the contributing guide

Research direction

Start in datafusion/ffi/src/query_planner.rs and datafusion/ffi/src/proto/physical_extension_codec.rs, then read the converter helpers in datafusion/proto/src/bytes/mod.rs. Reproduce the issue by round-tripping a plan with one DynamicFilterPhysicalExpr referenced by two nodes. Done means the FFI round trip preserves Arc::ptr_eq and relevant FFI paths are covered without changing the ABI.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
databases
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
70/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.