jayzhan211 commented on code in PR #27079:
URL: https://github.com/apache/datafusion/pull/27079#discussion_r4493427140
##########
datafusion/datasource/src/file_scan_config/sort_pushdown.rs:
##########
@@ -97,6 +97,29 @@ impl FileScanConfig {
new_config.file_source = new_file_source;
+ // Exact reverse: the source reads every file back to front, rows
+ // within each row group included, so with each group's files
+ // reversed too the scan emits the declared orderings reversed.
+ // Re-validate against the reversed groups before keeping them;
+ // if the ordering does not hold, clear it and the caller falls
+ // back to `Inexact`, which keeps the `SortExec`.
+ if reverse_file_groups && is_exact {
Review Comment:
This branch can reverse a scan that already has a limit, which then keeps
rows from the wrong end of the file. With the setup from my review comment,
this returns `40 39 38 37 36` instead of `5 4 3 2 1`. Main also gets this query
wrong through the `Inexact` path, so this guard fixes both. Can you add it,
with this query in the slt?
```sql
SELECT id FROM (SELECT * FROM t ORDER BY id ASC LIMIT 5) ORDER BY id DESC;
```
```diff
--- a/datafusion/datasource/src/file_scan_config/mod.rs
+++ b/datafusion/datasource/src/file_scan_config/mod.rs
) -> Result<SortOrderPushdownResult<Arc<dyn DataSource>>> {
+ // Reordering the scan changes which rows its limit keeps.
+ if self.limit.is_some() {
+ return Ok(SortOrderPushdownResult::Unsupported);
+ }
```
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -1029,6 +1066,27 @@ impl FileSource for ParquetSource {
})
});
+ // Exact reverse: when the reversed declared ordering satisfies the
+ // request and `enable_exact_reverse_scan` is set, read every file
+ // exactly back to front, row groups and the rows inside them, so
+ // the scan itself produces the requested order. Row groups keep
+ // their on-disk order before the flip: reordering them by
+ // statistics is not order-preserving when two row groups share a
+ // `min`. `FileScanConfig` reverses the files of each group and
+ // re-validates the ordering, falling back to `Inexact` if it does
+ // not hold.
+ if reversed_satisfies
+ && self.table_parquet_options.global.enable_exact_reverse_scan
+ {
+ let mut new_source = self.clone();
+ new_source.sort_order_for_reorder = None;
+ new_source.reverse_row_groups = true;
+ new_source.reverse_rows = true;
Review Comment:
A second pushdown on an already reversed scan flips the files and the
ordering back but leaves these flags on: I got `output_ordering=[id ASC]` with
rows still in `DESC` order. A single `PushdownSort` pass can't reach this, but
a second optimizer pass or a direct `try_pushdown_sort` call can. Toggling
fixes it, and a unit test that pushes down twice would cover it.
```suggestion
new_source.reverse_row_groups = !self.reverse_rows;
new_source.reverse_rows = !self.reverse_rows;
```
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -706,6 +743,82 @@ impl PushDecoderStreamState {
fn project_batch(&self, batch: &RecordBatch) -> Result<RecordBatch> {
self.decoder_projection.map(batch)
}
+
+ /// Decodes every remaining batch of `reader`'s row group, projects it,
+ /// reverses its rows, and stores the results in
+ /// [`Self::reversed_batches`]. Batches stay in decode order there, so
+ /// popping them yields the row group's rows last to first.
+ ///
+ /// Under [`Self::reverse_limit`] only the row group's last `limit` rows
+ /// are kept: batches that fall entirely before them are dropped as soon
+ /// as later batches cover the limit, so at most `limit` rows plus one
+ /// batch are held while decoding.
+ fn buffer_reversed_row_group(
+ &mut self,
+ reader: ParquetRecordBatchReader,
+ ) -> Result<()> {
+ let mut kept: VecDeque<RecordBatch> = VecDeque::new();
+ let mut kept_rows = 0;
+ for batch in reader {
+ let batch = batch.map_err(DataFusionError::from)?;
+ self.copy_arrow_reader_metrics();
+ let batch = self.project_batch(&batch)?;
+ if batch.num_rows() == 0 {
+ continue;
+ }
+ if let Some(reservation) = &self.reversed_reservation {
+ reservation.try_grow(batch.get_array_memory_size())?;
Review Comment:
Each batch charges the full size of buffers it shares with other batches, so
a large page dictionary is counted once per batch, and this scan can't spill.
With 1M rows and 4000 distinct strings (about 25 MB) a 100 MB pool fails here,
while the option-off plan finishes at 30 MB by spilling. It's fine to handle
this in a follow-up since it needs shared buffers counted once across batches,
but could you file an issue so it isn't lost before this option goes on by
default?
##########
datafusion/common/src/config.rs:
##########
@@ -1354,6 +1354,14 @@ config_namespace! {
/// pattern of selected rows.
pub force_filter_selections: bool, default = false
+ /// (reading) If true, a scan whose requested ordering is the reverse
of
+ /// the files' declared ordering reads each row group into memory and
+ /// emits its rows in reverse, so the scan is exactly ordered and the
+ /// `SortExec` above it can be removed. Peak memory per scanned file is
+ /// one decoded row group (after projection and filtering). If false,
+ /// only the row group order is reversed and the sort is kept.
Review Comment:
Optional cleanup: a later `SET` has no effect on an existing table in either
direction, so someone hitting the resources error can't turn this off that way.
Worth saying so here.
```suggestion
/// only the row group order is reversed and the sort is kept. The
value
/// is captured when the table is created, so a later `SET` does not
/// affect existing tables.
```
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -307,6 +309,22 @@ pub(crate) struct PushDecoderStreamState {
/// group at a time as they are decoded or skipped, and topped up to the
/// full range when the stream is dropped.
pub(crate) byte_progress: ByteProgress,
+ /// Emit each row group's rows in reverse order. The active reader is
+ /// drained into [`Self::reversed_batches`] before anything is returned,
+ /// so one decoded row group is held in memory at a time.
+ pub(crate) reverse_rows: bool,
Review Comment:
Optional cleanup: these four fields are only meaningful together, so one
optional struct would remove the `.filter(|_| prepared.reverse_rows)` calls in
the opener and the `self.reverse_rows` check here.
```rs
pub(crate) reverser: Option<RowGroupReverser>,
```
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -1029,6 +1066,27 @@ impl FileSource for ParquetSource {
})
});
+ // Exact reverse: when the reversed declared ordering satisfies the
+ // request and `enable_exact_reverse_scan` is set, read every file
+ // exactly back to front, row groups and the rows inside them, so
+ // the scan itself produces the requested order. Row groups keep
+ // their on-disk order before the flip: reordering them by
+ // statistics is not order-preserving when two row groups share a
+ // `min`. `FileScanConfig` reverses the files of each group and
+ // re-validates the ordering, falling back to `Inexact` if it does
+ // not hold.
+ if reversed_satisfies
Review Comment:
The source decides this with equivalence reasoning but `FileScanConfig`
compares literally, so `DESC NULLS LAST` on a NOT NULL column, `WHERE a = 1
ORDER BY b DESC` and `ORDER BY date_trunc('day', ts) DESC` get downgraded and
keep `reverse_rows` under a TopK. On lineitem at 8 partitions that took 2053 MB
vs 469 MB with the option off. If the config uses the same reasoning, these
become real `Exact` scans: 3 ms and 250 MB on that query.
Test 8 and `sort_pushdown_exact_without_file_reversal_downgraded_to_inexact`
then flip to `Exact`, and
`sort_pushdown_exact_source_out_of_order_returns_exact` must keep passing.
Could you add the other two shapes to the slt? I get `(1, 20), (1, 19), (1,
18)` for `SELECT * FROM ab WHERE a = 1 ORDER BY b DESC LIMIT 3` on a table
declared `(a ASC, b ASC)`.
```diff
--- a/datafusion/datasource/src/file_scan_config/sort_pushdown.rs
+++ b/datafusion/datasource/src/file_scan_config/sort_pushdown.rs
+fn reversed_file_groups(file_groups: Vec<FileGroup>) -> Vec<FileGroup> {
+ file_groups
+ .into_iter()
+ .map(|group| {
+ let mut files = group.into_inner();
+ files.reverse();
+ files.into()
+ })
+ .collect()
+}
+
+fn reversed_orderings(orderings: &[LexOrdering]) -> Vec<LexOrdering> {
+ orderings
+ .iter()
+ .filter_map(|ordering| {
+ LexOrdering::new(ordering.iter().map(|expr| expr.reverse()))
+ })
+ .collect()
+}
+
impl FileScanConfig {
+ /// Whether only a back-to-front read of this scan satisfies `order`:
the
+ /// declared orderings do not, but their reverse does.
+ fn reverse_scan_satisfies(&self, order: &[PhysicalSortExpr]) ->
Result<bool> {
+ if self.eq_properties().ordering_satisfy(order.iter().cloned())? {
+ return Ok(false);
+ }
+ let mut reversed = self.clone();
+ reversed.file_groups = reversed_file_groups(reversed.file_groups);
+ reversed.output_ordering =
reversed_orderings(&self.output_ordering);
+ reversed.eq_properties().ordering_satisfy(order.iter().cloned())
+ }
@@ rebuild_with_source
orderings
.iter()
.any(|ordering| ordering.is_reverse(&requested))
+ || (is_exact && self.reverse_scan_satisfies(order)?)
```
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -706,6 +743,82 @@ impl PushDecoderStreamState {
fn project_batch(&self, batch: &RecordBatch) -> Result<RecordBatch> {
self.decoder_projection.map(batch)
}
+
+ /// Decodes every remaining batch of `reader`'s row group, projects it,
+ /// reverses its rows, and stores the results in
+ /// [`Self::reversed_batches`]. Batches stay in decode order there, so
+ /// popping them yields the row group's rows last to first.
+ ///
+ /// Under [`Self::reverse_limit`] only the row group's last `limit` rows
+ /// are kept: batches that fall entirely before them are dropped as soon
+ /// as later batches cover the limit, so at most `limit` rows plus one
+ /// batch are held while decoding.
+ fn buffer_reversed_row_group(
+ &mut self,
+ reader: ParquetRecordBatchReader,
+ ) -> Result<()> {
+ let mut kept: VecDeque<RecordBatch> = VecDeque::new();
+ let mut kept_rows = 0;
+ for batch in reader {
Review Comment:
Optional cleanup: this decodes the whole row group inside one poll, about
150 ms per 1M rows on `SELECT *` here, and a query can't be cancelled until it
finishes. Making this function async and yielding between batches would keep it
cancellable.
```rs
tokio::task::yield_now().await;
```
--
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]
