Struct GroupsAccumulatorAdapter

pub struct GroupsAccumulatorAdapter { /* private fields */ }

An adapter that implements GroupsAccumulator for any Accumulator

While Accumulator are simpler to implement and can support more general calculations (like retractable window functions), they are not as fast as a specialized GroupsAccumulator. This interface bridges the gap so the group by operator only operates in terms of Accumulator.

Internally, this adapter creates a new Accumulator for each group which stores the state for that group. This both requires an allocation for each Accumulator, internal indices, as well as whatever internal allocations the Accumulator itself requires.

For example, a MinAccumulator that computes the minimum string value with a ScalarValue::Utf8. That will require at least two allocations per group (one for the MinAccumulator and one for the ScalarValue::Utf8).

                      ┌─────────────────────────────────┐
                      │MinAccumulator {                 │
               ┌─────▶│ min: ScalarValue::Utf8("A")     │───────┐
               │      │}                                │       │
               │      └─────────────────────────────────┘       └───────▶   "A"
   ┌─────┐     │      ┌─────────────────────────────────┐
   │  0  │─────┘      │MinAccumulator {                 │
   ├─────┤     ┌─────▶│ min: ScalarValue::Utf8("Z")     │───────────────▶   "Z"
   │  1  │─────┘      │}                                │
   └─────┘            └─────────────────────────────────┘                   ...
     ...                 ...
   ┌─────┐            ┌────────────────────────────────┐
   │ N-2 │            │MinAccumulator {                │
   ├─────┤            │  min: ScalarValue::Utf8("A")   │────────────────▶   "A"
   │ N-1 │─────┐      │}                               │
   └─────┘     │      └────────────────────────────────┘
               │      ┌────────────────────────────────┐        ┌───────▶   "Q"
               │      │MinAccumulator {                │        │
               └─────▶│  min: ScalarValue::Utf8("Q")   │────────┘
                      │}                               │
                      └────────────────────────────────┘


 Logical group         Current Min/Max value for that group stored
    number             as a ScalarValue which points to an
                       indivdually allocated String

Optimizations

The adapter minimizes the number of calls to Accumulator::update_batch by first collecting the input rows for each group into a contiguous array using compute::take

Implementations

impl GroupsAccumulatorAdapter

pub fn new<F>(factory: F) -> GroupsAccumulatorAdapter

where F: Fn() -> Result<Box<dyn Accumulator>, DataFusionError> + Send + 'static,

Create a new adapter that will create a new Accumulator for each group, using the specified factory function

Trait Implementations

impl GroupsAccumulator for GroupsAccumulatorAdapter

fn update_batch( &mut self, values: &[Arc<dyn Array>], group_indices: &[usize], opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<(), DataFusionError>

Updates the accumulator’s state from its arguments, encoded as a vector of ArrayRefs.

fn evaluate( &mut self, emit_to: EmitTo, ) -> Result<Arc<dyn Array>, DataFusionError>

Returns the final aggregate value for each group as a single RecordBatch, resetting the internal state.

fn state( &mut self, emit_to: EmitTo, ) -> Result<Vec<Arc<dyn Array>>, DataFusionError>

Returns the intermediate aggregate state for this accumulator, used for multi-phase grouping, resetting its internal state.

fn merge_batch( &mut self, values: &[Arc<dyn Array>], group_indices: &[usize], opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<(), DataFusionError>

Merges intermediate state (the output from Self::state) into this accumulator’s current state.

fn size(&self) -> usize

Amount of memory used to store the state of this accumulator, in bytes.

fn convert_to_state( &self, values: &[Arc<dyn Array>], opt_filter: Option<&BooleanArray>, ) -> Result<Vec<Arc<dyn Array>>, DataFusionError>

Converts an input batch directly to the intermediate aggregate state.

fn supports_convert_to_state(&self) -> bool

Returns true if Self::convert_to_state is implemented to support intermediate aggregate state conversion.