Method transition

datafusion/datasource-parquet/src/push_decoder.rs:149–205

Advances the decoder state machine until the next [`RecordBatch`] is produced, the file is fully consumed, or an error occurs. On each iteration the decoder is polled via [`ParquetPushDecoder::try_decode`]: - [`NeedsData`](DecodeResult::NeedsData) – the requested byte ranges are fetched from the [`AsyncFileReader`] and fed back into the decoder. - [`Data`](DecodeResult::Data) – a decoded batch is

(mut self)

    /// miri where `&mut self` creates a single opaque borrow that conflicts
    /// with `unfold`'s ownership across yield points.
    async fn transition(mut self) -> Option<(Result<RecordBatch>, Self)> {
        loop {
            if self.remaining_limit == Some(0) {
                return None;
            }
            match self.decoder.try_decode() {
                Ok(DecodeResult::NeedsData(ranges)) => {
                    let data = self
                        .reader
                        .get_byte_ranges(ranges.clone())
                        .await
                        .map_err(DataFusionError::from);
                    match data {
                        Ok(data) => {
                            if let Err(e) = self.decoder.push_ranges(ranges, data) {
                                return Some((Err(DataFusionError::from(e)), self));
                            }
                        }
                        Err(e) => return Some((Err(e), self)),
                    }
                }
                Ok(DecodeResult::Data(batch)) => {
                    let batch = if let Some(remaining_limit) = self.remaining_limit {
                        if batch.num_rows() > remaining_limit {
                            self.remaining_limit = Some(0);
                            batch.slice(0, remaining_limit)
                        } else {
                            self.remaining_limit =
                                Some(remaining_limit - batch.num_rows());
                            batch
                        }
                    } else {
                        batch
                    };
                    let mut timer = self.baseline_metrics.elapsed_compute().timer();
                    self.copy_arrow_reader_metrics();
                    let result = self.project_batch(&batch);
                    timer.stop();
                    // Release the borrow on baseline_metrics before moving self
                    drop(timer);
                    return Some((result, self));
                }
                Ok(DecodeResult::Finished) => {
                    // If there are pending decoders (e.g. for consecutive runs
                    // with different filter configurations), switch to the next.
                    if let Some(next) = self.pending_decoders.pop_front() {
                        self.decoder = next;
                        continue;
                    }
                    return None;
                }
                Err(e) => {
                    return Some((Err(DataFusionError::from(e)), self));
                }
            }
        }
    }