You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: stream.rs
    path of file: ./repos/cnosdb/query_server/query/src/extension/physical/plan_node/gapfill
    
    the code of the file until where you have to start completion: //! Implementation of [Stream] that performs gap-filling on tables.
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use datafusion::arrow;
use datafusion::arrow::array::{ArrayRef, TimestampNanosecondArray};
use datafusion::arrow::datatypes::SchemaRef;
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::error::{DataFusionError, Result};
use datafusion::execution::memory_pool::MemoryReservation;
use datafusion::physical_plan::expressions::Column;
use datafusion::physical_plan::metrics::{BaselineMetrics, RecordOutput};
use datafusion::physical_plan::{
    ExecutionPlan, PhysicalExpr, RecordBatchStream, SendableRecordBatchStream,
};
use futures::{ready, Stream, StreamExt};

use super::algo::GapFiller;
use super::buffered_input::BufferedInput;
use super::params::GapFillParams;
use super::GapFillExec;

/// An implementation of a gap-filling operato
    
    fn process(&mut self, mut input_batch: RecordBatch) -> Result<(RecordBatch, RecordBatch)> {
        let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();

        let input_time_array = self
            .time_expr
            .evaluate(&input_batch)?
            .into_array(input_batch.num_rows());
        let input_time_array: &TimestampNanosecondArray = input_time_array
            .as_any()
            .downcast_ref()
            .ok_or(DataFusionError::Internal(
                "time array must be a TimestampNanosecondArray".to_string(),
            ))?;
        let input_time_array = (expr_to_index(&self.time_expr), input_time_array);

        let group_arrays = self.group_arrays(&input_batch)?;
        let aggr_arrays = self.aggr_arrays(&input_batch)?;

        let timer = elapsed_compute.timer();
        let output_batch = self
            .gap_filler
            .build_gapfilled_output(
                Arc::clone(&self.schema),
                input_time_array,
                &group_arrays,
                &aggr_arrays,
            )
            .record_output(&self.baseline_metrics)?;
        timer.done();

        self.reservation
            .try_grow(output_batch.get_array_memory_size())?;

        // Slice the input to just what is needed moving forward, with one context
        // row before the next input offset.
        input_batch = self.gap_filler.slice_input_batch(input_batch)?;

        Ok((output_batch, input_batch))
    }