singhpratech opened a new pull request, #1336:
URL: https://github.com/apache/arrow-go/pull/1336
### Rationale for this change
The compute package has no way to compute a summary value from an array.
`FuncScalarAgg` exists as
a function kind but nothing implements it: `exec` has a kernel type for
scalar and vector kernels
only, there is no aggregate function type, and `execInternal` returns
`ErrNotImplemented` for that
kind. Adding a single aggregate function therefore means adding the
framework first, which is what
this pull request does, together with the first two kernels.
The design was discussed on #1296 and the interface follows the changes
asked for there.
### What changes are included in this PR?
The framework, in `arrow/compute/exec`:
* `ScalarAggKernel`, following the C++ `ScalarAggregateKernel`: an init
function that creates a
state, `Consume` to fold one `ExecSpan` into it, `Merge` to combine two
states, `Finalize` to
produce the result, plus an optional `Cleanup` and an `Ordered` flag.
* `AggregateResult`, a small carrier which holds either an owned
`scalar.Scalar` or owned
`arrow.ArrayData`. `exec` cannot import `compute` and so cannot name a
`Datum`, but an aggregate
result is not always a scalar, so the carrier keeps the exported interface
general; the compute
executor boxes it into a `Datum`.
* `AggKernel`, the interface the executor consumes, and `MergeAll`, for a
caller which aggregated
partitions of the input separately.
The executor, in `arrow/compute`:
* `ScalarAggregateFunction`, `funcImpl[exec.ScalarAggKernel]` with exact
dispatch, its own
`SetDefaultOptions`, and `AddKernel`/`AddNewKernel` which reject a kernel
missing any of the four
lifecycle functions.
* `scalarAggExecutor`, reached from `execInternal` for `FuncScalarAgg`. It
iterates the input in
spans without promoting scalars to arrays, as the C++ `ScalarAggExecutor`
does, so a kernel sees
a scalar weighted by the length of the span; it consumes into a single
state and finalizes once.
An empty input still reaches finalize, which is what lets `min_count`
decide the result of an
aggregation over no values.
* State cleanup that runs exactly once: when `Execute` returns, whether it
finalized or failed
part way, and through the executor's `Clear` on the paths where `Execute`
never ran, such as a
failure in `Init`. The result finalize returned owns its own value and is
unaffected.
The kernels, in `arrow/compute/internal/kernels/aggregate_basic.go` and
`arrow/compute/scalar_aggregate.go`:
* `count`, with `CountOptions` and the modes `CountOnlyValid`,
`CountOnlyNull` and `CountAllRows`,
over any input type. Counting the valid or the null values of a run-end
encoded, dictionary or
union array needs a logical null count that an `ArraySpan` does not carry,
so those return
`ErrNotImplemented`; `CountAllRows` works for them because it never looks
at validity.
* `sum` over int8..int64 (accumulating into int64), uint8..uint64 and bool
(into uint64),
float32/float64 (into float64) and the null type (into int64), with the
C++ semantics for nulls,
`skip_nulls` and `min_count`, wrap-around on integer overflow, and the C++
pairwise summation for
floating point input.
* `ScalarAggregateOptions`, `DefaultScalarAggregateOptions`,
`DefaultCountOptions`, the `Count` and
`Sum` convenience wrappers, and registration of both option types for
deserialization.
Aggregates stay out of `exprs`, which rejects any function whose kind is not
`FuncScalar`
(`arrow/compute/exprs/exec.go`, in the `*expr.ScalarFunction` case),
matching C++ where
aggregations run through Acero rather than through expressions.
Not in this pull request: `mean`, `min_max`, `min`, `max`, `any` and `all`,
which follow in a
second one; and `product`, `count_distinct`, `first`/`last`, decimals, the
statistical aggregates
and the `hash_*` family, which needs a grouper.
### Are these changes tested?
Yes.
* Table-driven tests per function over every input type, with and without
nulls, all null, empty,
sliced, chunked and scalar input, under the default options,
`skip_nulls=false`, and `min_count`
above and below the number of valid values. The expected values were taken
from the C++
implementation through pyarrow for the same fixtures and options; all 91
of them agree exactly.
* A differential fuzz run of 14,033 cases across the twelve input types,
plain, sliced, chunked and
scalar, with every option combination, compared against pyarrow: 0
mismatches, 0 error
asymmetries, 0 leaked bytes.
* Three tests cover the framework rather than the kernels. Every aggregate
runs at chunk sizes 1 to
65 and has to agree with the single-span result, which catches a kernel
that writes a cached null
count back into the span the executor re-slices. A kernel whose result is
an array, and one whose
result is a scalar backed by an allocated buffer, check that the value
finalize returned outlives
the cleanup of the state which produced it, and that cleanup runs exactly
once on success, on
empty input, on cancellation, and on a consume, finalize or cleanup error.
And the executor is
driven directly with a batch of scalars whose logical length is greater
than one, which the
`CallFunction` path cannot produce.
* The pairwise summation is pinned to 100001 doubles from a fixed generator;
the test also computes
a naive left-to-right sum of the same fixture and asserts that it differs,
so the pinned value can
only be reached with the C++ summation order.
* `go build ./...`, `go vet ./arrow/compute/...`, `go test
./arrow/compute/...` with and without
`-tags assert`, `go test -race`, and golangci-lint at 0 issues.
### Are there any user-facing changes?
Yes, all additive. `arrow/compute` gains the `count` and `sum` functions in
the default registry,
the `Count` and `Sum` wrappers, `ScalarAggregateOptions`, `CountOptions` and
their default helpers,
and `ScalarAggregateFunction`; `arrow/compute/exec` gains `ScalarAggKernel`,
`AggregateResult`,
`AggKernel` and `MergeAll`. Nothing existing changes behaviour:
`execInternal` previously fell through to its default
branch, `ErrNotImplemented: direct execution of ScalarAggregate`, and no
function of that kind was
registered.
--
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]
