You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "stuartcarnie (via GitHub)" <gi...@apache.org> on 2023/06/20 22:57:52 UTC
[GitHub] [arrow-datafusion] stuartcarnie commented on a diff in pull request #6729: Update documentation for creating User Defined Aggregates (AggregateUDF)
stuartcarnie commented on code in PR #6729:
URL: https://github.com/apache/arrow-datafusion/pull/6729#discussion_r1235982249
##########
datafusion/expr/src/function.rs:
##########
@@ -42,7 +42,7 @@ pub type ReturnTypeFunction =
/// Factory that returns an accumulator for the given aggregate, given
/// its return datatype.
-pub type AccumulatorFunctionImplementation =
+pub type AccumulatorFactoryFunction =
Review Comment:
Totally agree – this name is much clearer
##########
datafusion/expr/src/accumulator.rs:
##########
@@ -21,49 +21,161 @@ use arrow::array::ArrayRef;
use datafusion_common::{DataFusionError, Result, ScalarValue};
use std::fmt::Debug;
-/// Accumulates an aggregate's state.
+/// Describes an aggregate functions's state.
///
-/// `Accumulator`s are stateful objects that lives throughout the
+/// `Accumulator`s are stateful objects that live throughout the
/// evaluation of multiple rows and aggregate multiple values together
/// into a final output aggregate.
///
/// An accumulator knows how to:
-/// * update its state from inputs via `update_batch`
-/// * (optionally) retract an update to its state from given inputs via `retract_batch`
-/// * convert its internal state to a vector of aggregate values
-/// * update its state from multiple accumulators' states via `merge_batch`
-/// * compute the final value from its internal state via `evaluate`
+/// * update its state from inputs via [`update_batch`]
+///
+/// * compute the final value from its internal state via [`evaluate`]
+///
+/// * retract an update to its state from given inputs via
+/// [`retract_batch`] (when used as a window aggregate [window
+/// function])
+///
+/// * convert its internal state to a vector of aggregate values via
+/// [`state`] and combine the state from multiple accumulators'
+/// via [`merge_batch`], as part of efficient multi-phase grouping.
+///
+/// [`update_batch`]: Self::update_batch
+/// [`retract_batch`]: Self::retract_batch
+/// [`state`]: Self::state
+/// [`evaluate`]: Self::evaluate
+/// [`merge_batch`]: Self::merge_batch
+/// [window function]: https://en.wikipedia.org/wiki/Window_function_(SQL)
pub trait Accumulator: Send + Sync + Debug {
- /// Returns the partial intermediate state of the accumulator. This
- /// partial state is serialied as `Arrays` and then combined with
- /// other partial states from different instances of this
- /// accumulator (that ran on different partitions, for
- /// example).
+ /// Updates the accumulator's state from its input.
///
- /// The state can be and often is a different type than the output
- /// type of the [`Accumulator`].
+ /// `values` contains the arguments to this aggregate function.
+ ///
+ /// For example, the `SUM` accumulator maintains a running sum,
+ /// and `update_batch` adds each of the input values to the
+ /// running sum.
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()>;
+
+ /// Returns the final aggregate value.
+ ///
+ /// For example, the `SUM` accumulator maintains a running sum,
+ /// and `evaluate` will produce that running sum as its output.
+ fn evaluate(&self) -> Result<ScalarValue>;
+
+ /// Returns the allocated size required for this accumulator, in
+ /// bytes, including `Self`.
///
- /// See [`Self::merge_batch`] for more details on the merging process.
+ /// This value is used to calculate the memory used during
+ /// execution so DataFusion can stay within its allotted limit.
+ ///
+ /// "Allocated" means that for internal containers such as `Vec`,
+ /// the `capacity` should be used not the `len`.
+ fn size(&self) -> usize;
+
+ /// Returns the intermediate state of the accumulator.
+ ///
+ /// Intermediate state is used for "multi-phase" grouping in
+ /// DataFusion, where an aggregate is computed in parallel with
+ /// multiple `Accumulator` instances, as illustrated below:
+ ///
+ /// # MultiPhase Grouping
+ ///
+ /// ```text
+ /// ▲
+ /// │ evaluate() is called to
+ /// │ produce the final aggregate
+ /// │ value per group
+ /// │
+ /// ┌─────────────────────────┐
+ /// │GroupBy │
+ /// │(AggregateMode::Final) │ state() is called for each
+ /// │ │ group and the resulting
+ /// └─────────────────────────┘ RecordBatches passed to the
+ /// ▲
+ /// │
+ /// ┌────────────────┴───────────────┐
+ /// │ │
+ /// │ │
+ /// ┌─────────────────────────┐ ┌─────────────────────────┐
+ /// │ GroubyBy │ │ GroubyBy │
+ /// │(AggregateMode::Partial) │ │(AggregateMode::Partial) │
+ /// └─────────────────────────┘ └────────────▲────────────┘
+ /// ▲ │
+ /// │ │ update_batch() is called for
+ /// │ │ each input RecordBatch
+ /// .─────────. .─────────.
+ /// ,─' '─. ,─' '─.
+ /// ; Input : ; Input :
+ /// : Partition 0 ; : Partition 1 ;
+ /// ╲ ╱ ╲ ╱
+ /// '─. ,─' '─. ,─'
+ /// `───────' `───────'
+ /// ```
+ ///
+ /// The partial state is serialied as `Arrays` and then combined
+ /// with other partial states from different instances of this
+ /// Accumulator (that ran on different partitions, for example).
+ ///
+ /// The state can be and often is a different type than the output
+ /// type of the [`Accumulator`] and needs different merge
+ /// operations (for example, the partial state for `COUNT` needs
+ /// to be summed together)
///
/// Some accumulators can return multiple values for their
/// intermediate states. For example average, tracks `sum` and
/// `n`, and this function should return
/// a vector of two values, sum and n.
///
- /// `ScalarValue::List` can also be used to pass multiple values
- /// if the number of intermediate values is not known at planning
- /// time (e.g. median)
+ /// Note that [`ScalarValue::List`] can be used to pass multiple
+ /// values if the number of intermediate values is not known at
+ /// planning time (e.g. for `MEDIAN`)
fn state(&self) -> Result<Vec<ScalarValue>>;
- /// Updates the accumulator's state from a vector of arrays.
- fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()>;
+ /// Updates the accumulator's state from an `Array` containing one
+ /// or more intermediate values.
+ ///
+ /// For some aggregates (such as `SUM`), merge_batch is the same
+ /// as `update_batch`, but for some aggregrates (such as `COUNT`)
+ /// the operations differ. See [`Self::state`] for more details on how
+ /// state is used and merged.
+ ///
+ /// The `states` array passed was formed by concatenating the
+ /// results of calling [`Self::state`] on zero or more other
+ /// `Accumulator` instances.
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()>;
- /// Retracts an update (caused by the given inputs) to
+ /// Retracts (removed) an update (caused by the given inputs) to
/// accumulator's state.
///
/// This is the inverse operation of [`Self::update_batch`] and is used
- /// to incrementally calculate window aggregates where the OVER
+ /// to incrementally calculate window aggregates where the `OVER`
/// clause defines a bounded window.
+ ///
+ /// # Example
+ ///
+ /// For example, given the following input partition
+ ///
+ /// ```text
+ /// │ current │
+ /// window
+ /// │ │
+ /// ┌────┬────┬────┬────┬────┬────┬────┬────┬────┐
+ /// Input │ A │ B │ C │ D │ E │ F │ G │ H │ I │
+ /// partition └────┴────┴────┴────┼────┴────┴────┴────┼────┘
+ ///
+ /// │ next │
+ /// window
+ /// ```
+ ///
+ /// First, [`Self::evaluate`] will be called to produce the output
+ /// for the current window.
+ ///
+ /// Then, to advance to the next window:
+ ///
+ /// First, [`Self::retract_batch`] will be called with the values
+ /// that are leaving the window, `[B, C, D]` and then
+ /// [`Self::update_batch`] will be called with the values that are
+ /// entering the window, `[F, G, H]`.
Review Comment:
Very clear explanation 💯
##########
datafusion/expr/src/accumulator.rs:
##########
@@ -21,49 +21,161 @@ use arrow::array::ArrayRef;
use datafusion_common::{DataFusionError, Result, ScalarValue};
use std::fmt::Debug;
-/// Accumulates an aggregate's state.
+/// Describes an aggregate functions's state.
///
-/// `Accumulator`s are stateful objects that lives throughout the
+/// `Accumulator`s are stateful objects that live throughout the
/// evaluation of multiple rows and aggregate multiple values together
/// into a final output aggregate.
///
/// An accumulator knows how to:
-/// * update its state from inputs via `update_batch`
-/// * (optionally) retract an update to its state from given inputs via `retract_batch`
-/// * convert its internal state to a vector of aggregate values
-/// * update its state from multiple accumulators' states via `merge_batch`
-/// * compute the final value from its internal state via `evaluate`
+/// * update its state from inputs via [`update_batch`]
+///
+/// * compute the final value from its internal state via [`evaluate`]
+///
+/// * retract an update to its state from given inputs via
+/// [`retract_batch`] (when used as a window aggregate [window
+/// function])
+///
+/// * convert its internal state to a vector of aggregate values via
+/// [`state`] and combine the state from multiple accumulators'
+/// via [`merge_batch`], as part of efficient multi-phase grouping.
+///
+/// [`update_batch`]: Self::update_batch
+/// [`retract_batch`]: Self::retract_batch
+/// [`state`]: Self::state
+/// [`evaluate`]: Self::evaluate
+/// [`merge_batch`]: Self::merge_batch
+/// [window function]: https://en.wikipedia.org/wiki/Window_function_(SQL)
pub trait Accumulator: Send + Sync + Debug {
- /// Returns the partial intermediate state of the accumulator. This
- /// partial state is serialied as `Arrays` and then combined with
- /// other partial states from different instances of this
- /// accumulator (that ran on different partitions, for
- /// example).
+ /// Updates the accumulator's state from its input.
///
- /// The state can be and often is a different type than the output
- /// type of the [`Accumulator`].
+ /// `values` contains the arguments to this aggregate function.
+ ///
+ /// For example, the `SUM` accumulator maintains a running sum,
+ /// and `update_batch` adds each of the input values to the
+ /// running sum.
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()>;
+
+ /// Returns the final aggregate value.
+ ///
+ /// For example, the `SUM` accumulator maintains a running sum,
+ /// and `evaluate` will produce that running sum as its output.
+ fn evaluate(&self) -> Result<ScalarValue>;
+
+ /// Returns the allocated size required for this accumulator, in
+ /// bytes, including `Self`.
///
- /// See [`Self::merge_batch`] for more details on the merging process.
+ /// This value is used to calculate the memory used during
+ /// execution so DataFusion can stay within its allotted limit.
+ ///
+ /// "Allocated" means that for internal containers such as `Vec`,
+ /// the `capacity` should be used not the `len`.
+ fn size(&self) -> usize;
+
+ /// Returns the intermediate state of the accumulator.
+ ///
+ /// Intermediate state is used for "multi-phase" grouping in
+ /// DataFusion, where an aggregate is computed in parallel with
+ /// multiple `Accumulator` instances, as illustrated below:
+ ///
+ /// # MultiPhase Grouping
+ ///
+ /// ```text
+ /// ▲
+ /// │ evaluate() is called to
+ /// │ produce the final aggregate
+ /// │ value per group
+ /// │
+ /// ┌─────────────────────────┐
+ /// │GroupBy │
+ /// │(AggregateMode::Final) │ state() is called for each
+ /// │ │ group and the resulting
+ /// └─────────────────────────┘ RecordBatches passed to the
+ /// ▲
+ /// │
+ /// ┌────────────────┴───────────────┐
+ /// │ │
+ /// │ │
+ /// ┌─────────────────────────┐ ┌─────────────────────────┐
+ /// │ GroubyBy │ │ GroubyBy │
+ /// │(AggregateMode::Partial) │ │(AggregateMode::Partial) │
+ /// └─────────────────────────┘ └────────────▲────────────┘
+ /// ▲ │
+ /// │ │ update_batch() is called for
+ /// │ │ each input RecordBatch
+ /// .─────────. .─────────.
+ /// ,─' '─. ,─' '─.
+ /// ; Input : ; Input :
+ /// : Partition 0 ; : Partition 1 ;
+ /// ╲ ╱ ╲ ╱
+ /// '─. ,─' '─. ,─'
+ /// `───────' `───────'
+ /// ```
+ ///
+ /// The partial state is serialied as `Arrays` and then combined
+ /// with other partial states from different instances of this
+ /// Accumulator (that ran on different partitions, for example).
+ ///
+ /// The state can be and often is a different type than the output
+ /// type of the [`Accumulator`] and needs different merge
+ /// operations (for example, the partial state for `COUNT` needs
+ /// to be summed together)
///
/// Some accumulators can return multiple values for their
/// intermediate states. For example average, tracks `sum` and
/// `n`, and this function should return
/// a vector of two values, sum and n.
///
- /// `ScalarValue::List` can also be used to pass multiple values
- /// if the number of intermediate values is not known at planning
- /// time (e.g. median)
+ /// Note that [`ScalarValue::List`] can be used to pass multiple
+ /// values if the number of intermediate values is not known at
+ /// planning time (e.g. for `MEDIAN`)
fn state(&self) -> Result<Vec<ScalarValue>>;
Review Comment:
This updated documentation is fantastic, thank you @alamb
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org