fn without_group_by(&mut self) -> Result<Option<RecordBatch>> {
        let aggr_expr_count = self.aggr_expr.len();
        let mut accumulator_set = create_accumulators(&self.aggr_expr)?;

        while let Some(batch) = self.input.borrow_mut().next()? {
            for i in 0..aggr_expr_count {
                match &self.aggr_expr[i] {
                    RuntimeExpr::AggregateFunction { f, args, t, .. } => {
                        // evaluate argument to aggregate function
                        match args[0](&batch) {
                            Ok(array) => match f {
                                AggregateType::Min => accumulator_set
                                    .accumulate_scalar(i, array_min(array, &t)?),
                                AggregateType::Max => accumulator_set
                                    .accumulate_scalar(i, array_max(array, &t)?),
                                AggregateType::Sum => accumulator_set
                                    .accumulate_scalar(i, array_sum(array, &t)?),
                                _ => {
                                    return Err(ExecutionError::NotImplemented(
                                        "Unsupported aggregate function".to_string(),
                                    ));
                                }
                            },
                            Err(_) => {
                                return Err(ExecutionError::ExecutionError(
                                    "Failed to evaluate argument to aggregate function"
                                        .to_string(),
                                ));
                            }
                        }
                    }
                    _ => {
                        return Err(ExecutionError::General(
                            "Invalid aggregate expression".to_string(),
                        ));
                    }
                }
            }
        }

        let mut result_columns: Vec<ArrayRef> = vec![];

        for i in 0..aggr_expr_count {
            let accum = accumulator_set.aggr_values[i].borrow();
            match accum.data_type() {
                DataType::UInt8 => {
                    result_columns.push(array_from_scalar!(UInt8Builder, UInt8, accum)?)
                }
                DataType::UInt16 => {
                    result_columns.push(array_from_scalar!(UInt16Builder, UInt16, accum)?)
                }
                DataType::UInt32 => {
                    result_columns.push(array_from_scalar!(UInt32Builder, UInt32, accum)?)
                }
                DataType::UInt64 => {
                    result_columns.push(array_from_scalar!(UInt64Builder, UInt64, accum)?)
                }
                DataType::Int8 => {
                    result_columns.push(array_from_scalar!(Int8Builder, Int8, accum)?)
                }
                DataType::Int16 => {
                    result_columns.push(array_from_scalar!(Int16Builder, Int16, accum)?)
                }
                DataType::Int32 => {
                    result_columns.push(array_from_scalar!(Int32Builder, Int32, accum)?)
                }
                DataType::Int64 => {
                    result_columns.push(array_from_scalar!(Int64Builder, Int64, accum)?)
                }
                DataType::Float32 => result_columns.push(array_from_scalar!(
                    Float32Builder,
                    Float32,
                    accum
                )?),
                DataType::Float64 => result_columns.push(array_from_scalar!(
                    Float64Builder,
                    Float64,
                    accum
                )?),
                _ => return Err(ExecutionError::NotImplemented("tbd".to_string())),
            }
        }

        Ok(Some(RecordBatch::new(self.schema.clone(), result_columns)))
    }
