Method execute

datafusion/physical-plan/src/work_table.rs:196–224

Stream the batches that were written to the work table.

(
        &self,
        partition: usize,
        _context: Arc<TaskContext>,
    )

    /// Stream the batches that were written to the work table.
    fn execute(
        &self,
        partition: usize,
        _context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        // WorkTable streams must be the plan base.
        assert_eq_or_internal_err!(
            partition,
            0,
            "WorkTableExec got an invalid partition {partition} (expected 0)"
        );
        let ReservedBatches {
            mut batches,
            reservation,
        } = self.work_table.take()?;
        if let Some(projection) = &self.projection {
            // We apply the projection
            // TODO: it would be better to apply it as soon as possible and not only here
            // TODO: an aggressive projection makes the memory reservation smaller, even if we do not edit it
            batches = batches
                .into_iter()
                .map(|b| b.project(projection))
                .collect::<Result<Vec<_>, _>>()?;
        }

        let stream = MemoryStream::try_new(batches, Arc::clone(&self.schema), None)?
            .with_reservation(reservation);
        Ok(Box::pin(cooperative(stream)))
    }

    fn metrics(&self) -> Option<MetricsSet> {
        Some(self.metrics.clone_inner())