fix: output_ordering converted to Vec<Option<LexOrdering>>

[destrex271]

Which issue does this PR close?

• Closes FileScanConfig::output_ordering must be vector of optionals #17354

What changes are included in this PR?

• Updated the type of output_ordering in FileScanConfig to Vec<Option<LexOrdering>>
• Updated the get_projected_output_ordering function in file_scan_config.rs to return Vec<Option>
• Handle output generated by get_projected_output_ordering in various places by flattening the result and reconstructing Vec<LexOrdering> from Vec<Option<LexOrdering>> (This might need some discussion/additional reviews)

Are these changes tested?

Changes were mostly minor refactors.

[alamb]

@crepererum can you please review this PR (as you filed the request for #17354)

[destrex271]

@alamb , can you please trigger the test workflows again? Just updated the branch

[crepererum]

I think it's a start, but the actual bug of "we loose partitions during projections" was now moved out of get_projected_output_ordering into all these flatten calls. To actually fix this, I think we need to fix the return type of project as well.

[crepererum]

Could we eventually fix this API as well so that the API setter is the same as the internal state? If you wanna make the transition for existing users smoother, deprecate this method here and add a new one a la:

pub fn with_output_ordering_opt(mut self, output_ordering: Vec<LexOrdering>) -> Self { ... }

[destrex271]

Yes that makes sense, I'll refactor this function and related references.

[destrex271]

Should we also modify the type of projected_output_orderings here?

datafusion/datafusion/physical-plan/src/streaming.rs

Line 67 in da3d90a

projected_output_ordering: Vec<LexOrdering>,

This is a result of modifying the return type of the following functions to Result<Vec<Option<LexOrdering>>

1. datafusion/datafusion/physical-expr/src/physical_expr.rs

   Line 131 in da3d90a

   pub fn create_ordering(

2. datafusion/datafusion/core/src/datasource/listing/table.rs

   Line 1132 in da3d90a

   fn try_create_output_ordering(&self) -> Result<Vec<LexOrdering>> {

This is required since in

datafusion/datafusion/core/src/datasource/listing/table.rs

Line 1205 in da3d90a

let output_ordering = self.try_create_output_ordering()?;

we are creating the output orderings using the try_create_output_ordering function.

There are two solutions:

1. Either modify the return types of create_output_ordering and other functions that are affected by it
2. Preserve the current return type as Vec<LexOrdering> and just wrap each item in the vec into Some() before sending as args to with_output_ordering.

[crepererum]

I think the projected version needs to be adjusted too.

[destrex271]

Understood! I'll make the necessary changes.

[destrex271]

@crepererum

I guess we should skip modifying the type of expr to Option since at this stage having None seems to be a little useless.

#[derive(Debug, Clone)]
pub struct SortExec {
    /// Input schema
    pub(crate) input: Arc<dyn ExecutionPlan>,
    /// Sort expressions
    expr: LexOrdering,
    /// Containing all metrics set created during sort
    metrics_set: ExecutionPlanMetricsSet,
    /// Preserve partitions of input plan. If false, the input partitions
    /// will be sorted and merged into a single output partition.
    preserve_partitioning: bool,
    /// Fetch highest/lowest n results
    fetch: Option<usize>,
    /// Normalized common sort prefix between the input and the sort expressions (only used with fetch)
    common_sort_prefix: Vec<PhysicalSortExpr>,
    /// Cache holding plan properties like equivalences, output partitioning etc.
    cache: PlanProperties,
    /// Filter matching the state of the sort for dynamic filter pushdown.
    /// If `fetch` is `Some`, this will also be set and a TopK operator may be used.
    /// If `fetch` is `None`, this will be `None`.
    filter: Option<Arc<RwLock<TopKDynamicFilters>>>,
}

[crepererum]

I think we should display the None cases too, otherwise this is going to be pretty confusing. Imagine a projection that leads to Some, None, Some and you now gonna display 2 orderings (for 3 partitions). From that you cannot tell if it was None, Some, Some, Some, None, Some, or Some, Some, None.

[crepererum]

I'm not convinced that this is correct, with the same argument as for the display code. I think the return type here also needs to be Vec<Option<...>>.

[crepererum]

same as for the other display code.

[destrex271]

Understood. I'll refactor the display function and its relevant calls.