zhuqi-lucas commented on code in PR #26357:
URL: https://github.com/apache/datafusion/pull/26357#discussion_r4320945186
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
@@ -832,6 +835,419 @@ impl GroupsAccumulator for ArrayAggGroupsAccumulator {
}
}
+/// Converts each input row into a one-element list, the state shape of
+/// `ARRAY_AGG`.
+///
+/// Filtered rows become null list entries, which `merge_batch` skips. With
+/// `ignore_nulls`, null values also become null list entries; without it they
+/// stay as `[NULL]` so `merge_batch` retains them.
+fn single_value_lists(
+ input: &ArrayRef,
+ opt_filter: Option<&BooleanArray>,
+ ignore_nulls: bool,
+ datatype: &DataType,
+) -> ArrayRef {
+ // Offsets are [0, 1, 2, ..., n].
+ let offsets = OffsetBuffer::from_repeated_length(1, input.len());
+ let filter_nulls = opt_filter.map(filter_to_nulls);
+ let nulls = if ignore_nulls {
+ let logical = input.logical_nulls();
+ NullBuffer::union(filter_nulls.as_ref(), logical.as_ref())
+ } else {
+ filter_nulls
+ };
+ let field = Arc::new(Field::new_list_field(datatype.clone(), true));
+ Arc::new(ListArray::new(field, offsets, Arc::clone(input), nulls))
+}
+
+/// [`GroupsAccumulator`] for `ARRAY_AGG(DISTINCT x)`, optionally with
+/// `ORDER BY x`.
+///
+/// Distinct values are encoded once with a [`RowConverter`] and stored in a
+/// single [`Rows`] buffer shared by all groups, so a value that appears in
many
+/// groups is held once. Each group records the values it has seen as
+/// `(group_index, value_index)` pairs, deduplicated in one hash set rather
than
+/// one set per group.
+#[derive(Debug)]
+struct DistinctArrayAggGroupsAccumulator {
Review Comment:
Thanks @kosiew! Added it back in 8758c098d: group counts (100 / 1000 /
8192), distinct cardinality (~25 / ~7800), and emitting all at once vs in 8
`EmitTo::First` chunks, each against the adapter.
It shows the cost you suspected: `retain_after` rebuilds the remaining state
on every partial emission, so with 100 groups and high cardinality, emitting in
8 chunks is 1.55 ms vs 0.98 ms for the adapter (emitting all at once is 2.6x
faster). Every other case is faster, up to 41x. Filed #26512 to make the
compaction amortized.
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
@@ -341,6 +332,33 @@ impl ArrayAggAccumulator {
/// Returns the mask of non-null rows in `values` when `ignore_nulls` is set
Review Comment:
Good catch, moved it back in 8758c098d.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
