+ Ordering Requirement for an operator describes how the input data to that operator must be sorted for the operator to compute the correct result. It is the job of the planner to make sure that these requirements are satisfied during execution (See DataFusion EnforceSorting for an implementation of such a rule). ++ +There are various use cases where this type of analysis can be useful such as the following examples. +### Removing Unnecessary Sorts +Imagine a user wants to execute the following query: +```SQL +SELECT hostname, log_line +FROM telemetry ORDER BY time ASC limit 10 +``` +If we don't know anything about the `telemetry` table we need to sort it by `time ASC` and then retrieve the first 10 rows to get the correct result. However, if the table is already ordered by `time ASC`, we can simply retrieve the first 10 rows. This approach executes much faster and uses less memory compared to resorting the entire table, even when the [TopK] operator is used. + +[TopK]: https://docs.rs/datafusion/latest/datafusion/physical_plan/struct.TopK.html + +In order to avoid the sort the query optimizer must determine the data is already sorted. For simple queries the analysis is straightforward however it gets complicated fast. For example, what if your data is sorted by `[hostname, time ASC]` and your query is +```sql +SELECT hostname, log_line +FROM telemetry WHERE hostname = 'app.example.com' ORDER BY time ASC; +``` +In this case a sort still isn't needed, but the analysis must reason about the sortedness of the stream when it knows `hostname` has a single value. + +### Optimized Operator Implementations +As another use case, some operators can utilize the ordering information to change its underlying algorithm to execute more efficiently. Consider the following query: +```SQL +SELECT COUNT(log_line) +FROM telemetry GROUP BY hostname; +``` +Most analytic systems, including DataFusion, by default implement such a query using a hash table keyed on values of `hostname` to store the counts. However, if the `telemetry` table is sorted by `hostname`, there are much more efficient algorithms for grouping on `hostname` values than hashing every value and storing it in memory. However, the more efficient algorithm can only be used when the input is sorted correctly. To see this in practice, check out the [source](https://github.com/apache/datafusion/tree/main/datafusion/physical-plan/src/aggregates/order) for ordered variant of the `Aggregation` in `DataFusion`. + +### Streaming-Friendly Execution + +Stream processing aims to produce results immediately as they become available ensuring minimal latency for real-time workloads. However, some operators need to consume all input data before producing any output. Consider the `Sort` operation: before it can start generating output, the algorithm must first process all input data. As a result, data flow halts whenever such an operator is encountered until all input is consumed. When a physical query plan contains such an operator (`Sort`, `CrossJoin`, ..) we refer to this as pipeline breaking, meaning the query cannot be executed in a streaming fashion. + +For a query to be executed in a streaming fashion we need to satisfy 2 conditions: + +**Logically Streamable** +It should be possible to generate what user wants in streaming fashion. Consider following query: + +```SQL +SELECT SUM(amount) +FROM orders +``` +Here, the user wants to compute the sum of all amounts in the orders table. By the nature of the query this requires scanning the entire table to generate a result making it impossible to execute in a streaming fashion. + +**Streaming Aware Planner** +Being logically streamable does not guarantee that a query will execute in a streaming fashion. SQL is a declarative language, meaning it specifies 'WHAT' user wants. It is up to the planner 'HOW' to generate the result. In most cases there are many ways to compute the correct result for a given query. The query planner is responsible for choosing "a way" (ideally the best[*](#optimal) one) among the all alternatives to generate what user asks for. If a plan contains a pipeline-breaking operator the execution will not be streaming—even if the query is logically streamable. To generate truly streaming plans from logically streamable queries the planner must carefully analyze the existing orderings in the source tables to ensure that the final plan does not contain any pipeline-breaking operators. + + +## Analysis +Let's start by creating an example table that we will refer throughout the post. This table models the input data of an operator for the analysis: + +### Example Virtual Table + + + +
| amount | price | hostname | currency | time_bin | time | price_cloned | time_cloned | +
|---|---|---|---|---|---|---|---|
| 12 | 25 | app.example.com | USD | 08:00:00 | 08:01:30 | 25 | 08:01:30 | +
| 12 | 26 | app.example.com | USD | 08:00:00 | 08:11:30 | 26 | 08:11:30 | +
| 15 | 30 | app.example.com | USD | 08:00:00 | 08:41:30 | 30 | 08:41:30 | +
| 15 | 32 | app.example.com | USD | 08:00:00 | 08:55:15 | 32 | 08:55:15 | +
| 15 | 35 | app.example.com | USD | 09:00:00 | 09:10:23 | 35 | 09:10:23 | +
| 20 | 18 | app.example.com | USD | 09:00:00 | 09:20:33 | 18 | 09:20:33 | +
| 20 | 22 | app.example.com | USD | 09:00:00 | 09:40:15 | 22 | 09:40:15 | +
+How can a table have multiple orderings?: At first glance it may seem counterintuitive for a table to have more than one valid ordering. However, during query execution such scenarios can arise. + +For example consider the following query: + +```sql +SELECT time, date_bin('1 hour', time, '1970-01-01') as time_bin +FROM table; +``` +If we know that the table is ordered by+ +By inspection, you can see this table is sorted by the `amount` column, but It is also sorted by `time` and `time_bin` as well as the compound `(time_bin, amount)` and many other variations. While this example is an extreme case, real-world data often has multiple sort orders. + +A naive approach for analyzing whether the ordering requirement of an operator is satisfied by its input would be: + + - Store all the valid ordering expressions that the tables satisfies + - Check whether the ordering requirement by the operator is among valid orderings. + +This naive algorithm works and correct. However, listing all valid orderings can be quite lengthy and is of exponential complexity as the number of orderings grows. For the example table here is a (small) subset of the valid orderings: + +`[amount ASC]` +`[amount ASC, price ASC]` +`[amount ASC, price_cloned ASC]` +`[hostname ASC, amount ASC, price_cloned ASC]` +`[amount ASC, hostname ASC, price_cloned ASC]` +`[amount ASC, price_cloned ASC, hostname ASC]` +. +. +. + +As can be seen from the listing above storing all valid orderings is wasteful and contains significant redundancy. Here are some observations which suggest that we can do much better: + + +- Storing a prefix of another valid ordering is redundant. If the table satisfies the lexicographic ordering[1](#footnote1): `[amount ASC, price ASC]`, it already satisfies ordering `[amount ASC]` trivially. Hence, once we store `[amount ASC, price ASC]` storing `[amount ASC]` is redundant. + +- Using all columns that are equal to each other in the listings is redundant. If we know the table is ordered by `[amount ASC, price ASC]`, it is also ordered by `[amount ASC, price_cloned ASC]` since `price` and `price_cloned` are copy of each other. It is enough to use just one expression among the expressions that exact copy of each other. + +- Constant expressions can be inserted anywhere in a valid ordering with an arbitrary direction (e.g. `ASC`, `DESC`). Hence, if the table is ordered by `[amount ASC, price ASC]`, it is also ordered by:time ASCwe can infer thattime_bin ASCis also a valid ordering. This is because thedate_binfunction is monotonic, meaning it preserves the order of its input. + +DataFusion leverages these functional dependencies to infer new orderings as data flows through different query operators. For details on the implementation see the source code. +
+ Note: These properties are implemented in the+ +These properties allow us to analyze whether the ordering requirement is satisfied by the data already. + +### 1. Constant Expressions +Constant expressions are those where each row in the expression has the same value across all rows. Although constant expressions may seem odd in a table they can arise after operations like `Filter` or `Join` occur. + +For instance in the example table: + +- Columns `hostname` and `currency` are constant because every row in the table has the same value (`'app.example.com'` for `hostname`, and `'USD'` for `currency`) for these columns. + +EquivalencePropertiesstructure inDataFusion, please see the source for more details
+
+ Note: Constant expressions can arise during query execution. For example, in following query:+ +### 2. Equivalent Expression Groups +Equivalent expression groups are expressions that always hold the same value across rows. These expressions can be thought of as clones of each other and may arise from operations like `Filter`, `Join`, or `Projection`. + +In the example table, the expressions `price` and `price_cloned` form one equivalence group, and `time` and `time_cloned` form another equivalence group. + +
+SELECT hostname FROM logsWHERE hostname='app.example.com'
+ after filtering is done, for subsequent operators thehostnamecolumn will be constant. +
+ Note: Equivalent expression groups can arise during the query execution. For example, in the following query:+ +### 3. Succinct Encoding of Valid Orderings +Valid orderings are the orderings that the table already satisfies. However, naively listing them requires exponential space as the number of columns grows as discussed before. Instead, we list all valid orderings after following constraints are applied: + +- Do not use any constant expressions in the valid ordering construction +- Use only one entry (by convention the first entry) in the equivalent expression group. +- Lexicographic ordering shouldn't contain any leading ordering[2](#footnote2)except the first position [3](#footnote3). +- Do not use any prefix of a valid lexicographic ordering[4](#footnote4). + +After applying the first and second constraint, the example table simplifies to + + + +
+SELECT time, time as time_cloned FROM logs
+ after the projection is done, for subsequent operatorstimeandtime_clonedwill form an equivalence group. As another example, in the following query:
+SELECT employees.id, employees.name, departments.department_name+FROM employees+JOIN departments ON employees.department_id = departments.id;
+after joining,employees.department_idanddepartments.idwill form an equivalence group. +
| amount | price | time_bin | time | +
|---|---|---|---|
| 12 | 25 | 08:00:00 | 08:01:30 | +
| 12 | 26 | 08:00:00 | 08:11:30 | +
| 15 | 30 | 08:00:00 | 08:41:30 | +
| 15 | 32 | 08:00:00 | 08:55:15 | +
| 15 | 35 | 09:00:00 | 09:10:23 | +
| 20 | 18 | 09:00:00 | 09:20:33 | +
| 20 | 22 | 09:00:00 | 09:40:15 | +
[1]Lexicographic order is a way of ordering sequences (like strings, list of expressions) based on the order of their components, similar to how words are ordered in a dictionary. It compares each element of the sequences one by one, from left to right.
+[2]Leading ordering is the first ordering in a lexicographic ordering list. As an example, for the ordering: [amount ASC, price ASC], leading ordering will be: amount ASC.
[3]This means that, if we know that [amount ASC] and [time ASC] are both valid orderings for the table. We shouldn't enlist [amount ASC, time ASC] or [time ASC, amount ASC] as valid orderings. These orderings can be deduced if we know that table satisfies the ordering [amount ASC] and [time ASC].
[4]This means that, if ordering [amount ASC, price ASC] is a valid ordering for the table. We shouldn't enlist [amount ASC] as valid ordering. Validity of it can be deduced from the ordering: [amount ASC, price ASC]
+
[5]Leading ordering requirement is the first ordering requirement in the list of lexicographic ordering requirement expression. As an example for the requirement: [amount ASC, time_bin ASC, prices ASC, time ASC], leading ordering requirement is: amount ASC.
[6]Leading valid orderings are the first ordering for each valid ordering list in the table. As an example, for the valid orderings: [amount ASC, prices ASC], [time_bin ASC], [time ASC], leading valid orderings will be: amount ASC, time_bin ASC, time ASC.
*Best depends on the use case, DataFusion has many various flags to communicate what user thinks the best plan is (e.g. streamable, fastest, lowest memory, etc.). See configurations for detail.