Method run_input

datafusion/physical-plan/src/stream.rs:326–382

Runs the `partition` of the `input` ExecutionPlan on the tokio thread pool and writes its outputs to this stream If the input partition produces an error, the error will be sent to the output stream and no further results are sent.

(
        &mut self,
        input: Arc<dyn ExecutionPlan>,
        partition: usize,
        context: Arc<TaskContext>,
    )

    /// If the input partition produces an error, the error will be
    /// sent to the output stream and no further results are sent.
    pub(crate) fn run_input(
        &mut self,
        input: Arc<dyn ExecutionPlan>,
        partition: usize,
        context: Arc<TaskContext>,
    ) {
        let output = self.tx();
        let input_display = if log::log_enabled!(log::Level::Debug) {
            displayable(input.as_ref()).one_line().to_string()
        } else {
            String::new()
        };

        self.inner.spawn(async move {
            let mut stream = match input.execute(partition, context) {
                Err(e) => {
                    // If send fails, the plan being torn down, there
                    // is no place to send the error and no reason to continue.
                    output.send(Err(e)).await.ok();
                    debug!(
                        "Stopping execution: error executing input: {input_display}",
                    );
                    return Ok(());
                }
                Ok(stream) => stream,
            };

            // Drop the input early, as soon as we're done with it.
            // Holding on to it can cause delays in cancelling the child plan when the query is
            // cancelled.
            drop(input);

            // Transfer batches from inner stream to the output tx
            // immediately.
            while let Some(item) = stream.next().await {
                let is_err = item.is_err();

                // If send fails, plan being torn down, there is no
                // place to send the error and no reason to continue.
                if output.send(item).await.is_err() {
                    debug!(
                        "Stopping execution: output is gone, plan cancelling: {input_display}",
                    );
                    return Ok(());
                }

                // Stop after the first error is encountered (Don't
                // drive all streams to completion)
                if is_err {
                    debug!("Stopping execution: plan returned error: {input_display}");
                    return Ok(());
                }
            }

            Ok(())
        });
    }