var(distinct) support

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
complete description later.

Describe the solution you'd like
A clear and concise description of what you want to happen.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

Additional context
Add any other context or screenshots about the feature request here.

Checklist of things needed to be implemented. For reference existing implementation:

https://github.com/apache/datafusion/blob/2a08013af3ccf703bee202c959b40bb0d35bdea1/datafusion/functions-aggregate/src/variance.rs

Distinct accumulator for variance, see example for median:

datafusion/datafusion/functions-aggregate/src/median.rs

Lines 511 to 542 in 2a08013

    
           #[derive(Debug)] 
        
           struct DistinctMedianAccumulator<T: ArrowNumericType> { 
        
               distinct_values: GenericDistinctBuffer<T>, 
        
               data_type: DataType, 
        
           } 
        
           impl<T: ArrowNumericType + Debug> Accumulator for DistinctMedianAccumulator<T> { 
        
               fn state(&mut self) -> Result<Vec<ScalarValue>> { 
        
                   self.distinct_values.state() 
        
               } 
        
               fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> { 
        
                   self.distinct_values.update_batch(values) 
        
               } 
        
               fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { 
        
                   self.distinct_values.merge_batch(states) 
        
               } 
        
               fn evaluate(&mut self) -> Result<ScalarValue> { 
        
                   let d = std::mem::take(&mut self.distinct_values.values) 
        
                       .into_iter() 
        
                       .map(|v| v.0) 
        
                       .collect::<Vec<_>>(); 
        
                   let median = calculate_median::<T>(d); 
        
                   ScalarValue::new_primitive::<T>(median, &self.data_type) 
        
               } 
        
               fn size(&self) -> usize { 
        
                   size_of_val(self) + self.distinct_values.size() 
        
               } 
        
           }

Make sure to wire distinct accumulator into the UDAF:

datafusion/datafusion/functions-aggregate/src/median.rs

Lines 151 to 184 in 2a08013

    
           fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> { 
        
               macro_rules! helper { 
        
                   ($t:ty, $dt:expr) => { 
        
                       if acc_args.is_distinct { 
        
                           Ok(Box::new(DistinctMedianAccumulator::<$t> { 
        
                               data_type: $dt.clone(), 
        
                               distinct_values: GenericDistinctBuffer::new($dt), 
        
                           })) 
        
                       } else { 
        
                           Ok(Box::new(MedianAccumulator::<$t> { 
        
                               data_type: $dt.clone(), 
        
                               all_values: vec![], 
        
                           })) 
        
                       } 
        
                   }; 
        
               } 
        
               let dt = acc_args.expr_fields[0].data_type().clone(); 
        
               downcast_integer! { 
        
                   dt => (helper, dt), 
        
                   DataType::Float16 => helper!(Float16Type, dt), 
        
                   DataType::Float32 => helper!(Float32Type, dt), 
        
                   DataType::Float64 => helper!(Float64Type, dt), 
        
                   DataType::Decimal32(_, _) => helper!(Decimal32Type, dt), 
        
                   DataType::Decimal64(_, _) => helper!(Decimal64Type, dt), 
        
                   DataType::Decimal128(_, _) => helper!(Decimal128Type, dt), 
        
                   DataType::Decimal256(_, _) => helper!(Decimal256Type, dt), 
        
                   _ => Err(DataFusionError::NotImplemented(format!( 
        
                       "MedianAccumulator not supported for {} with {}", 
        
                       acc_args.name, 
        
                       dt, 
        
                   ))), 
        
               } 
        
           }

datafusion/datafusion/functions-aggregate/src/median.rs

Lines 132 to 149 in 2a08013

    
           fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> { 
        
               //Intermediate state is a list of the elements we have collected so far 
        
               let field = Field::new_list_field(args.input_fields[0].data_type().clone(), true); 
        
               let state_name = if args.is_distinct { 
        
                   "distinct_median" 
        
               } else { 
        
                   "median" 
        
               }; 
        
               Ok(vec![ 
        
                   Field::new( 
        
                       format_state_name(args.name, state_name), 
        
                       DataType::List(Arc::new(field)), 
        
                       true, 
        
                   ) 
        
                   .into(), 
        
               ]) 
        
           }

Add tests here:

https://github.com/apache/datafusion/blob/2a08013af3ccf703bee202c959b40bb0d35bdea1/datafusion/sqllogictest/test_files/aggregate.slt

implement var distinct (#19706)

implement var distinct (apache#19706)

Navigation Menu

Repository navigation

`var(distinct)` support #2410

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	#[derive(Debug)]
	struct DistinctMedianAccumulator<T: ArrowNumericType> {
	distinct_values: GenericDistinctBuffer<T>,
	data_type: DataType,
	}

	impl<T: ArrowNumericType + Debug> Accumulator for DistinctMedianAccumulator<T> {
	fn state(&mut self) -> Result<Vec<ScalarValue>> {
	self.distinct_values.state()
	}

	fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
	self.distinct_values.update_batch(values)
	}

	fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
	self.distinct_values.merge_batch(states)
	}

	fn evaluate(&mut self) -> Result<ScalarValue> {
	let d = std::mem::take(&mut self.distinct_values.values)
	.into_iter()
	.map(\|v\| v.0)
	.collect::<Vec<_>>();
	let median = calculate_median::<T>(d);
	ScalarValue::new_primitive::<T>(median, &self.data_type)
	}

	fn size(&self) -> usize {
	size_of_val(self) + self.distinct_values.size()
	}
	}

	fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
	macro_rules! helper {
	($t:ty, $dt:expr) => {
	if acc_args.is_distinct {
	Ok(Box::new(DistinctMedianAccumulator::<$t> {
	data_type: $dt.clone(),
	distinct_values: GenericDistinctBuffer::new($dt),
	}))
	} else {
	Ok(Box::new(MedianAccumulator::<$t> {
	data_type: $dt.clone(),
	all_values: vec![],
	}))
	}
	};
	}

	let dt = acc_args.expr_fields[0].data_type().clone();
	downcast_integer! {
	dt => (helper, dt),
	DataType::Float16 => helper!(Float16Type, dt),
	DataType::Float32 => helper!(Float32Type, dt),
	DataType::Float64 => helper!(Float64Type, dt),
	DataType::Decimal32(_, _) => helper!(Decimal32Type, dt),
	DataType::Decimal64(_, _) => helper!(Decimal64Type, dt),
	DataType::Decimal128(_, _) => helper!(Decimal128Type, dt),
	DataType::Decimal256(_, _) => helper!(Decimal256Type, dt),
	_ => Err(DataFusionError::NotImplemented(format!(
	"MedianAccumulator not supported for {} with {}",
	acc_args.name,
	dt,
	))),
	}
	}

	fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
	//Intermediate state is a list of the elements we have collected so far
	let field = Field::new_list_field(args.input_fields[0].data_type().clone(), true);
	let state_name = if args.is_distinct {
	"distinct_median"
	} else {
	"median"
	};

	Ok(vec![
	Field::new(
	format_state_name(args.name, state_name),
	DataType::List(Arc::new(field)),
	true,
	)
	.into(),
	])
	}

var(distinct) support #2410

Description

Activity

Jefffrey commented on Dec 10, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

`var(distinct)` support #2410