(skywalking-banyandb) 17/50: test(banyand/measure): storage-level bench for block_cursor decode (passthrough vs native)

[ha...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hanahmily pushed a commit to branch vectorized-query
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git

commit d2b877ef3034596f90816f43ee0c55af0a216879
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Sat May 9 04:28:48 2026 +0000

    test(banyand/measure): storage-level bench for block_cursor decode (passthrough vs native)
    
    Build a real tsTable, snapshot it, and time three storage emit paths
    over identical raw stored bytes:
    
      - row:         copyAllTo (legacy row path) → *MeasureResult
      - passthrough: copyAllToBatch with ColumnTypeTagValue/FieldValue
      - native:      copyAllToBatch with native primitives (Int64/String/...)
    
    This is the gate the package-level bench in pkg/query/vectorized/measure
    cannot model — its fixture builds *modelv1.TagValue at the
    model.MeasureResult boundary, so both passthrough and native end up
    reading pre-built wrappers and the storage-side savings are invisible.
    
    Numbers (100 series x 1000 rows, 2 tags + 1 field, count=3 benchtime=2s):
    
      Row         44ms / 1.21M allocs / 60.5MB
      Passthrough 43ms / 1.20M allocs / 57.7MB
      Native      14ms / 0.30M allocs / 20.8MB
    
    Native skips ~3 allocs per cell (the *modelv1.TagValue wrapper +
    oneof + inner struct that mustDecodeTagValue builds for the row and
    passthrough paths). 100K rows x 3 cells x 3 allocs/cell ≈ 904K alloc
    delta — matches the observed 1.207M - 0.303M.
    
    Native and passthrough are production-equivalent end-to-end: storage
    savings of native (3 allocs/cell) cancel the egress reconstruction
    cost (3 allocs/cell at columnValueToTagValue). The package-level
    bench's failure of native types is bench-fixture bias — its fixture
    skips storage decode entirely so native sees only the egress
    reconstruction, not the storage savings that balance it.
---
 banyand/measure/block_batch_benchmark_test.go | 248 ++++++++++++++++++++++++++
 1 file changed, 248 insertions(+)

diff --git a/banyand/measure/block_batch_benchmark_test.go b/banyand/measure/block_batch_benchmark_test.go
new file mode 100644
index 000000000..d4ae37a78
--- /dev/null
+++ b/banyand/measure/block_batch_benchmark_test.go
@@ -0,0 +1,248 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/apache/skywalking-banyandb/api/common"
+	"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+	"github.com/apache/skywalking-banyandb/banyand/protector"
+	"github.com/apache/skywalking-banyandb/pkg/convert"
+	"github.com/apache/skywalking-banyandb/pkg/fs"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+	"github.com/apache/skywalking-banyandb/pkg/query/model"
+	"github.com/apache/skywalking-banyandb/pkg/query/vectorized"
+	"github.com/apache/skywalking-banyandb/pkg/test"
+	"github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+// G5b storage-level bench: measures the cost of block_cursor decode at
+// the raw-bytes-to-cell boundary, where native column types can skip
+// the *modelv1.TagValue wrapper construction (3 allocs/cell) that the
+// row path and passthrough column types pay through mustDecodeTagValue.
+//
+// The package-level bench in pkg/query/vectorized/measure cannot model
+// this — its fixture builds *modelv1.TagValue at the model.MeasureResult
+// boundary, so both passthrough and native end up reading pre-built
+// wrappers. Here, we build a real tsTable, snapshot it, and time three
+// storage emit paths over identical raw stored bytes:
+//
+//   - row:         copyAllTo (legacy row path) → *MeasureResult
+//   - passthrough: copyAllToBatch with ColumnTypeTagValue/FieldValue
+//   - native:      copyAllToBatch with native primitives
+//
+// Run via:
+//
+//	go test ./banyand/measure -bench=BenchmarkStorageDecode -benchmem -run=^$ -count=3 -benchtime=2s
+
+const (
+	benchStorageSeries  = 100
+	benchStorageRowsPer = 1000
+)
+
+var benchStorageProj = []model.TagProjection{
+	{Family: "default", Names: []string{"svc", "env_id"}},
+}
+
+var benchStorageFieldProj = []string{"v_int"}
+
+var benchStorageSchemaTagTypes = map[string]pbv1.ValueType{
+	"svc":    pbv1.ValueTypeStr,
+	"env_id": pbv1.ValueTypeInt64,
+}
+
+// benchPassthroughSchema mirrors what BuildBatchSchema currently produces
+// in production: tag/field cells stay as *modelv1.TagValue / FieldValue.
+func benchPassthroughSchema() *vectorized.BatchSchema {
+	return vectorized.NewBatchSchema([]vectorized.ColumnDef{
+		{Role: vectorized.RoleTimestamp, Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleVersion, Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleSeriesID, Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleShardID, Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleTag, TagFamily: "default", Name: "svc", Type: vectorized.ColumnTypeTagValue},
+		{Role: vectorized.RoleTag, TagFamily: "default", Name: "env_id", Type: vectorized.ColumnTypeTagValue},
+		{Role: vectorized.RoleField, Name: "v_int", Type: vectorized.ColumnTypeFieldValue},
+	})
+}
+
+// benchNativeSchema declares the same columns with native primitive types,
+// so block_cursor's appendDecodedTagBytesAsTyped path is exercised.
+func benchNativeSchema() *vectorized.BatchSchema {
+	return vectorized.NewBatchSchema([]vectorized.ColumnDef{
+		{Role: vectorized.RoleTimestamp, Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleVersion, Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleSeriesID, Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleShardID, Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleTag, TagFamily: "default", Name: "svc", Type: vectorized.ColumnTypeString},
+		{Role: vectorized.RoleTag, TagFamily: "default", Name: "env_id", Type: vectorized.ColumnTypeInt64},
+		{Role: vectorized.RoleField, Name: "v_int", Type: vectorized.ColumnTypeInt64},
+	})
+}
+
+func buildBenchStorageDataPoints(seriesCount, rowsPerSeries int) *dataPoints {
+	total := seriesCount * rowsPerSeries
+	dps := &dataPoints{
+		seriesIDs:   make([]common.SeriesID, total),
+		timestamps:  make([]int64, total),
+		versions:    make([]int64, total),
+		tagFamilies: make([][]nameValues, total),
+		fields:      make([]nameValues, total),
+	}
+	for s := range seriesCount {
+		for r := range rowsPerSeries {
+			i := s*rowsPerSeries + r
+			dps.seriesIDs[i] = common.SeriesID(s + 1)
+			dps.timestamps[i] = int64(r + 1)
+			dps.versions[i] = 1
+			dps.tagFamilies[i] = []nameValues{
+				{
+					name: "default", values: []*nameValue{
+						{name: "svc", valueType: pbv1.ValueTypeStr, value: []byte("alpha")},
+						{name: "env_id", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(int64(s + 1))},
+					},
+				},
+			}
+			dps.fields[i] = nameValues{
+				name: "skipped", values: []*nameValue{
+					{name: "v_int", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(42)},
+				},
+			}
+		}
+	}
+	return dps
+}
+
+// benchStorageHarness owns the tsTable + snapshot + part list used by all
+// three storage decode benchmarks. Built once per benchmark function via
+// b.Helper / b.Cleanup so the per-iteration cost focuses on cursor decode.
+type benchStorageHarness struct {
+	tst       *tsTable
+	snap      *snapshot
+	parts     []*part
+	sids      []common.SeriesID
+	queryOpts queryOptions
+}
+
+func setupBenchStorageHarness(b *testing.B) *benchStorageHarness {
+	b.Helper()
+	tmpPath, defFn := test.Space(require.New(b))
+	tst, err := newTSTable(
+		fs.NewLocalFileSystem(),
+		tmpPath, common.Position{},
+		logger.GetLogger("bench"),
+		timestamp.TimeRange{},
+		option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+		nil,
+	)
+	require.NoError(b, err)
+	dps := buildBenchStorageDataPoints(benchStorageSeries, benchStorageRowsPer)
+	tst.mustAddDataPoints(dps)
+	time.Sleep(200 * time.Millisecond)
+
+	snap := tst.currentSnapshot()
+	require.NotNil(b, snap)
+
+	shardCache := storage.NewShardCache("bench-group", 0, 0)
+	parts, _ := snap.getParts(nil, shardCache, 0, int64(benchStorageRowsPer)+1)
+	require.NotEmpty(b, parts)
+
+	sids := make([]common.SeriesID, benchStorageSeries)
+	for i := range sids {
+		sids[i] = common.SeriesID(i + 1)
+	}
+
+	queryOpts := queryOptions{
+		schemaTagTypes: benchStorageSchemaTagTypes,
+		minTimestamp:   0,
+		maxTimestamp:   int64(benchStorageRowsPer) + 1,
+	}
+	queryOpts.MeasureQueryOptions.TagProjection = benchStorageProj
+	queryOpts.MeasureQueryOptions.FieldProjection = benchStorageFieldProj
+
+	b.Cleanup(func() {
+		snap.decRef()
+		tst.Close()
+		defFn()
+	})
+	return &benchStorageHarness{tst: tst, snap: snap, parts: parts, sids: sids, queryOpts: queryOpts}
+}
+
+// drainCursors walks every block in the harness's parts and invokes f
+// against each loaded cursor. Mirrors loadCursorsForBatch's per-block
+// step but inline so the bench stays single-goroutine and deterministic.
+func (h *benchStorageHarness) drainCursors(f func(bc *blockCursor)) {
+	ti := &tstIter{}
+	ti.init(h.parts, h.sids, h.queryOpts.minTimestamp, h.queryOpts.maxTimestamp)
+	for ti.nextBlock() {
+		bc := generateBlockCursor()
+		p := ti.piHeap[0]
+		opts := h.queryOpts
+		opts.TagProjection = benchStorageProj
+		opts.FieldProjection = benchStorageFieldProj
+		bc.init(p.p, p.curBlock, opts)
+		tmpBlock := generateBlock()
+		if bc.loadData(tmpBlock) {
+			f(bc)
+		}
+		releaseBlock(tmpBlock)
+		releaseBlockCursor(bc)
+	}
+}
+
+func BenchmarkStorageDecode_Row(b *testing.B) {
+	h := setupBenchStorageHarness(b)
+	b.ReportAllocs()
+	b.ResetTimer()
+	for b.Loop() {
+		h.drainCursors(func(bc *blockCursor) {
+			r := &model.MeasureResult{}
+			bc.copyAllTo(r, nil, benchStorageProj, false)
+		})
+	}
+}
+
+func BenchmarkStorageDecode_Passthrough(b *testing.B) {
+	h := setupBenchStorageHarness(b)
+	schema := benchPassthroughSchema()
+	b.ReportAllocs()
+	b.ResetTimer()
+	for b.Loop() {
+		h.drainCursors(func(bc *blockCursor) {
+			mb := newMeasureBatchForSchema(schema, len(bc.timestamps))
+			bc.copyAllToBatch(mb, schema, nil, false)
+		})
+	}
+}
+
+func BenchmarkStorageDecode_Native(b *testing.B) {
+	h := setupBenchStorageHarness(b)
+	schema := benchNativeSchema()
+	b.ReportAllocs()
+	b.ResetTimer()
+	for b.Loop() {
+		h.drainCursors(func(bc *blockCursor) {
+			mb := newMeasureBatchForSchema(schema, len(bc.timestamps))
+			bc.copyAllToBatch(mb, schema, nil, false)
+		})
+	}
+}