use std::any::Any;
use std::fmt::Debug;
use std::sync::Arc;
use crate::aggregate::row_accumulator::RowAccumulator;
use crate::{AggregateExpr, PhysicalExpr};
use arrow::array::Int64Array;
use arrow::compute;
use arrow::datatypes::DataType;
use arrow::{array::ArrayRef, datatypes::Field};
use datafusion_common::{downcast_value, ScalarValue};
use datafusion_common::{DataFusionError, Result};
use datafusion_expr::Accumulator;
use datafusion_row::accessor::RowAccessor;
use crate::expressions::format_state_name;
#[derive(Debug, Clone)]
pub struct Count {
name: String,
data_type: DataType,
nullable: bool,
expr: Arc<dyn PhysicalExpr>,
}
impl Count {
pub fn new(
expr: Arc<dyn PhysicalExpr>,
name: impl Into<String>,
data_type: DataType,
) -> Self {
Self {
name: name.into(),
expr,
data_type,
nullable: true,
}
}
}
impl AggregateExpr for Count {
fn as_any(&self) -> &dyn Any {
self
}
fn field(&self) -> Result<Field> {
Ok(Field::new(
&self.name,
self.data_type.clone(),
self.nullable,
))
}
fn state_fields(&self) -> Result<Vec<Field>> {
Ok(vec![Field::new(
format_state_name(&self.name, "count"),
self.data_type.clone(),
true,
)])
}
fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
vec![self.expr.clone()]
}
fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
Ok(Box::new(CountAccumulator::new()))
}
fn name(&self) -> &str {
&self.name
}
fn row_accumulator_supported(&self) -> bool {
true
}
fn supports_bounded_execution(&self) -> bool {
true
}
fn create_row_accumulator(
&self,
start_index: usize,
) -> Result<Box<dyn RowAccumulator>> {
Ok(Box::new(CountRowAccumulator::new(start_index)))
}
fn reverse_expr(&self) -> Option<Arc<dyn AggregateExpr>> {
Some(Arc::new(self.clone()))
}
fn create_sliding_accumulator(&self) -> Result<Box<dyn Accumulator>> {
Ok(Box::new(CountAccumulator::new()))
}
}
#[derive(Debug)]
struct CountAccumulator {
count: i64,
}
impl CountAccumulator {
pub fn new() -> Self {
Self { count: 0 }
}
}
impl Accumulator for CountAccumulator {
fn state(&self) -> Result<Vec<ScalarValue>> {
Ok(vec![ScalarValue::Int64(Some(self.count))])
}
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
let array = &values[0];
self.count += (array.len() - array.null_count()) as i64;
Ok(())
}
fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
let array = &values[0];
self.count -= (array.len() - array.null_count()) as i64;
Ok(())
}
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
let counts = downcast_value!(states[0], Int64Array);
let delta = &compute::sum(counts);
if let Some(d) = delta {
self.count += *d;
}
Ok(())
}
fn evaluate(&self) -> Result<ScalarValue> {
Ok(ScalarValue::Int64(Some(self.count)))
}
fn size(&self) -> usize {
std::mem::size_of_val(self)
}
}
#[derive(Debug)]
struct CountRowAccumulator {
state_index: usize,
}
impl CountRowAccumulator {
pub fn new(index: usize) -> Self {
Self { state_index: index }
}
}
impl RowAccumulator for CountRowAccumulator {
fn update_batch(
&mut self,
values: &[ArrayRef],
accessor: &mut RowAccessor,
) -> Result<()> {
let array = &values[0];
let delta = (array.len() - array.null_count()) as u64;
accessor.add_u64(self.state_index, delta);
Ok(())
}
fn merge_batch(
&mut self,
states: &[ArrayRef],
accessor: &mut RowAccessor,
) -> Result<()> {
let counts = downcast_value!(states[0], Int64Array);
let delta = &compute::sum(counts);
if let Some(d) = delta {
accessor.add_i64(self.state_index, *d);
}
Ok(())
}
fn evaluate(&self, accessor: &RowAccessor) -> Result<ScalarValue> {
Ok(ScalarValue::Int64(Some(
accessor.get_u64_opt(self.state_index()).unwrap_or(0) as i64,
)))
}
#[inline(always)]
fn state_index(&self) -> usize {
self.state_index
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::expressions::col;
use crate::expressions::tests::aggregate;
use crate::generic_test_op;
use arrow::record_batch::RecordBatch;
use arrow::{array::*, datatypes::*};
use datafusion_common::Result;
#[test]
fn count_elements() -> Result<()> {
let a: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
generic_test_op!(a, DataType::Int32, Count, ScalarValue::from(5i64))
}
#[test]
fn count_with_nulls() -> Result<()> {
let a: ArrayRef = Arc::new(Int32Array::from(vec![
Some(1),
Some(2),
None,
None,
Some(3),
None,
]));
generic_test_op!(a, DataType::Int32, Count, ScalarValue::from(3i64))
}
#[test]
fn count_all_nulls() -> Result<()> {
let a: ArrayRef = Arc::new(BooleanArray::from(vec![
None, None, None, None, None, None, None, None,
]));
generic_test_op!(a, DataType::Boolean, Count, ScalarValue::from(0i64))
}
#[test]
fn count_empty() -> Result<()> {
let a: Vec<bool> = vec![];
let a: ArrayRef = Arc::new(BooleanArray::from(a));
generic_test_op!(a, DataType::Boolean, Count, ScalarValue::from(0i64))
}
#[test]
fn count_utf8() -> Result<()> {
let a: ArrayRef =
Arc::new(StringArray::from(vec!["a", "bb", "ccc", "dddd", "ad"]));
generic_test_op!(a, DataType::Utf8, Count, ScalarValue::from(5i64))
}
#[test]
fn count_large_utf8() -> Result<()> {
let a: ArrayRef =
Arc::new(LargeStringArray::from(vec!["a", "bb", "ccc", "dddd", "ad"]));
generic_test_op!(a, DataType::LargeUtf8, Count, ScalarValue::from(5i64))
}
}