zhuqi-lucas commented on code in PR #27079:
URL: https://github.com/apache/datafusion/pull/27079#discussion_r4508975535
##########
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:
Done.
##########
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:
Done. The stream now pushes one batch per loop iteration and yields between
them, with the compute timer paused across the yield.
##########
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:
Done: `Option<RowGroupReverser>` with `push`, `finish` and `pop`.
--
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]
