andygrove opened a new issue, #27427:
URL: https://github.com/apache/datafusion/issues/27427
### Describe the bug
`FFI_ExecutionPlan` does not carry a plan's input requirements across the
FFI boundary. A `ForeignExecutionPlan` falls back to the `ExecutionPlan` trait
defaults for:
- `input_distribution_requirements` (and the deprecated
`required_input_distribution`) → `UnspecifiedDistribution` for every child
- `required_input_ordering` → `None` for every child
- `maintains_input_order` → `false` for every child
- `benefits_from_input_partitioning` → derived from the (lost) distribution
So when a physical optimizer rule runs on the other side of the boundary
from the plan it optimizes, it cannot see what the plan's nodes require.
`EnsureRequirements` then skips repartitioning and sorting that the plan needs.
Other rules, like the ones that reason about `maintains_input_order`, also see
wrong metadata.
This happens with a foreign `QueryPlanner` (`FFI_QueryPlanner`) that calls
`DefaultPhysicalPlanner` with a `ForeignSession`. The session's physical
optimizer rules belong to the host and run via `FFI_PhysicalOptimizerRule`. To
the host, every node the planner library built is an opaque
`ForeignExecutionPlan`.
Before #26461 this could silently produce a plan that broke its
requirements. In that case a `FinalPartitioned` aggregate gets round-robin
input, so groups can be split across partitions. `SanityCheckPlan` ran as a
host-side rule and was just as blind, so it let the plan through. Since #26461
the check runs natively in `DefaultPhysicalPlanner`, so 56.0.0 now rejects the
plan:
```text
SanityCheckPlan: the plan produced by physical optimizer rule
'EnsureRequirements' does not satisfy the order and distribution requirements
Plan:
AggregateExec: mode=FinalPartitioned, gby=[numbers.A % Int64(2)@0 as
numbers.A % Int64(2)], aggr=[count(Int64(1))]
FFI_ExecutionPlan: RepartitionExec, number_of_children=1
AggregateExec: mode=Partial, gby=[CAST(A@0 AS Int64) % 2 as numbers.A
% Int64(2)], aggr=[count(Int64(1))]
FFI_ExecutionPlan: DataSourceExec, number_of_children=0
does not satisfy distribution requirements: KeyPartitioned[[numbers.A %
Int64(2)@0]]
Child-0 output partitioning: RoundRobinBatch(4)
```
This blocks datafusion-python's upgrade to 56
(apache/datafusion-python#1826).
### To Reproduce
Wrap a native `FinalPartitioned` `AggregateExec` in `FFI_ExecutionPlan`,
force the foreign path, and compare:
```rust
let mut ffi = FFI_ExecutionPlan::new(Arc::clone(&native_final_agg), None);
ffi.library_marker_id = mock_foreign_marker_id;
let foreign: Arc<dyn ExecutionPlan> = (&ffi).try_into()?;
native.required_input_distribution() // [KeyPartitioned([Column { name:
"numbers.A % Int64(2)", index: 0 }])]
foreign.required_input_distribution() // [UnspecifiedDistribution]
```
End to end: in datafusion-python on #1826,
`examples/datafusion-ffi-query-planner-example`,
`test_spawning_plan_across_three_libraries` runs `SELECT "A" % 2, count(*) ...
GROUP BY 1` with `target_partitions = 4` through a foreign query planner. With
debug logging on, the plan after the host's `EnsureRequirements` (run over FFI)
has a round-robin `RepartitionExec` under `FinalPartitioned`. Planning the same
query natively gives `Hash([numbers.A % Int64(2)@0], 4)`.
### Expected behavior
A `ForeignExecutionPlan` reports the same input requirements as the plan it
wraps: distribution requirements (including co-partitioning), required input
ordering (hard/soft alternatives), `maintains_input_order` and
`benefits_from_input_partitioning`.
### Additional context
Proposed fix: add callbacks for these four methods to `FFI_ExecutionPlan`,
with FFI representations of `Distribution`, `InputDistributionRequirements`,
`OrderingRequirements` and `PhysicalSortRequirement`, and forward them from
`ForeignExecutionPlan`. This changes the `FFI_ExecutionPlan` layout, so it is
an FFI ABI change.
--
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]
