Skip to content

Commit 6c850b7

Browse files
committed
Fix the avg accumulator
1 parent 560040a commit 6c850b7

1 file changed

Lines changed: 29 additions & 8 deletions

File tree

‎datafusion/functions-aggregate/src/average.rs‎

Lines changed: 29 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -653,8 +653,15 @@ impl Accumulator for AvgAccumulator {
653653
}
654654

655655
fn state(&mut self) -> Result<Vec<ScalarValue>> {
656+
// With no non-NULL values both fields are NULL, matching
657+
// `AvgGroupsAccumulator::state`
658+
let (count, sum) = if self.count == 0 {
659+
(None, None)
660+
} else {
661+
(Some(self.count), self.sum)
662+
};
656663
Ok(vec![
657-
ScalarValue::from(self.count),
664+
ScalarValue::UInt64(count),
658665
ScalarValue::Float64(self.sum),
659666
])
660667
}
@@ -833,9 +840,16 @@ where
833840
}
834841

835842
fn state(&mut self) -> Result<Vec<ScalarValue>> {
843+
// With no non-NULL values both fields are NULL, matching
844+
// `AvgGroupsAccumulator::state`
845+
let (count, sum) = if self.count == 0 {
846+
(None, None)
847+
} else {
848+
(Some(self.count), self.sum)
849+
};
836850
Ok(vec![
837-
ScalarValue::from(self.count),
838-
ScalarValue::new_primitive::<S>(self.sum, &self.sum_data_type)?,
851+
ScalarValue::UInt64(count),
852+
ScalarValue::new_primitive::<S>(sum, &self.sum_data_type)?,
839853
])
840854
}
841855

@@ -916,14 +930,21 @@ impl Accumulator for DurationAvgAccumulator {
916930
}
917931

918932
fn state(&mut self) -> Result<Vec<ScalarValue>> {
933+
// With no non-NULL values both fields are NULL, matching
934+
// `AvgGroupsAccumulator::state`
935+
let (count, sum) = if self.count == 0 {
936+
(None, None)
937+
} else {
938+
(Some(self.count), self.sum)
939+
};
919940
let duration_value = match self.time_unit {
920-
TimeUnit::Second => ScalarValue::DurationSecond(self.sum),
921-
TimeUnit::Millisecond => ScalarValue::DurationMillisecond(self.sum),
922-
TimeUnit::Microsecond => ScalarValue::DurationMicrosecond(self.sum),
923-
TimeUnit::Nanosecond => ScalarValue::DurationNanosecond(self.sum),
941+
TimeUnit::Second => ScalarValue::DurationSecond(sum),
942+
TimeUnit::Millisecond => ScalarValue::DurationMillisecond(sum),
943+
TimeUnit::Microsecond => ScalarValue::DurationMicrosecond(sum),
944+
TimeUnit::Nanosecond => ScalarValue::DurationNanosecond(sum),
924945
};
925946

926-
Ok(vec![ScalarValue::from(self.count), duration_value])
947+
Ok(vec![ScalarValue::UInt64(count), duration_value])
927948
}
928949

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

0 commit comments

Comments
 (0)