[Feature] BanyanDB: COUNT_DISTINCT, standalone + distributed push-down

[hanahmily]

Part 3 of 5 of the tag aggregation and time bucketing design, shipped in apache/skywalking-banyandb#1360.

Design references: §6 Semantics · §7.3 COUNT_DISTINCT on a single node · §7.4 COUNT_DISTINCT across data nodes · §7.6 Memory safety · §12 Delivery order (stages 4, 4a, 5)

Depends on: #14088 (API foundation + tag binding).

This is the highest-stakes issue in the series — a mis-evaluated push-down condition does not crash, it silently returns a plausible-looking wrong count. Test it hardest.

Scope

Standalone

• pkg/query/aggregation/distinct.go: Distinct interface (In/Val/Reset), NewDistinct(), keyed by the existing appendKeyComponent encoder (already handles every type the §6 matrix admits).
• New aggSlot variant; §6 matrix's COUNT_DISTINCT column (string/int/bytes accepted, array/timestamp rejected).
• Per-value memory charge: Consume reserves on each newly-added value and refunds in Close; also fixes reduce.go:133's entrySize=0 so the liaison-side reduce isn't silently unbudgeted.
• NewMap/NewReduce keep returning errUnknownFunc for COUNT_DISTINCT — this is what makes a COUNT_DISTINCT TopNAggregation rule fail loudly instead of mis-aggregating; needs a pinning test, not new guard code.

Distributed push-down

• Analyzer condition: routingTags ⊆ groupByTags ∪ {target} (routing key = the measure's sharding_key if configured, else its entity) — reject otherwise, with an error naming the target, the entity/routing key, and the two ways out.
• Reject any request resolving to more than one lifecycle stage, bucketed or not (expand DefaultStages before counting — never silently drop a stage).
• Reject ambiguous bare tag names (same name in two families) — require the family qualifier.
• Map phase: add the shard-id column to keyIndices so each emitted partial is exactly one shard's contribution — a precondition fix, since today's partial is per-node, not per-shard.
• Reduce: prepend stage to the dedup key — (stage, shard, group_key) — since hot/warm nodes can both report the same shard id with different, disjoint data; needs a stage field added to the partial wire schema.
• distributedAggFunc maps COUNT_DISTINCT to the existing SUM reducer (no new wire type, no new Partial shape).

Parallel, non-gating investigation (§7.4 callout)

• The replicas=1 experiment: compare distributed SUM against a raw scan at replicas=1 vs. replicas=0. Its outcome decides only whether the shard-in-keyIndices fix above is also a live correctness fix for existing SUM/COUNT/MEAN (→ a CHANGES.md entry and possibly a patch release) — not whether to build the fix, which ships in this issue regardless.

Out of scope

• Time-bucket grouping — see the sibling issue in this series ([Feature] BanyanDB: time-bucket grouping, standalone + distributed #14089); this issue's push-down condition composes with bucketing once both land, with no special case (the bucket is just another leading group key).

Test / Definition of done

• New package tests in pkg/query/aggregation/distinct_test.go (zero tests exist there today): set semantics, reset, duplicate insertion doesn't grow the count.
• COUNT_DISTINCT over string/int/bytes targets; null values excluded; budget exhaustion fails with the existing "aggregation memory budget exceeded" message.
• Off-cadence disagreement fixture: one entity, two points in one bucket (09:01:00, 09:01:34 on a 1-minute measure) — assert COUNT reports 2 and COUNT_DISTINCT reports 1.
• Accept/reject table driven by routingTags, not entity tags.
• Sharding-key counterexample: entity {e}, sharding key {s}, the same e under two s values, COUNT_DISTINCT(e) ungrouped on s — must reject (an entity-only rule wrongly accepts this and returns 2 for one distinct value).
• Multi-stage rejection, including DefaultStages expansion; paired with the positive case — the same target in two disjoint buckets of one stage must count once per bucket and must not be rejected.
• Ambiguous tag name resolves by family qualifier; an unqualified reference fails naming both families.
• Rejection is asserted at analyze time, before any node is contacted; error text is checked (it is the entire remediation path for the caller).
• Replica dedup: the same (stage, shard, group) fed twice doesn't double-count; two partials differing only in shard do add.
• Standalone vs. distributed agreement; replicas=0 vs. replicas=1 agreement (the permanent regression form of the parallel investigation above).
• The investigation's written finding, plus a CHANGES.md entry if replicas=1 divergence reproduces.

[hanahmily]

Heads-up for whoever picks this up: the liaison-side reduce foundation this issue's distributed push-down reuses (distributedAggFunc → the existing AggModeMap/AggModeReduce machinery, i.e. ReduceRawFrames/ReducePartialBatches/resolveKeyIndices in pkg/query/vectorized/measure/reduce.go) got two fixes during #14089's review cycle (apache/skywalking-banyandb#1362, follow-up commit e48e23db):

1. resolveKeyIndices now resolves GroupBy key tags by (TagFamily, Name), not Name alone. A stale note in the design doc had claimed bare-name matching was safe because "tag names are unique per measure" — that directly contradicted the doc's own §5.1 (which already debunks that assumption) and was a real bug: a valid schema with the same tag name in two families could bind to the wrong family's column. distributedGroupByTagKey now threads the family through end to end.
2. ReducePartialBatches now sorts bucketed results into global bucket-ascending order (BatchAggregation.SortInsertionByBucket) before draining. Consuming node partials frame-by-frame only produced piecewise-ascending order across nodes, which would have mispaginated limit/offset on a bucketed distributed query.

Since this issue's scope note says "distributedAggFunc maps COUNT_DISTINCT to the existing SUM reducer (no new wire type, no new Partial shape)," COUNT_DISTINCT's distributed reduce goes through this same, now-fixed path — no extra work needed here for either issue, just flagging it so nobody re-derives or second-guesses it.

Design doc corrections tracking this (and two other doc/implementation gaps found during #14089): apache/skywalking-banyandb#1363.