Improve get_meet_of_orderings to check for common prefixes

[ozankabak]

Which issue does this PR close?

This PR addresses a concern raised by @mingmwang in his review of #5035.

Rationale for this change

As discussed here, we want to get the longest common prefix when finding the meet of the given orderings. If we don't do that, we may lose some coarse ordering information.

What changes are included in this PR?

This PR changes the internal logic of the get_meet_of_orderings function to add the desired functionality.

Are these changes tested?

New unit tests are added to verify the correct behavior in three relevant test cases.

Are there any user-facing changes?

No.

[mingmwang]

LGTM

[mingmwang]

Union allows its inputs to have different schema names or even different types. For example:
select a1， b1 from tab_a1 UNION ALL select a2， b2 from tab_a2

In case the tab_a1 is sorted by ('a1', 'b1') and tab_a2 is sorted by ('a2', 'b2'), we can safely derive the output ordering is ('a1', 'b1'). But I think you can leave it for future since the current implementation is already quite sophisticated.

[mingmwang]

Should move the calculation logic to the UnionExec's constructor ? Since maintains_input_order() will be invoked multiple times during the optimization process.

Or here I think we do not need to call ordering_satisfy with equivalence_properties, we can just checking the length
of the sort expressions, if the length matches, should be true.

    fn maintains_input_order(&self) -> Vec<bool> {
        // If the Union has an output ordering, it maintains at least one
        // child's ordering (i.e. the meet).
        // For instance, assume that the first child is SortExpr('a','b','c'),
        // the second child is SortExpr('a','b') and the third child is
        // SortExpr('a','b'). The output ordering would be SortExpr('a','b'),
        // which is the "meet" of all input orderings. In this example, this
        // function will return vec![false, true, true], indicating that we
        // preserve the orderings for the 2nd and the 3rd children.
        self.inputs()
            .iter()
            .map(|child| {
                ordering_satisfy(self.output_ordering(), child.output_ordering(), || {
                    child.equivalence_properties()
                })
            })
            .collect()
    }

[ozankabak]

You mean something like this?

fn maintains_input_order(&self) -> Vec<bool> {
    // ... comment text ...
    if let Some(output_ordering) = self.output_ordering() {
        self.inputs()
            .iter()
            .map(|child| {
                if let Some(child_ordering) = child.output_ordering() {
                    output_ordering.len() == child_ordering.len()
                } else {
                    false
                }
            })
            .collect()
    } else {
        vec![false; self.inputs().len()]
    }
}

[mingmwang]

You mean something like this?

fn maintains_input_order(&self) -> Vec<bool> {
    // ... comment text ...
    if let Some(output_ordering) = self.output_ordering() {
        self.inputs()
            .iter()
            .map(|child| {
                if let Some(child_ordering) = child.output_ordering() {
                    output_ordering.len() == child_ordering.len()
                } else {
                    false
                }
            })
            .collect()
    } else {
        vec![false; self.inputs().len()]
    }
}

Yes.

[ozankabak]

Done, thanks for taking a closer look.

Union allows its inputs to have different schema names or even different types. For example: select a1， b1 from tab_a1 UNION ALL select a2， b2 from tab_a2

In case the tab_a1 is sorted by ('a1', 'b1') and tab_a2 is sorted by ('a2', 'b2'), we can safely derive the output ordering is ('a1', 'b1'). But I think you can leave it for future since the current implementation is already quite sophisticated.

@mustafasrepo will take over and think about this in a follow-on PR.

[alamb]

Thank you @ozankabak

[alamb]

👍

[alamb]

👍

[alamb]

👍

[alamb]

I don't understand why we no longer need to check that the output order of the child is compatible with the output order of the input. Shouldn't this be calling get_meet_of_orderings to check rather than checking the length?

[ozankabak]

The call self.output_ordering() in turn calls get_meet_of_orderings and computes the common longest prefix. We then check whether any of the inputs are equal to that prefix (equality in length implies equality in this case). Note that we are checking for strict equality instead of an equivalence-aware check following @mingmwang's suggestion.

[alamb]

Thanks @ozankabak -- makes sense to me

[ursabot]

Benchmark runs are scheduled for baseline = 125a858 and contender = 1f7885b. 1f7885b is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] ec2-t3-xlarge-us-east-2
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] test-mac-arm
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] ursa-i9-9960x
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] ursa-thinkcentre-m75q
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java