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: histogram_fold.rs
    path of file: ./repos/greptimedb/src/promql/src/extension_plan
    
    the code of the file until where you have to start completion: // Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::any::Any;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::task::Poll;
use std::time::Instant;

use common_recordbatch::RecordBatch as GtRecordBatch;
use common_telemetry::warn;
use datafusion::arrow::array::AsArray;
use datafusion::arrow::compute::{self, concat_batches, SortOptions};
use datafusion::arrow::datatypes::{DataType, Float64Type, SchemaRef};
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::common::{DFSchema, DFSchemaRef};
use datafusion::error::{DataFusionError, Result as DataFusionResult};
use datafusion::execution::TaskContext;
use datafusion::logical_expr::{LogicalPlan, UserDefinedLogicalNodeCore};
use
    
    fn fold_buf(&mut self, bucket_num: usize, input: RecordBatch) -> DataFusionResult<()> {
        self.push_input_buf(input);
        // TODO(ruihang): this concat is avoidable.
        let batch = concat_batches(&self.input.schema(), self.input_buffer.drain(..).as_ref())?;
        let mut remaining_rows = self.input_buffered_rows;
        let mut cursor = 0;

        let gt_schema = GtSchema::try_from(self.input.schema()).unwrap();
        let batch = GtRecordBatch::try_from_df_record_batch(Arc::new(gt_schema), batch).unwrap();

        while remaining_rows >= bucket_num {
            // "sample" normal columns
            for normal_index in &self.normal_indices {
                let val = batch.column(*normal_index).get(cursor);
                self.output_buffer[*normal_index].push_value_ref(val.as_value_ref());
            }
            // "fold" `le` and field columns
            let le_array = batch.column(self.le_column_index);
            let field_array = batch.column(self.field_column_index);
            let mut bucket = vec![];
            let mut counters = vec![];
            for bias in 0..bucket_num {
                let le_str_val = le_array.get(cursor + bias);
                let le_str_val_ref = le_str_val.as_value_ref();
                let le_str = le_str_val_ref
                    .as_string()
                    .unwrap()
                    .expect("le column should not be nullable");
                let le = le_str.parse::<f64>().unwrap();
                bucket.push(le);

                let counter = field_array
                    .get(cursor + bias)
                    .as_value_ref()
                    .as_f64()
                    .unwrap()
                    .expect("field column should not be nullable");
                counters.push(counter);
            }
            let result = Self::evaluate_row(self.quantile, &bucket, &counters)?;
            self.output_buffer[self.field_column_index].push_value_ref(ValueRef::from(result));
            cursor += bucket_num;
            remaining_rows -= bucket_num;
            self.output_buffered_rows += 1;
        }

        let remaining_input_batch = batch.into_df_record_batch().slice(cursor, remaining_rows);
        self.input_buffered_rows = remaining_input_batch.num_rows();
        self.input_buffer.push(remaining_input_batch);

        Ok(())
    }