Skip to content

FFI_ExecutionPlan loses input distribution and ordering requirements, so optimizer rules across FFI produce invalid plans #27427

Open

Description

@andygrove

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:

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:

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.

Activity

  1. self-assigned this
    on Nov 14, 2026
  2. andygrove commented on Nov 14, 2026

    @andygrove
    MemberAuthor

    This is a regression in 56 for FFI query planners

    For users of a foreign QueryPlanner, this is a regression from 55 to 56: a query that plans and runs on 55.0.0 fails on 56 with the SanityCheckPlan error above.

    The gap in FFI_ExecutionPlan is not new: it never carried input requirements. 56 makes it visible because #26461 moved SanityCheckPlan out of the rule list and into DefaultPhysicalPlanner. On 55 the check ran as one of the host's rules over FFI, so it was just as blind as EnsureRequirements and passed the plan.

    On 55, this particular query still returned correct results, by luck. The partial aggregate produces a single partition with one row per key, so round-robin sends each key to exactly one final partition. With a multi-partition input, the same plan could split a key across partitions and return duplicate groups. So 56 turns a silent correctness hazard into a hard error, which is better behavior, but users who upgrade still see working queries start failing.

    The fix is contained to datafusion-ffi: FFI_ExecutionPlan gets callbacks for input_distribution_requirements, required_input_ordering, maintains_input_order and benefits_from_input_partitioning. It cherry-picks cleanly onto branch-56 apart from the upgrade-guide file, so it is a candidate for 56.0.0 or a 56.0.x patch release. PR to follow.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

bugSomething isn't workingregressionSomething that used to work no longer does

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions