Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Improve get_meet_of_orderings to check for common prefixes #5111

Merged
merged 2 commits into from
Jan 31, 2023
Merged

Improve get_meet_of_orderings to check for common prefixes #5111

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
44114c6
Improve get_meet_of_orderings to check for common prefixes among the …
ozankabak Jan 30, 2023
21d00a1
Incorporate review suggestions
ozankabak Jan 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
197 changes: 172 additions & 25 deletions datafusion/core/src/physical_plan/common.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ use arrow::error::ArrowError;
use arrow::error::Result as ArrowResult;
use arrow::ipc::writer::{FileWriter, IpcWriteOptions};
use arrow::record_batch::RecordBatch;
use datafusion_physical_expr::utils::ordering_satisfy;
use datafusion_physical_expr::PhysicalSortExpr;
use futures::{Future, Stream, StreamExt, TryStreamExt};
use log::debug;
Expand Down Expand Up @@ -324,34 +323,37 @@ pub fn transpose<T>(original: Vec<Vec<T>>) -> Vec<Vec<T>> {
}
}

/// Calculates the "meet" of children orderings
/// The meet is the finest ordering that satisfied by all the input
/// Calculates the "meet" of given orderings.
/// The meet is the finest ordering that satisfied by all the given
/// orderings, see https://en.wikipedia.org/wiki/Join_and_meet.
pub fn get_meet_of_orderings(
children: &[Arc<dyn ExecutionPlan>],
given: &[Arc<dyn ExecutionPlan>],
) -> Option<&[PhysicalSortExpr]> {
// To find the meet, we first find the smallest input ordering.
let mut smallest: Option<&[PhysicalSortExpr]> = None;
for item in children.iter() {
if let Some(ordering) = item.output_ordering() {
smallest = match smallest {
None => Some(ordering),
Some(expr) if ordering.len() < expr.len() => Some(ordering),
_ => continue,
given
.iter()
.map(|item| item.output_ordering())
.collect::<Option<Vec<_>>>()
.and_then(get_meet_of_orderings_helper)
}

fn get_meet_of_orderings_helper(
orderings: Vec<&[PhysicalSortExpr]>,
) -> Option<&[PhysicalSortExpr]> {
let mut idx = 0;
let first = orderings[0];
loop {
for ordering in orderings.iter() {
if idx >= ordering.len() {
return Some(ordering);
} else if ordering[idx] != first[idx] {
return if idx > 0 {
Some(&ordering[..idx])
} else {
None
};
}
} else {
return None;
}
}
// Check if the smallest ordering is a meet or not:
if children.iter().all(|child| {
ordering_satisfy(child.output_ordering(), smallest, || {
child.equivalence_properties()
})
}) {
smallest
} else {
None
idx += 1;
}
}

Expand All @@ -368,7 +370,152 @@ mod tests {
datatypes::{DataType, Field, Schema},
record_batch::RecordBatch,
};
use datafusion_physical_expr::expressions::col;
use datafusion_physical_expr::expressions::{col, Column};

#[test]
fn get_meet_of_orderings_helper_common_prefix_test() -> Result<()> {
let input1: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("b", 1)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("c", 2)),
options: SortOptions::default(),
},
];

let input2: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("b", 1)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("y", 2)),
options: SortOptions::default(),
},
];

let input3: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("x", 1)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("y", 2)),
options: SortOptions::default(),
},
];

let expected = vec![PhysicalSortExpr {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
}];

let result = get_meet_of_orderings_helper(vec![&input1, &input2, &input3]);
assert_eq!(result.unwrap(), expected);
Ok(())
}

#[test]
fn get_meet_of_orderings_helper_subset_test() -> Result<()> {
let input1: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("b", 1)),
options: SortOptions::default(),
},
];

let input2: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("b", 1)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("c", 2)),
options: SortOptions::default(),
},
];

let input3: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("b", 1)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("d", 2)),
options: SortOptions::default(),
},
];

let result = get_meet_of_orderings_helper(vec![&input1, &input2, &input3]);
assert_eq!(result.unwrap(), input1);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

Ok(())
}

#[test]
fn get_meet_of_orderings_helper_no_overlap_test() -> Result<()> {
let input1: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("b", 1)),
options: SortOptions::default(),
},
];

let input2: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("x", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 1)),
options: SortOptions::default(),
},
];

let input3: Vec<PhysicalSortExpr> = vec![
PhysicalSortExpr {
expr: Arc::new(Column::new("a", 0)),
options: SortOptions::default(),
},
PhysicalSortExpr {
expr: Arc::new(Column::new("y", 1)),
options: SortOptions::default(),
},
];

let result = get_meet_of_orderings_helper(vec![&input1, &input2, &input3]);
assert!(result.is_none());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

Ok(())
}

#[test]
fn test_meet_of_orderings() -> Result<()> {
Expand Down
21 changes: 13 additions & 8 deletions datafusion/core/src/physical_plan/union.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,6 @@ use crate::{
error::Result,
physical_plan::{expressions, metrics::BaselineMetrics},
};
use datafusion_physical_expr::utils::ordering_satisfy;
use tokio::macros::support::thread_rng_n;

/// `UnionExec`: `UNION ALL` execution plan.
Expand Down Expand Up @@ -240,14 +239,20 @@ impl ExecutionPlan for UnionExec {
// 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()
if let Some(output_ordering) = self.output_ordering() {
self.inputs()
.iter()
.map(|child| {
if let Some(child_ordering) = child.output_ordering() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

Copy link
Contributor Author

@ozankabak ozankabak Jan 30, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

output_ordering.len() == child_ordering.len()
} else {
false
}
})
})
.collect()
.collect()
} else {
vec![false; self.inputs().len()]
}
}

fn with_new_children(
Expand Down