You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ji...@apache.org on 2022/01/16 05:25:14 UTC
[arrow-datafusion] branch remove-update-merge created (now 1361197)
This is an automated email from the ASF dual-hosted git repository.
jiayuliu pushed a change to branch remove-update-merge
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git.
at 1361197 remove update and merge
This branch includes the following new commits:
new 1361197 remove update and merge
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[arrow-datafusion] 01/01: remove update and merge
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiayuliu pushed a commit to branch remove-update-merge
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git
commit 136119749e0f8ddcb88393452c5dad29a7d541ec
Author: Jiayu Liu <ji...@hey.com>
AuthorDate: Sun Jan 16 12:38:41 2022 +0800
remove update and merge
---
datafusion-examples/examples/simple_udaf.rs | 4 +-
datafusion/src/datasource/file_format/parquet.rs | 48 +++++++++++++++-----
datafusion/src/datasource/mod.rs | 4 +-
.../src/physical_plan/distinct_expressions.rs | 14 +++---
.../physical_plan/expressions/approx_distinct.rs | 17 --------
.../src/physical_plan/expressions/array_agg.rs | 7 +--
.../src/physical_plan/expressions/average.rs | 8 ----
datafusion/src/physical_plan/expressions/count.rs | 8 ----
.../src/physical_plan/expressions/covariance.rs | 8 ----
.../src/physical_plan/expressions/min_max.rs | 20 ---------
datafusion/src/physical_plan/expressions/stddev.rs | 8 ----
datafusion/src/physical_plan/expressions/sum.rs | 8 ----
.../src/physical_plan/expressions/variance.rs | 8 ----
datafusion/src/physical_plan/mod.rs | 51 ++--------------------
14 files changed, 56 insertions(+), 157 deletions(-)
diff --git a/datafusion-examples/examples/simple_udaf.rs b/datafusion-examples/examples/simple_udaf.rs
index 5a0e814..44e4907 100644
--- a/datafusion-examples/examples/simple_udaf.rs
+++ b/datafusion-examples/examples/simple_udaf.rs
@@ -82,7 +82,7 @@ impl Accumulator for GeometricMean {
// this function receives one entry per argument of this accumulator.
// DataFusion calls this function on every row, and expects this function to update the accumulator's state.
- fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
// this is a one-argument UDAF, and thus we use `0`.
let value = &values[0];
match value {
@@ -101,7 +101,7 @@ impl Accumulator for GeometricMean {
// this function receives states from other accumulators (Vec<ScalarValue>)
// and updates the accumulator.
- fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
let prod = &states[0];
let n = &states[1];
match (prod, n) {
diff --git a/datafusion/src/datasource/file_format/parquet.rs b/datafusion/src/datasource/file_format/parquet.rs
index 608795a..8a8787a 100644
--- a/datafusion/src/datasource/file_format/parquet.rs
+++ b/datafusion/src/datasource/file_format/parquet.rs
@@ -36,6 +36,9 @@ use parquet::file::statistics::Statistics as ParquetStatistics;
use super::FileFormat;
use super::PhysicalPlanConfig;
+use crate::arrow::array::{
+ BooleanArray, Float32Array, Float64Array, Int32Array, Int64Array,
+};
use crate::arrow::datatypes::{DataType, Field};
use crate::datasource::object_store::{ObjectReader, ObjectReaderStream};
use crate::datasource::{create_max_min_accs, get_col_stats};
@@ -47,7 +50,6 @@ use crate::physical_plan::expressions::{MaxAccumulator, MinAccumulator};
use crate::physical_plan::file_format::ParquetExec;
use crate::physical_plan::ExecutionPlan;
use crate::physical_plan::{Accumulator, Statistics};
-use crate::scalar::ScalarValue;
/// The default file exetension of parquet files
pub const DEFAULT_PARQUET_EXTENSION: &str = ".parquet";
@@ -132,7 +134,9 @@ fn summarize_min_max(
if let DataType::Boolean = fields[i].data_type() {
if s.has_min_max_set() {
if let Some(max_value) = &mut max_values[i] {
- match max_value.update(&[ScalarValue::Boolean(Some(*s.max()))]) {
+ match max_value.update_batch(&[Arc::new(BooleanArray::from(
+ vec![Some(*s.max())],
+ ))]) {
Ok(_) => {}
Err(_) => {
max_values[i] = None;
@@ -140,7 +144,9 @@ fn summarize_min_max(
}
}
if let Some(min_value) = &mut min_values[i] {
- match min_value.update(&[ScalarValue::Boolean(Some(*s.min()))]) {
+ match min_value.update_batch(&[Arc::new(BooleanArray::from(
+ vec![Some(*s.min())],
+ ))]) {
Ok(_) => {}
Err(_) => {
min_values[i] = None;
@@ -154,7 +160,10 @@ fn summarize_min_max(
if let DataType::Int32 = fields[i].data_type() {
if s.has_min_max_set() {
if let Some(max_value) = &mut max_values[i] {
- match max_value.update(&[ScalarValue::Int32(Some(*s.max()))]) {
+ match max_value.update_batch(&[Arc::new(Int32Array::from_value(
+ *s.max(),
+ 1,
+ ))]) {
Ok(_) => {}
Err(_) => {
max_values[i] = None;
@@ -162,7 +171,10 @@ fn summarize_min_max(
}
}
if let Some(min_value) = &mut min_values[i] {
- match min_value.update(&[ScalarValue::Int32(Some(*s.min()))]) {
+ match min_value.update_batch(&[Arc::new(Int32Array::from_value(
+ *s.min(),
+ 1,
+ ))]) {
Ok(_) => {}
Err(_) => {
min_values[i] = None;
@@ -176,7 +188,10 @@ fn summarize_min_max(
if let DataType::Int64 = fields[i].data_type() {
if s.has_min_max_set() {
if let Some(max_value) = &mut max_values[i] {
- match max_value.update(&[ScalarValue::Int64(Some(*s.max()))]) {
+ match max_value.update_batch(&[Arc::new(Int64Array::from_value(
+ *s.min(),
+ 1,
+ ))]) {
Ok(_) => {}
Err(_) => {
max_values[i] = None;
@@ -184,7 +199,10 @@ fn summarize_min_max(
}
}
if let Some(min_value) = &mut min_values[i] {
- match min_value.update(&[ScalarValue::Int64(Some(*s.min()))]) {
+ match min_value.update_batch(&[Arc::new(Int64Array::from_value(
+ *s.min(),
+ 1,
+ ))]) {
Ok(_) => {}
Err(_) => {
min_values[i] = None;
@@ -198,7 +216,9 @@ fn summarize_min_max(
if let DataType::Float32 = fields[i].data_type() {
if s.has_min_max_set() {
if let Some(max_value) = &mut max_values[i] {
- match max_value.update(&[ScalarValue::Float32(Some(*s.max()))]) {
+ match max_value.update_batch(&[Arc::new(Float32Array::from(
+ vec![Some(*s.max())],
+ ))]) {
Ok(_) => {}
Err(_) => {
max_values[i] = None;
@@ -206,7 +226,9 @@ fn summarize_min_max(
}
}
if let Some(min_value) = &mut min_values[i] {
- match min_value.update(&[ScalarValue::Float32(Some(*s.min()))]) {
+ match min_value.update_batch(&[Arc::new(Float32Array::from(
+ vec![Some(*s.min())],
+ ))]) {
Ok(_) => {}
Err(_) => {
min_values[i] = None;
@@ -220,7 +242,9 @@ fn summarize_min_max(
if let DataType::Float64 = fields[i].data_type() {
if s.has_min_max_set() {
if let Some(max_value) = &mut max_values[i] {
- match max_value.update(&[ScalarValue::Float64(Some(*s.max()))]) {
+ match max_value.update_batch(&[Arc::new(Float64Array::from(
+ vec![Some(*s.max())],
+ ))]) {
Ok(_) => {}
Err(_) => {
max_values[i] = None;
@@ -228,7 +252,9 @@ fn summarize_min_max(
}
}
if let Some(min_value) = &mut min_values[i] {
- match min_value.update(&[ScalarValue::Float64(Some(*s.min()))]) {
+ match min_value.update_batch(&[Arc::new(Float64Array::from(
+ vec![Some(*s.min())],
+ ))]) {
Ok(_) => {}
Err(_) => {
min_values[i] = None;
diff --git a/datafusion/src/datasource/mod.rs b/datafusion/src/datasource/mod.rs
index 6e119f0..f0ed6e5 100644
--- a/datafusion/src/datasource/mod.rs
+++ b/datafusion/src/datasource/mod.rs
@@ -71,7 +71,7 @@ pub async fn get_statistics_with_limit(
if let Some(max_value) = &mut max_values[i] {
if let Some(file_max) = cs.max_value.clone() {
- match max_value.update(&[file_max]) {
+ match max_value.update_batch(&[file_max.to_array_of_size(1)]) {
Ok(_) => {}
Err(_) => {
max_values[i] = None;
@@ -82,7 +82,7 @@ pub async fn get_statistics_with_limit(
if let Some(min_value) = &mut min_values[i] {
if let Some(file_min) = cs.min_value.clone() {
- match min_value.update(&[file_min]) {
+ match min_value.update_batch(&[file_min.to_array_of_size(1)]) {
Ok(_) => {}
Err(_) => {
min_values[i] = None;
diff --git a/datafusion/src/physical_plan/distinct_expressions.rs b/datafusion/src/physical_plan/distinct_expressions.rs
index ae60253..a96bd40 100644
--- a/datafusion/src/physical_plan/distinct_expressions.rs
+++ b/datafusion/src/physical_plan/distinct_expressions.rs
@@ -17,16 +17,16 @@
//! Implementations for DISTINCT expressions, e.g. `COUNT(DISTINCT c)`
+use ahash::RandomState;
+use array::array::ArrayRef;
+use array::array::ArrayRef;
+use arrow::datatypes::{DataType, Field};
use std::any::Any;
+use std::collections::HashSet;
use std::fmt::Debug;
use std::hash::Hash;
use std::sync::Arc;
-use arrow::datatypes::{DataType, Field};
-
-use ahash::RandomState;
-use std::collections::HashSet;
-
use crate::error::{DataFusionError, Result};
use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
use crate::scalar::ScalarValue;
@@ -132,7 +132,7 @@ struct DistinctCountAccumulator {
}
impl Accumulator for DistinctCountAccumulator {
- fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
// If a row has a NULL, it is not included in the final count.
if !values.iter().any(|v| v.is_null()) {
self.values.insert(DistinctScalarValues(values.to_vec()));
@@ -141,7 +141,7 @@ impl Accumulator for DistinctCountAccumulator {
Ok(())
}
- fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
if states.is_empty() {
return Ok(());
}
diff --git a/datafusion/src/physical_plan/expressions/approx_distinct.rs b/datafusion/src/physical_plan/expressions/approx_distinct.rs
index ac7dcb3..9900780 100644
--- a/datafusion/src/physical_plan/expressions/approx_distinct.rs
+++ b/datafusion/src/physical_plan/expressions/approx_distinct.rs
@@ -217,23 +217,6 @@ impl<T: Hash> TryFrom<&ScalarValue> for HyperLogLog<T> {
macro_rules! default_accumulator_impl {
() => {
- fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
- self.update_batch(
- values
- .iter()
- .map(|s| s.to_array() as ArrayRef)
- .collect::<Vec<_>>()
- .as_slice(),
- )
- }
-
- fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
- assert_eq!(1, states.len(), "expect only 1 element in the states");
- let other = HyperLogLog::try_from(&states[0])?;
- self.hll.merge(&other);
- Ok(())
- }
-
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
assert_eq!(1, states.len(), "expect only 1 element in the states");
let binary_array = states[0].as_any().downcast_ref::<BinaryArray>().unwrap();
diff --git a/datafusion/src/physical_plan/expressions/array_agg.rs b/datafusion/src/physical_plan/expressions/array_agg.rs
index 3139c87..efacba8 100644
--- a/datafusion/src/physical_plan/expressions/array_agg.rs
+++ b/datafusion/src/physical_plan/expressions/array_agg.rs
@@ -21,6 +21,7 @@ use super::format_state_name;
use crate::error::Result;
use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
use crate::scalar::ScalarValue;
+use arrow::array::ArrayRef;
use arrow::datatypes::{DataType, Field};
use std::any::Any;
use std::sync::Arc;
@@ -116,14 +117,14 @@ impl Accumulator for ArrayAggAccumulator {
)])
}
- fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
let value = &values[0];
- self.array.push(value.clone());
+ self.array.extend(value.clone());
Ok(())
}
- fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
if states.is_empty() {
return Ok(());
};
diff --git a/datafusion/src/physical_plan/expressions/average.rs b/datafusion/src/physical_plan/expressions/average.rs
index a57c525..be1624c 100644
--- a/datafusion/src/physical_plan/expressions/average.rs
+++ b/datafusion/src/physical_plan/expressions/average.rs
@@ -171,10 +171,6 @@ impl Accumulator for AvgAccumulator {
Ok(vec![ScalarValue::from(self.count), self.sum.clone()])
}
- fn update(&mut self, _values: &[ScalarValue]) -> Result<()> {
- unimplemented!("update_batch is implemented instead");
- }
-
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
let values = &values[0];
@@ -183,10 +179,6 @@ impl Accumulator for AvgAccumulator {
Ok(())
}
- fn merge(&mut self, _states: &[ScalarValue]) -> Result<()> {
- unimplemented!("merge_batch is implemented instead");
- }
-
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
let counts = states[0].as_any().downcast_ref::<UInt64Array>().unwrap();
// counts are summed
diff --git a/datafusion/src/physical_plan/expressions/count.rs b/datafusion/src/physical_plan/expressions/count.rs
index edd9aea..7d4d144 100644
--- a/datafusion/src/physical_plan/expressions/count.rs
+++ b/datafusion/src/physical_plan/expressions/count.rs
@@ -112,14 +112,6 @@ impl Accumulator for CountAccumulator {
Ok(())
}
- fn update(&mut self, _values: &[ScalarValue]) -> Result<()> {
- unimplemented!("update_batch is implemented instead");
- }
-
- fn merge(&mut self, _states: &[ScalarValue]) -> Result<()> {
- unimplemented!("merge_batch is implemented instead");
- }
-
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
let counts = states[0].as_any().downcast_ref::<UInt64Array>().unwrap();
let delta = &compute::sum(counts);
diff --git a/datafusion/src/physical_plan/expressions/covariance.rs b/datafusion/src/physical_plan/expressions/covariance.rs
index 454dc43..06b6687 100644
--- a/datafusion/src/physical_plan/expressions/covariance.rs
+++ b/datafusion/src/physical_plan/expressions/covariance.rs
@@ -358,14 +358,6 @@ impl Accumulator for CovarianceAccumulator {
Ok(())
}
- fn update(&mut self, _values: &[ScalarValue]) -> Result<()> {
- unimplemented!("update_batch is implemented instead");
- }
-
- fn merge(&mut self, _states: &[ScalarValue]) -> Result<()> {
- unimplemented!("merge_batch is implemented instead");
- }
-
fn evaluate(&self) -> Result<ScalarValue> {
let count = match self.stats_type {
StatsType::Population => self.count,
diff --git a/datafusion/src/physical_plan/expressions/min_max.rs b/datafusion/src/physical_plan/expressions/min_max.rs
index 8f6cd45..05532f6 100644
--- a/datafusion/src/physical_plan/expressions/min_max.rs
+++ b/datafusion/src/physical_plan/expressions/min_max.rs
@@ -440,16 +440,6 @@ impl Accumulator for MaxAccumulator {
Ok(())
}
- fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
- let value = &values[0];
- self.max = max(&self.max, value)?;
- Ok(())
- }
-
- fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
- self.update(states)
- }
-
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
self.update_batch(states)
}
@@ -543,12 +533,6 @@ impl Accumulator for MinAccumulator {
Ok(vec![self.min.clone()])
}
- fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
- let value = &values[0];
- self.min = min(&self.min, value)?;
- Ok(())
- }
-
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
let values = &values[0];
let delta = &min_batch(values)?;
@@ -556,10 +540,6 @@ impl Accumulator for MinAccumulator {
Ok(())
}
- fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
- self.update(states)
- }
-
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
self.update_batch(states)
}
diff --git a/datafusion/src/physical_plan/expressions/stddev.rs b/datafusion/src/physical_plan/expressions/stddev.rs
index 191ac07..aeda37e 100644
--- a/datafusion/src/physical_plan/expressions/stddev.rs
+++ b/datafusion/src/physical_plan/expressions/stddev.rs
@@ -220,14 +220,6 @@ impl Accumulator for StddevAccumulator {
])
}
- fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
- self.variance.update(values)
- }
-
- fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
- self.variance.merge(states)
- }
-
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
self.variance.update_batch(values)
}
diff --git a/datafusion/src/physical_plan/expressions/sum.rs b/datafusion/src/physical_plan/expressions/sum.rs
index 61691c4..3582efc 100644
--- a/datafusion/src/physical_plan/expressions/sum.rs
+++ b/datafusion/src/physical_plan/expressions/sum.rs
@@ -353,20 +353,12 @@ impl Accumulator for SumAccumulator {
Ok(vec![self.sum.clone()])
}
- fn update(&mut self, _values: &[ScalarValue]) -> Result<()> {
- unimplemented!("update_batch is implemented instead");
- }
-
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
let values = &values[0];
self.sum = sum(&self.sum, &sum_batch(values)?)?;
Ok(())
}
- fn merge(&mut self, _states: &[ScalarValue]) -> Result<()> {
- unimplemented!("merge_batch is implemented instead");
- }
-
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
// sum(sum1, sum2, sum3, ...) = sum1 + sum2 + sum3 + ...
self.update_batch(states)
diff --git a/datafusion/src/physical_plan/expressions/variance.rs b/datafusion/src/physical_plan/expressions/variance.rs
index 117350c..215e5cf 100644
--- a/datafusion/src/physical_plan/expressions/variance.rs
+++ b/datafusion/src/physical_plan/expressions/variance.rs
@@ -302,14 +302,6 @@ impl Accumulator for VarianceAccumulator {
Ok(())
}
- fn update(&mut self, _values: &[ScalarValue]) -> Result<()> {
- unimplemented!("update_batch is implemented instead");
- }
-
- fn merge(&mut self, _states: &[ScalarValue]) -> Result<()> {
- unimplemented!("merge_batch is implemented instead");
- }
-
fn evaluate(&self) -> Result<ScalarValue> {
let count = match self.stats_type {
StatsType::Population => self.count,
diff --git a/datafusion/src/physical_plan/mod.rs b/datafusion/src/physical_plan/mod.rs
index 836d399..3f7f59e 100644
--- a/datafusion/src/physical_plan/mod.rs
+++ b/datafusion/src/physical_plan/mod.rs
@@ -566,9 +566,9 @@ pub trait WindowExpr: Send + Sync + Debug {
/// generically accumulates values.
///
/// An accumulator knows how to:
-/// * update its state from inputs via `update`
+/// * update its state from inputs via `update_batch`
/// * convert its internal state to a vector of scalar values
-/// * update its state from multiple accumulators' states via `merge`
+/// * update its state from multiple accumulators' states via `merge_batch`
/// * compute the final value from its internal state via `evaluate`
pub trait Accumulator: Send + Sync + Debug {
/// Returns the state of the accumulator at the end of the accumulation.
@@ -576,54 +576,11 @@ pub trait Accumulator: Send + Sync + Debug {
// of two values, sum and n.
fn state(&self) -> Result<Vec<ScalarValue>>;
- /// Updates the accumulator's state from a vector of scalars
- /// (called by default implementation of [`update_batch`]).
- ///
- /// Note: this method is often the simplest to implement and is
- /// backwards compatible to help to lower the barrier to entry for
- /// new users to write `Accumulators`
- ///
- /// You should always implement `update_batch` instead of this
- /// method for production aggregators or if you find yourself
- /// wanting to use mathematical kernels for [`ScalarValue`] such as
- /// `ScalarValue::add`, `ScalarValue::mul`, etc
- fn update(&mut self, values: &[ScalarValue]) -> Result<()>;
-
/// updates the accumulator's state from a vector of arrays.
- fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- if values.is_empty() {
- return Ok(());
- };
- (0..values[0].len()).try_for_each(|index| {
- let v = values
- .iter()
- .map(|array| ScalarValue::try_from_array(array, index))
- .collect::<Result<Vec<_>>>()?;
- self.update(&v)
- })
- }
-
- /// Updates the accumulator's state from a vector of scalars.
- /// (called by default implementation of [`merge`]).
- ///
- /// You should always implement `merge_batch` instead of this
- /// method for production aggregators. Please see notes on
- /// [`update`] for more detail and rationale.
- fn merge(&mut self, states: &[ScalarValue]) -> Result<()>;
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()>;
/// updates the accumulator's state from a vector of states.
- fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
- if states.is_empty() {
- return Ok(());
- };
- (0..states[0].len()).try_for_each(|index| {
- let v = states
- .iter()
- .map(|array| ScalarValue::try_from_array(array, index))
- .collect::<Result<Vec<_>>>()?;
- self.merge(&v)
- })
- }
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()>;
/// returns its value based on its current state.
fn evaluate(&self) -> Result<ScalarValue>;