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/05/15 15:32:28 UTC
(skywalking-banyandb) branch vectorized-query updated (768d67dd9 -> b5f551319)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch vectorized-query
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
from 768d67dd9 test(query/vectorized/measure/plan): rewrite stale G9b fall-through tests + clarify empty-result hidden-strip skip
new 0b163e6d9 test(query/vectorized/measure): G9e.1 real plan-path bench harness
new b5f551319 perf(query/vectorized/measure): defer BatchTop row copy to heap admission
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
pkg/query/vectorized/measure/bench_gates_test.go | 85 ++--
pkg/query/vectorized/measure/bench_test.go | 598 ++++++++++++++++++++---
pkg/query/vectorized/measure/top.go | 55 ++-
3 files changed, 603 insertions(+), 135 deletions(-)
(skywalking-banyandb) 02/02: perf(query/vectorized/measure): defer BatchTop row copy to heap admission
Posted by 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 b5f5513196fba381d9a562d3acb4388a8f86fed6
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 15:27:18 2026 +0000
perf(query/vectorized/measure): defer BatchTop row copy to heap admission
BatchTop.materialize deep-copied every candidate input row (all
schema columns into fresh 1-row typed columns) before the bounded-heap
check, so high-input / low-N top-N paid a full per-row materialize for
rows immediately discarded. Split into a cheap extractKey (sort key
only, every row) and materializeCols (deep copy, only on heap
admission). Rejected rows now allocate nothing.
W3 Top-N bench (100k rows): vec 58.0M -> 8.75M ns/op, 1.40M -> 117.6K
allocs/op -- from 1.32x slower than the row path to ~4.85x faster.
Output is byte-identical (retained rows still fully copied while the
batch is valid; cmpTopVal/shouldReplace read only key fields):
integration parity gate green (321s, all fixtures proto.Equal),
TestBatchTop_* + measure/plan unit suites green.
---
pkg/query/vectorized/measure/top.go | 55 +++++++++++++++++++++++++------------
1 file changed, 38 insertions(+), 17 deletions(-)
diff --git a/pkg/query/vectorized/measure/top.go b/pkg/query/vectorized/measure/top.go
index f58fc99b3..bd9a91a05 100644
--- a/pkg/query/vectorized/measure/top.go
+++ b/pkg/query/vectorized/measure/top.go
@@ -149,16 +149,30 @@ func (t *BatchTop) Consume(_ context.Context, b *vectorized.RecordBatch) error {
}
active := activeIndices(b)
for _, rowIdx := range active {
- candidate := t.materialize(b, int(rowIdx))
- candidate.seq = t.inputCount
+ ri := int(rowIdx)
+ seq := t.inputCount
t.inputCount++
+ // Heap not yet full: the row is retained, so pay the full
+ // per-column deep copy now (the batch is recycled after Consume
+ // returns, so a retained row cannot defer its copy).
if t.heapState.Len() < t.n {
- heap.Push(t.heapState, candidate)
+ row := &topRow{seq: seq}
+ t.extractKey(b, ri, row)
+ row.cols = t.materializeCols(b, ri)
+ heap.Push(t.heapState, row)
continue
}
+ // Heap full: extract only the cheap sort key and compare against
+ // the root. The expensive per-column materialize is deferred
+ // until the row actually displaces the root — for high-input /
+ // low-N top-N the vast majority of rows lose here and never pay
+ // the copy. cmpTopVal/shouldReplace read only the key fields.
+ cand := topRow{seq: seq}
+ t.extractKey(b, ri, &cand)
root := t.heapState.rows[0]
- if t.shouldReplace(candidate, root) {
- t.heapState.rows[0] = candidate
+ if t.shouldReplace(&cand, root) {
+ cand.cols = t.materializeCols(b, ri)
+ t.heapState.rows[0] = &cand
heap.Fix(t.heapState, 0)
}
}
@@ -224,8 +238,10 @@ func (t *BatchTop) Close() error {
return nil
}
-// materialize copies row rowIdx of b into a new topRow, reading the sort key
-// from the configured field column.
+// extractKey reads only the sort key for row rowIdx from the configured
+// field column into row — no per-column deep copy. This is the cheap
+// half of the old materialize: it runs for every input row, while the
+// expensive materializeCols runs only for rows admitted to the heap.
//
// The key column may be a native typed column (ColumnTypeInt64 /
// ColumnTypeFloat64 — promoted when an Agg reduces over the field) or a
@@ -234,26 +250,20 @@ func (t *BatchTop) Close() error {
// fields as passthrough). Both are handled so the float/int decision is
// per-column-shape, matching the row path's schema-field-type dispatch
// (pkg/query/logical/measure.topOp.Execute).
-func (t *BatchTop) materialize(b *vectorized.RecordBatch, rowIdx int) *topRow {
- cols := make([]vectorized.Column, len(t.schema.Columns))
- for i, def := range t.schema.Columns {
- cols[i] = vectorized.NewColumnForType(def.Type, 1)
- copyOneValue(cols[i], b.Columns[i], rowIdx)
- }
- row := &topRow{cols: cols}
+func (t *BatchTop) extractKey(b *vectorized.RecordBatch, rowIdx int, row *topRow) {
keyCol := b.Columns[t.fieldCol]
switch c := keyCol.(type) {
case *vectorized.TypedColumn[float64]:
row.isFloat = true
if c.IsNull(rowIdx) {
row.isNull = true
- return row
+ return
}
row.floatVal = c.Data()[rowIdx]
case *vectorized.TypedColumn[int64]:
if c.IsNull(rowIdx) {
row.isNull = true
- return row
+ return
}
row.intVal = c.Data()[rowIdx]
case *vectorized.TypedColumn[*modelv1.FieldValue]:
@@ -275,5 +285,16 @@ func (t *BatchTop) materialize(b *vectorized.RecordBatch, rowIdx int) *topRow {
// rather than panicking.
row.isNull = true
}
- return row
+}
+
+// materializeCols deep-copies row rowIdx of b into schema-shaped 1-row
+// columns. Only called for rows admitted to the bounded heap (heap not
+// full, or the row displaces the root) — rejected rows never pay this.
+func (t *BatchTop) materializeCols(b *vectorized.RecordBatch, rowIdx int) []vectorized.Column {
+ cols := make([]vectorized.Column, len(t.schema.Columns))
+ for i, def := range t.schema.Columns {
+ cols[i] = vectorized.NewColumnForType(def.Type, 1)
+ copyOneValue(cols[i], b.Columns[i], rowIdx)
+ }
+ return cols
}
(skywalking-banyandb) 01/02: test(query/vectorized/measure): G9e.1 real plan-path bench harness
Posted by 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 0b163e6d9420bc4d9ba0860b898ee84164a117ad
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 15:17:00 2026 +0000
test(query/vectorized/measure): G9e.1 real plan-path bench harness
Replace the deprecated NewMIterator leaf-substitution bench (scan/decode
only; G7e operator wiring was reverted) with paired row/vec benchmarks
that drive the production operator pipeline (BatchScan -> BatchTop /
BuildOperators -> BatchLimit -> NewIteratorFromPipeline) against a
faithful row-path baseline replicated via pkg/query/aggregation.
Workloads W1-W8: scan (W1/W2), Top-N (W3), scalar reduce (W4), raw
GroupBy (W5), GroupBy+Agg (W6), hidden-tags egress strip (W7),
COUNT-on-float (W8). Gates opt-in via RUN_BENCH_GATES=1 (not in
test-ci). Aggregation shapes show vec 4-8x faster than row; W3 Top-N
currently exceeds the ns gate (vec ~1.3x) because BatchTop.materialize
deep-copies every candidate row before the bounded-heap check -- a
genuine operator regression tracked for the next fix, not absorbed by
loosening the gate.
---
pkg/query/vectorized/measure/bench_gates_test.go | 85 ++--
pkg/query/vectorized/measure/bench_test.go | 598 ++++++++++++++++++++---
2 files changed, 565 insertions(+), 118 deletions(-)
diff --git a/pkg/query/vectorized/measure/bench_gates_test.go b/pkg/query/vectorized/measure/bench_gates_test.go
index 11e24a0ff..0e0bed5f1 100644
--- a/pkg/query/vectorized/measure/bench_gates_test.go
+++ b/pkg/query/vectorized/measure/bench_gates_test.go
@@ -22,19 +22,34 @@ import (
"testing"
)
-// G5a acceptance gates per spec §"Performance Evaluation Plan". Ratios are
-// vectorized / row; failing the gate is a regression that blocks the
-// default-flip rollout.
+// G9e acceptance gates. Ratios are vectorized / row; failing the gate is a
+// regression that blocks the default-flip rollout. The vec half here is the
+// production compute pipeline (source → [BatchAggregation | BatchTop] →
+// BatchLimit drained through the public adapter), not the deprecated
+// NewMIterator leaf-substitution path.
//
-// The alloc gate is set to 1.005 (0.5% tolerance) rather than the spec's
-// literal 1.00. Reason: each runVectorizedPath call constructs a fresh
-// BatchSchema, BatchPool, BatchScan, Pipeline, and MIterator wrapper —
-// roughly 20 fixture allocations per query. The row path's resultMIterator
-// is a struct literal with effectively zero fixture cost. Spread over
-// W1's 10K rows that's a 0.05% per-iteration delta; over W3/W4's 100K rows,
-// 0.014%. The spec author's "architectural benefit must materialize"
-// intent is satisfied at 1.005 — a real per-row alloc regression would
-// blow far past 0.5%, while fixture noise stays under it.
+// Tolerance rationale:
+//
+// - Scan shapes (W1, W2, W7): ns ≤ 1.05, allocs ≤ 1.005, bytes ≤ 1.20.
+// The alloc gate is 1.005 (0.5%) rather than a literal 1.00 because each
+// runVectorizedPath call constructs a fresh BatchSchema, BatchPool,
+// BatchScan, Pipeline, and adapter — roughly 20 fixture allocations per
+// query. The row path's serializer is a struct literal with effectively
+// zero fixture cost. Spread over W1's 10K rows that is a 0.05%
+// per-iteration delta; over 100K-row workloads, 0.014%. The "per-cell
+// wrapper bias" rationale from G5a still applies: buildResults allocates
+// a fresh TagValue/FieldValue per cell so both paths pay the same
+// storage-decode cost; a real per-row alloc regression blows far past
+// 0.5% while fixture noise stays under it. bytes ≤ 1.20 absorbs the
+// RecordBatch column backing the row path never allocates.
+//
+// - Agg / Top / GroupBy shapes (W3, W4, W5, W6, W8): vec should win or
+// tie because aggregation/top runs on typed columns and the output row
+// count is bounded by group/heap cardinality, amortizing wrapper
+// reconstruction at egress. ns ≤ 1.05, allocs ≤ 1.05, bytes ≤ 1.20
+// initially — these can tighten once a baseline is recorded (the vec
+// advantage should let ns/allocs drop well below 1.0 for high-input,
+// low-output shapes).
type benchGate struct {
id string
maxNsRatio float64 // ns/op ≤ row × maxNsRatio
@@ -42,34 +57,30 @@ type benchGate struct {
maxBytesRatio float64 // B/op ≤ row × maxBytesRatio
}
-// W3's spec gate is `vec ≤ row × 1.00` — tighter than the others — because
-// W3 is "GroupBy + SUM/COUNT" and columnar should win outright once
-// aggregation runs on the columns. With G4's wiring, operators are not yet
-// wired into NewMIterator's pipeline, so W3 here measures the same scan +
-// serialize cost as W2 — the strict gate is shape-mismatched. Relaxed to
-// 1.05 to match the other scan-shape gates; tighten back to 1.00 once
-// BatchAggregation/BatchGroupBy execute end-to-end (post-G6b).
var benchGates = map[string]benchGate{
- "W1": {id: "W1", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
- "W2": {id: "W2", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
- "W3": {id: "W3", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
- "W4": {id: "W4", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
- "W5": {id: "W5", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
- "W2-MB": {id: "W2-MB", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
- "W4-MB": {id: "W4-MB", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
- "W5-MB": {id: "W5-MB", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
+ // Scan-shaped (regression continuity + hidden-tags egress strip).
+ "W1": {id: "W1", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
+ "W2": {id: "W2", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
+ "W7": {id: "W7", maxNsRatio: 1.05, maxAllocRatio: 1.005, maxBytesRatio: 1.20},
+ // Compute-shaped (Top / scalar reduce / raw GroupBy / GroupBy+Agg /
+ // COUNT-on-float). vec should win or tie; tighten once baselined.
+ "W3": {id: "W3", maxNsRatio: 1.05, maxAllocRatio: 1.05, maxBytesRatio: 1.20},
+ "W4": {id: "W4", maxNsRatio: 1.05, maxAllocRatio: 1.05, maxBytesRatio: 1.20},
+ "W5": {id: "W5", maxNsRatio: 1.05, maxAllocRatio: 1.05, maxBytesRatio: 1.20},
+ "W6": {id: "W6", maxNsRatio: 1.05, maxAllocRatio: 1.05, maxBytesRatio: 1.20},
+ "W8": {id: "W8", maxNsRatio: 1.05, maxAllocRatio: 1.05, maxBytesRatio: 1.20},
}
-// TestBenchGates_PerWorkload runs both serialization paths inside testing.B
-// harnesses and asserts the spec's vec/row ratios. A regression fails this
-// test, not just the markdown report — gates are enforced as code.
+// TestBenchGates_PerWorkload runs both paths inside testing.B harnesses and
+// asserts the vec/row ratios. A regression fails this test, not just the
+// markdown report — gates are enforced as code.
//
-// Skipped unless RUN_BENCH_GATES=1 is set (or short mode is off and the host
-// is not under load): this test takes ~10–20s of wall time per workload and
-// is gated on a CI-tunable knob to keep `go test ./...` fast.
+// Skipped unless RUN_BENCH_GATES=1 is set: this test takes ~10–20s of wall
+// time per workload and is gated on a CI-tunable knob to keep
+// `go test ./...` fast.
func TestBenchGates_PerWorkload(t *testing.T) {
if os.Getenv("RUN_BENCH_GATES") != "1" {
- t.Skip("set RUN_BENCH_GATES=1 to run G5a bench gates")
+ t.Skip("set RUN_BENCH_GATES=1 to run G9e bench gates")
}
if testing.Short() {
t.Skip("skipping bench gates in -short mode")
@@ -120,15 +131,15 @@ func timeWorkload(spec workloadSpec, vectorized bool) testing.BenchmarkResult {
results := buildResults(spec)
schema := buildSchema(spec)
opts := buildOpts(spec)
- cfg := VectorizedConfig{Enabled: true, BatchSize: 1024, QueryMemoryMiB: 64}
+ cfg := cfgFor()
return testing.Benchmark(func(b *testing.B) {
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
if vectorized {
- runVectorizedPath(results, schema, opts, cfg)
+ runVectorizedPath(spec, results, schema, opts, cfg)
} else {
- runRowPath(results, opts)
+ runRowPath(spec, results, opts)
}
}
})
diff --git a/pkg/query/vectorized/measure/bench_test.go b/pkg/query/vectorized/measure/bench_test.go
index b655d547e..a2886eee6 100644
--- a/pkg/query/vectorized/measure/bench_test.go
+++ b/pkg/query/vectorized/measure/bench_test.go
@@ -18,44 +18,99 @@
package measure
import (
+ "container/heap"
"context"
"testing"
"github.com/apache/skywalking-banyandb/api/common"
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/aggregation"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
"github.com/apache/skywalking-banyandb/pkg/query/model"
+ "github.com/apache/skywalking-banyandb/pkg/query/vectorized"
)
-// Microbenchmarks (G5a) — paired row-path vs vectorized-path benchmarks for
-// W1..W5 per the spec's Performance Evaluation Plan. Both paths consume the
-// same fake MeasureQueryResult; only the serialization implementation
-// differs. ns/op, B/op, allocs/op are reported via testing.B.
+// Microbenchmarks (G9e) — paired row-path vs vectorized-path benchmarks for
+// the G9 query shapes. The vec half is the *production* compute path: a
+// vectorized.Pipeline composed of the same measure-level operators
+// plan.Execute wires (source → [BatchAggregation | BatchTop] → BatchLimit →
+// NewIteratorFromPipeline), drained through the public adapter. The row half
+// is a faithful replica of pkg/query/logical/measure's reduce/top — it cannot
+// be imported (cycle: logical/measure → this package), so it is replicated
+// here exactly as diff_test.go documents for rowSerialize.
+//
+// The earlier harness benchmarked NewMIterator (the deprecated
+// leaf-substitution path) which is scan/decode only and never runs
+// Top/Agg/GroupBy — that did not measure the production vec compute the
+// parity gate validates. This file replaces it.
//
// Workload scales are bounded for unit-bench tractability — the integration
// macro suite at test/integration/standalone/benchmark/ exercises full-scale
// shapes against the real Measure module. Acceptance gates are ratios
-// (vec/row), so the relative comparison holds at any scale; absolute
-// throughput is not the gate.
+// (vec/row), so the relative comparison holds at any scale.
//
// Run via:
//
// go test ./pkg/query/vectorized/measure -bench=. -benchmem -count=5 -benchtime=2s
-// workloadSpec parameterizes a benchmark workload.
+// Repeated literals (goconst min-occurrences 4).
+const (
+ benchSvc = "svc"
+ benchValue = "value"
+ benchDefault = "default"
+ benchEnvID = "env_id"
+ benchVInt = "v_int"
+ benchVFloat = "v_float"
+ benchCriteria = "criteria"
+ defaultBatch = 1024
+ defaultMemMiB = 64
+ defaultTopN = 50
+ defaultLimitN = 100
+ groupKeyCardin = 16
+)
+
+// queryShape selects which operator pipeline a workload exercises. Each
+// shape names one production query class the vec subsystem handles via
+// plan.Dispatch (scan/top/scalarReduce/rawGroupBy/groupByAgg/hiddenTags/
+// countFloat); the row half reduces the same input with the row path's
+// algorithm so the gate ratio reflects the real compute trade-off.
+type queryShape int
+
+// queryShape values.
+const (
+ shapeScan queryShape = iota
+ shapeTop
+ shapeScalarReduce
+ shapeRawGroupBy
+ shapeGroupByAgg
+ shapeHiddenTags
+ shapeCountFloat
+)
+
+// workloadSpec parameterizes a benchmark workload and the query shape applied
+// on top of its scan output.
//
// chunksPerSeries > 1 splits each series's rowsPer rows into that many
// *model.MeasureResult instances, simulating the multi-block storage path
-// where queryResult.merge produces multiple Pull() results per series. The
-// vec adapter must accumulate rows across PullBatch calls; the row adapter
-// just sees more Pull invocations. Default 1 = single-block per series.
+// where queryResult.merge produces multiple Pull() results per series.
+// Default 1 = single-block per series.
type workloadSpec struct {
+ groupBy *model.MeasureGroupBy
+ agg *model.MeasureAgg
id string
+ topField string
+ criteriaTag string
tagFamilies []tagSpec
fields []fieldSpec
series int
rowsPer int
chunksPerSeries int
+ topN int
+ limitN int
+ shape queryShape
+ topAsc bool
}
type tagSpec struct {
@@ -69,76 +124,101 @@ type fieldSpec struct {
col databasev1.FieldType
}
-// w1..w5 mirror the spec's catalog, scaled for unit-bench memory budgets.
-// Total rows per workload is held near 100k so each benchmark iteration
-// completes quickly enough for -benchtime=2s to amortize fixture build cost.
+// Workload catalog: a scan baseline (W1/W2 for regression continuity), then
+// one workload per G9 query shape. Total rows per workload is held near 100k
+// so each iteration amortizes fixture build cost under -benchtime=2s.
var (
+ // W1 — single-series scan baseline (regression continuity).
w1 = workloadSpec{
id: "W1",
+ shape: shapeScan,
series: 1, rowsPer: 10000,
- fields: []fieldSpec{{name: "v_int", col: databasev1.FieldType_FIELD_TYPE_INT}},
+ fields: []fieldSpec{{name: benchVInt, col: databasev1.FieldType_FIELD_TYPE_INT}},
}
+ // W2 — multi-tag multi-series scan baseline (regression continuity).
w2 = workloadSpec{
id: "W2",
+ shape: shapeScan,
series: 100, rowsPer: 1000,
tagFamilies: []tagSpec{
- {family: "default", name: "svc", col: databasev1.TagType_TAG_TYPE_STRING},
- {family: "default", name: "env_id", col: databasev1.TagType_TAG_TYPE_INT},
+ {family: benchDefault, name: benchSvc, col: databasev1.TagType_TAG_TYPE_STRING},
+ {family: benchDefault, name: benchEnvID, col: databasev1.TagType_TAG_TYPE_INT},
},
fields: []fieldSpec{
- {name: "v_int", col: databasev1.FieldType_FIELD_TYPE_INT},
- {name: "v_float", col: databasev1.FieldType_FIELD_TYPE_FLOAT},
+ {name: benchVInt, col: databasev1.FieldType_FIELD_TYPE_INT},
+ {name: benchVFloat, col: databasev1.FieldType_FIELD_TYPE_FLOAT},
},
}
+ // W3 — Top-N by an int field over a scan (Scan → Top → Limit).
w3 = workloadSpec{
id: "W3",
- series: 1000, rowsPer: 100,
- tagFamilies: []tagSpec{{family: "default", name: "svc", col: databasev1.TagType_TAG_TYPE_STRING}},
- fields: []fieldSpec{{name: "v_int", col: databasev1.FieldType_FIELD_TYPE_INT}},
+ shape: shapeTop,
+ series: 100, rowsPer: 1000,
+ tagFamilies: []tagSpec{{family: benchDefault, name: benchSvc, col: databasev1.TagType_TAG_TYPE_STRING}},
+ fields: []fieldSpec{{name: benchValue, col: databasev1.FieldType_FIELD_TYPE_INT}},
+ topField: benchValue, topN: defaultTopN, topAsc: false, limitN: defaultLimitN,
}
+ // W4 — scalar reduce: SUM over an int field, no GroupBy (Agg only).
w4 = workloadSpec{
id: "W4",
+ shape: shapeScalarReduce,
series: 100, rowsPer: 1000,
- tagFamilies: []tagSpec{{family: "default", name: "svc", col: databasev1.TagType_TAG_TYPE_STRING}},
- fields: []fieldSpec{{name: "v_int", col: databasev1.FieldType_FIELD_TYPE_INT}},
+ tagFamilies: []tagSpec{{family: benchDefault, name: benchSvc, col: databasev1.TagType_TAG_TYPE_STRING}},
+ fields: []fieldSpec{{name: benchValue, col: databasev1.FieldType_FIELD_TYPE_INT}},
+ agg: &model.MeasureAgg{FieldName: benchValue, Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ limitN: defaultLimitN,
}
+ // W5 — raw GroupBy by svc, no Agg: first-seen row per group.
w5 = workloadSpec{
id: "W5",
- series: 1000, rowsPer: 100,
+ shape: shapeRawGroupBy,
+ series: groupKeyCardin, rowsPer: 6250,
+ tagFamilies: []tagSpec{{family: benchDefault, name: benchSvc, col: databasev1.TagType_TAG_TYPE_STRING}},
+ fields: []fieldSpec{{name: benchValue, col: databasev1.FieldType_FIELD_TYPE_INT}},
+ groupBy: &model.MeasureGroupBy{TagFamily: benchDefault, TagNames: []string{benchSvc}},
+ limitN: defaultLimitN,
+ }
+ // W6 — GroupBy by svc + SUM over an int field.
+ w6 = workloadSpec{
+ id: "W6",
+ shape: shapeGroupByAgg,
+ series: groupKeyCardin, rowsPer: 6250,
+ tagFamilies: []tagSpec{{family: benchDefault, name: benchSvc, col: databasev1.TagType_TAG_TYPE_STRING}},
+ fields: []fieldSpec{{name: benchValue, col: databasev1.FieldType_FIELD_TYPE_INT}},
+ groupBy: &model.MeasureGroupBy{TagFamily: benchDefault, TagNames: []string{benchSvc}},
+ agg: &model.MeasureAgg{FieldName: benchValue, Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ limitN: defaultLimitN,
+ }
+ // W7 — hidden-tags egress strip: a criteria tag materialized for
+ // storage-side filtering but absent from the visible projection, then
+ // stripped at egress (both paths pay the strip cost).
+ w7 = workloadSpec{
+ id: "W7",
+ shape: shapeHiddenTags,
+ series: 100, rowsPer: 1000,
tagFamilies: []tagSpec{
- {family: "default", name: "svc", col: databasev1.TagType_TAG_TYPE_STRING},
- {family: "default", name: "env_id", col: databasev1.TagType_TAG_TYPE_INT},
- {family: "default", name: "blob", col: databasev1.TagType_TAG_TYPE_DATA_BINARY},
- {family: "default", name: "ports", col: databasev1.TagType_TAG_TYPE_INT_ARRAY},
- {family: "default", name: "labels", col: databasev1.TagType_TAG_TYPE_STRING_ARRAY},
- },
- fields: []fieldSpec{
- {name: "v_int", col: databasev1.FieldType_FIELD_TYPE_INT},
- {name: "v_float", col: databasev1.FieldType_FIELD_TYPE_FLOAT},
- {name: "v_str", col: databasev1.FieldType_FIELD_TYPE_STRING},
- {name: "v_bytes", col: databasev1.FieldType_FIELD_TYPE_DATA_BINARY},
+ {family: benchDefault, name: benchSvc, col: databasev1.TagType_TAG_TYPE_STRING},
+ {family: benchDefault, name: benchCriteria, col: databasev1.TagType_TAG_TYPE_STRING},
},
+ fields: []fieldSpec{{name: benchVInt, col: databasev1.FieldType_FIELD_TYPE_INT}},
+ criteriaTag: benchCriteria,
+ }
+ // W8 — COUNT over a float field: exercises the float-typed COUNT slot
+ // (output follows input type, so COUNT-on-float emits a float).
+ w8 = workloadSpec{
+ id: "W8",
+ shape: shapeCountFloat,
+ series: groupKeyCardin, rowsPer: 6250,
+ tagFamilies: []tagSpec{{family: benchDefault, name: benchSvc, col: databasev1.TagType_TAG_TYPE_STRING}},
+ fields: []fieldSpec{{name: benchVFloat, col: databasev1.FieldType_FIELD_TYPE_FLOAT}},
+ groupBy: &model.MeasureGroupBy{TagFamily: benchDefault, TagNames: []string{benchSvc}},
+ agg: &model.MeasureAgg{FieldName: benchVFloat, Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_COUNT},
+ limitN: defaultLimitN,
}
- // Multi-block variants: same shape as their single-block counterpart but
- // with chunksPerSeries=4, so each series produces 4 *MeasureResult entries
- // instead of 1. Exercises BatchSourceFromBatchResult's cross-Pull
- // accumulation path that copies rows across PullBatch calls into a single
- // RecordBatch.
- w2mb = withChunks(w2, "W2-MB", 4)
- w4mb = withChunks(w4, "W4-MB", 4)
- w5mb = withChunks(w5, "W5-MB", 4)
-
- allWorkloads = []workloadSpec{w1, w2, w3, w4, w5, w2mb, w4mb, w5mb}
+ allWorkloads = []workloadSpec{w1, w2, w3, w4, w5, w6, w7, w8}
)
-func withChunks(base workloadSpec, id string, chunks int) workloadSpec {
- c := base
- c.id = id
- c.chunksPerSeries = chunks
- return c
-}
-
// buildResults materializes a deterministic []*model.MeasureResult for the
// workload. Each cell allocates a fresh *modelv1.TagValue / *modelv1.FieldValue
// — mirroring production storage's mustDecodeTagValue, which produces a new
@@ -150,6 +230,11 @@ func withChunks(base workloadSpec, id string, chunks int) workloadSpec {
// wrappers, all paths see the same storage-decode cost the production
// queryResult pays — the ratios reflect the real pipeline trade-off.
//
+// To exercise GroupBy/Top with realistic cardinality, the svc tag and the
+// numeric field vary per series/row: svc = "svc-<sid mod groupKeyCardin>"
+// so the number of distinct groups is bounded, and the numeric value is
+// (row index) so Top has a non-degenerate ordering.
+//
// chunksPerSeries > 1 splits each series's rows across multiple
// *model.MeasureResult entries, simulating multi-block heap-merge output
// from queryResult.merge.
@@ -158,6 +243,7 @@ func buildResults(spec workloadSpec) []*model.MeasureResult {
results := make([]*model.MeasureResult, 0, spec.series*chunks)
for s := range spec.series {
sid := common.SeriesID(s + 1)
+ group := s % groupKeyCardin
for c := range chunks {
start := (spec.rowsPer * c) / chunks
end := (spec.rowsPer * (c + 1)) / chunks
@@ -178,7 +264,7 @@ func buildResults(spec workloadSpec) []*model.MeasureResult {
for _, ts := range spec.tagFamilies {
values := make([]*modelv1.TagValue, n)
for i := range values {
- values[i] = freshTagValue(ts.col)
+ values[i] = freshTagValueFor(ts, group, start+i)
}
tags = append(tags, model.Tag{Name: ts.name, Values: values})
}
@@ -189,7 +275,7 @@ func buildResults(spec workloadSpec) []*model.MeasureResult {
for _, f := range spec.fields {
values := make([]*modelv1.FieldValue, n)
for i := range values {
- values[i] = freshFieldValue(f.col)
+ values[i] = freshFieldValueFor(f, start+i)
}
r.Fields = append(r.Fields, model.Field{Name: f.name, Values: values})
}
@@ -200,6 +286,39 @@ func buildResults(spec workloadSpec) []*model.MeasureResult {
return results
}
+// freshTagValueFor builds a brand-new *modelv1.TagValue per call. The svc
+// tag varies by group so GroupBy has bounded cardinality; everything else
+// follows freshTagValue's storage-decode-shaped wrappers.
+func freshTagValueFor(ts tagSpec, group, row int) *modelv1.TagValue {
+ if ts.col == databasev1.TagType_TAG_TYPE_STRING && ts.name == benchSvc {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Str{
+ Str: &modelv1.Str{Value: "svc-" + string(rune('a'+group%26))},
+ }}
+ }
+ if ts.col == databasev1.TagType_TAG_TYPE_STRING {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Str{
+ Str: &modelv1.Str{Value: "c-" + string(rune('a'+(row+group)%26))},
+ }}
+ }
+ return freshTagValue(ts.col)
+}
+
+// freshFieldValueFor varies the numeric value by row so Top/Agg see a
+// non-degenerate distribution; non-numeric fields fall back to the static
+// storage-decode-shaped wrapper.
+func freshFieldValueFor(f fieldSpec, row int) *modelv1.FieldValue {
+ switch f.col {
+ case databasev1.FieldType_FIELD_TYPE_INT:
+ return &modelv1.FieldValue{Value: &modelv1.FieldValue_Int{Int: &modelv1.Int{Value: int64(row)}}}
+ case databasev1.FieldType_FIELD_TYPE_FLOAT:
+ return &modelv1.FieldValue{Value: &modelv1.FieldValue_Float{Float: &modelv1.Float{Value: float64(row)}}}
+ case databasev1.FieldType_FIELD_TYPE_STRING, databasev1.FieldType_FIELD_TYPE_DATA_BINARY,
+ databasev1.FieldType_FIELD_TYPE_UNSPECIFIED:
+ return freshFieldValue(f.col)
+ }
+ return freshFieldValue(f.col)
+}
+
// freshTagValue builds a brand-new *modelv1.TagValue per call. The shape and
// inner alloc count mirror mustDecodeTagValue (wrapper + oneof + inner).
func freshTagValue(t databasev1.TagType) *modelv1.TagValue {
@@ -233,6 +352,8 @@ func freshFieldValue(t databasev1.FieldType) *modelv1.FieldValue {
return &modelv1.FieldValue{Value: &modelv1.FieldValue_Str{Str: &modelv1.Str{Value: "ok"}}}
case databasev1.FieldType_FIELD_TYPE_DATA_BINARY:
return &modelv1.FieldValue{Value: &modelv1.FieldValue_BinaryData{BinaryData: []byte{0xab, 0xcd}}}
+ case databasev1.FieldType_FIELD_TYPE_UNSPECIFIED:
+ return &modelv1.FieldValue{Value: &modelv1.FieldValue_Int{Int: &modelv1.Int{Value: 42}}}
}
return &modelv1.FieldValue{Value: &modelv1.FieldValue_Int{Int: &modelv1.Int{Value: 42}}}
}
@@ -255,9 +376,13 @@ func buildSchema(spec workloadSpec) *databasev1.Measure {
return m
}
-// buildOpts derives MeasureQueryOptions matching the workload's projection.
+// buildOpts derives MeasureQueryOptions matching the workload's projection,
+// including GroupBy/Agg so BuildBatchSchema promotes the referenced columns
+// to native typed columns exactly as the production planner does. For the
+// hidden-tags shape the criteria tag IS materialized (storage-side filter
+// input) but is removed from the *visible* projection by buildVisibleOpts.
func buildOpts(spec workloadSpec) model.MeasureQueryOptions {
- opts := model.MeasureQueryOptions{}
+ opts := model.MeasureQueryOptions{GroupBy: spec.groupBy, Agg: spec.agg}
if len(spec.tagFamilies) > 0 {
names := make([]string, 0, len(spec.tagFamilies))
for _, t := range spec.tagFamilies {
@@ -274,30 +399,341 @@ func buildOpts(spec workloadSpec) model.MeasureQueryOptions {
return opts
}
+// hiddenSet returns the criteria tags that were projected only for
+// storage-side filtering and must be stripped at egress, or an empty set.
+func hiddenSet(spec workloadSpec) logical.HiddenTagSet {
+ h := logical.NewHiddenTagSet()
+ if spec.criteriaTag != "" {
+ h.Add(spec.criteriaTag)
+ }
+ return h
+}
+
// benchSink prevents the compiler from eliding the work inside benchmark
// loops — every drained row is summed into it.
var benchSink int
-// runRowPath drains the row-path serializer once over a fresh cursor backed
-// by the supplied results, accumulating the row count into benchSink.
-func runRowPath(results []*model.MeasureResult, opts model.MeasureQueryOptions) {
- qr := &fakeMeasureQueryResult{seq: results}
- benchSink += len(rowSerialize(qr, opts))
+// Row-path baseline.
+//
+// rowSerialize (diff_test.go) replays the scan/serialize the row path's
+// resultMIterator runs. The Agg/Top/GroupBy reduces below replicate
+// pkg/query/logical/measure faithfully enough for representative cost — they
+// cannot be imported (cycle: logical/measure → this package), exactly as
+// diff_test.go:96-99 documents for rowSerialize. aggregation.* IS imported
+// (cycle-free, the same package the row path uses; see
+// measure_plan_aggregation.go:59,63,84,186,197).
+
+// runRowPath drains the row baseline once for the workload's shape over a
+// fresh cursor backed by cloned results, accumulating into benchSink.
+func runRowPath(spec workloadSpec, results []*model.MeasureResult, opts model.MeasureQueryOptions) {
+ qr := &fakeMeasureQueryResult{seq: cloneResults(results)}
+ dps := rowSerialize(qr, opts)
+ switch spec.shape {
+ case shapeScan:
+ benchSink += len(dps)
+ case shapeHiddenTags:
+ hidden := hiddenSet(spec)
+ for _, idp := range dps {
+ idp.DataPoint.TagFamilies = hidden.StripHiddenTags(idp.DataPoint.TagFamilies)
+ }
+ benchSink += len(dps)
+ case shapeTop:
+ benchSink += len(rowTopReduce(dps, spec))
+ case shapeScalarReduce, shapeGroupByAgg, shapeCountFloat:
+ benchSink += len(rowAggReduce(dps, spec))
+ case shapeRawGroupBy:
+ benchSink += len(rowGroupByFirst(dps, spec))
+ }
+}
+
+// rowFieldIndex finds the projected index of name in opts.FieldProjection,
+// matching the row path's logical.FieldRef.Spec.FieldIdx semantics
+// (DataPoint.Fields are appended in FieldProjection order by rowSerialize).
+func rowFieldIndex(opts model.MeasureQueryOptions, name string) int {
+ for i, f := range opts.FieldProjection {
+ if f == name {
+ return i
+ }
+ }
+ return -1
+}
+
+// rowTopReduce mirrors pkg/query/logical/measure.topOp.Execute: every data
+// point is inserted into one bounded TopQueue keyed on the field value, then
+// drained in sorted order. asc=false → top-N (largest); asc=true → bottom-N.
+func rowTopReduce(dps []*measurev1.InternalDataPoint, spec workloadSpec) []*measurev1.InternalDataPoint {
+ if spec.topN <= 0 {
+ return dps
+ }
+ opts := buildOpts(spec)
+ fieldIdx := rowFieldIndex(opts, spec.topField)
+ h := &rowTopHeap{asc: spec.topAsc}
+ for _, idp := range dps {
+ v := rowNumericFieldValue(idp.GetDataPoint().GetFields(), fieldIdx)
+ el := rowTopEl{idp: idp, val: v, seq: h.seq}
+ h.seq++
+ if h.Len() < spec.topN {
+ heap.Push(h, el)
+ continue
+ }
+ root := h.rows[0]
+ if (spec.topAsc && el.val < root.val) || (!spec.topAsc && el.val > root.val) {
+ h.rows[0] = el
+ heap.Fix(h, 0)
+ }
+ }
+ out := make([]*measurev1.InternalDataPoint, 0, h.Len())
+ for h.Len() > 0 {
+ out = append(out, heap.Pop(h).(rowTopEl).idp)
+ }
+ for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
+ out[i], out[j] = out[j], out[i]
+ }
+ return rowApplyLimit(out, spec.limitN)
+}
+
+func rowNumericFieldValue(fields []*measurev1.DataPoint_Field, idx int) float64 {
+ if idx < 0 || idx >= len(fields) {
+ return 0
+ }
+ switch v := fields[idx].GetValue().GetValue().(type) {
+ case *modelv1.FieldValue_Int:
+ return float64(v.Int.GetValue())
+ case *modelv1.FieldValue_Float:
+ return v.Float.GetValue()
+ }
+ return 0
+}
+
+type rowTopEl struct {
+ idp *measurev1.InternalDataPoint
+ val float64
+ seq int
}
-// runVectorizedPath drains the vectorized adapter once over a fresh cursor.
-func runVectorizedPath(results []*model.MeasureResult, schema *databasev1.Measure,
- opts model.MeasureQueryOptions, cfg VectorizedConfig,
+type rowTopHeap struct {
+ rows []rowTopEl
+ seq int
+ asc bool
+}
+
+func (h *rowTopHeap) Len() int { return len(h.rows) }
+
+func (h *rowTopHeap) Less(i, j int) bool {
+ if h.rows[i].val != h.rows[j].val {
+ if h.asc {
+ return h.rows[i].val > h.rows[j].val
+ }
+ return h.rows[i].val < h.rows[j].val
+ }
+ return h.rows[i].seq > h.rows[j].seq
+}
+
+func (h *rowTopHeap) Swap(i, j int) { h.rows[i], h.rows[j] = h.rows[j], h.rows[i] }
+
+func (h *rowTopHeap) Push(x any) { h.rows = append(h.rows, x.(rowTopEl)) }
+
+func (h *rowTopHeap) Pop() any {
+ n := len(h.rows)
+ x := h.rows[n-1]
+ h.rows = h.rows[:n-1]
+ return x
+}
+
+// rowAggReduce mirrors pkg/query/logical/measure.aggGroupIterator /
+// aggAllIterator: rows are partitioned by the GroupBy key (the whole result
+// when GroupBy is unset → scalar reduce), each group folded through an
+// aggregation.Map dispatched on the field's declared type (FIELD_TYPE_INT →
+// int64, FIELD_TYPE_FLOAT → float64), then emitted one row per group.
+func rowAggReduce(dps []*measurev1.InternalDataPoint, spec workloadSpec) []*measurev1.InternalDataPoint {
+ opts := buildOpts(spec)
+ fieldIdx := rowFieldIndex(opts, spec.agg.FieldName)
+ isFloat := false
+ for _, f := range spec.fields {
+ if f.name == spec.agg.FieldName && f.col == databasev1.FieldType_FIELD_TYPE_FLOAT {
+ isFloat = true
+ }
+ }
+ keyOf := rowGroupKeyFunc(spec)
+ order := make([]string, 0)
+ seen := make(map[string]struct{})
+ intMaps := make(map[string]aggregation.Map[int64])
+ floatMaps := make(map[string]aggregation.Map[float64])
+ tagsOf := make(map[string][]*modelv1.TagFamily)
+ for _, idp := range dps {
+ dp := idp.GetDataPoint()
+ key := keyOf(dp)
+ if _, ok := seen[key]; !ok {
+ seen[key] = struct{}{}
+ order = append(order, key)
+ tagsOf[key] = dp.GetTagFamilies()
+ if isFloat {
+ m, _ := aggregation.NewMap[float64](spec.agg.Func)
+ floatMaps[key] = m
+ } else {
+ m, _ := aggregation.NewMap[int64](spec.agg.Func)
+ intMaps[key] = m
+ }
+ }
+ fv := dp.GetFields()[fieldIdx].GetValue()
+ if isFloat {
+ n, _ := aggregation.FromFieldValue[float64](fv)
+ floatMaps[key].In(n)
+ } else {
+ n, _ := aggregation.FromFieldValue[int64](fv)
+ intMaps[key].In(n)
+ }
+ }
+ out := make([]*measurev1.InternalDataPoint, 0, len(order))
+ for _, key := range order {
+ var val *modelv1.FieldValue
+ if isFloat {
+ val, _ = aggregation.ToFieldValue(floatMaps[key].Val())
+ } else {
+ val, _ = aggregation.ToFieldValue(intMaps[key].Val())
+ }
+ out = append(out, &measurev1.InternalDataPoint{DataPoint: &measurev1.DataPoint{
+ TagFamilies: tagsOf[key],
+ Fields: []*measurev1.DataPoint_Field{{Name: spec.agg.FieldName, Value: val}},
+ }})
+ }
+ return rowApplyLimit(out, spec.limitN)
+}
+
+// rowGroupByFirst mirrors the raw-GroupBy row path: groupIterator yields the
+// whole group, processor.go keeps only current[0], so exactly the first-seen
+// row of each group surfaces in group-insertion order.
+func rowGroupByFirst(dps []*measurev1.InternalDataPoint, spec workloadSpec) []*measurev1.InternalDataPoint {
+ keyOf := rowGroupKeyFunc(spec)
+ seen := make(map[string]struct{})
+ out := make([]*measurev1.InternalDataPoint, 0)
+ for _, idp := range dps {
+ key := keyOf(idp.GetDataPoint())
+ if _, ok := seen[key]; ok {
+ continue
+ }
+ seen[key] = struct{}{}
+ out = append(out, idp)
+ }
+ return rowApplyLimit(out, spec.limitN)
+}
+
+// rowGroupKeyFunc returns a function computing the GroupBy key string from a
+// DataPoint's projected tag families. When GroupBy is unset every row maps to
+// the same key (scalar reduce).
+func rowGroupKeyFunc(spec workloadSpec) func(*measurev1.DataPoint) string {
+ if spec.groupBy == nil || len(spec.groupBy.TagNames) == 0 {
+ return func(*measurev1.DataPoint) string { return "" }
+ }
+ names := spec.groupBy.TagNames
+ family := spec.groupBy.TagFamily
+ return func(dp *measurev1.DataPoint) string {
+ var b []byte
+ for _, tf := range dp.GetTagFamilies() {
+ if tf.GetName() != family {
+ continue
+ }
+ for _, want := range names {
+ for _, tag := range tf.GetTags() {
+ if tag.GetKey() == want {
+ b = append(b, tag.GetValue().GetStr().GetValue()...)
+ b = append(b, 0)
+ }
+ }
+ }
+ }
+ return string(b)
+ }
+}
+
+func rowApplyLimit(in []*measurev1.InternalDataPoint, limitN int) []*measurev1.InternalDataPoint {
+ if limitN > 0 && len(in) > limitN {
+ return in[:limitN]
+ }
+ return in
+}
+
+// Vectorized path — the *production* compute pipeline.
+//
+// Constructed the same way plan.Execute does (executor.go): a
+// vectorized.PipelineBuilder with a shared MemoryTracker, source →
+// [BatchAggregation | BatchTop] → BatchLimit, built, Init'd, then wrapped via
+// NewIteratorFromPipeline and drained through the public adapter. package
+// plan cannot be imported (cycle: plan → measure), so the measure-level
+// operators are wired directly — this is byte-for-byte the operator graph
+// plan.Execute produces for these shapes.
+
+// runVectorizedPath drains the production vec pipeline once over a fresh
+// cursor backed by cloned results.
+func runVectorizedPath(spec workloadSpec, results []*model.MeasureResult,
+ schema *databasev1.Measure, opts model.MeasureQueryOptions, cfg VectorizedConfig,
) {
- qr := &fakeMeasureQueryResult{seq: results}
- it, err := NewMIterator(context.Background(), qr, schema, opts, cfg)
- if err != nil {
- panic(err)
+ qr := &fakeMeasureQueryResult{seq: cloneResults(results)}
+ batchSchema, schemaErr := BuildBatchSchema(schema, opts)
+ if schemaErr != nil {
+ panic(schemaErr)
}
- defer it.Close()
- for it.Next() {
- benchSink++
+ pool := vectorized.NewBatchPool(batchSchema, cfg.BatchSize)
+ source := NewBatchScan(qr, batchSchema, pool, cfg.BatchSize)
+
+ tracker := vectorized.NewMemoryTracker(int64(cfg.QueryMemoryMiB) * 1024 * 1024)
+ builder := vectorized.NewPipelineBuilder().WithMemoryTracker(tracker).From(source)
+
+ terminalSchema := batchSchema
+ switch spec.shape {
+ case shapeScan, shapeHiddenTags:
+ // Schema-preserving scan; no breaker.
+ case shapeTop:
+ fieldIdx, ok := batchSchema.FieldIndex(spec.topField)
+ if !ok {
+ panic("bench: top field not in schema")
+ }
+ builder.Break(NewBatchTop(batchSchema, fieldIdx, spec.topN, spec.topAsc, cfg.BatchSize))
+ case shapeScalarReduce, shapeRawGroupBy, shapeGroupByAgg, shapeCountFloat:
+ ops, opsErr := BuildOperators(opts, batchSchema, tracker, cfg.BatchSize)
+ if opsErr != nil {
+ panic(opsErr)
+ }
+ if len(ops) != 1 {
+ panic("bench: expected exactly one breaker operator")
+ }
+ builder.Break(ops[0])
+ terminalSchema = ops[0].OutputSchema()
}
+ if spec.limitN > 0 {
+ builder.Apply(NewBatchLimit(terminalSchema, 0, uint32(spec.limitN)))
+ }
+
+ pipeline, buildErr := builder.Build()
+ if buildErr != nil {
+ panic(buildErr)
+ }
+ if initErr := pipeline.Init(context.Background()); initErr != nil {
+ panic(initErr)
+ }
+ egressPool := vectorized.NewBatchPool(terminalSchema, cfg.BatchSize)
+ it := NewIteratorFromPipeline(context.Background(), pipeline, egressPool)
+ if spec.shape == shapeHiddenTags {
+ hidden := hiddenSet(spec)
+ for it.Next() {
+ for _, dp := range it.Current() {
+ dp.DataPoint.TagFamilies = hidden.StripHiddenTags(dp.DataPoint.TagFamilies)
+ }
+ benchSink++
+ }
+ } else {
+ for it.Next() {
+ benchSink++
+ }
+ }
+ if closeErr := it.Close(); closeErr != nil {
+ panic(closeErr)
+ }
+}
+
+// cfgFor returns the standard bench VectorizedConfig.
+func cfgFor() VectorizedConfig {
+ return VectorizedConfig{Enabled: true, BatchSize: defaultBatch, QueryMemoryMiB: defaultMemMiB}
}
func benchmarkRow(b *testing.B, spec workloadSpec) {
@@ -306,7 +742,7 @@ func benchmarkRow(b *testing.B, spec workloadSpec) {
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
- runRowPath(results, opts)
+ runRowPath(spec, results, opts)
}
}
@@ -314,15 +750,15 @@ func benchmarkVectorized(b *testing.B, spec workloadSpec) {
results := buildResults(spec)
schema := buildSchema(spec)
opts := buildOpts(spec)
- cfg := VectorizedConfig{Enabled: true, BatchSize: 1024, QueryMemoryMiB: 64}
+ cfg := cfgFor()
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
- runVectorizedPath(results, schema, opts, cfg)
+ runVectorizedPath(spec, results, schema, opts, cfg)
}
}
-// Paired benchmarks per spec.
+// Paired benchmarks per workload.
func BenchmarkRowPath_W1(b *testing.B) { benchmarkRow(b, w1) }
func BenchmarkVectorizedPath_W1(b *testing.B) { benchmarkVectorized(b, w1) }
@@ -339,11 +775,11 @@ func BenchmarkVectorizedPath_W4(b *testing.B) { benchmarkVectorized(b, w4) }
func BenchmarkRowPath_W5(b *testing.B) { benchmarkRow(b, w5) }
func BenchmarkVectorizedPath_W5(b *testing.B) { benchmarkVectorized(b, w5) }
-func BenchmarkRowPath_W2MB(b *testing.B) { benchmarkRow(b, w2mb) }
-func BenchmarkVectorizedPath_W2MB(b *testing.B) { benchmarkVectorized(b, w2mb) }
+func BenchmarkRowPath_W6(b *testing.B) { benchmarkRow(b, w6) }
+func BenchmarkVectorizedPath_W6(b *testing.B) { benchmarkVectorized(b, w6) }
-func BenchmarkRowPath_W4MB(b *testing.B) { benchmarkRow(b, w4mb) }
-func BenchmarkVectorizedPath_W4MB(b *testing.B) { benchmarkVectorized(b, w4mb) }
+func BenchmarkRowPath_W7(b *testing.B) { benchmarkRow(b, w7) }
+func BenchmarkVectorizedPath_W7(b *testing.B) { benchmarkVectorized(b, w7) }
-func BenchmarkRowPath_W5MB(b *testing.B) { benchmarkRow(b, w5mb) }
-func BenchmarkVectorizedPath_W5MB(b *testing.B) { benchmarkVectorized(b, w5mb) }
+func BenchmarkRowPath_W8(b *testing.B) { benchmarkRow(b, w8) }
+func BenchmarkVectorizedPath_W8(b *testing.B) { benchmarkVectorized(b, w8) }