`CooperativeExec` incorrectly implements `maintains_input_order`

[pepijnve]

Describe the bug

CooperativeExec implements maintains_input_order by delegating to its input. As a consequence it may return a Vec with a length that is smaller or larger than the number of children of CooperativeExec which is always exactly 1.

To Reproduce

No response

Expected behavior

No response

Additional context

No response

[pepijnve]

I came up with this as unit test, but that pretty much duplicates the implementation. Should I add this test case or is it not worth it?

What might be useful is to have more general test logic that takes in an ExecutionPlan and verifies (to the extent possible) that the post conditions required by the trait are met. For instance that the length of the various Vecs is correct. This can't check for correctness of the values (i.e. true or false in this case), but it can check for wellformedness of the result.

#[test]
fn maintains_input_order() -> Result<()> {
    let schema = Arc::new(Schema::empty());
    let exec = LazyMemoryExec::try_new(schema, vec![])?;
    let coop = CooperativeExec::new(Arc::new(exec));
    let maintains_input_order = coop.maintains_input_order();
    assert_eq!(maintains_input_order, vec![true;coop.children().len()]);
    Ok(())
}

[alamb]

What might be useful is to have more general test logic that takes in an ExecutionPlan and verifies (to the extent possible) that the post conditions required by the trait are met. For instance that the length of the various Vecs is correct. This can't check for correctness of the values (i.e. true or false in this case), but it can check for wellformedness of the result.

@wiedld started adding something like this with the ExecutionPlan::check_invariants:

https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html#method.check_invariants

Maybe we can use that mechanism?

[alamb]

Or we can extend here https://github.com/apache/datafusion/blob/c6d55207161e400e53645d5ee7d7bf16cd024c2f/datafusion/core/src/physical_planner.rs#L2356-L2355 with some basic checks