Method execute

datafusion/datasource/src/source.rs:387–411

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

    }

    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        let shared_state = self
            .execution_state
            .get_or_init(|| self.data_source.create_sibling_state())
            .clone();
        let args = OpenArgs::new(partition, Arc::clone(&context))
            .with_shared_state(shared_state);
        let stream = self.data_source.open_with_args(args)?;
        let batch_size = context.session_config().batch_size();

        log::debug!(
            "Batch splitting enabled for partition {partition}: batch_size={batch_size}"
        );
        let metrics = self.data_source.metrics();
        let split_metrics = SplitMetrics::new(&metrics, partition);
        Ok(Box::pin(BatchSplitStream::new(
            stream,
            batch_size,
            split_metrics,
        )))
    }

    fn metrics(&self) -> Option<MetricsSet> {
        let mut metrics = self.data_source.metrics().clone_inner();