/// Execute one partition and return an iterator over RecordBatch
Execute one partition and return an iterator over RecordBatch
Execute one partition and return an iterator over RecordBatch
fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        debug!(partition, "Start SchemaPivotExec::execute");
        if self.output_partitioning().partition_count() <= partition {
            return Err(Error::Internal(format!(
                "SchemaPivotExec invalid partition {}",
                partition
            )));
        }
        let baseline_metrics = BaselineMetrics::new(&self.metrics, partition);
        let input_schema = self.input.schema();
        let input_stream = self.input.execute(partition, context)?;
        // the operation performed in a separate task which is
        // then sent via a channel to the output
        let (tx, rx) = mpsc::channel(1);
        let task = tokio::task::spawn(schema_pivot(
            input_stream,
            input_schema,
            self.schema(),
            tx.clone(),
            baseline_metrics,
        ));
        // A second task watches the output of the worker task and reports errors
        let handle = tokio::task::spawn(watch_task(
            "schema_pivot",
            tx,
            AutoAbortJoinHandle::new(task),
        ));
        debug!(partition, "End SchemaPivotExec::execute");
        Ok(AdapterStream::adapt(
            self.schema(),
            rx,
            Some(Arc::new(AutoAbortJoinHandle::new(handle))),
        ))
    }
fn execute ( & self , partition : usize , context : Arc < TaskContext > , ) -> Result < SendableRecordBatchStream > { debug ! ( partition " " ) ; if self . output_partitioning ( ) . partition_count ( ) <= partition { return Err ( Error :: Internal ( format ! ( " " partition ) ) ) ; } let baseline_metrics = BaselineMetrics :: new ( & self . metrics , partition ) ; let input_schema = self . input . schema ( ) ; let input_stream = self . input . execute ( partition , context ) ? ; // the operation performed in a separate task which is // then sent via a channel to the output let ( tx , rx ) = mpsc :: channel ( 1 ) ; let task = tokio :: task :: spawn ( schema_pivot ( input_stream , input_schema , self . schema ( ) , tx . clone ( ) , baseline_metrics , ) ) ; // A second task watches the output of the worker task and reports errors let handle = tokio :: task :: spawn ( watch_task ( " " , tx , AutoAbortJoinHandle :: new ( task ) , ) ) ; debug ! ( partition " " ) ; Ok ( AdapterStream :: adapt ( self . schema ( ) , rx , Some ( Arc :: new ( AutoAbortJoinHandle :: new ( handle ) ) ) , ) ) }
Execute one partition and return an iterator over RecordBatch
Execute one partition and return an iterator over RecordBatch
&self
partition
usize
context
Arc<TaskContext>