You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "alamb (via GitHub)" <gi...@apache.org> on 2023/06/20 11:42:54 UTC
[GitHub] [arrow-datafusion] alamb opened a new pull request, #6729: Update documentation for creating User Defined Aggregates (AggregateUDF)
alamb opened a new pull request, #6729:
URL: https://github.com/apache/arrow-datafusion/pull/6729
# Which issue does this PR close?
related to https://github.com/apache/arrow-datafusion/issues/6611
# Rationale for this change
@stuartcarnie had some questions about how this API should work, and so I wanted to encode the answers into documentation for others as well
# What changes are included in this PR?
1. Update docs for `AggregateUDF`
2. Update docs for `Accumulator`
3. Rename `AccumulatorFunctionImplementation` to `AccumulatorFactoryFunction` to better describe what it does
# Are these changes tested?
Yes (existing tests + doc tests)
# Are there any user-facing changes?
1. Better docs
2. Different type alias name (`AccumulatorFunctionImplementation` to `AccumulatorFactoryFunction`)
--
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
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #6729: Update documentation for creating User Defined Aggregates (AggregateUDF)
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #6729:
URL: https://github.com/apache/arrow-datafusion/pull/6729#discussion_r1235168106
##########
datafusion/core/src/lib.rs:
##########
@@ -17,9 +17,9 @@
#![warn(missing_docs, clippy::needless_borrow)]
//! [DataFusion] is an extensible query engine written in Rust that
-//! uses [Apache Arrow] as its in-memory format. DataFusion's [use
-//! cases] include building very fast database and analytic systems,
-//! customized to particular workloads.
+//! uses [Apache Arrow] as its in-memory format. DataFusion's many [use
Review Comment:
this was a drive by cleanup as I pretended to be a new user navigating to the AggregateUDF page
##########
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:
This was just misleading, so I changed the name
##########
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.
Review Comment:
The trait is the same -- I did reorderd the methods to be better grouped together by use, but the actual methods are the same
--
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
[GitHub] [arrow-datafusion] stuartcarnie commented on a diff in pull request #6729: Update documentation for creating User Defined Aggregates (AggregateUDF)
Posted by "stuartcarnie (via GitHub)" <gi...@apache.org>.
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
[GitHub] [arrow-datafusion] alamb merged pull request #6729: Update documentation for creating User Defined Aggregates (AggregateUDF)
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #6729:
URL: https://github.com/apache/arrow-datafusion/pull/6729
--
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