| <!--- | |
| Licensed to the Apache Software Foundation (ASF) under one | |
| or more contributor license agreements. See the NOTICE file | |
| distributed with this work for additional information | |
| regarding copyright ownership. The ASF licenses this file | |
| to you under the Apache License, Version 2.0 (the | |
| "License"); you may not use this file except in compliance | |
| with the License. You may obtain a copy of the License at | |
| http://www.apache.org/licenses/LICENSE-2.0 | |
| Unless required by applicable law or agreed to in writing, | |
| software distributed under the License is distributed on an | |
| "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| KIND, either express or implied. See the License for the | |
| specific language governing permissions and limitations | |
| under the License. | |
| --> | |
| # Reading Explain Plans | |
| ## Introduction | |
| This section describes of how to read a DataFusion query plan. While fully | |
| comprehending all details of these plans requires significant expertise in the | |
| DataFusion engine, this guide will help you get started with the basics. | |
| Datafusion executes queries using a `query plan`. To see the plan without | |
| running the query, add the keyword `EXPLAIN` to your SQL query or call the | |
| [DataFrame::explain] method | |
| [dataframe::explain]: https://docs.rs/datafusion/latest/datafusion/dataframe/struct.DataFrame.html#method.explain | |
| ## Example: Select and filter | |
| In this section, we run example queries against the `hits.parquet` file. See | |
| [below](#data-in-this-example)) for information on how to get this file. | |
| Let's see how DataFusion runs a query that selects the top 5 watch lists for the | |
| site `http://domcheloveplanet.ru/`: | |
| ```sql | |
| EXPLAIN FORMAT INDENT SELECT "WatchID" AS wid, "hits.parquet"."ClientIP" AS ip | |
| FROM 'hits.parquet' | |
| WHERE starts_with("URL", 'http://domcheloveplanet.ru/') | |
| ORDER BY wid ASC, ip DESC | |
| LIMIT 5; | |
| ``` | |
| The output will look like | |
| ```text | |
| +---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| | plan_type | plan | | |
| +---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| | logical_plan | Sort: wid ASC NULLS LAST, ip DESC NULLS FIRST, fetch=5 | | |
| | | Projection: hits.parquet.WatchID AS wid, hits.parquet.ClientIP AS ip | | |
| | | Filter: starts_with(hits.parquet.URL, Utf8("http://domcheloveplanet.ru/")) | | |
| | | TableScan: hits.parquet projection=[WatchID, ClientIP, URL], partial_filters=[starts_with(hits.parquet.URL, Utf8("http://domcheloveplanet.ru/"))] | | |
| | physical_plan | SortPreservingMergeExec: [wid@0 ASC NULLS LAST,ip@1 DESC], fetch=5 | | |
| | | SortExec: TopK(fetch=5), expr=[wid@0 ASC NULLS LAST,ip@1 DESC], preserve_partitioning=[true] | | |
| | | ProjectionExec: expr=[WatchID@0 as wid, ClientIP@1 as ip] | | |
| | | CoalesceBatchesExec: target_batch_size=8192 | | |
| | | FilterExec: starts_with(URL@2, http://domcheloveplanet.ru/) | | |
| | | DataSourceExec: file_groups={16 groups: [[hits.parquet:0..923748528], ...]}, projection=[WatchID, ClientIP, URL], predicate=starts_with(URL@13, http://domcheloveplanet.ru/), file_type=parquet | | |
| +---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| 2 row(s) fetched. | |
| Elapsed 0.060 seconds. | |
| ``` | |
| There are two sections: logical plan and physical plan | |
| - **Logical Plan:** is a plan generated for a specific SQL query, DataFrame, or other language without the | |
| knowledge of the underlying data organization. | |
| - **Physical Plan:** is a plan generated from a logical plan along with consideration of the hardware | |
| configuration (e.g number of CPUs) and the underlying data organization (e.g number of files). | |
| This physical plan is specific to your hardware configuration and your data. If you load the same | |
| data to different hardware with different configurations, the same query may generate different query plans. | |
| Understanding a query plan can help to you understand its performance. For example, when the plan shows your query reads | |
| many files, it signals you to either add more filter in the query to read less data or to modify your file | |
| design to make fewer but larger files. This document focuses on how to read a query plan. How to make a | |
| query run faster depends on the reason it is slow and beyond the scope of this document. | |
| ## Query plans are trees | |
| A query plan is an upside down tree, and we always read from bottom up. The | |
| physical plan in Figure 1 in tree format will look like | |
| ```text | |
| ▲ | |
| │ | |
| │ | |
| ┌─────────────────────────────────────────────────┐ | |
| │ SortPreservingMergeExec │ | |
| │ [wid@0 ASC NULLS LAST,ip@1 DESC] │ | |
| │ fetch=5 │ | |
| └─────────────────────────────────────────────────┘ | |
| ▲ | |
| │ | |
| ┌─────────────────────────────────────────────────┐ | |
| │ SortExec TopK(fetch=5), │ | |
| │ expr=[wid@0 ASC NULLS LAST,ip@1 DESC], │ | |
| │ preserve_partitioning=[true] │ | |
| └─────────────────────────────────────────────────┘ | |
| ▲ | |
| │ | |
| ┌─────────────────────────────────────────────────┐ | |
| │ ProjectionExec │ | |
| │ expr=[WatchID@0 as wid, ClientIP@1 as ip] │ | |
| └─────────────────────────────────────────────────┘ | |
| ▲ | |
| │ | |
| ┌─────────────────────────────────────────────────┐ | |
| │ CoalesceBatchesExec │ | |
| └─────────────────────────────────────────────────┘ | |
| ▲ | |
| │ | |
| ┌─────────────────────────────────────────────────┐ | |
| │ FilterExec │ | |
| │ starts_with(URL@2, http://domcheloveplanet.ru/) │ | |
| └─────────────────────────────────────────────────┘ | |
| ▲ | |
| │ | |
| ┌────────────────────────────────────────────────┐ | |
| │ DataSourceExec │ | |
| │ hits.parquet (filter = ...) │ | |
| └────────────────────────────────────────────────┘ | |
| ``` | |
| Each node in the tree/plan ends with `Exec` and is sometimes also called an `operator` or `ExecutionPlan` where data is | |
| processed, transformed and sent up. | |
| 1. First, data in parquet the `hits.parquet` file is read in parallel using 16 cores in 16 "partitions" (more on this later) from `DataSourceExec`, which applies a first pass at filtering during the scan. | |
| 2. Next, the output is filtered using `FilterExec` to ensure only rows where `starts_with(URL, 'http://domcheloveplanet.ru/')` evaluates to true are passed on | |
| 3. The `CoalesceBatchesExec` then ensures that the data is grouped into larger batches for processing | |
| 4. The `ProjectionExec` then projects the data to rename the `WatchID` and `ClientIP` columns to `wid` and `ip` respectively. | |
| 5. The `SortExec` then sorts the data by `wid ASC, ip DESC`. The `Topk(fetch=5)` indicates that a special implementation is used that only tracks and emits the top 5 values in each partition. | |
| 6. Finally the `SortPreservingMergeExec` merges the sorted data from all partitions and returns the top 5 rows overall. | |
| ## Understanding large query plans | |
| A large query plan may look intimidating, but you can quickly understand what it does by following these steps | |
| 1. As always, read from bottom up, one operator at a time. | |
| 2. Understand the job of this operator by reading | |
| the [Physical Plan documentation](https://docs.rs/datafusion/latest/datafusion/physical_plan/index.html). | |
| 3. Understand the input data of the operator and how large/small it may be. | |
| 4. Understand how much data that operator produces and what it would look like. | |
| If you can answer those questions, you will be able to estimate how much work | |
| that plan has to do and thus how long it will take. However, the `EXPLAIN` just | |
| shows you the plan without executing it. | |
| If you want to know more about how much work each operator in query plan does, | |
| you can use the `EXPLAIN ANALYZE` to get the explain with runtime added (see | |
| next section) | |
| ## More Debugging Information: `EXPLAIN VERBOSE` | |
| If the plan has to read too many files, not all of them will be shown in the | |
| `EXPLAIN`. To see them, use `EXPLAIN VERBOSE`. Like `EXPLAIN`, `EXPLAIN VERBOSE` | |
| does not run the query. Instead it shows the full explain plan, with information | |
| that is omitted from the default explain, as well as all intermediate physical | |
| plans DataFusion generates before returning. This mode can be very helpful for | |
| debugging to see why and when DataFusion added and removed operators from a plan. | |
| ## Execution Counters: `EXPLAIN ANALYZE` | |
| During execution, DataFusion operators collect detailed metrics. You can access | |
| them programmatically via [`ExecutionPlan::metrics`] as well as with the | |
| `EXPLAIN ANALYZE` command. For example here is the same query as | |
| above but with `EXPLAIN ANALYZE` (note the output is edited for clarity) | |
| [`executionplan::metrics`]: https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html#method.metrics | |
| ```sql | |
| > EXPLAIN ANALYZE SELECT "WatchID" AS wid, "hits.parquet"."ClientIP" AS ip | |
| FROM 'hits.parquet' | |
| WHERE starts_with("URL", 'http://domcheloveplanet.ru/') | |
| ORDER BY wid ASC, ip DESC | |
| LIMIT 5; | |
| +-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| | plan_type | plan | | |
| +-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| | Plan with Metrics | SortPreservingMergeExec: [wid@0 ASC NULLS LAST,ip@1 DESC], fetch=5, metrics=[output_rows=5, elapsed_compute=2.375µs] | | |
| | | SortExec: TopK(fetch=5), expr=[wid@0 ASC NULLS LAST,ip@1 DESC], preserve_partitioning=[true], metrics=[output_rows=75, elapsed_compute=7.243038ms, row_replacements=482] | | |
| | | ProjectionExec: expr=[WatchID@0 as wid, ClientIP@1 as ip], metrics=[output_rows=811821, elapsed_compute=66.25µs] | | |
| | | FilterExec: starts_with(URL@2, http://domcheloveplanet.ru/), metrics=[output_rows=811821, elapsed_compute=1.36923816s] | | |
| | | DataSourceExec: file_groups={16 groups: [[hits.parquet:0..923748528], ...]}, projection=[WatchID, ClientIP, URL], predicate=starts_with(URL@13, http://domcheloveplanet.ru/), metrics=[output_rows=99997497, elapsed_compute=16ns, ... bytes_scanned=3703192723, ... time_elapsed_opening=308.203002ms, time_elapsed_scanning_total=8.350342183s, ...] | | |
| +-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| 1 row(s) fetched. | |
| Elapsed 0.720 seconds. | |
| ``` | |
| In this case, DataFusion actually ran the query, but discarded any results, and | |
| instead returned an annotated plan with a new field, `metrics=[...]` | |
| Most operators have the common metrics `output_rows` and `elapsed_compute` and | |
| some have operator specific metrics such as `DataSourceExec` with `ParquetSource` which has | |
| `bytes_scanned=3703192723`. Note that times and counters are reported across all | |
| cores, so if you have 16 cores, the time reported is the sum of the time taken | |
| by all 16 cores. | |
| Again, reading from bottom up: | |
| - `DataSourceExec` | |
| - `output_rows=99997497`: A total 99.9M rows were produced | |
| - `bytes_scanned=3703192723`: Of the 14GB file, 3.7GB were actually read (due to projection pushdown) | |
| - `bytes_processed=14779976446`: All 14GB were accounted for: the 3.7GB read, plus the bytes of row groups that pruning ruled out. Comparing this against the total size of the files in the plan tells you how far along a scan is | |
| - `time_elapsed_opening=308.203002ms`: It took 300ms to open the file and prepare to read it | |
| - `time_elapsed_scanning_total=8.350342183s`: It took 8.3 seconds of CPU time (across 16 cores) to actually decode the parquet data | |
| - `FilterExec` | |
| - `output_rows=811821`: Of the 99.9M rows at its input, only 811K rows passed the filter and were produced at the output | |
| - `elapsed_compute=1.36923816s`: In total, 1.36s of CPU time (across 16 cores) was spend evaluating the filter | |
| - `CoalesceBatchesExec` | |
| - `output_rows=811821`, `elapsed_compute=12.873379ms`: Produced 811K rows in 13ms | |
| - `ProjectionExec` | |
| - `output_rows=811821, elapsed_compute=66.25µs`: Produced 811K rows in 66µs (microseconds). This projection is almost instantaneous as it does not manipulate any data | |
| - `SortExec` | |
| - `output_rows=75`: Produced 75 rows in total. Each of 16 cores could produce up to 5 rows, but in this case not all cores did. | |
| - `elapsed_compute=7.243038ms`: 7ms was used to determine the top 5 rows | |
| - `row_replacements=482`: Internally, the TopK operator updated its top list 482 times | |
| - `SortPreservingMergeExec` | |
| - `output_rows=5`, `elapsed_compute=2.375µs`: Produced the final 5 rows in 2.375µs (microseconds) | |
| When predicate pushdown is enabled, `DataSourceExec` with `ParquetSource` gains the following metrics: | |
| - `output_rows_skew`: output skew score derived from per-partition `output_rows`. `0%` is perfectly balanced, `100%` is maximally skewed, and `N/A` means no output rows were produced. | |
| - `page_index_rows_pruned`: number of rows evaluated by page index filters. The metric reports both how many rows were considered in total and how many matched (were not pruned). | |
| - `page_index_pages_pruned`: number of pages evaluated by page index filters. The metric reports both how many pages were considered in total and how many matched (were not pruned). | |
| - `page_index_pages_skipped_by_fully_matched`: number of pages for which page-index pruning was skipped because row-group statistics proved the containing row group was fully matched. These pages are still scanned. | |
| - `row_groups_pruned_bloom_filter`: number of row groups evaluated by Bloom Filters, reporting both total checked groups and groups that matched. | |
| - `row_groups_pruned_statistics`: number of row groups evaluated by row-group statistics (min/max), reporting both total checked groups and groups that matched. | |
| - `limit_pruned_row_groups`: number of row groups pruned by the limit. | |
| - `limit_pruned_rows`: number of rows skipped by limit pruning when fully matched page ranges contain enough rows to satisfy the limit. | |
| - `pushdown_rows_matched`: rows that were tested by any of the above filters, and passed all of them. | |
| - `pushdown_rows_pruned`: rows that were tested by any of the above filters, and did not pass at least one of them. | |
| - `predicate_evaluation_errors`: number of times evaluating the filter expression failed (expected to be zero in normal operation) | |
| - `num_predicate_creation_errors`: number of errors creating predicates (expected to be zero in normal operation) | |
| - `bloom_filter_eval_time`: time spent parsing and evaluating Bloom Filters | |
| - `statistics_eval_time`: time spent parsing and evaluating row group-level statistics | |
| - `row_pushdown_eval_time`: time spent evaluating row-level filters | |
| - `page_index_eval_time`: time required to evaluate the page index filters | |
| `output_rows_skew` is also reported by operators that execute in multiple | |
| partitions: `RepartitionExec`, `HashJoinExec`, `AggregateExec`, | |
| `BoundedWindowAggExec` and `WindowAggExec`. It shows how evenly their output | |
| rows are spread across partitions, which helps spot data skew (for example, a | |
| hot join or grouping key). With a single partition the skew is always `0%`. | |
| ## Postgres-style `EXPLAIN (...)` options | |
| In addition to the legacy keyword form (`EXPLAIN ANALYZE VERBOSE FORMAT tree SELECT ...`), | |
| DataFusion accepts a Postgres-style option list on dialects whose | |
| [`supports_explain_with_utility_options`](https://docs.rs/sqlparser/latest/sqlparser/dialect/trait.Dialect.html#method.supports_explain_with_utility_options) | |
| returns `true`. This includes the default `GenericDialect`, `PostgreSqlDialect`, and | |
| `DuckDbDialect`, among others. | |
| ```sql | |
| EXPLAIN (ANALYZE, VERBOSE, METRICS 'rows,bytes', LEVEL dev) | |
| SELECT ... ; | |
| ``` | |
| The recognized options are: | |
| | Option | Argument | Effect | | |
| | --------- | ----------------- | ------------------------------------------------------------------------------------------------------------------------------------ | | |
| | `ANALYZE` | boolean, optional | Execute the plan and collect metrics. Defaults to `TRUE` when bare. Equivalent to the `ANALYZE` keyword. | | |
| | `VERBOSE` | boolean, optional | Show per-partition metrics and additional detail. Equivalent to the `VERBOSE` keyword. | | |
| | `FORMAT` | identifier/string | One of `indent`, `tree`, `pgjson`, `graphviz`. Equivalent to the `FORMAT <format>` clause. | | |
| | `METRICS` | string | Filter `ANALYZE` metrics by category. Accepts `'all'`, `'none'`, or any comma-separated subset of `rows,bytes,timing,uncategorized`. | | |
| | `LEVEL` | identifier/string | `summary` or `dev`. Controls metric verbosity for `ANALYZE`. | | |
| | `TIMING` | boolean | Sugar over `METRICS`: toggles inclusion of the `timing` category. | | |
| | `SUMMARY` | boolean | Sugar over `LEVEL`: `TRUE` → `summary`, `FALSE` → `dev`. | | |
| | `COSTS` | boolean | Include statistics in plain `EXPLAIN` output (equivalent to `SET datafusion.explain.show_statistics`). Not valid with `ANALYZE`. | | |
| Boolean arguments can be written bare (`ANALYZE` → `true`), as `TRUE`/`FALSE`, | |
| `ON`/`OFF`, or `0`/`1`. | |
| When combined with `ANALYZE`, `FORMAT` supports `indent` (the default) and | |
| `pgjson`; `tree` and `graphviz` are rejected. The `pgjson` form emits the | |
| physical plan with live metrics, which is handy for plan visualizers — see | |
| [`pgjson` format with `ANALYZE`](sql/explain.md#pgjson-format-with-analyze): | |
| ```sql | |
| EXPLAIN (ANALYZE, FORMAT pgjson) SELECT ...; | |
| ``` | |
| The statement-level options take precedence over session config, so you can leave | |
| the session defaults alone and override just for the current query: | |
| ```sql | |
| EXPLAIN (ANALYZE, LEVEL dev, METRICS 'rows,bytes') SELECT ...; | |
| ``` | |
| Postgres options that DataFusion does not model (`BUFFERS`, `WAL`, `SETTINGS`, | |
| `GENERIC_PLAN`, `MEMORY`) return a clear error rather than being silently | |
| accepted — use `METRICS` to filter what appears in the output. | |
| ## Partitions and Execution | |
| DataFusion determines the optimal number of cores to use as part of query | |
| planning. Roughly speaking, each "partition" in the plan is run independently using | |
| a separate core. Data crosses between cores only within certain operators such as | |
| `RepartitionExec`, `CoalescePartitions` and `SortPreservingMergeExec` | |
| You can read more about this in the [Partitioning Docs]. | |
| [partitioning docs]: https://docs.rs/datafusion/latest/datafusion/physical_expr/enum.Partitioning.html | |
| ## Example of an Aggregate Query | |
| Let us delve into an example query that aggregates data from the `hits.parquet` | |
| file. For example, this query from ClickBench finds the top 10 users by their | |
| number of hits: | |
| ```sql | |
| SELECT "UserID", COUNT(*) | |
| FROM 'hits.parquet' | |
| GROUP BY "UserID" | |
| ORDER BY COUNT(*) DESC | |
| LIMIT 10; | |
| ``` | |
| We can again see the query plan by using `EXPLAIN`: | |
| ```sql | |
| > EXPLAIN FORMAT INDENT SELECT "UserID", COUNT(*) FROM 'hits.parquet' GROUP BY "UserID" ORDER BY COUNT(*) DESC LIMIT 10; | |
| +---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| | plan_type | plan | | |
| +---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| | logical_plan | Limit: skip=0, fetch=10 | | |
| | | Sort: count(*) DESC NULLS FIRST, fetch=10 | | |
| | | Aggregate: groupBy=[[hits.parquet.UserID]], aggr=[[count(Int64(1)) AS count(*)]] | | |
| | | TableScan: hits.parquet projection=[UserID] | | |
| | physical_plan | GlobalLimitExec: skip=0, fetch=10 | | |
| | | SortPreservingMergeExec: [count(*)@1 DESC], fetch=10 | | |
| | | SortExec: TopK(fetch=10), expr=[count(*)@1 DESC], preserve_partitioning=[true] | | |
| | | AggregateExec: mode=FinalPartitioned, gby=[UserID@0 as UserID], aggr=[count(*)] | | |
| | | CoalesceBatchesExec: target_batch_size=8192 | | |
| | | RepartitionExec: partitioning=Hash([UserID@0], 10), input_partitions=10 | | |
| | | AggregateExec: mode=Partial, gby=[UserID@0 as UserID], aggr=[count(*)] | | |
| | | DataSourceExec: file_groups={10 groups: [[hits.parquet:0..1477997645], [hits.parquet:1477997645..2955995290], [hits.parquet:2955995290..4433992935], [hits.parquet:4433992935..5911990580], [hits.parquet:5911990580..7389988225], ...]}, projection=[UserID], file_type=parquet | | |
| | | | | |
| +---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | |
| ``` | |
| For this query, let's again read the plan from the bottom to the top: | |
| **Logical plan operators** | |
| - `TableScan` | |
| - `hits.parquet`: Scans data from the file `hits.parquet`. | |
| - `projection=[UserID]`: Reads only the `UserID` column | |
| - `Aggregate` | |
| - `groupBy=[[hits.parquet.UserID]]`: Groups by `UserID` column. | |
| - `aggr=[[count(Int64(1)) AS count(*)]]`: Applies the `COUNT` aggregate on each distinct group. | |
| - `Sort` | |
| - `count(*) DESC NULLS FIRST`: Sorts the data in descending count order. | |
| - `fetch=10`: Returns only the first 10 rows. | |
| - `Limit` | |
| - `skip=0`: Does not skip any data for the results. | |
| - `fetch=10`: Limits the results to 10 values. | |
| **Physical plan operators** | |
| - `DataSourceExec` | |
| - `file_groups={10 groups: [...]}`: Reads 10 groups in parallel from `hits.parquet`file. (The example above was run on a machine with 10 cores.) | |
| - `projection=[UserID]`: Pushes down projection of the `UserID` column. The parquet format is columnar and the DataFusion reader only decodes the columns required. | |
| - `AggregateExec` | |
| - `mode=Partial` Runs a [partial aggregation] in parallel across each of the 10 partitions from the `DataSourceExec` immediately after reading. | |
| - `gby=[UserID@0 as UserID]`: Represents `GROUP BY` in the [physical plan] and groups together the same values of `UserID`. | |
| - `aggr=[count(*)]`: Applies the `COUNT` aggregate on all rows for each group. | |
| - `RepartitionExec` | |
| - `partitioning=Hash([UserID@0], 10)`: Divides the input into 10 (new) output partitions based on the value of `hash(UserID)`. You can read more about this in the [partitioning] documentation. | |
| - `input_partitions=10`: Number of input partitions. | |
| - `CoalesceBatchesExec` | |
| - `target_batch_size=8192`: Combines smaller batches in to larger batches. In this case approximately 8192 rows in each batch. | |
| - `AggregateExec` | |
| - `mode=FinalPartitioned`: Performs the final aggregation on each group. See the [documentation on multi phase grouping] for more information. | |
| - `gby=[UserID@0 as UserID]`: Groups by `UserID`. | |
| - `aggr=[count(*)]`: Applies the `COUNT` aggregate on all rows for each group. | |
| - `SortExec` | |
| - `TopK(fetch=10)`: Use a special "TopK" sort that keeps only the largest 10 values in memory at a time. You can read more about this in the [TopK] documentation. | |
| - `expr=[count(*)@1 DESC]`: Sorts all rows in descending order. Note this represents the `ORDER BY` in the physical plan. | |
| - `preserve_partitioning=[true]`: The sort is done in parallel on each partition. In this case the top 10 values are found for each of the 10 partitions, in parallel. | |
| - `SortPreservingMergeExec` | |
| - `[count(*)@1 DESC]`: This operator merges the 10 distinct streams into a single stream using this expression. | |
| - `fetch=10`: Returns only the first 10 rows | |
| - `GlobalLimitExec` | |
| - `skip=0`: Does not skip any rows | |
| - `fetch=10`: Returns only the first 10 rows, denoted by `LIMIT 10` in the query. | |
| [partial aggregation]: https://docs.rs/datafusion/latest/datafusion/physical_plan/aggregates/enum.AggregateMode.html#variant.Partial | |
| [physical plan]: https://docs.rs/datafusion/latest/datafusion/physical_plan/aggregates/struct.PhysicalGroupBy.html | |
| [partitioning]: https://docs.rs/datafusion/latest/datafusion/physical_plan/repartition/struct.RepartitionExec.html | |
| [topk]: https://docs.rs/datafusion/latest/datafusion/physical_plan/struct.TopK.html | |
| [documentation on multi phase grouping]: https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.Accumulator.html#tymethod.state | |
| ### Data in this Example | |
| The examples in this section use data from [ClickBench], a benchmark for data | |
| analytics. The examples are in terms of the 14GB [`hits.parquet`] file and can be | |
| downloaded from the website or using the following commands: | |
| ```shell | |
| cd benchmarks | |
| ./bench.sh data clickbench_1 | |
| *************************** | |
| DataFusion Benchmark Runner and Data Generator | |
| COMMAND: data | |
| BENCHMARK: clickbench_1 | |
| DATA_DIR: /Users/andrewlamb/Software/datafusion2/benchmarks/data | |
| CARGO_COMMAND: cargo run --release | |
| PREFER_HASH_JOIN: true | |
| *************************** | |
| Checking hits.parquet...... found 14779976446 bytes ... Done | |
| ``` | |
| Then you can run `datafusion-cli` to get plans: | |
| ```shell | |
| cd datafusion/benchmarks/data | |
| datafusion-cli | |
| DataFusion CLI v41.0.0 | |
| > select count(*) from 'hits.parquet'; | |
| +----------+ | |
| | count(*) | | |
| +----------+ | |
| | 99997497 | | |
| +----------+ | |
| 1 row(s) fetched. | |
| Elapsed 0.062 seconds. | |
| > | |
| ``` | |
| [clickbench]: https://benchmark.clickhouse.com/ | |
| [`hits.parquet`]: https://datasets.clickhouse.com/hits_compatible/hits.parquet |