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/10/23 06:17:42 UTC
(skywalking-banyandb) branch main updated: Replace the multi-group Measure row merger with a batch merger (#1420)
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 57322bc8b Replace the multi-group Measure row merger with a batch merger (#1420)
57322bc8b is described below
commit 57322bc8b9f6dca5461eb0fe43401085e253804d
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Oct 23 14:17:23 2026 +0800
Replace the multi-group Measure row merger with a batch merger (#1420)
---
CHANGES.md | 1 +
banyand/query/measure_multi_group_dispatch_test.go | 688 +++++++++++++++++++++
banyand/query/processor.go | 53 +-
pkg/query/logical/measure/cross_group_merge.go | 131 ----
pkg/query/logical/measure/sortable_data_points.go | 49 --
pkg/query/logical/measure/sorted_miterator.go | 195 ------
pkg/query/logical/measure/sorted_miterator_test.go | 266 --------
.../vectorized/measure/plan/distributed_rows.go | 236 +++++--
.../measure/plan/distributed_rows_test.go | 23 +
.../vectorized/measure/plan/multi_group_merge.go | 444 +++++++++++++
.../measure/plan/multi_group_merge_test.go | 460 ++++++++++++++
pkg/query/vectorized/measure/plan/sort_key.go | 10 +-
pkg/query/vectorized/measure/plan/sort_key_test.go | 23 +-
pkg/query/vectorized/measure/serialize.go | 9 +
14 files changed, 1862 insertions(+), 726 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 06b3e3438..726f05e31 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -106,6 +106,7 @@ Release Notes.
- Bump mcp/canopy npm deps (hono, js-yaml, vitest 5) to clear Dependabot CVEs.
- Bump the pending mcp npm dependencies (`fast-uri` 4.2.1, `ip-address` 10.7.3, `proxy-addr` 2.0.8, and the `brace-expansion` 5.0.12 dev dependency) and regenerate the committed `mcp/LICENSE`, so `make check-req` passes on the Dependabot PRs that only edit `package-lock.json`.
- Stop `test-build-system.yml` reporting a version drift when the image carries the pinned version: `grep -q` inside a pipeline exited on its first match, SIGPIPEd the writer, and GitHub's default `bash -eo pipefail` turned that into a pipeline failure. The same pipeline shape hid a real regression in the "no version literal" check, where a false 141 skipped the check. Both now read from a here-string, and the platform probe no longer ends in `head -1`.
+- Merge the per-group results of a local multi-group Measure query as columnar batches (`vecplan.DispatchBatches` and `MultiGroupMerge`, built on the liaison row merger) instead of through the row merger, and remove `MergeGroupMIterators`, `sortableDataPoints` and `sortedMIterator`. Hidden criteria tags are stripped once, after the merge, and the trace plan string now reads `vec-multi-measure-batch[...]`. The query `offset` is applied once to the merged rows instead of to each group befo [...]
## 0.11.1
diff --git a/banyand/query/measure_multi_group_dispatch_test.go b/banyand/query/measure_multi_group_dispatch_test.go
new file mode 100644
index 000000000..08bacfe53
--- /dev/null
+++ b/banyand/query/measure_multi_group_dispatch_test.go
@@ -0,0 +1,688 @@
+// 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 query
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ logical_measure "github.com/apache/skywalking-banyandb/pkg/query/logical/measure"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized"
+ vmeasure "github.com/apache/skywalking-banyandb/pkg/query/vectorized/measure"
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized/measure/frame"
+)
+
+const (
+ mgGroupA = "group-a"
+ mgGroupB = "group-b"
+ mgFamily = "default"
+ mgTagSvc = "svc"
+ mgTagRegion = "region"
+ mgTagExtra = "extra"
+ mgField = "value"
+ mgIndexRule = "svc-rule"
+ mgTagLabels = "labels"
+)
+
+// mgRow is one stored row: the (ts, SID, version, value) tuple of the issue's oracle plus its tags.
+type mgRow struct {
+ tags map[string]string
+ arrays map[string][]string
+ ts int64
+ sid uint64
+ version int64
+ value int64
+}
+
+// mgGroup is a fake storage group: it serves its rows in the given storage batches.
+type mgGroup struct {
+ onEOF func()
+ queryErr error
+ schema *databasev1.Measure
+ result *mgResult
+ batches [][]mgRow
+ pullErrAt int
+}
+
+func (g *mgGroup) Query(_ context.Context, opts model.MeasureQueryOptions) (model.MeasureQueryResult, error) {
+ if g.queryErr != nil {
+ return nil, g.queryErr
+ }
+ schema, schemaErr := vmeasure.BuildBatchSchema(g.schema, opts)
+ if schemaErr != nil {
+ return nil, schemaErr
+ }
+ g.result = &mgResult{pullErrAt: g.pullErrAt, onEOF: g.onEOF}
+ for _, rows := range g.batches {
+ g.result.batches = append(g.result.batches, buildMgBatch(schema, rows))
+ }
+ return g.result, nil
+}
+
+func (g *mgGroup) VectorizedConfig() vmeasure.VectorizedConfig {
+ // One row per batch, so a group's rows reach the merge as several batches.
+ return vmeasure.VectorizedConfig{BatchSize: 1, QueryMemoryMiB: 16}
+}
+
+func (g *mgGroup) GetSchema() *databasev1.Measure { return g.schema }
+
+var _ vecExecutionContext = (*mgGroup)(nil)
+
+// mgResult is a MeasureBatchResult over prebuilt storage batches that counts Release calls.
+type mgResult struct {
+ onEOF func()
+ batches []*model.MeasureBatch
+ pullErrAt int
+ pulled int
+ released int
+}
+
+func (r *mgResult) Pull() *model.MeasureResult { return nil }
+
+func (r *mgResult) PullBatch(context.Context) (*model.MeasureBatch, error) {
+ if r.pullErrAt > 0 && r.pulled+1 == r.pullErrAt {
+ return nil, errors.New("storage read failed")
+ }
+ if r.pulled >= len(r.batches) {
+ if r.onEOF != nil {
+ r.onEOF()
+ r.onEOF = nil
+ }
+ return nil, nil
+ }
+ batch := r.batches[r.pulled]
+ r.pulled++
+ return batch, nil
+}
+
+func (r *mgResult) Release() { r.released++ }
+
+func buildMgBatch(schema *vectorized.BatchSchema, rows []mgRow) *model.MeasureBatch {
+ batch := &model.MeasureBatch{Schema: schema}
+ for _, row := range rows {
+ batch.Timestamps = append(batch.Timestamps, row.ts)
+ batch.Versions = append(batch.Versions, row.version)
+ batch.ShardIDs = append(batch.ShardIDs, 0)
+ batch.SeriesIDs = append(batch.SeriesIDs, common.SeriesID(row.sid))
+ }
+ for _, def := range schema.Columns {
+ switch def.Role {
+ case vectorized.RoleTag:
+ col := vectorized.NewColumnForType(def.Type, len(rows))
+ for _, row := range rows {
+ if arrCol, isArr := col.(*vectorized.TypedColumn[[]string]); isArr {
+ if values, has := row.arrays[def.Name]; has {
+ arrCol.Append(values)
+ } else {
+ col.AppendNull()
+ }
+ continue
+ }
+ value, ok := row.tags[def.Name]
+ strCol, isStr := col.(*vectorized.TypedColumn[string])
+ if !ok || !isStr {
+ col.AppendNull()
+ continue
+ }
+ strCol.Append(value)
+ }
+ batch.Tags = append(batch.Tags, col)
+ case vectorized.RoleField:
+ col := vectorized.NewColumnForType(def.Type, len(rows))
+ for _, row := range rows {
+ col.(*vectorized.TypedColumn[int64]).Append(row.value)
+ }
+ batch.Fields = append(batch.Fields, col)
+ case vectorized.RoleTimestamp, vectorized.RoleVersion, vectorized.RoleSeriesID,
+ vectorized.RoleShardID, vectorized.RoleElementID, vectorized.RoleOrderKey:
+ }
+ }
+ return batch
+}
+
+func mgMeasure(group string, tags ...string) *databasev1.Measure {
+ specs := make([]*databasev1.TagSpec, 0, len(tags))
+ for _, tag := range tags {
+ tagType := databasev1.TagType_TAG_TYPE_STRING
+ if tag == mgTagLabels {
+ tagType = databasev1.TagType_TAG_TYPE_STRING_ARRAY
+ }
+ specs = append(specs, &databasev1.TagSpec{Name: tag, Type: tagType})
+ }
+ return &databasev1.Measure{
+ Metadata: &commonv1.Metadata{Name: "m", Group: group},
+ Entity: &databasev1.Entity{TagNames: []string{mgTagSvc}},
+ TagFamilies: []*databasev1.TagFamilySpec{{Name: mgFamily, Tags: specs}},
+ Fields: []*databasev1.FieldSpec{{Name: mgField, FieldType: databasev1.FieldType_FIELD_TYPE_INT}},
+ }
+}
+
+func mgContext(t *testing.T, groups ...*mgGroup) *measureExecutionContext {
+ t.Helper()
+ rules := []*databasev1.IndexRule{
+ {
+ Metadata: &commonv1.Metadata{Name: mgIndexRule, Group: mgGroupA},
+ Tags: []string{mgTagSvc},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ },
+ {
+ Metadata: &commonv1.Metadata{Name: "labels-rule", Group: mgGroupA},
+ Tags: []string{mgTagLabels},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ },
+ {
+ Metadata: &commonv1.Metadata{Name: "region-rule", Group: mgGroupA},
+ Tags: []string{mgTagRegion},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ },
+ }
+ mctx := &measureExecutionContext{}
+ for _, g := range groups {
+ schema, schemaErr := logical_measure.BuildSchema(g.schema, rules)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ mctx.ecc = append(mctx.ecc, g)
+ mctx.schemas = append(mctx.schemas, schema)
+ mctx.metadata = append(mctx.metadata, g.schema.GetMetadata())
+ }
+ return mctx
+}
+
+// mgRequest projects svc and extra and filters on region, a criteria tag outside the projection.
+func mgRequest(sort modelv1.Sort, indexRule string, offset, limit uint32) *measurev1.QueryRequest {
+ return &measurev1.QueryRequest{
+ Name: "m",
+ Groups: []string{mgGroupA, mgGroupB},
+ TimeRange: &modelv1.TimeRange{
+ Begin: timestamppb.New(time.Unix(0, 0)),
+ End: timestamppb.New(time.Unix(1, 0)),
+ },
+ TagProjection: &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{
+ {Name: mgFamily, Tags: []string{mgTagSvc, mgTagExtra}},
+ }},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{mgField}},
+ Criteria: &modelv1.Criteria{Exp: &modelv1.Criteria_Condition{Condition: &modelv1.Condition{
+ Name: mgTagRegion,
+ Op: modelv1.Condition_BINARY_OP_EQ,
+ Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "r"}}},
+ }}},
+ OrderBy: &modelv1.QueryOrder{Sort: sort, IndexRuleName: indexRule},
+ Offset: offset,
+ Limit: limit,
+ }
+}
+
+// mgOracleGroups builds the issue's oracle: duplicate versions split across batches and groups.
+// Group B's second batch holds the lower version of the same (ts, SID), which must lose to version 3.
+func mgOracleGroups() (*mgGroup, *mgGroup) {
+ groupA := &mgGroup{
+ schema: mgMeasure(mgGroupA, mgTagSvc, mgTagRegion),
+ batches: [][]mgRow{
+ {{ts: 10, sid: 7, version: 1, value: 100, tags: map[string]string{mgTagSvc: "x", mgTagRegion: "r"}}},
+ {{ts: 20, sid: 8, version: 1, value: 200, tags: map[string]string{mgTagSvc: "y", mgTagRegion: "r"}}},
+ },
+ }
+ groupB := &mgGroup{
+ schema: mgMeasure(mgGroupB, mgTagSvc, mgTagRegion, mgTagExtra),
+ batches: [][]mgRow{
+ {{ts: 10, sid: 7, version: 3, value: 300, tags: map[string]string{mgTagSvc: "x", mgTagRegion: "r", mgTagExtra: "e"}}},
+ {{ts: 10, sid: 7, version: 2, value: 250, tags: map[string]string{mgTagSvc: "x", mgTagRegion: "r", mgTagExtra: "e"}}},
+ },
+ }
+ return groupA, groupB
+}
+
+// mgDispatch runs dispatchMeasure and turns a panic inside it into a test failure, so a
+// regression reads as a failed assertion rather than aborting the whole test binary.
+func mgDispatch(ctx context.Context, t *testing.T, req *measurev1.QueryRequest, mctx *measureExecutionContext,
+) (iter executor.MIterator, planStr string, err error) {
+ t.Helper()
+ defer func() {
+ if recovered := recover(); recovered != nil {
+ t.Fatalf("dispatchMeasure panicked: %v", recovered)
+ }
+ }()
+ return dispatchMeasure(ctx, req, mctx, false)
+}
+
+func drainMgRows(t *testing.T, iter executor.MIterator) ([]string, []*measurev1.DataPoint) {
+ t.Helper()
+ var rows []string
+ var points []*measurev1.DataPoint
+ for iter.Next() {
+ for _, idp := range iter.Current() {
+ dp := idp.GetDataPoint()
+ rows = append(rows, fmt.Sprintf("(%d,%d,%d,%d)", dp.GetTimestamp().AsTime().UnixNano(), dp.GetSid(), dp.GetVersion(),
+ dp.GetFields()[0].GetValue().GetInt().GetValue()))
+ points = append(points, dp)
+ }
+ }
+ return rows, points
+}
+
+func TestDispatchMeasureMultiGroupBatchMerge(t *testing.T) {
+ oracleAsc := []string{"(10,7,3,300)", "(20,8,1,200)"}
+ oracleDesc := []string{"(20,8,1,200)", "(10,7,3,300)"}
+ tests := []struct {
+ name string
+ indexRule string
+ want []string
+ sort modelv1.Sort
+ offset uint32
+ limit uint32
+ }{
+ {name: "time ascending", sort: modelv1.Sort_SORT_ASC, limit: 10, want: oracleAsc},
+ {name: "time descending", sort: modelv1.Sort_SORT_DESC, limit: 10, want: oracleDesc},
+ {name: "time ascending offset 1 limit 1", sort: modelv1.Sort_SORT_ASC, offset: 1, limit: 1, want: []string{"(20,8,1,200)"}},
+ {name: "index order ascending", indexRule: mgIndexRule, sort: modelv1.Sort_SORT_ASC, limit: 10, want: oracleAsc},
+ {name: "index order descending", indexRule: mgIndexRule, sort: modelv1.Sort_SORT_DESC, limit: 10, want: oracleDesc},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ mctx := mgContext(t, groupA, groupB)
+ iter, planStr, dispatchErr := mgDispatch(context.Background(), t, mgRequest(tt.sort, tt.indexRule, tt.offset, tt.limit), mctx)
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ if !strings.HasPrefix(planStr, "vec-multi-measure-batch[") {
+ t.Fatalf("plan %q must name the batch merge, vec-multi-measure-batch[...]", planStr)
+ }
+ rows, points := drainMgRows(t, iter)
+ if closeErr := iter.Close(); closeErr != nil {
+ t.Fatalf("Close: %v", closeErr)
+ }
+ if got, want := strings.Join(rows, " "), strings.Join(tt.want, " "); got != want {
+ t.Fatalf("rows = %s, want %s", got, want)
+ }
+ for _, dp := range points {
+ for _, family := range dp.GetTagFamilies() {
+ for _, tag := range family.GetTags() {
+ if tag.GetKey() == mgTagRegion {
+ t.Fatalf("hidden criteria tag %q leaked into %v", mgTagRegion, dp)
+ }
+ }
+ }
+ }
+ if groupA.result.released == 0 || groupB.result.released == 0 {
+ t.Fatalf("closing the merged iterator must release both groups, released a=%d b=%d", groupA.result.released, groupB.result.released)
+ }
+ // (20,8) comes from group A alone, which does not define the extra tag.
+ for _, dp := range points {
+ if dp.GetSid() != 8 {
+ continue
+ }
+ for _, family := range dp.GetTagFamilies() {
+ for _, tag := range family.GetTags() {
+ if tag.GetKey() == mgTagExtra && tag.GetValue().GetStr() != nil {
+ t.Fatalf("tag %q only group B defines must be null on group A rows, got %v", mgTagExtra, tag.GetValue())
+ }
+ }
+ }
+ }
+ })
+ }
+
+ t.Run("tag only group B defines is returned for its rows", testMultiGroupExtraTag)
+ t.Run("a group whose dispatch fails releases the batches already read", testMultiGroupDispatchFailure)
+ t.Run("a batch read failure releases every group", testMultiGroupReadFailure)
+ t.Run("a canceled context releases every group", testMultiGroupCanceled)
+ t.Run("a context canceled after the groups were read stops the merge", testMultiGroupCanceledDuringMerge)
+ t.Run("Top keeps cross-group order and version dedup", testMultiGroupTop)
+ t.Run("null index values form their own sort window", testMultiGroupNullSortKey)
+ t.Run("array index values order the merge", testMultiGroupArraySortKey)
+ t.Run("null and empty index values stay distinct windows", testMultiGroupNullAndEmptySortKey)
+ t.Run("an empty passthrough value sorts after a null one", testMultiGroupEmptyVersusNullPassthrough)
+ t.Run("raw GroupBy keeps cross-group version dedup", testMultiGroupRawGroupBy)
+ t.Run("Agg results are concatenated per group", testMultiGroupAgg)
+ t.Run("time-bucketed Agg results are merged by time", testMultiGroupBucketedAgg)
+ t.Run("the merged result emits a raw frame of the merged batches", testMultiGroupFrame)
+}
+
+func testMultiGroupExtraTag(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10), mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ _, points := drainMgRows(t, iter)
+ if len(points) == 0 {
+ t.Fatal("no rows")
+ }
+ var extra string
+ for _, family := range points[0].GetTagFamilies() {
+ for _, tag := range family.GetTags() {
+ if tag.GetKey() == mgTagExtra {
+ extra = tag.GetValue().GetStr().GetValue()
+ }
+ }
+ }
+ if extra != "e" {
+ t.Fatalf("row (10,7,3,300) comes from group B and must carry extra=e, got %q", extra)
+ }
+}
+
+func testMultiGroupDispatchFailure(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ groupB.queryErr = errors.New("group b unavailable")
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10), mgContext(t, groupA, groupB))
+ if dispatchErr == nil || !strings.Contains(dispatchErr.Error(), "group b unavailable") {
+ t.Fatalf("want the group b error, got iter=%v err=%v", iter, dispatchErr)
+ }
+ if groupA.result.released == 0 {
+ t.Fatal("group a was read before group b failed and must be released")
+ }
+}
+
+func testMultiGroupReadFailure(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ groupB.pullErrAt = 2
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10), mgContext(t, groupA, groupB))
+ if dispatchErr == nil || !strings.Contains(dispatchErr.Error(), "storage read failed") {
+ t.Fatalf("want the storage read error, got iter=%v err=%v", iter, dispatchErr)
+ }
+ if groupA.result.released == 0 || groupB.result.released == 0 {
+ t.Fatalf("every group must be released, released a=%d b=%d", groupA.result.released, groupB.result.released)
+ }
+}
+
+func testMultiGroupCanceled(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ iter, _, dispatchErr := mgDispatch(ctx, t, mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10), mgContext(t, groupA, groupB))
+ if !errors.Is(dispatchErr, context.Canceled) {
+ t.Fatalf("want context.Canceled, got iter=%v err=%v", iter, dispatchErr)
+ }
+ if groupA.result.released == 0 {
+ t.Fatal("group a must be released")
+ }
+}
+
+func testMultiGroupCanceledDuringMerge(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ // Cancel once the last group has been read, so only the merge phase can notice it.
+ groupB.onEOF = cancel
+ iter, _, dispatchErr := mgDispatch(ctx, t, mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10), mgContext(t, groupA, groupB))
+ if !errors.Is(dispatchErr, context.Canceled) {
+ t.Fatalf("want context.Canceled from the merge phase, got iter=%v err=%v", iter, dispatchErr)
+ }
+ if groupA.result.released == 0 || groupB.result.released == 0 {
+ t.Fatalf("every group must be released, released a=%d b=%d", groupA.result.released, groupB.result.released)
+ }
+}
+
+func testMultiGroupTop(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ req := mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10)
+ req.Top = &measurev1.QueryRequest_Top{Number: 10, FieldName: mgField, FieldValueSort: modelv1.Sort_SORT_DESC}
+ iter, planStr, dispatchErr := mgDispatch(context.Background(), t, req, mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ if !strings.HasPrefix(planStr, "vec-multi-measure-batch[") {
+ t.Fatalf("plan %q must name the batch merge", planStr)
+ }
+ rows, _ := drainMgRows(t, iter)
+ // Group B's Top returns both versions of (10,7); the merge keeps version 3 and orders by time.
+ if got, want := strings.Join(rows, " "), "(10,7,3,300) (20,8,1,200)"; got != want {
+ t.Fatalf("rows = %s, want %s", got, want)
+ }
+}
+
+// mgNullKeyGroups adds a row without an svc value, which sorts before every non-null value.
+func mgNullKeyGroups() (*mgGroup, *mgGroup) {
+ groupA, groupB := mgOracleGroups()
+ groupB.batches = append(groupB.batches, []mgRow{
+ {ts: 30, sid: 9, version: 1, value: 900, tags: map[string]string{mgTagRegion: "r"}},
+ })
+ return groupA, groupB
+}
+
+func testMultiGroupNullSortKey(t *testing.T) {
+ groupA, groupB := mgNullKeyGroups()
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, mgRequest(modelv1.Sort_SORT_ASC, mgIndexRule, 0, 10), mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ rows, _ := drainMgRows(t, iter)
+ // The null svc row (30,9) is its own window and comes first in ascending order.
+ if got, want := strings.Join(rows, " "), "(30,9,1,900) (10,7,3,300) (20,8,1,200)"; got != want {
+ t.Fatalf("rows = %s, want %s", got, want)
+ }
+}
+
+func testMultiGroupArraySortKey(t *testing.T) {
+ newGroup := func(group string, rows ...mgRow) *mgGroup {
+ batches := make([][]mgRow, 0, len(rows))
+ for _, row := range rows {
+ batches = append(batches, []mgRow{row})
+ }
+ return &mgGroup{schema: mgMeasure(group, mgTagSvc, mgTagRegion, mgTagLabels), batches: batches}
+ }
+ groupA := newGroup(mgGroupA,
+ mgRow{ts: 10, sid: 7, version: 1, value: 100, tags: map[string]string{mgTagRegion: "r"}, arrays: map[string][]string{mgTagLabels: {"b"}}},
+ mgRow{ts: 20, sid: 8, version: 1, value: 200, tags: map[string]string{mgTagRegion: "r"}, arrays: map[string][]string{mgTagLabels: {"a"}}})
+ groupB := newGroup(mgGroupB,
+ mgRow{ts: 30, sid: 9, version: 1, value: 300, tags: map[string]string{mgTagRegion: "r"}, arrays: map[string][]string{mgTagLabels: {"c"}}})
+ req := mgRequest(modelv1.Sort_SORT_ASC, "labels-rule", 0, 10)
+ // labels is not projected, so it is a hidden order-by column.
+ req.TagProjection = &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{{Name: mgFamily, Tags: []string{mgTagSvc}}}}
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, req, mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ rows, _ := drainMgRows(t, iter)
+ if got, want := strings.Join(rows, " "), "(20,8,1,200) (10,7,1,100) (30,9,1,300)"; got != want {
+ t.Fatalf("rows = %s, want %s", got, want)
+ }
+}
+
+func testMultiGroupNullAndEmptySortKey(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ groupA.batches = [][]mgRow{
+ {{ts: 10, sid: 7, version: 1, value: 100, tags: map[string]string{mgTagRegion: "r"}}},
+ {{ts: 20, sid: 9, version: 1, value: 200, tags: map[string]string{mgTagSvc: "x", mgTagRegion: "r"}}},
+ }
+ groupB.batches = [][]mgRow{
+ {{ts: 10, sid: 7, version: 3, value: 300, tags: map[string]string{mgTagRegion: "r"}}},
+ {{ts: 30, sid: 8, version: 1, value: 800, tags: map[string]string{mgTagSvc: "", mgTagRegion: "r"}}},
+ }
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, mgRequest(modelv1.Sort_SORT_ASC, mgIndexRule, 0, 10), mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ rows, _ := drainMgRows(t, iter)
+ // null < empty < "x"; the two null rows are one window, so version 3 replaces version 1.
+ if got, want := strings.Join(rows, " "), "(10,7,3,300) (30,8,1,800) (20,9,1,200)"; got != want {
+ t.Fatalf("rows = %s, want %s", got, want)
+ }
+}
+
+func testMultiGroupRawGroupBy(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ req := mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10)
+ req.GroupBy = &measurev1.QueryRequest_GroupBy{TagProjection: &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{
+ {Name: mgFamily, Tags: []string{mgTagSvc}},
+ }}, FieldName: mgField}
+ iter, planStr, dispatchErr := mgDispatch(context.Background(), t, req, mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ if !strings.HasPrefix(planStr, "vec-multi-measure-batch[") {
+ t.Fatalf("plan %q must name the batch merge", planStr)
+ }
+ rows, _ := drainMgRows(t, iter)
+ // Each group keeps its first row per svc; the merge then keeps version 3 of (10,7).
+ if got, want := strings.Join(rows, " "), "(10,7,3,300) (20,8,1,200)"; got != want {
+ t.Fatalf("rows = %s, want %s", got, want)
+ }
+}
+
+func mgFieldNumber(dp *measurev1.DataPoint) float64 {
+ value := dp.GetFields()[0].GetValue()
+ if value.GetFloat() != nil {
+ return value.GetFloat().GetValue()
+ }
+ return float64(value.GetInt().GetValue())
+}
+
+func testMultiGroupAgg(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ req := mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10)
+ req.Agg = &measurev1.QueryRequest_Aggregation{Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM, FieldName: mgField}
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, req, mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ var sums []float64
+ for iter.Next() {
+ sums = append(sums, mgFieldNumber(iter.Current()[0].GetDataPoint()))
+ }
+ // One partial sum per group, in group order: A = 100+200, B = 300+250. Reducing them across
+ // groups is #14069.
+ if len(sums) != 2 || sums[0] != 300 || sums[1] != 550 {
+ t.Fatalf("sums = %v, want [300 550]", sums)
+ }
+}
+
+func testMultiGroupBucketedAgg(t *testing.T) {
+ const second = int64(time.Second)
+ groupA, groupB := mgOracleGroups()
+ groupA.batches = [][]mgRow{
+ {{ts: 2 * second, sid: 7, version: 1, value: 20, tags: map[string]string{mgTagSvc: "x", mgTagRegion: "r"}}},
+ {{ts: 4 * second, sid: 7, version: 1, value: 40, tags: map[string]string{mgTagSvc: "x", mgTagRegion: "r"}}},
+ }
+ groupB.batches = [][]mgRow{
+ {{ts: 1 * second, sid: 7, version: 1, value: 10, tags: map[string]string{mgTagSvc: "x", mgTagRegion: "r"}}},
+ {{ts: 3 * second, sid: 7, version: 1, value: 30, tags: map[string]string{mgTagSvc: "x", mgTagRegion: "r"}}},
+ }
+ req := mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10)
+ req.Criteria = nil
+ req.TimeRange = &modelv1.TimeRange{Begin: timestamppb.New(time.Unix(0, 0)), End: timestamppb.New(time.Unix(10, 0))}
+ req.GroupBy = &measurev1.QueryRequest_GroupBy{TimeBucket: &measurev1.QueryRequest_GroupBy_TimeBucket{Width: "1s"}}
+ req.Agg = &measurev1.QueryRequest_Aggregation{Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM, FieldName: mgField}
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, req, mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ var buckets []string
+ for iter.Next() {
+ dp := iter.Current()[0].GetDataPoint()
+ buckets = append(buckets, fmt.Sprintf("%d:%v", dp.GetTimestamp().AsTime().Unix(), mgFieldNumber(dp)))
+ }
+ if got, want := strings.Join(buckets, " "), "1:10 2:20 3:30 4:40"; got != want {
+ t.Fatalf("buckets = %s, want %s (ordered by time across groups)", got, want)
+ }
+}
+
+func testMultiGroupFrame(t *testing.T) {
+ groupA, groupB := mgOracleGroups()
+ req := mgRequest(modelv1.Sort_SORT_ASC, "", 0, 10)
+ req.Criteria = nil
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, req, mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ // The batch merge ends in a vectorized pipeline over the merged batches; a result built from
+ // rows would have neither capability.
+ if _, ok := iter.(vmeasure.RawFrameSource); !ok {
+ t.Fatalf("%T must expose the merged pipeline as a RawFrameSource", iter)
+ }
+ emitter, ok := iter.(vmeasure.FrameEmitter)
+ if !ok {
+ t.Fatalf("%T must emit a raw frame", iter)
+ }
+ body, emitErr := emitter.EmitFrame(context.Background())
+ if emitErr != nil {
+ t.Fatalf("EmitFrame: %v", emitErr)
+ }
+ batch, decodeErr := frame.Decode(body)
+ if decodeErr != nil {
+ t.Fatalf("Decode: %v", decodeErr)
+ }
+ defer batch.ReleaseOwners()
+ if batch.ActiveLen() != 2 {
+ t.Fatalf("frame rows = %d, want 2", batch.ActiveLen())
+ }
+ timestamps := batch.Columns[batch.Schema.TimestampIndex()].(*vectorized.TypedColumn[int64]).Data()
+ if timestamps[0] != 10 || timestamps[1] != 20 {
+ t.Fatalf("frame timestamps = %v, want [10 20]", timestamps[:2])
+ }
+ if closeErr := iter.Close(); closeErr != nil {
+ t.Fatalf("Close: %v", closeErr)
+ }
+ if groupA.result.released == 0 || groupB.result.released == 0 {
+ t.Fatalf("closing the frame source must release both groups, released a=%d b=%d", groupA.result.released, groupB.result.released)
+ }
+}
+
+// testMultiGroupEmptyVersusNullPassthrough makes svc a string in group A and an int in group B, so
+// the merged column is a passthrough TagValue whose encoder returns no bytes for an empty string.
+// Group A's empty string must still sort after group B's null.
+func testMultiGroupEmptyVersusNullPassthrough(t *testing.T) {
+ groupA := &mgGroup{
+ schema: mgMeasure(mgGroupA, mgTagSvc, mgTagRegion),
+ batches: [][]mgRow{
+ {{ts: 10, sid: 1, version: 1, value: 100, tags: map[string]string{mgTagSvc: "", mgTagRegion: "r"}}},
+ },
+ }
+ schemaB := mgMeasure(mgGroupB, mgTagSvc, mgTagRegion)
+ schemaB.TagFamilies[0].Tags[0].Type = databasev1.TagType_TAG_TYPE_INT
+ groupB := &mgGroup{
+ schema: schemaB,
+ batches: [][]mgRow{
+ {{ts: 20, sid: 2, version: 1, value: 200, tags: map[string]string{mgTagRegion: "r"}}},
+ },
+ }
+ req := mgRequest(modelv1.Sort_SORT_ASC, mgIndexRule, 0, 1)
+ req.TagProjection = &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{{Name: mgFamily, Tags: []string{mgTagSvc}}}}
+ iter, _, dispatchErr := mgDispatch(context.Background(), t, req, mgContext(t, groupA, groupB))
+ if dispatchErr != nil {
+ t.Fatalf("dispatchMeasure: %v", dispatchErr)
+ }
+ defer iter.Close()
+ rows, _ := drainMgRows(t, iter)
+ if got, want := strings.Join(rows, " "), "(20,2,1,200)"; got != want {
+ t.Fatalf("rows = %s, want %s (the null svc row first)", got, want)
+ }
+}
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index 7575056f6..e675f6b2a 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -591,14 +591,9 @@ func executeMeasurePlan(
return mit, planStr, nil
}
-// dispatchMeasure is the thin adapter from measureExecutionContext into
-// the vec dispatch inputs. Multi-measure queries (len(ecc) > 1) are
-// handled per-group: each group is dispatched individually and the
-// per-group iterators are merged through the shared cross-group ordering
-// stack in logical_measure.MergeGroupMIterators (G9f.1, reusing
-// sortableDataPoints + sortedMIterator so the cross-group order and
-// version dedup are reproduced by construction, not by parallel
-// reimplementation).
+// dispatchMeasure adapts the native BatchResult boundary to final row serialization.
+// Multi-group queries use DispatchBatches and MultiGroupMerge to align schemas,
+// merge sort keys, deduplicate versions and apply pagination before serialization.
//
// Distributed Map-mode GroupBy+Agg (emitPartial=true) routes to vec via
// AggModeMap (G9f.2): vecplan.Dispatch receives emitPartial and the
@@ -655,39 +650,35 @@ func dispatchMeasureBatches(
if projErr := vecplan.ValidateMultiGroupProjection(queryCriteria, mctx.schemas, measureSchemas); projErr != nil {
return nil, "", projErr
}
- iters := make([]*vecplan.BatchResult, 0, len(mctx.ecc))
- planStrs := make([]string, 0, len(mctx.ecc))
- closeOpened := func() {
- for _, opened := range iters {
- _ = opened.Close()
- }
+ merge, mergeErr := vecplan.NewMultiGroupMerge(ctx, queryCriteria, mctx.schemas, measureSchemas, vecs[0].VectorizedConfig())
+ if mergeErr != nil {
+ return nil, "", mergeErr
}
+ groups := make([]*vecplan.GroupBatches, 0, len(mctx.ecc))
+ planStrs := make([]string, 0, len(mctx.ecc))
for groupIdx, vec := range vecs {
- mit, planStr, dispatchErr := vecplan.DispatchBatches(ctx, queryCriteria,
+ result, planStr, dispatchErr := vecplan.DispatchBatches(ctx, merge.GroupRequest(),
mctx.metadata[groupIdx], vec.GetSchema(), mctx.schemas[groupIdx], vec, vec.VectorizedConfig(), emitPartial, true)
+ if dispatchErr == nil {
+ var group *vecplan.GroupBatches
+ group, dispatchErr = merge.Collect(ctx, result, planStr)
+ if dispatchErr == nil {
+ groups = append(groups, group)
+ planStrs = append(planStrs, planStr)
+ }
+ }
if dispatchErr != nil {
- closeOpened()
+ for _, obtained := range groups {
+ obtained.Release()
+ }
return nil, "", dispatchErr
}
- iters = append(iters, mit)
- planStrs = append(planStrs, planStr)
- }
- order, orderErr := logical_measure.ResolveCrossGroupMergeOrder(queryCriteria, mctx.schemas)
- if orderErr != nil {
- closeOpened()
- return nil, "", orderErr
- }
- _ = order
- schema, schemaErr := vecplan.BuildMultiGroupBatchSchema(measureSchemas, queryCriteria)
- if schemaErr != nil {
- closeOpened()
- return nil, "", schemaErr
}
- merged, mergeErr := vecplan.MergeStandaloneBatches(ctx, iters, schema, queryCriteria, mctx.schemas, vecs[0].VectorizedConfig())
+ merged, mergeErr := merge.MergeBatches(ctx, groups)
if mergeErr != nil {
return nil, "", mergeErr
}
- return merged, fmt.Sprintf("vec-multi-measure%v", planStrs), nil
+ return merged, fmt.Sprintf("vec-multi-measure-batch%v", planStrs), nil
}
// collectInternalDataPoints collects InternalDataPoints from the iterator.
diff --git a/pkg/query/logical/measure/cross_group_merge.go b/pkg/query/logical/measure/cross_group_merge.go
deleted file mode 100644
index c7557df71..000000000
--- a/pkg/query/logical/measure/cross_group_merge.go
+++ /dev/null
@@ -1,131 +0,0 @@
-// 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 (
- "fmt"
-
- measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
- modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
- "github.com/apache/skywalking-banyandb/pkg/iter/sort"
- "github.com/apache/skywalking-banyandb/pkg/query/executor"
- "github.com/apache/skywalking-banyandb/pkg/query/logical"
-)
-
-// CrossGroupMergeOrder is the resolved ordering for a multi-group
-// (multi-measure) query: which field each per-group stream is sorted on
-// and whether the k-way merge is descending. It is derived from the
-// request's OrderBy against the projected merged multi-group schema,
-// reproducing exactly the row path's unresolvedMerger.Analyze ordering.
-type CrossGroupMergeOrder struct {
- sortTagSpec logical.TagSpec
- sortByTime bool
- desc bool
-}
-
-// resolveCrossGroupMergeOrder is the order-resolution body shared
-// verbatim by the row unresolvedMerger.Analyze and the exported
-// ResolveCrossGroupMergeOrder. s MUST be the projected merged schema
-// (post-ProjTags when a tag projection is set) so an index-rule
-// OrderBy's TagFamilyIdx/TagIdx match the projected InternalDataPoint
-// layout newComparableElement indexes into.
-func resolveCrossGroupMergeOrder(orderBy *modelv1.QueryOrder, s logical.Schema) (CrossGroupMergeOrder, error) {
- var order CrossGroupMergeOrder
- if orderBy == nil {
- order.sortByTime = true
- return order, nil
- }
- if orderBy.IndexRuleName == "" {
- order.sortByTime = true
- if orderBy.Sort == modelv1.Sort_SORT_DESC {
- order.desc = true
- }
- return order, nil
- }
-
- ok, indexRule := s.IndexRuleDefined(orderBy.IndexRuleName)
- if !ok {
- return order, fmt.Errorf("index rule %s not found", orderBy.IndexRuleName)
- }
- if len(indexRule.Tags) != 1 {
- return order, fmt.Errorf("index rule %s should have one tag", orderBy.IndexRuleName)
- }
- sortTagSpec := s.FindTagSpecByName(indexRule.Tags[0])
- if sortTagSpec == nil {
- return order, fmt.Errorf("tag %s not found", indexRule.Tags[0])
- }
- order.sortTagSpec = *sortTagSpec
- if orderBy.Sort == modelv1.Sort_SORT_DESC {
- order.desc = true
- }
- return order, nil
-}
-
-// ResolveCrossGroupMergeOrder reproduces the projected-merged-schema
-// resolution that Analyze + unresolvedMerger.Analyze perform for a
-// multi-group request, so a caller holding per-group executor.MIterator
-// streams (banyand/query/processor.go, G9f.1) merges them with the row
-// path's exact cross-group ordering. ss are the per-group logical
-// schemas in request order; they are merged via mergeSchema and, when
-// criteria carries a tag projection, projected via ProjTags exactly as
-// the row analyzer does before the OrderBy is resolved.
-func ResolveCrossGroupMergeOrder(criteria *measurev1.QueryRequest, ss []logical.Schema) (CrossGroupMergeOrder, error) {
- var order CrossGroupMergeOrder
- s, mergeErr := mergeSchema(ss)
- if mergeErr != nil {
- return order, mergeErr
- }
- if s == nil {
- if criteria.GetOrderBy() == nil || criteria.GetOrderBy().GetIndexRuleName() == "" {
- return resolveCrossGroupMergeOrder(criteria.GetOrderBy(), nil)
- }
- return order, fmt.Errorf("cannot resolve order_by %s: no schema", criteria.GetOrderBy().GetIndexRuleName())
- }
- tagProjection := logical.ToTags(criteria.GetTagProjection())
- if len(tagProjection) > 0 {
- projectionTagRefs, refErr := s.CreateTagRef(tagProjection...)
- if refErr != nil {
- return order, refErr
- }
- s = s.ProjTags(projectionTagRefs...)
- }
- return resolveCrossGroupMergeOrder(criteria.GetOrderBy(), s)
-}
-
-// MergeGroupMIterators wraps the per-group executor.MIterator streams in
-// the identical sortableDataPoints -> sort.NewItemIter -> sortedMIterator
-// stack mergePlan.Execute builds for the row path. Because the merge and
-// equal-SortedField version dedup machinery is shared verbatim, the
-// cross-group order and dedup are reproduced by construction, not by a
-// parallel reimplementation. The input iters are consumed lazily and are
-// closed by the returned iterator's Close.
-func MergeGroupMIterators(iters []executor.MIterator, order CrossGroupMergeOrder) executor.MIterator {
- sortableIters := make([]sort.Iterator[*comparableDataPoint], 0, len(iters))
- for _, it := range iters {
- sortableIters = append(sortableIters, &sortableDataPoints{
- iter: it,
- sortTagSpec: order.sortTagSpec,
- sortByTime: order.sortByTime,
- })
- }
- merged := &sortedMIterator{
- Iterator: sort.NewItemIter(sortableIters, order.desc),
- }
- merged.init()
- return merged
-}
diff --git a/pkg/query/logical/measure/sortable_data_points.go b/pkg/query/logical/measure/sortable_data_points.go
deleted file mode 100644
index 21d8c8758..000000000
--- a/pkg/query/logical/measure/sortable_data_points.go
+++ /dev/null
@@ -1,49 +0,0 @@
-// 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 (
- "github.com/apache/skywalking-banyandb/pkg/query/executor"
- "github.com/apache/skywalking-banyandb/pkg/query/logical"
-)
-
-type sortableDataPoints struct {
- iter executor.MIterator
- current *comparableDataPoint
- sortTagSpec logical.TagSpec
- sortByTime bool
-}
-
-func (s *sortableDataPoints) Next() bool {
- if !s.iter.Next() {
- return false
- }
-
- dp := s.iter.Current()[0]
- var err error
- s.current, err = newComparableElement(dp, s.sortByTime, s.sortTagSpec)
- return err == nil
-}
-
-func (s *sortableDataPoints) Val() *comparableDataPoint {
- return s.current
-}
-
-func (s *sortableDataPoints) Close() error {
- return s.iter.Close()
-}
diff --git a/pkg/query/logical/measure/sorted_miterator.go b/pkg/query/logical/measure/sorted_miterator.go
deleted file mode 100644
index 4e7a382b6..000000000
--- a/pkg/query/logical/measure/sorted_miterator.go
+++ /dev/null
@@ -1,195 +0,0 @@
-// 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 (
- "bytes"
- "container/list"
-
- measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
- "github.com/apache/skywalking-banyandb/pkg/convert"
- "github.com/apache/skywalking-banyandb/pkg/iter/sort"
- pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
- "github.com/apache/skywalking-banyandb/pkg/query/executor"
- "github.com/apache/skywalking-banyandb/pkg/query/logical"
-)
-
-var _ sort.Comparable = (*comparableDataPoint)(nil)
-
-type comparableDataPoint struct {
- *measurev1.InternalDataPoint
- sortField []byte
-}
-
-func newComparableElement(idp *measurev1.InternalDataPoint, sortByTime bool, sortTagSpec logical.TagSpec) (*comparableDataPoint, error) {
- dp := idp.GetDataPoint()
- var sortField []byte
- if sortByTime {
- sortField = convert.Uint64ToBytes(uint64(dp.Timestamp.AsTime().UnixNano()))
- } else {
- var err error
- sortField, err = pbv1.MarshalTagValue(dp.TagFamilies[sortTagSpec.TagFamilyIdx].Tags[sortTagSpec.TagIdx].Value)
- if err != nil {
- return nil, err
- }
- }
-
- return &comparableDataPoint{
- InternalDataPoint: idp,
- sortField: sortField,
- }, nil
-}
-
-func (e *comparableDataPoint) SortedField() []byte {
- return e.sortField
-}
-
-var _ executor.MIterator = (*sortedMIterator)(nil)
-
-type sortedMIterator struct {
- sort.Iterator[*comparableDataPoint]
- data *list.List
- uniqueData map[uint64]*measurev1.InternalDataPoint
- cur *measurev1.InternalDataPoint
- seenSids map[uint64]struct{}
- initialized bool
- exhausted bool
- closed bool
- indexMode bool
-}
-
-func (s *sortedMIterator) init() {
- if s.initialized {
- return
- }
- s.initialized = true
- if !s.Iterator.Next() {
- s.exhausted = true
- return
- }
- s.data = list.New()
- s.uniqueData = make(map[uint64]*measurev1.InternalDataPoint)
- if s.indexMode {
- s.seenSids = make(map[uint64]struct{})
- }
- s.loadDps()
-}
-
-func (s *sortedMIterator) Next() bool {
- if s.data == nil {
- return false
- }
- if s.data.Len() == 0 {
- s.loadDps()
- if s.data.Len() == 0 {
- return false
- }
- }
- idp := s.data.Front()
- s.data.Remove(idp)
- s.cur = idp.Value.(*measurev1.InternalDataPoint)
- return true
-}
-
-// loadDps consumes sort-field-equal groups from the underlying iterator and
-// pushes their deduplicated rows into s.data. When indexMode is true, rows
-// whose Sid was already emitted in a prior group are skipped — without this,
-// cross-node duplicates that carry different per-node "last-write" timestamps
-// would slip past hashDataPoint (which keys on Sid+timestamp) because they
-// land in different sort-field groups. When every row in a group is filtered,
-// the loop advances to the next group so callers see continued iteration.
-func (s *sortedMIterator) loadDps() {
- for !s.exhausted {
- s.loadOneGroup()
- if s.data.Len() > 0 {
- return
- }
- }
-}
-
-func (s *sortedMIterator) loadOneGroup() {
- for k := range s.uniqueData {
- delete(s.uniqueData, k)
- }
- first := s.Iterator.Val()
- s.uniqueData[hashDataPoint(first.GetDataPoint())] = first.InternalDataPoint
- for {
- if !s.Iterator.Next() {
- s.exhausted = true
- break
- }
- v := s.Iterator.Val()
- if bytes.Equal(first.SortedField(), v.SortedField()) {
- key := hashDataPoint(v.GetDataPoint())
- if existed, ok := s.uniqueData[key]; ok {
- if v.GetDataPoint().Version > existed.GetDataPoint().Version {
- s.uniqueData[key] = v.InternalDataPoint
- }
- } else {
- s.uniqueData[key] = v.InternalDataPoint
- }
- } else {
- break
- }
- }
- for _, v := range s.uniqueData {
- if s.indexMode {
- sid := v.GetDataPoint().GetSid()
- if _, dup := s.seenSids[sid]; dup {
- continue
- }
- s.seenSids[sid] = struct{}{}
- }
- s.data.PushBack(v)
- }
-}
-
-func (s *sortedMIterator) Current() []*measurev1.InternalDataPoint {
- return []*measurev1.InternalDataPoint{s.cur}
-}
-
-func (s *sortedMIterator) Close() error {
- if s.closed {
- return nil
- }
- s.closed = true
- s.exhausted = true
- s.data = nil
- s.uniqueData = nil
- s.seenSids = nil
- s.cur = nil
- if s.Iterator == nil {
- return nil
- }
- return s.Iterator.Close()
-}
-
-const (
- offset64 = 14695981039346656037
- prime64 = 1099511628211
-)
-
-// hashDataPoint calculates the hash value of a data point with fnv64a.
-// https://en.wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function
-func hashDataPoint(dp *measurev1.DataPoint) uint64 {
- h := uint64(offset64)
- h = (h ^ dp.Sid) * prime64
- h = (h ^ uint64(dp.Timestamp.Seconds)) * prime64
- h = (h ^ uint64(dp.Timestamp.Nanos)) * prime64
- return h
-}
diff --git a/pkg/query/logical/measure/sorted_miterator_test.go b/pkg/query/logical/measure/sorted_miterator_test.go
deleted file mode 100644
index ef98e8283..000000000
--- a/pkg/query/logical/measure/sorted_miterator_test.go
+++ /dev/null
@@ -1,266 +0,0 @@
-// 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 (
- "slices"
- "testing"
-
- "github.com/google/go-cmp/cmp"
- "google.golang.org/protobuf/testing/protocmp"
- timestamppb "google.golang.org/protobuf/types/known/timestamppb"
-
- measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
-)
-
-type mockIterator struct {
- data []*comparableDataPoint
- idx int
-}
-
-func (m *mockIterator) Next() bool {
- m.idx++
- return m.idx < len(m.data)
-}
-
-func (m *mockIterator) Val() *comparableDataPoint {
- return m.data[m.idx]
-}
-
-func (m *mockIterator) Close() error {
- return nil
-}
-
-func makeComparableDP(sid uint64, seconds, nanos int64, version int64, sortField byte) *comparableDataPoint {
- return &comparableDataPoint{
- InternalDataPoint: &measurev1.InternalDataPoint{
- DataPoint: &measurev1.DataPoint{
- Sid: sid,
- Timestamp: ×tamppb.Timestamp{Seconds: seconds, Nanos: int32(nanos)},
- Version: version,
- },
- },
- sortField: []byte{sortField},
- }
-}
-
-func TestSortedMIterator_IndexModeDedup(t *testing.T) {
- // Cross-node duplicates for an IndexMode measure carry the same Sid but
- // different per-node "last-write" timestamps. They live in different
- // sort-field groups when sorted by time, so hashDataPoint (which keys on
- // Sid+timestamp) cannot collapse them — only Sid-based cross-group dedup
- // can. See the package-level comment on sortedMIterator.loadDps.
- testCases := []struct {
- name string
- data []*comparableDataPoint
- wantSids []uint64
- indexMode bool
- }{
- {
- // Sort-by-time, two nodes return the same Sid at different ts.
- // IndexMode=true → only first-seen Sid survives.
- name: "indexMode collapses same Sid across sort-field groups",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 0, 1, 1),
- makeComparableDP(1, 2, 0, 1, 2),
- },
- indexMode: true,
- wantSids: []uint64{1},
- },
- {
- // Same input, IndexMode=false → existing behavior preserved.
- name: "row-path keeps Sid duplicates with different timestamps",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 0, 1, 1),
- makeComparableDP(1, 2, 0, 1, 2),
- },
- indexMode: false,
- wantSids: []uint64{1, 1},
- },
- {
- // Sort-by-tag (same sort field) — both modes already dedup via
- // hashDataPoint here. Verify IndexMode does not regress that.
- name: "indexMode collapses same Sid within one sort-field group",
- data: []*comparableDataPoint{
- makeComparableDP(2, 1, 0, 1, 5),
- makeComparableDP(2, 2, 0, 1, 5),
- },
- indexMode: true,
- wantSids: []uint64{2},
- },
- {
- // Three Sids interleaved across two sort-field groups.
- // IndexMode=true keeps one per Sid (first-seen).
- name: "indexMode preserves distinct Sids across groups",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 0, 1, 1),
- makeComparableDP(2, 1, 0, 1, 1),
- makeComparableDP(1, 2, 0, 1, 2),
- makeComparableDP(3, 2, 0, 1, 2),
- },
- indexMode: true,
- wantSids: []uint64{1, 2, 3},
- },
- {
- // Group at sf=2 contains only already-seen Sids — loadDps's outer
- // loop must advance past the fully-filtered group to reach sf=3
- // instead of stopping iteration. Without that, `Next` returns
- // false after the first emission and Sid=2 is lost.
- name: "indexMode advances past a fully-filtered intermediate group",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 0, 1, 1),
- makeComparableDP(1, 2, 0, 1, 2),
- makeComparableDP(2, 3, 0, 1, 3),
- },
- indexMode: true,
- wantSids: []uint64{1, 2},
- },
- }
- for _, tc := range testCases {
- t.Run(tc.name, func(t *testing.T) {
- iter := &sortedMIterator{
- Iterator: &mockIterator{data: tc.data, idx: -1},
- indexMode: tc.indexMode,
- }
- iter.init()
- gotSids := make([]uint64, 0, len(tc.wantSids))
- for iter.Next() {
- gotSids = append(gotSids, iter.Current()[0].GetDataPoint().GetSid())
- }
- slices.Sort(gotSids)
- wantSids := append([]uint64(nil), tc.wantSids...)
- slices.Sort(wantSids)
- if diff := cmp.Diff(wantSids, gotSids); diff != "" {
- t.Errorf("Sids mismatch (-want +got):\n%s", diff)
- }
- })
- }
-}
-
-func TestSortedMIterator(t *testing.T) {
- testCases := []struct {
- name string
- data []*comparableDataPoint
- want []*measurev1.DataPoint
- }{
- {
- name: "empty data",
- data: []*comparableDataPoint{},
- want: []*measurev1.DataPoint{},
- },
- {
- name: "all data points are the same",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 1, 1, 1),
- makeComparableDP(1, 1, 1, 1, 1),
- },
- want: []*measurev1.DataPoint{
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 1, Nanos: 1}, Version: 1},
- },
- },
- {
- name: "identical data points with different sort fields",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 1, 1, 1),
- makeComparableDP(1, 1, 1, 1, 2),
- },
- want: []*measurev1.DataPoint{
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 1, Nanos: 1}, Version: 1},
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 1, Nanos: 1}, Version: 1},
- },
- },
- {
- name: "different data points with different sort fields",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 1, 1, 1),
- makeComparableDP(1, 2, 2, 1, 2),
- },
- want: []*measurev1.DataPoint{
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 1, Nanos: 1}, Version: 1},
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 2, Nanos: 2}, Version: 1},
- },
- },
- {
- name: "identical data points with different versions",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 1, 1, 1),
- makeComparableDP(1, 2, 2, 1, 1),
- makeComparableDP(1, 1, 1, 2, 1),
- },
- want: []*measurev1.DataPoint{
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 1, Nanos: 1}, Version: 2},
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 2, Nanos: 2}, Version: 1},
- },
- },
- {
- name: "multiple sids with different versions",
- data: []*comparableDataPoint{
- makeComparableDP(1, 1, 1, 1, 1),
- makeComparableDP(2, 1, 1, 1, 1),
- makeComparableDP(1, 1, 1, 2, 1),
- makeComparableDP(1, 2, 2, 2, 2),
- makeComparableDP(1, 2, 2, 1, 2),
- makeComparableDP(2, 2, 2, 2, 2),
- makeComparableDP(1, 3, 3, 1, 3),
- makeComparableDP(1, 3, 3, 2, 3),
- makeComparableDP(3, 3, 3, 2, 3),
- },
- want: []*measurev1.DataPoint{
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 1, Nanos: 1}, Version: 2},
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 2, Nanos: 2}, Version: 2},
- {Sid: 1, Timestamp: ×tamppb.Timestamp{Seconds: 3, Nanos: 3}, Version: 2},
- {Sid: 2, Timestamp: ×tamppb.Timestamp{Seconds: 1, Nanos: 1}, Version: 1},
- {Sid: 2, Timestamp: ×tamppb.Timestamp{Seconds: 2, Nanos: 2}, Version: 2},
- {Sid: 3, Timestamp: ×tamppb.Timestamp{Seconds: 3, Nanos: 3}, Version: 2},
- },
- },
- }
-
- for _, tc := range testCases {
- t.Run(tc.name, func(t *testing.T) {
- iter := &sortedMIterator{
- Iterator: &mockIterator{data: tc.data, idx: -1},
- }
-
- iter.init()
-
- got := make([]*measurev1.DataPoint, 0)
- for iter.Next() {
- got = append(got, iter.Current()[0].GetDataPoint())
- }
-
- slices.SortFunc(got, func(a, b *measurev1.DataPoint) int {
- if diff := a.Sid - b.Sid; diff != 0 {
- return int(diff)
- }
- if diff := a.Timestamp.Seconds - b.Timestamp.Seconds; diff != 0 {
- return int(diff)
- }
- if diff := a.Timestamp.Nanos - b.Timestamp.Nanos; diff != 0 {
- return int(diff)
- }
- return int(a.Version - b.Version)
- })
-
- if diff := cmp.Diff(tc.want, got,
- protocmp.Transform(), protocmp.IgnoreUnknown()); diff != "" {
- t.Errorf("sortedMIterator mismatch (-want +got):\n%s", diff)
- }
- })
- }
-}
diff --git a/pkg/query/vectorized/measure/plan/distributed_rows.go b/pkg/query/vectorized/measure/plan/distributed_rows.go
index 6c5893579..2ce446cae 100644
--- a/pkg/query/vectorized/measure/plan/distributed_rows.go
+++ b/pkg/query/vectorized/measure/plan/distributed_rows.go
@@ -25,6 +25,8 @@ import (
"sort"
"strings"
+ "google.golang.org/protobuf/proto"
+
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
itersort "github.com/apache/skywalking-banyandb/pkg/iter/sort"
"github.com/apache/skywalking-banyandb/pkg/query/vectorized"
@@ -62,6 +64,8 @@ type seenSIDKey struct {
// column without going through TagIndex lookup. The merger prefers the
// explicit index when >= 0; otherwise it resolves from the names.
type distributedRowsSpec struct {
+ // Interrupt, when set, is polled while merging; a non-nil result aborts the merge with it.
+ Interrupt func() error
Tracker *vectorized.MemoryTracker
OrderByFamily string
OrderByTagName string
@@ -156,7 +160,7 @@ func newDistributedRowSourceIter(
if encodeErr != nil {
return nil, encodeErr
}
- sortKeys[rowIdx] = key
+ sortKeys[rowIdx] = markSortKey(key, isNullSortCell(batch.Columns[sortColIdx], rowIdx))
}
iter.sortKeys = sortKeys
// Per-source local sort by the OrderBy column's encoded key so the
@@ -207,9 +211,16 @@ func (s *distributedRowSourceIter) Next() bool {
}
rowIdx := s.indices[s.pos]
s.pos++
- sidCol := s.batch.Columns[s.sidIdx].(*vectorized.TypedColumn[int64])
tsCol := s.batch.Columns[s.tsIdx].(*vectorized.TypedColumn[int64])
- verCol := s.batch.Columns[s.verIdx].(*vectorized.TypedColumn[int64])
+ // A schema without series-id or version columns (a time-bucketed aggregate) has nothing to
+ // dedup on; the emitter keeps every such row.
+ var sid, ver int64
+ if s.sidIdx >= 0 {
+ sid = s.batch.Columns[s.sidIdx].(*vectorized.TypedColumn[int64]).Data()[rowIdx]
+ }
+ if s.verIdx >= 0 {
+ ver = s.batch.Columns[s.verIdx].(*vectorized.TypedColumn[int64]).Data()[rowIdx]
+ }
var sortField []byte
if s.sortColIdx >= 0 {
// Phase 2: reuse the cached OrderBy sort key. encodeSortKey was
@@ -227,9 +238,9 @@ func (s *distributedRowSourceIter) Next() bool {
source: s.source,
group: s.group,
seq: s.seq,
- sid: sidCol.Data()[rowIdx],
+ sid: sid,
ts: tsCol.Data()[rowIdx],
- ver: verCol.Data()[rowIdx],
+ ver: ver,
sortField: sortField,
}
s.seq++
@@ -313,6 +324,7 @@ func runDistributedRowMerge(
indexMode: spec.IndexMode,
borrowOutput: spec.BorrowOutput,
window: make(map[distributedRowKey]*distributedRowItem),
+ noDedup: schema.SeriesIDIndex() < 0 || schema.VersionIndex() < 0,
}
defer func() {
if retErr != nil {
@@ -324,15 +336,23 @@ func runDistributedRowMerge(
}
}()
- for merger.Next() {
+ for rows := 0; merger.Next(); rows++ {
+ if spec.Interrupt != nil && rows%interruptCheckRows == 0 {
+ if interruptErr := spec.Interrupt(); interruptErr != nil {
+ return nil, interruptErr
+ }
+ }
item := merger.Val()
- if emitter.windowKey != nil && !bytes.Equal(item.sortField, emitter.windowKey) {
+ // A null sort cell encodes as a nil key, so "no window open" is tracked explicitly: a
+ // nil windowKey alone would merge a null row's window into the next non-null row's.
+ if emitter.windowOpen && !bytes.Equal(item.sortField, emitter.windowKey) {
if err := emitter.flushWindow(); err != nil {
return nil, err
}
}
- if emitter.windowKey == nil {
+ if !emitter.windowOpen {
emitter.windowKey = item.sortField
+ emitter.windowOpen = true
}
emitter.accept(item)
}
@@ -345,6 +365,91 @@ func runDistributedRowMerge(
return emitter.output, nil
}
+// interruptCheckRows is how many merged rows pass between checks of distributedRowsSpec.Interrupt.
+const interruptCheckRows = 1024
+
+// markSortKey prefixes an encoded index sort key with a null marker so a null cell and an empty
+// value stay distinct under the heap's byte comparison: null sorts first. Nullness is passed in
+// rather than inferred from the key, because some encoders (pbv1.MarshalTagValue) return a nil
+// key for a non-null empty string or empty array too.
+func markSortKey(key []byte, isNull bool) []byte {
+ if isNull {
+ return []byte{0}
+ }
+ out := make([]byte, 1+len(key))
+ out[0] = 1
+ copy(out[1:], key)
+ return out
+}
+
+// isNullSortCell reports whether the sort cell holds no value: a null in the column's validity,
+// or a passthrough TagValue that is absent or the null value.
+func isNullSortCell(col vectorized.Column, rowIdx int) bool {
+ if col.IsNull(rowIdx) {
+ return true
+ }
+ if tvCol, ok := col.(*vectorized.TypedColumn[*modelv1.TagValue]); ok {
+ tv := tvCol.Data()[rowIdx]
+ if tv == nil {
+ return true
+ }
+ _, isNullValue := tv.GetValue().(*modelv1.TagValue_Null)
+ return isNullValue
+ }
+ return false
+}
+
+// columnBytes estimates the memory a column's cells hold, counting the variable-size payloads
+// that a fixed per-row width misses (strings, bytes, arrays, passthrough protobuf values).
+func columnBytes(col vectorized.Column) int64 {
+ var total int64
+ switch c := col.(type) {
+ case *vectorized.TypedColumn[int64]:
+ total = int64(len(c.Data())) * 8
+ case *vectorized.TypedColumn[float64]:
+ total = int64(len(c.Data())) * 8
+ case *vectorized.TypedColumn[string]:
+ for _, v := range c.Data() {
+ total += 16 + int64(len(v))
+ }
+ case *vectorized.TypedColumn[[]byte]:
+ for _, v := range c.Data() {
+ total += 24 + int64(len(v))
+ }
+ case *vectorized.TypedColumn[[]int64]:
+ for _, v := range c.Data() {
+ total += 24 + int64(len(v))*8
+ }
+ case *vectorized.TypedColumn[[]string]:
+ for _, v := range c.Data() {
+ total += 24
+ for _, e := range v {
+ total += 16 + int64(len(e))
+ }
+ }
+ case *vectorized.TypedColumn[*modelv1.TagValue]:
+ for _, v := range c.Data() {
+ total += 16 + int64(proto.Size(v))
+ }
+ case *vectorized.TypedColumn[*modelv1.FieldValue]:
+ for _, v := range c.Data() {
+ total += 16 + int64(proto.Size(v))
+ }
+ default:
+ total = int64(col.Len()) * 16
+ }
+ return total
+}
+
+// batchBytes estimates the memory a batch's columns hold.
+func batchBytes(batch *vectorized.RecordBatch) int64 {
+ var total int64
+ for _, col := range batch.Columns {
+ total += columnBytes(col)
+ }
+ return total
+}
+
// releaseSources drops the references the decoded source batches hold on their frames.
func releaseSources(sources []decodedBatchSource) {
for _, src := range sources {
@@ -422,9 +527,11 @@ type distributedRowEmitter struct {
output []*vectorized.RecordBatch
batchSize int
rowWidth int64
+ pendingStart, pendingRows int
indexMode bool
borrowOutput bool
- pendingStart, pendingRows int
+ windowOpen bool
+ noDedup bool
}
// accept adds an incoming row to the current sort-group dedup map. The window
@@ -433,6 +540,10 @@ type distributedRowEmitter struct {
// row-path baseline.
func (e *distributedRowEmitter) accept(item *distributedRowItem) {
key := distributedRowKey{sid: item.sid, ts: item.ts}
+ if e.noDedup {
+ // Unique per row: without series-id and version there is no (sid, ts) identity.
+ key = distributedRowKey{sid: int64(item.source), ts: int64(item.seq)}
+ }
existing, ok := e.window[key]
if !ok || item.ver > existing.ver {
e.window[key] = item
@@ -466,7 +577,7 @@ func (e *distributedRowEmitter) flushWindow() error {
return emit[i].source < emit[j].source
})
for _, row := range emit {
- if e.indexMode {
+ if e.indexMode && !e.noDedup {
sidKey := seenSIDKey{group: row.group, sid: row.sid}
if _, dup := e.seenSID[sidKey]; dup {
continue
@@ -479,6 +590,7 @@ func (e *distributedRowEmitter) flushWindow() error {
}
clear(e.window)
e.windowKey = nil
+ e.windowOpen = false
return nil
}
@@ -711,7 +823,7 @@ func appendColumnRangeCoerced(dst, src vectorized.Column, srcPos, n int) error {
case *vectorized.TypedColumn[[]int64]:
tv = &modelv1.TagValue{Value: &modelv1.TagValue_IntArray{IntArray: &modelv1.IntArray{Value: slices.Clone(sc.Data()[rowIdx])}}}
case *vectorized.TypedColumn[[]string]:
- tv = &modelv1.TagValue{Value: &modelv1.TagValue_StrArray{StrArray: &modelv1.StrArray{Value: slices.Clone(sc.Data()[rowIdx])}}}
+ tv = &modelv1.TagValue{Value: &modelv1.TagValue_StrArray{StrArray: &modelv1.StrArray{Value: cloneStrings(sc.Data()[rowIdx])}}}
default:
return fmt.Errorf("appendColumnRangeCoerced: cannot coerce %s to tagvalue", src.Type())
}
@@ -754,6 +866,16 @@ func appendColumnRangeCoerced(dst, src vectorized.Column, srcPos, n int) error {
return measure.AppendColumnRange(dst, src, srcPos, n)
}
+// cloneStrings copies the slice and every element's bytes: a cell may borrow a received frame
+// or a storage block, and the passthrough value travels on to the response.
+func cloneStrings(values []string) []string {
+ out := make([]string, len(values))
+ for i, v := range values {
+ out[i] = strings.Clone(v)
+ }
+ return out
+}
+
// decodeMultiGroupSources decodes each groupFrame body and reshapes the
// resulting batch into the merged schema by null-filling columns that exist
// in the merged schema but are absent from the source group's frame schema.
@@ -779,40 +901,12 @@ func decodeMultiGroupSources(groupFrames []groupFrame, mergedSchema *vectorized.
}
continue
}
- // Build a column-index map from the source schema so we can look up
- // which source column corresponds to each merged-schema destination
- // column by (role, family, name) key.
- srcColMap := make(map[string]int, len(rawBatch.Schema.Columns))
- for ci, cd := range rawBatch.Schema.Columns {
- srcColMap[schemaColKey(cd)] = ci
- }
- // Reshape the source batch into the merged schema. Columns present in
- // the source are copied; columns absent from the source are null-filled.
- reshapedBatch := vectorized.NewRecordBatch(mergedSchema, rawBatch.Len)
- reshapedBatch.Len = rawBatch.Len
- reshapedBatch.Selection = rawBatch.Selection
- for destColIdx, destDef := range mergedSchema.Columns {
- destKey := schemaColKey(destDef)
- srcColIdx, hasSrc := srcColMap[destKey]
- if hasSrc {
- if copyErr := appendColumnRangeCoerced(
- reshapedBatch.Columns[destColIdx],
- rawBatch.Columns[srcColIdx],
- 0, rawBatch.Len,
- ); copyErr != nil {
- rawBatch.ReleaseOwners()
- releaseSources(sources)
- return nil, fmt.Errorf("reshape group %d frame %d col %d: %w", gf.group, frameIdx, destColIdx, copyErr)
- }
- } else {
- for range rawBatch.Len {
- reshapedBatch.Columns[destColIdx].AppendNull()
- }
- }
+ reshapedBatch, reshapeErr := reshapeToMergedSchema(rawBatch, mergedSchema)
+ if reshapeErr != nil {
+ rawBatch.ReleaseOwners()
+ releaseSources(sources)
+ return nil, fmt.Errorf("reshape group %d frame %d: %w", gf.group, frameIdx, reshapeErr)
}
- // The reshaped batch shares the raw batch's cells, so it takes over the raw batch's
- // references on the frame.
- rawBatch.Owners.TransferTo(&reshapedBatch.Owners)
sources = append(sources, decodedBatchSource{
batch: reshapedBatch,
source: frameIdx,
@@ -822,6 +916,42 @@ func decodeMultiGroupSources(groupFrames []groupFrame, mergedSchema *vectorized.
return sources, nil
}
+// reshapeToMergedSchema reshapes rawBatch into mergedSchema: columns present in
+// the source (matched by role, family and name) are copied, coerced to the merged
+// column type, and columns absent from the source are null-filled. The reshaped
+// batch shares rawBatch's cells, so it takes over rawBatch's references on the
+// memory they borrow. On error rawBatch is left untouched for the caller to release.
+func reshapeToMergedSchema(rawBatch *vectorized.RecordBatch, mergedSchema *vectorized.BatchSchema) (*vectorized.RecordBatch, error) {
+ // Build a column-index map from the source schema so we can look up
+ // which source column corresponds to each merged-schema destination
+ // column by (role, family, name) key.
+ srcColMap := make(map[string]int, len(rawBatch.Schema.Columns))
+ for ci, cd := range rawBatch.Schema.Columns {
+ srcColMap[schemaColKey(cd)] = ci
+ }
+ reshapedBatch := vectorized.NewRecordBatch(mergedSchema, rawBatch.Len)
+ reshapedBatch.Len = rawBatch.Len
+ reshapedBatch.Selection = rawBatch.Selection
+ for destColIdx, destDef := range mergedSchema.Columns {
+ srcColIdx, hasSrc := srcColMap[schemaColKey(destDef)]
+ if !hasSrc {
+ for range rawBatch.Len {
+ reshapedBatch.Columns[destColIdx].AppendNull()
+ }
+ continue
+ }
+ if copyErr := appendColumnRangeCoerced(
+ reshapedBatch.Columns[destColIdx],
+ rawBatch.Columns[srcColIdx],
+ 0, rawBatch.Len,
+ ); copyErr != nil {
+ return nil, fmt.Errorf("col %d: %w", destColIdx, copyErr)
+ }
+ }
+ rawBatch.Owners.TransferTo(&reshapedBatch.Owners)
+ return reshapedBatch, nil
+}
+
// schemaColKey returns a string key that uniquely identifies a column within a
// BatchSchema for the multi-group reshape map in decodeMultiGroupSources. The
// key encodes role, tag-family, and name with two separator bytes (0x01, 0x02)
@@ -842,19 +972,27 @@ func mergeDistributedRowsMulti(groupFrames []groupFrame, mergedSchema *vectorize
if mergedSchema.SeriesIDIndex() < 0 || mergedSchema.TimestampIndex() < 0 || mergedSchema.VersionIndex() < 0 {
return nil, fmt.Errorf("row merge requires series-id, timestamp, and version columns")
}
- batchSize := spec.BatchSize
- if batchSize <= 0 {
- batchSize = vectorized.DefaultBatchSize
- }
sources, decodeErr := decodeMultiGroupSources(groupFrames, mergedSchema)
if decodeErr != nil {
return nil, decodeErr
}
- // The output batches pin whatever the sources borrow (see appendRowToCurrent).
+ return mergeBatchSources(sources, mergedSchema, spec)
+}
+
+// mergeBatchSources runs the k-way heap merger over sources whose batches are
+// already shaped as mergedSchema, whether they were decoded from frames or come
+// straight from a local plan. It takes ownership of sources: the output batches
+// pin whatever the sources borrow (see appendRowToCurrent), so the sources' own
+// references are released when the merge returns.
+func mergeBatchSources(sources []decodedBatchSource, mergedSchema *vectorized.BatchSchema, spec distributedRowsSpec) ([]*vectorized.RecordBatch, error) {
defer releaseSources(sources)
if len(sources) == 0 {
return nil, nil
}
+ batchSize := spec.BatchSize
+ if batchSize <= 0 {
+ batchSize = vectorized.DefaultBatchSize
+ }
sortColIdx, sortErr := resolveDistributedSortColumn(mergedSchema, spec)
if sortErr != nil {
return nil, sortErr
diff --git a/pkg/query/vectorized/measure/plan/distributed_rows_test.go b/pkg/query/vectorized/measure/plan/distributed_rows_test.go
index 363203a2d..98c15cf56 100644
--- a/pkg/query/vectorized/measure/plan/distributed_rows_test.go
+++ b/pkg/query/vectorized/measure/plan/distributed_rows_test.go
@@ -24,6 +24,7 @@ import (
"sync"
"testing"
"time"
+ "unsafe"
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
@@ -1079,3 +1080,25 @@ func TestApplyBatchGroupByFirstToRows_OverUint16Boundary_KeepsEveryGroup(t *test
t.Fatalf("GroupByFirst: commonB first-seen should be row 1 (value=1): %+v", r)
}
}
+
+// TestAppendColumnRangeCoerced_StringArrayDoesNotBorrowElements pins that promoting a native
+// string-array cell to a passthrough TagValue copies every element's bytes: the source cell may
+// borrow storage memory that is released before the response is serialized.
+func TestAppendColumnRangeCoerced_StringArrayDoesNotBorrowElements(t *testing.T) {
+ src := vectorized.NewStrArrayColumn(1)
+ borrowed := []string{strings.Clone("alpha"), strings.Clone("beta")}
+ src.Append(borrowed)
+ dst := vectorized.NewTagValueColumn(1)
+ if err := appendColumnRangeCoerced(dst, src, 0, 1); err != nil {
+ t.Fatalf("appendColumnRangeCoerced: %v", err)
+ }
+ got := dst.Data()[0].GetStrArray().GetValue()
+ if len(got) != 2 || got[0] != "alpha" || got[1] != "beta" {
+ t.Fatalf("coerced array = %v, want [alpha beta]", got)
+ }
+ for i := range got {
+ if unsafe.StringData(got[i]) == unsafe.StringData(borrowed[i]) {
+ t.Fatalf("element %d still shares its bytes with the source cell", i)
+ }
+ }
+}
diff --git a/pkg/query/vectorized/measure/plan/multi_group_merge.go b/pkg/query/vectorized/measure/plan/multi_group_merge.go
new file mode 100644
index 000000000..067bb3cad
--- /dev/null
+++ b/pkg/query/vectorized/measure/plan/multi_group_merge.go
@@ -0,0 +1,444 @@
+// 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 plan
+
+import (
+ "context"
+ "fmt"
+ "math"
+
+ "google.golang.org/protobuf/proto"
+
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized"
+ vmeasure "github.com/apache/skywalking-banyandb/pkg/query/vectorized/measure"
+)
+
+// MultiGroupMerge merges the per-group batch results of a local multi-group
+// measure query. It derives the per-group request, so every group returns enough
+// rows and the order-by column, runs the shared k-way row merger over the groups'
+// batches, and exposes the merged batches before final row serialization.
+//
+// A result that keeps the series-id, timestamp and version columns (plain rows,
+// Top and raw GroupBy) is merged by sort key with version dedup. A time-bucketed
+// aggregate keeps only the timestamp, so it is merged by time without dedup. Any
+// other Agg output has no ordering column, so its per-group results are concatenated
+// in group order, as each group's plan already applied its own limit; reducing them
+// across groups is #14069.
+//
+// Every batch the groups hand over, the reshaped copies and the merged output are
+// charged to one memory tracker sized by the query memory budget.
+type MultiGroupMerge struct {
+ groupReq *measurev1.QueryRequest
+ orderBy *resolvedOrderByTag
+ hidden logical.HiddenTagSet
+ tracker *vectorized.MemoryTracker
+ cfg vmeasure.VectorizedConfig
+ offset uint32
+ limit uint32
+ desc bool
+ indexMode bool
+ plainRows bool
+}
+
+// NewMultiGroupMerge resolves how the groups of req are merged. schemas and
+// measureSchemas are the per-group logical and measure schemas in request order.
+func NewMultiGroupMerge(
+ ctx context.Context,
+ req *measurev1.QueryRequest,
+ schemas []logical.Schema,
+ measureSchemas []*databasev1.Measure,
+ cfg vmeasure.VectorizedConfig,
+) (*MultiGroupMerge, error) {
+ merge := &MultiGroupMerge{
+ groupReq: req,
+ cfg: cfg,
+ hidden: logical.NewHiddenTagSet(),
+ tracker: vectorized.QueryMemoryTracker(ctx, int64(cfg.QueryMemoryMiB)*1024*1024),
+ plainRows: req.GetGroupBy() == nil && req.GetAgg() == nil && req.GetTop() == nil,
+ offset: req.GetOffset(),
+ limit: req.GetLimit(),
+ desc: req.GetOrderBy().GetSort() == modelv1.Sort_SORT_DESC,
+ }
+ if req.GetAgg() != nil {
+ return merge, nil
+ }
+ merge.indexMode = len(measureSchemas) > 0
+ for _, ms := range measureSchemas {
+ if !ms.GetIndexMode() {
+ merge.indexMode = false
+ break
+ }
+ }
+ groupReq := proto.Clone(req).(*measurev1.QueryRequest)
+ merge.groupReq = groupReq
+ if merge.plainRows {
+ if merge.limit == 0 {
+ merge.limit = defaultLimit
+ }
+ // A group can only return the first offset+limit rows of the merged order, and the
+ // offset is applied once, after the merge.
+ groupReq.Offset = 0
+ groupReq.Limit = uint32(min(uint64(merge.offset)+uint64(merge.limit), math.MaxUint32))
+ }
+ if indexRule := req.GetOrderBy().GetIndexRuleName(); indexRule != "" {
+ resolved, resolveErr := resolveMultiGroupOrderBy(schemas, measureSchemas, indexRule)
+ if resolveErr != nil {
+ return nil, resolveErr
+ }
+ merge.orderBy = &resolved
+ if !orderByProjectionVisible(req.GetTagProjection(), resolved) {
+ // The merge sorts on this column, so every group must project it; it is
+ // stripped again after the merge.
+ groupReq.TagProjection = appendOrderByToProjection(groupReq.GetTagProjection(), resolved)
+ merge.hidden.Add(resolved.tag)
+ }
+ }
+ return merge, nil
+}
+
+// resolveMultiGroupOrderBy finds the tag the index rule sorts on in the first
+// group that defines the rule.
+func resolveMultiGroupOrderBy(schemas []logical.Schema, measureSchemas []*databasev1.Measure, indexRuleName string) (resolvedOrderByTag, error) {
+ for groupIdx, schema := range schemas {
+ if schema == nil || groupIdx >= len(measureSchemas) {
+ continue
+ }
+ defined, rule := schema.IndexRuleDefined(indexRuleName)
+ if !defined {
+ continue
+ }
+ return resolveOrderByTag(measureSchemas[groupIdx], []*databasev1.IndexRule{rule}, indexRuleName)
+ }
+ return resolvedOrderByTag{}, fmt.Errorf("index rule %s not found", indexRuleName)
+}
+
+// GroupRequest returns the request every group is dispatched with.
+func (m *MultiGroupMerge) GroupRequest() *measurev1.QueryRequest { return m.groupReq }
+
+// Tracker returns the shared query memory tracker used by collection and merging.
+func (m *MultiGroupMerge) Tracker() *vectorized.MemoryTracker { return m.tracker }
+
+// Merge returns a native iterator over merged batches, transferring Close ownership.
+func (m *MultiGroupMerge) Merge(ctx context.Context, groups []*GroupBatches) (executor.MIterator, error) {
+ result, err := m.MergeBatches(ctx, groups)
+ if err != nil {
+ return nil, err
+ }
+ return result.NativeIterator(ctx)
+}
+
+// MergeBatches takes ownership of groups and returns their merged native batches.
+// Groups and reservations are released on failure or when the result is closed.
+func (m *MultiGroupMerge) MergeBatches(ctx context.Context, groups []*GroupBatches) (*BatchResult, error) {
+ var reserved int64
+ fail := func(err error) (*BatchResult, error) {
+ m.tracker.Release(reserved)
+ for _, group := range groups {
+ group.Release()
+ }
+ return nil, err
+ }
+ var schemas []*vectorized.BatchSchema
+ hidden := logical.NewHiddenTagSet()
+ for tag := range m.hidden {
+ hidden.Add(tag)
+ }
+ for _, group := range groups {
+ if group.Schema != nil {
+ schemas = append(schemas, group.Schema)
+ }
+ for tag := range group.Hidden {
+ hidden.Add(tag)
+ }
+ }
+ if len(schemas) == 0 {
+ for _, group := range groups {
+ group.Release()
+ }
+ return NewBatchResult(ctx, &batchSliceSource{schema: vectorized.NewBatchSchema(nil)}, vectorized.NewBatchSchema(nil), m.cfg.BatchSize)
+ }
+ merged, unionErr := unionGroupSchemas(schemas)
+ if unionErr != nil {
+ return fail(unionErr)
+ }
+ sources, sourceErr := m.reshapeGroupBatches(ctx, groups, merged, &reserved)
+ if sourceErr != nil {
+ return fail(sourceErr)
+ }
+ offset, limit := uint32(0), uint32(math.MaxUint32)
+ var output []*vectorized.RecordBatch
+ if merged.TimestampIndex() >= 0 {
+ if m.plainRows {
+ offset, limit = m.offset, m.limit
+ }
+ spec := distributedRowsSpec{
+ Desc: m.desc,
+ IndexMode: m.indexMode,
+ BatchSize: m.cfg.BatchSize,
+ OrderByColIdx: -1,
+ Interrupt: ctx.Err,
+ Tracker: m.tracker,
+ }
+ if m.orderBy != nil {
+ spec.OrderByFamily = m.orderBy.family
+ spec.OrderByTagName = m.orderBy.tag
+ }
+ // The merged output can be as large as the reshaped input; reserve that bound while both
+ // exist so the peak stays inside the budget.
+ reshaped := reserved
+ if reserveErr := m.tracker.Reserve(reshaped); reserveErr != nil {
+ releaseSources(sources)
+ return fail(fmt.Errorf("vec multi-group merge: %w", reserveErr))
+ }
+ reserved += reshaped
+ // Index-order merging copies every row's sort key.
+ keyBytes := m.sortKeyBytes(sources, merged)
+ if reserveErr := m.tracker.Reserve(keyBytes); reserveErr != nil {
+ releaseSources(sources)
+ return fail(fmt.Errorf("vec multi-group merge: %w", reserveErr))
+ }
+ reserved += keyBytes
+ // Keep the shared merger's transient batch guard in addition to retained output.
+ var mergeErr error
+ output, mergeErr = mergeBatchSources(sources, merged, spec)
+ if mergeErr != nil {
+ return fail(fmt.Errorf("vec multi-group merge: %w", mergeErr))
+ }
+ // The reshaped batches are released; the merged output keeps what it actually holds.
+ var outBytes int64
+ for _, batch := range output {
+ outBytes += batchBytes(batch)
+ }
+ outBytes = min(outBytes, reshaped)
+ m.tracker.Release(reserved - outBytes)
+ reserved = outBytes
+ } else {
+ for _, src := range sources {
+ output = append(output, src.batch)
+ }
+ }
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ releaseBatches(output)
+ return fail(fmt.Errorf("vec multi-group merge: %w", ctxErr))
+ }
+ result, resultErr := m.resultFromBatches(ctx, output, merged, offset, limit, groups, reserved)
+ if resultErr != nil {
+ return nil, resultErr
+ }
+ if !hidden.IsEmpty() {
+ projected := newHiddenBatchProjection(result, hidden)
+ return NewBatchResult(ctx, projected, projected.OutputSchema(), m.cfg.BatchSize)
+ }
+ return result, nil
+}
+
+// sortKeyBytes estimates the sort keys the merge allocates: the order-by column's payload plus a
+// slice header and marker per row. Time ordering allocates a fixed 8 bytes per emitted row, which
+// the output reservation covers.
+func (m *MultiGroupMerge) sortKeyBytes(sources []decodedBatchSource, merged *vectorized.BatchSchema) int64 {
+ if m.orderBy == nil {
+ return 0
+ }
+ colIdx, ok := merged.TagIndex(m.orderBy.family, m.orderBy.tag)
+ if !ok {
+ return 0
+ }
+ var total int64
+ for _, src := range sources {
+ total += columnBytes(src.batch.Columns[colIdx]) + int64(src.batch.Len)*25
+ }
+ return total
+}
+
+// resultFromBatches applies pagination while keeping the result in batches.
+func (m *MultiGroupMerge) resultFromBatches(ctx context.Context, batches []*vectorized.RecordBatch,
+ schema *vectorized.BatchSchema, offset, limit uint32, groups []*GroupBatches, reserved int64,
+) (*BatchResult, error) {
+ source := &groupReleasingSource{
+ PullOperator: &batchSliceSource{batches: batches, schema: schema}, groups: groups, tracker: m.tracker, reserved: reserved,
+ }
+ builder := vectorized.NewPipelineBuilder().WithMemoryTracker(m.tracker).From(source)
+ builder.Apply(vmeasure.NewBatchLimit(schema, offset, limit))
+ pipeline, err := builder.Build()
+ if err != nil {
+ _ = source.Close()
+ return nil, fmt.Errorf("vec multi-group merge: build output pipeline: %w", err)
+ }
+ if err = pipeline.Init(ctx); err != nil {
+ _ = pipeline.Close()
+ return nil, fmt.Errorf("vec multi-group merge: init output pipeline: %w", err)
+ }
+ return NewBatchResult(ctx, &pipelineBatchSource{pipeline: pipeline, schema: schema}, schema, m.cfg.BatchSize)
+}
+
+// reshapeGroupBatches aligns every group's batches to the merged schema, null-filling
+// the columns a group lacks, and charges each reshaped batch to the tracker (added to
+// *reserved). Sources are numbered in group order so rows that tie on the sort key keep
+// a deterministic order. On error the reshaped batches are released and the groups are
+// left for the caller.
+func (m *MultiGroupMerge) reshapeGroupBatches(
+ ctx context.Context,
+ groups []*GroupBatches,
+ merged *vectorized.BatchSchema,
+ reserved *int64,
+) ([]decodedBatchSource, error) {
+ var sources []decodedBatchSource
+ for groupIdx, group := range groups {
+ for batchIdx, raw := range group.Batches {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ releaseSources(sources)
+ return nil, fmt.Errorf("vec multi-group merge: %w", ctxErr)
+ }
+ if raw == nil || raw.ActiveLen() == 0 {
+ continue
+ }
+ // Cells are shared or copied by value, and columns the group lacks are null-filled.
+ bytes := batchBytes(raw) + int64(raw.Len)*16*int64(max(0, len(merged.Columns)-len(raw.Schema.Columns)))
+ if reserveErr := m.tracker.Reserve(bytes); reserveErr != nil {
+ releaseSources(sources)
+ return nil, fmt.Errorf("vec multi-group merge: %w", reserveErr)
+ }
+ *reserved += bytes
+ reshaped, reshapeErr := reshapeToMergedSchema(raw, merged)
+ if reshapeErr != nil {
+ releaseSources(sources)
+ return nil, fmt.Errorf("vec multi-group merge: reshape group %d batch %d: %w", groupIdx, batchIdx, reshapeErr)
+ }
+ sources = append(sources, decodedBatchSource{batch: reshaped, source: len(sources), group: groupIdx})
+ }
+ }
+ return sources, nil
+}
+
+// unionGroupSchemas builds the merged schema of the groups' output schemas: the
+// columns in first-seen order, keyed by role, family and name. A tag or field whose
+// type differs between groups is promoted to the passthrough column type.
+func unionGroupSchemas(schemas []*vectorized.BatchSchema) (*vectorized.BatchSchema, error) {
+ var cols []vectorized.ColumnDef
+ index := make(map[string]int)
+ for _, schema := range schemas {
+ for _, def := range schema.Columns {
+ key := schemaColKey(def)
+ colIdx, seen := index[key]
+ if !seen {
+ index[key] = len(cols)
+ cols = append(cols, def)
+ continue
+ }
+ if cols[colIdx].Type == def.Type {
+ continue
+ }
+ switch def.Role {
+ case vectorized.RoleTag:
+ cols[colIdx].Type = vectorized.ColumnTypeTagValue
+ case vectorized.RoleField:
+ cols[colIdx].Type = vectorized.ColumnTypeFieldValue
+ default:
+ return nil, fmt.Errorf("vec multi-group merge: column %q type diverges across groups on role %v: %v vs %v",
+ def.Name, def.Role, cols[colIdx].Type, def.Type)
+ }
+ }
+ }
+ return vectorized.NewBatchSchema(cols), nil
+}
+
+// groupReleasingSource keeps storage owners alive through final batch consumption.
+type groupReleasingSource struct {
+ vectorized.PullOperator
+ tracker *vectorized.MemoryTracker
+ groups []*GroupBatches
+ reserved int64
+}
+
+func (g *groupReleasingSource) Close() error {
+ err := g.PullOperator.Close()
+ for _, group := range g.groups {
+ group.Release()
+ }
+ g.groups = nil
+ g.tracker.Release(g.reserved)
+ g.reserved = 0
+ return err
+}
+
+// GroupBatches owns copies of one group's borrowed batch descriptors and columns.
+// The native result stays open until Release to retain any borrowed cell payloads.
+type GroupBatches struct {
+ tracker *vectorized.MemoryTracker
+ closer *BatchResult
+ Schema *vectorized.BatchSchema
+ Hidden logical.HiddenTagSet
+ PlanString string
+ Batches []*vectorized.RecordBatch
+ reserved int64
+}
+
+// Release closes the native group and releases copied output and reservations.
+func (g *GroupBatches) Release() {
+ if g == nil {
+ return
+ }
+ releaseBatches(g.Batches)
+ g.Batches = nil
+ if g.closer != nil {
+ _ = g.closer.Close()
+ g.closer = nil
+ }
+ if g.tracker != nil {
+ g.tracker.Release(g.reserved)
+ g.reserved = 0
+ }
+}
+
+// Collect takes ownership of a native group result without materializing rows.
+func (m *MultiGroupMerge) Collect(ctx context.Context, result *BatchResult, planStr string) (*GroupBatches, error) {
+ group := &GroupBatches{closer: result, tracker: m.tracker, Schema: result.OutputSchema(), PlanString: planStr}
+ for {
+ raw, err := result.NextBatch(ctx)
+ if err != nil {
+ group.Release()
+ return nil, fmt.Errorf("vec dispatch: read batches: %w", err)
+ }
+ if raw == nil {
+ return group, nil
+ }
+ bytes := batchBytes(raw)
+ if err = m.tracker.Reserve(bytes); err != nil {
+ group.Release()
+ return nil, fmt.Errorf("vec dispatch: read batches: %w", err)
+ }
+ group.reserved += bytes
+ // NextBatch lends descriptors and columns only until the next pull. Copy both
+ // before advancing and transfer the owners that pin borrowed cell payloads.
+ copyBatch, err := reshapeToMergedSchema(raw, raw.Schema)
+ if err != nil {
+ group.Release()
+ return nil, err
+ }
+ copyBatch.Selection = append([]uint16(nil), raw.Selection...)
+ if raw.Selection != nil && len(raw.Selection) == 0 {
+ copyBatch.Selection = []uint16{}
+ }
+ group.Batches = append(group.Batches, copyBatch)
+ }
+}
diff --git a/pkg/query/vectorized/measure/plan/multi_group_merge_test.go b/pkg/query/vectorized/measure/plan/multi_group_merge_test.go
new file mode 100644
index 000000000..ce088cfb9
--- /dev/null
+++ b/pkg/query/vectorized/measure/plan/multi_group_merge_test.go
@@ -0,0 +1,460 @@
+// 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 plan
+
+import (
+ "context"
+ "fmt"
+ "testing"
+ "time"
+
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+ logicalmeasure "github.com/apache/skywalking-banyandb/pkg/query/logical/measure"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized"
+ measure "github.com/apache/skywalking-banyandb/pkg/query/vectorized/measure"
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized/measure/frame"
+)
+
+// mgRowSpec is one stored (ts, SID, version, value) row.
+type mgRowSpec struct {
+ tag string
+ ts, sid, version, value int64
+}
+
+// mgBatchEC serves each storage batch of rows as one MeasureBatch.
+type mgBatchEC struct {
+ schema *databasev1.Measure
+ result *mgBatchResult
+ batches [][]mgRowSpec
+}
+
+func (e *mgBatchEC) Query(_ context.Context, opts model.MeasureQueryOptions) (model.MeasureQueryResult, error) {
+ schema, schemaErr := measure.BuildBatchSchema(e.schema, opts)
+ if schemaErr != nil {
+ return nil, schemaErr
+ }
+ result := &mgBatchResult{}
+ e.result = result
+ for _, rows := range e.batches {
+ batch := &model.MeasureBatch{Schema: schema}
+ for _, row := range rows {
+ batch.Timestamps = append(batch.Timestamps, row.ts)
+ batch.Versions = append(batch.Versions, row.version)
+ batch.ShardIDs = append(batch.ShardIDs, 0)
+ batch.SeriesIDs = append(batch.SeriesIDs, common.SeriesID(row.sid))
+ }
+ for _, def := range schema.Columns {
+ switch def.Role {
+ case vectorized.RoleTag:
+ col := vectorized.NewColumnForType(def.Type, len(rows))
+ for _, row := range rows {
+ tag := row.tag
+ if tag == "" {
+ tag = "x"
+ }
+ col.(*vectorized.TypedColumn[string]).Append(tag)
+ }
+ batch.Tags = append(batch.Tags, col)
+ case vectorized.RoleField:
+ col := vectorized.NewColumnForType(def.Type, len(rows))
+ for _, row := range rows {
+ col.(*vectorized.TypedColumn[int64]).Append(row.value)
+ }
+ batch.Fields = append(batch.Fields, col)
+ case vectorized.RoleTimestamp, vectorized.RoleVersion, vectorized.RoleSeriesID,
+ vectorized.RoleShardID, vectorized.RoleElementID, vectorized.RoleOrderKey:
+ }
+ }
+ result.batches = append(result.batches, batch)
+ }
+ return result, nil
+}
+
+type mgBatchResult struct {
+ batches []*model.MeasureBatch
+ next int
+ released int
+}
+
+func (r *mgBatchResult) Pull() *model.MeasureResult { return nil }
+
+func (r *mgBatchResult) PullBatch(context.Context) (*model.MeasureBatch, error) {
+ if r.next >= len(r.batches) {
+ return nil, nil
+ }
+ batch := r.batches[r.next]
+ r.next++
+ return batch, nil
+}
+
+func (r *mgBatchResult) Release() { r.released++ }
+
+// TestMultiGroupMerge_MaterializesRowsOnce pins that the groups are merged as batches: the rows
+// are converted to InternalDataPoints once, after the merge, and only the rows that are returned.
+// A group turned into rows before the merge would add its own rows to the count.
+func TestMultiGroupMerge_MaterializesRowsOnce(t *testing.T) {
+ cfg := measure.VectorizedConfig{BatchSize: 1, QueryMemoryMiB: 16}
+ groups := []*mgBatchEC{
+ {schema: testMeasureSchemaForGroup("a"), batches: [][]mgRowSpec{{{ts: 10, sid: 7, version: 1, value: 100}}, {{ts: 20, sid: 8, version: 1, value: 200}}}},
+ {schema: testMeasureSchemaForGroup("b"), batches: [][]mgRowSpec{{{ts: 10, sid: 7, version: 3, value: 300}}, {{ts: 10, sid: 7, version: 2, value: 250}}}},
+ }
+ var schemas []logical.Schema
+ var measureSchemas []*databasev1.Measure
+ for _, g := range groups {
+ schema, schemaErr := logicalmeasure.BuildSchema(g.schema, nil)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ schemas = append(schemas, schema)
+ measureSchemas = append(measureSchemas, g.schema)
+ }
+ req := &measurev1.QueryRequest{
+ Name: "demo",
+ Groups: []string{"a", "b"},
+ TagProjection: projTagProj(),
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
+ TimeRange: &modelv1.TimeRange{Begin: timestamppb.New(time.Unix(0, 0)), End: timestamppb.New(time.Unix(1, 0))},
+ Limit: 10,
+ }
+ merge, mergeErr := NewMultiGroupMerge(context.Background(), req, schemas, measureSchemas, cfg)
+ if mergeErr != nil {
+ t.Fatalf("NewMultiGroupMerge: %v", mergeErr)
+ }
+ serializedBefore := measure.SerializedRowCount()
+ var batches []*GroupBatches
+ for i, g := range groups {
+ result, planStr, dispatchErr := DispatchBatches(context.Background(), merge.GroupRequest(),
+ &commonv1.Metadata{Name: "demo", Group: g.schema.GetMetadata().GetGroup()}, g.schema, schemas[i], g, cfg, false, true)
+ var group *GroupBatches
+ if dispatchErr == nil {
+ group, dispatchErr = merge.Collect(context.Background(), result, planStr)
+ }
+ if dispatchErr != nil {
+ t.Fatalf("DispatchBatches: %v", dispatchErr)
+ }
+ batches = append(batches, group)
+ }
+ iter, iterErr := merge.Merge(context.Background(), batches)
+ if iterErr != nil {
+ t.Fatalf("Merge: %v", iterErr)
+ }
+ defer iter.Close()
+ var versions []int64
+ for iter.Next() {
+ versions = append(versions, iter.Current()[0].GetDataPoint().GetVersion())
+ }
+ if len(versions) != 2 || versions[0] != 3 || versions[1] != 1 {
+ t.Fatalf("versions = %v, want [3 1] (version 3 wins (10,7); (20,8) keeps version 1)", versions)
+ }
+ if serialized := measure.SerializedRowCount() - serializedBefore; serialized != int64(len(versions)) {
+ t.Fatalf("%d rows were materialized for a %d-row result: groups must be merged as batches", serialized, len(versions))
+ }
+}
+
+func mgTestSetup(t *testing.T, groups []*mgBatchEC) ([]logical.Schema, []*databasev1.Measure) {
+ t.Helper()
+ var schemas []logical.Schema
+ var measureSchemas []*databasev1.Measure
+ for _, g := range groups {
+ schema, schemaErr := logicalmeasure.BuildSchema(g.schema, nil)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ schemas = append(schemas, schema)
+ measureSchemas = append(measureSchemas, g.schema)
+ }
+ return schemas, measureSchemas
+}
+
+func mgTestRequest(limit uint32) *measurev1.QueryRequest {
+ return &measurev1.QueryRequest{
+ Name: "demo",
+ Groups: []string{"a", "b"},
+ TagProjection: projTagProj(),
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
+ TimeRange: &modelv1.TimeRange{Begin: timestamppb.New(time.Unix(0, 0)), End: timestamppb.New(time.Unix(1, 0))},
+ Limit: limit,
+ }
+}
+
+func mgDispatchAll(t *testing.T, merge *MultiGroupMerge, groups []*mgBatchEC, schemas []logical.Schema, cfg measure.VectorizedConfig) ([]*GroupBatches, error) {
+ t.Helper()
+ var batches []*GroupBatches
+ for i, g := range groups {
+ result, planStr, dispatchErr := DispatchBatches(context.Background(), merge.GroupRequest(),
+ &commonv1.Metadata{Name: "demo", Group: g.schema.GetMetadata().GetGroup()}, g.schema, schemas[i], g, cfg, false, true)
+ var group *GroupBatches
+ if dispatchErr == nil {
+ group, dispatchErr = merge.Collect(context.Background(), result, planStr)
+ }
+ if dispatchErr != nil {
+ for _, obtained := range batches {
+ obtained.Release()
+ }
+ return nil, dispatchErr
+ }
+ batches = append(batches, group)
+ }
+ return batches, nil
+}
+
+// TestMultiGroupMerge_RetainedMemoryIsBounded pins that the batches the groups hand over, their
+// reshaped copies and the merged output are all charged to the query memory budget: two groups
+// of 6,000 rows fit as raw input but not once the reshaped copy exists as well.
+func TestMultiGroupMerge_RetainedMemoryIsBounded(t *testing.T) {
+ cfg := measure.VectorizedConfig{BatchSize: 1024, QueryMemoryMiB: 1}
+ rows := func(base int64) [][]mgRowSpec {
+ batch := make([]mgRowSpec, 6000)
+ for i := range batch {
+ batch[i] = mgRowSpec{ts: base + int64(i), sid: base + int64(i), version: 1, value: 1}
+ }
+ return [][]mgRowSpec{batch}
+ }
+ groups := []*mgBatchEC{
+ {schema: testMeasureSchemaForGroup("a"), batches: rows(0)},
+ {schema: testMeasureSchemaForGroup("b"), batches: rows(100000)},
+ }
+ schemas, measureSchemas := mgTestSetup(t, groups)
+ merge, mergeErr := NewMultiGroupMerge(context.Background(), mgTestRequest(20000), schemas, measureSchemas, cfg)
+ if mergeErr != nil {
+ t.Fatalf("NewMultiGroupMerge: %v", mergeErr)
+ }
+ batches, dispatchErr := mgDispatchAll(t, merge, groups, schemas, cfg)
+ if dispatchErr != nil {
+ t.Fatalf("the raw batches alone must fit the budget: %v", dispatchErr)
+ }
+ iter, iterErr := merge.Merge(context.Background(), batches)
+ if iterErr == nil {
+ iter.Close()
+ t.Fatal("Merge must fail once the reshaped copies exceed the memory budget")
+ }
+ if used := merge.Tracker().Used(); used != 0 {
+ t.Fatalf("a failed merge must give back its reservations, %d bytes still held", used)
+ }
+ for i, g := range groups {
+ if g.result.released == 0 {
+ t.Fatalf("group %d was not released after the failed merge", i)
+ }
+ }
+}
+
+// TestMultiGroupMerge_EmitFrameDecodesAndReleases covers the raw-frame path of the merged result:
+// the frame decodes to the merged rows and closing the iterator releases every group.
+func TestMultiGroupMerge_EmitFrameDecodesAndReleases(t *testing.T) {
+ cfg := measure.VectorizedConfig{BatchSize: 1, QueryMemoryMiB: 16}
+ groups := []*mgBatchEC{
+ {schema: testMeasureSchemaForGroup("a"), batches: [][]mgRowSpec{{{ts: 10, sid: 7, version: 1, value: 100}}, {{ts: 20, sid: 8, version: 1, value: 200}}}},
+ {schema: testMeasureSchemaForGroup("b"), batches: [][]mgRowSpec{{{ts: 10, sid: 7, version: 3, value: 300}}}},
+ }
+ schemas, measureSchemas := mgTestSetup(t, groups)
+ merge, mergeErr := NewMultiGroupMerge(context.Background(), mgTestRequest(10), schemas, measureSchemas, cfg)
+ if mergeErr != nil {
+ t.Fatalf("NewMultiGroupMerge: %v", mergeErr)
+ }
+ batches, dispatchErr := mgDispatchAll(t, merge, groups, schemas, cfg)
+ if dispatchErr != nil {
+ t.Fatalf("DispatchBatches: %v", dispatchErr)
+ }
+ iter, iterErr := merge.Merge(context.Background(), batches)
+ if iterErr != nil {
+ t.Fatalf("Merge: %v", iterErr)
+ }
+ emitter, ok := iter.(measure.FrameEmitter)
+ if !ok {
+ t.Fatalf("%T must emit a raw frame", iter)
+ }
+ body, emitErr := emitter.EmitFrame(context.Background())
+ if emitErr != nil {
+ t.Fatalf("EmitFrame: %v", emitErr)
+ }
+ decoded, decodeErr := frame.Decode(body)
+ if decodeErr != nil {
+ t.Fatalf("Decode: %v", decodeErr)
+ }
+ defer decoded.ReleaseOwners()
+ timestamps := decoded.Columns[decoded.Schema.TimestampIndex()].(*vectorized.TypedColumn[int64]).Data()
+ versions := decoded.Columns[decoded.Schema.VersionIndex()].(*vectorized.TypedColumn[int64]).Data()
+ if decoded.ActiveLen() != 2 || timestamps[0] != 10 || timestamps[1] != 20 || versions[0] != 3 || versions[1] != 1 {
+ t.Fatalf("frame ts=%v versions=%v rows=%d, want ts=[10 20] versions=[3 1]", timestamps, versions, decoded.ActiveLen())
+ }
+ if closeErr := iter.Close(); closeErr != nil {
+ t.Fatalf("Close: %v", closeErr)
+ }
+ for i, g := range groups {
+ if g.result.released == 0 {
+ t.Fatalf("group %d was not released by Close", i)
+ }
+ }
+ if used := merge.Tracker().Used(); used != 0 {
+ t.Fatalf("Close must give back the output reservation, %d bytes still held", used)
+ }
+}
+
+// TestMultiGroupMerge_VariableSizePayloadIsCharged pins that retained memory is charged by what
+// the cells hold: 300 rows with 4 KiB strings are 1.2 MB, far over a 1 MiB budget, although a
+// fixed per-row width would price them at a few tens of KB.
+func TestMultiGroupMerge_VariableSizePayloadIsCharged(t *testing.T) {
+ cfg := measure.VectorizedConfig{BatchSize: 1024, QueryMemoryMiB: 1}
+ big := string(make([]byte, 4096))
+ batch := make([]mgRowSpec, 300)
+ for i := range batch {
+ batch[i] = mgRowSpec{ts: int64(i), sid: int64(i), version: 1, value: 1, tag: big}
+ }
+ groups := []*mgBatchEC{
+ {schema: testMeasureSchemaForGroup("a"), batches: [][]mgRowSpec{batch}},
+ {schema: testMeasureSchemaForGroup("b"), batches: [][]mgRowSpec{batch}},
+ }
+ schemas, measureSchemas := mgTestSetup(t, groups)
+ merge, mergeErr := NewMultiGroupMerge(context.Background(), mgTestRequest(1000), schemas, measureSchemas, cfg)
+ if mergeErr != nil {
+ t.Fatalf("NewMultiGroupMerge: %v", mergeErr)
+ }
+ if _, dispatchErr := mgDispatchAll(t, merge, groups, schemas, cfg); dispatchErr == nil {
+ t.Fatal("DispatchBatches must fail: the 4 KiB strings exceed the memory budget")
+ }
+ if used := merge.Tracker().Used(); used != 0 {
+ t.Fatalf("a failed dispatch must give back its reservations, %d bytes still held", used)
+ }
+ if groups[0].result.released == 0 {
+ t.Fatal("the group that failed its reservation must be released")
+ }
+}
+
+func TestMultiGroupMerge_PreservesEmptyGroupSchemas(t *testing.T) {
+ for _, withRows := range []bool{false, true} {
+ t.Run(fmt.Sprintf("rows=%t", withRows), func(t *testing.T) {
+ ctx := vectorized.WithQueryMemoryTracker(context.Background(), 16<<20)
+ cfg := measure.VectorizedConfig{BatchSize: 4, QueryMemoryMiB: 16}
+ schema := distributedRowsTestSchema()
+ defs := append([]vectorized.ColumnDef(nil), schema.Columns...)
+ defs = append(defs, vectorized.ColumnDef{Name: "zone", TagFamily: "extra", Role: vectorized.RoleTag, Type: vectorized.ColumnTypeString})
+ groups := []*GroupBatches{{Schema: schema}, {Schema: vectorized.NewBatchSchema(defs)}}
+ if withRows {
+ groups[0].Batches = []*vectorized.RecordBatch{buildDistributedRowsBatchDirect(schema,
+ []rowMergeOutput{{ts: 10, sid: 1, ver: 1, value: 42}})}
+ }
+ merge, err := NewMultiGroupMerge(ctx, &measurev1.QueryRequest{Limit: 10}, nil, nil, cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ result, err := merge.MergeBatches(ctx, groups)
+ if err != nil {
+ t.Fatal(err)
+ }
+ idx, ok := result.OutputSchema().TagIndex("extra", "zone")
+ if !ok {
+ t.Fatal("column defined by empty group was dropped")
+ }
+ batch, err := result.NextBatch(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if withRows && (batch == nil || !batch.Columns[idx].IsNull(0)) {
+ t.Fatal("missing tag was not null-filled")
+ }
+ if !withRows && batch != nil {
+ t.Fatal("empty groups produced rows")
+ }
+ if err = result.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if used := merge.Tracker().Used(); used != 0 {
+ t.Fatalf("retained %d bytes after close", used)
+ }
+ })
+ }
+}
+
+func TestMultiGroupMerge_CollectCopiesBorrowedBatches(t *testing.T) {
+ ctx := vectorized.WithQueryMemoryTracker(context.Background(), 16<<20)
+ cfg := measure.VectorizedConfig{BatchSize: 4, QueryMemoryMiB: 16}
+ schema := vectorized.NewBatchSchema([]vectorized.ColumnDef{{Name: "value", Role: vectorized.RoleField, Type: vectorized.ColumnTypeInt64}})
+ owner := &standaloneTestOwner{}
+ source := &recycledGroupSource{schema: schema, batch: vectorized.NewRecordBatch(schema, 2), owner: owner}
+ result, err := NewBatchResult(ctx, source, schema, cfg.BatchSize)
+ if err != nil {
+ t.Fatal(err)
+ }
+ merge, err := NewMultiGroupMerge(ctx, &measurev1.QueryRequest{}, nil, nil, cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ group, err := merge.Collect(ctx, result, "recycled")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(group.Batches) != 2 {
+ t.Fatalf("got %d batches", len(group.Batches))
+ }
+ for idx, batch := range group.Batches {
+ value := batch.Columns[0].(*vectorized.TypedColumn[int64]).Data()[0]
+ if value != int64(idx+1) || batch.Selection[0] != uint16(idx) {
+ t.Fatalf("borrowed batch changed after next pull: value=%d selection=%v", value, batch.Selection)
+ }
+ }
+ if owner.refs != 2 {
+ t.Fatalf("collected batches retained %d owner pins", owner.refs)
+ }
+ group.Release()
+ group.Release()
+ if owner.refs != 0 || merge.Tracker().Used() != 0 || !source.closed {
+ t.Fatal("collected group leaked resources")
+ }
+}
+
+type recycledGroupSource struct {
+ schema *vectorized.BatchSchema
+ batch *vectorized.RecordBatch
+ owner *standaloneTestOwner
+ next int
+ closed bool
+}
+
+func (s *recycledGroupSource) Init(ctx context.Context) error { return ctx.Err() }
+func (s *recycledGroupSource) OutputSchema() *vectorized.BatchSchema { return s.schema }
+func (s *recycledGroupSource) NextBatch(context.Context) (*vectorized.RecordBatch, error) {
+ if s.next >= 2 {
+ return nil, nil
+ }
+ col := s.batch.Columns[0].(*vectorized.TypedColumn[int64])
+ col.Reset()
+ col.Append(int64(s.next + 1))
+ col.Append(int64(s.next + 11))
+ s.batch.Len = 2
+ if s.batch.Selection == nil {
+ s.batch.Selection = []uint16{0}
+ }
+ s.batch.Selection[0] = uint16(s.next)
+ s.batch.Owners.Pin(s.owner)
+ s.next++
+ return s.batch, nil
+}
+
+func (s *recycledGroupSource) Close() error {
+ s.batch.ReleaseOwners()
+ s.closed = true
+ return nil
+}
diff --git a/pkg/query/vectorized/measure/plan/sort_key.go b/pkg/query/vectorized/measure/plan/sort_key.go
index 6194f660f..df6cb829b 100644
--- a/pkg/query/vectorized/measure/plan/sort_key.go
+++ b/pkg/query/vectorized/measure/plan/sort_key.go
@@ -43,12 +43,13 @@ import (
// -Inf < -finite < +/-0 (the two zeros compare equal) < +finite < +Inf.
// - string: raw UTF-8 bytes.
// - []byte: raw bytes.
+// - []int64 / []string (array tags): pbv1.MarshalTagValue of the array TagValue, the
+// same bytes the row path's newComparableElement produced.
// - *modelv1.TagValue passthrough: pbv1.MarshalTagValue applied to the
// cell. This is the same encoding the row path's newComparableElement
// uses, so equal-sort-field windows align with the row baseline.
//
-// Returns an error on unsupported column types (e.g. []int64, []string,
-// FieldValue) — those should be rejected at plan-analysis time before the
+// Returns an error on unsupported column types (e.g. FieldValue) — those should be rejected at plan-analysis time before the
// merger ever sees them. A null cell encodes as nil so it sorts before any
// non-null value in lex order.
func encodeSortKey(col vectorized.Column, rowIdx int) ([]byte, error) {
@@ -99,6 +100,11 @@ func encodeSortKey(col vectorized.Column, rowIdx int) ([]byte, error) {
return nil, nil
}
return pbv1.MarshalTagValue(tv)
+ case *vectorized.TypedColumn[[]int64]:
+ // An array tag sorts by the same encoding the row path gave its TagValue.
+ return pbv1.MarshalTagValue(&modelv1.TagValue{Value: &modelv1.TagValue_IntArray{IntArray: &modelv1.IntArray{Value: c.Data()[rowIdx]}}})
+ case *vectorized.TypedColumn[[]string]:
+ return pbv1.MarshalTagValue(&modelv1.TagValue{Value: &modelv1.TagValue_StrArray{StrArray: &modelv1.StrArray{Value: c.Data()[rowIdx]}}})
}
return nil, fmt.Errorf("encodeSortKey: unsupported column type %s", col.Type())
}
diff --git a/pkg/query/vectorized/measure/plan/sort_key_test.go b/pkg/query/vectorized/measure/plan/sort_key_test.go
index 9bff7e520..94a3bc993 100644
--- a/pkg/query/vectorized/measure/plan/sort_key_test.go
+++ b/pkg/query/vectorized/measure/plan/sort_key_test.go
@@ -150,10 +150,10 @@ func TestEncodeSortKey_TagValue_PassthroughMatchesMarshalTagValue(t *testing.T)
}
func TestEncodeSortKey_UnsupportedType(t *testing.T) {
- col := vectorized.NewInt64ArrayColumn(1)
- col.Append([]int64{1, 2, 3})
+ col := vectorized.NewFieldValueColumn(1)
+ col.Append(&modelv1.FieldValue{Value: &modelv1.FieldValue_Int{Int: &modelv1.Int{Value: 1}}})
if _, err := encodeSortKey(col, 0); err == nil {
- t.Fatal("encodeSortKey on []int64 column must error")
+ t.Fatal("encodeSortKey on a FieldValue column must error")
}
}
@@ -180,3 +180,20 @@ func TestEncodeSortKeyConstColumn(t *testing.T) {
})
}
}
+
+func TestEncodeSortKey_ArraysMatchRowPathTagValueEncoding(t *testing.T) {
+ strCol := vectorized.NewStrArrayColumn(1)
+ strCol.Append([]string{"a", "b"})
+ intCol := vectorized.NewInt64ArrayColumn(1)
+ intCol.Append([]int64{1, 2})
+ wantStr, err := pbv1.MarshalTagValue(&modelv1.TagValue{Value: &modelv1.TagValue_StrArray{StrArray: &modelv1.StrArray{Value: []string{"a", "b"}}}})
+ require.NoError(t, err)
+ wantInt, err := pbv1.MarshalTagValue(&modelv1.TagValue{Value: &modelv1.TagValue_IntArray{IntArray: &modelv1.IntArray{Value: []int64{1, 2}}}})
+ require.NoError(t, err)
+ gotStr, err := encodeSortKey(strCol, 0)
+ require.NoError(t, err)
+ gotInt, err := encodeSortKey(intCol, 0)
+ require.NoError(t, err)
+ require.Equal(t, wantStr, gotStr)
+ require.Equal(t, wantInt, gotInt)
+}
diff --git a/pkg/query/vectorized/measure/serialize.go b/pkg/query/vectorized/measure/serialize.go
index 4638dc6a5..6a7d48265 100644
--- a/pkg/query/vectorized/measure/serialize.go
+++ b/pkg/query/vectorized/measure/serialize.go
@@ -20,6 +20,7 @@ package measure
import (
"slices"
"strings"
+ "sync/atomic"
"time"
"google.golang.org/protobuf/types/known/timestamppb"
@@ -37,6 +38,7 @@ import (
// Row order matches batch row order (respecting Selection); the row-based path
// produces messages in the same order from MeasureResult iteration.
func serializeBatchToProto(b *vectorized.RecordBatch, dst []*measurev1.InternalDataPoint) []*measurev1.InternalDataPoint {
+ serializedRows.Add(int64(b.ActiveLen()))
if dst == nil {
dst = make([]*measurev1.InternalDataPoint, 0, b.ActiveLen())
}
@@ -345,3 +347,10 @@ func copyBytes(src []byte) []byte {
copy(buf, src)
return buf
}
+
+// serializedRows counts the rows serialized into row-form InternalDataPoints by this process.
+var serializedRows atomic.Int64
+
+// SerializedRowCount returns how many rows this process has materialized as InternalDataPoints.
+// Tests use its delta to check that a result is converted to rows only once, at egress.
+func SerializedRowCount() int64 { return serializedRows.Load() }