zhuqi-lucas opened a new pull request, #27079:
URL: https://github.com/apache/datafusion/pull/27079
## Which issue does this PR close?
- Part of #17172 (Part 1, order inversion). Follows #19064 (`Inexact`
reverse scan) and #19329 (`Exact` for the natural ordering).
- Related: #19048 (row group size and memory for the reverse scan).
## Rationale for this change
When a query asks for the reverse of the files' declared ordering (for
example `ORDER BY ts DESC LIMIT 10` on files written in `ts ASC`), the parquet
scan already reverses the row group order, but rows inside each row group stay
in on-disk order. The pushdown is therefore `Inexact` and a `SortExec` (usually
`TopK`) stays above the scan, which has to consume at least one full row group
per file before it can emit anything.
Reading each row group back to front makes the scan exactly ordered, so the
`SortExec` can go and the `LIMIT` can be pushed into the scan.
## What changes are included in this PR?
- New option `datafusion.execution.parquet.enable_exact_reverse_scan`
(default `false`).
- `ParquetSource::try_pushdown_sort` returns `Exact` when the reversed
declared ordering satisfies the request and the option is set. The source sets
a new `reverse_rows` flag next to `reverse_row_groups`, and does not reorder
row groups by statistics, since that is not order-preserving for row groups
that share a `min`.
- The parquet stream drains each row group's reader, reverses the rows of
every batch, and emits the batches last to first. At most one decoded row group
(after projection and filtering) is held at a time.
- A limit is counted in decode order, which the reversal does not follow, so
a reversing scan does not pass it to the decoder or use it for row group / page
pruning; the file stream applies it to the reversed output. Under a limit the
scan reads less instead:
- a row group whose output is known up front (no predicate, or fully
matched by statistics) is cut to its last `limit` rows with a row selection,
and the offset index is loaded so the pages ahead of them are not fetched;
- for other row groups only the last `limit` decoded rows are kept, so at
most `limit` rows plus one batch are held.
- The buffered rows are charged to the task's memory pool through a new
`FileSource::with_memory_pool` hook (default: no accounting; `FileScanConfig`
calls it when the scan is opened). An oversized row group fails the scan with a
resources error instead of growing the process.
- `FileScanConfig::rebuild_with_source` reverses the files of each group,
declares the reversed orderings, and re-validates them, falling back to
`Inexact` when they do not hold (for example overlapping files). For a source
that is `Exact` but whose request is not literally the reverse of the declared
ordering, the forward `Exact` path now also checks that the kept ordering
satisfies the request, so such a claim is downgraded instead of keeping an
ordering that does not describe the scan.
- `reverse_rows` is carried in `ParquetScanExecNode` and the option in
`ParquetOptions` proto.
## Are these changes tested?
- Opener: rows reversed within row groups spanning several batches, limit
not applied in decode order, pushed-down filter before reversal, limit with a
filter (selected tail for the fully matched row group, kept tail for the
other), fewer bytes fetched under a limit, resources error from a small pool,
reservation released after the scan.
- `PreparedAccessPlan::select_tail_for_limit`: tail selection, spanning row
groups, stopping at a row group whose output is unknown or that already has a
selection.
- `ParquetSource::try_pushdown_sort`: `Exact` with the option, `Inexact`
without.
- `FileScanConfig`: files and ordering reversed, overlapping files
downgraded, an `Exact` claim without file reversal downgraded.
- Proto round trip of a plan with `reverse_rows`.
- `sort_pushdown_exact_reverse.slt`: single and multiple files, filter,
overlapping files, multi-column prefix, NULLs, NOT NULL `DESC NULLS LAST`, two
partitions, option off.
### Benchmark
`sort_pushdown_sorted` (TPC-H SF1 `lineitem`, 3 files sorted by
`l_orderkey`, `WITH ORDER`), median of 5 iterations, this branch with the
option off vs on (run with `dfbench sort-pushdown --sorted` as in `bench.sh run
sort_pushdown_sorted`):
| Query | | off | on | |
|---|---|---:|---:|---:|
| q1 | 3 cols, `ORDER BY l_orderkey` | 97.62 ms | 96.48 ms | 1.01x |
| q2 | 3 cols, `ORDER BY l_orderkey LIMIT 100` | 2.20 ms | 2.11 ms | 1.04x |
| q3 | `*`, `ORDER BY l_orderkey` | 313.01 ms | 313.18 ms | 1.00x |
| q4 | `*`, `ORDER BY l_orderkey LIMIT 100` | 4.74 ms | 4.76 ms | 1.00x |
| q5 | 3 cols, `DESC LIMIT 100` | 7.65 ms | 2.08 ms | 3.68x |
| q6 | 3 cols, `DESC LIMIT 1000` | 10.50 ms | 2.18 ms | 4.82x |
| q7 | `*`, `DESC LIMIT 100` | 13.61 ms | 3.99 ms | 3.41x |
| q8 | `*`, `DESC LIMIT 1000` | 15.54 ms | 4.04 ms | 3.85x |
The reversed `LIMIT` queries now cost about the same as their forward
counterparts (q5 vs q2, q7 vs q4). The `sort_pushdown_inexact*` benchmarks are
unaffected: their files overlap, so the exact path does not apply.
## Are there any user-facing changes?
A new config option, off by default, and a new
`FileSource::with_memory_pool` method with a default implementation. With the
option on, plans for reversed requests drop the `SortExec` and show
`reverse_rows=true` on the scan. A scan without a limit still holds one decoded
row group per partition (charged to the memory pool), which is why the option
is off by default; page-level reverse decoding (#19486) would bound it further.
--
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]
