alamb commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2121812355
##########
datafusion/physical-expr-common/src/sort_expr.rs:
##########
@@ -129,23 +138,55 @@ impl PhysicalSortExpr {
to_str(&self.options)
)
}
-}
-/// Access the PhysicalSortExpr as a PhysicalExpr
-impl AsRef<dyn PhysicalExpr> for PhysicalSortExpr {
- fn as_ref(&self) -> &(dyn PhysicalExpr + 'static) {
- self.expr.as_ref()
+ /// Evaluates the sort expression into a `SortColumn` that can be passed
+ /// into the arrow sort kernel.
+ pub fn evaluate_to_sort_column(&self, batch: &RecordBatch) ->
Result<SortColumn> {
+ let array_to_sort = match self.expr.evaluate(batch)? {
+ ColumnarValue::Array(array) => array,
+ ColumnarValue::Scalar(scalar) =>
scalar.to_array_of_size(batch.num_rows())?,
+ };
+ Ok(SortColumn {
+ values: array_to_sort,
+ options: Some(self.options),
+ })
+ }
+
+ /// Checks whether this sort expression satisfies the given `requirement`.
Review Comment:
I really like these APIs. I think it would help future readers of we also
linked to a definition of what "satisfies" means
For example, maybe we could define precisely what "satisfies" and
"compatible" mean on the comments of `options_compatible` and then link to
`options_compatible` in these comments
##########
datafusion/physical-expr-common/src/sort_expr.rs:
##########
@@ -334,177 +313,142 @@ fn to_str(options: &SortOptions) -> &str {
}
}
-///`LexOrdering` contains a `Vec<PhysicalSortExpr>`, which represents
-/// a lexicographical ordering.
-///
-/// For example, `vec![a ASC, b DESC]` represents a lexicographical ordering
-/// that first sorts by column `a` in ascending order, then by column `b` in
-/// descending order.
-#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
-pub struct LexOrdering {
- inner: Vec<PhysicalSortExpr>,
-}
-
-impl AsRef<LexOrdering> for LexOrdering {
- fn as_ref(&self) -> &LexOrdering {
- self
+// Cross-conversion utilities between `PhysicalSortExpr` and
`PhysicalSortRequirement`
+impl From<PhysicalSortExpr> for PhysicalSortRequirement {
+ fn from(value: PhysicalSortExpr) -> Self {
+ Self::new(value.expr, Some(value.options))
}
}
-impl LexOrdering {
- /// Creates a new [`LexOrdering`] from a vector
- pub fn new(inner: Vec<PhysicalSortExpr>) -> Self {
- Self { inner }
- }
-
- /// Return an empty LexOrdering (no expressions)
- pub fn empty() -> &'static LexOrdering {
- static EMPTY_ORDER: LazyLock<LexOrdering> =
LazyLock::new(LexOrdering::default);
- &EMPTY_ORDER
- }
-
- /// Returns the number of elements that can be stored in the LexOrdering
- /// without reallocating.
- pub fn capacity(&self) -> usize {
- self.inner.capacity()
- }
-
- /// Clears the LexOrdering, removing all elements.
- pub fn clear(&mut self) {
- self.inner.clear()
- }
-
- /// Takes ownership of the actual vector of `PhysicalSortExpr`s in the
LexOrdering.
- pub fn take_exprs(self) -> Vec<PhysicalSortExpr> {
- self.inner
- }
-
- /// Returns `true` if the LexOrdering contains `expr`
- pub fn contains(&self, expr: &PhysicalSortExpr) -> bool {
- self.inner.contains(expr)
- }
-
- /// Add all elements from `iter` to the LexOrdering.
- pub fn extend<I: IntoIterator<Item = PhysicalSortExpr>>(&mut self, iter:
I) {
- self.inner.extend(iter)
- }
-
- /// Remove all elements from the LexOrdering where `f` evaluates to
`false`.
- pub fn retain<F>(&mut self, f: F)
- where
- F: FnMut(&PhysicalSortExpr) -> bool,
- {
- self.inner.retain(f)
- }
-
- /// Returns `true` if the LexOrdering contains no elements.
- pub fn is_empty(&self) -> bool {
- self.inner.is_empty()
- }
-
- /// Returns an iterator over each `&PhysicalSortExpr` in the LexOrdering.
- pub fn iter(&self) -> core::slice::Iter<PhysicalSortExpr> {
- self.inner.iter()
+impl From<PhysicalSortRequirement> for PhysicalSortExpr {
+ /// The default sort options `ASC, NULLS LAST` when the requirement does
+ /// not specify sort options. This default is consistent with PostgreSQL.
+ ///
+ /// Reference: <https://www.postgresql.org/docs/current/queries-order.html>
+ fn from(value: PhysicalSortRequirement) -> Self {
+ let options = value
+ .options
+ .unwrap_or_else(|| SortOptions::new(false, false));
+ Self::new(value.expr, options)
}
+}
- /// Returns the number of elements in the LexOrdering.
- pub fn len(&self) -> usize {
- self.inner.len()
- }
+/// This object represents a lexicographical ordering and contains a vector
+/// of `PhysicalSortExpr` objects.
+///
+/// For example, a `vec![a ASC, b DESC]` represents a lexicographical ordering
+/// that first sorts by column `a` in ascending order, then by column `b` in
+/// descending order. The ordering is non-degenerate, meaning it contains at
+/// least one element, and it is duplicate-free, meaning it does not contain
+/// multiple entries for the same column.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct LexOrdering {
+ exprs: Vec<PhysicalSortExpr>,
+ set: HashSet<Arc<dyn PhysicalExpr>>,
Review Comment:
I think it would help here to mention that `set` contains pointers to the
same underlying PhysicalExpr in `exprs` and is included to allow quick
insertion check
Also it occurs to me while reviewing this code that `ConstExpr` uses
`IndexSet` instead of a Vec + HashSet. I wonder if that could be an improvement
##########
datafusion/functions-aggregate/src/first_last.rs:
##########
@@ -782,14 +756,92 @@ where
}
}
}
+
+#[derive(Debug)]
Review Comment:
This is a neat idea -- could you please add some documentation? I think the
idea is it simply takes the first value it sees (with no order)
##########
datafusion/physical-expr-common/src/sort_expr.rs:
##########
@@ -159,38 +200,17 @@ impl Display for PhysicalSortExpr {
}
}
-impl PhysicalSortExpr {
- /// evaluate the sort expression into SortColumn that can be passed into
arrow sort kernel
- pub fn evaluate_to_sort_column(&self, batch: &RecordBatch) ->
Result<SortColumn> {
- let value_to_sort = self.expr.evaluate(batch)?;
- let array_to_sort = match value_to_sort {
- ColumnarValue::Array(array) => array,
- ColumnarValue::Scalar(scalar) =>
scalar.to_array_of_size(batch.num_rows())?,
- };
- Ok(SortColumn {
- values: array_to_sort,
- options: Some(self.options),
- })
- }
-
- /// Checks whether this sort expression satisfies the given `requirement`.
- /// If sort options are unspecified in `requirement`, only expressions are
- /// compared for inequality.
- pub fn satisfy(
- &self,
- requirement: &PhysicalSortRequirement,
- schema: &Schema,
- ) -> bool {
+/// Returns whether the given two [`SortOptions`] are compatible.
Review Comment:
It would help to document what `nullable` means in this context -- I think
if it is false it means the input is known not to have nulls
##########
datafusion/core/tests/dataframe/mod.rs:
##########
@@ -2581,7 +2581,7 @@ async fn test_count_wildcard_on_sort() -> Result<()> {
| | TableScan: t1 projection=[b]
|
| physical_plan | ProjectionExec: expr=[b@0 as b, count(*)@1 as count(*)]
|
| | SortPreservingMergeExec: [count(Int64(1))@2 ASC NULLS
LAST] |
- | | SortExec: expr=[count(Int64(1))@2 ASC NULLS LAST],
preserve_partitioning=[true] |
+ | | SortExec: expr=[count(*)@1 ASC NULLS LAST],
preserve_partitioning=[true] |
Review Comment:
It is not clear to me why this test changes, but it seems like an
improvement to me
##########
datafusion/functions-aggregate/src/first_last.rs:
##########
@@ -1195,13 +1234,90 @@ impl AggregateUDFImpl for LastValue {
}
}
+#[derive(Debug)]
+pub struct TrivialLastValueAccumulator {
Review Comment:
Likewise here it would help to add a note that this is used when there is no
ordering and simply picks the last value that is seen
##########
datafusion/physical-expr-common/src/sort_expr.rs:
##########
@@ -516,162 +460,240 @@ impl Display for LexOrdering {
}
}
-impl FromIterator<PhysicalSortExpr> for LexOrdering {
- fn from_iter<T: IntoIterator<Item = PhysicalSortExpr>>(iter: T) -> Self {
- let mut lex_ordering = LexOrdering::default();
-
- for i in iter {
- lex_ordering.push(i);
- }
+impl IntoIterator for LexOrdering {
+ type Item = PhysicalSortExpr;
+ type IntoIter = IntoIter<Self::Item>;
- lex_ordering
+ fn into_iter(self) -> Self::IntoIter {
+ self.exprs.into_iter()
}
}
-impl Index<usize> for LexOrdering {
- type Output = PhysicalSortExpr;
+impl<'a> IntoIterator for &'a LexOrdering {
+ type Item = &'a PhysicalSortExpr;
+ type IntoIter = std::slice::Iter<'a, PhysicalSortExpr>;
- fn index(&self, index: usize) -> &Self::Output {
- &self.inner[index]
+ fn into_iter(self) -> Self::IntoIter {
+ self.exprs.iter()
}
}
-impl Index<Range<usize>> for LexOrdering {
- type Output = [PhysicalSortExpr];
-
- fn index(&self, range: Range<usize>) -> &Self::Output {
- &self.inner[range]
+impl From<LexOrdering> for Vec<PhysicalSortExpr> {
+ fn from(ordering: LexOrdering) -> Self {
+ ordering.exprs
}
}
-impl Index<RangeFrom<usize>> for LexOrdering {
- type Output = [PhysicalSortExpr];
+/// This object represents a lexicographical ordering requirement and contains
+/// a vector of `PhysicalSortRequirement` objects.
+///
+/// For example, a `vec![a Some(ASC), b None]` represents a lexicographical
+/// requirement that firsts imposes an ordering by column `a` in ascending
+/// order, then by column `b` in *any* (ascending or descending) order. The
+/// ordering is non-degenerate, meaning it contains at least one element, and
+/// it is duplicate-free, meaning it does not contain multiple entries for the
+/// same column.
+///
+/// Note that a `LexRequirement` need not enforce the uniqueness of its sort
+/// expressions after construction like a `LexOrdering` does, because it
provides
+/// no mutation methods. If such methods become necessary, we will need to
+/// enforce uniqueness like the latter object.
+#[derive(Debug, Clone, PartialEq)]
+pub struct LexRequirement {
+ reqs: Vec<PhysicalSortRequirement>,
+}
+
+impl LexRequirement {
+ /// Creates a new [`LexRequirement`] from the given vector of sort
expressions.
+ /// If the vector is empty, returns `None`.
+ pub fn new(reqs: impl IntoIterator<Item = PhysicalSortRequirement>) ->
Option<Self> {
Review Comment:
I think calling this `try_new` might be more "idiomatic" to signal it can
return `None`
##########
datafusion/physical-expr-common/src/sort_expr.rs:
##########
@@ -334,177 +313,142 @@ fn to_str(options: &SortOptions) -> &str {
}
}
-///`LexOrdering` contains a `Vec<PhysicalSortExpr>`, which represents
-/// a lexicographical ordering.
-///
-/// For example, `vec![a ASC, b DESC]` represents a lexicographical ordering
-/// that first sorts by column `a` in ascending order, then by column `b` in
-/// descending order.
-#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
-pub struct LexOrdering {
- inner: Vec<PhysicalSortExpr>,
-}
-
-impl AsRef<LexOrdering> for LexOrdering {
- fn as_ref(&self) -> &LexOrdering {
- self
+// Cross-conversion utilities between `PhysicalSortExpr` and
`PhysicalSortRequirement`
+impl From<PhysicalSortExpr> for PhysicalSortRequirement {
+ fn from(value: PhysicalSortExpr) -> Self {
+ Self::new(value.expr, Some(value.options))
}
}
-impl LexOrdering {
- /// Creates a new [`LexOrdering`] from a vector
- pub fn new(inner: Vec<PhysicalSortExpr>) -> Self {
- Self { inner }
- }
-
- /// Return an empty LexOrdering (no expressions)
- pub fn empty() -> &'static LexOrdering {
- static EMPTY_ORDER: LazyLock<LexOrdering> =
LazyLock::new(LexOrdering::default);
- &EMPTY_ORDER
- }
-
- /// Returns the number of elements that can be stored in the LexOrdering
- /// without reallocating.
- pub fn capacity(&self) -> usize {
- self.inner.capacity()
- }
-
- /// Clears the LexOrdering, removing all elements.
- pub fn clear(&mut self) {
- self.inner.clear()
- }
-
- /// Takes ownership of the actual vector of `PhysicalSortExpr`s in the
LexOrdering.
- pub fn take_exprs(self) -> Vec<PhysicalSortExpr> {
- self.inner
- }
-
- /// Returns `true` if the LexOrdering contains `expr`
- pub fn contains(&self, expr: &PhysicalSortExpr) -> bool {
- self.inner.contains(expr)
- }
-
- /// Add all elements from `iter` to the LexOrdering.
- pub fn extend<I: IntoIterator<Item = PhysicalSortExpr>>(&mut self, iter:
I) {
- self.inner.extend(iter)
- }
-
- /// Remove all elements from the LexOrdering where `f` evaluates to
`false`.
- pub fn retain<F>(&mut self, f: F)
- where
- F: FnMut(&PhysicalSortExpr) -> bool,
- {
- self.inner.retain(f)
- }
-
- /// Returns `true` if the LexOrdering contains no elements.
- pub fn is_empty(&self) -> bool {
- self.inner.is_empty()
- }
-
- /// Returns an iterator over each `&PhysicalSortExpr` in the LexOrdering.
- pub fn iter(&self) -> core::slice::Iter<PhysicalSortExpr> {
- self.inner.iter()
+impl From<PhysicalSortRequirement> for PhysicalSortExpr {
+ /// The default sort options `ASC, NULLS LAST` when the requirement does
+ /// not specify sort options. This default is consistent with PostgreSQL.
+ ///
+ /// Reference: <https://www.postgresql.org/docs/current/queries-order.html>
+ fn from(value: PhysicalSortRequirement) -> Self {
+ let options = value
+ .options
+ .unwrap_or_else(|| SortOptions::new(false, false));
+ Self::new(value.expr, options)
}
+}
- /// Returns the number of elements in the LexOrdering.
- pub fn len(&self) -> usize {
- self.inner.len()
- }
+/// This object represents a lexicographical ordering and contains a vector
+/// of `PhysicalSortExpr` objects.
+///
+/// For example, a `vec![a ASC, b DESC]` represents a lexicographical ordering
+/// that first sorts by column `a` in ascending order, then by column `b` in
+/// descending order. The ordering is non-degenerate, meaning it contains at
+/// least one element, and it is duplicate-free, meaning it does not contain
+/// multiple entries for the same column.
Review Comment:
Here is a minor OCD formatting suggestion:
```suggestion
/// descending order.
///
/// # Invariants
/// The following always hold true for the orderings:
/// 1. It is non-degenerate, meaning it contains at least one element
/// 2. It is duplicate-free, meaning it does not contain multiple entries
for the same column.
```
##########
datafusion/physical-expr-common/src/sort_expr.rs:
##########
@@ -516,162 +460,240 @@ impl Display for LexOrdering {
}
}
-impl FromIterator<PhysicalSortExpr> for LexOrdering {
- fn from_iter<T: IntoIterator<Item = PhysicalSortExpr>>(iter: T) -> Self {
- let mut lex_ordering = LexOrdering::default();
-
- for i in iter {
- lex_ordering.push(i);
- }
+impl IntoIterator for LexOrdering {
+ type Item = PhysicalSortExpr;
+ type IntoIter = IntoIter<Self::Item>;
- lex_ordering
+ fn into_iter(self) -> Self::IntoIter {
+ self.exprs.into_iter()
}
}
-impl Index<usize> for LexOrdering {
- type Output = PhysicalSortExpr;
+impl<'a> IntoIterator for &'a LexOrdering {
+ type Item = &'a PhysicalSortExpr;
+ type IntoIter = std::slice::Iter<'a, PhysicalSortExpr>;
- fn index(&self, index: usize) -> &Self::Output {
- &self.inner[index]
+ fn into_iter(self) -> Self::IntoIter {
+ self.exprs.iter()
}
}
-impl Index<Range<usize>> for LexOrdering {
- type Output = [PhysicalSortExpr];
-
- fn index(&self, range: Range<usize>) -> &Self::Output {
- &self.inner[range]
+impl From<LexOrdering> for Vec<PhysicalSortExpr> {
+ fn from(ordering: LexOrdering) -> Self {
+ ordering.exprs
}
}
-impl Index<RangeFrom<usize>> for LexOrdering {
- type Output = [PhysicalSortExpr];
+/// This object represents a lexicographical ordering requirement and contains
+/// a vector of `PhysicalSortRequirement` objects.
+///
+/// For example, a `vec![a Some(ASC), b None]` represents a lexicographical
+/// requirement that firsts imposes an ordering by column `a` in ascending
+/// order, then by column `b` in *any* (ascending or descending) order. The
+/// ordering is non-degenerate, meaning it contains at least one element, and
+/// it is duplicate-free, meaning it does not contain multiple entries for the
+/// same column.
+///
+/// Note that a `LexRequirement` need not enforce the uniqueness of its sort
+/// expressions after construction like a `LexOrdering` does, because it
provides
+/// no mutation methods. If such methods become necessary, we will need to
+/// enforce uniqueness like the latter object.
+#[derive(Debug, Clone, PartialEq)]
+pub struct LexRequirement {
+ reqs: Vec<PhysicalSortRequirement>,
+}
+
+impl LexRequirement {
+ /// Creates a new [`LexRequirement`] from the given vector of sort
expressions.
+ /// If the vector is empty, returns `None`.
+ pub fn new(reqs: impl IntoIterator<Item = PhysicalSortRequirement>) ->
Option<Self> {
+ let (non_empty, requirements) = Self::construct(reqs);
+ non_empty.then_some(requirements)
+ }
+
+ /// Returns the leading `PhysicalSortRequirement` of the `LexRequirement`.
+ /// Note that this function does not return an `Option`, as a
`LexRequirement`
+ /// is always non-degenerate (i.e. it contains at least one element).
+ pub fn first(&self) -> &PhysicalSortRequirement {
+ // Can safely `unwrap` because `LexRequirement` is non-degenerate:
+ self.reqs.first().unwrap()
+ }
+
+ /// Constructs a new `LexRequirement` from the given sort requirements w/o
+ /// enforcing non-degeneracy. This function is used internally and is not
+ /// meant (or safe) for external use.
+ fn construct(
+ reqs: impl IntoIterator<Item = PhysicalSortRequirement>,
+ ) -> (bool, Self) {
+ let mut set = HashSet::new();
+ let reqs = reqs
+ .into_iter()
+ .filter_map(|r| set.insert(Arc::clone(&r.expr)).then_some(r))
+ .collect();
+ (!set.is_empty(), Self { reqs })
+ }
+}
- fn index(&self, range_from: RangeFrom<usize>) -> &Self::Output {
- &self.inner[range_from]
+impl<const N: usize> From<[PhysicalSortRequirement; N]> for LexRequirement {
+ fn from(value: [PhysicalSortRequirement; N]) -> Self {
+ // TODO: Replace this assertion with a condition on the generic
parameter
+ // when Rust supports it.
+ assert!(N > 0);
+ let (non_empty, requirement) = Self::construct(value);
+ debug_assert!(non_empty);
+ requirement
}
}
-impl Index<RangeTo<usize>> for LexOrdering {
- type Output = [PhysicalSortExpr];
+impl Deref for LexRequirement {
+ type Target = [PhysicalSortRequirement];
- fn index(&self, range_to: RangeTo<usize>) -> &Self::Output {
- &self.inner[range_to]
+ fn deref(&self) -> &Self::Target {
+ self.reqs.as_slice()
}
}
-impl IntoIterator for LexOrdering {
- type Item = PhysicalSortExpr;
- type IntoIter = IntoIter<PhysicalSortExpr>;
+impl IntoIterator for LexRequirement {
+ type Item = PhysicalSortRequirement;
+ type IntoIter = IntoIter<Self::Item>;
fn into_iter(self) -> Self::IntoIter {
- self.inner.into_iter()
+ self.reqs.into_iter()
}
}
-///`LexOrderingRef` is an alias for the type &`[PhysicalSortExpr]`, which
represents
-/// a reference to a lexicographical ordering.
-#[deprecated(since = "43.0.0", note = "use &LexOrdering instead")]
-pub type LexOrderingRef<'a> = &'a [PhysicalSortExpr];
+impl<'a> IntoIterator for &'a LexRequirement {
+ type Item = &'a PhysicalSortRequirement;
+ type IntoIter = std::slice::Iter<'a, PhysicalSortRequirement>;
-///`LexRequirement` is an struct containing a `Vec<PhysicalSortRequirement>`,
which
-/// represents a lexicographical ordering requirement.
-#[derive(Debug, Default, Clone, PartialEq)]
-pub struct LexRequirement {
- pub inner: Vec<PhysicalSortRequirement>,
+ fn into_iter(self) -> Self::IntoIter {
+ self.reqs.iter()
+ }
}
-impl LexRequirement {
- pub fn new(inner: Vec<PhysicalSortRequirement>) -> Self {
- Self { inner }
+impl From<LexRequirement> for Vec<PhysicalSortRequirement> {
+ fn from(requirement: LexRequirement) -> Self {
+ requirement.reqs
}
+}
- pub fn is_empty(&self) -> bool {
- self.inner.is_empty()
+// Cross-conversion utilities between `LexOrdering` and `LexRequirement`
+impl From<LexOrdering> for LexRequirement {
+ fn from(value: LexOrdering) -> Self {
+ // Can construct directly as `value` is non-degenerate:
+ let (non_empty, requirements) =
+ Self::construct(value.into_iter().map(Into::into));
+ debug_assert!(non_empty);
+ requirements
}
+}
- pub fn iter(&self) -> impl Iterator<Item = &PhysicalSortRequirement> {
- self.inner.iter()
+impl From<LexRequirement> for LexOrdering {
+ fn from(value: LexRequirement) -> Self {
+ // Can construct directly as `value` is non-degenerate:
+ let (non_empty, ordering) =
Self::construct(value.into_iter().map(Into::into));
+ debug_assert!(non_empty);
+ ordering
+ }
+}
+
+/// Represents a plan's input ordering requirements. Vector elements represent
Review Comment:
This is so much nicer 😍
Are there any restrictions on the requirements such as that there are no
duplicate elements or that they are not empty? It seems the intention is that
`Option<OrderingRequirement>` is used to represent no ordering requirement, and
I think all the APIs require at least one LexRequirement.
It might be good to document that as another restriction
##########
datafusion/physical-expr-common/src/sort_expr.rs:
##########
@@ -334,177 +313,142 @@ fn to_str(options: &SortOptions) -> &str {
}
}
-///`LexOrdering` contains a `Vec<PhysicalSortExpr>`, which represents
-/// a lexicographical ordering.
-///
-/// For example, `vec![a ASC, b DESC]` represents a lexicographical ordering
-/// that first sorts by column `a` in ascending order, then by column `b` in
-/// descending order.
-#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
-pub struct LexOrdering {
- inner: Vec<PhysicalSortExpr>,
-}
-
-impl AsRef<LexOrdering> for LexOrdering {
- fn as_ref(&self) -> &LexOrdering {
- self
+// Cross-conversion utilities between `PhysicalSortExpr` and
`PhysicalSortRequirement`
+impl From<PhysicalSortExpr> for PhysicalSortRequirement {
+ fn from(value: PhysicalSortExpr) -> Self {
+ Self::new(value.expr, Some(value.options))
}
}
-impl LexOrdering {
- /// Creates a new [`LexOrdering`] from a vector
- pub fn new(inner: Vec<PhysicalSortExpr>) -> Self {
- Self { inner }
- }
-
- /// Return an empty LexOrdering (no expressions)
- pub fn empty() -> &'static LexOrdering {
- static EMPTY_ORDER: LazyLock<LexOrdering> =
LazyLock::new(LexOrdering::default);
- &EMPTY_ORDER
- }
-
- /// Returns the number of elements that can be stored in the LexOrdering
- /// without reallocating.
- pub fn capacity(&self) -> usize {
- self.inner.capacity()
- }
-
- /// Clears the LexOrdering, removing all elements.
- pub fn clear(&mut self) {
- self.inner.clear()
- }
-
- /// Takes ownership of the actual vector of `PhysicalSortExpr`s in the
LexOrdering.
- pub fn take_exprs(self) -> Vec<PhysicalSortExpr> {
- self.inner
- }
-
- /// Returns `true` if the LexOrdering contains `expr`
- pub fn contains(&self, expr: &PhysicalSortExpr) -> bool {
- self.inner.contains(expr)
- }
-
- /// Add all elements from `iter` to the LexOrdering.
- pub fn extend<I: IntoIterator<Item = PhysicalSortExpr>>(&mut self, iter:
I) {
- self.inner.extend(iter)
- }
-
- /// Remove all elements from the LexOrdering where `f` evaluates to
`false`.
- pub fn retain<F>(&mut self, f: F)
- where
- F: FnMut(&PhysicalSortExpr) -> bool,
- {
- self.inner.retain(f)
- }
-
- /// Returns `true` if the LexOrdering contains no elements.
- pub fn is_empty(&self) -> bool {
- self.inner.is_empty()
- }
-
- /// Returns an iterator over each `&PhysicalSortExpr` in the LexOrdering.
- pub fn iter(&self) -> core::slice::Iter<PhysicalSortExpr> {
- self.inner.iter()
+impl From<PhysicalSortRequirement> for PhysicalSortExpr {
+ /// The default sort options `ASC, NULLS LAST` when the requirement does
+ /// not specify sort options. This default is consistent with PostgreSQL.
+ ///
+ /// Reference: <https://www.postgresql.org/docs/current/queries-order.html>
+ fn from(value: PhysicalSortRequirement) -> Self {
+ let options = value
+ .options
+ .unwrap_or_else(|| SortOptions::new(false, false));
+ Self::new(value.expr, options)
}
+}
- /// Returns the number of elements in the LexOrdering.
- pub fn len(&self) -> usize {
- self.inner.len()
- }
+/// This object represents a lexicographical ordering and contains a vector
+/// of `PhysicalSortExpr` objects.
+///
+/// For example, a `vec![a ASC, b DESC]` represents a lexicographical ordering
+/// that first sorts by column `a` in ascending order, then by column `b` in
+/// descending order. The ordering is non-degenerate, meaning it contains at
+/// least one element, and it is duplicate-free, meaning it does not contain
+/// multiple entries for the same column.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct LexOrdering {
+ exprs: Vec<PhysicalSortExpr>,
+ set: HashSet<Arc<dyn PhysicalExpr>>,
+}
- /// Removes the last element from the LexOrdering and returns it, or
`None` if it is empty.
- pub fn pop(&mut self) -> Option<PhysicalSortExpr> {
- self.inner.pop()
+impl LexOrdering {
+ /// Creates a new [`LexOrdering`] from the given vector of sort
expressions.
+ /// If the vector is empty, returns `None`.
+ pub fn new(exprs: impl IntoIterator<Item = PhysicalSortExpr>) ->
Option<Self> {
+ let (non_empty, ordering) = Self::construct(exprs);
+ non_empty.then_some(ordering)
}
- /// Appends an element to the back of the LexOrdering.
- pub fn push(&mut self, physical_sort_expr: PhysicalSortExpr) {
- self.inner.push(physical_sort_expr)
+ /// Appends an element to the back of the `LexOrdering`.
+ pub fn push(&mut self, sort_expr: PhysicalSortExpr) {
+ if self.set.insert(Arc::clone(&sort_expr.expr)) {
+ self.exprs.push(sort_expr);
+ }
}
- /// Truncates the LexOrdering, keeping only the first `len` elements.
- pub fn truncate(&mut self, len: usize) {
- self.inner.truncate(len)
+ /// Add all elements from `iter` to the `LexOrdering`.
+ pub fn extend(&mut self, sort_exprs: impl IntoIterator<Item =
PhysicalSortExpr>) {
+ for sort_expr in sort_exprs {
+ if self.set.insert(Arc::clone(&sort_expr.expr)) {
+ self.exprs.push(sort_expr);
+ }
Review Comment:
Very minor: I think this is the same as `push`:
```suggestion
self.push(sort_expr);
```
##########
datafusion/physical-expr/src/equivalence/ordering.rs:
##########
@@ -164,51 +124,40 @@ impl OrderingEquivalenceClass {
/// For example, if `orderings[idx]` is `[a ASC, b ASC, c DESC]` and
/// `orderings[pre_idx]` is `[b ASC, c DESC]`, then the function will trim
/// `orderings[idx]` to `[a ASC]`.
- fn resolve_overlap(&mut self, idx: usize, pre_idx: usize) -> bool {
+ fn resolve_overlap(&mut self, idx: usize, pre_idx: usize) -> Option<bool> {
Review Comment:
it would help to update this doc with what returning `None` means (it looks
like it means there was no overlap? and `Some(res)` means there was overlap and
`res` is if the ordering was actually truncated)
##########
datafusion/sqllogictest/test_files/topk.slt:
##########
@@ -380,7 +380,7 @@ explain select number + 1 as number_plus, number, number +
1 as other_number_plu
----
physical_plan
01)SortPreservingMergeExec: [number_plus@0 DESC, number@1 DESC,
other_number_plus@2 DESC, age@3 ASC NULLS LAST], fetch=3
-02)--SortExec: TopK(fetch=3), expr=[number_plus@0 DESC, number@1 DESC,
other_number_plus@2 DESC, age@3 ASC NULLS LAST], preserve_partitioning=[true],
sort_prefix=[number_plus@0 DESC, number@1 DESC]
+02)--SortExec: TopK(fetch=3), expr=[number_plus@0 DESC, number@1 DESC, age@3
ASC NULLS LAST], preserve_partitioning=[true], sort_prefix=[number_plus@0 DESC,
number@1 DESC]
Review Comment:
likewise this seems like an improvment
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -1053,25 +1051,9 @@ impl TableProvider for ListingTable {
file_extension: self.options().format.get_ext(),
};
- let order_requirements = if !self.options().file_sort_order.is_empty()
{
- // Multiple sort orders in outer vec are equivalent, so we pass
only the first one
- let orderings = self.try_create_output_ordering()?;
- let Some(ordering) = orderings.first() else {
- return internal_err!(
- "Expected ListingTable to have a sort order, but none
found!"
- );
- };
- // Converts Vec<Vec<SortExpr>> into type required by execution
plan to specify its required input ordering
- Some(LexRequirement::new(
- ordering
- .into_iter()
- .cloned()
- .map(PhysicalSortRequirement::from)
- .collect::<Vec<_>>(),
- ))
- } else {
- None
- };
+ let orderings = self.try_create_output_ordering()?;
+ // It is sufficient to pass only one of the equivalent orderings:
Review Comment:
maybe in the future (not this PR) the code could pass all the orderings and
let the writer decide what to use
##########
datafusion/core/tests/physical_optimizer/enforce_sorting.rs:
##########
@@ -210,24 +175,27 @@ async fn test_remove_unnecessary_sort5() -> Result<()> {
let left_schema = create_test_schema2()?;
let right_schema = create_test_schema3()?;
let left_input = memory_exec(&left_schema);
- let parquet_sort_exprs = vec![sort_expr("a", &right_schema)];
- let right_input = parquet_exec_sorted(&right_schema, parquet_sort_exprs);
-
+ let parquet_ordering = [sort_expr("a", &right_schema)].into();
+ let right_input =
+ parquet_exec_with_sort(right_schema.clone(), vec![parquet_ordering]);
let on = vec![(
Arc::new(Column::new_with_schema("col_a", &left_schema)?) as _,
Arc::new(Column::new_with_schema("c", &right_schema)?) as _,
)];
let join = hash_join_exec(left_input, right_input, on, None,
&JoinType::Inner)?;
- let physical_plan = sort_exec(vec![sort_expr("a", &join.schema())], join);
+ let physical_plan = sort_exec([sort_expr("a", &join.schema())].into(),
join);
- let expected_input = ["SortExec: expr=[a@2 ASC],
preserve_partitioning=[false]",
+ let expected_input = [
Review Comment:
It is unfortunate we haven't rewritten these tests to insta yet
##########
datafusion/core/tests/physical_optimizer/enforce_distribution.rs:
##########
@@ -3208,10 +3215,11 @@ fn do_not_preserve_ordering_through_repartition() ->
Result<()> {
#[test]
fn no_need_for_sort_after_filter() -> Result<()> {
let schema = schema();
- let sort_key = LexOrdering::new(vec![PhysicalSortExpr {
- expr: col("c", &schema).unwrap(),
+ let sort_key: LexOrdering = [PhysicalSortExpr {
Review Comment:
FWIW another way to write this that i find sometimes clearer is:
```rst
let sort_key = LexOrdering::from([PhysicalSortExpr { ...}]);
```
It also often avoids an extra line.
Totally not needed for this PR -- I am just pointing it out
##########
datafusion/physical-expr-common/src/sort_expr.rs:
##########
@@ -516,162 +460,240 @@ impl Display for LexOrdering {
}
}
-impl FromIterator<PhysicalSortExpr> for LexOrdering {
- fn from_iter<T: IntoIterator<Item = PhysicalSortExpr>>(iter: T) -> Self {
- let mut lex_ordering = LexOrdering::default();
-
- for i in iter {
- lex_ordering.push(i);
- }
+impl IntoIterator for LexOrdering {
+ type Item = PhysicalSortExpr;
+ type IntoIter = IntoIter<Self::Item>;
- lex_ordering
+ fn into_iter(self) -> Self::IntoIter {
+ self.exprs.into_iter()
}
}
-impl Index<usize> for LexOrdering {
- type Output = PhysicalSortExpr;
+impl<'a> IntoIterator for &'a LexOrdering {
+ type Item = &'a PhysicalSortExpr;
+ type IntoIter = std::slice::Iter<'a, PhysicalSortExpr>;
- fn index(&self, index: usize) -> &Self::Output {
- &self.inner[index]
+ fn into_iter(self) -> Self::IntoIter {
+ self.exprs.iter()
}
}
-impl Index<Range<usize>> for LexOrdering {
- type Output = [PhysicalSortExpr];
-
- fn index(&self, range: Range<usize>) -> &Self::Output {
- &self.inner[range]
+impl From<LexOrdering> for Vec<PhysicalSortExpr> {
+ fn from(ordering: LexOrdering) -> Self {
+ ordering.exprs
}
}
-impl Index<RangeFrom<usize>> for LexOrdering {
- type Output = [PhysicalSortExpr];
+/// This object represents a lexicographical ordering requirement and contains
+/// a vector of `PhysicalSortRequirement` objects.
+///
+/// For example, a `vec![a Some(ASC), b None]` represents a lexicographical
+/// requirement that firsts imposes an ordering by column `a` in ascending
+/// order, then by column `b` in *any* (ascending or descending) order. The
+/// ordering is non-degenerate, meaning it contains at least one element, and
+/// it is duplicate-free, meaning it does not contain multiple entries for the
+/// same column.
+///
+/// Note that a `LexRequirement` need not enforce the uniqueness of its sort
+/// expressions after construction like a `LexOrdering` does, because it
provides
+/// no mutation methods. If such methods become necessary, we will need to
+/// enforce uniqueness like the latter object.
+#[derive(Debug, Clone, PartialEq)]
+pub struct LexRequirement {
+ reqs: Vec<PhysicalSortRequirement>,
+}
+
+impl LexRequirement {
+ /// Creates a new [`LexRequirement`] from the given vector of sort
expressions.
+ /// If the vector is empty, returns `None`.
+ pub fn new(reqs: impl IntoIterator<Item = PhysicalSortRequirement>) ->
Option<Self> {
+ let (non_empty, requirements) = Self::construct(reqs);
+ non_empty.then_some(requirements)
+ }
+
+ /// Returns the leading `PhysicalSortRequirement` of the `LexRequirement`.
+ /// Note that this function does not return an `Option`, as a
`LexRequirement`
+ /// is always non-degenerate (i.e. it contains at least one element).
+ pub fn first(&self) -> &PhysicalSortRequirement {
+ // Can safely `unwrap` because `LexRequirement` is non-degenerate:
+ self.reqs.first().unwrap()
+ }
+
+ /// Constructs a new `LexRequirement` from the given sort requirements w/o
+ /// enforcing non-degeneracy. This function is used internally and is not
+ /// meant (or safe) for external use.
+ fn construct(
+ reqs: impl IntoIterator<Item = PhysicalSortRequirement>,
+ ) -> (bool, Self) {
+ let mut set = HashSet::new();
+ let reqs = reqs
+ .into_iter()
+ .filter_map(|r| set.insert(Arc::clone(&r.expr)).then_some(r))
+ .collect();
+ (!set.is_empty(), Self { reqs })
+ }
+}
- fn index(&self, range_from: RangeFrom<usize>) -> &Self::Output {
- &self.inner[range_from]
+impl<const N: usize> From<[PhysicalSortRequirement; N]> for LexRequirement {
+ fn from(value: [PhysicalSortRequirement; N]) -> Self {
+ // TODO: Replace this assertion with a condition on the generic
parameter
+ // when Rust supports it.
+ assert!(N > 0);
+ let (non_empty, requirement) = Self::construct(value);
+ debug_assert!(non_empty);
+ requirement
}
}
-impl Index<RangeTo<usize>> for LexOrdering {
- type Output = [PhysicalSortExpr];
+impl Deref for LexRequirement {
+ type Target = [PhysicalSortRequirement];
- fn index(&self, range_to: RangeTo<usize>) -> &Self::Output {
- &self.inner[range_to]
+ fn deref(&self) -> &Self::Target {
+ self.reqs.as_slice()
}
}
-impl IntoIterator for LexOrdering {
- type Item = PhysicalSortExpr;
- type IntoIter = IntoIter<PhysicalSortExpr>;
+impl IntoIterator for LexRequirement {
+ type Item = PhysicalSortRequirement;
+ type IntoIter = IntoIter<Self::Item>;
fn into_iter(self) -> Self::IntoIter {
- self.inner.into_iter()
+ self.reqs.into_iter()
}
}
-///`LexOrderingRef` is an alias for the type &`[PhysicalSortExpr]`, which
represents
-/// a reference to a lexicographical ordering.
-#[deprecated(since = "43.0.0", note = "use &LexOrdering instead")]
-pub type LexOrderingRef<'a> = &'a [PhysicalSortExpr];
+impl<'a> IntoIterator for &'a LexRequirement {
+ type Item = &'a PhysicalSortRequirement;
+ type IntoIter = std::slice::Iter<'a, PhysicalSortRequirement>;
-///`LexRequirement` is an struct containing a `Vec<PhysicalSortRequirement>`,
which
-/// represents a lexicographical ordering requirement.
-#[derive(Debug, Default, Clone, PartialEq)]
-pub struct LexRequirement {
- pub inner: Vec<PhysicalSortRequirement>,
+ fn into_iter(self) -> Self::IntoIter {
+ self.reqs.iter()
+ }
}
-impl LexRequirement {
- pub fn new(inner: Vec<PhysicalSortRequirement>) -> Self {
- Self { inner }
+impl From<LexRequirement> for Vec<PhysicalSortRequirement> {
+ fn from(requirement: LexRequirement) -> Self {
+ requirement.reqs
}
+}
- pub fn is_empty(&self) -> bool {
- self.inner.is_empty()
+// Cross-conversion utilities between `LexOrdering` and `LexRequirement`
+impl From<LexOrdering> for LexRequirement {
+ fn from(value: LexOrdering) -> Self {
+ // Can construct directly as `value` is non-degenerate:
+ let (non_empty, requirements) =
+ Self::construct(value.into_iter().map(Into::into));
+ debug_assert!(non_empty);
+ requirements
}
+}
- pub fn iter(&self) -> impl Iterator<Item = &PhysicalSortRequirement> {
- self.inner.iter()
+impl From<LexRequirement> for LexOrdering {
+ fn from(value: LexRequirement) -> Self {
+ // Can construct directly as `value` is non-degenerate:
+ let (non_empty, ordering) =
Self::construct(value.into_iter().map(Into::into));
+ debug_assert!(non_empty);
+ ordering
+ }
+}
+
+/// Represents a plan's input ordering requirements. Vector elements represent
+/// alternative ordering requirements in the order of preference.
+#[derive(Debug, Clone, PartialEq)]
+pub enum OrderingRequirements {
+ /// The operator is not able to work without one of these requirements.
+ Hard(Vec<LexRequirement>),
+ /// The operator can benefit from these input orderings when available,
+ /// but can still work in the absence of any input ordering.
+ Soft(Vec<LexRequirement>),
+}
+
+impl OrderingRequirements {
+ /// Creates a new instance from the given alternatives. If an empty list of
+ /// alternatives are given, returns `None`.
+ pub fn new_alternatives(
+ alternatives: impl IntoIterator<Item = LexRequirement>,
+ soft: bool,
+ ) -> Option<Self> {
+ let alternatives = alternatives.into_iter().collect::<Vec<_>>();
+ (!alternatives.is_empty()).then(|| {
+ if soft {
+ Self::Soft(alternatives)
+ } else {
+ Self::Hard(alternatives)
+ }
+ })
}
- pub fn push(&mut self, physical_sort_requirement: PhysicalSortRequirement)
{
- self.inner.push(physical_sort_requirement)
+ /// Creates a new instance with a single hard requirement.
+ pub fn new(requirement: LexRequirement) -> Self {
+ Self::Hard(vec![requirement])
}
- /// Create a new [`LexRequirement`] from a [`LexOrdering`]
- ///
- /// Returns [`LexRequirement`] that requires the exact
- /// sort of the [`PhysicalSortExpr`]s in `ordering`
- pub fn from_lex_ordering(ordering: LexOrdering) -> Self {
- Self::new(
- ordering
- .into_iter()
- .map(PhysicalSortRequirement::from)
- .collect(),
- )
+ /// Creates a new instance with a single soft requirement.
+ pub fn new_soft(requirement: LexRequirement) -> Self {
+ Self::Soft(vec![requirement])
}
- /// Constructs a duplicate-free `LexOrderingReq` by filtering out
- /// duplicate entries that have same physical expression inside.
- ///
- /// For example, `vec![a Some(ASC), a Some(DESC)]` collapses to `vec![a
- /// Some(ASC)]`.
- pub fn collapse(self) -> Self {
- let mut output = Vec::<PhysicalSortRequirement>::new();
- for item in self {
- if !output.iter().any(|req| req.expr.eq(&item.expr)) {
- output.push(item);
- }
+ /// Adds an alternative requirement to the list of alternatives.
+ pub fn add_alternative(&mut self, requirement: LexRequirement) {
+ match self {
+ Self::Hard(alts) | Self::Soft(alts) => alts.push(requirement),
}
- LexRequirement::new(output)
}
-}
-impl From<LexOrdering> for LexRequirement {
- fn from(value: LexOrdering) -> Self {
- Self::from_lex_ordering(value)
+ /// Returns the first (i.e. most preferred) `LexRequirement` among
+ /// alternative requirements.
+ pub fn into_single(self) -> LexRequirement {
+ match self {
+ Self::Hard(mut alts) | Self::Soft(mut alts) => alts.swap_remove(0),
+ }
}
-}
-impl Deref for LexRequirement {
- type Target = [PhysicalSortRequirement];
-
- fn deref(&self) -> &Self::Target {
- self.inner.as_slice()
+ /// Returns a reference to the first (i.e. most preferred) `LexRequirement`
+ /// among alternative requirements.
+ pub fn first(&self) -> &LexRequirement {
+ match self {
+ Self::Hard(alts) | Self::Soft(alts) => &alts[0],
+ }
}
-}
-
-impl FromIterator<PhysicalSortRequirement> for LexRequirement {
- fn from_iter<T: IntoIterator<Item = PhysicalSortRequirement>>(iter: T) ->
Self {
- let mut lex_requirement = LexRequirement::new(vec![]);
- for i in iter {
- lex_requirement.inner.push(i);
+ /// Returns all alternatives as a vector of `LexRequirement` objects and a
+ /// boolean value indicating softness/hardness of the requirements.
+ pub fn get_alternatives(self) -> (Vec<LexRequirement>, bool) {
Review Comment:
Maybe this could be called `into_alternates` to signal it consumes `self`
##########
datafusion/physical-expr/src/equivalence/ordering.rs:
##########
@@ -16,115 +16,76 @@
// under the License.
use std::fmt::Display;
-use std::hash::Hash;
+use std::ops::Deref;
use std::sync::Arc;
use std::vec::IntoIter;
-use crate::equivalence::add_offset_to_expr;
-use crate::{LexOrdering, PhysicalExpr};
+use crate::expressions::with_new_schema;
+use crate::{add_offset_to_physical_sort_exprs, LexOrdering, PhysicalExpr};
use arrow::compute::SortOptions;
-use datafusion_common::HashSet;
+use arrow::datatypes::SchemaRef;
+use datafusion_common::{HashSet, Result};
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
-/// An `OrderingEquivalenceClass` object keeps track of different alternative
-/// orderings than can describe a schema. For example, consider the following
table:
+/// An `OrderingEquivalenceClass` keeps track of distinct alternative orderings
+/// than can describe a table. For example, consider the following table:
///
/// ```text
-/// |a|b|c|d|
-/// |1|4|3|1|
-/// |2|3|3|2|
-/// |3|1|2|2|
-/// |3|2|1|3|
+/// ┌───┬───┬───┬───┐
+/// │ a │ b │ c │ d │
+/// ├───┼───┼───┼───┤
+/// │ 1 │ 4 │ 3 │ 1 │
+/// │ 2 │ 3 │ 3 │ 2 │
+/// │ 3 │ 1 │ 2 │ 2 │
+/// │ 3 │ 2 │ 1 │ 3 │
+/// └───┴───┴───┴───┘
/// ```
///
-/// Here, both `vec![a ASC, b ASC]` and `vec![c DESC, d ASC]` describe the
table
+/// Here, both `[a ASC, b ASC]` and `[c DESC, d ASC]` describe the table
/// ordering. In this case, we say that these orderings are equivalent.
Review Comment:
I wonder would it help here to describe any requirements related to
`orderings`? Specifically what does an empty `ordering` mean?
##########
datafusion/sqllogictest/test_files/topk.slt:
##########
@@ -370,7 +370,7 @@ query TT
explain select number, letter, age, number as column4, letter as column5 from
partial_sorted order by number desc, column4 desc, letter asc, column5 asc, age
desc limit 3;
----
physical_plan
-01)SortExec: TopK(fetch=3), expr=[number@0 DESC, column4@3 DESC, letter@1 ASC
NULLS LAST, column5@4 ASC NULLS LAST, age@2 DESC],
preserve_partitioning=[false], sort_prefix=[number@0 DESC, letter@1 ASC NULLS
LAST]
+01)SortExec: TopK(fetch=3), expr=[number@0 DESC, letter@1 ASC NULLS LAST,
age@2 DESC], preserve_partitioning=[false], sort_prefix=[number@0 DESC,
letter@1 ASC NULLS LAST]
Review Comment:
this seems like a plan improvement 👍 as now there are fewer exprs in the
sort expression (given the comments in the test, it seems like this was the
expected results anyways)
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -173,9 +173,6 @@ SELECT approx_percentile_cont(c12) WITHIN GROUP (ORDER BY
c12) FROM aggregate_te
statement error DataFusion error: This feature is not implemented: Tdigest
max_size value for 'APPROX_PERCENTILE_CONT' must be a literal
SELECT approx_percentile_cont(0.95, c5) WITHIN GROUP (ORDER BY c12) FROM
aggregate_test_100
-statement error DataFusion error: This feature is not implemented: Conflicting
ordering requirements in aggregate functions is not supported
Review Comment:
Why was this test removed? I think it is still a valid test isn't it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For additional commands, e-mail: github-h...@datafusion.apache.org
