You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/09/18 05:35:24 UTC
(skywalking-banyandb) branch main updated: feat(query): group measure queries by time bucket (#1362)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 4ad47b655 feat(query): group measure queries by time bucket (#1362)
4ad47b655 is described below
commit 4ad47b65580b7a111ab967eaf966d0b5cfdba0be
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Sep 18 13:35:17 2026 +0800
feat(query): group measure queries by time bucket (#1362)
* feat(query): group measure queries by time bucket
Adds measure.v1.QueryRequest.GroupBy.time_bucket support (issue #14089):
GroupBy can now bucket by tumbling time windows, alone or combined with a
tag GroupBy, in both standalone and distributed queries. Width comes from
the request or falls back to the measure's own interval.
BatchTimeBucket streams bucket-by-bucket on the part-scan path (bounding
live memory to one bucket's groups) and falls back to a bounded
non-streaming map for index-mode measures, which have no
ascending-timestamp guarantee. It is a genuine vectorized.PullOperator
that pulls upstream lazily via a new PipelineBuilder.Transform primitive,
rather than a BreakerOperator that would have to drain the whole scan
before emitting anything.
A mixed-version cluster where an older data node ignores time_bucket is
rejected with a clear error instead of silently collapsing the series to
one row. Bucketed output is kept globally ascending in both the
distributed reduce (multi-node partials) and the index-mode map fallback,
which each independently lose that ordering without an explicit sort.
Integration coverage added under test/cases/measure for both standalone
and distributed query.
* fix(query): resolve distributed GroupBy keys by family, drain map-mode buckets incrementally
Addresses two real issues from Copilot's review of PR #1362:
- Thread (TagFamily, tagNames) through the liaison-side distributed reduce
(distributedGroupByTagKey, ReduceRawFrames/ReducePartialBatches,
resolveKeyIndices) instead of matching key tags by bare name. Tag-family
validation does not reject the same tag name in two families, so a valid
schema with a repeated name could bind a bucketed or tag-keyed distributed
reduce to the wrong family's column. Corrects a stale design-doc note that
had claimed this was safe, contradicting the doc's own §5.1.
- Drain BatchTimeBucket's map-mode terminal aggregator one page at a time
(drainMapModeAggregator) instead of materializing every page into the
pending queue before returning the first one. That aggregator can
represent the whole scan in index mode, so the old behavior defeated the
bounded-memory intent the PullOperator rewrite exists for. Streaming
mode's per-bucket flushes are unaffected (already bounded).
A third comment (build_test.go) was verified as a false positive: the test
already uses the correct columns and passes.
---
CHANGES.md | 3 +-
docs/design/0.12.0/tag-aggregation/README.md | 18 +-
pkg/query/model/model.go | 24 +-
.../vectorized/measure/agg_value_path_test.go | 6 +
pkg/query/vectorized/measure/aggregation.go | 117 ++++-
pkg/query/vectorized/measure/plan.go | 66 +++
pkg/query/vectorized/measure/plan/analyzer.go | 126 ++++-
pkg/query/vectorized/measure/plan/analyzer_test.go | 195 +++++++-
pkg/query/vectorized/measure/plan/build_test.go | 81 +++
pkg/query/vectorized/measure/plan/distributed.go | 22 +-
.../vectorized/measure/plan/distributed_test.go | 61 +++
pkg/query/vectorized/measure/plan/groupby_agg.go | 42 +-
pkg/query/vectorized/measure/plan_test.go | 61 +++
pkg/query/vectorized/measure/raw_emit.go | 6 +-
.../vectorized/measure/raw_emit_bench_test.go | 2 +-
pkg/query/vectorized/measure/raw_emit_test.go | 4 +-
pkg/query/vectorized/measure/reduce.go | 65 ++-
pkg/query/vectorized/measure/reduce_test.go | 283 +++++++++++
pkg/query/vectorized/measure/timebucket.go | 465 ++++++++++++++++++
pkg/query/vectorized/measure/timebucket_test.go | 542 +++++++++++++++++++++
.../vectorized/measure/topology_matrix_test.go | 4 +-
pkg/query/vectorized/pipeline.go | 29 ++
pkg/query/vectorized/pipeline_test.go | 108 ++++
pkg/test/helpers/constant.go | 6 +
.../testdata/measures/time_bucket_metric.json | 36 ++
test/cases/init.go | 7 +
test/cases/measure/data/data.go | 22 +-
.../measure/data/input/group_time_bucket.yaml | 27 +
.../data/input/group_time_bucket_with_tag.yaml | 35 ++
.../data/testdata/time_bucket_metric_data.json | 152 ++++++
.../cases/measure/data/want/group_time_bucket.yaml | 33 ++
.../data/want/group_time_bucket_with_tag.yaml | 66 +++
test/cases/measure/measure.go | 7 +
33 files changed, 2616 insertions(+), 105 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 8480cd32f..7a6e01204 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -11,7 +11,8 @@ Release Notes.
- Add logging related flags to the FODC proxy and agent. Every BanyanDB binary now shares the same logging flags and the matching `BYDB_LOGGING_*` environment variables.
- [Breaking Change] Remove the row-based query execution path from the Stream, Measure and Trace query engines. Measure TopN pre-aggregation is unchanged and still executes row-based. The `--stream-vectorized-enabled`, `--trace-vectorized-enabled` and `--measure-vectorized-enabled` flags stay registered but no longer select an engine; `=false` now fails fast at startup on standalone, data and liaison nodes. A query shape the vectorized engine cannot plan now returns an error instead of f [...]
- Push the criteria tag filter ahead of the vectorized stream merge, so a filtered index-order query bounds its merge at limit+offset. For such a query, where several rows share an ElementID, the criteria is now evaluated first and the element is represented by its first matching row in the requested sort order. Filtered timestamp-order queries are unchanged.
-- `measure.v1.QueryRequest.Aggregation` can now target a tag (`tag_name` + `tag_family`) instead of a field: `SUM`/`MIN`/`MAX`/`MEAN` over an `INT` tag, and `COUNT` over any scalar tag, in both standalone and distributed queries. Wire-additive only — existing clients that only ever set `field_name` are unaffected. `COUNT_DISTINCT` and time-bucketed `GROUP BY` are not yet implemented; see the [tag aggregation and time bucketing design](docs/design/0.12.0/tag-aggregation/README.md).
+- `measure.v1.QueryRequest.Aggregation` can now target a tag (`tag_name` + `tag_family`) instead of a field: `SUM`/`MIN`/`MAX`/`MEAN` over an `INT` tag, and `COUNT` over any scalar tag, in both standalone and distributed queries. Wire-additive only — existing clients that only ever set `field_name` are unaffected. `COUNT_DISTINCT` is not yet implemented; see the [tag aggregation and time bucketing design](docs/design/0.12.0/tag-aggregation/README.md).
+- `measure.v1.QueryRequest.GroupBy.time_bucket` groups a measure query by tumbling time windows, alone or combined with a tag `GroupBy`, in both standalone and distributed queries. The bucket width comes from the request or falls back to the measure's own `interval`. The part-scan path streams bucket-by-bucket so live memory never exceeds one bucket's groups; index-mode measures (no ascending-timestamp guarantee) use a bounded non-streaming fallback instead. A mixed-version cluster where [...]
### Bug Fixes
diff --git a/docs/design/0.12.0/tag-aggregation/README.md b/docs/design/0.12.0/tag-aggregation/README.md
index 8f854fa2e..08ab65cdf 100644
--- a/docs/design/0.12.0/tag-aggregation/README.md
+++ b/docs/design/0.12.0/tag-aggregation/README.md
@@ -1208,12 +1208,18 @@ retired. A clear error is more honest than a second implementation that will rot
- **Follow-up: gap filling.** Needs a declared bucket set and a zero value per aggregate; SUM and
COUNT fill with 0, MIN/MAX/MEAN have no defensible fill.
-### Known adjacent inconsistency (note, do not fix here)
-
-`reduce.go:169 resolveKeyIndices` matches key tags by `Name` only, ignoring `TagFamily`, unlike
-`plan.go:140` and `distributed.go:945`, which both compare the family. It is safe in practice —
-tag names are unique per measure (§5.1) — and `distributedGroupByTagNames` (`:1131`) discards the
-family before calling it anyway. New code should resolve on `(family, name)`.
+### Fixed: adjacent inconsistency in the liaison-side key resolution
+
+`reduce.go resolveKeyIndices` used to match key tags by `Name` only, ignoring `TagFamily`, unlike
+`plan.go:140` and `distributed.go:945`, which both compare the family. An earlier version of
+this note claimed that was safe because "tag names are unique per measure" — that directly
+contradicts §5.1 above, which debunks exactly that assumption (tag-family validation does not
+reject a name repeated across families, and `pkg/query/vectorized/schema_test.go` exercises it
+as a valid case). A bucketed or tag-keyed distributed reduce over such a schema could silently
+bind to the wrong family's column. Fixed by threading the key's `TagFamily` alongside its names
+end to end — `distributedGroupByTagKey` (`distributed.go`) now returns `(family, names)` instead
+of discarding the family, and `resolveKeyIndices` matches on `(family, name)`, same as `plan.go`
+and `distributed.go`'s row-merge path.
---
diff --git a/pkg/query/model/model.go b/pkg/query/model/model.go
index 9d87c36df..bc9a9c24a 100644
--- a/pkg/query/model/model.go
+++ b/pkg/query/model/model.go
@@ -52,22 +52,30 @@ type TagProjection struct {
// MeasureGroupBy describes a GroupBy clause for a measure query. v1 supports
// a single tag family; each entry in TagNames is a key column. An empty
-// TagNames slice means the query carries no GroupBy clause.
+// TagNames slice means the query carries no tag GroupBy clause — the query
+// may still bucket by time alone (TimeBucket set, TagNames empty).
//
-// TimeBucket carries the request's optional time_bucket clause (design §5.3);
-// a nil TimeBucket means no bucketing. Resolution and execution land in a
-// later delivery stage — this field is only plumbed through from the wire.
+// TimeBucket carries the request's resolved time_bucket clause (design §5.3);
+// a nil TimeBucket means no bucketing.
type MeasureGroupBy struct {
TimeBucket *MeasureTimeBucket
TagFamily string
TagNames []string
}
-// MeasureTimeBucket describes a GroupBy.time_bucket clause. Width is the
-// requested bucket duration string; empty means "use the measure's own
-// interval" (design §5.3).
+// MeasureTimeBucket describes a resolved GroupBy.time_bucket clause (design
+// §5.3). Width is the raw requested/resolved duration string, kept for
+// diagnostics; WidthNanos is its parsed value in nanoseconds and is what
+// execution actually uses — always > 0 once translateGroupBy has resolved it
+// (a zero or negative width is rejected at analyze time, never carried this
+// far). UseIndexModeMap is set when measureSchema.GetIndexMode() is true: an
+// index-mode measure's scan has no ascending-timestamp guarantee, so
+// execution must take the non-streaming map path (design §7.2) instead of
+// the streaming operator, decided once here rather than guessed at runtime.
type MeasureTimeBucket struct {
- Width string
+ Width string
+ WidthNanos int64
+ UseIndexModeMap bool
}
// MeasureAgg describes a single aggregation for a measure query. v1 supports
diff --git a/pkg/query/vectorized/measure/agg_value_path_test.go b/pkg/query/vectorized/measure/agg_value_path_test.go
index c1bc5e022..f83ffa6cc 100644
--- a/pkg/query/vectorized/measure/agg_value_path_test.go
+++ b/pkg/query/vectorized/measure/agg_value_path_test.go
@@ -42,7 +42,9 @@ func TestReduceRawFramesTracing_Path_Typed(t *testing.T) {
_, path, reduceErr := ReducePartialBatches(
[]*vectorized.RecordBatch{b},
+ "f",
[]string{"g"},
+ false,
[]AggReduceSpec{{OutputName: "sum_v", Func: AggSum}},
64,
vectorized.NewMemoryTracker(1<<30),
@@ -76,7 +78,9 @@ func TestReduceRawFramesTracing_Path_FieldValueFallback(t *testing.T) {
_, path, reduceErr := ReducePartialBatches(
[]*vectorized.RecordBatch{b},
+ "f",
[]string{"g"},
+ false,
[]AggReduceSpec{{OutputName: "sum_v", Func: AggSum}},
64,
vectorized.NewMemoryTracker(1<<30),
@@ -107,7 +111,9 @@ func TestReduceRawFramesTracing_Path_Unresolved(t *testing.T) {
_, path, reduceErr := ReducePartialBatches(
[]*vectorized.RecordBatch{b},
+ "f",
[]string{"g"},
+ false,
[]AggReduceSpec{{OutputName: "sum_v", Func: AggSum}},
64,
vectorized.NewMemoryTracker(1<<30),
diff --git a/pkg/query/vectorized/measure/aggregation.go b/pkg/query/vectorized/measure/aggregation.go
index 2f2888753..bcea2164c 100644
--- a/pkg/query/vectorized/measure/aggregation.go
+++ b/pkg/query/vectorized/measure/aggregation.go
@@ -141,9 +141,11 @@ type BatchAggregation struct {
aggInputCountIdx []int
insertion []*aggGroup
outputShardIdx int
+ outputTimeIdx int
tagOutOffset int
mode AggMode
shardIDIdx int
+ bucketIdx int
entrySize int64
reserved int64
rowsIn int64
@@ -153,7 +155,7 @@ type BatchAggregation struct {
closed bool
}
-// aggGroup carries one bucket's reduction state plus a copy of every
+// aggGroup carries one group's reduction state plus a copy of every
// projected tag column for this group. tagCols is indexed by position in
// BatchAggregation.tagIndices (NOT just the GroupBy keys), so non-key
// projected tags can be emitted as their first-seen value.
@@ -165,11 +167,18 @@ type BatchAggregation struct {
// read by AggModeMap emit; AggModeAll ignores it. Stays zero when
// keyIndices is empty (scalar reduce) so the emitted partial matches the
// row path's aggAllIterator.Current() hardcoding ShardId: 0.
+//
+// timeValue is captured by newGroup from the input batch's (already
+// bucket-floored) RoleTimestamp column at the group's creating row, exactly
+// like shardID — it is constant for every row in the group by construction,
+// since the bucket-floored timestamp is itself part of the group key
+// (design §7.2). Read only when bucketIdx >= 0; ignored otherwise.
type aggGroup struct {
- key string
- tagCols []vectorized.Column
- slots []aggSlot
- shardID int64
+ key string
+ tagCols []vectorized.Column
+ slots []aggSlot
+ shardID int64
+ timeValue int64
}
// aggSlot holds either:
@@ -205,13 +214,28 @@ type aggSlot struct {
// reserved per new group bucket (key columns + slots + map entry overhead).
// Pass entrySize=0 to disable per-group bookkeeping. tracker must not be nil
// — use a large NewMemoryTracker for unit tests that don't care about budget.
+//
+// Time bucketing (design §7.2) needs no dedicated parameter: when keyIndices
+// contains a RoleTimestamp column, that column is treated as the bucket key
+// — grouping by it is already handled generically by computeKey, and the
+// only bucket-specific behavior (conditional D2 reversal: re-emitting the
+// bucket start as a RoleTimestamp output column) is derived here from that
+// same column. RoleTimestamp never legitimately appears in keyIndices for
+// any other reason, so this inference is unambiguous.
func NewBatchAggregation(
input *vectorized.BatchSchema, keyIndices []int,
aggs []AggSpec, mode AggMode, batchSize int,
tracker *vectorized.MemoryTracker, entrySize int64,
) *BatchAggregation {
+ bucketIdx := -1
+ for _, idx := range keyIndices {
+ if input.Columns[idx].Role == vectorized.RoleTimestamp {
+ bucketIdx = idx
+ break
+ }
+ }
tagIndices := collectTagIndices(input, keyIndices, hiddenTagExclusionSet(aggs))
- layout := buildAggOutputLayout(input, tagIndices, aggs, mode)
+ layout := buildAggOutputLayout(input, tagIndices, aggs, mode, bucketIdx)
return &BatchAggregation{
inputSchema: input,
outputSchema: layout.schema,
@@ -223,8 +247,10 @@ func NewBatchAggregation(
aggOutOffsets: layout.aggOutOffsets,
aggHasCount: layout.aggHasCount,
outputShardIdx: layout.outputShardIdx,
+ outputTimeIdx: layout.outputTimestampIdx,
tagOutOffset: layout.tagOutOffset,
shardIDIdx: findShardIDIndex(input),
+ bucketIdx: bucketIdx,
aggInputCountIdx: buildAggInputCountIdx(input, aggs, mode),
mode: mode,
aggValuePath: deriveAggValuePath(input, aggs),
@@ -420,6 +446,34 @@ func (a *BatchAggregation) NextBatch(_ context.Context) (*vectorized.RecordBatch
return out, nil
}
+// SortInsertionByBucket stable-sorts emission order by each group's
+// captured bucket timestamp (aggGroup.timeValue, design §7.2), ascending.
+// A no-op when this aggregation carries no bucket key (bucketIdx < 0).
+//
+// This exists for BatchTimeBucket's map-mode drain: its single persistent
+// aggregator accumulates groups across the whole scan in first-seen order,
+// but map mode exists precisely because the input may not be
+// time-ordered — so emission order needs an explicit sort to satisfy the
+// bucket-ascending output contract (design §7.5). Calling this on a
+// streaming-mode instance is also safe but a no-op in effect: each such
+// instance covers exactly one bucket, so every group already shares the
+// same timeValue and the stable sort leaves insertion order untouched.
+func (a *BatchAggregation) SortInsertionByBucket() {
+ if a.bucketIdx < 0 {
+ return
+ }
+ slices.SortStableFunc(a.insertion, func(x, y *aggGroup) int {
+ switch {
+ case x.timeValue < y.timeValue:
+ return -1
+ case x.timeValue > y.timeValue:
+ return 1
+ default:
+ return 0
+ }
+ })
+}
+
// Close releases the group map and refunds the outstanding memory
// reservation. Idempotent.
func (a *BatchAggregation) Close() error {
@@ -488,6 +542,17 @@ func (a *BatchAggregation) newGroup(b *vectorized.RecordBatch, rowIdx int, key s
}
}
}
+ // Capture the bucket start for a time-bucketed group (design §7.2). Every
+ // row in the group shares the same value by construction — the bucket is
+ // itself part of keyIndices — so the creating row's value is definitive.
+ if a.bucketIdx >= 0 {
+ if tsCol, ok := b.Columns[a.bucketIdx].(*vectorized.TypedColumn[int64]); ok {
+ data := tsCol.Data()
+ if rowIdx >= 0 && rowIdx < len(data) {
+ g.timeValue = data[rowIdx]
+ }
+ }
+ }
return g, nil
}
@@ -533,6 +598,11 @@ func (a *BatchAggregation) emitGroupRow(out *vectorized.RecordBatch, group *aggG
if a.outputShardIdx >= 0 {
out.Columns[a.outputShardIdx].(*vectorized.TypedColumn[int64]).Append(group.shardID)
}
+ // Conditional D2 reversal (design §7.2): re-emit the bucket start as a
+ // RoleTimestamp column, present only when this aggregation is bucketed.
+ if a.outputTimeIdx >= 0 {
+ out.Columns[a.outputTimeIdx].(*vectorized.TypedColumn[int64]).Append(group.timeValue)
+ }
// Projected tag columns, in tagIndices order — including non-key tags
// carried forward as the first-seen value. tagOutOffset is 0 in
// AggModeAll and 1 in AggModeMap (after the shard-id column).
@@ -699,25 +769,34 @@ type aggOutputLayout struct {
// RoleShardID column. -1 in AggModeAll (no shard column emitted);
// 0 in AggModeMap (the partial batch always carries shard id first).
outputShardIdx int
+ // outputTimestampIdx is the output-batch column index of the RoleTimestamp
+ // column re-emitted for a time-bucketed aggregation (design §7.2's
+ // conditional D2 reversal). -1 when the aggregation is not bucketed —
+ // the unbucketed case still drops the timestamp entirely, unchanged.
+ outputTimestampIdx int
// tagOutOffset is the output-batch column index where the tag columns
- // begin (0 in AggModeAll; 1 in AggModeMap).
+ // begin: 0 in AggModeAll, plus 1 for a leading shard-id column (AggModeMap)
+ // and plus 1 more for a leading bucket timestamp column (if bucketed).
tagOutOffset int
}
// buildAggOutputLayout derives the output-batch ColumnDef list AND the
-// per-agg / shard-id index bookkeeping for a given (input schema, tag
-// indices, agg specs, mode). It is the sole place that decides Map-mode's
-// shard-id-first + MEAN-emits-two-columns layout, so emit-time code can
-// just consult precomputed offsets.
+// per-agg / shard-id / timestamp index bookkeeping for a given (input
+// schema, tag indices, agg specs, mode, bucket index). It is the sole place
+// that decides Map-mode's shard-id-first + MEAN-emits-two-columns layout,
+// plus the bucketed case's leading timestamp column, so emit-time code can
+// just consult precomputed offsets. bucketIdx is the input-schema index of
+// the bucket key column, or -1 when the aggregation is not bucketed.
func buildAggOutputLayout(
- input *vectorized.BatchSchema, tagIndices []int, aggs []AggSpec, mode AggMode,
+ input *vectorized.BatchSchema, tagIndices []int, aggs []AggSpec, mode AggMode, bucketIdx int,
) aggOutputLayout {
- // Worst-case capacity: shard-id (1) + tags + 2 per agg (MEAN value + count).
- defs := make([]vectorized.ColumnDef, 0, 1+len(tagIndices)+2*len(aggs))
+ // Worst-case capacity: shard-id (1) + timestamp (1) + tags + 2 per agg.
+ defs := make([]vectorized.ColumnDef, 0, 2+len(tagIndices)+2*len(aggs))
layout := aggOutputLayout{
- aggOutOffsets: make([]int, len(aggs)),
- aggHasCount: make([]bool, len(aggs)),
- outputShardIdx: -1,
+ aggOutOffsets: make([]int, len(aggs)),
+ aggHasCount: make([]bool, len(aggs)),
+ outputShardIdx: -1,
+ outputTimestampIdx: -1,
}
if mode == AggModeMap {
defs = append(defs, vectorized.ColumnDef{
@@ -727,6 +806,10 @@ func buildAggOutputLayout(
})
layout.outputShardIdx = 0
}
+ if bucketIdx >= 0 {
+ layout.outputTimestampIdx = len(defs)
+ defs = append(defs, vectorized.ColumnDef{Role: vectorized.RoleTimestamp, Type: vectorized.ColumnTypeInt64})
+ }
layout.tagOutOffset = len(defs)
for _, ti := range tagIndices {
defs = append(defs, input.Columns[ti])
diff --git a/pkg/query/vectorized/measure/plan.go b/pkg/query/vectorized/measure/plan.go
index 8e96ef916..767b46e2a 100644
--- a/pkg/query/vectorized/measure/plan.go
+++ b/pkg/query/vectorized/measure/plan.go
@@ -140,6 +140,72 @@ func aggTargetName(agg *model.MeasureAgg) string {
return agg.FieldName
}
+// BuildTimeBucketOperator builds the BatchTimeBucket operator for a
+// time-bucketed GroupBy (design §7.2), wrapping upstream directly. Unlike
+// BuildOperators, this does not return a BreakerOperator to be attached via
+// PipelineBuilder.Break: BatchTimeBucket is a PullOperator that pulls from
+// upstream lazily inside its own NextBatch, which is what makes its memory
+// bound real (see BatchTimeBucket's doc). Callers wire it in via
+// PipelineBuilder.Transform instead. Check opts.GroupBy.GetTimeBucket() != nil
+// before calling this rather than BuildOperators.
+//
+// tagKeyIndices are the already resolved tag GroupBy keys (empty for a
+// bucket-only GroupBy, i.e. no tag key at all). Agg is optional, exactly as
+// for the non-bucketed case: when absent this is a bucketed raw GroupBy
+// (first-seen row per (bucket, tags)), when present it is a bucketed
+// aggregate — both shapes reuse the same BatchTimeBucket, since
+// BatchAggregation with zero AggSpecs already produces the raw-GroupBy
+// shape. (The analyzer currently rejects the no-Agg case for real requests
+// — BatchAggregation's empty-AggSpec output layout drops every projected
+// field, unlike BatchGroupByFirst's full-schema passthrough — but
+// BuildTimeBucketOperator itself stays agnostic to that policy choice.)
+func BuildTimeBucketOperator(
+ upstream vectorized.PullOperator, opts model.MeasureQueryOptions, schema *vectorized.BatchSchema,
+ tracker *vectorized.MemoryTracker, batchSize int, mode AggMode,
+) (*BatchTimeBucket, error) {
+ if tracker == nil {
+ return nil, fmt.Errorf("vectorized.measure: BuildTimeBucketOperator requires a non-nil shared MemoryTracker")
+ }
+ if batchSize <= 0 {
+ return nil, fmt.Errorf("vectorized.measure: batchSize must be > 0, got %d", batchSize)
+ }
+ if mode == AggModeReduce {
+ return nil, fmt.Errorf("vectorized.measure: BuildTimeBucketOperator does not build AggModeReduce — that operator is built by the liaison reduce plan")
+ }
+ var tagKeyIndices []int
+ if opts.GroupBy != nil && opts.GroupBy.TagFamily != "" && len(opts.GroupBy.TagNames) > 0 {
+ var keyErr error
+ tagKeyIndices, keyErr = lookupGroupByKeyIndices(schema, opts.GroupBy)
+ if keyErr != nil {
+ return nil, keyErr
+ }
+ }
+ timestampIdx := schema.TimestampIndex()
+ if timestampIdx < 0 {
+ return nil, fmt.Errorf("vectorized.measure: time_bucket requires a RoleTimestamp column in the scan schema")
+ }
+ var specs []AggSpec
+ if opts.Agg != nil {
+ inputIdx, inputErr := lookupAggInputColumnIndex(schema, opts.Agg)
+ if inputErr != nil {
+ return nil, inputErr
+ }
+ aggFn, fnErr := protoAggFuncToInternal(opts.Agg.Func)
+ if fnErr != nil {
+ return nil, fnErr
+ }
+ specs = []AggSpec{{
+ Func: aggFn,
+ InputCol: inputIdx,
+ Output: aggTargetName(opts.Agg),
+ HideTag: opts.Agg.HideTag,
+ }}
+ }
+ tb := opts.GroupBy.TimeBucket
+ return NewBatchTimeBucket(upstream, schema, tagKeyIndices, timestampIdx, tb.WidthNanos,
+ specs, mode, batchSize, tracker, aggEntrySize, !tb.UseIndexModeMap), nil
+}
+
// lookupGroupByKeyIndices resolves each GroupBy tag name to its column index
// in schema. All names must exist within the configured tag family.
func lookupGroupByKeyIndices(schema *vectorized.BatchSchema, gb *model.MeasureGroupBy) ([]int, error) {
diff --git a/pkg/query/vectorized/measure/plan/analyzer.go b/pkg/query/vectorized/measure/plan/analyzer.go
index f9feb5e77..a3b44e010 100644
--- a/pkg/query/vectorized/measure/plan/analyzer.go
+++ b/pkg/query/vectorized/measure/plan/analyzer.go
@@ -181,38 +181,114 @@ func buildTagProjection(req *measurev1.QueryRequest) []model.TagProjection {
return out
}
-// translateGroupBy builds the model GroupBy struct from the proto,
-// validating that:
-// - the GroupBy tag_projection names exactly one family with non-empty tags
-// (v1 single-family limitation)
-// - the named GroupBy tags exist in the Measure schema
+// translateGroupBy builds the model GroupBy struct from the proto. A
+// GroupBy must set tag_projection, time_bucket, or both (design §7.2 —
+// bucketing alone, with no tag key, is a legal grouping). When
+// tag_projection is present it must name exactly one family with non-empty
+// tags (v1 single-family limitation), and every named tag must exist in the
+// Measure schema. When time_bucket is present, its width is resolved per
+// §5.3 and the request is checked against the bucketed streaming
+// precondition (time-ascending input).
func translateGroupBy(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (*model.MeasureGroupBy, error) {
- families := req.GetGroupBy().GetTagProjection().GetTagFamilies()
- if len(families) == 0 {
- return nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection must list at least one tag family")
+ groupByProto := req.GetGroupBy()
+ families := groupByProto.GetTagProjection().GetTagFamilies()
+ tbProto := groupByProto.GetTimeBucket()
+ if len(families) == 0 && tbProto == nil {
+ return nil, fmt.Errorf("plan.Analyze: GroupBy must set tag_projection, time_bucket, or both")
}
- if len(families) > 1 {
- return nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection v1 supports a single tag family, got %d", len(families))
+
+ gb := &model.MeasureGroupBy{}
+ if len(families) > 0 {
+ if len(families) > 1 {
+ return nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection v1 supports a single tag family, got %d", len(families))
+ }
+ family := families[0]
+ if len(family.GetTags()) == 0 {
+ return nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection family %q has no tags", family.GetName())
+ }
+ gb.TagFamily = family.GetName()
+ gb.TagNames = append([]string(nil), family.GetTags()...)
+ if validateErr := validateGroupByTags(measureSchema, gb); validateErr != nil {
+ return nil, validateErr
+ }
+ }
+
+ if tbProto != nil {
+ // A time-bucketed GroupBy without an Agg has no execution support:
+ // BatchTimeBucket's raw (no-AggSpec) shape reuses BatchAggregation's
+ // output layout, which — unlike BatchGroupByFirst's full schema
+ // passthrough — carries only tags and the bucket timestamp, silently
+ // dropping every projected field (and, on the distributed path, the
+ // series-id/version columns raw row merging requires). Reject rather
+ // than silently lose data; a bucketed raw GroupBy is a possible
+ // follow-up, not something this design ships.
+ if req.GetAgg() == nil {
+ return nil, fmt.Errorf("plan.Analyze: time_bucket requires Agg; a bucketed raw GroupBy (no aggregate) is not supported yet")
+ }
+ tb, tbErr := resolveTimeBucket(tbProto, measureSchema)
+ if tbErr != nil {
+ return nil, tbErr
+ }
+ if orderErr := validateBucketableOrdering(req); orderErr != nil {
+ return nil, orderErr
+ }
+ gb.TimeBucket = tb
}
- family := families[0]
- if len(family.GetTags()) == 0 {
- return nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection family %q has no tags", family.GetName())
+ return gb, nil
+}
+
+// resolveTimeBucket implements the §5.3 width-resolution table: an explicit
+// tb.Width wins; an empty Width falls back to the measure's own interval;
+// both empty is rejected, and so is any width that fails to parse or is not
+// strictly positive. There is deliberately no "must be a multiple of the
+// interval" rule — Measure.interval is a declared write cadence, not an
+// enforced storage invariant, so every positive width is equally meaningful.
+func resolveTimeBucket(tb *measurev1.QueryRequest_GroupBy_TimeBucket, measureSchema *databasev1.Measure) (*model.MeasureTimeBucket, error) {
+ raw := tb.GetWidth()
+ if raw == "" {
+ raw = measureSchema.GetInterval()
+ if raw == "" {
+ return nil, fmt.Errorf("plan.Analyze: time_bucket needs a width: measure %q declares no interval", measureSchema.GetMetadata().GetName())
+ }
}
- gb := &model.MeasureGroupBy{
- TagFamily: family.GetName(),
- TagNames: append([]string(nil), family.GetTags()...),
+ width, parseErr := timestamp.ParseDuration(raw)
+ if parseErr != nil {
+ return nil, fmt.Errorf("plan.Analyze: time_bucket width %q is not a valid duration: %w", raw, parseErr)
}
- if validateErr := validateGroupByTags(measureSchema, gb); validateErr != nil {
- return nil, validateErr
+ if width <= 0 {
+ return nil, fmt.Errorf("plan.Analyze: time_bucket width %q must be positive, got %s", raw, width)
}
- // time_bucket is wire-additive only in this delivery stage: the value is
- // carried onto the model unchanged, but nothing resolves or validates it
- // yet, and no operator acts on it (design §12 stage 0). Resolution,
- // validation, and the bucketing operator land in a later stage (§5.3, §7.2).
- if tb := req.GetGroupBy().GetTimeBucket(); tb != nil {
- gb.TimeBucket = &model.MeasureTimeBucket{Width: tb.GetWidth()}
+ return &model.MeasureTimeBucket{
+ Width: raw,
+ WidthNanos: int64(width),
+ UseIndexModeMap: measureSchema.GetIndexMode(),
+ }, nil
+}
+
+// validateBucketableOrdering rejects a bucketed request whose order_by names
+// a non-time index rule (design §7.2: both the streaming operator and the
+// index-mode map fallback assume the request carries no ordering that would
+// contradict time-ascending scan input). An aggregation request already
+// carries no effective order_by today — Analyze never reads
+// req.GetOrderBy(), and the distributed planner skips OrderBy resolution
+// whenever Agg is set — so this is a belt-and-suspenders guard against a
+// caller relying on a setting the engine would otherwise silently ignore,
+// applied uniformly regardless of the streaming/map choice for a
+// consistent contract.
+func validateBucketableOrdering(req *measurev1.QueryRequest) error {
+ orderBy := req.GetOrderBy()
+ if ruleName := orderBy.GetIndexRuleName(); ruleName != "" {
+ return fmt.Errorf("plan.Analyze: time_bucket requires time ordering; order_by.index_rule_name %q is not supported on a bucketed query", ruleName)
+ }
+ // An empty index_rule_name with Sort == SORT_DESC still resolves to a
+ // time-ordered scan (index.OrderByTypeTime) per applyMeasureQueryOrdering
+ // — just descending instead of ascending. BatchTimeBucket's streaming
+ // path assumes ascending input specifically, not merely "time-ordered",
+ // so this must be rejected too.
+ if orderBy.GetSort() == modelv1.Sort_SORT_DESC {
+ return fmt.Errorf("plan.Analyze: time_bucket requires ascending time order; order_by.sort SORT_DESC is not supported on a bucketed query")
}
- return gb, nil
+ return nil
}
// translateAgg builds the model Agg struct from the proto. Exactly one of
diff --git a/pkg/query/vectorized/measure/plan/analyzer_test.go b/pkg/query/vectorized/measure/plan/analyzer_test.go
index 9e9c6e32b..babbc4b47 100644
--- a/pkg/query/vectorized/measure/plan/analyzer_test.go
+++ b/pkg/query/vectorized/measure/plan/analyzer_test.go
@@ -580,12 +580,11 @@ func TestAnalyze_AggTagTarget_AlreadyProjected_HideTagFalse(t *testing.T) {
}
}
-// TestAnalyze_GroupByTimeBucket_WireOnly_NoExecutionYet pins this issue's
-// scope boundary (design §12 stage 0): GroupBy.time_bucket is carried onto
-// model.MeasureGroupBy unchanged, but nothing resolves, validates, or
-// executes it yet — Analyze must not error and must not alter the plan
-// shape just because time_bucket is set.
-func TestAnalyze_GroupByTimeBucket_WireOnly_NoExecutionYet(t *testing.T) {
+// TestAnalyze_GroupByTimeBucket_ResolvesWidth pins design §5.3's
+// explicit-width case end to end: GroupBy.time_bucket resolves to a
+// model.MeasureTimeBucket carrying both the raw string and its parsed
+// nanosecond width.
+func TestAnalyze_GroupByTimeBucket_ResolvesWidth(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
@@ -595,14 +594,19 @@ func TestAnalyze_GroupByTimeBucket_WireOnly_NoExecutionYet(t *testing.T) {
FieldName: fieldValue,
TimeBucket: &measurev1.QueryRequest_GroupBy_TimeBucket{Width: "5m"},
},
+ Agg: &measurev1.QueryRequest_Aggregation{
+ Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
+ FieldName: fieldValue,
+ },
}
p, err := Analyze(req, testMeasureSchema(), measure.AggModeAll)
if err != nil {
- t.Fatalf("a set time_bucket must not error in this delivery stage: %v", err)
+ t.Fatalf("Analyze: %v", err)
}
gba := p.(*Limit).Child.(*GroupByAgg)
- if gba.GroupBy.TimeBucket == nil || gba.GroupBy.TimeBucket.Width != "5m" {
- t.Fatalf("GroupBy.TimeBucket must carry the wire value through, got %+v", gba.GroupBy.TimeBucket)
+ const wantNanos = int64(5 * 60 * 1e9)
+ if gba.GroupBy.TimeBucket == nil || gba.GroupBy.TimeBucket.Width != "5m" || gba.GroupBy.TimeBucket.WidthNanos != wantNanos {
+ t.Fatalf("GroupBy.TimeBucket: want Width=5m WidthNanos=%d, got %+v", wantNanos, gba.GroupBy.TimeBucket)
}
}
@@ -638,3 +642,176 @@ func TestQueryRequest_TagAggAndTimeBucket_SurviveWireRoundTrip(t *testing.T) {
t.Fatalf("GroupBy.TimeBucket did not survive the wire round trip: %+v", got.GetGroupBy())
}
}
+
+// timeBucketReq builds a request with a bare time_bucket GroupBy (no tag
+// key) and no Agg, for exercising resolveTimeBucket's width table directly
+// through Analyze.
+func timeBucketReq(width string) *measurev1.QueryRequest {
+ return &measurev1.QueryRequest{
+ Name: "demo",
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
+ GroupBy: &measurev1.QueryRequest_GroupBy{
+ TimeBucket: &measurev1.QueryRequest_GroupBy_TimeBucket{Width: width},
+ },
+ // A bucketed GroupBy requires Agg (translateGroupBy rejects a
+ // bucketed raw GroupBy — no execution support carries the projected
+ // fields for that shape yet), so every width-table case needs one.
+ Agg: &measurev1.QueryRequest_Aggregation{
+ Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
+ FieldName: fieldValue,
+ },
+ }
+}
+
+// TestAnalyze_TimeBucket_WidthResolutionTable pins the full §5.3 table as
+// literal test cases, including the explicit non-rejection of a width that
+// is not a multiple of the measure's interval — the rule an earlier draft
+// wrongly proposed.
+func TestAnalyze_TimeBucket_WidthResolutionTable(t *testing.T) {
+ schemaWithInterval := testMeasureSchema()
+ schemaWithInterval.Interval = "1m"
+
+ cases := []struct {
+ schema *databasev1.Measure
+ name string
+ width string
+ wantNanos int64
+ wantErr bool
+ }{
+ {name: "width from request", width: "5m", schema: testMeasureSchema(), wantNanos: 5 * 60 * 1e9},
+ {name: "width empty, measure interval present", width: "", schema: schemaWithInterval, wantNanos: 60 * 1e9},
+ {name: "width empty, measure interval absent", width: "", schema: testMeasureSchema(), wantErr: true},
+ {name: "width unparseable", width: "banana", schema: testMeasureSchema(), wantErr: true},
+ {name: "width zero", width: "0s", schema: testMeasureSchema(), wantErr: true},
+ {name: "width negative", width: "-5m", schema: testMeasureSchema(), wantErr: true},
+ {
+ name: "width not a multiple of the interval is accepted", width: "7m",
+ schema: schemaWithInterval, wantNanos: 7 * 60 * 1e9,
+ },
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ p, err := Analyze(timeBucketReq(tc.width), tc.schema, measure.AggModeAll)
+ if tc.wantErr {
+ if err == nil {
+ t.Fatal("want error, got nil")
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("Analyze: %v", err)
+ }
+ gba := p.(*Limit).Child.(*GroupByAgg)
+ if gba.GroupBy.TimeBucket.WidthNanos != tc.wantNanos {
+ t.Fatalf("WidthNanos: want %d, got %d", tc.wantNanos, gba.GroupBy.TimeBucket.WidthNanos)
+ }
+ })
+ }
+}
+
+// TestAnalyze_TimeBucket_BoundaryTimestampAcceptedAsWidth is a design-§5.3
+// aside folded into the width table: a boundary-aligned width string is
+// just an ordinary positive duration — nothing about "boundary" is special
+// at resolution time (only bucketStart's floor arithmetic cares about
+// boundaries), so this doubles as a plain sanity check on parsing.
+func TestAnalyze_TimeBucket_BoundaryTimestampAcceptedAsWidth(t *testing.T) {
+ p, err := Analyze(timeBucketReq("1000ns"), testMeasureSchema(), measure.AggModeAll)
+ if err != nil {
+ t.Fatalf("Analyze: %v", err)
+ }
+ gba := p.(*Limit).Child.(*GroupByAgg)
+ if gba.GroupBy.TimeBucket.WidthNanos != 1000 {
+ t.Fatalf("WidthNanos: want 1000, got %d", gba.GroupBy.TimeBucket.WidthNanos)
+ }
+}
+
+// TestAnalyze_TimeBucket_BucketOnly_NoTagKey pins design §7.2: a GroupBy may
+// bucket by time alone, with no tag key at all.
+func TestAnalyze_TimeBucket_BucketOnly_NoTagKey(t *testing.T) {
+ p, err := Analyze(timeBucketReq("5m"), testMeasureSchema(), measure.AggModeAll)
+ if err != nil {
+ t.Fatalf("Analyze: %v", err)
+ }
+ gba := p.(*Limit).Child.(*GroupByAgg)
+ if gba.GroupBy.TagFamily != "" || len(gba.GroupBy.TagNames) != 0 {
+ t.Fatalf("bucket-only GroupBy must carry no tag key, got %+v", gba.GroupBy)
+ }
+ if gba.GroupBy.TimeBucket == nil {
+ t.Fatal("bucket-only GroupBy must still carry TimeBucket")
+ }
+}
+
+// TestAnalyze_GroupBy_NeitherTagProjectionNorTimeBucket_Errors pins that an
+// empty GroupBy message (no tag_projection, no time_bucket) is rejected —
+// there is nothing to group by.
+func TestAnalyze_GroupBy_NeitherTagProjectionNorTimeBucket_Errors(t *testing.T) {
+ req := &measurev1.QueryRequest{
+ Name: "demo",
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
+ GroupBy: &measurev1.QueryRequest_GroupBy{},
+ }
+ if _, err := Analyze(req, testMeasureSchema(), measure.AggModeAll); err == nil {
+ t.Fatal("an empty GroupBy message must error")
+ }
+}
+
+// TestAnalyze_TimeBucket_RejectsNonTimeOrderBy pins the analyzer guard
+// (design §7.2): a bucketed request whose order_by names a non-time index
+// rule is rejected rather than silently ignored.
+func TestAnalyze_TimeBucket_RejectsNonTimeOrderBy(t *testing.T) {
+ req := timeBucketReq("5m")
+ req.OrderBy = &modelv1.QueryOrder{IndexRuleName: "some_index"}
+ if _, err := Analyze(req, testMeasureSchema(), measure.AggModeAll); err == nil {
+ t.Fatal("a bucketed request with a non-time order_by must error")
+ }
+}
+
+// TestAnalyze_TimeBucket_RejectsDescendingTimeOrderBy pins the other half of
+// the ordering guard: an empty index_rule_name with Sort=SORT_DESC still
+// resolves to a time-ordered scan (index.OrderByTypeTime), just descending
+// instead of ascending — which would trip BatchTimeBucket's monotonicity
+// guard on the very first bucket transition. This must be rejected at
+// analyze time with a clear error, not surface as a confusing "bucket
+// regressed" failure deep in the scan.
+func TestAnalyze_TimeBucket_RejectsDescendingTimeOrderBy(t *testing.T) {
+ req := timeBucketReq("5m")
+ req.OrderBy = &modelv1.QueryOrder{Sort: modelv1.Sort_SORT_DESC}
+ if _, err := Analyze(req, testMeasureSchema(), measure.AggModeAll); err == nil {
+ t.Fatal("a bucketed request with order_by.sort=SORT_DESC must error")
+ }
+}
+
+// TestAnalyze_TimeBucket_WithoutAgg_Rejected pins that a bucketed raw
+// GroupBy (time_bucket set, Agg nil) is rejected rather than silently
+// dropping every projected field: BatchTimeBucket's no-Agg shape reuses
+// BatchAggregation's output layout (tags + bucket timestamp only), unlike
+// BatchGroupByFirst's full-schema passthrough for the non-bucketed raw
+// GroupBy case.
+func TestAnalyze_TimeBucket_WithoutAgg_Rejected(t *testing.T) {
+ req := &measurev1.QueryRequest{
+ Name: "demo",
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
+ GroupBy: &measurev1.QueryRequest_GroupBy{
+ TimeBucket: &measurev1.QueryRequest_GroupBy_TimeBucket{Width: "5m"},
+ },
+ }
+ if _, err := Analyze(req, testMeasureSchema(), measure.AggModeAll); err == nil {
+ t.Fatal("a bucketed GroupBy without Agg must error")
+ }
+}
+
+// TestAnalyze_TimeBucket_UsesIndexModeMap pins that the analyzer decides
+// streaming vs. the index-mode map fallback once, from
+// measureSchema.GetIndexMode(), rather than leaving it to be guessed later.
+func TestAnalyze_TimeBucket_UsesIndexModeMap(t *testing.T) {
+ indexModeSchema := testMeasureSchema()
+ indexModeSchema.IndexMode = true
+ p, err := Analyze(timeBucketReq("5m"), indexModeSchema, measure.AggModeAll)
+ if err != nil {
+ t.Fatalf("Analyze: %v", err)
+ }
+ gba := p.(*Limit).Child.(*GroupByAgg)
+ if !gba.GroupBy.TimeBucket.UseIndexModeMap {
+ t.Fatal("an index-mode measure must resolve UseIndexModeMap=true")
+ }
+}
diff --git a/pkg/query/vectorized/measure/plan/build_test.go b/pkg/query/vectorized/measure/plan/build_test.go
index 8d9e5b292..2f5506c45 100644
--- a/pkg/query/vectorized/measure/plan/build_test.go
+++ b/pkg/query/vectorized/measure/plan/build_test.go
@@ -19,6 +19,7 @@ package plan
import (
"context"
+ "fmt"
"testing"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
@@ -231,3 +232,83 @@ func TestGroupByAgg_Schema_DropsTimestampAddsAggField(t *testing.T) {
t.Fatalf("col 1 should be 'value' (row-path parity), got %s", out.Columns[1].Name)
}
}
+
+// TestGroupByAgg_TimeBucket_Schema_KeepsTimestampAtBucketStart is the
+// bucketed sibling of TestGroupByAgg_Schema_DropsTimestampAddsAggField
+// (design §7.2's conditional D2 reversal): the two together pin D2 as
+// conditional rather than repealed — an unbucketed aggregation still drops
+// the timestamp, a bucketed one carries it as the bucket start.
+func TestGroupByAgg_TimeBucket_Schema_KeepsTimestampAtBucketStart(t *testing.T) {
+ schema, batch := buildScanInput(t)
+ src := &fakePullSource{schema: schema, batches: []*vectorized.RecordBatch{batch}}
+ scan := NewScan(schema, ScanParams{})
+ scan.Source = src
+ gba, err := NewGroupByAgg(scan,
+ &model.MeasureGroupBy{
+ TagFamily: "default",
+ TagNames: []string{tagSvc},
+ TimeBucket: &model.MeasureTimeBucket{WidthNanos: 2},
+ },
+ &model.MeasureAgg{FieldName: fieldValue, Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ measure.AggModeAll,
+ )
+ if err != nil {
+ t.Fatalf("NewGroupByAgg: %v", err)
+ }
+ out := gba.Schema()
+ if out == nil {
+ t.Fatal("GroupByAgg.Schema must not be nil")
+ }
+ if out.TimestampIndex() < 0 {
+ t.Fatal("a bucketed aggregation must carry a RoleTimestamp column (design §7.2 D2 reversal)")
+ }
+
+ root := NewLimit(gba, 0, 10)
+ tracker := vectorized.NewMemoryTracker(1 << 20)
+ bc := &BuildContext{
+ Builder: vectorized.NewPipelineBuilder().WithMemoryTracker(tracker),
+ Tracker: tracker,
+ Config: measure.VectorizedConfig{BatchSize: 1024, QueryMemoryMiB: 1},
+ }
+ if buildErr := root.Build(context.Background(), bc); buildErr != nil {
+ t.Fatalf("Build: %v", buildErr)
+ }
+ pipeline, buildErr := bc.Builder.Build()
+ if buildErr != nil {
+ t.Fatalf("PipelineBuilder.Build: %v", buildErr)
+ }
+ if initErr := pipeline.Init(context.Background()); initErr != nil {
+ t.Fatal(initErr)
+ }
+ batches := drainPipeline(t, pipeline)
+
+ // buildScanInput rows: (ts=1,a,1) (ts=2,b,4) (ts=3,a,2) (ts=4,a,3) (ts=5,b,5).
+ // Width 2 buckets: ts1->0, ts2/ts3->2, ts4/ts5->4.
+ got := map[string]int64{}
+ for _, b := range batches {
+ tsIdx := b.Schema.TimestampIndex()
+ if tsIdx < 0 {
+ t.Fatalf("bucketed output batch must carry a RoleTimestamp column, got schema %+v", b.Schema.Columns)
+ }
+ tsCol := b.Columns[tsIdx].(*vectorized.TypedColumn[int64])
+ svcCol := b.Columns[tsIdx+1].(*vectorized.TypedColumn[string])
+ sumCol := b.Columns[tsIdx+2].(*vectorized.TypedColumn[int64])
+ for i := 0; i < b.Len; i++ {
+ key := svcColKey(tsCol.Data()[i], svcCol.Data()[i])
+ got[key] = sumCol.Data()[i]
+ }
+ }
+ want := map[string]int64{svcColKey(0, "a"): 1, svcColKey(2, "b"): 4, svcColKey(2, "a"): 2, svcColKey(4, "a"): 3, svcColKey(4, "b"): 5}
+ for k, v := range want {
+ if got[k] != v {
+ t.Errorf("sum[%s] = %d, want %d (full result: %v)", k, got[k], v, got)
+ }
+ }
+ if len(got) != len(want) {
+ t.Fatalf("result set size = %d, want %d: %v", len(got), len(want), got)
+ }
+}
+
+func svcColKey(bucket int64, svc string) string {
+ return fmt.Sprintf("%d|%s", bucket, svc)
+}
diff --git a/pkg/query/vectorized/measure/plan/distributed.go b/pkg/query/vectorized/measure/plan/distributed.go
index e010bf2c3..8f8a7713e 100644
--- a/pkg/query/vectorized/measure/plan/distributed.go
+++ b/pkg/query/vectorized/measure/plan/distributed.go
@@ -636,7 +636,8 @@ func collectRawFrameResponsesWithNodes(ff []bus.Future) ([][]byte, []*commonv1.T
}
func (p *DistributedPlan) executeAgg(ctx context.Context, frames [][]byte, req *measurev1.QueryRequest) (executor.MIterator, error) {
- keyTagNames := distributedGroupByTagNames(req.GetGroupBy())
+ keyTagFamily, keyTagNames := distributedGroupByTagKey(req.GetGroupBy())
+ bucketed := req.GetGroupBy().GetTimeBucket() != nil
aggFunc, aggErr := distributedAggFunc(req.GetAgg().GetFunction())
if aggErr != nil {
return nil, aggErr
@@ -651,7 +652,7 @@ func (p *DistributedPlan) executeAgg(ctx context.Context, frames [][]byte, req *
addTraceTagf(reduceSpan, tracelabels.TagFramesIn, "%d", len(frames))
frameDecodeDurations := collectFrameDecodeDurations(frames)
// nolint:contextcheck // pure in-memory reducer; no cancelable I/O downstream
- batches, aggValuePath, reduceErr := vmeasure.ReduceRawFrames(frames, keyTagNames, aggSpecs, p.cfg.BatchSize, tracker)
+ batches, aggValuePath, reduceErr := vmeasure.ReduceRawFrames(frames, keyTagFamily, keyTagNames, bucketed, aggSpecs, p.cfg.BatchSize, tracker)
if reduceErr != nil {
if reduceSpan != nil {
reduceSpan.Error(reduceErr)
@@ -1140,15 +1141,24 @@ func aggOutputName(agg *measurev1.QueryRequest_Aggregation) string {
return agg.GetFieldName()
}
-func distributedGroupByTagNames(groupBy *measurev1.QueryRequest_GroupBy) []string {
+// distributedGroupByTagKey extracts the GroupBy key tag family and names
+// from req.GroupBy.TagProjection (first family, v1 single-family
+// limitation — same convention as applyBatchGroupByFirstToRows). The family
+// must travel with the names: tag-family validation does not reject the
+// same tag name in two families (design doc §5.1), so a bare name is
+// ambiguous on a valid schema and the liaison-side reduce must resolve key
+// columns the same (family, name) way the node-side GroupBy already does
+// (plan.go's lookupGroupByKeyIndices, distributed.go's
+// applyBatchGroupByFirstToRows).
+func distributedGroupByTagKey(groupBy *measurev1.QueryRequest_GroupBy) (family string, names []string) {
if groupBy == nil || groupBy.GetTagProjection() == nil {
- return nil
+ return "", nil
}
families := groupBy.GetTagProjection().GetTagFamilies()
if len(families) == 0 {
- return nil
+ return "", nil
}
- return append([]string(nil), families[0].GetTags()...)
+ return families[0].GetName(), append([]string(nil), families[0].GetTags()...)
}
func distributedAggFunc(fn modelv1.AggregationFunction) (vmeasure.AggFunc, error) {
diff --git a/pkg/query/vectorized/measure/plan/distributed_test.go b/pkg/query/vectorized/measure/plan/distributed_test.go
index 0d210f1e7..416a7b393 100644
--- a/pkg/query/vectorized/measure/plan/distributed_test.go
+++ b/pkg/query/vectorized/measure/plan/distributed_test.go
@@ -220,6 +220,40 @@ func TestAnalyzeDistributed_NodeTemplatePushesAggPartials_TagTarget(t *testing.T
}
}
+// TestDistributedGroupByTagKey_PreservesFamily is the regression pin for
+// the PR review finding on executeAgg's reduce call: distributedGroupByTagKey
+// must return the tag family alongside the names, not just names, because
+// tag-family validation does not reject the same tag name in two families
+// (design doc §5.1) — dropping the family would let resolveKeyIndices bind
+// a bucketed distributed reduce to the wrong family's column on a valid
+// schema with a repeated tag name.
+func TestDistributedGroupByTagKey_PreservesFamily(t *testing.T) {
+ groupBy := &measurev1.QueryRequest_GroupBy{
+ TagProjection: &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{
+ {Name: "fam2", Tags: []string{"g"}},
+ }},
+ }
+ family, names := distributedGroupByTagKey(groupBy)
+ if family != "fam2" {
+ t.Fatalf("family = %q, want %q", family, "fam2")
+ }
+ if len(names) != 1 || names[0] != "g" {
+ t.Fatalf("names = %v, want [g]", names)
+ }
+}
+
+// TestDistributedGroupByTagKey_NilGroupBy pins the nil/empty-projection
+// no-op cases return an empty family alongside nil names, matching
+// resolveKeyIndices's existing "no keyTagNames" contract.
+func TestDistributedGroupByTagKey_NilGroupBy(t *testing.T) {
+ if family, names := distributedGroupByTagKey(nil); family != "" || names != nil {
+ t.Fatalf("nil GroupBy: got (%q, %v), want (\"\", nil)", family, names)
+ }
+ if family, names := distributedGroupByTagKey(&measurev1.QueryRequest_GroupBy{}); family != "" || names != nil {
+ t.Fatalf("no TagProjection: got (%q, %v), want (\"\", nil)", family, names)
+ }
+}
+
// TestAggOutputName_ResolvesTagOrFieldTarget is the regression pin for the
// distributed-reduce OutputName bug this issue fixed: executeAgg used to
// read req.GetAgg().GetFieldName() unconditionally, which is empty for a
@@ -236,6 +270,33 @@ func TestAggOutputName_ResolvesTagOrFieldTarget(t *testing.T) {
}
}
+// TestAnalyzeDistributed_NodeTemplatePushesTimeBucket pins design §7.2's
+// distributed seam: GroupBy.time_bucket travels to the node template
+// unchanged (proto.Clone carries it verbatim), so each data node floors
+// independently and no liaison-side coordination is needed to distribute it.
+func TestAnalyzeDistributed_NodeTemplatePushesTimeBucket(t *testing.T) {
+ req := &measurev1.QueryRequest{
+ Name: "demo",
+ TagProjection: projTagProj(),
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
+ GroupBy: &measurev1.QueryRequest_GroupBy{
+ TagProjection: projTagProj(),
+ TimeBucket: &measurev1.QueryRequest_GroupBy_TimeBucket{Width: "5m"},
+ },
+ Agg: &measurev1.QueryRequest_Aggregation{
+ Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
+ FieldName: fieldValue,
+ },
+ }
+ p, analyzeErr := AnalyzeDistributed(req, []*databasev1.Measure{testMeasureSchema()}, nil, vmeasure.VectorizedConfig{BatchSize: 4, QueryMemoryMiB: 1})
+ if analyzeErr != nil {
+ t.Fatalf("AnalyzeDistributed: %v", analyzeErr)
+ }
+ if got := p.nodeTemplate.GetGroupBy().GetTimeBucket().GetWidth(); got != "5m" {
+ t.Fatalf("node template should push time_bucket unchanged, got width %q", got)
+ }
+}
+
// TestAnalyzeDistributed_TopAggUnboundsNodeLimit_Matrix is the regression
// gate for the per-node Limit truncation bug in distributed Top-over-Agg:
// each (Top.N, request Limit) combination must produce a node template with
diff --git a/pkg/query/vectorized/measure/plan/groupby_agg.go b/pkg/query/vectorized/measure/plan/groupby_agg.go
index 076785b74..10d3cc9d7 100644
--- a/pkg/query/vectorized/measure/plan/groupby_agg.go
+++ b/pkg/query/vectorized/measure/plan/groupby_agg.go
@@ -70,8 +70,9 @@ func NewGroupByAgg(child VecPlan, groupBy *model.MeasureGroupBy, agg *model.Meas
}
// Schema returns the aggregation output schema. The schema is computed
-// lazily on first call by running BuildOperators against the child
-// schema; subsequent calls return the cached value.
+// lazily on first call by running BuildOperators (or, for a time-bucketed
+// GroupBy, BuildTimeBucketOperator with a nil upstream — see its doc)
+// against the child schema; subsequent calls return the cached value.
func (g *GroupByAgg) Schema() *vectorized.BatchSchema {
if g.outputCache != nil {
return g.outputCache
@@ -88,6 +89,14 @@ func (g *GroupByAgg) Schema() *vectorized.BatchSchema {
// A throwaway tracker; we only need the resulting operator's
// OutputSchema, not its bookkeeping.
tracker := vectorized.NewMemoryTracker(1 << 30)
+ if g.GroupBy != nil && g.GroupBy.TimeBucket != nil {
+ bucket, err := vmeasure.BuildTimeBucketOperator(nil, opts, inputSchema, tracker, 1024, g.Mode)
+ if err != nil {
+ return nil
+ }
+ g.outputCache = bucket.OutputSchema()
+ return g.outputCache
+ }
ops, err := vmeasure.BuildOperators(opts, inputSchema, tracker, 1024, g.Mode)
if err != nil || len(ops) != 1 {
return nil
@@ -99,15 +108,38 @@ func (g *GroupByAgg) Schema() *vectorized.BatchSchema {
// Children returns the single child.
func (g *GroupByAgg) Children() []VecPlan { return []VecPlan{g.Child} }
-// Build recurses into child, then constructs the BatchAggregation via
-// BuildOperators and attaches it as a breaker. The pipeline-shared
-// MemoryTracker from bc threads through.
+// Build recurses into child, then attaches the aggregation operator. A
+// time-bucketed GroupBy wires BatchTimeBucket in via
+// PipelineBuilder.Transform, since it is a PullOperator that pulls from
+// upstream directly rather than a BreakerOperator driven by Break's
+// Consume/Finalize contract (see BatchTimeBucket's doc for why that
+// distinction is what makes its memory bound real). Every other shape
+// attaches its BatchAggregation/BatchGroupBy via Break, as before. The
+// pipeline-shared MemoryTracker from bc threads through either way.
func (g *GroupByAgg) Build(ctx context.Context, bc *BuildContext) error {
if buildErr := g.Child.Build(ctx, bc); buildErr != nil {
return buildErr
}
inputSchema := g.Child.Schema()
opts := model.MeasureQueryOptions{GroupBy: g.GroupBy, Agg: g.Agg}
+
+ if g.GroupBy != nil && g.GroupBy.TimeBucket != nil {
+ var bucket *vmeasure.BatchTimeBucket
+ var buildErr error
+ bc.Builder.Transform(func(upstream vectorized.PullOperator) vectorized.PullOperator {
+ bucket, buildErr = vmeasure.BuildTimeBucketOperator(upstream, opts, inputSchema, bc.Tracker, bc.Config.BatchSize, g.Mode)
+ if buildErr != nil {
+ return upstream
+ }
+ return bucket
+ })
+ if buildErr != nil {
+ return fmt.Errorf("plan.GroupByAgg.Build: %w", buildErr)
+ }
+ g.outputCache = bucket.OutputSchema()
+ return nil
+ }
+
ops, opsErr := vmeasure.BuildOperators(opts, inputSchema, bc.Tracker, bc.Config.BatchSize, g.Mode)
if opsErr != nil {
return fmt.Errorf("plan.GroupByAgg.Build: %w", opsErr)
diff --git a/pkg/query/vectorized/measure/plan_test.go b/pkg/query/vectorized/measure/plan_test.go
index e3fe67341..18a2f386b 100644
--- a/pkg/query/vectorized/measure/plan_test.go
+++ b/pkg/query/vectorized/measure/plan_test.go
@@ -120,6 +120,67 @@ func TestBuildOperators_GroupByWithoutAgg_EmitsFirstOnlyGroupBy(t *testing.T) {
}
}
+// TestBuildTimeBucketOperator_BucketOnly_ReturnsBatchTimeBucket pins the
+// bucket-only GroupBy shape (design §7.2): no tag key at all, grouping
+// purely by the resolved bucket width. This exercises BuildTimeBucketOperator
+// directly (opts.Agg nil) to pin the mechanical routing; plan/analyzer.go's
+// Analyze rejects this combination for real requests today (a bucketed raw
+// GroupBy has no execution support — BatchAggregation's empty-AggSpec output
+// layout drops every projected field, unlike BatchGroupByFirst's
+// full-schema passthrough for the non-bucketed case), so this shape isn't
+// reachable outside a direct BuildTimeBucketOperator call like this one.
+// upstream is nil: this test never calls Init/NextBatch on the result.
+func TestBuildTimeBucketOperator_BucketOnly_ReturnsBatchTimeBucket(t *testing.T) {
+ opts := model.MeasureQueryOptions{
+ GroupBy: &model.MeasureGroupBy{TimeBucket: &model.MeasureTimeBucket{WidthNanos: 1000}},
+ }
+ bucket, err := BuildTimeBucketOperator(nil, opts, bucketTestSchema(), vectorized.NewMemoryTracker(1<<20), 1024, AggModeAll)
+ if err != nil {
+ t.Fatalf("BuildTimeBucketOperator: %v", err)
+ }
+ if bucket == nil {
+ t.Fatal("want a non-nil *BatchTimeBucket")
+ }
+}
+
+// TestBuildTimeBucketOperator_BucketPlusTagGroupByPlusAgg pins the combined
+// shape: bucket + tag GroupBy + Agg all route to the same BatchTimeBucket
+// operator (design §7.2), which internally binds the agg spec exactly like
+// the non-bucketed path.
+func TestBuildTimeBucketOperator_BucketPlusTagGroupByPlusAgg(t *testing.T) {
+ opts := model.MeasureQueryOptions{
+ GroupBy: &model.MeasureGroupBy{
+ TagFamily: "default",
+ TagNames: []string{"g"},
+ TimeBucket: &model.MeasureTimeBucket{WidthNanos: 1000},
+ },
+ Agg: &model.MeasureAgg{FieldName: "v", Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ }
+ bucket, err := BuildTimeBucketOperator(nil, opts, bucketTestSchema(), vectorized.NewMemoryTracker(1<<20), 1024, AggModeAll)
+ if err != nil {
+ t.Fatalf("BuildTimeBucketOperator: %v", err)
+ }
+ if bucket == nil {
+ t.Fatal("want a non-nil *BatchTimeBucket")
+ }
+}
+
+// TestBuildTimeBucketOperator_MissingTimestampColumn_Errors pins the
+// defensive guard: a schema with no RoleTimestamp column (should never
+// happen in production — BuildBatchSchema always emits one) is rejected
+// rather than panicking downstream.
+func TestBuildTimeBucketOperator_MissingTimestampColumn_Errors(t *testing.T) {
+ schema := vectorized.NewBatchSchema([]vectorized.ColumnDef{
+ {Role: vectorized.RoleTag, TagFamily: "default", Name: "g", Type: vectorized.ColumnTypeString},
+ })
+ opts := model.MeasureQueryOptions{
+ GroupBy: &model.MeasureGroupBy{TimeBucket: &model.MeasureTimeBucket{WidthNanos: 1000}},
+ }
+ if _, err := BuildTimeBucketOperator(nil, opts, schema, vectorized.NewMemoryTracker(1<<20), 1024, AggModeAll); err == nil {
+ t.Fatal("a schema with no RoleTimestamp column must error, not panic")
+ }
+}
+
// TestBuildOperators_AggWithoutGroupBy_EmitsBatchAggregation pins the
// scalar-reduce shape: a BatchAggregation with no key columns, so every
// row collapses into a single output row carrying the first-seen tags
diff --git a/pkg/query/vectorized/measure/raw_emit.go b/pkg/query/vectorized/measure/raw_emit.go
index ed4a43e17..ca21d4ac7 100644
--- a/pkg/query/vectorized/measure/raw_emit.go
+++ b/pkg/query/vectorized/measure/raw_emit.go
@@ -588,13 +588,17 @@ func appendActive(dst, src *vectorized.RecordBatch) {
// an empty distributed result.
func ReduceFramesToInternalDataPoints(
frames [][]byte,
+ keyTagFamily string,
keyTagNames []string,
aggSpecs []AggReduceSpec,
topSpec *ReduceTopSpec,
batchSize int,
tracker *vectorized.MemoryTracker,
) ([]*measurev1.InternalDataPoint, error) {
- reduced, _, reduceErr := ReduceRawFrames(frames, keyTagNames, aggSpecs, batchSize, tracker)
+ // This bridge predates time bucketing (see the doc comment above) and
+ // has no caller that could ever set a bucketed GroupBy through it, so
+ // it is unconditionally not bucketed.
+ reduced, _, reduceErr := ReduceRawFrames(frames, keyTagFamily, keyTagNames, false, aggSpecs, batchSize, tracker)
if reduceErr != nil {
return nil, reduceErr
}
diff --git a/pkg/query/vectorized/measure/raw_emit_bench_test.go b/pkg/query/vectorized/measure/raw_emit_bench_test.go
index c8084e2c9..b29996334 100644
--- a/pkg/query/vectorized/measure/raw_emit_bench_test.go
+++ b/pkg/query/vectorized/measure/raw_emit_bench_test.go
@@ -56,7 +56,7 @@ func BenchmarkG9f_DistributedFanout(b *testing.B) {
b.ReportAllocs()
for i := 0; i < b.N; i++ {
idps, err := ReduceFramesToInternalDataPoints(frames,
- []string{"g"},
+ "default", []string{"g"},
[]AggReduceSpec{{OutputName: "out", Func: AggSum}},
nil, 1024, vectorized.NewMemoryTracker(1<<30))
if err != nil {
diff --git a/pkg/query/vectorized/measure/raw_emit_test.go b/pkg/query/vectorized/measure/raw_emit_test.go
index c95886e0c..a9449e4c0 100644
--- a/pkg/query/vectorized/measure/raw_emit_test.go
+++ b/pkg/query/vectorized/measure/raw_emit_test.go
@@ -131,7 +131,7 @@ func TestReduceFramesToInternalDataPoints_AggRoundTrip(t *testing.T) {
frames := [][]byte{shard1Body, shard1Body, shard2Body, nil}
idps, reduceErr := ReduceFramesToInternalDataPoints(frames,
- []string{"g"},
+ "default", []string{"g"},
[]AggReduceSpec{{OutputName: "sum_v", Func: AggSum}},
nil, 1024, vectorized.NewMemoryTracker(1<<30))
if reduceErr != nil {
@@ -376,7 +376,7 @@ func TestReduceFramesToInternalDataPoints_MeanInt(t *testing.T) {
}
idps, reduceErr := ReduceFramesToInternalDataPoints([][]byte{body},
- []string{"g"},
+ "default", []string{"g"},
[]AggReduceSpec{{OutputName: "value", Func: AggMean}},
nil, 1024, vectorized.NewMemoryTracker(1<<30))
if reduceErr != nil {
diff --git a/pkg/query/vectorized/measure/reduce.go b/pkg/query/vectorized/measure/reduce.go
index f95daa663..4a3b04df1 100644
--- a/pkg/query/vectorized/measure/reduce.go
+++ b/pkg/query/vectorized/measure/reduce.go
@@ -66,18 +66,31 @@ type ReduceTopSpec struct {
// any structural mismatch is a producer bug, not a recoverable data error,
// and is reported loudly.
//
-// keyTagNames selects which tags form the group key — the same names the
-// request's GroupBy used. Tags present in the partial schema but NOT
-// listed here are still carried forward in the output as the first-seen
-// value per group (mirrors the data-node operator's first-seen non-key
-// tag rule).
+// keyTagFamily and keyTagNames select which tags form the group key — the
+// same family and names the request's GroupBy used. The family must travel
+// with the names: tag-family validation does not reject the same tag name
+// in two families (design doc §5.1), so resolveKeyIndices matches on
+// (family, name), not name alone — matching the node-side GroupBy's own
+// (plan.go's lookupGroupByKeyIndices). Tags present in the partial schema
+// but NOT listed here are still carried forward in the output as the
+// first-seen value per group (mirrors the data-node operator's first-seen
+// non-key tag rule).
+//
+// bucketed is true for a time-bucketed GroupBy (design §7.2): the bucket
+// column is resolved as an additional leading key, ahead of keyTagNames.
+// A bucketed reduce whose reference schema carries no RoleTimestamp column
+// is the mixed-version case (design §10) — an older data node ignored
+// time_bucket and returned whole-range partials — and resolveKeyIndices
+// hard-errors rather than silently collapsing the series to one row.
//
// Returns the final reduced batches in group-insertion order (paginated
// by batchSize) and the AggValuePath that describes how the value column
// was resolved. Callers walk the batches sequentially; one row per group.
func ReduceRawFrames(
frames [][]byte,
+ keyTagFamily string,
keyTagNames []string,
+ bucketed bool,
aggSpecs []AggReduceSpec,
batchSize int,
tracker *vectorized.MemoryTracker,
@@ -93,7 +106,7 @@ func ReduceRawFrames(
}
decoded = append(decoded, b)
}
- return ReducePartialBatches(decoded, keyTagNames, aggSpecs, batchSize, tracker)
+ return ReducePartialBatches(decoded, keyTagFamily, keyTagNames, bucketed, aggSpecs, batchSize, tracker)
}
// ReducePartialBatches is the in-memory counterpart of ReduceRawFrames —
@@ -106,7 +119,9 @@ func ReduceRawFrames(
// value column was resolved (typed, fieldvalue-fallback, or unresolved).
func ReducePartialBatches(
partials []*vectorized.RecordBatch,
+ keyTagFamily string,
keyTagNames []string,
+ bucketed bool,
aggSpecs []AggReduceSpec,
batchSize int,
tracker *vectorized.MemoryTracker,
@@ -122,7 +137,7 @@ func ReducePartialBatches(
if refSchema == nil {
return nil, AggValuePathTyped, nil
}
- keyIndices, indicesErr := resolveKeyIndices(refSchema, keyTagNames)
+ keyIndices, indicesErr := resolveKeyIndices(refSchema, keyTagFamily, keyTagNames, bucketed)
if indicesErr != nil {
return nil, AggValuePathUnresolved, indicesErr
}
@@ -149,6 +164,14 @@ func ReducePartialBatches(
if finalErr := op.Finalize(context.Background()); finalErr != nil {
return nil, path, fmt.Errorf("ReducePartialBatches: finalize: %w", finalErr)
}
+ // Each partial arrives already bucket-ascending (design §7.2 streams
+ // per node), but op consumes partials frame by frame, so insertion
+ // order is only piecewise ascending — e.g. node A's [2000, 3000]
+ // followed by node B's [1000, 2000] inserts as [2000, 3000, 1000].
+ // Sort once before draining so the merged result is globally
+ // bucket-ascending, which iteratorFromBatches's offset/limit pagination
+ // (design §7.5) requires to page correctly. A no-op when unbucketed.
+ op.SortInsertionByBucket()
var out []*vectorized.RecordBatch
for {
nb, nextErr := op.NextBatch(context.Background())
@@ -163,24 +186,34 @@ func ReducePartialBatches(
return out, path, nil
}
-// resolveKeyIndices binds keyTagNames to column indices in the partial
-// schema. Returns an error if a name does not resolve. Empty list is
-// allowed (scalar reduce) and returns nil keyIndices.
-func resolveKeyIndices(schema *vectorized.BatchSchema, keyTagNames []string) ([]int, error) {
- if len(keyTagNames) == 0 {
- return nil, nil
+// resolveKeyIndices binds the bucket column (when bucketed) and
+// (keyTagFamily, keyTagNames) to column indices in the partial schema.
+// Matching requires both family and name — tag-family validation does not
+// reject the same tag name in two families (design doc §5.1), so matching
+// by name alone could silently bind to the wrong family's column on a
+// valid schema. Returns an error if a tag does not resolve, or if bucketed
+// is true but the schema carries no RoleTimestamp column (design §10's
+// mixed-version guard). Neither bucketed nor any keyTagNames is allowed
+// (scalar reduce) and returns nil keyIndices; keyTagFamily is unused then.
+func resolveKeyIndices(schema *vectorized.BatchSchema, keyTagFamily string, keyTagNames []string, bucketed bool) ([]int, error) {
+ var out []int
+ if bucketed {
+ idx := schema.TimestampIndex()
+ if idx < 0 {
+ return nil, fmt.Errorf("bucketed query expected a RoleTimestamp column in the partial schema (an older data node may have ignored time_bucket)")
+ }
+ out = append(out, idx)
}
- out := make([]int, 0, len(keyTagNames))
for _, name := range keyTagNames {
idx := -1
for i, def := range schema.Columns {
- if def.Role == vectorized.RoleTag && def.Name == name {
+ if def.Role == vectorized.RoleTag && def.TagFamily == keyTagFamily && def.Name == name {
idx = i
break
}
}
if idx < 0 {
- return nil, fmt.Errorf("key tag %q not present in partial schema", name)
+ return nil, fmt.Errorf("key tag %s.%s not present in partial schema", keyTagFamily, name)
}
out = append(out, idx)
}
diff --git a/pkg/query/vectorized/measure/reduce_test.go b/pkg/query/vectorized/measure/reduce_test.go
new file mode 100644
index 000000000..c4673187e
--- /dev/null
+++ b/pkg/query/vectorized/measure/reduce_test.go
@@ -0,0 +1,283 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "testing"
+
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized"
+)
+
+// bucketPartialSchema is "shard_id, timestamp (bucket), tag.default.g,
+// field out (sum partial)" — the AggModeMap partial shape for a bucketed
+// tag GroupBy + Agg (design §7.2's leading-timestamp-key layout).
+func bucketPartialSchema() *vectorized.BatchSchema {
+ return vectorized.NewBatchSchema([]vectorized.ColumnDef{
+ {Role: vectorized.RoleShardID, Name: shardIDOutputName, Type: vectorized.ColumnTypeInt64},
+ {Role: vectorized.RoleTimestamp, Type: vectorized.ColumnTypeInt64},
+ {Role: vectorized.RoleTag, TagFamily: "default", Name: "g", Type: vectorized.ColumnTypeString},
+ {Role: vectorized.RoleField, Name: "out", Type: vectorized.ColumnTypeInt64},
+ })
+}
+
+// bucketPartialRow is one input row for bucketPartialBatch.
+type bucketPartialRow struct {
+ g string
+ shard int64
+ ts int64
+ v int64
+}
+
+func bucketPartialBatch(rows ...bucketPartialRow) *vectorized.RecordBatch {
+ s := bucketPartialSchema()
+ b := vectorized.NewRecordBatch(s, len(rows))
+ shardCol := b.Columns[0].(*vectorized.TypedColumn[int64])
+ tsCol := b.Columns[1].(*vectorized.TypedColumn[int64])
+ gCol := b.Columns[2].(*vectorized.TypedColumn[string])
+ vCol := b.Columns[3].(*vectorized.TypedColumn[int64])
+ for _, r := range rows {
+ shardCol.Append(r.shard)
+ tsCol.Append(r.ts)
+ gCol.Append(r.g)
+ vCol.Append(r.v)
+ }
+ b.Len = len(rows)
+ return b
+}
+
+// TestResolveKeyIndices_Bucketed_ResolvesTimestampAsLeadingKey pins design
+// §7.2's distributed seam: bucketed=true resolves the RoleTimestamp column
+// index ahead of any tag key indices.
+func TestResolveKeyIndices_Bucketed_ResolvesTimestampAsLeadingKey(t *testing.T) {
+ s := bucketPartialSchema()
+ idx, err := resolveKeyIndices(s, "default", []string{"g"}, true)
+ if err != nil {
+ t.Fatalf("resolveKeyIndices: %v", err)
+ }
+ if len(idx) != 2 || idx[0] != 1 || idx[1] != 2 {
+ t.Fatalf("keyIndices = %v, want [1 2] (timestamp then tag g)", idx)
+ }
+}
+
+// TestResolveKeyIndices_BucketOnly_NoTagKeys pins the bucket-only GroupBy
+// case: bucketed=true with no keyTagNames still resolves a non-nil
+// keyIndices (grouping by bucket alone), not the scalar-reduce nil.
+func TestResolveKeyIndices_BucketOnly_NoTagKeys(t *testing.T) {
+ s := bucketPartialSchema()
+ idx, err := resolveKeyIndices(s, "default", nil, true)
+ if err != nil {
+ t.Fatalf("resolveKeyIndices: %v", err)
+ }
+ if len(idx) != 1 || idx[0] != 1 {
+ t.Fatalf("keyIndices = %v, want [1] (timestamp only)", idx)
+ }
+}
+
+// TestResolveKeyIndices_Bucketed_MissingTimestampColumn_Errors pins design
+// §10's mixed-version guard: a bucketed reduce whose schema carries no
+// RoleTimestamp column — an older data node ignored time_bucket — hard-errors
+// rather than silently collapsing the series to one row per tag group.
+func TestResolveKeyIndices_Bucketed_MissingTimestampColumn_Errors(t *testing.T) {
+ s := vectorized.NewBatchSchema([]vectorized.ColumnDef{
+ {Role: vectorized.RoleShardID, Name: shardIDOutputName, Type: vectorized.ColumnTypeInt64},
+ {Role: vectorized.RoleTag, TagFamily: "default", Name: "g", Type: vectorized.ColumnTypeString},
+ {Role: vectorized.RoleField, Name: "out", Type: vectorized.ColumnTypeInt64},
+ })
+ if _, err := resolveKeyIndices(s, "default", []string{"g"}, true); err == nil {
+ t.Fatal("a bucketed reduce over a schema with no RoleTimestamp column must error")
+ }
+}
+
+// TestResolveKeyIndices_DuplicateNameAcrossFamilies_ResolvesRequestedFamily
+// pins the PR review finding that motivated adding keyTagFamily:
+// tag-family validation does not reject the same tag name in two families
+// (design doc §5.1), so a valid schema can have "g" in both "fam1" and
+// "fam2". Matching by name alone would silently bind to whichever family
+// happens to come first in schema.Columns, corrupting a distributed
+// bucketed reduce whose GroupBy actually targets the other family. The fix
+// must resolve the column belonging to the REQUESTED family regardless of
+// column order.
+func TestResolveKeyIndices_DuplicateNameAcrossFamilies_ResolvesRequestedFamily(t *testing.T) {
+ s := vectorized.NewBatchSchema([]vectorized.ColumnDef{
+ {Role: vectorized.RoleShardID, Name: shardIDOutputName, Type: vectorized.ColumnTypeInt64},
+ {Role: vectorized.RoleTag, TagFamily: "fam1", Name: "g", Type: vectorized.ColumnTypeString},
+ {Role: vectorized.RoleTag, TagFamily: "fam2", Name: "g", Type: vectorized.ColumnTypeString},
+ {Role: vectorized.RoleField, Name: "out", Type: vectorized.ColumnTypeInt64},
+ })
+ idx, err := resolveKeyIndices(s, "fam2", []string{"g"}, false)
+ if err != nil {
+ t.Fatalf("resolveKeyIndices: %v", err)
+ }
+ if len(idx) != 1 || idx[0] != 2 {
+ t.Fatalf("keyIndices = %v, want [2] (fam2's g column, not fam1's at index 1)", idx)
+ }
+}
+
+// TestResolveKeyIndices_NotBucketed_Unaffected pins that bucketed=false
+// preserves the pre-existing behavior exactly (no timestamp resolution).
+func TestResolveKeyIndices_NotBucketed_Unaffected(t *testing.T) {
+ idx, err := resolveKeyIndices(bucketPartialSchema(), "default", nil, false)
+ if err != nil {
+ t.Fatalf("resolveKeyIndices: %v", err)
+ }
+ if idx != nil {
+ t.Fatalf("keyIndices = %v, want nil (scalar reduce, unaffected by an unrelated timestamp column)", idx)
+ }
+}
+
+// TestReducePartialBatches_Bucketed_CombinesAcrossShards pins the
+// distributed reduce end to end: two shards' partials for the same
+// (bucket, tag) combine into one summed row, and a different bucket stays
+// separate.
+func TestReducePartialBatches_Bucketed_CombinesAcrossShards(t *testing.T) {
+ p1 := bucketPartialBatch(
+ bucketPartialRow{shard: 1, ts: 0, g: "a", v: 3},
+ bucketPartialRow{shard: 1, ts: 1000, g: "a", v: 7},
+ )
+ p2 := bucketPartialBatch(
+ bucketPartialRow{shard: 2, ts: 0, g: "a", v: 4},
+ )
+ batches, _, err := ReducePartialBatches(
+ []*vectorized.RecordBatch{p1, p2},
+ "default", []string{"g"}, true,
+ []AggReduceSpec{{OutputName: "out", Func: AggSum}},
+ 64, vectorized.NewMemoryTracker(1<<30),
+ )
+ if err != nil {
+ t.Fatalf("ReducePartialBatches: %v", err)
+ }
+ got := map[string]int64{}
+ for _, b := range batches {
+ tsCol := b.Columns[0].(*vectorized.TypedColumn[int64])
+ gCol := b.Columns[1].(*vectorized.TypedColumn[string])
+ vCol := b.Columns[2].(*vectorized.TypedColumn[int64])
+ for i := 0; i < b.Len; i++ {
+ got[formatBucketKey(tsCol.Data()[i], gCol.Data()[i])] = vCol.Data()[i]
+ }
+ }
+ want := map[string]int64{"a|0": 7, "a|1000": 7}
+ for k, v := range want {
+ if got[k] != v {
+ t.Errorf("sum[%s] = %d, want %d (full result: %v)", k, got[k], v, got)
+ }
+ }
+ if len(got) != len(want) {
+ t.Fatalf("result set size = %d, want %d: %v", len(got), len(want), got)
+ }
+}
+
+// TestReducePartialBatches_Bucketed_GloballySortsAcrossNodePartials pins the
+// review finding that motivated BatchAggregation.SortInsertionByBucket:
+// each node partial arrives already bucket-ascending (design §7.2 streams
+// per node), but consuming them frame by frame only yields piecewise
+// ascending insertion order — node A's [2000, 3000] followed by node B's
+// [1000, 2000] would insert as [2000, 3000, 1000] without a final sort.
+// iteratorFromBatches's offset/limit pagination (design §7.5) requires the
+// reduced result to be globally bucket-ascending, not just per-partial.
+func TestReducePartialBatches_Bucketed_GloballySortsAcrossNodePartials(t *testing.T) {
+ nodeA := bucketPartialBatch(
+ bucketPartialRow{shard: 1, ts: 2000, g: "a", v: 2},
+ bucketPartialRow{shard: 1, ts: 3000, g: "a", v: 3},
+ )
+ nodeB := bucketPartialBatch(
+ bucketPartialRow{shard: 2, ts: 1000, g: "a", v: 1},
+ bucketPartialRow{shard: 2, ts: 2000, g: "a", v: 20},
+ )
+ batches, _, err := ReducePartialBatches(
+ []*vectorized.RecordBatch{nodeA, nodeB},
+ "default", []string{"g"}, true,
+ []AggReduceSpec{{OutputName: "out", Func: AggSum}},
+ 64, vectorized.NewMemoryTracker(1<<30),
+ )
+ if err != nil {
+ t.Fatalf("ReducePartialBatches: %v", err)
+ }
+ var gotTS []int64
+ for _, b := range batches {
+ tsCol := b.Columns[0].(*vectorized.TypedColumn[int64])
+ for i := 0; i < b.Len; i++ {
+ gotTS = append(gotTS, tsCol.Data()[i])
+ }
+ }
+ wantTS := []int64{1000, 2000, 3000}
+ if len(gotTS) != len(wantTS) {
+ t.Fatalf("bucket timestamps = %v, want %v", gotTS, wantTS)
+ }
+ for i, want := range wantTS {
+ if gotTS[i] != want {
+ t.Fatalf("bucket timestamps = %v, want globally ascending %v", gotTS, wantTS)
+ }
+ }
+}
+
+// TestReducePartialBatches_Bucketed_MixedVersionOldNodeFirst_Errors pins
+// design §10: when the first non-empty partial lacks a RoleTimestamp
+// column (an old node that ignored time_bucket), the reduce hard-errors
+// immediately rather than silently reducing on tags alone.
+func TestReducePartialBatches_Bucketed_MixedVersionOldNodeFirst_Errors(t *testing.T) {
+ oldNodeSchema := vectorized.NewBatchSchema([]vectorized.ColumnDef{
+ {Role: vectorized.RoleShardID, Name: shardIDOutputName, Type: vectorized.ColumnTypeInt64},
+ {Role: vectorized.RoleTag, TagFamily: "default", Name: "g", Type: vectorized.ColumnTypeString},
+ {Role: vectorized.RoleField, Name: "out", Type: vectorized.ColumnTypeInt64},
+ })
+ oldPartial := vectorized.NewRecordBatch(oldNodeSchema, 1)
+ oldPartial.Columns[0].(*vectorized.TypedColumn[int64]).Append(1)
+ oldPartial.Columns[1].(*vectorized.TypedColumn[string]).Append("a")
+ oldPartial.Columns[2].(*vectorized.TypedColumn[int64]).Append(5)
+ oldPartial.Len = 1
+
+ _, _, err := ReducePartialBatches(
+ []*vectorized.RecordBatch{oldPartial},
+ "default", []string{"g"}, true,
+ []AggReduceSpec{{OutputName: "out", Func: AggSum}},
+ 64, vectorized.NewMemoryTracker(1<<30),
+ )
+ if err == nil {
+ t.Fatal("a bucketed reduce whose first partial has no RoleTimestamp column must error")
+ }
+}
+
+// TestReducePartialBatches_Bucketed_MixedVersionOldNodeSecond_Errors is the
+// other ordering of the same mixed-version scenario: a new node's partial
+// (with RoleTimestamp) defines the schema, and an old node's partial (without
+// it) arrives later — caught by the existing schemaCompatible structural
+// check rather than resolveKeyIndices.
+func TestReducePartialBatches_Bucketed_MixedVersionOldNodeSecond_Errors(t *testing.T) {
+ newPartial := bucketPartialBatch(bucketPartialRow{shard: 1, ts: 0, g: "a", v: 1})
+
+ oldNodeSchema := vectorized.NewBatchSchema([]vectorized.ColumnDef{
+ {Role: vectorized.RoleShardID, Name: shardIDOutputName, Type: vectorized.ColumnTypeInt64},
+ {Role: vectorized.RoleTag, TagFamily: "default", Name: "g", Type: vectorized.ColumnTypeString},
+ {Role: vectorized.RoleField, Name: "out", Type: vectorized.ColumnTypeInt64},
+ })
+ oldPartial := vectorized.NewRecordBatch(oldNodeSchema, 1)
+ oldPartial.Columns[0].(*vectorized.TypedColumn[int64]).Append(2)
+ oldPartial.Columns[1].(*vectorized.TypedColumn[string]).Append("a")
+ oldPartial.Columns[2].(*vectorized.TypedColumn[int64]).Append(5)
+ oldPartial.Len = 1
+
+ _, _, err := ReducePartialBatches(
+ []*vectorized.RecordBatch{newPartial, oldPartial},
+ "default", []string{"g"}, true,
+ []AggReduceSpec{{OutputName: "out", Func: AggSum}},
+ 64, vectorized.NewMemoryTracker(1<<30),
+ )
+ if err == nil {
+ t.Fatal("a schema-mismatched later partial must error")
+ }
+}
diff --git a/pkg/query/vectorized/measure/timebucket.go b/pkg/query/vectorized/measure/timebucket.go
new file mode 100644
index 000000000..5435077df
--- /dev/null
+++ b/pkg/query/vectorized/measure/timebucket.go
@@ -0,0 +1,465 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "context"
+ "fmt"
+ "slices"
+
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized"
+)
+
+// bucketAggPoolCapacity bounds the internal output-pool capacity of each
+// per-bucket BatchAggregation instance openBucket creates. That instance is
+// throwaway — its output is copied into the shared, properly batchSize-sized
+// outputBuilder (see appendOutput) and the instance is closed immediately
+// after — so its own pool capacity is an internal working-buffer size, not
+// the pipeline's output chunk size. Using bt.batchSize (often 1024) there
+// would allocate a full-capacity batch for even a one-group bucket; a small
+// bucket, the common sparse-time-series case, would waste most of it. A
+// bucket with more groups than this simply pages internally (openBucket's
+// caller already loops NextBatch to drain every page), so this only trades
+// a little pagination overhead for dense buckets against a large reduction
+// in wasted allocation for sparse ones.
+const bucketAggPoolCapacity = 64
+
+// bucketStart mirrors pkg/flow/streaming/sliding_window.go:302 getWindowStart
+// — the formula the TopN pre-aggregation path already uses to assign
+// tumbling windows. It is exactly DATE_BIN with a Unix-epoch origin.
+//
+// Negative timestamps are unreachable in practice (measure timestamps are
+// Unix milliseconds and the write path rejects non-positive ones), but Go's
+// % truncates toward zero rather than flooring, so getWindowStart's plain
+// "ts - ts%width" skews by one bucket for ts < 0. This corrects for that
+// rather than depending on a caller's invariant — the two formulas agree
+// exactly for ts >= 0, which is the entire reachable domain.
+func bucketStart(ts, width int64) int64 {
+ remainder := ts % width
+ if remainder < 0 {
+ remainder += width
+ }
+ return ts - remainder
+}
+
+// floorTimestampColumn rewrites schema's RoleTimestamp column in place,
+// replacing each non-null value with its bucket start. This is the "map
+// operator that rewrites the timestamp column to its bucket start" from
+// design §7.2: once floored, the column is an ordinary int64 group key and
+// needs no special handling from computeKey/appendKeyComponent.
+func floorTimestampColumn(b *vectorized.RecordBatch, timestampIdx int, widthNanos int64) {
+ col, ok := b.Columns[timestampIdx].(*vectorized.TypedColumn[int64])
+ if !ok {
+ return
+ }
+ data := col.Data()
+ for i, ts := range data {
+ if col.IsNull(i) {
+ continue
+ }
+ col.SetAt(i, bucketStart(ts, widthNanos))
+ }
+}
+
+// BatchTimeBucket is the time-bucketed counterpart of BatchAggregation
+// (design §7.2). It floors each incoming batch's timestamp column to the
+// bucket start, then delegates the actual grouping/folding to a sequence of
+// ordinary BatchAggregation instances — one per closed bucket in streaming
+// mode, or a single instance covering the whole scan in map mode — so every
+// bucket-aware behavior (the conditional D2 reversal, tag carry-forward,
+// AggModeMap/Reduce partial shapes) is exactly BatchAggregation's existing
+// behavior, never duplicated here.
+//
+// BatchTimeBucket is a PullOperator, not a BreakerOperator: it owns upstream
+// directly and pulls from it lazily inside NextBatch, rather than being
+// driven by breakerStage's "Consume the entire upstream, then serve"
+// contract. That distinction is load-bearing for the memory-bound claim
+// below — see NewBatchTimeBucket's wiring note.
+//
+// Streaming mode (the part-scan default) holds only the current bucket's
+// groups live at a time: consumeBatch splits each upstream batch into
+// contiguous same-bucket runs (the input is time-ascending, so a run, once
+// left behind, never recurs) and flushes+closes the open BatchAggregation
+// whenever the bucket advances, bounding live group state to one bucket's
+// groups instead of buckets × tagGroups. Because NextBatch drains any
+// already-flushed output before pulling the next upstream batch, the
+// flushed-but-not-yet-emitted queue is bounded too: at most the buckets that
+// closed while processing a single upstream batch (typically one, since a
+// batch is usually a thin, densely-packed time slice), not every bucket the
+// whole scan will ever produce. A row whose bucket regresses is a
+// monotonicity violation — streaming is correct only while the input is
+// time-ordered — and consumeBatch fails loudly rather than silently
+// reopening a flushed bucket.
+//
+// Map mode (index-mode measures, which have no ascending-timestamp
+// guarantee — design §7.2) never flushes mid-scan: every row folds into one
+// persistent BatchAggregation keyed on (bucket, tags…) like any other
+// GroupBy, and the whole result is produced once upstream is exhausted.
+// This trades the memory bound for correctness on out-of-order input;
+// NewBatchTimeBucket's streaming argument selects which mode a given
+// operator instance runs.
+type BatchTimeBucket struct {
+ upstream vectorized.PullOperator
+ inputSchema *vectorized.BatchSchema
+ outputSchema *vectorized.BatchSchema
+ tracker *vectorized.MemoryTracker
+ currentAgg *BatchAggregation
+ outputPool *vectorized.BatchPool
+ outputBuilder *vectorized.RecordBatch
+ pending []*vectorized.RecordBatch
+ aggKeyIndices []int
+ aggs []AggSpec
+ widthNanos int64
+ currentBucket int64
+ timestampIdx int
+ batchSize int
+ entrySize int64
+ mode AggMode
+ hasBucket bool
+ streaming bool
+ upstreamDone bool
+ mapDrainReady bool
+ closed bool
+}
+
+// NewBatchTimeBucket constructs a BatchTimeBucket wrapping upstream, the
+// operator this one pulls batches from directly (see the type doc for why
+// that — rather than a Consume/Finalize breaker contract — is what makes
+// the memory bound real). upstream may be nil for a schema-only probe (no
+// Init/NextBatch/Close call reaches it in that case) — GroupByAgg.Schema()
+// uses this to read OutputSchema without a real pipeline.
+//
+// keyIndices are the tag GroupBy keys (may be empty for a bucket-only
+// GroupBy); timestampIdx is the input schema's RoleTimestamp column;
+// widthNanos is the resolved bucket width (design §5.3), always > 0.
+// streaming selects the part-scan streaming path (true) or the index-mode
+// map fallback (false) — decided once by the analyzer from
+// measureSchema.GetIndexMode(), never guessed here. tracker and entrySize
+// are forwarded to each underlying BatchAggregation unchanged.
+func NewBatchTimeBucket(
+ upstream vectorized.PullOperator,
+ input *vectorized.BatchSchema, keyIndices []int, timestampIdx int, widthNanos int64,
+ aggs []AggSpec, mode AggMode, batchSize int, tracker *vectorized.MemoryTracker, entrySize int64,
+ streaming bool,
+) *BatchTimeBucket {
+ aggKeyIndices := append(append([]int(nil), keyIndices...), timestampIdx)
+ // A throwaway instance only to read its precomputed OutputSchema — mirrors
+ // GroupByAgg.Schema()'s existing pattern of probing shape without Init.
+ probe := NewBatchAggregation(input, aggKeyIndices, aggs, mode, batchSize, tracker, entrySize)
+ outputSchema := probe.OutputSchema()
+ return &BatchTimeBucket{
+ upstream: upstream,
+ inputSchema: input,
+ outputSchema: outputSchema,
+ tracker: tracker,
+ outputPool: vectorized.NewBatchPool(outputSchema, batchSize),
+ aggKeyIndices: aggKeyIndices,
+ aggs: slices.Clone(aggs),
+ widthNanos: widthNanos,
+ timestampIdx: timestampIdx,
+ batchSize: batchSize,
+ entrySize: entrySize,
+ mode: mode,
+ streaming: streaming,
+ }
+}
+
+// Init initializes upstream. The first bucket opens lazily on the first
+// NextBatch call.
+func (bt *BatchTimeBucket) Init(ctx context.Context) error { return bt.upstream.Init(ctx) }
+
+// OutputSchema returns the schema of emitted batches, precomputed at
+// construction time.
+func (bt *BatchTimeBucket) OutputSchema() *vectorized.BatchSchema { return bt.outputSchema }
+
+// NextBatch drains any already-flushed output first; only once that queue is
+// empty does it pull the next batch from upstream. This ordering is what
+// bounds the flushed-but-unemitted queue to "whatever closed while
+// processing one upstream batch" instead of "everything the whole scan will
+// ever produce" — see the type doc.
+//
+// Map mode's terminal aggregator is the one exception a per-upstream-batch
+// bound doesn't cover on its own: it can represent the entire scan, not one
+// bucket. Once upstream is exhausted, drainMapModeAggregator takes over and
+// pulls that aggregator one page at a time — interleaved with normal
+// pending/commitBuilder draining — rather than flushCurrentBucket's
+// loop-until-nil, which would materialize every page into pending before
+// this call ever returns the first one.
+func (bt *BatchTimeBucket) NextBatch(ctx context.Context) (*vectorized.RecordBatch, error) {
+ for {
+ if len(bt.pending) > 0 {
+ nb := bt.pending[0]
+ bt.pending = bt.pending[1:]
+ return nb, nil
+ }
+ if bt.upstreamDone {
+ if bt.currentAgg == nil {
+ return nil, nil
+ }
+ drained, drainErr := bt.drainMapModeAggregator(ctx)
+ if drainErr != nil {
+ return nil, drainErr
+ }
+ if !drained {
+ continue
+ }
+ bt.commitBuilder()
+ continue
+ }
+ b, pullErr := bt.upstream.NextBatch(ctx)
+ if pullErr != nil {
+ return nil, pullErr
+ }
+ if b == nil {
+ bt.upstreamDone = true
+ if bt.streaming {
+ // The scan has ended; flush whatever bucket is still
+ // open — bounded to one bucket's groups, same as any
+ // mid-scan advance-driven flush, so draining it in one
+ // shot here is fine.
+ if flushErr := bt.flushCurrentBucket(ctx); flushErr != nil {
+ return nil, flushErr
+ }
+ bt.commitBuilder()
+ }
+ // Map mode leaves currentAgg open; the loop's upstreamDone
+ // branch above drains it incrementally instead.
+ continue
+ }
+ if consumeErr := bt.consumeBatch(ctx, b); consumeErr != nil {
+ return nil, consumeErr
+ }
+ // Commit whatever the buckets that closed while processing this one
+ // upstream batch compacted into the builder — see commitBuilder's
+ // doc for why this boundary, not batchSize, is what must gate a
+ // commit: it is the same point at which the pre-compaction code
+ // already made queued output visible, so latency is unchanged.
+ bt.commitBuilder()
+ }
+}
+
+// drainMapModeAggregator pulls exactly one page from the terminal (map-mode)
+// aggregator into the output builder, finalizing and sorting it first on the
+// first call. Returns drained=true once the aggregator is exhausted and
+// closed (currentAgg is nil at that point) — the caller should then check
+// pending/commitBuilder's result rather than loop here, keeping this method
+// symmetric with the main pull loop's one-unit-of-work-per-call discipline.
+func (bt *BatchTimeBucket) drainMapModeAggregator(ctx context.Context) (drained bool, err error) {
+ if !bt.mapDrainReady {
+ if finalizeErr := bt.currentAgg.Finalize(ctx); finalizeErr != nil {
+ return false, finalizeErr
+ }
+ bt.currentAgg.SortInsertionByBucket()
+ bt.mapDrainReady = true
+ }
+ nb, nextErr := bt.currentAgg.NextBatch(ctx)
+ if nextErr != nil {
+ return false, nextErr
+ }
+ if nb == nil {
+ closeErr := bt.currentAgg.Close()
+ bt.currentAgg = nil
+ return true, closeErr
+ }
+ bt.appendOutput(nb)
+ return false, nil
+}
+
+// Close is idempotent; it closes any still-open bucket aggregator (the
+// error path — the scan never reached EOF), releases the pending queue, and
+// closes upstream.
+func (bt *BatchTimeBucket) Close() error {
+ if bt.closed {
+ return nil
+ }
+ bt.closed = true
+ var firstErr error
+ if bt.currentAgg != nil {
+ firstErr = bt.currentAgg.Close()
+ bt.currentAgg = nil
+ }
+ bt.pending = nil
+ bt.outputBuilder = nil
+ if bt.upstream != nil {
+ if closeErr := bt.upstream.Close(); closeErr != nil && firstErr == nil {
+ firstErr = closeErr
+ }
+ }
+ return firstErr
+}
+
+// consumeBatch floors b's timestamp column to bucket starts, then either
+// feeds every active row into the single persistent aggregator (map mode)
+// or splits the batch into contiguous same-bucket runs and flushes the open
+// bucket whenever the run's bucket advances past it (streaming mode). A row
+// whose bucket regresses in streaming mode is a monotonicity violation and
+// fails loudly rather than reopening an already-flushed bucket.
+func (bt *BatchTimeBucket) consumeBatch(ctx context.Context, b *vectorized.RecordBatch) error {
+ active := activeIndices(b)
+ if len(active) == 0 {
+ return nil
+ }
+ floorTimestampColumn(b, bt.timestampIdx, bt.widthNanos)
+ tsCol, ok := b.Columns[bt.timestampIdx].(*vectorized.TypedColumn[int64])
+ if !ok {
+ return fmt.Errorf("vectorized.measure: BatchTimeBucket: column %d is not a RoleTimestamp int64 column", bt.timestampIdx)
+ }
+ tsData := tsCol.Data()
+
+ if !bt.streaming {
+ if !bt.hasBucket {
+ bt.hasBucket = true
+ if openErr := bt.openBucket(ctx); openErr != nil {
+ return openErr
+ }
+ }
+ return bt.feedRun(ctx, b, active)
+ }
+
+ runStart := 0
+ for i, rowIdx := range active {
+ bucket := tsData[rowIdx]
+ switch {
+ case !bt.hasBucket:
+ bt.hasBucket = true
+ bt.currentBucket = bucket
+ if openErr := bt.openBucket(ctx); openErr != nil {
+ return openErr
+ }
+ runStart = i
+ case bucket == bt.currentBucket:
+ // Same run; keep accumulating.
+ case bucket > bt.currentBucket:
+ if feedErr := bt.feedRun(ctx, b, active[runStart:i]); feedErr != nil {
+ return feedErr
+ }
+ if flushErr := bt.flushCurrentBucket(ctx); flushErr != nil {
+ return flushErr
+ }
+ bt.currentBucket = bucket
+ if openErr := bt.openBucket(ctx); openErr != nil {
+ return openErr
+ }
+ runStart = i
+ default:
+ return fmt.Errorf("vectorized.measure: BatchTimeBucket: input is not time-ascending — row bucket %d precedes the current bucket %d", bucket, bt.currentBucket)
+ }
+ }
+ return bt.feedRun(ctx, b, active[runStart:])
+}
+
+// openBucket constructs and initializes a fresh BatchAggregation to become
+// the currently open bucket's aggregator.
+func (bt *BatchTimeBucket) openBucket(ctx context.Context) error {
+ agg := NewBatchAggregation(bt.inputSchema, bt.aggKeyIndices, bt.aggs, bt.mode, min(bt.batchSize, bucketAggPoolCapacity), bt.tracker, bt.entrySize)
+ if initErr := agg.Init(ctx); initErr != nil {
+ return initErr
+ }
+ bt.currentAgg = agg
+ return nil
+}
+
+// feedRun consumes the rows named by rows (a Selection into b's columns,
+// pre-floored) through the currently open bucket's aggregator.
+func (bt *BatchTimeBucket) feedRun(ctx context.Context, b *vectorized.RecordBatch, rows []uint16) error {
+ if len(rows) == 0 {
+ return nil
+ }
+ view := &vectorized.RecordBatch{Schema: b.Schema, Columns: b.Columns, Selection: rows, Len: b.Len}
+ return bt.currentAgg.Consume(ctx, view)
+}
+
+// flushCurrentBucket finalizes and drains a closed streaming-mode bucket's
+// aggregator, compacting its output into the shared builder (see
+// appendOutput), then closes it. A no-op when no bucket is open.
+//
+// Streaming-only: a streaming instance's aggregator holds exactly one
+// bucket's groups by construction, so draining it in one shot here is
+// bounded the same way the group map itself is. Map mode's terminal
+// aggregator can span the whole scan and is drained incrementally instead
+// by drainMapModeAggregator, which flushCurrentBucket is never called for.
+//
+// A closed bucket's own BatchAggregation.NextBatch pages come from a
+// pool sized for a full batchSize batch regardless of how many groups the
+// bucket actually held — a bucket with one group would otherwise queue one
+// full-capacity, mostly-empty RecordBatch per bucket. When a single
+// upstream batch spans many small buckets (e.g. one sample per bucket),
+// that is O(buckets) such allocations queued before NextBatch ever returns
+// one to the caller. appendOutput compacts them into shared
+// batchSize-capacity batches instead, so queued output is proportional to
+// row count, not bucket count.
+func (bt *BatchTimeBucket) flushCurrentBucket(ctx context.Context) error {
+ if bt.currentAgg == nil {
+ return nil
+ }
+ if finalizeErr := bt.currentAgg.Finalize(ctx); finalizeErr != nil {
+ return finalizeErr
+ }
+ for {
+ nb, nextErr := bt.currentAgg.NextBatch(ctx)
+ if nextErr != nil {
+ return nextErr
+ }
+ if nb == nil {
+ break
+ }
+ bt.appendOutput(nb)
+ }
+ closeErr := bt.currentAgg.Close()
+ bt.currentAgg = nil
+ return closeErr
+}
+
+// appendOutput compacts nb's rows into the shared output builder, queuing
+// the builder to pending once it reaches batchSize. nb already at
+// batchSize capacity (the common case for a bucket with many groups, or
+// for map mode's single final drain) is queued directly with no copy —
+// compaction only matters for the small/partial pages a sparse bucket
+// produces.
+func (bt *BatchTimeBucket) appendOutput(nb *vectorized.RecordBatch) {
+ if bt.outputBuilder == nil && nb.Len >= bt.batchSize {
+ bt.pending = append(bt.pending, nb)
+ return
+ }
+ for i := 0; i < nb.Len; i++ {
+ if bt.outputBuilder == nil {
+ bt.outputBuilder = bt.outputPool.Get()
+ }
+ for c, col := range bt.outputBuilder.Columns {
+ copyOneValue(col, nb.Columns[c], i)
+ }
+ bt.outputBuilder.Len++
+ if bt.outputBuilder.Len >= bt.batchSize {
+ bt.pending = append(bt.pending, bt.outputBuilder)
+ bt.outputBuilder = nil
+ }
+ }
+}
+
+// commitBuilder queues whatever the builder has accumulated so far, even if
+// under batchSize. Called at each NextBatch pull/consume boundary — the
+// same point at which the pre-compaction code already made a bucket's
+// flushed output visible — so compacting several small buckets into one
+// batch does not add pull-ahead latency beyond what NextBatch already did.
+func (bt *BatchTimeBucket) commitBuilder() {
+ if bt.outputBuilder == nil || bt.outputBuilder.Len == 0 {
+ return
+ }
+ bt.pending = append(bt.pending, bt.outputBuilder)
+ bt.outputBuilder = nil
+}
diff --git a/pkg/query/vectorized/measure/timebucket_test.go b/pkg/query/vectorized/measure/timebucket_test.go
new file mode 100644
index 000000000..5fbab10c3
--- /dev/null
+++ b/pkg/query/vectorized/measure/timebucket_test.go
@@ -0,0 +1,542 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "context"
+ "fmt"
+ "strconv"
+ "testing"
+
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized"
+)
+
+// TestBucketStart_BoundaryTimestamp_StaysPut pins design §11: a timestamp
+// exactly on a bucket boundary (ts % width == 0) floors to itself.
+func TestBucketStart_BoundaryTimestamp_StaysPut(t *testing.T) {
+ if got := bucketStart(5000, 1000); got != 5000 {
+ t.Fatalf("bucketStart(5000, 1000) = %d, want 5000", got)
+ }
+}
+
+// TestBucketStart_AgreesWithGetWindowStart_ForNonNegativeTimestamps pins
+// agreement with pkg/flow/streaming/sliding_window.go:302 getWindowStart
+// ("ts - ts%width") across the entire reachable domain (timestamps are Unix
+// milliseconds; the write path rejects non-positive ones) — keeping the
+// query path and the TopN pre-aggregation path from drifting into two
+// definitions of "the 5-minute bucket".
+func TestBucketStart_AgreesWithGetWindowStart_ForNonNegativeTimestamps(t *testing.T) {
+ cases := []struct{ ts, width int64 }{
+ {0, 1000},
+ {1, 1000},
+ {999, 1000},
+ {1000, 1000},
+ {1001, 1000},
+ {5999, 1000},
+ {6000, 1000},
+ {123456789, 60000},
+ }
+ for _, c := range cases {
+ got := bucketStart(c.ts, c.width)
+ want := c.ts - c.ts%c.width // getWindowStart's own formula
+ if got != want {
+ t.Errorf("bucketStart(%d, %d) = %d, want %d (getWindowStart)", c.ts, c.width, got, want)
+ }
+ }
+}
+
+// TestBucketStart_NegativeTimestamp_FloorsRatherThanTruncates pins design
+// §7.2's explicit correction: Go's % truncates toward zero, which would
+// skew getWindowStart's plain formula by one bucket for ts < 0 — negative
+// timestamps are unreachable in practice, but bucketStart floors correctly
+// regardless of a caller's invariant. -1 belongs to the bucket [-1000, 0),
+// so its floor is -1000, not 0 (what a naive "ts - ts%width" port yields).
+func TestBucketStart_NegativeTimestamp_FloorsRatherThanTruncates(t *testing.T) {
+ if got := bucketStart(-1, 1000); got != -1000 {
+ t.Fatalf("bucketStart(-1, 1000) = %d, want -1000 (floor, not truncate-toward-zero)", got)
+ }
+ if got := bucketStart(-1000, 1000); got != -1000 {
+ t.Fatalf("bucketStart(-1000, 1000) = %d, want -1000 (boundary stays put)", got)
+ }
+}
+
+// bucketTestRow is one input row for bucketTestSchema / feedBucketBatch.
+type bucketTestRow struct {
+ g string
+ ts int64
+ v int64
+}
+
+// bucketTestSchema is "timestamp (int64), tag.default.g (string), field v
+// (int64)" — a minimal scan-shaped schema for BatchTimeBucket tests.
+func bucketTestSchema() *vectorized.BatchSchema {
+ return vectorized.NewBatchSchema([]vectorized.ColumnDef{
+ {Role: vectorized.RoleTimestamp, Type: vectorized.ColumnTypeInt64},
+ {Role: vectorized.RoleTag, TagFamily: "default", Name: "g", Type: vectorized.ColumnTypeString},
+ {Role: vectorized.RoleField, Name: "v", Type: vectorized.ColumnTypeInt64},
+ })
+}
+
+func buildBucketBatch(schema *vectorized.BatchSchema, rows ...bucketTestRow) *vectorized.RecordBatch {
+ b := vectorized.NewRecordBatch(schema, len(rows))
+ tsCol := b.Columns[0].(*vectorized.TypedColumn[int64])
+ gCol := b.Columns[1].(*vectorized.TypedColumn[string])
+ vCol := b.Columns[2].(*vectorized.TypedColumn[int64])
+ for _, r := range rows {
+ tsCol.Append(r.ts)
+ gCol.Append(r.g)
+ vCol.Append(r.v)
+ }
+ b.Len = len(rows)
+ return b
+}
+
+// fakeBucketUpstream is a minimal vectorized.PullOperator serving a fixed
+// batch sequence, standing in for the Scan chain BatchTimeBucket wraps in
+// production.
+type fakeBucketUpstream struct {
+ schema *vectorized.BatchSchema
+ batches []*vectorized.RecordBatch
+ idx int
+}
+
+func (f *fakeBucketUpstream) Init(_ context.Context) error { return nil }
+func (f *fakeBucketUpstream) OutputSchema() *vectorized.BatchSchema { return f.schema }
+func (f *fakeBucketUpstream) Close() error { return nil }
+func (f *fakeBucketUpstream) NextBatch(_ context.Context) (*vectorized.RecordBatch, error) {
+ if f.idx >= len(f.batches) {
+ return nil, nil
+ }
+ b := f.batches[f.idx]
+ f.idx++
+ return b, nil
+}
+
+// sumByBucketAndTag reads a drained batch shaped [timestamp, g, v(sum)] into
+// a map keyed "bucket|tag" for order-independent assertions.
+func sumByBucketAndTag(t *testing.T, batches []*vectorized.RecordBatch) map[string]int64 {
+ t.Helper()
+ out := make(map[string]int64)
+ for _, b := range batches {
+ tsCol := b.Columns[0].(*vectorized.TypedColumn[int64])
+ gCol := b.Columns[1].(*vectorized.TypedColumn[string])
+ vCol := b.Columns[2].(*vectorized.TypedColumn[int64])
+ for i := 0; i < b.Len; i++ {
+ key := formatBucketKey(tsCol.Data()[i], gCol.Data()[i])
+ out[key] = vCol.Data()[i]
+ }
+ }
+ return out
+}
+
+func formatBucketKey(ts int64, g string) string {
+ return g + "|" + strconv.FormatInt(ts, 10)
+}
+
+func drainBucket(t *testing.T, bt *BatchTimeBucket) []*vectorized.RecordBatch {
+ t.Helper()
+ var out []*vectorized.RecordBatch
+ for {
+ nb, err := bt.NextBatch(context.Background())
+ if err != nil {
+ t.Fatalf("NextBatch: %v", err)
+ }
+ if nb == nil {
+ break
+ }
+ out = append(out, nb)
+ }
+ return out
+}
+
+// TestBatchTimeBucket_Streaming_GroupsPerBucketPerTag pins the core
+// streaming shape end to end, through the public PullOperator interface:
+// rows across two buckets and two tag groups are summed per (bucket, tag),
+// and the output timestamp column carries the bucket start (design §7.2's
+// conditional D2 reversal).
+func TestBatchTimeBucket_Streaming_GroupsPerBucketPerTag(t *testing.T) {
+ s := bucketTestSchema()
+ keyIndices := []int{1} // tag "g"
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ batch := buildBucketBatch(s,
+ bucketTestRow{"a", 100, 1}, // bucket 0
+ bucketTestRow{"b", 200, 4}, // bucket 0
+ bucketTestRow{"a", 300, 2}, // bucket 0
+ bucketTestRow{"a", 1100, 3}, // bucket 1000
+ bucketTestRow{"b", 1200, 5}, // bucket 1000
+ )
+ upstream := &fakeBucketUpstream{schema: s, batches: []*vectorized.RecordBatch{batch}}
+ bt := NewBatchTimeBucket(upstream, s, keyIndices, 0, 1000, spec, AggModeAll, 8, vectorized.NewMemoryTracker(1<<30), 0, true)
+ if initErr := bt.Init(context.Background()); initErr != nil {
+ t.Fatal(initErr)
+ }
+ defer bt.Close()
+
+ got := sumByBucketAndTag(t, drainBucket(t, bt))
+ want := map[string]int64{"a|0": 3, "b|0": 4, "a|1000": 3, "b|1000": 5}
+ for k, v := range want {
+ if got[k] != v {
+ t.Errorf("sum[%s] = %d, want %d (full result: %v)", k, got[k], v, got)
+ }
+ }
+ if len(got) != len(want) {
+ t.Fatalf("result set size = %d, want %d: %v", len(got), len(want), got)
+ }
+}
+
+// TestBatchTimeBucket_Streaming_NextBatch_ReturnsClosedBucketBeforeUpstreamEOF
+// pins the pipelining property (design §7.2): the first NextBatch call
+// returns bucket 0's output as soon as a later row in the SAME upstream
+// batch advances past it — without ever asking upstream for more input, let
+// alone reaching its EOF. Only the second NextBatch call, which does drive
+// upstream to EOF, flushes the still-open final bucket. The operator never
+// buffers more than what closed while processing one upstream batch.
+func TestBatchTimeBucket_Streaming_NextBatch_ReturnsClosedBucketBeforeUpstreamEOF(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ batch := buildBucketBatch(s,
+ bucketTestRow{"a", 100, 1},
+ bucketTestRow{"a", 1100, 2}, // advances past bucket 0
+ )
+ upstream := &fakeBucketUpstream{schema: s, batches: []*vectorized.RecordBatch{batch}}
+ bt := NewBatchTimeBucket(upstream, s, nil, 0, 1000, spec, AggModeAll, 8, vectorized.NewMemoryTracker(1<<30), 0, true)
+ _ = bt.Init(context.Background())
+ defer bt.Close()
+
+ first, err := bt.NextBatch(context.Background())
+ if err != nil {
+ t.Fatalf("NextBatch: %v", err)
+ }
+ if first == nil {
+ t.Fatal("bucket 0 must flush and be returned by the first NextBatch call")
+ }
+ if upstream.idx != 1 {
+ t.Fatalf("the first NextBatch call must pull upstream exactly once, got %d pulls", upstream.idx)
+ }
+ tsCol := first.Columns[0].(*vectorized.TypedColumn[int64])
+ if tsCol.Data()[0] != 0 {
+ t.Fatalf("first flushed bucket timestamp = %d, want 0", tsCol.Data()[0])
+ }
+
+ second, err := bt.NextBatch(context.Background())
+ if err != nil {
+ t.Fatalf("NextBatch: %v", err)
+ }
+ if second == nil {
+ t.Fatal("bucket 1000 must flush on the second NextBatch call (upstream EOF)")
+ }
+ if second.Columns[0].(*vectorized.TypedColumn[int64]).Data()[0] != 1000 {
+ t.Fatalf("second flushed bucket timestamp = %d, want 1000", second.Columns[0].(*vectorized.TypedColumn[int64]).Data()[0])
+ }
+
+ third, err := bt.NextBatch(context.Background())
+ if err != nil || third != nil {
+ t.Fatalf("want EOF (nil, nil) after both buckets drained, got (%v, %v)", third, err)
+ }
+}
+
+// TestBatchTimeBucket_Streaming_MonotonicityViolation_Errors pins the
+// operator-side guard (design §7.2): a row whose bucket regresses is an
+// invariant violation and must fail loudly, not silently reopen a
+// already-flushed bucket. Exercises consumeBatch directly (package-internal
+// access) since the violation is a property of one Consume-shaped call, not
+// of NextBatch's surrounding pull loop.
+func TestBatchTimeBucket_Streaming_MonotonicityViolation_Errors(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ bt := NewBatchTimeBucket(nil, s, nil, 0, 1000, spec, AggModeAll, 8, vectorized.NewMemoryTracker(1<<30), 0, true)
+ defer bt.Close()
+
+ batch := buildBucketBatch(s,
+ bucketTestRow{"a", 1100, 1}, // bucket 1000
+ bucketTestRow{"a", 100, 2}, // bucket 0 — regresses
+ )
+ if consumeErr := bt.consumeBatch(context.Background(), batch); consumeErr == nil {
+ t.Fatal("a regressing bucket must error, not silently reopen bucket 0")
+ }
+}
+
+// TestBatchTimeBucket_MapMode_HandlesOutOfOrderInput pins design §7.2's
+// index-mode fallback: input arriving as buckets B, A, B (which would
+// violate streaming's monotonicity guard) must still produce one correct
+// result per bucket when streaming=false — proving it took the map path.
+func TestBatchTimeBucket_MapMode_HandlesOutOfOrderInput(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ batch := buildBucketBatch(s,
+ bucketTestRow{"a", 1100, 10}, // bucket 1000 (B)
+ bucketTestRow{"a", 100, 1}, // bucket 0 (A) — out of order, must not error
+ bucketTestRow{"a", 1200, 20}, // bucket 1000 (B) again
+ )
+ upstream := &fakeBucketUpstream{schema: s, batches: []*vectorized.RecordBatch{batch}}
+ bt := NewBatchTimeBucket(upstream, s, nil, 0, 1000, spec, AggModeAll, 8, vectorized.NewMemoryTracker(1<<30), 0, false)
+ if initErr := bt.Init(context.Background()); initErr != nil {
+ t.Fatal(initErr)
+ }
+ defer bt.Close()
+
+ got := sumByBucketAndTag(t, drainBucket(t, bt))
+ want := map[string]int64{"a|0": 1, "a|1000": 30}
+ for k, v := range want {
+ if got[k] != v {
+ t.Errorf("sum[%s] = %d, want %d (full result: %v)", k, got[k], v, got)
+ }
+ }
+}
+
+// TestBatchTimeBucket_MapMode_EmitsBucketsInAscendingOrder pins the review
+// finding that motivated BatchAggregation.SortInsertionByBucket: map mode's
+// single persistent aggregator accumulates groups in first-seen order over
+// input that may not be time-ordered (the whole reason map mode exists),
+// so without an explicit sort, out-of-order input like [2100, 1100, 2200]
+// would emit buckets [2000, 1000] instead of [1000, 2000] — silently
+// violating the bucket-ascending output contract (design §7.5) that a
+// downstream Limit relies on.
+func TestBatchTimeBucket_MapMode_EmitsBucketsInAscendingOrder(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ batch := buildBucketBatch(s,
+ bucketTestRow{"a", 2100, 2}, // bucket 2000
+ bucketTestRow{"a", 1100, 1}, // bucket 1000 — out of order
+ bucketTestRow{"a", 2200, 20},
+ )
+ upstream := &fakeBucketUpstream{schema: s, batches: []*vectorized.RecordBatch{batch}}
+ bt := NewBatchTimeBucket(upstream, s, nil, 0, 1000, spec, AggModeAll, 8, vectorized.NewMemoryTracker(1<<30), 0, false)
+ if initErr := bt.Init(context.Background()); initErr != nil {
+ t.Fatal(initErr)
+ }
+ defer bt.Close()
+
+ drained := drainBucket(t, bt)
+ var gotTS []int64
+ for _, b := range drained {
+ tsCol := b.Columns[0].(*vectorized.TypedColumn[int64])
+ for i := 0; i < b.Len; i++ {
+ gotTS = append(gotTS, tsCol.Data()[i])
+ }
+ }
+ wantTS := []int64{1000, 2000}
+ if len(gotTS) != len(wantTS) {
+ t.Fatalf("bucket timestamps = %v, want %v", gotTS, wantTS)
+ }
+ for i, want := range wantTS {
+ if gotTS[i] != want {
+ t.Fatalf("bucket timestamps = %v, want ascending %v", gotTS, wantTS)
+ }
+ }
+}
+
+// TestBatchTimeBucket_MapMode_DrainsTerminalAggregatorIncrementally pins the
+// PR review finding that motivated drainMapModeAggregator: map mode's
+// terminal aggregator represents the whole scan (unlike a streaming
+// instance, which only ever holds one bucket), so materializing every one
+// of its pages into pending before the first NextBatch call returns would
+// re-introduce the same unbounded-queue shape the PullOperator rewrite
+// exists to avoid. With one group per bucket, batchSize=8, and the
+// per-bucket aggregator pool capped at bucketAggPoolCapacity=64 internally,
+// draining must proceed in small increments — not all 100 groups at once —
+// so no single NextBatch call should return a batch of more than batchSize
+// rows, and pending must never hold more than a couple of batches' worth at
+// a time while the drain is in progress.
+func TestBatchTimeBucket_MapMode_DrainsTerminalAggregatorIncrementally(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ const numGroups = 100
+ const batchSize = 8
+ rows := make([]bucketTestRow, 0, numGroups)
+ for i := 0; i < numGroups; i++ {
+ // Distinct tag per row forces one group per row; map mode never
+ // flushes mid-scan regardless of bucket, so all numGroups groups
+ // stay live in the single terminal aggregator until upstream EOF.
+ rows = append(rows, bucketTestRow{fmt.Sprintf("tag-%d", i), 0, int64(i)})
+ }
+ batch := buildBucketBatch(s, rows...)
+ upstream := &fakeBucketUpstream{schema: s, batches: []*vectorized.RecordBatch{batch}}
+ bt := NewBatchTimeBucket(upstream, s, []int{1}, 0, 1000, spec, AggModeAll, batchSize, vectorized.NewMemoryTracker(1<<30), 0, false)
+ if initErr := bt.Init(context.Background()); initErr != nil {
+ t.Fatal(initErr)
+ }
+ defer bt.Close()
+
+ total := 0
+ batches := 0
+ for {
+ if len(bt.pending) > 2 {
+ t.Fatalf("pending holds %d batches mid-drain, want the terminal aggregator drained incrementally, not materialized all at once", len(bt.pending))
+ }
+ nb, err := bt.NextBatch(context.Background())
+ if err != nil {
+ t.Fatalf("NextBatch: %v", err)
+ }
+ if nb == nil {
+ break
+ }
+ batches++
+ if nb.Len > batchSize {
+ t.Fatalf("output batch Len = %d, exceeds batchSize %d", nb.Len, batchSize)
+ }
+ total += nb.Len
+ }
+ if total != numGroups {
+ t.Fatalf("total rows drained = %d, want %d", total, numGroups)
+ }
+ if wantBatches := (numGroups + batchSize - 1) / batchSize; batches != wantBatches {
+ t.Fatalf("got %d output batches, want %d", batches, wantBatches)
+ }
+}
+
+// TestBatchTimeBucket_Close_ReleasesLiveState pins that closing the
+// operator mid-stream (error path, or caller giving up early) releases the
+// open bucket's memory reservation rather than leaking it. Exercises
+// consumeBatch directly so the reservation can be inspected before any
+// flush would otherwise release it.
+func TestBatchTimeBucket_Close_ReleasesLiveState(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ tracker := vectorized.NewMemoryTracker(1 << 30)
+ bt := NewBatchTimeBucket(nil, s, []int{1}, 0, 1000, spec, AggModeAll, 8, tracker, 512, true)
+
+ batch := buildBucketBatch(s, bucketTestRow{"a", 100, 1}, bucketTestRow{"b", 200, 2})
+ if consumeErr := bt.consumeBatch(context.Background(), batch); consumeErr != nil {
+ t.Fatalf("consumeBatch: %v", consumeErr)
+ }
+ if tracker.Used() == 0 {
+ t.Fatal("open bucket must have reserved memory for its live groups")
+ }
+ if closeErr := bt.Close(); closeErr != nil {
+ t.Fatalf("Close: %v", closeErr)
+ }
+ if tracker.Used() != 0 {
+ t.Fatalf("Close must release the open bucket's reservation, tracker.Used() = %d", tracker.Used())
+ }
+}
+
+// TestBatchTimeBucket_NextBatch_PullsUpstreamLazily is the regression pin
+// for the P1 review finding this design was rewritten to fix: a
+// BreakerOperator-shaped implementation is driven by breakerStage, which
+// pulls the ENTIRE upstream — closing every bucket the scan will ever
+// produce — before the breaker's own NextBatch is ever called, so the
+// flushed-but-unemitted queue would hold every bucket, not just the one
+// that just closed. As a genuine PullOperator that owns upstream directly,
+// BatchTimeBucket must only pull as much upstream as it needs to produce
+// ONE output batch, leaving later buckets' upstream batches unpulled.
+func TestBatchTimeBucket_NextBatch_PullsUpstreamLazily(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ // Three widely-spaced buckets, one upstream batch each — a scan that,
+ // under the old Breaker shape, would be fully drained (all 3 pulls) by
+ // the very first NextBatch call.
+ batches := []*vectorized.RecordBatch{
+ buildBucketBatch(s, bucketTestRow{"a", 0, 1}),
+ buildBucketBatch(s, bucketTestRow{"a", 1000, 2}),
+ buildBucketBatch(s, bucketTestRow{"a", 2000, 3}),
+ }
+ upstream := &fakeBucketUpstream{schema: s, batches: batches}
+ bt := NewBatchTimeBucket(upstream, s, nil, 0, 1000, spec, AggModeAll, 8, vectorized.NewMemoryTracker(1<<30), 0, true)
+ _ = bt.Init(context.Background())
+ defer bt.Close()
+
+ first, err := bt.NextBatch(context.Background())
+ if err != nil {
+ t.Fatalf("NextBatch: %v", err)
+ }
+ if first == nil {
+ t.Fatal("want the first flushed bucket back immediately")
+ }
+ // Only enough upstream batches to detect ONE bucket transition (bucket
+ // 0's opening row, then bucket 1000's opening row triggers the flush) —
+ // not all three, which the old Breaker shape would have pulled.
+ if upstream.idx != 2 {
+ t.Fatalf("first NextBatch call pulled upstream %d times, want 2 (must not race ahead into unopened buckets)", upstream.idx)
+ }
+ if len(bt.pending) != 0 {
+ t.Fatalf("pending queue must be drained by the return, not accumulate: len=%d", len(bt.pending))
+ }
+}
+
+// TestBatchTimeBucket_Streaming_SparseBuckets_CompactsOutputBatches pins the
+// review finding that motivated appendOutput/commitBuilder: one upstream
+// batch spanning many distinct, single-row buckets must not queue one
+// full-batchSize-capacity RecordBatch per bucket (O(buckets) allocations for
+// O(buckets) rows). With batchSize=8 and 20 buckets closed while processing
+// a single upstream batch, output must be packed into ceil(20/8)=3 batches,
+// not 20 — and no output batch may exceed batchSize rows.
+func TestBatchTimeBucket_Streaming_SparseBuckets_CompactsOutputBatches(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ const numBuckets = 20
+ const batchSize = 8
+ rows := make([]bucketTestRow, 0, numBuckets)
+ for i := 0; i < numBuckets; i++ {
+ rows = append(rows, bucketTestRow{"a", int64(i) * 1000, int64(i)})
+ }
+ batch := buildBucketBatch(s, rows...)
+ upstream := &fakeBucketUpstream{schema: s, batches: []*vectorized.RecordBatch{batch}}
+ bt := NewBatchTimeBucket(upstream, s, nil, 0, 1000, spec, AggModeAll, batchSize, vectorized.NewMemoryTracker(1<<30), 0, true)
+ if initErr := bt.Init(context.Background()); initErr != nil {
+ t.Fatal(initErr)
+ }
+ defer bt.Close()
+
+ drained := drainBucket(t, bt)
+ // The last bucket (opened by the batch's final row) never sees an
+ // advancing row within this batch, so only numBuckets-1 buckets flush
+ // here; the final one flushes at upstream EOF as its own commit.
+ wantBatches := (numBuckets-1+batchSize-1)/batchSize + 1
+ if len(drained) != wantBatches {
+ t.Fatalf("got %d output batches, want %d (compaction must pack multiple closed buckets per batch)", len(drained), wantBatches)
+ }
+ total := 0
+ for _, b := range drained {
+ if b.Len > batchSize {
+ t.Errorf("output batch Len = %d, exceeds batchSize %d", b.Len, batchSize)
+ }
+ total += b.Len
+ }
+ if total != numBuckets {
+ t.Fatalf("total rows across output batches = %d, want %d", total, numBuckets)
+ }
+}
+
+// TestBatchTimeBucket_Streaming_BucketAdvanceReleasesPriorBucketMemory pins
+// the memory-bound property this whole design exists for (§7.2): flushing a
+// closed bucket must release ITS reservation immediately — not just at
+// final Close — so live memory never grows past one bucket's groups
+// regardless of how many buckets the scan crosses. Exercises consumeBatch
+// directly to inspect the tracker right after the internal flush, before
+// NextBatch's own draining would otherwise complicate the picture.
+func TestBatchTimeBucket_Streaming_BucketAdvanceReleasesPriorBucketMemory(t *testing.T) {
+ s := bucketTestSchema()
+ spec := []AggSpec{{Func: AggSum, InputCol: 2, Output: "v"}}
+ tracker := vectorized.NewMemoryTracker(1 << 30)
+ bt := NewBatchTimeBucket(nil, s, []int{1}, 0, 1000, spec, AggModeAll, 8, tracker, 512, true)
+ defer bt.Close()
+
+ batch := buildBucketBatch(s,
+ bucketTestRow{"a", 100, 1}, // opens bucket 0, reserves memory
+ bucketTestRow{"a", 1100, 2}, // advances past bucket 0: must flush + release it
+ )
+ if consumeErr := bt.consumeBatch(context.Background(), batch); consumeErr != nil {
+ t.Fatalf("consumeBatch: %v", consumeErr)
+ }
+ // Only bucket 1000 should still be live; bucket 0's reservation must
+ // already be released, not deferred until upstream EOF or Close.
+ if got := tracker.Used(); got != 512 {
+ t.Fatalf("tracker.Used() = %d, want 512 (only the current bucket's one group live)", got)
+ }
+}
diff --git a/pkg/query/vectorized/measure/topology_matrix_test.go b/pkg/query/vectorized/measure/topology_matrix_test.go
index 641e4fac3..734bdc547 100644
--- a/pkg/query/vectorized/measure/topology_matrix_test.go
+++ b/pkg/query/vectorized/measure/topology_matrix_test.go
@@ -84,7 +84,7 @@ func TestTopologyMatrix(t *testing.T) {
t.Run(name, func(t *testing.T) {
frames := buildPartialFrames(t, rows, shards, replicas, tc.fn)
got, _, reduceErr := ReduceRawFrames(frames,
- []string{"g"},
+ "default", []string{"g"}, false,
[]AggReduceSpec{{OutputName: "out", Func: tc.fn}},
1024, vectorized.NewMemoryTracker(1<<30))
if reduceErr != nil {
@@ -343,7 +343,7 @@ func TestTopologyMatrix_WithTop(t *testing.T) {
t.Run(name, func(t *testing.T) {
frames := buildPartialFrames(t, rows, shards, replicas, AggSum)
reduced, _, reduceErr := ReduceRawFrames(frames,
- []string{"g"},
+ "default", []string{"g"}, false,
[]AggReduceSpec{{OutputName: "out", Func: AggSum}},
1024, vectorized.NewMemoryTracker(1<<30))
if reduceErr != nil {
diff --git a/pkg/query/vectorized/pipeline.go b/pkg/query/vectorized/pipeline.go
index 8394e9a29..ab7d87d1a 100644
--- a/pkg/query/vectorized/pipeline.go
+++ b/pkg/query/vectorized/pipeline.go
@@ -114,6 +114,35 @@ func (b *PipelineBuilder) Break(br BreakerOperator) *PipelineBuilder {
return b
}
+// Transform closes the current segment — any queued Applies, plus any
+// completed Break steps — into a single concrete PullOperator, passes it to
+// fn, and makes fn's result the new base for whatever Apply/Break/Transform
+// calls follow (and for the eventual Build).
+//
+// Unlike Break, the operator fn returns is a plain PullOperator, not a
+// BreakerOperator driven by breakerStage's "drain the entire upstream via
+// Consume, then serve via NextBatch" contract. Use Transform for an
+// operator that needs to pull from upstream lazily, batch by batch,
+// interleaving its own flushed output with further upstream consumption —
+// e.g. one whose memory bound depends on never holding more than what a
+// single upstream batch can produce. Wrapping such an operator as a
+// BreakerOperator instead would force the entire upstream to be buffered
+// before anything downstream is emitted, defeating the point.
+func (b *PipelineBuilder) Transform(fn func(upstream PullOperator) PullOperator) *PipelineBuilder {
+ head := b.source
+ for _, step := range b.steps {
+ head = newFusedStage(head, step.preFused)
+ head = newBreakerStage(head, step.breaker)
+ }
+ if len(b.pendingFused) > 0 {
+ head = newFusedStage(head, b.pendingFused)
+ }
+ b.source = fn(head)
+ b.steps = nil
+ b.pendingFused = nil
+ return b
+}
+
// Build validates and constructs the Pipeline. Each closed segment
// becomes fusedStage(prev, preFused) → breakerStage(_, breaker); any
// fusibles still queued after the last Break form a final fused stage
diff --git a/pkg/query/vectorized/pipeline_test.go b/pkg/query/vectorized/pipeline_test.go
index 2c80ba2b4..8776bacb7 100644
--- a/pkg/query/vectorized/pipeline_test.go
+++ b/pkg/query/vectorized/pipeline_test.go
@@ -273,3 +273,111 @@ func TestPipelineBuilder_ApplyAfterBreak_RunsAfterBreaker(t *testing.T) {
t.Fatalf("post-break fusible should see breaker output %v (not source rows), got %v", wantPost, postSeen)
}
}
+
+// doublingPull wraps an upstream PullOperator and doubles every int64 value
+// it sees — a minimal stand-in for an operator that genuinely transforms
+// its upstream via NextBatch, the shape Transform exists for.
+type doublingPull struct {
+ upstream PullOperator
+}
+
+func (d *doublingPull) Init(ctx context.Context) error { return d.upstream.Init(ctx) }
+func (d *doublingPull) OutputSchema() *BatchSchema { return d.upstream.OutputSchema() }
+func (d *doublingPull) Close() error { return d.upstream.Close() }
+func (d *doublingPull) NextBatch(ctx context.Context) (*RecordBatch, error) {
+ b, err := d.upstream.NextBatch(ctx)
+ if err != nil || b == nil {
+ return b, err
+ }
+ col := b.Columns[0].(*TypedColumn[int64])
+ data := col.Data()
+ for i, v := range data {
+ col.SetAt(i, v*2)
+ }
+ return b, nil
+}
+
+func TestPipelineBuilder_Transform_WrapsSourceDirectly(t *testing.T) {
+ s := NewBatchSchema([]ColumnDef{{Role: RoleTimestamp, Type: ColumnTypeInt64}})
+ src := &fakePull{schema: s, batches: []*RecordBatch{mkInt64Batch(s, 1, 2, 3)}}
+ p, err := NewPipelineBuilder().
+ From(src).
+ Transform(func(upstream PullOperator) PullOperator { return &doublingPull{upstream: upstream} }).
+ Build()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if initErr := p.head.Init(context.Background()); initErr != nil {
+ t.Fatal(initErr)
+ }
+ out, err := p.Next(context.Background())
+ if err != nil || out == nil {
+ t.Fatalf("Transform-wrapped pipeline should yield a batch: out=%v err=%v", out, err)
+ }
+ got := out.Columns[0].(*TypedColumn[int64]).Data()
+ want := []int64{2, 4, 6}
+ for i := range want {
+ if got[i] != want[i] {
+ t.Fatalf("Transform should apply doublingPull to the source's rows: got %v, want %v", got, want)
+ }
+ }
+}
+
+// TestPipelineBuilder_Transform_ClosesPendingSegmentFirst pins that
+// Transform sees the fully-fused upstream — any Applied fusibles queued
+// before it must already have run — not the raw source.
+func TestPipelineBuilder_Transform_ClosesPendingSegmentFirst(t *testing.T) {
+ s := NewBatchSchema([]ColumnDef{{Role: RoleTimestamp, Type: ColumnTypeInt64}})
+ src := &fakePull{schema: s, batches: []*RecordBatch{mkInt64Batch(s, 1, 2, 3)}}
+ incrementBeforeTransform := &fakeFusible{schema: s} // +1
+ p, err := NewPipelineBuilder().
+ From(src).
+ Apply(incrementBeforeTransform).
+ Transform(func(upstream PullOperator) PullOperator { return &doublingPull{upstream: upstream} }).
+ Build()
+ if err != nil {
+ t.Fatal(err)
+ }
+ _ = p.head.Init(context.Background())
+ out, err := p.Next(context.Background())
+ if err != nil || out == nil {
+ t.Fatalf("pipeline should yield a batch: out=%v err=%v", out, err)
+ }
+ got := out.Columns[0].(*TypedColumn[int64]).Data()
+ want := []int64{4, 6, 8} // (v+1)*2
+ for i := range want {
+ if got[i] != want[i] {
+ t.Fatalf("Transform's upstream must be the fully-fused chain (+1 then *2): got %v, want %v", got, want)
+ }
+ }
+}
+
+// TestPipelineBuilder_BreakAfterTransform_RunsOnTransformedOutput pins that
+// a Break following a Transform drains the Transform-wrapped operator, not
+// the original source — proving Transform correctly becomes the new base
+// for subsequent stages.
+func TestPipelineBuilder_BreakAfterTransform_RunsOnTransformedOutput(t *testing.T) {
+ s := NewBatchSchema([]ColumnDef{{Role: RoleTimestamp, Type: ColumnTypeInt64}})
+ src := &fakePull{schema: s, batches: []*RecordBatch{mkInt64Batch(s, 1, 2, 3)}}
+ br := &fakeBreaker{schema: s}
+ p, err := NewPipelineBuilder().
+ From(src).
+ Transform(func(upstream PullOperator) PullOperator { return &doublingPull{upstream: upstream} }).
+ Break(br).
+ Build()
+ if err != nil {
+ t.Fatal(err)
+ }
+ _ = p.head.Init(context.Background())
+ _, _ = p.Next(context.Background())
+ if len(br.consumed) != 1 {
+ t.Fatalf("breaker should have consumed exactly 1 batch, got %d", len(br.consumed))
+ }
+ got := br.consumed[0].Columns[0].(*TypedColumn[int64]).Data()
+ want := []int64{2, 4, 6}
+ for i := range want {
+ if got[i] != want[i] {
+ t.Fatalf("breaker after Transform should see doubled rows: got %v, want %v", got, want)
+ }
+ }
+}
diff --git a/pkg/test/helpers/constant.go b/pkg/test/helpers/constant.go
index 8706b1308..65a780000 100644
--- a/pkg/test/helpers/constant.go
+++ b/pkg/test/helpers/constant.go
@@ -80,6 +80,12 @@ type Args struct {
WantErr bool
DisOrder bool
IgnoreElementID bool
+ // SkipQL skips the BydbQL round-trip check (input/<Input>.ql) for this
+ // case. Temporary escape hatch for query shapes BydbQL cannot express
+ // yet — e.g. GroupBy.time_bucket, which has no TIME_BUCKET(...) grammar
+ // until apache/skywalking#14091 lands. Remove a case's SkipQL (and add
+ // its .ql fixture) once BydbQL supports the shape.
+ SkipQL bool
}
// UnmarshalYAML decodes YAML raw bytes to proto.Message.
diff --git a/pkg/test/measure/testdata/measures/time_bucket_metric.json b/pkg/test/measure/testdata/measures/time_bucket_metric.json
new file mode 100644
index 000000000..26e19eab8
--- /dev/null
+++ b/pkg/test/measure/testdata/measures/time_bucket_metric.json
@@ -0,0 +1,36 @@
+{
+ "metadata": {
+ "name": "time_bucket_metric",
+ "group": "sw_metric"
+ },
+ "tag_families": [
+ {
+ "name": "default",
+ "tags": [
+ {
+ "name": "id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "entity_id",
+ "type": "TAG_TYPE_STRING"
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "name": "value",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ }
+ ],
+ "entity": {
+ "tag_names": [
+ "entity_id"
+ ]
+ },
+ "interval": "20s",
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}
diff --git a/test/cases/init.go b/test/cases/init.go
index 8d33da6fe..f4569c2c8 100644
--- a/test/cases/init.go
+++ b/test/cases/init.go
@@ -71,6 +71,13 @@ func Initialize(addr string, now time.Time) {
casesmeasuredata.Write(conn, "endpoint_resp_time_minute", "sw_metric", "endpoint_resp_time_minute_data1.json", now.Add(10*time.Second), interval)
casesmeasuredata.Write(conn, "service_instance_metric_topn_test", "sw_metric", "service_instance_metric_topn_test_data.json", now, interval)
casesmeasuredata.Write(conn, "service_instance_float_metric", "sw_metric", "service_instance_float_metric_data.json", now, interval)
+ // time_bucket_metric is seeded at its own 20s interval (not the shared
+ // 1m `interval`) so its 6 points land 20s apart ending at `now`. `now`
+ // is guaranteed minute-aligned (see the suite's `now` construction), so
+ // a 1-minute time_bucket width deterministically splits them 2/3/1
+ // across three buckets regardless of the wall-clock minute `now` lands
+ // on — see test/cases/measure/data/input/group_time_bucket*.yaml.
+ casesmeasuredata.Write(conn, "time_bucket_metric", "sw_metric", "time_bucket_metric_data.json", now, 20*time.Second)
casesmeasuredata.WriteMixed(conn, now.Add(30*time.Minute), interval,
casesmeasuredata.WriteSpec{
Metadata: &commonv1.Metadata{Name: "service_cpm_minute", Group: "sw_spec"},
diff --git a/test/cases/measure/data/data.go b/test/cases/measure/data/data.go
index 594d2c322..f3fa73698 100644
--- a/test/cases/measure/data/data.go
+++ b/test/cases/measure/data/data.go
@@ -109,10 +109,18 @@ func verifyWithContext(ctx context.Context, innerGm gm.Gomega, sharedContext hel
return a.Timestamp.AsTime().Compare(b.Timestamp.AsTime())
})
}
- for i := range resp.DataPoints {
- if resp.DataPoints[i].Timestamp != nil {
- innerGm.Expect(resp.DataPoints[i].Version).Should(gm.BeNumerically(">", 0))
- innerGm.Expect(resp.DataPoints[i].Sid).Should(gm.BeNumerically(">", 0))
+ // Version/Sid describe a single stored row, so this only holds for a
+ // non-aggregated result (raw scan or raw GroupBy's first-seen-row
+ // passthrough). An aggregated result may still carry a non-nil
+ // Timestamp — a time-bucketed GroupBy re-emits the bucket start (design
+ // §7.2's D2 reversal) — but that bucket row can merge many underlying
+ // rows (and series), so no single Version/Sid describes it.
+ if query.GetAgg() == nil {
+ for i := range resp.DataPoints {
+ if resp.DataPoints[i].Timestamp != nil {
+ innerGm.Expect(resp.DataPoints[i].Version).Should(gm.BeNumerically(">", 0))
+ innerGm.Expect(resp.DataPoints[i].Sid).Should(gm.BeNumerically(">", 0))
+ }
}
}
success := innerGm.Expect(cmp.Equal(resp, want,
@@ -153,8 +161,10 @@ var VerifyFn = func(innerGm gm.Gomega, sharedContext helpers.SharedContext, args
// verifyQLWithRequest ensures the generated QL matches the YAML request specification.
func verifyQLWithRequest(ctx context.Context, innerGm gm.Gomega, args helpers.Args, yamlQuery *measurev1.QueryRequest, conn *grpclib.ClientConn) {
- // if the test case expects an error, skip the QL verification.
- if args.WantErr {
+ // if the test case expects an error, or explicitly opts out (a query
+ // shape BydbQL cannot express yet — see helpers.Args.SkipQL), skip the
+ // QL verification.
+ if args.WantErr || args.SkipQL {
return
}
qlContent, err := qlFS.ReadFile("input/" + args.Input + ".ql")
diff --git a/test/cases/measure/data/input/group_time_bucket.yaml b/test/cases/measure/data/input/group_time_bucket.yaml
new file mode 100644
index 000000000..7536ffc9c
--- /dev/null
+++ b/test/cases/measure/data/input/group_time_bucket.yaml
@@ -0,0 +1,27 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: "time_bucket_metric"
+groups: ["sw_metric"]
+fieldProjection:
+ names: ["value"]
+groupBy:
+ timeBucket:
+ width: "1m"
+agg:
+ function: "AGGREGATION_FUNCTION_SUM"
+ fieldName: "value"
diff --git a/test/cases/measure/data/input/group_time_bucket_with_tag.yaml b/test/cases/measure/data/input/group_time_bucket_with_tag.yaml
new file mode 100644
index 000000000..6586db221
--- /dev/null
+++ b/test/cases/measure/data/input/group_time_bucket_with_tag.yaml
@@ -0,0 +1,35 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: "time_bucket_metric"
+groups: ["sw_metric"]
+tagProjection:
+ tagFamilies:
+ - name: "default"
+ tags: ["id"]
+fieldProjection:
+ names: ["value"]
+groupBy:
+ tagProjection:
+ tagFamilies:
+ - name: "default"
+ tags: ["id"]
+ timeBucket:
+ width: "1m"
+agg:
+ function: "AGGREGATION_FUNCTION_SUM"
+ fieldName: "value"
diff --git a/test/cases/measure/data/testdata/time_bucket_metric_data.json b/test/cases/measure/data/testdata/time_bucket_metric_data.json
new file mode 100644
index 000000000..35d5a29eb
--- /dev/null
+++ b/test/cases/measure/data/testdata/time_bucket_metric_data.json
@@ -0,0 +1,152 @@
+[
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svcA"
+ }
+ },
+ {
+ "str": {
+ "value": "entity_1"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 1
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svcA"
+ }
+ },
+ {
+ "str": {
+ "value": "entity_2"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 2
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svcA"
+ }
+ },
+ {
+ "str": {
+ "value": "entity_3"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 3
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svcB"
+ }
+ },
+ {
+ "str": {
+ "value": "entity_4"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 4
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svcA"
+ }
+ },
+ {
+ "str": {
+ "value": "entity_5"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 5
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svcB"
+ }
+ },
+ {
+ "str": {
+ "value": "entity_6"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 6
+ }
+ }
+ ]
+ }
+]
diff --git a/test/cases/measure/data/want/group_time_bucket.yaml b/test/cases/measure/data/want/group_time_bucket.yaml
new file mode 100644
index 000000000..b97a7b89f
--- /dev/null
+++ b/test/cases/measure/data/want/group_time_bucket.yaml
@@ -0,0 +1,33 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+dataPoints:
+- fields:
+ - name: value
+ value:
+ int:
+ value: "3"
+- fields:
+ - name: value
+ value:
+ int:
+ value: "12"
+- fields:
+ - name: value
+ value:
+ int:
+ value: "6"
diff --git a/test/cases/measure/data/want/group_time_bucket_with_tag.yaml b/test/cases/measure/data/want/group_time_bucket_with_tag.yaml
new file mode 100644
index 000000000..40a011aa1
--- /dev/null
+++ b/test/cases/measure/data/want/group_time_bucket_with_tag.yaml
@@ -0,0 +1,66 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+dataPoints:
+- fields:
+ - name: value
+ value:
+ int:
+ value: "3"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: id
+ value:
+ str:
+ value: svcA
+- fields:
+ - name: value
+ value:
+ int:
+ value: "8"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: id
+ value:
+ str:
+ value: svcA
+- fields:
+ - name: value
+ value:
+ int:
+ value: "4"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: id
+ value:
+ str:
+ value: svcB
+- fields:
+ - name: value
+ value:
+ int:
+ value: "6"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: id
+ value:
+ str:
+ value: svcB
diff --git a/test/cases/measure/measure.go b/test/cases/measure/measure.go
index ae5c8b0cf..7108579d4 100644
--- a/test/cases/measure/measure.go
+++ b/test/cases/measure/measure.go
@@ -113,6 +113,13 @@ var measureEntries = []any{
g.Entry("filter by tag with GE", helpers.Args{Input: "tag_filter_ge", Duration: 25 * time.Minute, Offset: -20 * time.Minute, DisOrder: true}),
g.Entry("complex AND/OR", helpers.Args{Input: "complex_and_or", Duration: 25 * time.Minute, Offset: -20 * time.Minute, DisOrder: true}),
g.Entry("group sum with filter", helpers.Args{Input: "group_sum_with_filter", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
+ // time_bucket_metric is seeded at now-100s..now (20s apart); a 1-minute
+ // bucket deterministically splits them 2/3/1 across three buckets
+ // regardless of which wall-clock minute `now` lands on (see
+ // test/cases/init.go). SkipQL: BydbQL has no TIME_BUCKET(...) grammar
+ // yet — apache/skywalking#14091.
+ g.Entry("group by time bucket", helpers.Args{Input: "group_time_bucket", Duration: 4 * time.Minute, Offset: -3 * time.Minute, SkipQL: true}),
+ g.Entry("group by time bucket and tag", helpers.Args{Input: "group_time_bucket_with_tag", Duration: 4 * time.Minute, Offset: -3 * time.Minute, SkipQL: true}),
g.Entry("top N with filter", helpers.Args{Input: "top_with_filter", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("index mode filter by NE", helpers.Args{Input: "index_mode_ne", Duration: 25 * time.Minute, Offset: -20 * time.Minute, DisOrder: true}),
g.Entry("index mode filter by LE on int", helpers.Args{Input: "index_mode_le", Duration: 25 * time.Minute, Offset: -20 * time.Minute, DisOrder: true}),