singhpratech commented on issue #1296:
URL: https://github.com/apache/arrow-go/issues/1296#issuecomment-5593008436
Design for the scalar aggregate kernels, before the PR. It follows the C++
`ScalarAggregateKernel` (init, consume, merge, finalize) and the C++
semantics for nulls,
`skip_nulls` and `min_count`; the two Go-specific choices are the questions
at the end.
The framework and the eight kernels below are written and pass their tests
locally against
main (281776d), with results compared against pyarrow; I will push once you
have commented on the shape.
This is longer than a comment usually is because it adds an exported kernel
interface; every
item below is either a decision you may want to change or a fact you can
check in the diff.
**Kernel type** (package `exec`; the name follows the existing
`FuncScalarAgg` abbreviation,
the function type below keeps the full name to match `ScalarFunction` and
`VectorFunction`):
```go
// ScalarAggConsume folds one span into the state in ctx.State.
type ScalarAggConsume = func(ctx *KernelCtx, span *ExecSpan) error
// ScalarAggMerge folds src into dst; both come from the kernel's init
function.
type ScalarAggMerge = func(ctx *KernelCtx, src, dst KernelState) error
// ScalarAggFinalize produces the result from ctx.State.
type ScalarAggFinalize = func(ctx *KernelCtx) (scalar.Scalar, error)
type ScalarAggKernel struct {
kernel
ConsumeFn ScalarAggConsume
MergeFn ScalarAggMerge
FinalizeFn ScalarAggFinalize
}
func NewScalarAggKernel(in []InputType, out OutputType, init KernelInitFn,
consume ScalarAggConsume, merge ScalarAggMerge, finalize
ScalarAggFinalize) ScalarAggKernel
```
Finalize returns a `scalar.Scalar` rather than a Datum because `exec` cannot
import `compute`;
every kernel here returns a scalar (`min_max` a `*scalar.Struct`) and the
executor boxes it with
`NewDatum`. Merge order is the C++ one, `merge(ctx, src&&, dst*)`. State
lives in
`KernelCtx.State` from the existing `KernelInitFn`; `AddKernel` rejects a
nil init, since the
executor cannot consume without a state. `MergeFn` is required on every
kernel and tested, so a
parallel consume can be added later without touching kernels (no executor is
parallel today, so
that is new work, not a hook).
**Function type and executor.** `kernelType` gains `exec.ScalarAggKernel`;
`ScalarAggregateFunction` is `funcImpl[exec.ScalarAggKernel]` with `Kind()
== FuncScalarAgg`,
its own `SetDefaultOptions`, and `DispatchBest = DispatchExact` (one kernel
per exact input
type, as `VectorFunction`). `execInternal` gets a pooled `scalarAggExecutor`
for
`FuncScalarAgg`: it iterates spans without scalar promotion, as the C++
`ScalarAggExecutor` does
(so each kernel handles a scalar input weighted by the span length),
consumes sequentially into
one state, then finalizes once. An empty input still reaches Finalize, so
`min_count` decides the
result, as in C++. The kernels count nulls of each span from its bitmap and
never write to the
iterator's span: `ArraySpan.SetSlice` carries a cached null count into the
next slice, and a
kernel that calls `UpdateNullCount()` on the shared span makes later slices
wrong once
`ChunkSize` is below the input length (`and_kleene`/`or_kleene` on main show
it; filed as
#1305 with a fix). A test runs every aggregate at chunk sizes 1 to 65
against the default. Aggregates stay outside `exprs` (it accepts scalar
functions only), which
matches C++ where aggregates run through Acero, not expressions.
**Options.** C++ field names and tags, so expression serialization
round-trips:
```go
type ScalarAggregateOptions struct {
SkipNulls bool `compute:"skip_nulls"`
MinCount uint32 `compute:"min_count"`
}
func DefaultScalarAggregateOptions() *ScalarAggregateOptions {
return &ScalarAggregateOptions{SkipNulls: true, MinCount: 1}
}
type CountMode int8
const (
CountOnlyValid CountMode = iota
CountOnlyNull
CountAll
)
type CountOptions struct {
Mode CountMode `compute:"mode"`
}
```
Each function registers `DefaultScalarAggregateOptions()` as its default, so
`CallFunction` with
nil options behaves as C++ defaults. A zero-value `ScalarAggregateOptions{}`
means
`skip_nulls=false, min_count=0`, which is what C++ gives for
`ScalarAggregateOptions(false, 0)`.
The package has precedent for both ways of handling that:
`ArithmeticOptions.NoCheckOverflow`
inverts the field so the zero value is the default, `TakeOptions` keeps the
C++ name and relies on
`DefaultTakeOptions()`. This note takes the second; see question 2.
**Kernels, first PRs.** Semantics are C++'s, checked row by row against
pyarrow built from main:
| function | inputs → output | result is null when |
|---|---|---|
| `sum` | int8..int64 → int64; uint8..uint64 → uint64; float32/64 → float64;
bool → uint64; null → int64 | `!skip_nulls` and a null was seen, or fewer than
`min_count` values |
| `mean` | numeric and bool → float64; null → float64 | same; with
`min_count=0` and no values: NaN (0.0 for the null type) |
| `min_max` | numeric and bool → `struct<min: T, max: T>`; null → struct of
nulls | `!skip_nulls` and a null was seen, or fewer than `min_count` values,
with `min_count` forced to at least 1 |
| `min`, `max` | same inputs → T | the `min_max` state; Finalize picks one
field |
| `count` | any type → int64 (`CountOptions`) | never; ignores
`ScalarAggregateOptions`, counts logical nulls |
| `any`, `all` | bool → bool | Kleene: `any` is null when `!skip_nulls` and
no true and a null was seen; `all` when `!skip_nulls` and no false and a null
was seen; both when fewer than `min_count` values |
Integer `sum` wraps, which is what C++ produces in practice; there is no
checked scalar aggregate
in C++ either. Float `sum` and `mean` use the pairwise summation of C++
`SumArray`, so results
match pyarrow exactly; `mean` accumulates in float64 for numeric and bool
inputs. `min_max` starts
from NaN with C `fmin`/`fmax` semantics: NaN never wins against a value, the
result is NaN only
when every non-null value is NaN, and -0.0 orders below +0.0 (so `min` of
`[0.0, -0.0]` is
-0.0, as pyarrow returns). Thin wrappers `Sum`, `Mean`, `MinMax`, `Min`,
`Max`, `Count`,
`Any`, `All` mirror `api_aggregate.h`.
Two PRs: the framework with `count` and `sum` first, then the other six
kernels. Not in them:
`product`, `count_distinct`, `first_last`/`first`/`last` (which add the C++
`ordered` flag),
decimals, the statistical aggregates, and the `hash_*` family, which needs a
grouper and its own
note. Tests are table-driven per function in `compute_test`: nulls, no
nulls, all null, empty,
scalar and chunked inputs under default options, `skip_nulls=false` and
`min_count` above and
below the valid count, plus a consume-halves-then-merge check per kernel.
Two questions:
1. `FinalizeFn` returning `scalar.Scalar` instead of a Datum, forced by the
`exec` → `compute`
import direction: acceptable?
2. `skip_nulls` as a plain bool with the C++ name (zero value false), or
inverted like
`NoCheckOverflow` so the zero value is the default? `min_count` cannot be
inverted either way.
--
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]
