zhuqi-lucas commented on code in PR #27079:
URL: https://github.com/apache/datafusion/pull/27079#discussion_r4508974869
##########
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:
Added in ec8e932c6, with this query as test 13 in the slt and a unit test.
##########
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:
Done with your helper in ec8e932c6. Test 8 and the unit test now expect
`Exact`, `sort_pushdown_exact_source_out_of_order_returns_exact` still passes,
and the other two shapes are tests 14 and 15. Test 14 needs `pushdown_filters =
true` so the scan sees `a = 1`. As a safety net, an `Exact` that still has to
be downgraded now falls back to the original source, so no row reversal is left
under a sort.
##########
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:
Done, with `try_pushdown_sort_exact_reverse_twice_reads_forward`.
##########
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:
Agreed, I will file a follow-up issue for this before the option goes on by
default.
--
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]
