diff --git a/datafusion/physical-expr-common/src/sort_expr.rs b/datafusion/physical-expr-common/src/sort_expr.rs index 8af850e759c3f..7f61536f31379 100644 --- a/datafusion/physical-expr-common/src/sort_expr.rs +++ b/datafusion/physical-expr-common/src/sort_expr.rs @@ -344,6 +344,11 @@ pub fn options_compatible( /// * `ORDER BY A DESC NULLS FIRST` matches /// * `ORDER BY A ASC NULLS FIRST` matches (`ASC` and `NULL` options ignored) /// * `ORDER BY B DESC NULLS FIRST` does not match (different expr) +/// +/// See [Required orderings] for where requirements come from and how they +/// are checked against the orderings an input provides. +/// +/// [Required orderings]: https://docs.rs/datafusion-physical-expr/latest/datafusion_physical_expr/equivalence/struct.EquivalenceProperties.html#required-orderings #[derive(Clone, Debug)] pub struct PhysicalSortRequirement { /// Physical expression representing the column to sort @@ -697,6 +702,11 @@ impl From for Vec { /// 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. +/// +/// See [Required orderings] for where requirements come from and how they +/// are checked against the orderings an input provides. +/// +/// [Required orderings]: https://docs.rs/datafusion-physical-expr/latest/datafusion_physical_expr/equivalence/struct.EquivalenceProperties.html#required-orderings #[derive(Debug, Clone, PartialEq)] pub struct LexRequirement { reqs: Vec, @@ -807,6 +817,11 @@ impl From for LexOrdering { /// 1. It is non-degenerate, meaning it contains at least one ordering. The /// absence of an input ordering requirement is represented by a `None` value /// in `ExecutionPlan` APIs, which return an `Option`. +/// +/// See [Required orderings] for where requirements come from and how they +/// are checked against the orderings an input provides. +/// +/// [Required orderings]: https://docs.rs/datafusion-physical-expr/latest/datafusion_physical_expr/equivalence/struct.EquivalenceProperties.html#required-orderings #[derive(Debug, Clone, PartialEq)] pub enum OrderingRequirements { /// The operator is not able to work without one of these requirements. diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs index a94a38f47f75d..7a5c9e1f2f05a 100644 --- a/datafusion/physical-expr/src/equivalence/class.rs +++ b/datafusion/physical-expr/src/equivalence/class.rs @@ -24,16 +24,19 @@ use datafusion_physical_expr_common::physical_expr::{ use indexmap::{IndexMap, IndexSet}; -/// Represents whether a constant expression's value is uniform or varies across -/// partitions. Has two variants: -/// - `Heterogeneous`: The constant expression may have different values for -/// different partitions. -/// - `Uniform(Option)`: The constant expression has the same value -/// across all partitions, or is `None` if the value is unknown. +/// How a constant expression value varies across partitions. +/// +/// See [Constants] for why the difference matters. +/// +/// [Constants]: crate::equivalence::EquivalenceProperties#constants #[derive(Clone, Debug, Default, Eq, PartialEq)] pub enum AcrossPartitions { + /// The constant may have different values for different partitions. #[default] Heterogeneous, + /// The constant has the same value across all partitions. + /// + /// `Some` if the value is exactly known, `None` otherwise. Uniform(Option), } @@ -52,20 +55,10 @@ impl Display for AcrossPartitions { } } -/// A structure representing a expression known to be constant in a physical -/// execution plan. -/// -/// The `ConstExpr` struct encapsulates an expression that is constant during -/// the execution of a query. For example if a filter like `A = 5` appears -/// earlier in the plan, `A` would become a constant in subsequent operations. -/// -/// # Fields +/// An expression known to be constant during the execution of a query. /// -/// - `expr`: Constant expression for a node in the physical plan. -/// - `across_partitions`: A boolean flag indicating whether the constant -/// expression is the same across partitions. If set to `true`, the constant -/// expression has same value for all partitions. If set to `false`, the -/// constant expression may have different values for different partitions. +/// For example after a filter like `A = 5` in a the plan, the expression `A` +/// is know to be constant in subsequent operations. See [Constants] for more details. /// /// # Example /// @@ -76,11 +69,13 @@ impl Display for AcrossPartitions { /// // Create a constant expression from a physical expression: /// let const_expr = ConstExpr::from(col); /// ``` +/// +/// [Constants]: crate::equivalence::EquivalenceProperties#constants #[derive(Clone, Debug)] pub struct ConstExpr { /// The expression that is known to be constant (e.g. a `Column`). pub expr: Arc, - /// Indicates whether the constant have the same value across all partitions. + /// Does the constant have the same value across all partitions? pub across_partitions: AcrossPartitions, } // TODO: The `ConstExpr` definition above can be in an inconsistent state where @@ -157,13 +152,17 @@ impl From> for ConstExpr { } } -/// An `EquivalenceClass` is a set of [`Arc`]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. +/// A set of [`Arc`]s that are known +/// to have the same value for all tuples in a relation. +/// +/// `EquivaluenceClass`es are generated by equality predicates such as `a = b`, +/// typically equi-join conditions and conditions in filters. See +/// [Equivalent expressions] for more detail. /// -/// Two `EquivalenceClass`es are equal if they contains the same expressions in +/// Two `EquivalenceClass`es are equal if they contain the same expressions in /// without any ordering. +/// +/// [Equivalent expressions]: crate::equivalence::EquivalenceProperties#equivalent-expressions #[derive(Clone, Debug, Default, Eq, PartialEq)] pub struct EquivalenceClass { /// The expressions in this equivalence class. The order doesn't matter for @@ -300,8 +299,12 @@ type AugmentedMapping<'a> = IndexMap< (&'a ProjectionTargets, Option<&'a EquivalenceClass>), >; -/// A collection of distinct `EquivalenceClass`es. This object supports fast -/// lookups of expressions and their equivalence classes. +/// A set of distinct [`EquivalenceClass`]es. +/// +/// `EquivalenceGroup` supports fast lookups of expressions and their +/// equivalence classes. See [`EquivalenceProperties`] for more details. +/// +/// [`EquivalenceProperties`]: crate::equivalence::EquivalenceProperties #[derive(Clone, Debug, Default)] pub struct EquivalenceGroup { /// A mapping from expressions to their equivalence class key. diff --git a/datafusion/physical-expr/src/equivalence/grouping.rs b/datafusion/physical-expr/src/equivalence/grouping.rs index 2fce6dad8292d..ab84fb1f49836 100644 --- a/datafusion/physical-expr/src/equivalence/grouping.rs +++ b/datafusion/physical-expr/src/equivalence/grouping.rs @@ -22,10 +22,38 @@ use datafusion_physical_expr_common::physical_expr::format_physical_expr_list; /// order. Contiguity applies to the complete partition stream, not separately /// to each record batch. /// +/// For example, consider the following table: +/// +/// ```text +/// ┌───┬───┬───┐ +/// │ a │ b │ c │ +/// ├───┼───┼───┤ +/// │ 2 │ 1 │ 5 │ +/// │ 2 │ 1 │ 3 │ +/// │ 1 │ 3 │ 4 │ +/// │ 2 │ 2 │ 1 │ +/// │ 2 │ 2 │ 2 │ +/// └───┴───┴───┘ +/// ``` +/// +/// * Is grouped by `[a, b]`, as rows with the same `(a, b)` are contiguous, +/// even though the table is not sorted by `a`, `b` nor `(a, b)`. +/// * Is grouped by `[b]`, since each value of `b` is in one run. +/// * Is NOT grouped by `[a]`, because `a = 2` appears in two separate runs. +/// /// Expression order within an entry is not significant: `[a, b]` and `[b, a]` /// describe the same grouping. Entries do not imply properties for subsets; -/// `[a, b]` alone says nothing about whether all rows with the same `a` are +/// as the table shows, `[a, b]` does not imply `[a]` as the same `a` may not be /// contiguous. +/// +/// # See also +/// * [`OrderingEquivalenceClass`] for the stronger property that rows are +/// sorted. +/// * [`EquivalenceProperties`] for how groupings relate to the other +/// properties of a plan node. +/// +/// [`EquivalenceProperties`]: crate::equivalence::EquivalenceProperties +/// [`OrderingEquivalenceClass`]: crate::equivalence::OrderingEquivalenceClass #[derive(Clone, Debug, Default)] pub struct GroupingEquivalenceClass { groupings: Vec>>, diff --git a/datafusion/physical-expr/src/equivalence/mod.rs b/datafusion/physical-expr/src/equivalence/mod.rs index ff57146efcce9..3724469083451 100644 --- a/datafusion/physical-expr/src/equivalence/mod.rs +++ b/datafusion/physical-expr/src/equivalence/mod.rs @@ -2,6 +2,15 @@ // SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership // SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements +//! Equivalence properties: the orderings, equality, constants and +//! groupings known to hold for the output of a plan node, which the planner +//! uses to avoid sorts and repartitions. +//! +//! # See Also +//! +//! * [`EquivalenceProperties`], which explains the concepts and how the +//! pieces below fit together. + use std::borrow::Borrow; use std::sync::Arc; diff --git a/datafusion/physical-expr/src/equivalence/ordering.rs b/datafusion/physical-expr/src/equivalence/ordering.rs index 7de47767f62fd..05fe1d2c4610e 100644 --- a/datafusion/physical-expr/src/equivalence/ordering.rs +++ b/datafusion/physical-expr/src/equivalence/ordering.rs @@ -15,8 +15,9 @@ use arrow::datatypes::SchemaRef; use datafusion_common::Result; use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr; -/// An `OrderingEquivalenceClass` keeps track of distinct alternative orderings -/// than can describe a table. For example, consider the following table: +/// Tracks distinct alternative orderings than describe a table. +/// +/// For example, consider the following table: /// /// ```text /// ┌───┬───┬───┬───┐ @@ -30,16 +31,25 @@ use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr; /// ``` /// /// 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. +/// ordering, so these orderings are equivalent. See [Equivalent sort +/// expressions] for more details. +/// +/// An `OrderingEquivalenceClass` is a set of such equivalent orderings. +/// +/// [`GroupingEquivalenceClass`] is used when groups of equal values are +/// contiguous, but not in order. /// -/// An `OrderingEquivalenceClass` is a set of such equivalent orderings, which -/// is represented by a vector of `LexOrdering`s. The set does not store any -/// redundant information by enforcing the invariant that no suffix of an -/// ordering in the equivalence class is a prefix of another ordering in the -/// equivalence class. The set can be empty, which means that there are no -/// orderings that describe the table. +/// [Equivalent sort expressions]: crate::equivalence::EquivalenceProperties#equivalent-sort-expressions +/// [`GroupingEquivalenceClass`]: crate::equivalence::GroupingEquivalenceClass #[derive(Clone, Debug, Default, Eq, PartialEq)] pub struct OrderingEquivalenceClass { + /// Set of equivalent orderings + /// + /// Invariant: no suffix of an ordering in the equivalence class is a prefix + /// of another ordering in the equivalence class. This avoids redundant + /// information. + /// + /// If empty, means the table is unordered orderings: Vec, } @@ -216,18 +226,11 @@ impl OrderingEquivalenceClass { /// `ASC NULLS LAST` and `DESC NULLS FIRST`. Rows sorted both ways all have /// the same value (or are all null). /// - /// This is how a value that is constant only within groups of rows is - /// recorded, since a `ConstExpr` is constant across all rows. For example, - /// `SUM(x) OVER (PARTITION BY k)` has one value per `k`, so the window adds - /// `[k, w]` once for each of the four directions of `w`. - /// - /// ### Future Improvements + /// This is how a whole-partition window aggregate records that its value + /// is constant within each window partition. See + /// [Per-partition constants from window functions]. /// - /// This function may become unnecessary if any of the following improvements - /// are implemented: - /// 1. `SortOptions` supports encoding constantness information. - /// 2. `EquivalenceProperties` gains `FunctionalDependency` awareness, eliminating - /// the need for `Constant` and `Constraints`. + /// [Per-partition constants from window functions]: crate::equivalence::EquivalenceProperties#per-partition-constants-from-window-functions pub fn is_expr_partial_const(&self, expr: &Arc) -> bool { is_partial_const(self.iter().map(|ordering| ordering.first()), expr) } diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs index 1206315bbb274..d8efbc5e7b916 100644 --- a/datafusion/physical-expr/src/equivalence/properties/mod.rs +++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs @@ -38,10 +38,12 @@ use datafusion_expr::sort_properties::{ExprProperties, SortProperties}; use indexmap::IndexSet; use itertools::{Either, Itertools}; -/// `EquivalenceProperties` stores information about the output of a plan node -/// that can be used to optimize the plan. Currently, it keeps track of: +/// `EquivalenceProperties` stores information about how values are distributed +/// across the output of a plan node that can be used to optimize the plan. +/// +/// Currently, it tracks: /// - Sort expressions (orderings), -/// - Complete expression tuples whose values are contiguous, +/// - When all rows of equal values are contiguous for certain tuples (groupings) /// - Equivalent expressions; i.e. expressions known to have the same value. /// - Constants expressions; i.e. expressions known to contain a single constant /// value. @@ -50,7 +52,25 @@ use itertools::{Either, Itertools}; /// /// [Using Ordering for Better Plans]: https://datafusion.apache.org/blog/2025/03/11/ordering-analysis/ /// -/// # Example equivalent sort expressions +/// This document explains the concepts and how the pieces fit together. +/// +/// # Overview +/// +/// | Concept | Example | Stored in | +/// |------------------------|-------------------------------------------|---------------------------------------------------------------------| +/// | Equivalent expressions | `a` and `b` after the filter `a = b` | An [`EquivalenceGroup`] of [`EquivalenceClass`]es | +/// | Constants | `a` after the filter `a = 5` | The `constant` of an [`EquivalenceClass`], added as a [`ConstExpr`] | +/// | Orderings | `[a ASC, b DESC]` after a sort | An [`OrderingEquivalenceClass`] | +/// | Groupings | `[a, b]` after a hash aggregate on `a, b` | A [`GroupingEquivalenceClass`] | +/// | Constraints | A primary key or unique constraint | [`Constraints`] | +/// +/// [`EquivalenceClass`]: crate::equivalence::EquivalenceClass +/// [Normal form]: #normal-form +/// [Required orderings]: #required-orderings +/// [`ExecutionPlan::required_input_ordering`]: https://docs.rs/datafusion-physical-plan/latest/datafusion_physical_plan/trait.ExecutionPlan.html#method.required_input_ordering +/// [Checking whether a requirement is satisfied]: #checking-whether-a-requirement-is-satisfied +/// +/// # Equivalent sort expressions /// /// Consider table below: /// @@ -67,11 +87,11 @@ use itertools::{Either, Itertools}; /// /// In this case, both `a ASC` and `b DESC` can describe the table ordering. /// `EquivalenceProperties` tracks these different valid sort expressions and -/// treat `a ASC` and `b DESC` on an equal footing. For example, if the query +/// treats `a ASC` and `b DESC` on an equal footing. For example, if the query /// specifies the output sorted by EITHER `a ASC` or `b DESC`, the sort can be /// avoided. /// -/// # Example equivalent expressions +/// # Equivalent expressions /// /// Similarly, consider the table below: /// @@ -86,11 +106,241 @@ use itertools::{Either, Itertools}; /// └---┴---┘ /// ``` /// -/// In this case, columns `a` and `b` always have the same value. With this -/// information, Datafusion can optimize various operations. For example, if -/// the partition requirement is `Hash(a)` and output partitioning is -/// `Hash(b)`, then DataFusion avoids repartitioning the data as the existing -/// partitioning satisfies the requirement. +/// In this case, columns `a` and `b` always have the same value. DataFusion +/// uses this information to optimize various operations. For example, if the +/// partition requirement is `Hash(a)` and output partitioning is `Hash(b)`, +/// then DataFusion avoids repartitioning the data as the existing partitioning +/// satisfies the requirement. +/// +/// # Constants +/// +/// A constant is an expression with a single value in the output of a plan +/// node. A constant is *uniform* if it has the same value in every output +/// partition, and *heterogeneous* if it has one value within each output +/// partition but may differ between output partitions (see +/// [`AcrossPartitions`]). +/// +/// ## Example: Uniform Constant from a Filter +/// +/// Consider an input sorted by `b ASC` and the filter `a = 5`, which produces +/// the following table: +/// +/// ```text +/// ┌───┬───┐ +/// │ a │ b │ +/// ├───┼───┤ +/// │ 5 │ 1 │ +/// │ 5 │ 2 │ +/// │ 5 │ 4 │ +/// └───┴───┘ +/// ``` +/// +/// The filter adds `a` as a uniform [`ConstExpr`]. If the query requests +/// sorting on `[a ASC, b ASC]` DataFusion knows it can skip sorting. +/// +/// ## Example: Heterogeneous Constant from `UNION ALL` +/// +/// Consider `SELECT 1 AS c, x FROM t1 UNION ALL SELECT 2 AS c, x FROM t2`, +/// where each input is sorted by `x ASC` and has one output partition. The +/// union has two output partitions, one from each input: +/// +/// ```text +/// Partition 0 Partition 1 +/// ┌───┬───┐ ┌───┬───┐ +/// │ c │ x │ │ c │ x │ +/// ├───┼───┤ ├───┼───┤ +/// │ 1 │ 1 │ │ 2 │ 2 │ +/// │ 1 │ 3 │ │ 2 │ 5 │ +/// └───┴───┘ └───┴───┘ +/// ``` +/// +/// Observations: +/// - Within each output partition `c` has one value, but the values differ +/// between output partitions, so `c` is an +/// [`AcrossPartitions::Heterogeneous`] constant. +/// - Each output partition is sorted by `[x ASC]`, and therefore also by +/// `[c ASC, x ASC]`. +/// - However, when the output partitions are merged by `[x ASC]` alone, the +/// result is NOT sorted by `c`, unlike with an [`AcrossPartitions::Uniform`] +/// constant, which would have the same value in every output partition and +/// thus be sorted by `[c ASC, x ASC]` as well. +/// +/// Given those observations, `c` must stay in the ordering `[c ASC, x ASC]` +/// rather than being removed as a uniform constant would be. +/// +/// # Constants within a window partition +/// +/// NOTE: it is important to distinguish between *window partition*, the rows +/// with equal `PARTITION BY` keys, and *output partition*, how the data is +/// divided up across execution units during execution. +/// +/// ## Example: Window Functions with `PARTITION BY` +/// +/// Consider an input sorted by `k ASC` and the window expression +/// `SUM(x) OVER (PARTITION BY k) AS w`, which produces the following rows: +/// +/// ```text +/// ┌───┬───┬────┐ +/// │ k │ x │ w │ +/// ├───┼───┼────┤ +/// │ 1 │ 3 │ 5 │ +/// │ 1 │ 2 │ 5 │ +/// │ 2 │ 4 │ 10 │ +/// │ 2 │ 6 │ 10 │ +/// │ 3 │ 1 │ 1 │ +/// └───┴───┴────┘ +/// ``` +/// +/// Observations: +/// - `w` is not a constant because it differs between window partitions. +/// - The output is sorted by `k` (the window partition key), because +/// DataFusion's window implementation keeps the input order. +/// - Within each run of equal `k`, `w` has one value. +/// +/// Given those observations, the output is sorted by `[k, w]` in *every* +/// direction of `w`: ascending or descending, nulls first or last: +/// +/// - `[k ASC, w ASC NULLS FIRST]` +/// - `[k ASC, w ASC NULLS LAST]` +/// - `[k ASC, w DESC NULLS FIRST]` +/// - `[k ASC, w DESC NULLS LAST]` +/// +/// ## Example: Window Functions without `PARTITION BY` +/// +/// Consider the same input and the window expression `SUM(x) OVER () AS w`, +/// which produces the following rows: +/// +/// ```text +/// ┌───┬───┬────┐ +/// │ k │ x │ w │ +/// ├───┼───┼────┤ +/// │ 1 │ 3 │ 16 │ +/// │ 1 │ 2 │ 16 │ +/// │ 2 │ 4 │ 16 │ +/// │ 2 │ 6 │ 16 │ +/// │ 3 │ 1 │ 16 │ +/// └───┴───┴────┘ +/// ``` +/// +/// Observations: +/// - The frame of every row is the whole table, so `w` has one value. +/// - The output is still sorted by `k`, as the input was. +/// +/// Given those observations, `w` is added as a constant (a [`ConstExpr`]) +/// rather than as orderings. +/// +/// ## Example: Set-Monotonic Window Functions +/// +/// Consider an input sorted by `[ts ASC]` and the window function +/// `MAX(x) OVER (ORDER BY ts) AS m`, which produces the following rows: +/// +/// ```text +/// ┌────┬───┬───┐ +/// │ ts │ x │ m │ +/// ├────┼───┼───┤ +/// │ 1 │ 3 │ 3 │ +/// │ 2 │ 1 │ 3 │ +/// │ 3 │ 5 │ 5 │ +/// │ 4 │ 2 │ 5 │ +/// └────┴───┴───┘ +/// ``` +/// +/// Observations: +/// - When written with explicit defaults, this query is equivalent to +/// `SELECT ts, x, MAX(x) OVER (ORDER BY ts RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS m`, +/// so each row's frame contains every earlier row's frame. +/// - `MAX` is set-monotonically increasing: adding rows to its input can only +/// raise its value or leave it unchanged (see [`SetMonotonicity`]). +/// +/// Given those observations, `m` never decreases as `ts` increases, so the +/// output sorting is described by both `[ts ASC]` and `[m ASC NULLS LAST]`. +/// +/// If the query had a `PARTITION BY k`, with the input sorted by `k` first, +/// the added ordering would be `[k, m ASC NULLS LAST]`. +/// +/// A decreasing aggregate such as `MIN` results in `DESC NULLS LAST`. +/// +/// A frame that shrinks as rows advance, such as `CURRENT ROW` to +/// `UNBOUNDED FOLLOWING`, reverses the direction: `MAX` then results in +/// `DESC NULLS FIRST`. +/// +/// [`SetMonotonicity`]: datafusion_expr::SetMonotonicity +/// +/// # Required orderings +/// +/// Some operators only work on sorted input. A sort merge join needs +/// inputs sorted on the join keys, and a streaming window needs its input +/// sorted by its `PARTITION BY` and then its `ORDER BY` expressions. +/// +/// Each operator describes its requirements via +/// [`ExecutionPlan::required_input_ordering`]. +/// +/// **Orderings** describe how the data is actually distributed. +/// **Requirements** describe what an operator *needs*. The planner compares +/// orderings to requirements with [`Self::ordering_satisfy_requirement`] and +/// inserts or removes Sort operations so that they match (see [Checking whether +/// a requirement is satisfied]). +/// +/// # Normal form +/// +/// Orderings and requirements (see [Required orderings]) are compared in +/// *normal form*: every expression is replaced by the first member of its +/// equivalence class, uniform constants and expressions that repeat an earlier +/// one are removed. +/// +/// For example, if `a` and `b` are equal and `d` is a uniform constant, the +/// requirement `[b ASC, d DESC, c ASC, a ASC]` normalizes to `[a ASC, c ASC]`. +/// +/// # Checking whether a requirement is satisfied +/// +/// The requirement `[a, b]` holds if the rows are sorted by `a`, and the rows +/// that tie on `a` are sorted by `b`. +/// +/// [`Self::ordering_satisfy_requirement`] checks whether a requirement is +/// satisfied by an ordering, and is quite complex. +/// +/// # Discovering orderings +/// +/// `EquivalenceProperties` discovers and accounts for additional orderings that +/// may not be obvious from various additional information such as constants and +/// equality. +/// +/// ## Example: Ordering Implied by a `CAST` +/// +/// Consider a table `t` sorted by `[c ASC, b ASC]` and the query +/// `SELECT a, b, c FROM t WHERE c = CAST(a AS BIGINT)`, which produces the +/// following rows: +/// +/// ```text +/// ┌───┬───┬───┐ +/// │ a │ b │ c │ +/// ├───┼───┼───┤ +/// │ 1 │ 2 │ 1 │ +/// │ 1 │ 5 │ 1 │ +/// │ 2 │ 1 │ 2 │ +/// │ 3 │ 4 │ 3 │ +/// └───┴───┴───┘ +/// ``` +/// +/// Observations: +/// - The filter keeps the input order, so the rows are still sorted by +/// `[c ASC, b ASC]` +/// - Casting an `INT` to a `BIGINT` preserves order and never maps two +/// different values to the same one, so rows with different `a` are in the +/// order of `c`, which is the order of `a`. +/// - Rows with equal `c` have equal `a`, and are also still sorted by `b`. +/// +/// Given those observations, the rows are also sorted by `[a ASC, b ASC]`, +/// even though no operator sorted on `a`. `EquivalenceProperties` derives +/// this ordering when the equality is added, so a later `ORDER BY a, b` +/// needs no re-sorting. +/// +/// This holds in general for a function `f` whose output follows the +/// lexicographical order of its arguments, which `f` declares by setting +/// [`ExprProperties::preserves_lex_ordering`]. +/// +/// [`Self::add_equal_conditions`] and [`Self::add_constants`] look for such +/// orderings whenever they learn something new. /// /// # Code Example /// ``` @@ -107,10 +357,10 @@ use itertools::{Either, Itertools}; /// # let col_a = col("a", &schema).unwrap(); /// # let col_b = col("b", &schema).unwrap(); /// # 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); +/// // Single constant value of b /// eq_properties.add_constants(vec![ConstExpr::from(col_b)]); +/// // ORDER BY a ASC, c DESC /// eq_properties.add_ordering([ /// PhysicalSortExpr::new_default(col_a).asc(), /// PhysicalSortExpr::new_default(col_c).desc(), @@ -364,8 +614,9 @@ impl EquivalenceProperties { self.geq_cache.clear(); } - /// Removes constant expressions that may change across partitions. - /// This method should be used when merging data from different partitions. + /// Removes constant expressions that may change across output partitions. + /// This method should be used when merging data from different output + /// partitions. pub fn clear_per_partition_constants(&mut self) { if self.eq_group.clear_per_partition_constants() { // Renormalize orderings if the equivalence group changes: @@ -754,6 +1005,10 @@ impl EquivalenceProperties { /// tie on `a`. /// 3. The orderings now start at `[b, c]` and `[d]`. /// 4. The function returns `true` as `[d]` satisfies `d`. + /// + /// See [Required orderings] for what a requirement is. + /// + /// [Required orderings]: EquivalenceProperties#required-orderings pub fn ordering_satisfy_requirement( &self, given: impl IntoIterator, @@ -1425,7 +1680,7 @@ impl EquivalenceProperties { /// Returns a `Some` value if the expression is constant according to /// equivalence group, and `None` otherwise. The `Some` variant contains /// an `AcrossPartitions` value indicating whether the expression is - /// constant across partitions, and its actual value (if available). + /// constant across output partitions, and its actual value (if available). pub fn is_expr_constant( &self, expr: &Arc, diff --git a/datafusion/physical-expr/src/equivalence/properties/walk.rs b/datafusion/physical-expr/src/equivalence/properties/walk.rs index e0487cdd23dfa..5b8b602626cbc 100644 --- a/datafusion/physical-expr/src/equivalence/properties/walk.rs +++ b/datafusion/physical-expr/src/equivalence/properties/walk.rs @@ -22,29 +22,127 @@ use itertools::Itertools; /// The state of one walk along a sequence of sort keys, such as the keys of a /// required ordering. /// -/// A sort key only matters among the rows that tie on the keys before it. So -/// once a key is satisfied, the walk *pins* it: for the rest of the walk, the -/// key has one value, like a constant. An expression is *fixed* if it has one -/// value among those rows: it is a literal, a constant, pinned, a leaf that -/// reports itself as a singleton (such as an uncorrelated scalar subquery), or -/// not volatile and computed only from fixed expressions. The *head* of an -/// ordering is its first expression that is not fixed; an ordering whose -/// expressions are all fixed has no head. With orderings `[a, b]` and -/// `[a, c]`, the requirement `[a, c]` is satisfied, because pinning `a` makes -/// `c` the head of the second ordering. +/// [`EquivalenceProperties`] uses a `SortKeyWalk` to decide if a set of +/// orderings satisfy a requirement. See [Checking whether a requirement is +/// satisfied] for more details. /// -/// Pins live in the walk; the equivalence properties it reads do not change. +/// # Intuition and Introduction, and Definitions +/// +/// An ordering `[x, y, z]` is a chain of claims about the rows: they are sorted +/// by x; among rows with equal x, sorted by y; among rows with equal x and y, +/// sorted by z. +/// +/// ## Pinning and Fixed Expressions +/// +/// During a walk, the code scans the ordering requirement left to right. When +/// it finds key `k` satisfied, it *pins* `k`, meaning that for the next steps +/// in the walk, `k` is effectively a constant across the remaining ordering +/// requirement. +/// +/// An expression is *fixed* if it has one value for all rows that tie on the +/// currently pinned expressions. Fixed expressions the pinned expressions +/// themselves, as well as literals, constants, and singleton leaf expressions +/// (such as an uncorrelated scalar subquery). See [`Self::is_fixed`] for more +/// details. +/// +/// ## Heads +/// +/// For each equivalent ordering, the walk tracks its current *head* expression, +/// which is the current claim that is valid given the currently pinned +/// expressions. When an expression is pinned, it advances the head of any +/// existing claim about that key. An ordering with nothing left to promise is +/// used up. +/// +/// For example, if we have data ordered by `[a, b]` as well as `[c]`, and we +/// walk the requirement `[a, c]`, pinning `a` after the first step results in: +/// +/// | Step | Requirement Key | Pinned | Head `[a, b]` | Head `[c]` | Outcome | +/// |------|-----------------|--------|---------------|------------|-----------| +/// | 1 | a | {} | a | c | pin a | +/// | 2 | c | {a} | b | c | satisfied | +/// +/// It may be confusing that the head of `[c]` does not advance. The intuition +/// is that `[c]` still promises the data is sorted by `c` for all equal values +/// of `a`. This is true because the data was entirely sorted by `[c]` to begin +/// with, and pinning `a` does not change the order of the rows that tie on `a`, +/// as illustrated in this example: +/// +/// ```text +/// ┌───┬───┬───┐ +/// │ a │ b │ c │ +/// ├───┼───┼───┤ +/// │ 1 │ 1 │ 2 │ +/// │ 1 │ 3 │ 5 │ +/// │ 2 │ 1 │ 7 │ +/// │ 2 │ 2 │ 9 │ +/// └───┴───┴───┘ +/// ``` +/// +/// # Example: Walking a Requirement +/// +/// * Orderings: `[a ASC, b ASC, c ASC]` and `[d ASC]` +/// * Requirement: `[a ASC, b ASC]` +/// +/// | Step | Requirement Key | Pinned | Head `[a, b, c]` | Head `[d]` | Outcome | +/// |------|-----------------|--------|------------------|------------|-----------| +/// | 1 | a | {} | a | d | pin a | +/// | 2 | b | {a} | b | d | satisfied | +/// +/// This ordering is satisfied, (intuitively since the requirement is a prefix +/// of the first ordering). The second ordering plays no part. +/// +/// * Orderings: `[a ASC, b ASC]` and `[b ASC, c ASC]` +/// * Requirement: `[a ASC, b ASC, c ASC]` +/// +/// | Step | Requirement Key | Pinned | Head `[a, b]` | Head `[b, c]` | Outcome | +/// |------|-----------------|--------|---------------|---------------|-----------| +/// | 1 | a | {} | a | b | pin a | +/// | 2 | b | {a} | b | b | pin b | +/// | 3 | c | {a, b} | - | c | satisfied | +/// +/// Note that After step 2, `[a, b]` is used up as both of its expressions are +/// pinned. Among the rows that tie on `a` and `b`, the second ordering says the +/// rows are sorted by `c`, which can satisfy the third key. +/// +/// * Orderings: `[a ASC, c ASC]` and `[c ASC, b ASC]` +/// * Requirement: `[a ASC, b ASC, c ASC]` +/// +/// | Step | Requirement Key | Pinned | Head `[a, c]` | Head `[c, b]` | Outcome | +/// |------|-----------------|--------|---------------|---------------|-------------| +/// | 1 | a | {} | a | c | pin a | +/// | 2 | b | {a} | c | c | unsatisfied | +/// +/// After pinning `a`, both orderings are headed by `c`. Since nothing says the +/// rows that tie on `a` are sorted by `b`, the walk stops, and reports that the +/// requirement is not satisfied. +/// +/// * Orderings: `[a ASC, b ASC]` and `[a ASC, c ASC]` +/// * Requirement: `[a ASC, c ASC]` +/// +/// | Step | Requirement Key | Pinned | Head `[a, b]` | Head `[a, c]` | Outcome | +/// |------|-----------------|--------|---------------|---------------|-----------| +/// | 1 | a | {} | a | a | pin a | +/// | 2 | c | {a} | b | c | satisfied | +/// +/// [Checking whether a requirement is satisfied]: EquivalenceProperties#checking-whether-a-requirement-is-satisfied pub(super) struct SortKeyWalk<'a> { props: &'a EquivalenceProperties, - /// Pinned expressions. Pinning a member of an equivalence class pins all - /// of its members. + /// Pinned expressions. pinned: HashSet>, /// Orderings that hold among the rows that tie on the pinned expressions, /// found by [`EquivalenceProperties::implied_ordering`] as the walk goes. implied: Vec, - /// For each ordering (the normal orderings of `props`, then `implied`), - /// the index of its head. Empty until the first pin, so that a walk with - /// nothing pinned allocates nothing; until then every head is at index 0. + /// The position of the head of each ordering. Empty until the first pin. + /// + /// `head_idxs[i]` tracks the position in the `LexOrdering` at index `i`, + /// as returned by [`Self::ordering`]. + /// + /// The ordering at index `i` is: + /// * if `i < self.props.normal_cls.len()` then `self.props.normal_cls[i]` + /// * else, `self.implied[i - self.props.normal_cls.len()]` otherwise. See + /// + /// The value is either an index into that ordering's sort expressions, or + /// its length when the ordering has been exhausted. head_idxs: Vec, } @@ -132,10 +230,13 @@ impl<'a> SortKeyWalk<'a> { Ok(()) } - /// Pins `expr` and the rest of its equivalence class (but, unlike - /// [`Self::pin_key`], not its children), then moves the head of each - /// ordering past the expressions that are now fixed, and adds the - /// orderings that the heads imply, until no new ordering is found. + /// Pins `expr`. + /// + /// See [`SortKeyWalk`] for an explanation of pinning. + /// + /// + /// This method pins `expr` and thus by extension all members of its + /// equivalence class), but not its children. See [`Self::pin_key`] for more details. pub(super) fn pin(&mut self, expr: Arc) -> Result<()> { match self.props.eq_group.get_equivalence_class(&expr) { Some(class) => self.pinned.extend(class.iter().cloned()), @@ -250,11 +351,12 @@ impl<'a> SortKeyWalk<'a> { // strict: node.data.strictly_order_preserving = true; } - // Now, check what we know about orderings. A whole-partition window - // such as `SUM(x) OVER (PARTITION BY k)` adds `[k, w]` in all four - // directions of `w` (see `OrderingEquivalenceClass::is_expr_partial_const`). - // Once `k` is pinned, `w` heads orderings in opposite directions, so it - // has one value among the rows that tie on `k`: + // Now, check what we know about orderings. A window aggregate over a + // whole window partition, such as `SUM(x) OVER (PARTITION BY k)`, adds + // `[k, w]` in all four directions of `w` (see `EquivalenceProperties` + // documentation for more details). Once `k` is pinned, `w` heads + // orderings in opposite directions, so it has one value among the rows + // that tie on `k`: let normal_expr = self.props.eq_group.normalize_expr(Arc::clone(&node.expr)); if self.is_fixed(&normal_expr) || is_partial_const(self.heads(), &normal_expr) { node.data.sort_properties = SortProperties::Singleton; @@ -264,10 +366,13 @@ impl<'a> SortKeyWalk<'a> { Ok(Transformed::yes(node)) } - /// Whether `expr` is fixed (see [`SortKeyWalk`]). Callers pass `expr` in - /// normal form. Pins are stored as given, so a pinned expression that is - /// not in normal form is found only through its equivalence class, whose - /// members are all pinned. + /// Whether `expr` is fixed + /// + /// Note `expr` must be in [normal form]. + /// + /// See [`SortKeyWalk`] for more details + /// + /// [normal form]: EquivalenceProperties#normal-form fn is_fixed(&self, expr: &Arc) -> bool { if expr.is::() || self.pinned.contains(expr) { return true; @@ -427,8 +532,10 @@ mod tests { Ok(()) } - /// A window over a whole partition, such as `SUM(..) OVER (PARTITION BY - /// k)`, adds `[k, x]` in all four directions of `x`. Pinning `k` leaves `x` + /// A window over a whole window partition, such as `SUM(..) OVER + /// (PARTITION BY k)`, adds `[k, x]` in all four directions of `x`. + /// + /// Pinning `k` leaves `x` /// heading orderings in opposite directions, so `x` has one value, and so /// does `CAST(x AS BIGINT)`: `[k, CAST(x AS BIGINT)]` holds in every /// direction. Without the partial constant check, `x` would only be diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs index b236cc2f7ce00..fce0a8391b173 100644 --- a/datafusion/physical-plan/src/execution_plan.rs +++ b/datafusion/physical-plan/src/execution_plan.rs @@ -205,6 +205,11 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync { /// NOTE that checking `!is_empty()` does **not** check for a /// required input ordering. Instead, the correct check is that at /// least one entry must be `Some` + /// + /// See [Required orderings] for how the planner checks these requirements + /// against the orderings a child provides. + /// + /// [Required orderings]: EquivalenceProperties#required-orderings fn required_input_ordering(&self) -> Vec> { vec![None; self.children().len()] } diff --git a/datafusion/physical-plan/src/windows/mod.rs b/datafusion/physical-plan/src/windows/mod.rs index 1e5eea0b01285..dc792984e4f6a 100644 --- a/datafusion/physical-plan/src/windows/mod.rs +++ b/datafusion/physical-plan/src/windows/mod.rs @@ -364,6 +364,12 @@ pub(crate) fn get_partition_by_sort_exprs( } } +/// Computes the [`EquivalenceProperties`] of a window operator's output. +/// +/// See [Constants within a window partition] for what this computes along with +/// examples. +/// +/// [Constants within a window partition]: EquivalenceProperties#constants-within-a-window-partition pub(crate) fn window_equivalence_properties( schema: &SchemaRef, input: &Arc,