You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "ozankabak (via GitHub)" <gi...@apache.org> on 2025/05/30 13:17:06 UTC
[PR] [MAJOREquivalence System Overhaul [datafusion]
ozankabak opened a new pull request, #16217:
URL: https://github.com/apache/datafusion/pull/16217
## Which issue does this PR close?
- Closes #13748.
## Rationale for this change
DF's theoretical approach to representing orderings, equivalences and constants is sound, but it suffers from many implementation problems that lead to various issues. Some of these issues include:
- High-complexity planning times for queries involving UNION and ORDER BY clauses (see the referred issue).
- A continuous stream of subtle bugs that arise from the fact that our objects allow multiple representations of the same underlying state (e.g. a non-existing ordering is sometimes represented by a `None` value for a type `Option<LexOrdering>`, and sometimes with a `Some` value with the payload being "empty").
- The inability to relay ordering preference information to optimizer rules such as `EnforceSorting`, which results in missed optimization opportunities. For example, a windowing operator prefers its input to be sorted w.r.t. both PARTITION BY and ORDER BY keys, but if the former is not available, it is still preferable to have just the latter instead of a fully unsorted input.
## What changes are included in this PR?
- The equivalence system is rewritten to solve the computational complexity problem. The new system avoids handling constants separately from other equivalences and treats them on an equal footing within the same mechanism, avoiding unnecessary work during `ordering_satisfy` checks. Also, the new system caches normalization results for equivalent orderings, which was one of the major sources of excessive computational complexity during sort enforcement.
- The `LexOrdering` and `LexRequirement` objects do not allow degenerate orderings anymore, whenever you see this object, you can be sure there will be *some* ordering. Possible absence of an ordering will henceforth be properly represented by an `Option<LexOrdering>`.
- The `OrderingRequirements` object is introduced to capture ordering preferences of operators.
## Are these changes tested?
Yes.
## Are there any user-facing changes?
Yes -- this is a major API change and will break old code. We spent months looking for ways to (1) break this PR down into smaller ones, (2) yet lump all API changes in one PR -- but couldn't find an effective way to do this.
Therefore, I invite all contributors who are also downstream stakeholders to help by participating in a "pre-scream" test. Please test and report how this PR breaks your code, so we can collectively prepare an upgrade guide for the community at large.
@alamb, @andygrove, @berkaysynnada -- feel free to tag others.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2123545316
##########
datafusion/physical-expr/src/equivalence/properties/dependency.rs:
##########
@@ -907,72 +796,13 @@ mod tests {
for (exprs, expected) in test_cases {
let exprs = exprs.into_iter().cloned().collect::<Vec<_>>();
let expected = convert_to_sort_exprs(&expected);
- let (actual, _) = eq_properties.find_longest_permutation(&exprs);
+ let (actual, _) = eq_properties.find_longest_permutation(&exprs)?;
assert_eq!(actual, expected);
}
Ok(())
}
- #[test]
- fn test_get_finer() -> Result<()> {
Review Comment:
Note to other reviewers, this test appears to have been moved down farther in the file and renamed to `test_requirements_compatible` (not removed)
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -125,40 +120,85 @@ use itertools::Itertools;
/// # let col_c = col("c", &schema).unwrap();
/// // This object represents data that is sorted by a ASC, c DESC
/// // with a single constant value of b
-/// let mut eq_properties = EquivalenceProperties::new(schema)
-/// .with_constants(vec![ConstExpr::from(col_b)]);
-/// eq_properties.add_new_ordering(LexOrdering::new(vec![
+/// let mut eq_properties = EquivalenceProperties::new(schema);
+/// eq_properties.add_constants(vec![ConstExpr::from(col_b)]);
+/// eq_properties.add_ordering([
/// PhysicalSortExpr::new_default(col_a).asc(),
/// PhysicalSortExpr::new_default(col_c).desc(),
-/// ]));
+/// ]);
///
-/// assert_eq!(eq_properties.to_string(), "order: [[a@0 ASC, c@2 DESC]], const: [b@1(heterogeneous)]")
+/// assert_eq!(eq_properties.to_string(), "order: [[a@0 ASC, c@2 DESC]], eq: [{members: [b@1], constant: (heterogeneous)}]");
/// ```
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug)]
pub struct EquivalenceProperties {
- /// Distinct equivalence classes (exprs known to have the same expressions)
+ /// Distinct equivalence classes (i.e. expressions with the same value).
eq_group: EquivalenceGroup,
- /// Equivalent sort expressions
+ /// Equivalent sort expressions (i.e. those define the same ordering).
oeq_class: OrderingEquivalenceClass,
- /// Expressions whose values are constant
- ///
- /// TODO: We do not need to track constants separately, they can be tracked
- /// inside `eq_group` as `Literal` expressions.
- constants: Vec<ConstExpr>,
- /// Table constraints
+ /// Cache storing equivalent sort expressions in normal form (i.e. without
+ /// constants/duplicates and in standard form) and a map associating leading
+ /// terms with full sort expressions.
+ oeq_cache: OrderingEquivalenceCache,
Review Comment:
👍
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -190,382 +241,363 @@ impl EquivalenceProperties {
&self.oeq_class
}
- /// Return the inner OrderingEquivalenceClass, consuming self
- pub fn into_oeq_class(self) -> OrderingEquivalenceClass {
- self.oeq_class
- }
-
/// Returns a reference to the equivalence group within.
pub fn eq_group(&self) -> &EquivalenceGroup {
&self.eq_group
}
- /// Returns a reference to the constant expressions
- pub fn constants(&self) -> &[ConstExpr] {
- &self.constants
- }
-
+ /// Returns a reference to the constraints within.
pub fn constraints(&self) -> &Constraints {
&self.constraints
}
- /// Returns the output ordering of the properties.
- pub fn output_ordering(&self) -> Option<LexOrdering> {
- let constants = self.constants();
- let mut output_ordering = self.oeq_class().output_ordering().unwrap_or_default();
- // Prune out constant expressions
- output_ordering
- .retain(|sort_expr| !const_exprs_contains(constants, &sort_expr.expr));
- (!output_ordering.is_empty()).then_some(output_ordering)
+ /// Returns all the known constants expressions.
+ pub fn constants(&self) -> Vec<ConstExpr> {
+ self.eq_group
+ .iter()
+ .filter_map(|c| {
+ c.constant.as_ref().and_then(|across| {
+ c.canonical_expr()
+ .map(|expr| ConstExpr::new(Arc::clone(expr), across.clone()))
+ })
+ })
+ .collect()
}
- /// Returns the normalized version of the ordering equivalence class within.
- /// Normalization removes constants and duplicates as well as standardizing
- /// expressions according to the equivalence group within.
- pub fn normalized_oeq_class(&self) -> OrderingEquivalenceClass {
- OrderingEquivalenceClass::new(
- self.oeq_class
- .iter()
- .map(|ordering| self.normalize_sort_exprs(ordering))
- .collect(),
- )
+ /// Returns the output ordering of the properties.
+ pub fn output_ordering(&self) -> Option<LexOrdering> {
+ let concat = self.oeq_class.iter().flat_map(|o| o.iter().cloned());
+ self.normalize_sort_exprs(concat)
}
/// Extends this `EquivalenceProperties` with the `other` object.
- pub fn extend(mut self, other: Self) -> Self {
- self.eq_group.extend(other.eq_group);
- self.oeq_class.extend(other.oeq_class);
- self.with_constants(other.constants)
+ pub fn extend(mut self, other: Self) -> Result<Self> {
+ self.constraints.extend(other.constraints);
+ self.add_equivalence_group(other.eq_group)?;
+ self.add_orderings(other.oeq_class);
+ Ok(self)
}
/// Clears (empties) the ordering equivalence class within this object.
/// Call this method when existing orderings are invalidated.
pub fn clear_orderings(&mut self) {
self.oeq_class.clear();
+ self.oeq_cache.clear();
}
/// Removes constant expressions that may change across partitions.
- /// This method should be used when data from different partitions are merged.
+ /// This method should be used when merging data from different partitions.
pub fn clear_per_partition_constants(&mut self) {
- self.constants.retain(|item| {
- matches!(item.across_partitions(), AcrossPartitions::Uniform(_))
- })
- }
-
- /// Extends this `EquivalenceProperties` by adding the orderings inside the
- /// ordering equivalence class `other`.
- pub fn add_ordering_equivalence_class(&mut self, other: OrderingEquivalenceClass) {
- self.oeq_class.extend(other);
+ if self.eq_group.clear_per_partition_constants() {
+ // Renormalize orderings if the equivalence group changes:
+ let normal_orderings = self
+ .oeq_class
+ .iter()
+ .cloned()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache = OrderingEquivalenceCache::new(normal_orderings);
+ }
}
/// Adds new orderings into the existing ordering equivalence class.
- pub fn add_new_orderings(
+ pub fn add_orderings(
&mut self,
- orderings: impl IntoIterator<Item = LexOrdering>,
+ orderings: impl IntoIterator<Item = impl IntoIterator<Item = PhysicalSortExpr>>,
) {
- self.oeq_class.add_new_orderings(orderings);
+ let orderings: Vec<_> =
+ orderings.into_iter().filter_map(LexOrdering::new).collect();
+ let normal_orderings: Vec<_> = orderings
+ .iter()
+ .cloned()
+ .filter_map(|o| self.normalize_sort_exprs(o))
+ .collect();
+ if !normal_orderings.is_empty() {
+ self.oeq_class.extend(orderings);
+ // Normalize given orderings to update the cache:
+ self.oeq_cache.normal_cls.extend(normal_orderings);
+ // TODO: If no ordering is found to be redunant during extension, we
+ // can use a shortcut algorithm to update the leading map.
+ self.oeq_cache.update_map();
+ }
}
/// Adds a single ordering to the existing ordering equivalence class.
- pub fn add_new_ordering(&mut self, ordering: LexOrdering) {
- self.add_new_orderings([ordering]);
+ pub fn add_ordering(&mut self, ordering: impl IntoIterator<Item = PhysicalSortExpr>) {
+ self.add_orderings(std::iter::once(ordering));
}
/// Incorporates the given equivalence group to into the existing
/// equivalence group within.
- pub fn add_equivalence_group(&mut self, other_eq_group: EquivalenceGroup) {
- self.eq_group.extend(other_eq_group);
+ pub fn add_equivalence_group(
+ &mut self,
+ other_eq_group: EquivalenceGroup,
+ ) -> Result<()> {
+ if !other_eq_group.is_empty() {
+ self.eq_group.extend(other_eq_group);
+ // Renormalize orderings if the equivalence group changes:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls
+ .into_iter()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings based on the new equivalence classes:
+ let leading_exprs: Vec<_> =
+ self.oeq_cache.leading_map.keys().cloned().collect();
+ for expr in leading_exprs {
+ self.discover_new_orderings(expr)?;
+ }
+ }
+ Ok(())
+ }
+
+ /// Returns the ordering equivalence class within in normal form.
+ /// Normalization standardizes expressions according to the equivalence
+ /// group within, and removes constants/duplicates.
+ pub fn normalized_oeq_class(&self) -> OrderingEquivalenceClass {
+ self.oeq_class
+ .iter()
+ .cloned()
+ .filter_map(|ordering| self.normalize_sort_exprs(ordering))
+ .collect::<Vec<_>>()
+ .into()
}
/// Adds a new equality condition into the existing equivalence group.
/// If the given equality defines a new equivalence class, adds this new
/// equivalence class to the equivalence group.
pub fn add_equal_conditions(
&mut self,
- left: &Arc<dyn PhysicalExpr>,
- right: &Arc<dyn PhysicalExpr>,
+ left: Arc<dyn PhysicalExpr>,
+ right: Arc<dyn PhysicalExpr>,
) -> Result<()> {
- // Discover new constants in light of new the equality:
- if self.is_expr_constant(left) {
- // Left expression is constant, add right as constant
- if !const_exprs_contains(&self.constants, right) {
- let const_expr = ConstExpr::from(right)
- .with_across_partitions(self.get_expr_constant_value(left));
- self.constants.push(const_expr);
- }
- } else if self.is_expr_constant(right) {
- // Right expression is constant, add left as constant
- if !const_exprs_contains(&self.constants, left) {
- let const_expr = ConstExpr::from(left)
- .with_across_partitions(self.get_expr_constant_value(right));
- self.constants.push(const_expr);
- }
+ // Add equal expressions to the state:
+ if self.eq_group.add_equal_conditions(Arc::clone(&left), right) {
+ // Renormalize orderings if the equivalence group changes:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls
+ .into_iter()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings:
+ self.discover_new_orderings(left)?;
}
-
- // Add equal expressions to the state
- self.eq_group.add_equal_conditions(left, right);
-
- // Discover any new orderings
- self.discover_new_orderings(left)?;
Ok(())
}
/// Track/register physical expressions with constant values.
- #[deprecated(since = "43.0.0", note = "Use [`with_constants`] instead")]
- pub fn add_constants(self, constants: impl IntoIterator<Item = ConstExpr>) -> Self {
- self.with_constants(constants)
- }
-
- /// Remove the specified constant
- pub fn remove_constant(mut self, c: &ConstExpr) -> Self {
- self.constants.retain(|existing| existing != c);
- self
- }
-
- /// Track/register physical expressions with constant values.
- pub fn with_constants(
- mut self,
+ pub fn add_constants(
+ &mut self,
constants: impl IntoIterator<Item = ConstExpr>,
- ) -> Self {
- let normalized_constants = constants
- .into_iter()
- .filter_map(|c| {
- let across_partitions = c.across_partitions();
- let expr = c.owned_expr();
- let normalized_expr = self.eq_group.normalize_expr(expr);
-
- if const_exprs_contains(&self.constants, &normalized_expr) {
- return None;
- }
-
- let const_expr = ConstExpr::from(normalized_expr)
- .with_across_partitions(across_partitions);
-
- Some(const_expr)
+ ) -> Result<()> {
+ // Add the new constant to the equivalence group:
+ for constant in constants {
+ self.eq_group.add_constant(constant);
+ }
+ // Renormalize the orderings after adding new constants by removing
+ // the constants from existing orderings:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls.into_iter().map(|ordering| {
+ ordering.into_iter().filter(|sort_expr| {
+ self.eq_group.is_expr_constant(&sort_expr.expr).is_none()
})
- .collect::<Vec<_>>();
-
- // Add all new normalized constants
- self.constants.extend(normalized_constants);
-
- // Discover any new orderings based on the constants
- for ordering in self.normalized_oeq_class().iter() {
- if let Err(e) = self.discover_new_orderings(&ordering[0].expr) {
- log::debug!("error discovering new orderings: {e}");
- }
+ });
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings based on the constants:
+ let leading_exprs: Vec<_> = self.oeq_cache.leading_map.keys().cloned().collect();
+ for expr in leading_exprs {
+ self.discover_new_orderings(expr)?;
}
-
- self
+ Ok(())
}
- // Discover new valid orderings in light of a new equality.
- // Accepts a single argument (`expr`) which is used to determine
- // which orderings should be updated.
- // When constants or equivalence classes are changed, there may be new orderings
- // that can be discovered with the new equivalence properties.
- // For a discussion, see: https://github.com/apache/datafusion/issues/9812
- fn discover_new_orderings(&mut self, expr: &Arc<dyn PhysicalExpr>) -> Result<()> {
- let normalized_expr = self.eq_group().normalize_expr(Arc::clone(expr));
+ /// Discover new valid orderings in light of a new equality. Accepts a single
+ /// argument (`expr`) which is used to determine the orderings to update.
+ /// When constants or equivalence classes change, there may be new orderings
+ /// that can be discovered with the new equivalence properties.
+ /// For a discussion, see: <https://github.com/apache/datafusion/issues/9812>
+ fn discover_new_orderings(
+ &mut self,
+ normal_expr: Arc<dyn PhysicalExpr>,
+ ) -> Result<()> {
+ let Some(ordering_idxs) = self.oeq_cache.leading_map.get(&normal_expr) else {
+ return Ok(());
+ };
let eq_class = self
.eq_group
- .iter()
- .find_map(|class| {
- class
- .contains(&normalized_expr)
- .then(|| class.clone().into_vec())
- })
- .unwrap_or_else(|| vec![Arc::clone(&normalized_expr)]);
-
- let mut new_orderings: Vec<LexOrdering> = vec![];
- for ordering in self.normalized_oeq_class().iter() {
- if !ordering[0].expr.eq(&normalized_expr) {
- continue;
- }
+ .get_equivalence_class(&normal_expr)
+ .map_or_else(|| vec![normal_expr], |class| class.clone().into());
+ let mut new_orderings = vec![];
+ for idx in ordering_idxs {
+ let ordering = &self.oeq_cache.normal_cls[*idx];
let leading_ordering_options = ordering[0].options;
- for equivalent_expr in &eq_class {
+ 'exprs: for equivalent_expr in &eq_class {
let children = equivalent_expr.children();
if children.is_empty() {
continue;
}
-
- // Check if all children match the next expressions in the ordering
- let mut all_children_match = true;
+ // Check if all children match the next expressions in the ordering:
let mut child_properties = vec![];
-
- // Build properties for each child based on the next expressions
- for (i, child) in children.iter().enumerate() {
- if let Some(next) = ordering.get(i + 1) {
- if !child.as_ref().eq(next.expr.as_ref()) {
- all_children_match = false;
- break;
- }
- child_properties.push(ExprProperties {
- sort_properties: SortProperties::Ordered(next.options),
- range: Interval::make_unbounded(
- &child.data_type(&self.schema)?,
- )?,
- preserves_lex_ordering: true,
- });
- } else {
- all_children_match = false;
- break;
+ // Build properties for each child based on the next expression:
+ for (i, child) in children.into_iter().enumerate() {
+ let Some(next) = ordering.get(i + 1) else {
+ break 'exprs;
+ };
+ if !next.expr.eq(child) {
+ break 'exprs;
}
+ let data_type = child.data_type(&self.schema)?;
+ child_properties.push(ExprProperties {
+ sort_properties: SortProperties::Ordered(next.options),
+ range: Interval::make_unbounded(&data_type)?,
+ preserves_lex_ordering: true,
+ });
}
-
- if all_children_match {
- // Check if the expression is monotonic in all arguments
- if let Ok(expr_properties) =
- equivalent_expr.get_properties(&child_properties)
- {
- if expr_properties.preserves_lex_ordering
- && SortProperties::Ordered(leading_ordering_options)
- == expr_properties.sort_properties
- {
- // Assume existing ordering is [c ASC, a ASC, b ASC]
- // When equality c = f(a,b) is given, if we know that given ordering `[a ASC, b ASC]`,
- // ordering `[f(a,b) ASC]` is valid, then we can deduce that ordering `[a ASC, b ASC]` is also valid.
- // Hence, ordering `[a ASC, b ASC]` can be added to the state as a valid ordering.
- // (e.g. existing ordering where leading ordering is removed)
- new_orderings.push(LexOrdering::new(ordering[1..].to_vec()));
- break;
- }
- }
+ // Check if the expression is monotonic in all arguments:
+ let expr_properties =
+ equivalent_expr.get_properties(&child_properties)?;
+ if expr_properties.preserves_lex_ordering
+ && expr_properties.sort_properties
+ == SortProperties::Ordered(leading_ordering_options)
+ {
+ // Assume that `[c ASC, a ASC, b ASC]` is among existing
+ // orderings. If equality `c = f(a, b)` is given, ordering
+ // `[a ASC, b ASC]` implies the ordering `[c ASC]`. Thus,
+ // ordering `[a ASC, b ASC]` is also a valid ordering.
Review Comment:
I agree that the example only holds when `f` preserves the lexographic ordering of a,b. In fact that I think the check for `expr_properties.preserves_lex_ordering` is doing on line 463 above
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -579,302 +611,289 @@ impl EquivalenceProperties {
// From the analysis above, we know that `[a ASC]` is satisfied. Then,
// we add column `a` as constant to the algorithm state. This enables us
// to deduce that `(b + c) ASC` is satisfied, given `a` is constant.
- eq_properties = eq_properties.with_constants(std::iter::once(
- ConstExpr::from(Arc::clone(&normalized_req.expr)),
- ));
+ let const_expr = ConstExpr::from(element.expr);
+ eq_properties.add_constants(std::iter::once(const_expr))?;
}
+ Ok(true)
+ }
- // All requirements are satisfied.
- normalized_reqs.len()
+ /// Returns the number of consecutive sort expressions (starting from the
+ /// left) that are satisfied by the existing ordering.
+ fn common_sort_prefix_length(&self, normal_ordering: LexOrdering) -> Result<usize> {
+ let full_length = normal_ordering.len();
+ // Check whether the given ordering is satisfied by constraints:
+ if self.satisfied_by_constraints_ordering(&normal_ordering) {
+ // If constraints satisfy all sort expressions, return the full
+ // length:
+ return Ok(full_length);
+ }
+ let schema = self.schema();
+ let mut eq_properties = self.clone();
Review Comment:
I think you can avoid this clone by using `Cow` and only clone it if it needs to be updated
```diff
@@ -1157,7 +1158,8 @@ impl EquivalenceProperties {
&self,
exprs: &[Arc<dyn PhysicalExpr>],
) -> Result<(Vec<PhysicalSortExpr>, Vec<usize>)> {
- let mut eq_properties = self.clone();
+ // Avoid cloing `self` if not necessary:
+ let mut eq_properties = Cow::Borrowed(self);
let mut result = vec![];
// The algorithm is as follows:
// - Iterate over all the expressions and insert ordered expressions
@@ -1204,7 +1206,7 @@ impl EquivalenceProperties {
// an implementation strategy confined to this function.
for (PhysicalSortExpr { expr, .. }, idx) in &ordered_exprs {
let const_expr = ConstExpr::from(Arc::clone(expr));
- eq_properties.add_constants(std::iter::once(const_expr))?;
+ eq_properties.to_mut().add_constants(std::iter::once(const_expr))?;
search_indices.shift_remove(idx);
}
// Add new ordered section to the state.
```
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -579,302 +611,289 @@ impl EquivalenceProperties {
// From the analysis above, we know that `[a ASC]` is satisfied. Then,
// we add column `a` as constant to the algorithm state. This enables us
// to deduce that `(b + c) ASC` is satisfied, given `a` is constant.
- eq_properties = eq_properties.with_constants(std::iter::once(
- ConstExpr::from(Arc::clone(&normalized_req.expr)),
- ));
+ let const_expr = ConstExpr::from(element.expr);
+ eq_properties.add_constants(std::iter::once(const_expr))?;
}
+ Ok(true)
+ }
- // All requirements are satisfied.
- normalized_reqs.len()
+ /// Returns the number of consecutive sort expressions (starting from the
+ /// left) that are satisfied by the existing ordering.
+ fn common_sort_prefix_length(&self, normal_ordering: LexOrdering) -> Result<usize> {
+ let full_length = normal_ordering.len();
+ // Check whether the given ordering is satisfied by constraints:
+ if self.satisfied_by_constraints_ordering(&normal_ordering) {
+ // If constraints satisfy all sort expressions, return the full
+ // length:
+ return Ok(full_length);
+ }
+ let schema = self.schema();
+ let mut eq_properties = self.clone();
+ for (idx, element) in normal_ordering.into_iter().enumerate() {
+ // Check whether given ordering is satisfied:
+ let ExprProperties {
+ sort_properties, ..
+ } = eq_properties.get_expr_properties(Arc::clone(&element.expr));
+ let satisfy = match sort_properties {
+ SortProperties::Ordered(options) => options_compatible(
+ &options,
+ &element.options,
+ element.expr.nullable(schema).unwrap_or(true),
+ ),
+ // Singleton expressions satisfy any ordering.
+ SortProperties::Singleton => true,
+ SortProperties::Unordered => false,
+ };
+ if !satisfy {
+ // As soon as one sort expression is unsatisfied, return how
+ // many we've satisfied so far:
+ return Ok(idx);
+ }
+ // Treat satisfied keys as constants in subsequent iterations. We
+ // can do this because the "next" key only matters in a lexicographical
+ // ordering when the keys to its left have the same values.
+ //
+ // Note that these expressions are not properly "constants". This is just
+ // an implementation strategy confined to this function.
+ //
+ // For example, assume that the requirement is `[a ASC, (b + c) ASC]`,
+ // and existing equivalent orderings are `[a ASC, b ASC]` and `[c ASC]`.
+ // From the analysis above, we know that `[a ASC]` is satisfied. Then,
+ // we add column `a` as constant to the algorithm state. This enables us
+ // to deduce that `(b + c) ASC` is satisfied, given `a` is constant.
+ let const_expr = ConstExpr::from(element.expr);
+ eq_properties.add_constants(std::iter::once(const_expr))?
+ }
+ // All sort expressions are satisfied, return full length:
+ Ok(full_length)
}
- /// Determines the longest prefix of `reqs` that is satisfied by the existing ordering.
- /// Returns that prefix as a new `LexRequirement`, and a boolean indicating if all the requirements are satisfied.
+ /// Determines the longest normal prefix of `ordering` satisfied by the
+ /// existing ordering. Returns that prefix as a new `LexOrdering`, and a
+ /// boolean indicating whether all the sort expressions are satisfied.
pub fn extract_common_sort_prefix(
&self,
- reqs: &LexRequirement,
- ) -> (LexRequirement, bool) {
- // First, standardize the given requirement:
- let normalized_reqs = self.normalize_sort_requirements(reqs);
-
- let prefix_len = self.compute_common_sort_prefix_length(&normalized_reqs);
- (
- LexRequirement::new(normalized_reqs[..prefix_len].to_vec()),
- prefix_len == normalized_reqs.len(),
- )
- }
-
- /// Checks whether the given sort requirements are satisfied by any of the
- /// existing orderings.
- pub fn ordering_satisfy_requirement(&self, reqs: &LexRequirement) -> bool {
- self.extract_common_sort_prefix(reqs).1
+ ordering: LexOrdering,
+ ) -> Result<(Vec<PhysicalSortExpr>, bool)> {
+ // First, standardize the given ordering:
+ let Some(normal_ordering) = self.normalize_sort_exprs(ordering) else {
+ // If the ordering vanishes after normalization, it is satisfied:
+ return Ok((vec![], true));
+ };
+ let prefix_len = self.common_sort_prefix_length(normal_ordering.clone())?;
+ let flag = prefix_len == normal_ordering.len();
+ let mut sort_exprs: Vec<_> = normal_ordering.into();
+ if !flag {
+ sort_exprs.truncate(prefix_len);
+ }
+ Ok((sort_exprs, flag))
}
- /// Checks if the sort requirements are satisfied by any of the table constraints (primary key or unique).
- /// Returns true if any constraint fully satisfies the requirements.
- fn satisfied_by_constraints(
+ /// Checks if the sort expressions are satisfied by any of the table
+ /// constraints (primary key or unique). Returns true if any constraint
+ /// fully satisfies the expressions (i.e. constraint indices form a valid
+ /// prefix of an existing ordering that matches the expressions). For
+ /// unique constraints, also verifies nullable columns.
+ fn satisfied_by_constraints_ordering(
&self,
- normalized_reqs: &[PhysicalSortRequirement],
+ normal_exprs: &[PhysicalSortExpr],
) -> bool {
self.constraints.iter().any(|constraint| match constraint {
- Constraint::PrimaryKey(indices) | Constraint::Unique(indices) => self
- .satisfied_by_constraint(
- normalized_reqs,
- indices,
- matches!(constraint, Constraint::Unique(_)),
- ),
- })
- }
-
- /// Checks if sort requirements are satisfied by a constraint (primary key or unique).
- /// Returns true if the constraint indices form a valid prefix of an existing ordering
- /// that matches the requirements. For unique constraints, also verifies nullable columns.
- fn satisfied_by_constraint(
- &self,
- normalized_reqs: &[PhysicalSortRequirement],
- indices: &[usize],
- check_null: bool,
- ) -> bool {
- // Requirements must contain indices
- if indices.len() > normalized_reqs.len() {
- return false;
- }
-
- // Iterate over all orderings
- self.oeq_class.iter().any(|ordering| {
- if indices.len() > ordering.len() {
- return false;
- }
-
- // Build a map of column positions in the ordering
- let mut col_positions = HashMap::with_capacity(ordering.len());
- for (pos, req) in ordering.iter().enumerate() {
- if let Some(col) = req.expr.as_any().downcast_ref::<Column>() {
- col_positions.insert(
- col.index(),
- (pos, col.nullable(&self.schema).unwrap_or(true)),
- );
- }
- }
-
- // Check if all constraint indices appear in valid positions
- if !indices.iter().all(|&idx| {
- col_positions
- .get(&idx)
- .map(|&(pos, nullable)| {
- // For unique constraints, verify column is not nullable if it's first/last
- !check_null
- || (pos != 0 && pos != ordering.len() - 1)
- || !nullable
+ Constraint::PrimaryKey(indices) | Constraint::Unique(indices) => {
+ let check_null = matches!(constraint, Constraint::Unique(_));
+ let normalized_size = normal_exprs.len();
+ indices.len() <= normalized_size
+ && self.oeq_class.iter().any(|ordering| {
+ let length = ordering.len();
+ if indices.len() > length || normalized_size < length {
+ return false;
+ }
+ // Build a map of column positions in the ordering:
+ let mut col_positions = HashMap::with_capacity(length);
+ for (pos, req) in ordering.iter().enumerate() {
+ if let Some(col) = req.expr.as_any().downcast_ref::<Column>()
+ {
+ let nullable = col.nullable(&self.schema).unwrap_or(true);
+ col_positions.insert(col.index(), (pos, nullable));
+ }
+ }
+ // Check if all constraint indices appear in valid positions:
+ if !indices.iter().all(|idx| {
+ col_positions.get(idx).is_some_and(|&(pos, nullable)| {
+ // For unique constraints, verify column is not nullable if it's first/last:
+ !check_null
+ || !nullable
+ || (pos != 0 && pos != length - 1)
+ })
+ }) {
+ return false;
+ }
+ // Check if this ordering matches the prefix:
+ normal_exprs.iter().zip(ordering).all(|(given, existing)| {
+ existing.satisfy_expr(given, &self.schema)
+ })
})
- .unwrap_or(false)
- }) {
- return false;
}
-
- // Check if this ordering matches requirements prefix
- let ordering_len = ordering.len();
- normalized_reqs.len() >= ordering_len
- && normalized_reqs[..ordering_len].iter().zip(ordering).all(
- |(req, existing)| {
- req.expr.eq(&existing.expr)
- && req
- .options
- .is_none_or(|req_opts| req_opts == existing.options)
- },
- )
})
}
- /// Determines whether the ordering specified by the given sort requirement
- /// is satisfied based on the orderings within, equivalence classes, and
- /// constant expressions.
- ///
- /// # Parameters
- ///
- /// - `req`: A reference to a `PhysicalSortRequirement` for which the ordering
- /// satisfaction check will be done.
- ///
- /// # Returns
- ///
- /// Returns `true` if the specified ordering is satisfied, `false` otherwise.
- fn ordering_satisfy_single(&self, req: &PhysicalSortRequirement) -> bool {
- let ExprProperties {
- sort_properties, ..
- } = self.get_expr_properties(Arc::clone(&req.expr));
- match sort_properties {
- SortProperties::Ordered(options) => {
- let sort_expr = PhysicalSortExpr {
- expr: Arc::clone(&req.expr),
- options,
- };
- sort_expr.satisfy(req, self.schema())
+ /// Checks if the sort requirements are satisfied by any of the table
+ /// constraints (primary key or unique). Returns true if any constraint
+ /// fully satisfies the requirements (i.e. constraint indices form a valid
+ /// prefix of an existing ordering that matches the requirements). For
+ /// unique constraints, also verifies nullable columns.
+ fn satisfied_by_constraints(&self, normal_reqs: &[PhysicalSortRequirement]) -> bool {
+ self.constraints.iter().any(|constraint| match constraint {
+ Constraint::PrimaryKey(indices) | Constraint::Unique(indices) => {
+ let check_null = matches!(constraint, Constraint::Unique(_));
+ let normalized_size = normal_reqs.len();
+ indices.len() <= normalized_size
+ && self.oeq_class.iter().any(|ordering| {
+ let length = ordering.len();
+ if indices.len() > length || normalized_size < length {
+ return false;
+ }
+ // Build a map of column positions in the ordering:
+ let mut col_positions = HashMap::with_capacity(length);
+ for (pos, req) in ordering.iter().enumerate() {
+ if let Some(col) = req.expr.as_any().downcast_ref::<Column>()
+ {
+ let nullable = col.nullable(&self.schema).unwrap_or(true);
+ col_positions.insert(col.index(), (pos, nullable));
+ }
+ }
+ // Check if all constraint indices appear in valid positions:
+ if !indices.iter().all(|idx| {
+ col_positions.get(idx).is_some_and(|&(pos, nullable)| {
+ // For unique constraints, verify column is not nullable if it's first/last:
+ !check_null
+ || !nullable
+ || (pos != 0 && pos != length - 1)
+ })
+ }) {
+ return false;
+ }
+ // Check if this ordering matches the prefix:
+ normal_reqs.iter().zip(ordering).all(|(given, existing)| {
+ existing.satisfy(given, &self.schema)
+ })
+ })
}
- // Singleton expressions satisfies any ordering.
- SortProperties::Singleton => true,
- SortProperties::Unordered => false,
- }
+ })
}
/// Checks whether the `given` sort requirements are equal or more specific
/// than the `reference` sort requirements.
pub fn requirements_compatible(
&self,
- given: &LexRequirement,
- reference: &LexRequirement,
+ given: LexRequirement,
+ reference: LexRequirement,
) -> bool {
- let normalized_given = self.normalize_sort_requirements(given);
- let normalized_reference = self.normalize_sort_requirements(reference);
-
- (normalized_reference.len() <= normalized_given.len())
- && normalized_reference
+ let Some(normal_given) = self.normalize_sort_requirements(given) else {
+ return true;
+ };
+ let Some(normal_reference) = self.normalize_sort_requirements(reference) else {
+ return true;
+ };
+
+ (normal_reference.len() <= normal_given.len())
+ && normal_reference
.into_iter()
- .zip(normalized_given)
+ .zip(normal_given)
.all(|(reference, given)| given.compatible(&reference))
}
- /// Returns the finer ordering among the orderings `lhs` and `rhs`, breaking
- /// any ties by choosing `lhs`.
- ///
- /// The finer ordering is the ordering that satisfies both of the orderings.
- /// If the orderings are incomparable, returns `None`.
- ///
- /// For example, the finer ordering among `[a ASC]` and `[a ASC, b ASC]` is
- /// the latter.
- pub fn get_finer_ordering(
- &self,
- lhs: &LexOrdering,
- rhs: &LexOrdering,
- ) -> Option<LexOrdering> {
- // Convert the given sort expressions to sort requirements:
- let lhs = LexRequirement::from(lhs.clone());
- let rhs = LexRequirement::from(rhs.clone());
- let finer = self.get_finer_requirement(&lhs, &rhs);
- // Convert the chosen sort requirements back to sort expressions:
- finer.map(LexOrdering::from)
- }
-
- /// Returns the finer ordering among the requirements `lhs` and `rhs`,
- /// breaking any ties by choosing `lhs`.
+ /// Modify existing orderings by substituting sort expressions with appropriate
+ /// targets from the projection mapping. We substitute a sort expression when
+ /// its physical expression has a one-to-one functional relationship with a
+ /// target expression in the mapping.
///
- /// The finer requirements are the ones that satisfy both of the given
- /// requirements. If the requirements are incomparable, returns `None`.
+ /// After substitution, we may generate more than one `LexOrdering` for each
+ /// existing equivalent ordering. For example, `[a ASC, b ASC]` will turn
+ /// into `[CAST(a) ASC, b ASC]` and `[a ASC, b ASC]` when applying projection
+ /// expressions `a, b, CAST(a)`.
///
- /// For example, the finer requirements among `[a ASC]` and `[a ASC, b ASC]`
- /// is the latter.
- pub fn get_finer_requirement(
- &self,
- req1: &LexRequirement,
- req2: &LexRequirement,
- ) -> Option<LexRequirement> {
- let mut lhs = self.normalize_sort_requirements(req1);
- let mut rhs = self.normalize_sort_requirements(req2);
- lhs.inner
- .iter_mut()
- .zip(rhs.inner.iter_mut())
- .all(|(lhs, rhs)| {
- lhs.expr.eq(&rhs.expr)
- && match (lhs.options, rhs.options) {
- (Some(lhs_opt), Some(rhs_opt)) => lhs_opt == rhs_opt,
- (Some(options), None) => {
- rhs.options = Some(options);
- true
- }
- (None, Some(options)) => {
- lhs.options = Some(options);
- true
- }
- (None, None) => true,
- }
- })
- .then_some(if lhs.len() >= rhs.len() { lhs } else { rhs })
- }
-
- /// we substitute the ordering according to input expression type, this is a simplified version
- /// In this case, we just substitute when the expression satisfy the following condition:
- /// I. just have one column and is a CAST expression
- /// TODO: Add one-to-ones analysis for monotonic ScalarFunctions.
- /// TODO: we could precompute all the scenario that is computable, for example: atan(x + 1000) should also be substituted if
- /// x is DESC or ASC
- /// After substitution, we may generate more than 1 `LexOrdering`. As an example,
- /// `[a ASC, b ASC]` will turn into `[a ASC, b ASC], [CAST(a) ASC, b ASC]` when projection expressions `a, b, CAST(a)` is applied.
- pub fn substitute_ordering_component(
- &self,
+ /// TODO: Handle all scenarios that allow substitution; e.g. when `x` is
Review Comment:
I recommend we file a ticket to track this work and add a link to the ticket in the comments
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2931862394
🤖 `./gh_compare_branch_bench.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch_bench.sh) Running
Linux aal-dev 6.11.0-1013-gcp #13~24.04.1-Ubuntu SMP Wed Apr 2 16:34:16 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing required-input-ordering (a75ae4464888d3317b5869eb7ae3aa8a5b150294) to 900279c5d189a62a5585210c64ab6868af4640e5 [diff](https://github.com/apache/datafusion/compare/900279c5d189a62a5585210c64ab6868af4640e5..a75ae4464888d3317b5869eb7ae3aa8a5b150294)
BENCH_NAME=sql_planner
BENCH_COMMAND=cargo bench --bench sql_planner
BENCH_FILTER=
BENCH_BRANCH_NAME=required-input-ordering
Results will be posted here when complete
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126760915
##########
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:
Agreed, I will add such clarifications.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2127435736
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -579,302 +611,289 @@ impl EquivalenceProperties {
// From the analysis above, we know that `[a ASC]` is satisfied. Then,
// we add column `a` as constant to the algorithm state. This enables us
// to deduce that `(b + c) ASC` is satisfied, given `a` is constant.
- eq_properties = eq_properties.with_constants(std::iter::once(
- ConstExpr::from(Arc::clone(&normalized_req.expr)),
- ));
+ let const_expr = ConstExpr::from(element.expr);
+ eq_properties.add_constants(std::iter::once(const_expr))?;
}
+ Ok(true)
+ }
- // All requirements are satisfied.
- normalized_reqs.len()
+ /// Returns the number of consecutive sort expressions (starting from the
+ /// left) that are satisfied by the existing ordering.
+ fn common_sort_prefix_length(&self, normal_ordering: LexOrdering) -> Result<usize> {
+ let full_length = normal_ordering.len();
+ // Check whether the given ordering is satisfied by constraints:
+ if self.satisfied_by_constraints_ordering(&normal_ordering) {
+ // If constraints satisfy all sort expressions, return the full
+ // length:
+ return Ok(full_length);
+ }
+ let schema = self.schema();
+ let mut eq_properties = self.clone();
Review Comment:
I didn't really understand how this would help (probably because I'm tired after a long day :)), but we can make a tiny follow-on to do this if you think it will improve performance.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2930483562
I plan to begin a careful review of this PR today
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126733166
##########
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:
Agreed, enriching comments accordingly.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "Omega359 (via GitHub)" <gi...@apache.org>.
Omega359 commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2951240245
I'll be able to do my tests Sat afternoon or Sun morning. I'm good with doing that against either main or the branch.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2951959753
Thanks @Omega359. Given that we didn't hear any concerns yesterday, I will go ahead and merge this.
Thanks everyone for all the reviews!
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2922389594
@wiedld @xudong963 and @suremarc perhaps you are also interested in this PR given your past interest in the equivalence code. Also cc @chenkovsky
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2124375915
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -190,382 +241,363 @@ impl EquivalenceProperties {
&self.oeq_class
}
- /// Return the inner OrderingEquivalenceClass, consuming self
- pub fn into_oeq_class(self) -> OrderingEquivalenceClass {
- self.oeq_class
- }
-
/// Returns a reference to the equivalence group within.
pub fn eq_group(&self) -> &EquivalenceGroup {
&self.eq_group
}
- /// Returns a reference to the constant expressions
- pub fn constants(&self) -> &[ConstExpr] {
- &self.constants
- }
-
+ /// Returns a reference to the constraints within.
pub fn constraints(&self) -> &Constraints {
&self.constraints
}
- /// Returns the output ordering of the properties.
- pub fn output_ordering(&self) -> Option<LexOrdering> {
- let constants = self.constants();
- let mut output_ordering = self.oeq_class().output_ordering().unwrap_or_default();
- // Prune out constant expressions
- output_ordering
- .retain(|sort_expr| !const_exprs_contains(constants, &sort_expr.expr));
- (!output_ordering.is_empty()).then_some(output_ordering)
+ /// Returns all the known constants expressions.
+ pub fn constants(&self) -> Vec<ConstExpr> {
+ self.eq_group
+ .iter()
+ .filter_map(|c| {
+ c.constant.as_ref().and_then(|across| {
+ c.canonical_expr()
+ .map(|expr| ConstExpr::new(Arc::clone(expr), across.clone()))
+ })
+ })
+ .collect()
}
- /// Returns the normalized version of the ordering equivalence class within.
- /// Normalization removes constants and duplicates as well as standardizing
- /// expressions according to the equivalence group within.
- pub fn normalized_oeq_class(&self) -> OrderingEquivalenceClass {
- OrderingEquivalenceClass::new(
- self.oeq_class
- .iter()
- .map(|ordering| self.normalize_sort_exprs(ordering))
- .collect(),
- )
+ /// Returns the output ordering of the properties.
+ pub fn output_ordering(&self) -> Option<LexOrdering> {
+ let concat = self.oeq_class.iter().flat_map(|o| o.iter().cloned());
+ self.normalize_sort_exprs(concat)
}
/// Extends this `EquivalenceProperties` with the `other` object.
- pub fn extend(mut self, other: Self) -> Self {
- self.eq_group.extend(other.eq_group);
- self.oeq_class.extend(other.oeq_class);
- self.with_constants(other.constants)
+ pub fn extend(mut self, other: Self) -> Result<Self> {
+ self.constraints.extend(other.constraints);
+ self.add_equivalence_group(other.eq_group)?;
+ self.add_orderings(other.oeq_class);
+ Ok(self)
}
/// Clears (empties) the ordering equivalence class within this object.
/// Call this method when existing orderings are invalidated.
pub fn clear_orderings(&mut self) {
self.oeq_class.clear();
+ self.oeq_cache.clear();
}
/// Removes constant expressions that may change across partitions.
- /// This method should be used when data from different partitions are merged.
+ /// This method should be used when merging data from different partitions.
pub fn clear_per_partition_constants(&mut self) {
- self.constants.retain(|item| {
- matches!(item.across_partitions(), AcrossPartitions::Uniform(_))
- })
- }
-
- /// Extends this `EquivalenceProperties` by adding the orderings inside the
- /// ordering equivalence class `other`.
- pub fn add_ordering_equivalence_class(&mut self, other: OrderingEquivalenceClass) {
- self.oeq_class.extend(other);
+ if self.eq_group.clear_per_partition_constants() {
+ // Renormalize orderings if the equivalence group changes:
+ let normal_orderings = self
+ .oeq_class
+ .iter()
+ .cloned()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache = OrderingEquivalenceCache::new(normal_orderings);
+ }
}
/// Adds new orderings into the existing ordering equivalence class.
- pub fn add_new_orderings(
+ pub fn add_orderings(
&mut self,
- orderings: impl IntoIterator<Item = LexOrdering>,
+ orderings: impl IntoIterator<Item = impl IntoIterator<Item = PhysicalSortExpr>>,
) {
- self.oeq_class.add_new_orderings(orderings);
+ let orderings: Vec<_> =
+ orderings.into_iter().filter_map(LexOrdering::new).collect();
+ let normal_orderings: Vec<_> = orderings
+ .iter()
+ .cloned()
+ .filter_map(|o| self.normalize_sort_exprs(o))
+ .collect();
+ if !normal_orderings.is_empty() {
+ self.oeq_class.extend(orderings);
+ // Normalize given orderings to update the cache:
+ self.oeq_cache.normal_cls.extend(normal_orderings);
+ // TODO: If no ordering is found to be redunant during extension, we
+ // can use a shortcut algorithm to update the leading map.
+ self.oeq_cache.update_map();
+ }
}
/// Adds a single ordering to the existing ordering equivalence class.
- pub fn add_new_ordering(&mut self, ordering: LexOrdering) {
- self.add_new_orderings([ordering]);
+ pub fn add_ordering(&mut self, ordering: impl IntoIterator<Item = PhysicalSortExpr>) {
+ self.add_orderings(std::iter::once(ordering));
}
/// Incorporates the given equivalence group to into the existing
/// equivalence group within.
- pub fn add_equivalence_group(&mut self, other_eq_group: EquivalenceGroup) {
- self.eq_group.extend(other_eq_group);
+ pub fn add_equivalence_group(
+ &mut self,
+ other_eq_group: EquivalenceGroup,
+ ) -> Result<()> {
+ if !other_eq_group.is_empty() {
+ self.eq_group.extend(other_eq_group);
+ // Renormalize orderings if the equivalence group changes:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls
+ .into_iter()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings based on the new equivalence classes:
+ let leading_exprs: Vec<_> =
+ self.oeq_cache.leading_map.keys().cloned().collect();
+ for expr in leading_exprs {
+ self.discover_new_orderings(expr)?;
+ }
+ }
+ Ok(())
+ }
+
+ /// Returns the ordering equivalence class within in normal form.
+ /// Normalization standardizes expressions according to the equivalence
+ /// group within, and removes constants/duplicates.
+ pub fn normalized_oeq_class(&self) -> OrderingEquivalenceClass {
+ self.oeq_class
+ .iter()
+ .cloned()
+ .filter_map(|ordering| self.normalize_sort_exprs(ordering))
+ .collect::<Vec<_>>()
+ .into()
}
/// Adds a new equality condition into the existing equivalence group.
/// If the given equality defines a new equivalence class, adds this new
/// equivalence class to the equivalence group.
pub fn add_equal_conditions(
&mut self,
- left: &Arc<dyn PhysicalExpr>,
- right: &Arc<dyn PhysicalExpr>,
+ left: Arc<dyn PhysicalExpr>,
+ right: Arc<dyn PhysicalExpr>,
) -> Result<()> {
- // Discover new constants in light of new the equality:
- if self.is_expr_constant(left) {
- // Left expression is constant, add right as constant
- if !const_exprs_contains(&self.constants, right) {
- let const_expr = ConstExpr::from(right)
- .with_across_partitions(self.get_expr_constant_value(left));
- self.constants.push(const_expr);
- }
- } else if self.is_expr_constant(right) {
- // Right expression is constant, add left as constant
- if !const_exprs_contains(&self.constants, left) {
- let const_expr = ConstExpr::from(left)
- .with_across_partitions(self.get_expr_constant_value(right));
- self.constants.push(const_expr);
- }
+ // Add equal expressions to the state:
+ if self.eq_group.add_equal_conditions(Arc::clone(&left), right) {
+ // Renormalize orderings if the equivalence group changes:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls
+ .into_iter()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings:
+ self.discover_new_orderings(left)?;
}
-
- // Add equal expressions to the state
- self.eq_group.add_equal_conditions(left, right);
-
- // Discover any new orderings
- self.discover_new_orderings(left)?;
Ok(())
}
/// Track/register physical expressions with constant values.
- #[deprecated(since = "43.0.0", note = "Use [`with_constants`] instead")]
- pub fn add_constants(self, constants: impl IntoIterator<Item = ConstExpr>) -> Self {
- self.with_constants(constants)
- }
-
- /// Remove the specified constant
- pub fn remove_constant(mut self, c: &ConstExpr) -> Self {
- self.constants.retain(|existing| existing != c);
- self
- }
-
- /// Track/register physical expressions with constant values.
- pub fn with_constants(
- mut self,
+ pub fn add_constants(
+ &mut self,
constants: impl IntoIterator<Item = ConstExpr>,
- ) -> Self {
- let normalized_constants = constants
- .into_iter()
- .filter_map(|c| {
- let across_partitions = c.across_partitions();
- let expr = c.owned_expr();
- let normalized_expr = self.eq_group.normalize_expr(expr);
-
- if const_exprs_contains(&self.constants, &normalized_expr) {
- return None;
- }
-
- let const_expr = ConstExpr::from(normalized_expr)
- .with_across_partitions(across_partitions);
-
- Some(const_expr)
+ ) -> Result<()> {
+ // Add the new constant to the equivalence group:
+ for constant in constants {
+ self.eq_group.add_constant(constant);
+ }
+ // Renormalize the orderings after adding new constants by removing
+ // the constants from existing orderings:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls.into_iter().map(|ordering| {
+ ordering.into_iter().filter(|sort_expr| {
+ self.eq_group.is_expr_constant(&sort_expr.expr).is_none()
})
- .collect::<Vec<_>>();
-
- // Add all new normalized constants
- self.constants.extend(normalized_constants);
-
- // Discover any new orderings based on the constants
- for ordering in self.normalized_oeq_class().iter() {
- if let Err(e) = self.discover_new_orderings(&ordering[0].expr) {
- log::debug!("error discovering new orderings: {e}");
- }
+ });
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings based on the constants:
+ let leading_exprs: Vec<_> = self.oeq_cache.leading_map.keys().cloned().collect();
+ for expr in leading_exprs {
+ self.discover_new_orderings(expr)?;
}
-
- self
+ Ok(())
}
- // Discover new valid orderings in light of a new equality.
- // Accepts a single argument (`expr`) which is used to determine
- // which orderings should be updated.
- // When constants or equivalence classes are changed, there may be new orderings
- // that can be discovered with the new equivalence properties.
- // For a discussion, see: https://github.com/apache/datafusion/issues/9812
- fn discover_new_orderings(&mut self, expr: &Arc<dyn PhysicalExpr>) -> Result<()> {
- let normalized_expr = self.eq_group().normalize_expr(Arc::clone(expr));
+ /// Discover new valid orderings in light of a new equality. Accepts a single
+ /// argument (`expr`) which is used to determine the orderings to update.
+ /// When constants or equivalence classes change, there may be new orderings
+ /// that can be discovered with the new equivalence properties.
+ /// For a discussion, see: <https://github.com/apache/datafusion/issues/9812>
+ fn discover_new_orderings(
+ &mut self,
+ normal_expr: Arc<dyn PhysicalExpr>,
+ ) -> Result<()> {
+ let Some(ordering_idxs) = self.oeq_cache.leading_map.get(&normal_expr) else {
+ return Ok(());
+ };
let eq_class = self
.eq_group
- .iter()
- .find_map(|class| {
- class
- .contains(&normalized_expr)
- .then(|| class.clone().into_vec())
- })
- .unwrap_or_else(|| vec![Arc::clone(&normalized_expr)]);
-
- let mut new_orderings: Vec<LexOrdering> = vec![];
- for ordering in self.normalized_oeq_class().iter() {
- if !ordering[0].expr.eq(&normalized_expr) {
- continue;
- }
+ .get_equivalence_class(&normal_expr)
+ .map_or_else(|| vec![normal_expr], |class| class.clone().into());
+ let mut new_orderings = vec![];
+ for idx in ordering_idxs {
+ let ordering = &self.oeq_cache.normal_cls[*idx];
let leading_ordering_options = ordering[0].options;
- for equivalent_expr in &eq_class {
+ 'exprs: for equivalent_expr in &eq_class {
let children = equivalent_expr.children();
if children.is_empty() {
continue;
}
-
- // Check if all children match the next expressions in the ordering
- let mut all_children_match = true;
+ // Check if all children match the next expressions in the ordering:
let mut child_properties = vec![];
-
- // Build properties for each child based on the next expressions
- for (i, child) in children.iter().enumerate() {
- if let Some(next) = ordering.get(i + 1) {
- if !child.as_ref().eq(next.expr.as_ref()) {
- all_children_match = false;
- break;
- }
- child_properties.push(ExprProperties {
- sort_properties: SortProperties::Ordered(next.options),
- range: Interval::make_unbounded(
- &child.data_type(&self.schema)?,
- )?,
- preserves_lex_ordering: true,
- });
- } else {
- all_children_match = false;
- break;
+ // Build properties for each child based on the next expression:
+ for (i, child) in children.into_iter().enumerate() {
+ let Some(next) = ordering.get(i + 1) else {
+ break 'exprs;
+ };
+ if !next.expr.eq(child) {
+ break 'exprs;
}
+ let data_type = child.data_type(&self.schema)?;
+ child_properties.push(ExprProperties {
+ sort_properties: SortProperties::Ordered(next.options),
+ range: Interval::make_unbounded(&data_type)?,
+ preserves_lex_ordering: true,
+ });
}
-
- if all_children_match {
- // Check if the expression is monotonic in all arguments
- if let Ok(expr_properties) =
- equivalent_expr.get_properties(&child_properties)
- {
- if expr_properties.preserves_lex_ordering
- && SortProperties::Ordered(leading_ordering_options)
- == expr_properties.sort_properties
- {
- // Assume existing ordering is [c ASC, a ASC, b ASC]
- // When equality c = f(a,b) is given, if we know that given ordering `[a ASC, b ASC]`,
- // ordering `[f(a,b) ASC]` is valid, then we can deduce that ordering `[a ASC, b ASC]` is also valid.
- // Hence, ordering `[a ASC, b ASC]` can be added to the state as a valid ordering.
- // (e.g. existing ordering where leading ordering is removed)
- new_orderings.push(LexOrdering::new(ordering[1..].to_vec()));
- break;
- }
- }
+ // Check if the expression is monotonic in all arguments:
+ let expr_properties =
+ equivalent_expr.get_properties(&child_properties)?;
+ if expr_properties.preserves_lex_ordering
+ && expr_properties.sort_properties
+ == SortProperties::Ordered(leading_ordering_options)
+ {
+ // Assume that `[c ASC, a ASC, b ASC]` is among existing
+ // orderings. If equality `c = f(a, b)` is given, ordering
+ // `[a ASC, b ASC]` implies the ordering `[c ASC]`. Thus,
+ // ordering `[a ASC, b ASC]` is also a valid ordering.
Review Comment:
Got it. I saw the `preserves_lex_ordering` check.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2940673323
Thanks for all the reviews. I went through each and sent a commit that addresses all the suggestions, I also tried to answer any questions I saw.
I will later resolve the conflicts and get this ready to go. I see that we are close to getting ready for the 48 release too, so it seems like we are making a good effort to align them in the way we want.
Thanks all 🚀
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126896630
##########
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:
Yep, let's make a push for those after we merge this PR
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak merged PR #16217:
URL: https://github.com/apache/datafusion/pull/16217
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2952309229
🚀 woohoo!
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2922481568
Let's definitely time the merge and DF48 release in a cooperative way. We have two competing realities at play: (1) The PR attracts a lot of conflicts and it is not easy to resolve them without accidents, (2) We don't want to dump a large API change on to DF48 without giving prep time to people.
Checking out #15771, it seems we can either:
1. Move fast with the release while this gets reviews, which has the advantage of making DF48 a "checkpoint" before this PR. The downside is that we will need to do extensive testing *both* for the release and then one for the PR.
2. Move slower with the release (maybe release it a week later than usual) and merge this relatively quickly. The advantage of this choice would be performing extensive testing only once.
I am fine with both approaches as long as we arrive at a merge point quickly, before or after the release.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2932142471
🤖 `./gh_compare_branch_bench.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch_bench.sh) Running
Linux aal-dev 6.11.0-1013-gcp #13~24.04.1-Ubuntu SMP Wed Apr 2 16:34:16 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing required-input-ordering (a75ae4464888d3317b5869eb7ae3aa8a5b150294) to 900279c5d189a62a5585210c64ab6868af4640e5 [diff](https://github.com/apache/datafusion/compare/900279c5d189a62a5585210c64ab6868af4640e5..a75ae4464888d3317b5869eb7ae3aa8a5b150294)
BENCH_NAME=sql_planner
BENCH_COMMAND=cargo bench --bench sql_planner
BENCH_FILTER=
BENCH_BRANCH_NAME=required-input-ordering
Results will be posted here when complete
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2951003937
OK, I resolved the conflicts and all seems OK. Should we go ahead after CI passes?
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126812465
##########
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:
Right, I will update the documentation to reflect this.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126889241
##########
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:
The old equivalence system was not performing normalizations in certain cases. Now, as we always store orderings in normal form (in addition to their original form), we don't miss these.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126883202
##########
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:
Actually not -- this query works now. I looked at what is going on and it seems like the requirements are not really conflicting, seems like the old equivalence system wasn't able to handle it.
It'd be great if you can double check as well, so we don't accidentally start accepting queries we can't currently execute correctly :)
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2932136822
There are some pretty nice looking results in the benchmark results:
```
group main required-input-ordering
----- ---- -----------------------
...
physical_select_aggregates_from_200 1.19 26.3±0.10ms ? ?/sec 1.00 22.1±0.15ms ? ?/sec
...
physical_sorted_union_orderby 1.36 63.9±0.39ms ? ?/sec 1.00 47.0±0.31ms ? ?/sec
```
There are a few that appear to get slightly slower -- I will rerun the benchmarks to see if it reproduces and run a profile locally
```
physical_plan_clickbench_q43 1.00 3.1±0.04ms ? ?/sec 1.06 3.3±0.03ms ? ?/sec
physical_plan_tpch_q10 1.00 3.3±0.02ms ? ?/sec 1.06 3.5±0.04ms ? ?/sec
physical_plan_tpch_q7 1.00 4.4±0.02ms ? ?/sec 1.06 4.6±0.02ms ? ?/sec
```
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2922392615
Thanks for the quick response @alamb -- I added some reviewing tips and tricks to the PR body to help you guys out.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2124440817
##########
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 {
Review Comment:
I need to trace the change of `required_input_ordering` in other files here as it looks confusing at the first look between `LexRequirement` and `OrderingRequirements`. Maybe `InputOrderingRequirements` or `InputOrderingReq`?
Another question is, this looks like can be a separate change in API.
Is this also a required part of so called equivalence overhaul?
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2934983867
Thanks for all the reviews. I will address them today 🚀
With the computational complexity issue removed, I agree that we can progressively reduce the "constant factor" in running time over a few future PRs. To give reviewers some context: Complexity issue is solved by computing normalizations only once (when equivalences are modified) and handling constants on an equal footing with other equivalences. This makes ordering checks fast, but some parts of the current codebase probably make modifications eagerly, which is the likely source of the small slowdown @alamb is seeing. I expect refactoring those parts of the code a little bit will mitigate this.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2122154995
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -125,40 +120,85 @@ use itertools::Itertools;
/// # let col_c = col("c", &schema).unwrap();
/// // This object represents data that is sorted by a ASC, c DESC
/// // with a single constant value of b
-/// let mut eq_properties = EquivalenceProperties::new(schema)
-/// .with_constants(vec![ConstExpr::from(col_b)]);
-/// eq_properties.add_new_ordering(LexOrdering::new(vec![
+/// let mut eq_properties = EquivalenceProperties::new(schema);
+/// eq_properties.add_constants(vec![ConstExpr::from(col_b)]);
+/// eq_properties.add_ordering([
/// PhysicalSortExpr::new_default(col_a).asc(),
/// PhysicalSortExpr::new_default(col_c).desc(),
-/// ]));
+/// ]);
///
-/// assert_eq!(eq_properties.to_string(), "order: [[a@0 ASC, c@2 DESC]], const: [b@1(heterogeneous)]")
+/// assert_eq!(eq_properties.to_string(), "order: [[a@0 ASC, c@2 DESC]], eq: [{members: [b@1], constant: (heterogeneous)}]");
/// ```
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug)]
pub struct EquivalenceProperties {
- /// Distinct equivalence classes (exprs known to have the same expressions)
+ /// Distinct equivalence classes (i.e. expressions with the same value).
eq_group: EquivalenceGroup,
- /// Equivalent sort expressions
+ /// Equivalent sort expressions (i.e. those define the same ordering).
oeq_class: OrderingEquivalenceClass,
- /// Expressions whose values are constant
- ///
- /// TODO: We do not need to track constants separately, they can be tracked
- /// inside `eq_group` as `Literal` expressions.
- constants: Vec<ConstExpr>,
- /// Table constraints
+ /// Cache storing equivalent sort expressions in normal form (i.e. without
+ /// constants/duplicates and in standard form) and a map associating leading
+ /// terms with full sort expressions.
+ oeq_cache: OrderingEquivalenceCache,
+ /// Table constraints that factor in equivalence calculations.
constraints: Constraints,
/// Schema associated with this object.
schema: SchemaRef,
}
+/// This object serves as a cache for storing equivalent sort expressions
+/// in normal form, and a map associating leading sort expressions with
+/// full lexicographical orderings. With this information, DataFusion can
+/// efficiently determine whether a given ordering is satisfied by the
+/// existing orderings, and discover new orderings based on the existing
+/// equivalence properties.
+#[derive(Clone, Debug, Default)]
+struct OrderingEquivalenceCache {
+ /// Equivalent sort expressions in normal form.
+ normal_cls: OrderingEquivalenceClass,
Review Comment:
If this is an efficient way to leverage the info provided by `OrderingEquivalenceClass`, I wonder why you don't just incorporate cache into `OrderingEquivalenceClass` so we don't need to expose an additional cache?
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -190,382 +241,363 @@ impl EquivalenceProperties {
&self.oeq_class
}
- /// Return the inner OrderingEquivalenceClass, consuming self
- pub fn into_oeq_class(self) -> OrderingEquivalenceClass {
- self.oeq_class
- }
-
/// Returns a reference to the equivalence group within.
pub fn eq_group(&self) -> &EquivalenceGroup {
&self.eq_group
}
- /// Returns a reference to the constant expressions
- pub fn constants(&self) -> &[ConstExpr] {
- &self.constants
- }
-
+ /// Returns a reference to the constraints within.
pub fn constraints(&self) -> &Constraints {
&self.constraints
}
- /// Returns the output ordering of the properties.
- pub fn output_ordering(&self) -> Option<LexOrdering> {
- let constants = self.constants();
- let mut output_ordering = self.oeq_class().output_ordering().unwrap_or_default();
- // Prune out constant expressions
- output_ordering
- .retain(|sort_expr| !const_exprs_contains(constants, &sort_expr.expr));
- (!output_ordering.is_empty()).then_some(output_ordering)
+ /// Returns all the known constants expressions.
+ pub fn constants(&self) -> Vec<ConstExpr> {
+ self.eq_group
+ .iter()
+ .filter_map(|c| {
+ c.constant.as_ref().and_then(|across| {
+ c.canonical_expr()
+ .map(|expr| ConstExpr::new(Arc::clone(expr), across.clone()))
+ })
+ })
+ .collect()
}
- /// Returns the normalized version of the ordering equivalence class within.
- /// Normalization removes constants and duplicates as well as standardizing
- /// expressions according to the equivalence group within.
- pub fn normalized_oeq_class(&self) -> OrderingEquivalenceClass {
- OrderingEquivalenceClass::new(
- self.oeq_class
- .iter()
- .map(|ordering| self.normalize_sort_exprs(ordering))
- .collect(),
- )
+ /// Returns the output ordering of the properties.
+ pub fn output_ordering(&self) -> Option<LexOrdering> {
+ let concat = self.oeq_class.iter().flat_map(|o| o.iter().cloned());
+ self.normalize_sort_exprs(concat)
}
/// Extends this `EquivalenceProperties` with the `other` object.
- pub fn extend(mut self, other: Self) -> Self {
- self.eq_group.extend(other.eq_group);
- self.oeq_class.extend(other.oeq_class);
- self.with_constants(other.constants)
+ pub fn extend(mut self, other: Self) -> Result<Self> {
+ self.constraints.extend(other.constraints);
+ self.add_equivalence_group(other.eq_group)?;
+ self.add_orderings(other.oeq_class);
+ Ok(self)
}
/// Clears (empties) the ordering equivalence class within this object.
/// Call this method when existing orderings are invalidated.
pub fn clear_orderings(&mut self) {
self.oeq_class.clear();
+ self.oeq_cache.clear();
}
/// Removes constant expressions that may change across partitions.
- /// This method should be used when data from different partitions are merged.
+ /// This method should be used when merging data from different partitions.
pub fn clear_per_partition_constants(&mut self) {
- self.constants.retain(|item| {
- matches!(item.across_partitions(), AcrossPartitions::Uniform(_))
- })
- }
-
- /// Extends this `EquivalenceProperties` by adding the orderings inside the
- /// ordering equivalence class `other`.
- pub fn add_ordering_equivalence_class(&mut self, other: OrderingEquivalenceClass) {
- self.oeq_class.extend(other);
+ if self.eq_group.clear_per_partition_constants() {
+ // Renormalize orderings if the equivalence group changes:
+ let normal_orderings = self
+ .oeq_class
+ .iter()
+ .cloned()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache = OrderingEquivalenceCache::new(normal_orderings);
+ }
}
/// Adds new orderings into the existing ordering equivalence class.
- pub fn add_new_orderings(
+ pub fn add_orderings(
&mut self,
- orderings: impl IntoIterator<Item = LexOrdering>,
+ orderings: impl IntoIterator<Item = impl IntoIterator<Item = PhysicalSortExpr>>,
) {
- self.oeq_class.add_new_orderings(orderings);
+ let orderings: Vec<_> =
+ orderings.into_iter().filter_map(LexOrdering::new).collect();
+ let normal_orderings: Vec<_> = orderings
+ .iter()
+ .cloned()
+ .filter_map(|o| self.normalize_sort_exprs(o))
+ .collect();
+ if !normal_orderings.is_empty() {
+ self.oeq_class.extend(orderings);
+ // Normalize given orderings to update the cache:
+ self.oeq_cache.normal_cls.extend(normal_orderings);
+ // TODO: If no ordering is found to be redunant during extension, we
+ // can use a shortcut algorithm to update the leading map.
+ self.oeq_cache.update_map();
+ }
}
/// Adds a single ordering to the existing ordering equivalence class.
- pub fn add_new_ordering(&mut self, ordering: LexOrdering) {
- self.add_new_orderings([ordering]);
+ pub fn add_ordering(&mut self, ordering: impl IntoIterator<Item = PhysicalSortExpr>) {
+ self.add_orderings(std::iter::once(ordering));
}
/// Incorporates the given equivalence group to into the existing
/// equivalence group within.
- pub fn add_equivalence_group(&mut self, other_eq_group: EquivalenceGroup) {
- self.eq_group.extend(other_eq_group);
+ pub fn add_equivalence_group(
+ &mut self,
+ other_eq_group: EquivalenceGroup,
+ ) -> Result<()> {
+ if !other_eq_group.is_empty() {
+ self.eq_group.extend(other_eq_group);
+ // Renormalize orderings if the equivalence group changes:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls
+ .into_iter()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings based on the new equivalence classes:
+ let leading_exprs: Vec<_> =
+ self.oeq_cache.leading_map.keys().cloned().collect();
+ for expr in leading_exprs {
+ self.discover_new_orderings(expr)?;
+ }
+ }
+ Ok(())
+ }
+
+ /// Returns the ordering equivalence class within in normal form.
+ /// Normalization standardizes expressions according to the equivalence
+ /// group within, and removes constants/duplicates.
+ pub fn normalized_oeq_class(&self) -> OrderingEquivalenceClass {
+ self.oeq_class
+ .iter()
+ .cloned()
+ .filter_map(|ordering| self.normalize_sort_exprs(ordering))
+ .collect::<Vec<_>>()
+ .into()
}
/// Adds a new equality condition into the existing equivalence group.
/// If the given equality defines a new equivalence class, adds this new
/// equivalence class to the equivalence group.
pub fn add_equal_conditions(
&mut self,
- left: &Arc<dyn PhysicalExpr>,
- right: &Arc<dyn PhysicalExpr>,
+ left: Arc<dyn PhysicalExpr>,
+ right: Arc<dyn PhysicalExpr>,
) -> Result<()> {
- // Discover new constants in light of new the equality:
- if self.is_expr_constant(left) {
- // Left expression is constant, add right as constant
- if !const_exprs_contains(&self.constants, right) {
- let const_expr = ConstExpr::from(right)
- .with_across_partitions(self.get_expr_constant_value(left));
- self.constants.push(const_expr);
- }
- } else if self.is_expr_constant(right) {
- // Right expression is constant, add left as constant
- if !const_exprs_contains(&self.constants, left) {
- let const_expr = ConstExpr::from(left)
- .with_across_partitions(self.get_expr_constant_value(right));
- self.constants.push(const_expr);
- }
+ // Add equal expressions to the state:
+ if self.eq_group.add_equal_conditions(Arc::clone(&left), right) {
+ // Renormalize orderings if the equivalence group changes:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls
+ .into_iter()
+ .map(|o| self.eq_group.normalize_sort_exprs(o));
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings:
+ self.discover_new_orderings(left)?;
}
-
- // Add equal expressions to the state
- self.eq_group.add_equal_conditions(left, right);
-
- // Discover any new orderings
- self.discover_new_orderings(left)?;
Ok(())
}
/// Track/register physical expressions with constant values.
- #[deprecated(since = "43.0.0", note = "Use [`with_constants`] instead")]
- pub fn add_constants(self, constants: impl IntoIterator<Item = ConstExpr>) -> Self {
- self.with_constants(constants)
- }
-
- /// Remove the specified constant
- pub fn remove_constant(mut self, c: &ConstExpr) -> Self {
- self.constants.retain(|existing| existing != c);
- self
- }
-
- /// Track/register physical expressions with constant values.
- pub fn with_constants(
- mut self,
+ pub fn add_constants(
+ &mut self,
constants: impl IntoIterator<Item = ConstExpr>,
- ) -> Self {
- let normalized_constants = constants
- .into_iter()
- .filter_map(|c| {
- let across_partitions = c.across_partitions();
- let expr = c.owned_expr();
- let normalized_expr = self.eq_group.normalize_expr(expr);
-
- if const_exprs_contains(&self.constants, &normalized_expr) {
- return None;
- }
-
- let const_expr = ConstExpr::from(normalized_expr)
- .with_across_partitions(across_partitions);
-
- Some(const_expr)
+ ) -> Result<()> {
+ // Add the new constant to the equivalence group:
+ for constant in constants {
+ self.eq_group.add_constant(constant);
+ }
+ // Renormalize the orderings after adding new constants by removing
+ // the constants from existing orderings:
+ let normal_cls = mem::take(&mut self.oeq_cache.normal_cls);
+ let normal_orderings = normal_cls.into_iter().map(|ordering| {
+ ordering.into_iter().filter(|sort_expr| {
+ self.eq_group.is_expr_constant(&sort_expr.expr).is_none()
})
- .collect::<Vec<_>>();
-
- // Add all new normalized constants
- self.constants.extend(normalized_constants);
-
- // Discover any new orderings based on the constants
- for ordering in self.normalized_oeq_class().iter() {
- if let Err(e) = self.discover_new_orderings(&ordering[0].expr) {
- log::debug!("error discovering new orderings: {e}");
- }
+ });
+ self.oeq_cache.normal_cls = OrderingEquivalenceClass::new(normal_orderings);
+ self.oeq_cache.update_map();
+ // Discover any new orderings based on the constants:
+ let leading_exprs: Vec<_> = self.oeq_cache.leading_map.keys().cloned().collect();
+ for expr in leading_exprs {
+ self.discover_new_orderings(expr)?;
}
-
- self
+ Ok(())
}
- // Discover new valid orderings in light of a new equality.
- // Accepts a single argument (`expr`) which is used to determine
- // which orderings should be updated.
- // When constants or equivalence classes are changed, there may be new orderings
- // that can be discovered with the new equivalence properties.
- // For a discussion, see: https://github.com/apache/datafusion/issues/9812
- fn discover_new_orderings(&mut self, expr: &Arc<dyn PhysicalExpr>) -> Result<()> {
- let normalized_expr = self.eq_group().normalize_expr(Arc::clone(expr));
+ /// Discover new valid orderings in light of a new equality. Accepts a single
+ /// argument (`expr`) which is used to determine the orderings to update.
+ /// When constants or equivalence classes change, there may be new orderings
+ /// that can be discovered with the new equivalence properties.
+ /// For a discussion, see: <https://github.com/apache/datafusion/issues/9812>
+ fn discover_new_orderings(
+ &mut self,
+ normal_expr: Arc<dyn PhysicalExpr>,
+ ) -> Result<()> {
+ let Some(ordering_idxs) = self.oeq_cache.leading_map.get(&normal_expr) else {
+ return Ok(());
+ };
let eq_class = self
.eq_group
- .iter()
- .find_map(|class| {
- class
- .contains(&normalized_expr)
- .then(|| class.clone().into_vec())
- })
- .unwrap_or_else(|| vec![Arc::clone(&normalized_expr)]);
-
- let mut new_orderings: Vec<LexOrdering> = vec![];
- for ordering in self.normalized_oeq_class().iter() {
- if !ordering[0].expr.eq(&normalized_expr) {
- continue;
- }
+ .get_equivalence_class(&normal_expr)
+ .map_or_else(|| vec![normal_expr], |class| class.clone().into());
+ let mut new_orderings = vec![];
+ for idx in ordering_idxs {
+ let ordering = &self.oeq_cache.normal_cls[*idx];
let leading_ordering_options = ordering[0].options;
- for equivalent_expr in &eq_class {
+ 'exprs: for equivalent_expr in &eq_class {
let children = equivalent_expr.children();
if children.is_empty() {
continue;
}
-
- // Check if all children match the next expressions in the ordering
- let mut all_children_match = true;
+ // Check if all children match the next expressions in the ordering:
let mut child_properties = vec![];
-
- // Build properties for each child based on the next expressions
- for (i, child) in children.iter().enumerate() {
- if let Some(next) = ordering.get(i + 1) {
- if !child.as_ref().eq(next.expr.as_ref()) {
- all_children_match = false;
- break;
- }
- child_properties.push(ExprProperties {
- sort_properties: SortProperties::Ordered(next.options),
- range: Interval::make_unbounded(
- &child.data_type(&self.schema)?,
- )?,
- preserves_lex_ordering: true,
- });
- } else {
- all_children_match = false;
- break;
+ // Build properties for each child based on the next expression:
+ for (i, child) in children.into_iter().enumerate() {
+ let Some(next) = ordering.get(i + 1) else {
+ break 'exprs;
+ };
+ if !next.expr.eq(child) {
+ break 'exprs;
}
+ let data_type = child.data_type(&self.schema)?;
+ child_properties.push(ExprProperties {
+ sort_properties: SortProperties::Ordered(next.options),
+ range: Interval::make_unbounded(&data_type)?,
+ preserves_lex_ordering: true,
+ });
}
-
- if all_children_match {
- // Check if the expression is monotonic in all arguments
- if let Ok(expr_properties) =
- equivalent_expr.get_properties(&child_properties)
- {
- if expr_properties.preserves_lex_ordering
- && SortProperties::Ordered(leading_ordering_options)
- == expr_properties.sort_properties
- {
- // Assume existing ordering is [c ASC, a ASC, b ASC]
- // When equality c = f(a,b) is given, if we know that given ordering `[a ASC, b ASC]`,
- // ordering `[f(a,b) ASC]` is valid, then we can deduce that ordering `[a ASC, b ASC]` is also valid.
- // Hence, ordering `[a ASC, b ASC]` can be added to the state as a valid ordering.
- // (e.g. existing ordering where leading ordering is removed)
- new_orderings.push(LexOrdering::new(ordering[1..].to_vec()));
- break;
- }
- }
+ // Check if the expression is monotonic in all arguments:
+ let expr_properties =
+ equivalent_expr.get_properties(&child_properties)?;
+ if expr_properties.preserves_lex_ordering
+ && expr_properties.sort_properties
+ == SortProperties::Ordered(leading_ordering_options)
+ {
+ // Assume that `[c ASC, a ASC, b ASC]` is among existing
+ // orderings. If equality `c = f(a, b)` is given, ordering
+ // `[a ASC, b ASC]` implies the ordering `[c ASC]`. Thus,
+ // ordering `[a ASC, b ASC]` is also a valid ordering.
Review Comment:
I know you just rephrase these comments so the meaning is same. But I wonder if it is correct? Shouldn't it depend on what `f` is? If `f` is not monotonic increasing, how does `[a ASC, b ASC]` imply the ordering `[c ASC]`.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2932076551
🤖: Benchmark completed
<details><summary>Details</summary>
<p>
```
group main required-input-ordering
----- ---- -----------------------
logical_aggregate_with_join 1.00 731.5±5.31µs ? ?/sec 1.00 730.4±2.85µs ? ?/sec
logical_select_all_from_1000 1.01 125.1±0.22ms ? ?/sec 1.00 124.3±0.50ms ? ?/sec
logical_select_one_from_700 1.00 411.1±1.44µs ? ?/sec 1.00 411.1±1.53µs ? ?/sec
logical_trivial_join_high_numbered_columns 1.00 375.1±1.38µs ? ?/sec 1.00 375.0±6.11µs ? ?/sec
logical_trivial_join_low_numbered_columns 1.01 361.3±1.64µs ? ?/sec 1.00 359.1±2.25µs ? ?/sec
physical_intersection 1.00 840.3±3.07µs ? ?/sec 1.02 854.0±4.38µs ? ?/sec
physical_join_consider_sort 1.00 1358.7±7.27µs ? ?/sec 1.03 1400.3±5.71µs ? ?/sec
physical_join_distinct 1.00 350.4±1.86µs ? ?/sec 1.00 351.6±2.31µs ? ?/sec
physical_many_self_joins 1.00 10.0±0.03ms ? ?/sec 1.00 10.1±0.02ms ? ?/sec
physical_plan_clickbench_all 1.01 148.9±1.99ms ? ?/sec 1.00 147.9±1.17ms ? ?/sec
physical_plan_clickbench_q1 1.00 1726.0±17.47µs ? ?/sec 1.01 1735.0±22.65µs ? ?/sec
physical_plan_clickbench_q10 1.00 2.5±0.02ms ? ?/sec 1.03 2.6±0.06ms ? ?/sec
physical_plan_clickbench_q11 1.00 2.6±0.03ms ? ?/sec 1.02 2.6±0.03ms ? ?/sec
physical_plan_clickbench_q12 1.00 2.7±0.03ms ? ?/sec 1.01 2.8±0.02ms ? ?/sec
physical_plan_clickbench_q13 1.00 2.4±0.03ms ? ?/sec 1.01 2.4±0.03ms ? ?/sec
physical_plan_clickbench_q14 1.00 2.6±0.03ms ? ?/sec 1.01 2.7±0.02ms ? ?/sec
physical_plan_clickbench_q15 1.00 2.5±0.03ms ? ?/sec 1.02 2.5±0.03ms ? ?/sec
physical_plan_clickbench_q16 1.00 2.3±0.04ms ? ?/sec 1.01 2.4±0.02ms ? ?/sec
physical_plan_clickbench_q17 1.00 2.5±0.03ms ? ?/sec 1.04 2.5±0.02ms ? ?/sec
physical_plan_clickbench_q18 1.02 2.0±0.03ms ? ?/sec 1.00 1998.0±27.62µs ? ?/sec
physical_plan_clickbench_q19 1.00 2.9±0.02ms ? ?/sec 1.03 3.0±0.03ms ? ?/sec
physical_plan_clickbench_q2 1.00 1958.0±29.88µs ? ?/sec 1.01 1968.1±27.08µs ? ?/sec
physical_plan_clickbench_q20 1.02 1709.3±22.03µs ? ?/sec 1.00 1669.7±22.20µs ? ?/sec
physical_plan_clickbench_q21 1.00 1984.7±20.59µs ? ?/sec 1.03 2.0±0.03ms ? ?/sec
physical_plan_clickbench_q22 1.00 2.7±0.04ms ? ?/sec 1.01 2.7±0.02ms ? ?/sec
physical_plan_clickbench_q23 1.00 3.0±0.03ms ? ?/sec 1.03 3.0±0.06ms ? ?/sec
physical_plan_clickbench_q24 1.00 4.7±0.03ms ? ?/sec 1.00 4.7±0.02ms ? ?/sec
physical_plan_clickbench_q25 1.00 2.0±0.02ms ? ?/sec 1.02 2.1±0.02ms ? ?/sec
physical_plan_clickbench_q26 1.00 1857.1±19.45µs ? ?/sec 1.01 1871.1±18.16µs ? ?/sec
physical_plan_clickbench_q27 1.00 2.1±0.02ms ? ?/sec 1.01 2.1±0.03ms ? ?/sec
physical_plan_clickbench_q28 1.00 2.9±0.03ms ? ?/sec 1.01 3.0±0.05ms ? ?/sec
physical_plan_clickbench_q29 1.00 3.7±0.03ms ? ?/sec 1.02 3.7±0.05ms ? ?/sec
physical_plan_clickbench_q3 1.00 1936.2±20.30µs ? ?/sec 1.01 1953.3±21.10µs ? ?/sec
physical_plan_clickbench_q30 1.12 15.4±0.22ms ? ?/sec 1.00 13.8±0.14ms ? ?/sec
physical_plan_clickbench_q31 1.00 3.0±0.03ms ? ?/sec 1.00 2.9±0.03ms ? ?/sec
physical_plan_clickbench_q32 1.00 2.9±0.03ms ? ?/sec 1.02 3.0±0.05ms ? ?/sec
physical_plan_clickbench_q33 1.00 2.5±0.02ms ? ?/sec 1.01 2.5±0.02ms ? ?/sec
physical_plan_clickbench_q34 1.00 2.1±0.02ms ? ?/sec 1.02 2.2±0.02ms ? ?/sec
physical_plan_clickbench_q35 1.00 2.2±0.02ms ? ?/sec 1.02 2.3±0.03ms ? ?/sec
physical_plan_clickbench_q36 1.00 3.0±0.03ms ? ?/sec 1.02 3.1±0.03ms ? ?/sec
physical_plan_clickbench_q37 1.00 3.1±0.03ms ? ?/sec 1.01 3.1±0.02ms ? ?/sec
physical_plan_clickbench_q38 1.00 3.1±0.03ms ? ?/sec 1.00 3.1±0.03ms ? ?/sec
physical_plan_clickbench_q39 1.00 2.8±0.03ms ? ?/sec 1.00 2.8±0.02ms ? ?/sec
physical_plan_clickbench_q4 1.00 1676.1±20.68µs ? ?/sec 1.01 1685.4±20.14µs ? ?/sec
physical_plan_clickbench_q40 1.00 3.5±0.03ms ? ?/sec 1.01 3.5±0.01ms ? ?/sec
physical_plan_clickbench_q41 1.00 3.0±0.02ms ? ?/sec 1.01 3.1±0.03ms ? ?/sec
physical_plan_clickbench_q42 1.01 3.0±0.03ms ? ?/sec 1.00 3.0±0.02ms ? ?/sec
physical_plan_clickbench_q43 1.00 3.1±0.04ms ? ?/sec 1.06 3.3±0.03ms ? ?/sec
physical_plan_clickbench_q44 1.01 1833.2±23.49µs ? ?/sec 1.00 1820.9±17.43µs ? ?/sec
physical_plan_clickbench_q45 1.00 1835.8±24.12µs ? ?/sec 1.00 1843.0±44.55µs ? ?/sec
physical_plan_clickbench_q46 1.00 2.3±0.02ms ? ?/sec 1.01 2.3±0.02ms ? ?/sec
physical_plan_clickbench_q47 1.00 2.9±0.02ms ? ?/sec 1.03 2.9±0.05ms ? ?/sec
physical_plan_clickbench_q48 1.00 3.5±0.03ms ? ?/sec 1.02 3.5±0.06ms ? ?/sec
physical_plan_clickbench_q49 1.03 4.0±0.03ms ? ?/sec 1.00 3.9±0.07ms ? ?/sec
physical_plan_clickbench_q5 1.00 1867.4±19.68µs ? ?/sec 1.01 1886.0±23.34µs ? ?/sec
physical_plan_clickbench_q50 1.00 3.5±0.03ms ? ?/sec 1.01 3.6±0.03ms ? ?/sec
physical_plan_clickbench_q51 1.00 2.4±0.02ms ? ?/sec 1.01 2.4±0.02ms ? ?/sec
physical_plan_clickbench_q52 1.00 3.2±0.04ms ? ?/sec 1.02 3.3±0.03ms ? ?/sec
physical_plan_clickbench_q6 1.00 1871.9±20.29µs ? ?/sec 1.01 1889.0±21.57µs ? ?/sec
physical_plan_clickbench_q7 1.00 1732.6±17.57µs ? ?/sec 1.00 1728.1±30.43µs ? ?/sec
physical_plan_clickbench_q8 1.00 2.4±0.02ms ? ?/sec 1.03 2.5±0.03ms ? ?/sec
physical_plan_clickbench_q9 1.00 2.3±0.03ms ? ?/sec 1.00 2.3±0.02ms ? ?/sec
physical_plan_tpcds_all 1.00 1098.4±6.12ms ? ?/sec 1.03 1134.3±6.62ms ? ?/sec
physical_plan_tpch_all 1.00 65.3±0.16ms ? ?/sec 1.04 67.9±0.24ms ? ?/sec
physical_plan_tpch_q1 1.00 2.2±0.03ms ? ?/sec 1.03 2.3±0.02ms ? ?/sec
physical_plan_tpch_q10 1.00 3.3±0.02ms ? ?/sec 1.06 3.5±0.04ms ? ?/sec
physical_plan_tpch_q11 1.00 3.5±0.01ms ? ?/sec 1.05 3.7±0.07ms ? ?/sec
physical_plan_tpch_q12 1.00 1971.3±9.14µs ? ?/sec 1.04 2.0±0.01ms ? ?/sec
physical_plan_tpch_q13 1.00 1497.8±13.37µs ? ?/sec 1.06 1592.7±6.37µs ? ?/sec
physical_plan_tpch_q14 1.00 2.0±0.01ms ? ?/sec 1.05 2.1±0.02ms ? ?/sec
physical_plan_tpch_q16 1.00 2.6±0.03ms ? ?/sec 1.06 2.8±0.05ms ? ?/sec
physical_plan_tpch_q17 1.00 2.6±0.08ms ? ?/sec 1.02 2.7±0.01ms ? ?/sec
physical_plan_tpch_q18 1.00 2.8±0.02ms ? ?/sec 1.05 2.9±0.01ms ? ?/sec
physical_plan_tpch_q19 1.00 4.0±0.02ms ? ?/sec 1.01 4.0±0.01ms ? ?/sec
physical_plan_tpch_q2 1.00 5.7±0.02ms ? ?/sec 1.04 5.9±0.02ms ? ?/sec
physical_plan_tpch_q20 1.00 3.3±0.04ms ? ?/sec 1.04 3.5±0.04ms ? ?/sec
physical_plan_tpch_q21 1.00 4.4±0.03ms ? ?/sec 1.03 4.6±0.02ms ? ?/sec
physical_plan_tpch_q22 1.00 3.0±0.03ms ? ?/sec 1.04 3.1±0.02ms ? ?/sec
physical_plan_tpch_q3 1.00 2.3±0.01ms ? ?/sec 1.05 2.4±0.01ms ? ?/sec
physical_plan_tpch_q4 1.00 1668.8±15.02µs ? ?/sec 1.04 1734.0±14.60µs ? ?/sec
physical_plan_tpch_q5 1.00 3.3±0.02ms ? ?/sec 1.04 3.4±0.01ms ? ?/sec
physical_plan_tpch_q6 1.00 943.1±7.94µs ? ?/sec 1.02 963.1±6.34µs ? ?/sec
physical_plan_tpch_q7 1.00 4.4±0.02ms ? ?/sec 1.06 4.6±0.02ms ? ?/sec
physical_plan_tpch_q8 1.00 5.3±0.03ms ? ?/sec 1.05 5.6±0.03ms ? ?/sec
physical_plan_tpch_q9 1.00 4.2±0.02ms ? ?/sec 1.06 4.4±0.02ms ? ?/sec
physical_select_aggregates_from_200 1.19 26.3±0.10ms ? ?/sec 1.00 22.1±0.15ms ? ?/sec
physical_select_all_from_1000 1.01 139.5±0.33ms ? ?/sec 1.00 138.6±0.27ms ? ?/sec
physical_select_one_from_700 1.00 1041.5±5.37µs ? ?/sec 1.00 1036.6±4.65µs ? ?/sec
physical_sorted_union_orderby 1.36 63.9±0.39ms ? ?/sec 1.00 47.0±0.31ms ? ?/sec
physical_theta_join_consider_sort 1.00 1735.5±6.65µs ? ?/sec 1.01 1760.8±8.88µs ? ?/sec
physical_unnest_to_join 1.00 1329.7±8.03µs ? ?/sec 1.00 1336.1±8.43µs ? ?/sec
with_param_values_many_columns 1.00 165.3±0.78µs ? ?/sec 1.00 164.6±1.97µs ? ?/sec
```
</p>
</details>
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2122097178
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -175,307 +135,398 @@ impl ConstExpr {
}
}
+impl PartialEq for ConstExpr {
+ fn eq(&self, other: &Self) -> bool {
+ self.across_partitions == other.across_partitions && self.expr.eq(&other.expr)
+ }
+}
+
impl Display for ConstExpr {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.expr)?;
- match &self.across_partitions {
- AcrossPartitions::Heterogeneous => {
- write!(f, "(heterogeneous)")?;
- }
- AcrossPartitions::Uniform(value) => {
- if let Some(val) = value {
- write!(f, "(uniform: {val})")?;
- } else {
- write!(f, "(uniform: unknown)")?;
- }
- }
- }
- Ok(())
+ write!(f, "{}", self.across_partitions)
}
}
impl From<Arc<dyn PhysicalExpr>> for ConstExpr {
fn from(expr: Arc<dyn PhysicalExpr>) -> Self {
- Self::new(expr)
- }
-}
-
-impl From<&Arc<dyn PhysicalExpr>> for ConstExpr {
- fn from(expr: &Arc<dyn PhysicalExpr>) -> Self {
- Self::new(Arc::clone(expr))
+ // By default, assume constant expressions are not same across partitions.
+ // However, if we have a literal, it will have a single value that is the
+ // same across all partitions.
+ let across = if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+ AcrossPartitions::Uniform(Some(lit.value().clone()))
+ } else {
+ AcrossPartitions::Heterogeneous
+ };
+ Self {
+ expr,
+ across_partitions: across,
+ }
}
}
-/// Checks whether `expr` is among in the `const_exprs`.
-pub fn const_exprs_contains(
- const_exprs: &[ConstExpr],
- expr: &Arc<dyn PhysicalExpr>,
-) -> bool {
- const_exprs
- .iter()
- .any(|const_expr| const_expr.expr.eq(expr))
-}
-
/// An `EquivalenceClass` is a set of [`Arc<dyn PhysicalExpr>`]s that are known
/// to have the same value for all tuples in a relation. These are generated by
/// equality predicates (e.g. `a = b`), typically equi-join conditions and
/// equality conditions in filters.
///
/// Two `EquivalenceClass`es are equal if they contains the same expressions in
/// without any ordering.
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct EquivalenceClass {
- /// The expressions in this equivalence class. The order doesn't
- /// matter for equivalence purposes
- ///
- exprs: IndexSet<Arc<dyn PhysicalExpr>>,
-}
-
-impl PartialEq for EquivalenceClass {
- /// Returns true if other is equal in the sense
- /// of bags (multi-sets), disregarding their orderings.
- fn eq(&self, other: &Self) -> bool {
- self.exprs.eq(&other.exprs)
- }
+ /// The expressions in this equivalence class. The order doesn't matter for
+ /// equivalence purposes.
+ pub(crate) exprs: IndexSet<Arc<dyn PhysicalExpr>>,
+ /// Indicates whether the expressions in this equivalence class have a
+ /// constant value. A `Some` value indicates constant-ness.
+ pub(crate) constant: Option<AcrossPartitions>,
}
impl EquivalenceClass {
- /// Create a new empty equivalence class
- pub fn new_empty() -> Self {
- Self {
- exprs: IndexSet::new(),
- }
- }
-
- // Create a new equivalence class from a pre-existing `Vec`
- pub fn new(exprs: Vec<Arc<dyn PhysicalExpr>>) -> Self {
- Self {
- exprs: exprs.into_iter().collect(),
+ // Create a new equivalence class from a pre-existing collection.
+ pub fn new(exprs: impl IntoIterator<Item = Arc<dyn PhysicalExpr>>) -> Self {
+ let mut class = Self::default();
+ for expr in exprs {
+ class.push(expr);
}
- }
-
- /// Return the inner vector of expressions
- pub fn into_vec(self) -> Vec<Arc<dyn PhysicalExpr>> {
- self.exprs.into_iter().collect()
+ class
}
/// Return the "canonical" expression for this class (the first element)
- /// if any
- fn canonical_expr(&self) -> Option<Arc<dyn PhysicalExpr>> {
- self.exprs.iter().next().cloned()
+ /// if non-empty.
+ pub fn canonical_expr(&self) -> Option<&Arc<dyn PhysicalExpr>> {
+ self.exprs.iter().next()
}
/// Insert the expression into this class, meaning it is known to be equal to
- /// all other expressions in this class
+ /// all other expressions in this class.
pub fn push(&mut self, expr: Arc<dyn PhysicalExpr>) {
+ if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+ let expr_across = AcrossPartitions::Uniform(Some(lit.value().clone()));
+ if let Some(across) = self.constant.as_mut() {
+ // TODO: Return an error if constant values do not agree.
+ if *across == AcrossPartitions::Heterogeneous {
+ *across = expr_across;
+ }
Review Comment:
It is safer by checking the constant literal values are actually the same if the existing `constant` is `Uniform` and its value is known.
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -175,307 +135,398 @@ impl ConstExpr {
}
}
+impl PartialEq for ConstExpr {
+ fn eq(&self, other: &Self) -> bool {
+ self.across_partitions == other.across_partitions && self.expr.eq(&other.expr)
+ }
+}
+
impl Display for ConstExpr {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.expr)?;
- match &self.across_partitions {
- AcrossPartitions::Heterogeneous => {
- write!(f, "(heterogeneous)")?;
- }
- AcrossPartitions::Uniform(value) => {
- if let Some(val) = value {
- write!(f, "(uniform: {val})")?;
- } else {
- write!(f, "(uniform: unknown)")?;
- }
- }
- }
- Ok(())
+ write!(f, "{}", self.across_partitions)
}
}
impl From<Arc<dyn PhysicalExpr>> for ConstExpr {
fn from(expr: Arc<dyn PhysicalExpr>) -> Self {
- Self::new(expr)
- }
-}
-
-impl From<&Arc<dyn PhysicalExpr>> for ConstExpr {
- fn from(expr: &Arc<dyn PhysicalExpr>) -> Self {
- Self::new(Arc::clone(expr))
+ // By default, assume constant expressions are not same across partitions.
+ // However, if we have a literal, it will have a single value that is the
+ // same across all partitions.
+ let across = if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+ AcrossPartitions::Uniform(Some(lit.value().clone()))
+ } else {
+ AcrossPartitions::Heterogeneous
+ };
+ Self {
+ expr,
+ across_partitions: across,
+ }
}
}
-/// Checks whether `expr` is among in the `const_exprs`.
-pub fn const_exprs_contains(
- const_exprs: &[ConstExpr],
- expr: &Arc<dyn PhysicalExpr>,
-) -> bool {
- const_exprs
- .iter()
- .any(|const_expr| const_expr.expr.eq(expr))
-}
-
/// An `EquivalenceClass` is a set of [`Arc<dyn PhysicalExpr>`]s that are known
/// to have the same value for all tuples in a relation. These are generated by
/// equality predicates (e.g. `a = b`), typically equi-join conditions and
/// equality conditions in filters.
///
/// Two `EquivalenceClass`es are equal if they contains the same expressions in
/// without any ordering.
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct EquivalenceClass {
- /// The expressions in this equivalence class. The order doesn't
- /// matter for equivalence purposes
- ///
- exprs: IndexSet<Arc<dyn PhysicalExpr>>,
-}
-
-impl PartialEq for EquivalenceClass {
- /// Returns true if other is equal in the sense
- /// of bags (multi-sets), disregarding their orderings.
- fn eq(&self, other: &Self) -> bool {
- self.exprs.eq(&other.exprs)
- }
+ /// The expressions in this equivalence class. The order doesn't matter for
+ /// equivalence purposes.
+ pub(crate) exprs: IndexSet<Arc<dyn PhysicalExpr>>,
+ /// Indicates whether the expressions in this equivalence class have a
+ /// constant value. A `Some` value indicates constant-ness.
+ pub(crate) constant: Option<AcrossPartitions>,
}
impl EquivalenceClass {
- /// Create a new empty equivalence class
- pub fn new_empty() -> Self {
- Self {
- exprs: IndexSet::new(),
- }
- }
-
- // Create a new equivalence class from a pre-existing `Vec`
- pub fn new(exprs: Vec<Arc<dyn PhysicalExpr>>) -> Self {
- Self {
- exprs: exprs.into_iter().collect(),
+ // Create a new equivalence class from a pre-existing collection.
+ pub fn new(exprs: impl IntoIterator<Item = Arc<dyn PhysicalExpr>>) -> Self {
+ let mut class = Self::default();
+ for expr in exprs {
+ class.push(expr);
}
- }
-
- /// Return the inner vector of expressions
- pub fn into_vec(self) -> Vec<Arc<dyn PhysicalExpr>> {
- self.exprs.into_iter().collect()
+ class
}
/// Return the "canonical" expression for this class (the first element)
- /// if any
- fn canonical_expr(&self) -> Option<Arc<dyn PhysicalExpr>> {
- self.exprs.iter().next().cloned()
+ /// if non-empty.
+ pub fn canonical_expr(&self) -> Option<&Arc<dyn PhysicalExpr>> {
+ self.exprs.iter().next()
}
/// Insert the expression into this class, meaning it is known to be equal to
- /// all other expressions in this class
+ /// all other expressions in this class.
pub fn push(&mut self, expr: Arc<dyn PhysicalExpr>) {
+ if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+ let expr_across = AcrossPartitions::Uniform(Some(lit.value().clone()));
+ if let Some(across) = self.constant.as_mut() {
+ // TODO: Return an error if constant values do not agree.
+ if *across == AcrossPartitions::Heterogeneous {
Review Comment:
Why a new `Uniform` constant overwrites existing `Heterogeneous`? Shouldn't it be an error?
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126826372
##########
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 {
Review Comment:
We can change the name if there is a consensus on a better name. I don't really have a strong opinion on this. Regarding your last question, since `LexRequirement`s become non-degenerate, a lot of the call sites require updating. I tried organizing the changes relating to `OrderingRequirements` in a separate PR, but it didn't really reduce the diff because the call sites required updating anyway.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126895689
##########
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:
Agreed, let's keep this in mind in future PRs.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2950910237
I think there are some incoming big PRs, so I'd like to go ahead with this and help with fixing any issues @Omega359 finds after merge. In addition to the time cost of going through merges, I am fearful of accidentally introducing bugs every time I merge.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2934797274
> 🤖 `./gh_compare_branch_bench.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch_bench.sh) Running
I think the planning slowdown is real and reproducible.
I did some profiling and it seems it should be fairly easy to fix (we could do it in a follow on even) by optimizing the projection removal code:
<img width="1725" alt="Screenshot 2025-06-03 at 7 03 07 AM" src="https://github.com/user-attachments/assets/164e604d-7339-4b71-8f24-157a1851c9d0" />
<img width="1305" alt="Screenshot 2025-06-03 at 7 04 35 AM" src="https://github.com/user-attachments/assets/7178bdb8-d888-4be4-8bb8-e2c1116d8ca4" />
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2124345816
##########
datafusion/catalog/src/listing_schema.rs:
##########
@@ -143,7 +141,7 @@ impl ListingSchemaProvider {
order_exprs: vec![],
unbounded: false,
options: Default::default(),
- constraints: Constraints::empty(),
+ constraints: Default::default(),
Review Comment:
I just suggest to have a consistent way to represent it...you can see there are a lot `Constraints::default()` below.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2123722886
##########
datafusion/catalog/src/listing_schema.rs:
##########
@@ -143,7 +141,7 @@ impl ListingSchemaProvider {
order_exprs: vec![],
unbounded: false,
options: Default::default(),
- constraints: Constraints::empty(),
+ constraints: Default::default(),
Review Comment:
I think Default::default() is the same thing
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2124418429
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -845,7 +840,7 @@ pub struct SortExec {
/// 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: LexOrdering,
+ common_sort_prefix: Vec<PhysicalSortExpr>,
Review Comment:
`ExternalSorter`'s `expr` is changed from `PhysicalSortExpr`s to `LexOrdering`, but this changes from `LexOrdering`s to `PhysicalSortExpr`s. Can you clarify the reason? Why not having a consistent representation for sort expressions?
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -845,7 +840,7 @@ pub struct SortExec {
/// 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: LexOrdering,
+ common_sort_prefix: Vec<PhysicalSortExpr>,
Review Comment:
`ExternalSorter`'s `expr` is changed from `PhysicalSortExpr`s to `LexOrdering`, but this changes from `LexOrdering` to `PhysicalSortExpr`s. Can you clarify the reason? Why not having a consistent representation for sort expressions?
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2922442995
One thing on my mind is that maybe it's better to hold up the PR until DF48 release, then we can have enough time to check and verify the PR after it merges.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2922489274
> Move fast with the release while this gets reviews, which has the advantage of making DF48 a "checkpoint" before this PR. The downside is that we will need to do extensive testing both for the release and then one for the PR.
I think this is my preferred choice -- let's get DF 48 out and then give this one the maximum bake time. I think we can be careful of changes we merge to try and minimize the conflicts that this will collect
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2922519861
Sounds good -- let's go with the first choice (moving fast with the release and merging this right after) unless someone else brings a good argument to the contrary.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126871133
##########
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:
Exactly, will update the docstring accordingly.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "Omega359 (via GitHub)" <gi...@apache.org>.
Omega359 commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2954068858
fyi I did run this through my test suite and it seemed to work correctly :)
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126830302
##########
datafusion/functions-aggregate/src/first_last.rs:
##########
@@ -782,14 +756,92 @@ where
}
}
}
+
+#[derive(Debug)]
Review Comment:
Right -- will do.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126999507
##########
datafusion/catalog/src/listing_schema.rs:
##########
@@ -143,7 +141,7 @@ impl ListingSchemaProvider {
order_exprs: vec![],
unbounded: false,
options: Default::default(),
- constraints: Constraints::empty(),
+ constraints: Default::default(),
Review Comment:
I tried to conform to the existing style in the file. Some files use the `Default::default` pattern, while others use `TypeName::default` pattern. I agree that it'd be good if we have a uniform idiom for this case across the DF codebase. Maybe we can make this a PR in the future if there turns out to be a consensus on the style to use.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2922387217
😮
I will put this on my list to review asap
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2124345816
##########
datafusion/catalog/src/listing_schema.rs:
##########
@@ -143,7 +141,7 @@ impl ListingSchemaProvider {
order_exprs: vec![],
unbounded: false,
options: Default::default(),
- constraints: Constraints::empty(),
+ constraints: Default::default(),
Review Comment:
I just to have a consistent way to represent it...you can see there are a lot `Constraints::default()` below.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126736424
##########
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:
Right, updating accordingly.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126810121
##########
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 initially thought so too, but seems like Rust's convention is to use `try_new` when returning a `Result`, and simply `new` when returning an `Option`. Here is an example from the standard library: https://doc.rust-lang.org/std/num/type.NonZeroU32.html
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126891646
##########
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:
Agreed 👍
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126995806
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -175,307 +135,398 @@ impl ConstExpr {
}
}
+impl PartialEq for ConstExpr {
+ fn eq(&self, other: &Self) -> bool {
+ self.across_partitions == other.across_partitions && self.expr.eq(&other.expr)
+ }
+}
+
impl Display for ConstExpr {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.expr)?;
- match &self.across_partitions {
- AcrossPartitions::Heterogeneous => {
- write!(f, "(heterogeneous)")?;
- }
- AcrossPartitions::Uniform(value) => {
- if let Some(val) = value {
- write!(f, "(uniform: {val})")?;
- } else {
- write!(f, "(uniform: unknown)")?;
- }
- }
- }
- Ok(())
+ write!(f, "{}", self.across_partitions)
}
}
impl From<Arc<dyn PhysicalExpr>> for ConstExpr {
fn from(expr: Arc<dyn PhysicalExpr>) -> Self {
- Self::new(expr)
- }
-}
-
-impl From<&Arc<dyn PhysicalExpr>> for ConstExpr {
- fn from(expr: &Arc<dyn PhysicalExpr>) -> Self {
- Self::new(Arc::clone(expr))
+ // By default, assume constant expressions are not same across partitions.
+ // However, if we have a literal, it will have a single value that is the
+ // same across all partitions.
+ let across = if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+ AcrossPartitions::Uniform(Some(lit.value().clone()))
+ } else {
+ AcrossPartitions::Heterogeneous
+ };
+ Self {
+ expr,
+ across_partitions: across,
+ }
}
}
-/// Checks whether `expr` is among in the `const_exprs`.
-pub fn const_exprs_contains(
- const_exprs: &[ConstExpr],
- expr: &Arc<dyn PhysicalExpr>,
-) -> bool {
- const_exprs
- .iter()
- .any(|const_expr| const_expr.expr.eq(expr))
-}
-
/// An `EquivalenceClass` is a set of [`Arc<dyn PhysicalExpr>`]s that are known
/// to have the same value for all tuples in a relation. These are generated by
/// equality predicates (e.g. `a = b`), typically equi-join conditions and
/// equality conditions in filters.
///
/// Two `EquivalenceClass`es are equal if they contains the same expressions in
/// without any ordering.
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct EquivalenceClass {
- /// The expressions in this equivalence class. The order doesn't
- /// matter for equivalence purposes
- ///
- exprs: IndexSet<Arc<dyn PhysicalExpr>>,
-}
-
-impl PartialEq for EquivalenceClass {
- /// Returns true if other is equal in the sense
- /// of bags (multi-sets), disregarding their orderings.
- fn eq(&self, other: &Self) -> bool {
- self.exprs.eq(&other.exprs)
- }
+ /// The expressions in this equivalence class. The order doesn't matter for
+ /// equivalence purposes.
+ pub(crate) exprs: IndexSet<Arc<dyn PhysicalExpr>>,
+ /// Indicates whether the expressions in this equivalence class have a
+ /// constant value. A `Some` value indicates constant-ness.
+ pub(crate) constant: Option<AcrossPartitions>,
}
impl EquivalenceClass {
- /// Create a new empty equivalence class
- pub fn new_empty() -> Self {
- Self {
- exprs: IndexSet::new(),
- }
- }
-
- // Create a new equivalence class from a pre-existing `Vec`
- pub fn new(exprs: Vec<Arc<dyn PhysicalExpr>>) -> Self {
- Self {
- exprs: exprs.into_iter().collect(),
+ // Create a new equivalence class from a pre-existing collection.
+ pub fn new(exprs: impl IntoIterator<Item = Arc<dyn PhysicalExpr>>) -> Self {
+ let mut class = Self::default();
+ for expr in exprs {
+ class.push(expr);
}
- }
-
- /// Return the inner vector of expressions
- pub fn into_vec(self) -> Vec<Arc<dyn PhysicalExpr>> {
- self.exprs.into_iter().collect()
+ class
}
/// Return the "canonical" expression for this class (the first element)
- /// if any
- fn canonical_expr(&self) -> Option<Arc<dyn PhysicalExpr>> {
- self.exprs.iter().next().cloned()
+ /// if non-empty.
+ pub fn canonical_expr(&self) -> Option<&Arc<dyn PhysicalExpr>> {
+ self.exprs.iter().next()
}
/// Insert the expression into this class, meaning it is known to be equal to
- /// all other expressions in this class
+ /// all other expressions in this class.
pub fn push(&mut self, expr: Arc<dyn PhysicalExpr>) {
+ if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+ let expr_across = AcrossPartitions::Uniform(Some(lit.value().clone()));
+ if let Some(across) = self.constant.as_mut() {
+ // TODO: Return an error if constant values do not agree.
+ if *across == AcrossPartitions::Heterogeneous {
+ *across = expr_across;
+ }
Review Comment:
I agree 👍 It is indeed one of the follow-on TODOs to fix in the near term.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2941555626
I removed the conflicts, this is in a good state now.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "Omega359 (via GitHub)" <gi...@apache.org>.
Omega359 commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2944040199
I plan on running my test suite against this side branch hopefully tomorrow or early next week to verify 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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2932356312
🤖: Benchmark completed
<details><summary>Details</summary>
<p>
```
group main required-input-ordering
----- ---- -----------------------
logical_aggregate_with_join 1.01 736.0±4.72µs ? ?/sec 1.00 730.8±3.47µs ? ?/sec
logical_select_all_from_1000 1.01 125.3±0.28ms ? ?/sec 1.00 124.5±0.75ms ? ?/sec
logical_select_one_from_700 1.00 410.9±1.54µs ? ?/sec 1.00 412.0±1.37µs ? ?/sec
logical_trivial_join_high_numbered_columns 1.00 377.2±1.11µs ? ?/sec 1.00 377.5±2.10µs ? ?/sec
logical_trivial_join_low_numbered_columns 1.00 362.6±1.18µs ? ?/sec 1.00 363.5±2.03µs ? ?/sec
physical_intersection 1.00 842.7±6.01µs ? ?/sec 1.02 863.4±10.79µs ? ?/sec
physical_join_consider_sort 1.00 1356.9±5.64µs ? ?/sec 1.03 1395.7±5.69µs ? ?/sec
physical_join_distinct 1.01 353.9±3.21µs ? ?/sec 1.00 351.4±1.13µs ? ?/sec
physical_many_self_joins 1.00 10.1±0.02ms ? ?/sec 1.00 10.1±0.04ms ? ?/sec
physical_plan_clickbench_all 1.00 146.6±1.33ms ? ?/sec 1.01 148.4±1.44ms ? ?/sec
physical_plan_clickbench_q1 1.00 1705.1±20.55µs ? ?/sec 1.01 1725.0±15.43µs ? ?/sec
physical_plan_clickbench_q10 1.00 2.5±0.03ms ? ?/sec 1.02 2.5±0.04ms ? ?/sec
physical_plan_clickbench_q11 1.00 2.6±0.02ms ? ?/sec 1.02 2.6±0.04ms ? ?/sec
physical_plan_clickbench_q12 1.00 2.7±0.04ms ? ?/sec 1.03 2.8±0.02ms ? ?/sec
physical_plan_clickbench_q13 1.00 2.4±0.02ms ? ?/sec 1.02 2.4±0.02ms ? ?/sec
physical_plan_clickbench_q14 1.00 2.6±0.02ms ? ?/sec 1.03 2.6±0.03ms ? ?/sec
physical_plan_clickbench_q15 1.00 2.5±0.03ms ? ?/sec 1.04 2.6±0.02ms ? ?/sec
physical_plan_clickbench_q16 1.00 2.3±0.03ms ? ?/sec 1.05 2.4±0.04ms ? ?/sec
physical_plan_clickbench_q17 1.00 2.4±0.04ms ? ?/sec 1.05 2.6±0.04ms ? ?/sec
physical_plan_clickbench_q18 1.00 1977.1±11.26µs ? ?/sec 1.02 2.0±0.03ms ? ?/sec
physical_plan_clickbench_q19 1.00 2.9±0.01ms ? ?/sec 1.05 3.0±0.02ms ? ?/sec
physical_plan_clickbench_q2 1.00 1957.0±25.76µs ? ?/sec 1.02 1994.0±18.70µs ? ?/sec
physical_plan_clickbench_q20 1.00 1674.5±15.67µs ? ?/sec 1.01 1683.8±25.23µs ? ?/sec
physical_plan_clickbench_q21 1.00 1993.9±24.54µs ? ?/sec 1.00 1999.9±19.07µs ? ?/sec
physical_plan_clickbench_q22 1.00 2.6±0.03ms ? ?/sec 1.02 2.7±0.01ms ? ?/sec
physical_plan_clickbench_q23 1.00 3.0±0.02ms ? ?/sec 1.02 3.0±0.03ms ? ?/sec
physical_plan_clickbench_q24 1.00 4.7±0.03ms ? ?/sec 1.02 4.7±0.04ms ? ?/sec
physical_plan_clickbench_q25 1.00 2.0±0.03ms ? ?/sec 1.03 2.1±0.03ms ? ?/sec
physical_plan_clickbench_q26 1.00 1834.0±27.02µs ? ?/sec 1.03 1887.5±26.64µs ? ?/sec
physical_plan_clickbench_q27 1.00 2.1±0.03ms ? ?/sec 1.02 2.1±0.03ms ? ?/sec
physical_plan_clickbench_q28 1.00 2.9±0.03ms ? ?/sec 1.01 3.0±0.03ms ? ?/sec
physical_plan_clickbench_q29 1.00 3.6±0.03ms ? ?/sec 1.02 3.7±0.04ms ? ?/sec
physical_plan_clickbench_q3 1.00 1929.8±19.37µs ? ?/sec 1.04 2.0±0.02ms ? ?/sec
physical_plan_clickbench_q30 1.09 15.1±0.15ms ? ?/sec 1.00 13.9±0.13ms ? ?/sec
physical_plan_clickbench_q31 1.00 2.9±0.02ms ? ?/sec 1.03 3.0±0.03ms ? ?/sec
physical_plan_clickbench_q32 1.00 2.9±0.02ms ? ?/sec 1.02 3.0±0.02ms ? ?/sec
physical_plan_clickbench_q33 1.00 2.5±0.03ms ? ?/sec 1.01 2.5±0.01ms ? ?/sec
physical_plan_clickbench_q34 1.00 2.1±0.02ms ? ?/sec 1.03 2.2±0.02ms ? ?/sec
physical_plan_clickbench_q35 1.00 2.2±0.03ms ? ?/sec 1.02 2.3±0.03ms ? ?/sec
physical_plan_clickbench_q36 1.00 3.0±0.03ms ? ?/sec 1.02 3.1±0.03ms ? ?/sec
physical_plan_clickbench_q37 1.00 3.1±0.04ms ? ?/sec 1.01 3.1±0.03ms ? ?/sec
physical_plan_clickbench_q38 1.00 3.1±0.03ms ? ?/sec 1.02 3.1±0.03ms ? ?/sec
physical_plan_clickbench_q39 1.00 2.8±0.03ms ? ?/sec 1.01 2.9±0.04ms ? ?/sec
physical_plan_clickbench_q4 1.00 1668.3±14.38µs ? ?/sec 1.03 1714.2±24.87µs ? ?/sec
physical_plan_clickbench_q40 1.00 3.5±0.03ms ? ?/sec 1.02 3.5±0.03ms ? ?/sec
physical_plan_clickbench_q41 1.00 3.0±0.02ms ? ?/sec 1.02 3.1±0.03ms ? ?/sec
physical_plan_clickbench_q42 1.00 3.0±0.04ms ? ?/sec 1.00 3.0±0.04ms ? ?/sec
physical_plan_clickbench_q43 1.00 3.1±0.04ms ? ?/sec 1.07 3.3±0.11ms ? ?/sec
physical_plan_clickbench_q44 1.00 1830.7±25.31µs ? ?/sec 1.00 1835.2±18.71µs ? ?/sec
physical_plan_clickbench_q45 1.00 1844.6±20.01µs ? ?/sec 1.00 1844.1±30.61µs ? ?/sec
physical_plan_clickbench_q46 1.00 2.3±0.03ms ? ?/sec 1.01 2.3±0.02ms ? ?/sec
physical_plan_clickbench_q47 1.00 2.9±0.03ms ? ?/sec 1.01 2.9±0.03ms ? ?/sec
physical_plan_clickbench_q48 1.00 3.5±0.04ms ? ?/sec 1.01 3.5±0.05ms ? ?/sec
physical_plan_clickbench_q49 1.03 3.9±0.04ms ? ?/sec 1.00 3.8±0.03ms ? ?/sec
physical_plan_clickbench_q5 1.00 1882.5±26.37µs ? ?/sec 1.02 1929.0±23.74µs ? ?/sec
physical_plan_clickbench_q50 1.00 3.5±0.02ms ? ?/sec 1.00 3.6±0.02ms ? ?/sec
physical_plan_clickbench_q51 1.00 2.4±0.02ms ? ?/sec 1.01 2.4±0.01ms ? ?/sec
physical_plan_clickbench_q52 1.00 3.2±0.03ms ? ?/sec 1.02 3.3±0.02ms ? ?/sec
physical_plan_clickbench_q6 1.00 1865.0±14.75µs ? ?/sec 1.03 1926.6±35.69µs ? ?/sec
physical_plan_clickbench_q7 1.00 1714.9±17.28µs ? ?/sec 1.02 1747.1±25.25µs ? ?/sec
physical_plan_clickbench_q8 1.00 2.4±0.02ms ? ?/sec 1.04 2.5±0.02ms ? ?/sec
physical_plan_clickbench_q9 1.00 2.3±0.03ms ? ?/sec 1.02 2.4±0.03ms ? ?/sec
physical_plan_tpcds_all 1.00 1090.9±3.37ms ? ?/sec 1.04 1129.2±7.87ms ? ?/sec
physical_plan_tpch_all 1.00 65.2±0.92ms ? ?/sec 1.04 67.6±0.82ms ? ?/sec
physical_plan_tpch_q1 1.00 2.2±0.02ms ? ?/sec 1.04 2.3±0.01ms ? ?/sec
physical_plan_tpch_q10 1.00 3.3±0.02ms ? ?/sec 1.05 3.5±0.01ms ? ?/sec
physical_plan_tpch_q11 1.00 3.5±0.01ms ? ?/sec 1.05 3.6±0.02ms ? ?/sec
physical_plan_tpch_q12 1.00 1962.6±9.99µs ? ?/sec 1.04 2.0±0.02ms ? ?/sec
physical_plan_tpch_q13 1.00 1497.3±6.27µs ? ?/sec 1.06 1579.9±25.05µs ? ?/sec
physical_plan_tpch_q14 1.00 2.0±0.01ms ? ?/sec 1.05 2.1±0.02ms ? ?/sec
physical_plan_tpch_q16 1.00 2.6±0.01ms ? ?/sec 1.05 2.7±0.02ms ? ?/sec
physical_plan_tpch_q17 1.00 2.6±0.01ms ? ?/sec 1.02 2.6±0.01ms ? ?/sec
physical_plan_tpch_q18 1.00 2.8±0.01ms ? ?/sec 1.05 2.9±0.03ms ? ?/sec
physical_plan_tpch_q19 1.00 4.0±0.03ms ? ?/sec 1.01 4.0±0.03ms ? ?/sec
physical_plan_tpch_q2 1.00 5.7±0.03ms ? ?/sec 1.04 5.9±0.01ms ? ?/sec
physical_plan_tpch_q20 1.00 3.3±0.04ms ? ?/sec 1.03 3.4±0.01ms ? ?/sec
physical_plan_tpch_q21 1.00 4.4±0.03ms ? ?/sec 1.03 4.5±0.06ms ? ?/sec
physical_plan_tpch_q22 1.00 3.0±0.03ms ? ?/sec 1.03 3.1±0.03ms ? ?/sec
physical_plan_tpch_q3 1.00 2.3±0.01ms ? ?/sec 1.05 2.4±0.03ms ? ?/sec
physical_plan_tpch_q4 1.00 1662.8±8.32µs ? ?/sec 1.04 1729.9±12.29µs ? ?/sec
physical_plan_tpch_q5 1.00 3.2±0.01ms ? ?/sec 1.04 3.4±0.03ms ? ?/sec
physical_plan_tpch_q6 1.00 944.0±7.01µs ? ?/sec 1.02 964.7±6.33µs ? ?/sec
physical_plan_tpch_q7 1.00 4.4±0.02ms ? ?/sec 1.05 4.6±0.03ms ? ?/sec
physical_plan_tpch_q8 1.00 5.3±0.05ms ? ?/sec 1.03 5.5±0.02ms ? ?/sec
physical_plan_tpch_q9 1.00 4.2±0.01ms ? ?/sec 1.05 4.4±0.03ms ? ?/sec
physical_select_aggregates_from_200 1.19 26.1±0.07ms ? ?/sec 1.00 21.9±0.08ms ? ?/sec
physical_select_all_from_1000 1.01 139.7±0.27ms ? ?/sec 1.00 138.6±0.23ms ? ?/sec
physical_select_one_from_700 1.00 1038.2±5.50µs ? ?/sec 1.01 1045.3±5.50µs ? ?/sec
physical_sorted_union_orderby 1.37 64.2±1.20ms ? ?/sec 1.00 47.0±0.34ms ? ?/sec
physical_theta_join_consider_sort 1.00 1736.6±7.42µs ? ?/sec 1.01 1762.2±8.80µs ? ?/sec
physical_unnest_to_join 1.00 1329.5±14.09µs ? ?/sec 1.01 1337.2±5.08µs ? ?/sec
with_param_values_many_columns 1.01 166.3±1.24µs ? ?/sec 1.00 164.2±0.89µs ? ?/sec
```
</p>
</details>
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2122106227
##########
datafusion/catalog/src/listing_schema.rs:
##########
@@ -143,7 +141,7 @@ impl ListingSchemaProvider {
order_exprs: vec![],
unbounded: false,
options: Default::default(),
- constraints: Constraints::empty(),
+ constraints: Default::default(),
Review Comment:
Shouldn't it be `Constraints::default()`?
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2122154995
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -125,40 +120,85 @@ use itertools::Itertools;
/// # let col_c = col("c", &schema).unwrap();
/// // This object represents data that is sorted by a ASC, c DESC
/// // with a single constant value of b
-/// let mut eq_properties = EquivalenceProperties::new(schema)
-/// .with_constants(vec![ConstExpr::from(col_b)]);
-/// eq_properties.add_new_ordering(LexOrdering::new(vec![
+/// let mut eq_properties = EquivalenceProperties::new(schema);
+/// eq_properties.add_constants(vec![ConstExpr::from(col_b)]);
+/// eq_properties.add_ordering([
/// PhysicalSortExpr::new_default(col_a).asc(),
/// PhysicalSortExpr::new_default(col_c).desc(),
-/// ]));
+/// ]);
///
-/// assert_eq!(eq_properties.to_string(), "order: [[a@0 ASC, c@2 DESC]], const: [b@1(heterogeneous)]")
+/// assert_eq!(eq_properties.to_string(), "order: [[a@0 ASC, c@2 DESC]], eq: [{members: [b@1], constant: (heterogeneous)}]");
/// ```
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug)]
pub struct EquivalenceProperties {
- /// Distinct equivalence classes (exprs known to have the same expressions)
+ /// Distinct equivalence classes (i.e. expressions with the same value).
eq_group: EquivalenceGroup,
- /// Equivalent sort expressions
+ /// Equivalent sort expressions (i.e. those define the same ordering).
oeq_class: OrderingEquivalenceClass,
- /// Expressions whose values are constant
- ///
- /// TODO: We do not need to track constants separately, they can be tracked
- /// inside `eq_group` as `Literal` expressions.
- constants: Vec<ConstExpr>,
- /// Table constraints
+ /// Cache storing equivalent sort expressions in normal form (i.e. without
+ /// constants/duplicates and in standard form) and a map associating leading
+ /// terms with full sort expressions.
+ oeq_cache: OrderingEquivalenceCache,
+ /// Table constraints that factor in equivalence calculations.
constraints: Constraints,
/// Schema associated with this object.
schema: SchemaRef,
}
+/// This object serves as a cache for storing equivalent sort expressions
+/// in normal form, and a map associating leading sort expressions with
+/// full lexicographical orderings. With this information, DataFusion can
+/// efficiently determine whether a given ordering is satisfied by the
+/// existing orderings, and discover new orderings based on the existing
+/// equivalence properties.
+#[derive(Clone, Debug, Default)]
+struct OrderingEquivalenceCache {
+ /// Equivalent sort expressions in normal form.
+ normal_cls: OrderingEquivalenceClass,
Review Comment:
If this is an efficient way to leverage the info provided by `OrderingEquivalenceClass`, I wonder why you don't just incorporate cache into `OrderingEquivalenceClass` so we don't need to expose an additional cache?
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-3165532117
We found a regression related to this PR for `string_agg` not respecting `ORDER BY`
- https://github.com/apache/datafusion/issues/17011
@nuno-faria and @findepi have prepared PRs to fix this, and it would be great if @ozankabak you or someone on your team could also help review it
https://github.com/apache/datafusion/pull/17058
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2124399512
##########
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:
So wondering why the previous one doesn't do what the comment said.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126725189
##########
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:
Looks good, will incorporate.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126721050
##########
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:
Agreed, will add a remark to the docstring.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126827741
##########
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:
Agreed, will follow your suggestion.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126840903
##########
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:
An ordering equivalence class is basically a set of orderings, and if the table in question has no orderings, the set will be empty (which is a common/valid case). I will add some comments clarifying this.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126835220
##########
datafusion/functions-aggregate/src/first_last.rs:
##########
@@ -1195,13 +1234,90 @@ impl AggregateUDFImpl for LastValue {
}
}
+#[derive(Debug)]
+pub struct TrivialLastValueAccumulator {
Review Comment:
Right, will do.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126879246
##########
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:
IIRC this is a byproduct of how constants are handled (i.e. without special treatment in equivalence checks). It is indeed non-obvious, but is safe (unless I'm missing something).
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2127000873
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -579,302 +611,289 @@ impl EquivalenceProperties {
// From the analysis above, we know that `[a ASC]` is satisfied. Then,
// we add column `a` as constant to the algorithm state. This enables us
// to deduce that `(b + c) ASC` is satisfied, given `a` is constant.
- eq_properties = eq_properties.with_constants(std::iter::once(
- ConstExpr::from(Arc::clone(&normalized_req.expr)),
- ));
+ let const_expr = ConstExpr::from(element.expr);
+ eq_properties.add_constants(std::iter::once(const_expr))?;
}
+ Ok(true)
+ }
- // All requirements are satisfied.
- normalized_reqs.len()
+ /// Returns the number of consecutive sort expressions (starting from the
+ /// left) that are satisfied by the existing ordering.
+ fn common_sort_prefix_length(&self, normal_ordering: LexOrdering) -> Result<usize> {
+ let full_length = normal_ordering.len();
+ // Check whether the given ordering is satisfied by constraints:
+ if self.satisfied_by_constraints_ordering(&normal_ordering) {
+ // If constraints satisfy all sort expressions, return the full
+ // length:
+ return Ok(full_length);
+ }
+ let schema = self.schema();
+ let mut eq_properties = self.clone();
+ for (idx, element) in normal_ordering.into_iter().enumerate() {
+ // Check whether given ordering is satisfied:
+ let ExprProperties {
+ sort_properties, ..
+ } = eq_properties.get_expr_properties(Arc::clone(&element.expr));
+ let satisfy = match sort_properties {
+ SortProperties::Ordered(options) => options_compatible(
+ &options,
+ &element.options,
+ element.expr.nullable(schema).unwrap_or(true),
+ ),
+ // Singleton expressions satisfy any ordering.
+ SortProperties::Singleton => true,
+ SortProperties::Unordered => false,
+ };
+ if !satisfy {
+ // As soon as one sort expression is unsatisfied, return how
+ // many we've satisfied so far:
+ return Ok(idx);
+ }
+ // Treat satisfied keys as constants in subsequent iterations. We
+ // can do this because the "next" key only matters in a lexicographical
+ // ordering when the keys to its left have the same values.
+ //
+ // Note that these expressions are not properly "constants". This is just
+ // an implementation strategy confined to this function.
+ //
+ // For example, assume that the requirement is `[a ASC, (b + c) ASC]`,
+ // and existing equivalent orderings are `[a ASC, b ASC]` and `[c ASC]`.
+ // From the analysis above, we know that `[a ASC]` is satisfied. Then,
+ // we add column `a` as constant to the algorithm state. This enables us
+ // to deduce that `(b + c) ASC` is satisfied, given `a` is constant.
+ let const_expr = ConstExpr::from(element.expr);
+ eq_properties.add_constants(std::iter::once(const_expr))?
+ }
+ // All sort expressions are satisfied, return full length:
+ Ok(full_length)
}
- /// Determines the longest prefix of `reqs` that is satisfied by the existing ordering.
- /// Returns that prefix as a new `LexRequirement`, and a boolean indicating if all the requirements are satisfied.
+ /// Determines the longest normal prefix of `ordering` satisfied by the
+ /// existing ordering. Returns that prefix as a new `LexOrdering`, and a
+ /// boolean indicating whether all the sort expressions are satisfied.
pub fn extract_common_sort_prefix(
&self,
- reqs: &LexRequirement,
- ) -> (LexRequirement, bool) {
- // First, standardize the given requirement:
- let normalized_reqs = self.normalize_sort_requirements(reqs);
-
- let prefix_len = self.compute_common_sort_prefix_length(&normalized_reqs);
- (
- LexRequirement::new(normalized_reqs[..prefix_len].to_vec()),
- prefix_len == normalized_reqs.len(),
- )
- }
-
- /// Checks whether the given sort requirements are satisfied by any of the
- /// existing orderings.
- pub fn ordering_satisfy_requirement(&self, reqs: &LexRequirement) -> bool {
- self.extract_common_sort_prefix(reqs).1
+ ordering: LexOrdering,
+ ) -> Result<(Vec<PhysicalSortExpr>, bool)> {
+ // First, standardize the given ordering:
+ let Some(normal_ordering) = self.normalize_sort_exprs(ordering) else {
+ // If the ordering vanishes after normalization, it is satisfied:
+ return Ok((vec![], true));
+ };
+ let prefix_len = self.common_sort_prefix_length(normal_ordering.clone())?;
+ let flag = prefix_len == normal_ordering.len();
+ let mut sort_exprs: Vec<_> = normal_ordering.into();
+ if !flag {
+ sort_exprs.truncate(prefix_len);
+ }
+ Ok((sort_exprs, flag))
}
- /// Checks if the sort requirements are satisfied by any of the table constraints (primary key or unique).
- /// Returns true if any constraint fully satisfies the requirements.
- fn satisfied_by_constraints(
+ /// Checks if the sort expressions are satisfied by any of the table
+ /// constraints (primary key or unique). Returns true if any constraint
+ /// fully satisfies the expressions (i.e. constraint indices form a valid
+ /// prefix of an existing ordering that matches the expressions). For
+ /// unique constraints, also verifies nullable columns.
+ fn satisfied_by_constraints_ordering(
&self,
- normalized_reqs: &[PhysicalSortRequirement],
+ normal_exprs: &[PhysicalSortExpr],
) -> bool {
self.constraints.iter().any(|constraint| match constraint {
- Constraint::PrimaryKey(indices) | Constraint::Unique(indices) => self
- .satisfied_by_constraint(
- normalized_reqs,
- indices,
- matches!(constraint, Constraint::Unique(_)),
- ),
- })
- }
-
- /// Checks if sort requirements are satisfied by a constraint (primary key or unique).
- /// Returns true if the constraint indices form a valid prefix of an existing ordering
- /// that matches the requirements. For unique constraints, also verifies nullable columns.
- fn satisfied_by_constraint(
- &self,
- normalized_reqs: &[PhysicalSortRequirement],
- indices: &[usize],
- check_null: bool,
- ) -> bool {
- // Requirements must contain indices
- if indices.len() > normalized_reqs.len() {
- return false;
- }
-
- // Iterate over all orderings
- self.oeq_class.iter().any(|ordering| {
- if indices.len() > ordering.len() {
- return false;
- }
-
- // Build a map of column positions in the ordering
- let mut col_positions = HashMap::with_capacity(ordering.len());
- for (pos, req) in ordering.iter().enumerate() {
- if let Some(col) = req.expr.as_any().downcast_ref::<Column>() {
- col_positions.insert(
- col.index(),
- (pos, col.nullable(&self.schema).unwrap_or(true)),
- );
- }
- }
-
- // Check if all constraint indices appear in valid positions
- if !indices.iter().all(|&idx| {
- col_positions
- .get(&idx)
- .map(|&(pos, nullable)| {
- // For unique constraints, verify column is not nullable if it's first/last
- !check_null
- || (pos != 0 && pos != ordering.len() - 1)
- || !nullable
+ Constraint::PrimaryKey(indices) | Constraint::Unique(indices) => {
+ let check_null = matches!(constraint, Constraint::Unique(_));
+ let normalized_size = normal_exprs.len();
+ indices.len() <= normalized_size
+ && self.oeq_class.iter().any(|ordering| {
+ let length = ordering.len();
+ if indices.len() > length || normalized_size < length {
+ return false;
+ }
+ // Build a map of column positions in the ordering:
+ let mut col_positions = HashMap::with_capacity(length);
+ for (pos, req) in ordering.iter().enumerate() {
+ if let Some(col) = req.expr.as_any().downcast_ref::<Column>()
+ {
+ let nullable = col.nullable(&self.schema).unwrap_or(true);
+ col_positions.insert(col.index(), (pos, nullable));
+ }
+ }
+ // Check if all constraint indices appear in valid positions:
+ if !indices.iter().all(|idx| {
+ col_positions.get(idx).is_some_and(|&(pos, nullable)| {
+ // For unique constraints, verify column is not nullable if it's first/last:
+ !check_null
+ || !nullable
+ || (pos != 0 && pos != length - 1)
+ })
+ }) {
+ return false;
+ }
+ // Check if this ordering matches the prefix:
+ normal_exprs.iter().zip(ordering).all(|(given, existing)| {
+ existing.satisfy_expr(given, &self.schema)
+ })
})
- .unwrap_or(false)
- }) {
- return false;
}
-
- // Check if this ordering matches requirements prefix
- let ordering_len = ordering.len();
- normalized_reqs.len() >= ordering_len
- && normalized_reqs[..ordering_len].iter().zip(ordering).all(
- |(req, existing)| {
- req.expr.eq(&existing.expr)
- && req
- .options
- .is_none_or(|req_opts| req_opts == existing.options)
- },
- )
})
}
- /// Determines whether the ordering specified by the given sort requirement
- /// is satisfied based on the orderings within, equivalence classes, and
- /// constant expressions.
- ///
- /// # Parameters
- ///
- /// - `req`: A reference to a `PhysicalSortRequirement` for which the ordering
- /// satisfaction check will be done.
- ///
- /// # Returns
- ///
- /// Returns `true` if the specified ordering is satisfied, `false` otherwise.
- fn ordering_satisfy_single(&self, req: &PhysicalSortRequirement) -> bool {
- let ExprProperties {
- sort_properties, ..
- } = self.get_expr_properties(Arc::clone(&req.expr));
- match sort_properties {
- SortProperties::Ordered(options) => {
- let sort_expr = PhysicalSortExpr {
- expr: Arc::clone(&req.expr),
- options,
- };
- sort_expr.satisfy(req, self.schema())
+ /// Checks if the sort requirements are satisfied by any of the table
+ /// constraints (primary key or unique). Returns true if any constraint
+ /// fully satisfies the requirements (i.e. constraint indices form a valid
+ /// prefix of an existing ordering that matches the requirements). For
+ /// unique constraints, also verifies nullable columns.
+ fn satisfied_by_constraints(&self, normal_reqs: &[PhysicalSortRequirement]) -> bool {
+ self.constraints.iter().any(|constraint| match constraint {
+ Constraint::PrimaryKey(indices) | Constraint::Unique(indices) => {
+ let check_null = matches!(constraint, Constraint::Unique(_));
+ let normalized_size = normal_reqs.len();
+ indices.len() <= normalized_size
+ && self.oeq_class.iter().any(|ordering| {
+ let length = ordering.len();
+ if indices.len() > length || normalized_size < length {
+ return false;
+ }
+ // Build a map of column positions in the ordering:
+ let mut col_positions = HashMap::with_capacity(length);
+ for (pos, req) in ordering.iter().enumerate() {
+ if let Some(col) = req.expr.as_any().downcast_ref::<Column>()
+ {
+ let nullable = col.nullable(&self.schema).unwrap_or(true);
+ col_positions.insert(col.index(), (pos, nullable));
+ }
+ }
+ // Check if all constraint indices appear in valid positions:
+ if !indices.iter().all(|idx| {
+ col_positions.get(idx).is_some_and(|&(pos, nullable)| {
+ // For unique constraints, verify column is not nullable if it's first/last:
+ !check_null
+ || !nullable
+ || (pos != 0 && pos != length - 1)
+ })
+ }) {
+ return false;
+ }
+ // Check if this ordering matches the prefix:
+ normal_reqs.iter().zip(ordering).all(|(given, existing)| {
+ existing.satisfy(given, &self.schema)
+ })
+ })
}
- // Singleton expressions satisfies any ordering.
- SortProperties::Singleton => true,
- SortProperties::Unordered => false,
- }
+ })
}
/// Checks whether the `given` sort requirements are equal or more specific
/// than the `reference` sort requirements.
pub fn requirements_compatible(
&self,
- given: &LexRequirement,
- reference: &LexRequirement,
+ given: LexRequirement,
+ reference: LexRequirement,
) -> bool {
- let normalized_given = self.normalize_sort_requirements(given);
- let normalized_reference = self.normalize_sort_requirements(reference);
-
- (normalized_reference.len() <= normalized_given.len())
- && normalized_reference
+ let Some(normal_given) = self.normalize_sort_requirements(given) else {
+ return true;
+ };
+ let Some(normal_reference) = self.normalize_sort_requirements(reference) else {
+ return true;
+ };
+
+ (normal_reference.len() <= normal_given.len())
+ && normal_reference
.into_iter()
- .zip(normalized_given)
+ .zip(normal_given)
.all(|(reference, given)| given.compatible(&reference))
}
- /// Returns the finer ordering among the orderings `lhs` and `rhs`, breaking
- /// any ties by choosing `lhs`.
- ///
- /// The finer ordering is the ordering that satisfies both of the orderings.
- /// If the orderings are incomparable, returns `None`.
- ///
- /// For example, the finer ordering among `[a ASC]` and `[a ASC, b ASC]` is
- /// the latter.
- pub fn get_finer_ordering(
- &self,
- lhs: &LexOrdering,
- rhs: &LexOrdering,
- ) -> Option<LexOrdering> {
- // Convert the given sort expressions to sort requirements:
- let lhs = LexRequirement::from(lhs.clone());
- let rhs = LexRequirement::from(rhs.clone());
- let finer = self.get_finer_requirement(&lhs, &rhs);
- // Convert the chosen sort requirements back to sort expressions:
- finer.map(LexOrdering::from)
- }
-
- /// Returns the finer ordering among the requirements `lhs` and `rhs`,
- /// breaking any ties by choosing `lhs`.
+ /// Modify existing orderings by substituting sort expressions with appropriate
+ /// targets from the projection mapping. We substitute a sort expression when
+ /// its physical expression has a one-to-one functional relationship with a
+ /// target expression in the mapping.
///
- /// The finer requirements are the ones that satisfy both of the given
- /// requirements. If the requirements are incomparable, returns `None`.
+ /// After substitution, we may generate more than one `LexOrdering` for each
+ /// existing equivalent ordering. For example, `[a ASC, b ASC]` will turn
+ /// into `[CAST(a) ASC, b ASC]` and `[a ASC, b ASC]` when applying projection
+ /// expressions `a, b, CAST(a)`.
///
- /// For example, the finer requirements among `[a ASC]` and `[a ASC, b ASC]`
- /// is the latter.
- pub fn get_finer_requirement(
- &self,
- req1: &LexRequirement,
- req2: &LexRequirement,
- ) -> Option<LexRequirement> {
- let mut lhs = self.normalize_sort_requirements(req1);
- let mut rhs = self.normalize_sort_requirements(req2);
- lhs.inner
- .iter_mut()
- .zip(rhs.inner.iter_mut())
- .all(|(lhs, rhs)| {
- lhs.expr.eq(&rhs.expr)
- && match (lhs.options, rhs.options) {
- (Some(lhs_opt), Some(rhs_opt)) => lhs_opt == rhs_opt,
- (Some(options), None) => {
- rhs.options = Some(options);
- true
- }
- (None, Some(options)) => {
- lhs.options = Some(options);
- true
- }
- (None, None) => true,
- }
- })
- .then_some(if lhs.len() >= rhs.len() { lhs } else { rhs })
- }
-
- /// we substitute the ordering according to input expression type, this is a simplified version
- /// In this case, we just substitute when the expression satisfy the following condition:
- /// I. just have one column and is a CAST expression
- /// TODO: Add one-to-ones analysis for monotonic ScalarFunctions.
- /// TODO: we could precompute all the scenario that is computable, for example: atan(x + 1000) should also be substituted if
- /// x is DESC or ASC
- /// After substitution, we may generate more than 1 `LexOrdering`. As an example,
- /// `[a ASC, b ASC]` will turn into `[a ASC, b ASC], [CAST(a) ASC, b ASC]` when projection expressions `a, b, CAST(a)` is applied.
- pub fn substitute_ordering_component(
- &self,
+ /// TODO: Handle all scenarios that allow substitution; e.g. when `x` is
Review Comment:
Makes sense, will do.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2126992798
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -175,307 +135,398 @@ impl ConstExpr {
}
}
+impl PartialEq for ConstExpr {
+ fn eq(&self, other: &Self) -> bool {
+ self.across_partitions == other.across_partitions && self.expr.eq(&other.expr)
+ }
+}
+
impl Display for ConstExpr {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.expr)?;
- match &self.across_partitions {
- AcrossPartitions::Heterogeneous => {
- write!(f, "(heterogeneous)")?;
- }
- AcrossPartitions::Uniform(value) => {
- if let Some(val) = value {
- write!(f, "(uniform: {val})")?;
- } else {
- write!(f, "(uniform: unknown)")?;
- }
- }
- }
- Ok(())
+ write!(f, "{}", self.across_partitions)
}
}
impl From<Arc<dyn PhysicalExpr>> for ConstExpr {
fn from(expr: Arc<dyn PhysicalExpr>) -> Self {
- Self::new(expr)
- }
-}
-
-impl From<&Arc<dyn PhysicalExpr>> for ConstExpr {
- fn from(expr: &Arc<dyn PhysicalExpr>) -> Self {
- Self::new(Arc::clone(expr))
+ // By default, assume constant expressions are not same across partitions.
+ // However, if we have a literal, it will have a single value that is the
+ // same across all partitions.
+ let across = if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+ AcrossPartitions::Uniform(Some(lit.value().clone()))
+ } else {
+ AcrossPartitions::Heterogeneous
+ };
+ Self {
+ expr,
+ across_partitions: across,
+ }
}
}
-/// Checks whether `expr` is among in the `const_exprs`.
-pub fn const_exprs_contains(
- const_exprs: &[ConstExpr],
- expr: &Arc<dyn PhysicalExpr>,
-) -> bool {
- const_exprs
- .iter()
- .any(|const_expr| const_expr.expr.eq(expr))
-}
-
/// An `EquivalenceClass` is a set of [`Arc<dyn PhysicalExpr>`]s that are known
/// to have the same value for all tuples in a relation. These are generated by
/// equality predicates (e.g. `a = b`), typically equi-join conditions and
/// equality conditions in filters.
///
/// Two `EquivalenceClass`es are equal if they contains the same expressions in
/// without any ordering.
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct EquivalenceClass {
- /// The expressions in this equivalence class. The order doesn't
- /// matter for equivalence purposes
- ///
- exprs: IndexSet<Arc<dyn PhysicalExpr>>,
-}
-
-impl PartialEq for EquivalenceClass {
- /// Returns true if other is equal in the sense
- /// of bags (multi-sets), disregarding their orderings.
- fn eq(&self, other: &Self) -> bool {
- self.exprs.eq(&other.exprs)
- }
+ /// The expressions in this equivalence class. The order doesn't matter for
+ /// equivalence purposes.
+ pub(crate) exprs: IndexSet<Arc<dyn PhysicalExpr>>,
+ /// Indicates whether the expressions in this equivalence class have a
+ /// constant value. A `Some` value indicates constant-ness.
+ pub(crate) constant: Option<AcrossPartitions>,
}
impl EquivalenceClass {
- /// Create a new empty equivalence class
- pub fn new_empty() -> Self {
- Self {
- exprs: IndexSet::new(),
- }
- }
-
- // Create a new equivalence class from a pre-existing `Vec`
- pub fn new(exprs: Vec<Arc<dyn PhysicalExpr>>) -> Self {
- Self {
- exprs: exprs.into_iter().collect(),
+ // Create a new equivalence class from a pre-existing collection.
+ pub fn new(exprs: impl IntoIterator<Item = Arc<dyn PhysicalExpr>>) -> Self {
+ let mut class = Self::default();
+ for expr in exprs {
+ class.push(expr);
}
- }
-
- /// Return the inner vector of expressions
- pub fn into_vec(self) -> Vec<Arc<dyn PhysicalExpr>> {
- self.exprs.into_iter().collect()
+ class
}
/// Return the "canonical" expression for this class (the first element)
- /// if any
- fn canonical_expr(&self) -> Option<Arc<dyn PhysicalExpr>> {
- self.exprs.iter().next().cloned()
+ /// if non-empty.
+ pub fn canonical_expr(&self) -> Option<&Arc<dyn PhysicalExpr>> {
+ self.exprs.iter().next()
}
/// Insert the expression into this class, meaning it is known to be equal to
- /// all other expressions in this class
+ /// all other expressions in this class.
pub fn push(&mut self, expr: Arc<dyn PhysicalExpr>) {
+ if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+ let expr_across = AcrossPartitions::Uniform(Some(lit.value().clone()));
+ if let Some(across) = self.constant.as_mut() {
+ // TODO: Return an error if constant values do not agree.
+ if *across == AcrossPartitions::Heterogeneous {
Review Comment:
No, in this case we get partition uniformity information about a constant we didn't have before, so we only update our "books".
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#discussion_r2127003566
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -845,7 +840,7 @@ pub struct SortExec {
/// 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: LexOrdering,
+ common_sort_prefix: Vec<PhysicalSortExpr>,
Review Comment:
Good observation -- this is because we don't have a proper ordering here, it is just a prefix of sort expressions (which can be empty unlike an ordering).
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-2950730918
> I removed the conflicts, this is in a good state now. Did we get to the 48 cut-off point yet?
We are close I think
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [MAJOR] Equivalence System Overhaul [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #16217:
URL: https://github.com/apache/datafusion/pull/16217#issuecomment-3166735927
> We found a regression related to this PR for `string_agg` not respecting `ORDER BY`
>
> * [`string_agg` does not respect `ORDER BY` on `49.0.0` #17011](https://github.com/apache/datafusion/issues/17011)
>
> @nuno-faria and @findepi have prepared PRs to fix this, and it would be great if @ozankabak you or someone on your team could also help review it
>
> * [[branch-49] fix: string_agg not respecting ORDER BY #17058](https://github.com/apache/datafusion/pull/17058)
I'll take a look at that today
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org