Method poll_next_impl

datafusion/physical-plan/src/unnest.rs:348–401

Separate implementation function that unpins the [`UnnestStream`] so that partial borrows work correctly

(
        &mut self,
        cx: &mut std::task::Context<'_>,
    )

    /// Separate implementation function that unpins the [`UnnestStream`] so
    /// that partial borrows work correctly
    fn poll_next_impl(
        &mut self,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Result<RecordBatch>>> {
        loop {
            return Poll::Ready(match ready!(self.input.poll_next_unpin(cx)) {
                Some(Ok(batch)) => {
                    let elapsed_compute =
                        self.metrics.baseline_metrics.elapsed_compute().clone();
                    let timer = elapsed_compute.timer();
                    self.metrics.input_batches.add(1);
                    self.metrics.input_rows.add(batch.num_rows());
                    let result = build_batch(
                        &batch,
                        &self.schema,
                        &self.list_type_columns,
                        &self.struct_column_indices,
                        &self.options,
                    )?;
                    timer.done();
                    let Some(result_batch) = result else {
                        continue;
                    };
                    (&result_batch).record_output(&self.metrics.baseline_metrics);

                    // Empty record batches should not be emitted.
                    // They need to be treated as  [`Option<RecordBatch>`]es and handled separately
                    debug_assert!(result_batch.num_rows() > 0);
                    Some(Ok(result_batch))
                }
                // If the stream is depleted or returned an error, log the finish message:
                other => {
                    trace!(
                        "Processed {} probe-side input batches containing {} rows and \
                        produced {} output batches containing {} rows in {}",
                        self.metrics.input_batches,
                        self.metrics.input_rows,
                        self.metrics.baseline_metrics.output_batches(),
                        self.metrics.baseline_metrics.output_rows(),
                        self.metrics.baseline_metrics.elapsed_compute(),
                    );

                    // In the non-error case, i.e., input is simply depleted:
                    if other.is_none() {
                        // Release the input pipeline's resources.
                        let input_schema = self.input.schema();
                        self.input = Box::pin(EmptyRecordBatchStream::new(input_schema));
                    }

                    other
                }
            });
        }
    }
}

/// Given a set of struct column indices to flatten