/// perform simple aggregate on entire columns without grouping logic
perform simple aggregate on entire columns without grouping logic
perform simple aggregate on entire columns without grouping logic
fn without_group_by(&mut self) -> Result<Option<RecordBatch>> {
        let aggr_expr_count = self.aggr_expr.len();
        let mut accumulator_set = create_accumulators(&self.aggr_expr)?;
        while let Some(batch) = self.input.borrow_mut().next()? {
            for i in 0..aggr_expr_count {
                match &self.aggr_expr[i] {
                    RuntimeExpr::AggregateFunction { f, args, t, .. } => {
                        // evaluate argument to aggregate function
                        match args[0](&batch) {
                            Ok(array) => match f {
                                AggregateType::Min => accumulator_set
                                    .accumulate_scalar(i, array_min(array, &t)?),
                                AggregateType::Max => accumulator_set
                                    .accumulate_scalar(i, array_max(array, &t)?),
                                AggregateType::Sum => accumulator_set
                                    .accumulate_scalar(i, array_sum(array, &t)?),
                                _ => {
                                    return Err(ExecutionError::NotImplemented(
                                        "Unsupported aggregate function".to_string(),
                                    ));
                                }
                            },
                            Err(_) => {
                                return Err(ExecutionError::ExecutionError(
                                    "Failed to evaluate argument to aggregate function"
                                        .to_string(),
                                ));
                            }
                        }
                    }
                    _ => {
                        return Err(ExecutionError::General(
                            "Invalid aggregate expression".to_string(),
                        ));
                    }
                }
            }
        }
        let mut result_columns: Vec<ArrayRef> = vec![];
        for i in 0..aggr_expr_count {
            let accum = accumulator_set.aggr_values[i].borrow();
            match accum.data_type() {
                DataType::UInt8 => {
                    result_columns.push(array_from_scalar!(UInt8Builder, UInt8, accum)?)
                }
                DataType::UInt16 => {
                    result_columns.push(array_from_scalar!(UInt16Builder, UInt16, accum)?)
                }
                DataType::UInt32 => {
                    result_columns.push(array_from_scalar!(UInt32Builder, UInt32, accum)?)
                }
                DataType::UInt64 => {
                    result_columns.push(array_from_scalar!(UInt64Builder, UInt64, accum)?)
                }
                DataType::Int8 => {
                    result_columns.push(array_from_scalar!(Int8Builder, Int8, accum)?)
                }
                DataType::Int16 => {
                    result_columns.push(array_from_scalar!(Int16Builder, Int16, accum)?)
                }
                DataType::Int32 => {
                    result_columns.push(array_from_scalar!(Int32Builder, Int32, accum)?)
                }
                DataType::Int64 => {
                    result_columns.push(array_from_scalar!(Int64Builder, Int64, accum)?)
                }
                DataType::Float32 => result_columns.push(array_from_scalar!(
                    Float32Builder,
                    Float32,
                    accum
                )?),
                DataType::Float64 => result_columns.push(array_from_scalar!(
                    Float64Builder,
                    Float64,
                    accum
                )?),
                _ => return Err(ExecutionError::NotImplemented("tbd".to_string())),
            }
        }
        Ok(Some(RecordBatch::new(self.schema.clone(), result_columns)))
    }
fn without_group_by ( & mut self ) -> Result < Option < RecordBatch > > { let aggr_expr_count = self . aggr_expr . len ( ) ; let mut accumulator_set = create_accumulators ( & self . aggr_expr ) ? ; while let Some ( batch ) = self . input . borrow_mut ( ) . next ( ) ? { for i in 0 .. aggr_expr_count { match & self . aggr_expr [ i ] { RuntimeExpr :: AggregateFunction { f , args , t , .. } => { // evaluate argument to aggregate function match args [ 0 ] ( & batch ) { Ok ( array ) => match f { AggregateType :: Min => accumulator_set . accumulate_scalar ( i , array_min ( array , & t ) ? ) , AggregateType :: Max => accumulator_set . accumulate_scalar ( i , array_max ( array , & t ) ? ) , AggregateType :: Sum => accumulator_set . accumulate_scalar ( i , array_sum ( array , & t ) ? ) , _ => { return Err ( ExecutionError :: NotImplemented ( " " . to_string ( ) , ) ) ; } } , Err ( _ ) => { return Err ( ExecutionError :: ExecutionError ( " " . to_string ( ) , ) ) ; } } } _ => { return Err ( ExecutionError :: General ( " " . to_string ( ) , ) ) ; } } } } let mut result_columns : Vec < ArrayRef > = vec ! [ ] ; for i in 0 .. aggr_expr_count { let accum = accumulator_set . aggr_values [ i ] . borrow ( ) ; match accum . data_type ( ) { DataType :: UInt8 => { result_columns . push ( array_from_scalar ! ( UInt8Builder UInt8 accum ) ? ) } DataType :: UInt16 => { result_columns . push ( array_from_scalar ! ( UInt16Builder UInt16 accum ) ? ) } DataType :: UInt32 => { result_columns . push ( array_from_scalar ! ( UInt32Builder UInt32 accum ) ? ) } DataType :: UInt64 => { result_columns . push ( array_from_scalar ! ( UInt64Builder UInt64 accum ) ? ) } DataType :: Int8 => { result_columns . push ( array_from_scalar ! ( Int8Builder Int8 accum ) ? ) } DataType :: Int16 => { result_columns . push ( array_from_scalar ! ( Int16Builder Int16 accum ) ? ) } DataType :: Int32 => { result_columns . push ( array_from_scalar ! ( Int32Builder Int32 accum ) ? ) } DataType :: Int64 => { result_columns . push ( array_from_scalar ! ( Int64Builder Int64 accum ) ? ) } DataType :: Float32 => result_columns . push ( array_from_scalar ! ( Float32Builder Float32 accum ) ? ) , DataType :: Float64 => result_columns . push ( array_from_scalar ! ( Float64Builder Float64 accum ) ? ) , _ => return Err ( ExecutionError :: NotImplemented ( " " . to_string ( ) ) ) , } } Ok ( Some ( RecordBatch :: new ( self . schema . clone ( ) , result_columns ) ) ) }
perform simple aggregate on entire columns without grouping logic
perform simple aggregate on entire columns without grouping logic
&mut self