Method execute

datafusion/physical-plan/src/analyze.rs:164–227

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

    }

    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        assert_eq_or_internal_err!(
            partition,
            0,
            "AnalyzeExec invalid partition. Expected 0, got {partition}"
        );

        // Gather futures that will run each input partition in
        // parallel (on a separate tokio task) using a JoinSet to
        // cancel outstanding futures on drop
        let num_input_partitions = self.input.output_partitioning().partition_count();
        let mut builder =
            RecordBatchReceiverStream::builder(self.schema(), num_input_partitions);

        for input_partition in 0..num_input_partitions {
            builder.run_input(
                Arc::clone(&self.input),
                input_partition,
                Arc::clone(&context),
            );
        }

        // Create future that computes the final output
        let start = Instant::now();
        let captured_input = Arc::clone(&self.input);
        let captured_schema = Arc::clone(&self.schema);
        let verbose = self.verbose;
        let show_statistics = self.show_statistics;
        let metric_types = self.metric_types.clone();
        let metric_categories = self.metric_categories.clone();

        // future that gathers the results from all the tasks in the
        // JoinSet that computes the overall row count and final
        // record batch
        let mut input_stream = builder.build();
        let output = async move {
            let mut total_rows = 0;
            while let Some(batch) = input_stream.next().await.transpose()? {
                total_rows += batch.num_rows();
            }
            drop(input_stream);

            let duration = Instant::now() - start;
            create_output_batch(
                verbose,
                show_statistics,
                total_rows,
                duration,
                &captured_input,
                &captured_schema,
                &metric_types,
                metric_categories.as_deref(),
            )
        };