zhuqi-lucas commented on code in PR #27079:
URL: https://github.com/apache/datafusion/pull/27079#discussion_r4508975400
##########
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, filed #27254 to fix 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]
