alamb commented on code in PR #27250:
URL: https://github.com/apache/datafusion/pull/27250#discussion_r4514372078
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -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.
Review Comment:
Thank you - I verified your fix:
```
mkdir -p /tmp/cast_order && cd /tmp/cast_order
cat > t.csv <<'EOF'
a,b,c
1,2,1
1,5,1
2,1,2
3,4,3
EOF
```
```sql
DataFusion CLI v57.0.0-dev
> -- Case 1: table sorted by [c ASC, b ASC] (as written in the PR): SortExec
remains
CREATE EXTERNAL TABLE t_cb (a INT NOT NULL, b INT NOT NULL, c BIGINT NOT
NULL)
STORED AS CSV
WITH ORDER (c ASC, b ASC)
LOCATION '/tmp/cast_order/t.csv'
OPTIONS ('format.has_header' 'true');
EXPLAIN SELECT a, b, c FROM t_cb WHERE c = CAST(a AS BIGINT) ORDER BY a, b;
0 row(s) fetched.
Elapsed 0.021 seconds.
+---------------+-------------------------------+
| plan_type | plan |
+---------------+-------------------------------+
| physical_plan | ┌───────────────────────────┐ |
| | │ SortPreservingMergeExec │ |
| | │ -------------------- │ |
| | │ a ASC NULLS LAST, b ASC │ |
| | │ NULLS LAST │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ SortExec │ |
| | │ -------------------- │ |
| | │ a@0 ASC NULLS LAST, b@1 │ |
| | │ ASC NULLS LAST │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ FilterExec │ |
| | │ -------------------- │ |
| | │ predicate: │ |
| | │ c = CAST(a AS Int64) │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ RepartitionExec │ |
| | │ -------------------- │ |
| | │ partition_count(in->out): │ |
| | │ 1 -> 16 │ |
| | │ │ |
| | │ partitioning_scheme: │ |
| | │ RoundRobinBatch(16) │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ DataSourceExec │ |
| | │ -------------------- │ |
| | │ files: 1 │ |
| | │ format: csv │ |
| | └───────────────────────────┘ |
| | |
+---------------+-------------------------------+
1 row(s) fetched.
```
Here is your suggesetion (which does the right thing 👍 )
```sql
-- Case 2: table sorted by [c ASC, a ASC, b ASC] (jayzhan211's suggestion):
SortExec removed
CREATE EXTERNAL TABLE t_cab (a INT NOT NULL, b INT NOT NULL, c BIGINT NOT
NULL)
STORED AS CSV
WITH ORDER (c ASC, a ASC, b ASC)
LOCATION '/tmp/cast_order/t.csv'
OPTIONS ('format.has_header' 'true');
EXPLAIN SELECT a, b, c FROM t_cab WHERE c = CAST(a AS BIGINT) ORDER BY a, b;
SELECT a, b, c FROM t_cab WHERE c = CAST(a AS BIGINT) ORDER BY a, b;
```
```sql
+---------------+-------------------------------+
| plan_type | plan |
+---------------+-------------------------------+
| physical_plan | ┌───────────────────────────┐ |
| | │ SortPreservingMergeExec │ |
| | │ -------------------- │ |
| | │ a ASC NULLS LAST, b ASC │ |
| | │ NULLS LAST │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ FilterExec │ |
| | │ -------------------- │ |
| | │ predicate: │ |
| | │ c = CAST(a AS Int64) │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ RepartitionExec │ |
| | │ -------------------- │ |
| | │ partition_count(in->out): │ |
| | │ 1 -> 16 │ |
| | │ │ |
| | │ partitioning_scheme: │ |
| | │ RoundRobinBatch(16) │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ DataSourceExec │ |
| | │ -------------------- │ |
| | │ files: 1 │ |
| | │ format: csv │ |
| | └───────────────────────────┘ |
| | |
+---------------+-------------------------------+
1 row(s) fetched.
Elapsed 0.001 seconds.
+---+---+---+
| a | b | c |
+---+---+---+
| 1 | 2 | 1 |
| 1 | 5 | 1 |
| 2 | 1 | 2 |
| 3 | 4 | 3 |
+---+---+---+
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
