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 12:22:30 UTC
(skywalking-banyandb) branch vectorized-query updated (07adc30db -> 768d67dd9)
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
omit 07adc30db chore(scripts/soak): drop year-glob and remove eval in soak-monitor.sh
omit 2c00450b7 perf(query/vectorized/measure): zero-alloc egress on nil Selection + reuse null singletons
omit ad96b5aa9 fix(query/vectorized): correct Apply-after-Break stage ordering in PipelineBuilder
omit d61e73dc2 Merge remote-tracking branch 'origin/main' into vectorized-query
omit 475abf206 docs(changelog): note GroupBy+Agg coverage in vectorized measure entry
omit a59dabeef fix(banyand/query): fall through to row path on emitPartial GroupBy+Agg requests
omit 32fb720a3 feat(query/vectorized/measure)!: remove AggregationEnabled flag — GroupBy+Agg dispatch is default-on
omit b750fa827 chore: fix lint issues
omit 4b54f1d4f test(query/integration): flip --measure-vectorized-aggregation-enabled=true in parity gate (G8d.3)
omit 40fcf37b7 feat(query/vectorized/measure): BatchAggregation row-path parity for MEAN type and non-key projected tags (G8d.3)
omit 348c09c29 fix(query/vectorized/measure/plan): thread GroupBy/Agg into Scan schema so vec/storage column types agree (G8d.3)
omit 1cd7e768c docs(test/integration): record GroupBy+Agg bridge bug behind vec aggregation flag (G8d.3)
omit 7c0d9259f fix(query/vectorized/measure): agg output column inherits input field name for row-path parity (G8d.3 partial)
omit 0b079e752 feat(banyand/measure): plumb --measure-vectorized-aggregation-enabled CLI flag (G8d.2)
omit ad9941657 feat(query/vectorized/measure/plan): flag-gated GroupBy+Agg dispatch (G8d.2)
omit 253c56538 feat(query/vectorized/measure): BuildBatchSchema emits native columns for GroupBy keys + Agg field (G8d.2)
omit 7117d9fa5 test(query/integration): assert vec dispatch fires in parity gate (G8e)
omit c4e3ea866 feat(query/vectorized/measure/plan): dispatch eligibility hardening + observability counters (G8e)
omit cf44971fd feat(banyand/query): wire vec dispatch into executeMeasurePlan (G8d.2)
omit d71bc8890 feat(query/vectorized/measure/plan): vec dispatch entry point (G8d.1)
omit bb1f75141 feat(query/vectorized/measure/plan): vec executor (G8c)
omit 84cdf6b8b feat(query/vectorized/measure): expose NewIteratorFromPipeline
omit e55a6fa60 chore(query/measure): broaden row-path deprecation to all primary plan types
omit 89900e661 feat(query/vectorized/measure/plan): vec analyzer (G8b)
omit bd9f9a08c feat(query/vectorized/measure/plan): VecPlan tree scaffolding (G8a)
omit e2f30bda6 feat(query/vectorized): expose Pipeline.Init for multi-stage cascades
omit d420fa893 chore(query/measure): mark row-path Analyze/DistributedAnalyze as deprecated
omit 8008aa30f Revert "feat(query/vectorized/measure): wire aggregation planner into NewMIterator (G7e)"
omit 275866aca feat(query/vectorized/measure): wire aggregation planner into NewMIterator (G7e)
omit cb61bca97 feat(query/measure): MeasureGroupBy/Agg options and BuildOperators planner (G7d)
omit 4a0479e86 feat(query/vectorized): shared per-pipeline MemoryTracker (G7a)
omit 89c3119ff chore(measure): fix stale G5e config test and lint-blocking typos
omit a0dc29e95 feat(measure): flip vectorized query path to enabled by default (G5e)
omit 48c0f76be docs(soak): G5d Phase D summary
omit a9b77692f fix(soak): monitor tolerates missing Phase 0 files
omit 0c79de2ab feat(soak): tapered monitor for unattended G5d runs
omit 90aacda91 feat(banyand/measure): typed PullBatch for indexSortResult (closes US-003 multi-block deferral)
omit 141aee43a docs(test): add CLAUDE.md with G5d soak-harness lessons
omit abedc3f06 feat(soak): G5d soak harness — docker-compose + soak-driver + runbook
omit 36140ca94 feat(banyand/measure): batch-aware multi-block merge for PullBatch (US-003 strict completion)
omit d2b877ef3 test(banyand/measure): storage-level bench for block_cursor decode (passthrough vs native)
omit 3560a960a test(query/vectorized/measure): realistic per-cell bench fixture + multi-block workloads
omit b640c51a8 docs(query/vectorized/measure): pin passthrough as production type + drop redundant deep-copy in batch source
omit b2d3c0802 feat(query/vectorized/measure): wire NewMIterator through MeasureBatchResult.PullBatch (US-007 partial)
omit 55d3baf40 feat(banyand/measure): native typed-column emit on blockCursor (US-003 strict, single-block fast path)
omit ba6e82787 feat(query/vectorized/measure): native typed-column dispatch in BuildMeasureBatchFromResult (G5b/G5c partial)
omit a421813e1 feat(query/model): add MeasureBatchResult interface; storage dual-emits (G5b foundation)
omit 5de3e0eb2 perf(query/vectorized/measure): close G5a gap via passthrough columns + precomputed schema layout
omit 2614f0981 feat(query/vectorized/measure): add bench gates infrastructure (gate G5a)
omit 0e0638741 test(integration): restore SharedContext in vectorized parity AfterAll
omit fe9d38eb3 feat(query/vectorized/measure): add integration & parity (gate G4)
omit 11d0f9884 chore: fix lint issues in vectorized/measure operators
omit 278b29490 feat(query/vectorized/measure): add operators + output (gate G3)
omit 4e432ecb1 chore: fix lint issues in vectorized/measure package
omit 3755984c4 feat(query/vectorized/measure): add measure scan layer (gate G2)
omit 3711bcee7 chore: fix lint issues in vectorized package
omit b39ed477b feat(query/vectorized): add vectorized pipeline foundation (gate G1)
add 2804bdf87 feat: vectorized measure query subsystem (scan + GroupBy+Agg) (#1129)
add af5970f0c fix(queue/pub): keep pub-node-probe alive after caller ctx cancel (#1130)
new 8f21383d9 feat(query/vectorized/measure/plan): resolve order_by in vec dispatch
new fedf8cf43 feat(query/vectorized/measure/plan): G9c boundary-error parity (nil TimeRange, projection, empty result)
new 02d0c1940 feat(query/vectorized/measure/plan): G9d hidden criteria tags egress-strip wrapper
new cadfec0bf feat(query/vectorized/measure/plan): G9b GroupBy/Agg completeness (scalar reduce, raw GroupBy, projection auto-coverage)
new 9128d9fc2 feat(query/vectorized/measure/plan): G9a single-node Top dispatch + analyzer sort-direction fix
new 2cb0ddcfe fix(query/vectorized/measure): COUNT preserves input type for row-path parity (float_top_count)
new 087624546 chore(query/vectorized/measure/plan): fix goconst/dogsled lint in G9 test files
new 768d67dd9 test(query/vectorized/measure/plan): rewrite stale G9b fall-through tests + clarify empty-result hidden-strip skip
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (07adc30db)
\
N -- N -- N refs/heads/vectorized-query (768d67dd9)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 8 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:
CHANGES.md | 2 +-
banyand/queue/pub/client.go | 8 +-
pkg/query/vectorized/measure/aggregation.go | 36 +-
pkg/query/vectorized/measure/groupby.go | 33 +-
pkg/query/vectorized/measure/plan.go | 46 +-
pkg/query/vectorized/measure/plan/analyzer.go | 135 ++++--
pkg/query/vectorized/measure/plan/analyzer_test.go | 143 +++++--
pkg/query/vectorized/measure/plan/build_test.go | 16 +-
pkg/query/vectorized/measure/plan/dispatch.go | 278 ++++++-------
pkg/query/vectorized/measure/plan/dispatch_test.go | 463 ++++++++++++++++-----
pkg/query/vectorized/measure/plan/executor_test.go | 123 +++++-
pkg/query/vectorized/measure/plan/groupby_agg.go | 39 +-
pkg/query/vectorized/measure/plan/hidden_tags.go | 94 +++++
pkg/query/vectorized/measure/plan/top.go | 8 +-
pkg/query/vectorized/measure/plan_test.go | 44 +-
pkg/query/vectorized/measure/top.go | 54 ++-
pkg/query/vectorized/measure/top_test.go | 44 ++
.../data/input/{tag_filter.ql => empty_result.ql} | 2 +-
...ter_non_existent_tag.yaml => empty_result.yaml} | 4 +-
...er_tag_asc.ql => project_non_existent_field.ql} | 4 +-
...nt_tag.yaml => project_non_existent_field.yaml} | 4 +-
test/cases/measure/measure.go | 2 +
.../standalone/query/vectorized_test.go | 10 +-
23 files changed, 1175 insertions(+), 417 deletions(-)
create mode 100644 pkg/query/vectorized/measure/plan/hidden_tags.go
copy test/cases/measure/data/input/{tag_filter.ql => empty_result.ql} (96%)
copy test/cases/measure/data/input/{filter_non_existent_tag.yaml => empty_result.yaml} (94%)
copy test/cases/measure/data/input/{order_tag_asc.ql => project_non_existent_field.ql} (91%)
copy test/cases/measure/data/input/{project_non_existent_tag.yaml => project_non_existent_field.yaml} (93%)
(skywalking-banyandb) 08/08: test(query/vectorized/measure/plan): rewrite stale G9b fall-through tests + clarify empty-result hidden-strip skip
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 768d67dd9a49617493c3e4c5e88a5ceff0aac147
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 10:34:04 2026 +0000
test(query/vectorized/measure/plan): rewrite stale G9b fall-through tests + clarify empty-result hidden-strip skip
---
pkg/query/vectorized/measure/plan/dispatch.go | 4 +-
pkg/query/vectorized/measure/plan/dispatch_test.go | 88 ++++++++++++++--------
2 files changed, 61 insertions(+), 31 deletions(-)
diff --git a/pkg/query/vectorized/measure/plan/dispatch.go b/pkg/query/vectorized/measure/plan/dispatch.go
index c1864a21e..6ea641ca4 100644
--- a/pkg/query/vectorized/measure/plan/dispatch.go
+++ b/pkg/query/vectorized/measure/plan/dispatch.go
@@ -256,7 +256,9 @@ func Dispatch(
// Next()==false immediately and Close()==nil, so the client
// observes an empty []*measurev1.InternalDataPoint. Emit the same
// empty MIterator directly with handled=true instead of borrowing
- // the row machinery.
+ // the row machinery. Hidden-tag egress strip (below) is
+ // intentionally skipped here: an empty result has no DataPoints,
+ // so there is nothing to strip.
return emptyMIterator{}, p.String(), true, nil
}
diff --git a/pkg/query/vectorized/measure/plan/dispatch_test.go b/pkg/query/vectorized/measure/plan/dispatch_test.go
index 3b9fcf0ec..f62b796f4 100644
--- a/pkg/query/vectorized/measure/plan/dispatch_test.go
+++ b/pkg/query/vectorized/measure/plan/dispatch_test.go
@@ -25,6 +25,7 @@ import (
"google.golang.org/protobuf/types/known/timestamppb"
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"
@@ -66,54 +67,77 @@ func TestDispatch_NotEnabled_FallsThrough(t *testing.T) {
}
}
-// TestDispatch_GroupByWithoutAgg_FallsThrough covers the pair check
-// (GroupBy and Agg must travel together).
-func TestDispatch_GroupByWithoutAgg_FallsThrough(t *testing.T) {
+// dispatchSchemaFixture builds the (measureSchema, logicalSchema,
+// metadata, fakeEC) tuple the post-G9 ReachesEcQuery tests share. fakeEC
+// returns (nil, nil) so dispatch falls through after ec.Query (empty
+// result); what each test asserts is that ec.Query was reached at all,
+// proving the eligibility gate admitted the request.
+func dispatchSchemaFixture(t *testing.T) (*databasev1.Measure, logical.Schema, *commonv1.Metadata, *fakeEC) {
+ t.Helper()
+ measureSchema := testMeasureSchema()
+ // nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
+ logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ return measureSchema, logicalSchema, &commonv1.Metadata{Name: "demo", Group: "default"}, &fakeEC{}
+}
+
+// TestDispatch_RawGroupBy_ReachesEcQuery confirms G9b: GroupBy without
+// Agg (raw grouping) is now handled by the vec subsystem rather than
+// falling through to the row path.
+func TestDispatch_RawGroupBy_ReachesEcQuery(t *testing.T) {
+ measureSchema, logicalSchema, metadata, ec := dispatchSchemaFixture(t)
+
req := bareReq()
req.GroupBy = &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
FieldName: fieldValue,
}
_, _, handled, err := Dispatch(context.Background(),
- req, nil, nil, nil, nil, dispatchCfg(true))
+ req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
if err != nil {
- t.Fatalf("GroupBy-without-Agg fallthrough must not error: %v", err)
+ t.Fatalf("raw GroupBy must not error before ec.Query: %v", err)
}
- if handled {
- t.Fatal("GroupBy without Agg must fall through to row path")
+ if !ec.called {
+ t.Fatal("raw GroupBy (no Agg) must reach ec.Query post-G9b, not fall through")
+ }
+ if !handled {
+ t.Fatal("raw GroupBy reached ec.Query (empty result) — dispatch must report handled=true")
}
}
-// TestDispatch_AggWithoutGroupBy_FallsThrough is the Agg-only counterpart.
-// Scalar reduce is deferred; Agg without GroupBy falls through.
-func TestDispatch_AggWithoutGroupBy_FallsThrough(t *testing.T) {
+// TestDispatch_ScalarReduce_ReachesEcQuery confirms G9b: Agg without
+// GroupBy (scalar reduce) is now handled by the vec subsystem.
+func TestDispatch_ScalarReduce_ReachesEcQuery(t *testing.T) {
+ measureSchema, logicalSchema, metadata, ec := dispatchSchemaFixture(t)
+
req := bareReq()
req.Agg = &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
FieldName: fieldValue,
}
_, _, handled, err := Dispatch(context.Background(),
- req, nil, nil, nil, nil, dispatchCfg(true))
+ req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
if err != nil {
- t.Fatalf("Agg-without-GroupBy fallthrough must not error: %v", err)
+ t.Fatalf("scalar reduce must not error before ec.Query: %v", err)
}
- if handled {
- t.Fatal("Agg without GroupBy must fall through to row path")
+ if !ec.called {
+ t.Fatal("scalar reduce (Agg, no GroupBy) must reach ec.Query post-G9b, not fall through")
+ }
+ if !handled {
+ t.Fatal("scalar reduce reached ec.Query (empty result) — dispatch must report handled=true")
}
}
-// TestDispatch_GroupByAggUncoveredProjection_FallsThrough covers the
-// G8d.2 projection-coverage gate. BatchAggregation locates its key +
-// value columns by name inside the BatchSchema, which is built from
-// TagProjection + FieldProjection. When GroupBy keys or the Agg field
-// are not in the request's projection, dispatch falls through so the
-// row path can either extend projection implicitly or surface its
-// canonical error.
-func TestDispatch_GroupByAggUncoveredProjection_FallsThrough(t *testing.T) {
+// TestDispatch_GroupByAggUncoveredProjection_ReachesEcQuery confirms
+// G9b's projection auto-coverage: when GroupBy keys or the Agg field are
+// absent from the request's projection, plan.Analyze extends the
+// projection so the request is admitted instead of falling through.
+func TestDispatch_GroupByAggUncoveredProjection_ReachesEcQuery(t *testing.T) {
cases := []struct {
- name string
- mutate func(*measurev1.QueryRequest)
- comment string
+ mutate func(*measurev1.QueryRequest)
+ name string
}{
{
name: "groupby_tag_not_in_projection",
@@ -142,22 +166,26 @@ func TestDispatch_GroupByAggUncoveredProjection_FallsThrough(t *testing.T) {
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
FieldName: fieldValue,
}
- // Strip the value field from FieldProjection so the Agg field is uncovered.
+ // Strip the value field from FieldProjection; G9b auto-coverage re-adds it.
req.FieldProjection = nil
},
},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
+ measureSchema, logicalSchema, metadata, ec := dispatchSchemaFixture(t)
req := bareReq()
c.mutate(req)
_, _, handled, err := Dispatch(context.Background(),
- req, nil, nil, nil, nil, dispatchCfg(true))
+ req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
+ if err == nil && !handled {
+ t.Fatal("auto-covered projection must reach ec.Query (handled=true), not fall through")
+ }
if err != nil {
- t.Fatalf("uncovered projection must not error: %v", err)
+ t.Fatalf("auto-covered projection must not error: %v", err)
}
- if handled {
- t.Fatal("uncovered GroupBy / Agg projection must fall through")
+ if !ec.called {
+ t.Fatal("uncovered GroupBy/Agg projection must reach ec.Query post-G9b (auto-coverage)")
}
})
}
(skywalking-banyandb) 02/08: feat(query/vectorized/measure/plan): G9c boundary-error parity (nil TimeRange, projection, empty result)
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 fedf8cf43a9ba59088d71e75ed407b09f51e50dc
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 10:04:56 2026 +0000
feat(query/vectorized/measure/plan): G9c boundary-error parity (nil TimeRange, projection, empty result)
---
pkg/query/vectorized/measure/plan/dispatch.go | 107 +++++++++------
pkg/query/vectorized/measure/plan/dispatch_test.go | 152 +++++++++++++++------
test/cases/measure/data/input/empty_result.ql | 21 +++
test/cases/measure/data/input/empty_result.yaml | 32 +++++
.../data/input/project_non_existent_field.ql | 19 +++
.../data/input/project_non_existent_field.yaml | 25 ++++
test/cases/measure/measure.go | 2 +
7 files changed, 277 insertions(+), 81 deletions(-)
diff --git a/pkg/query/vectorized/measure/plan/dispatch.go b/pkg/query/vectorized/measure/plan/dispatch.go
index 3ada53ef1..baa870a2b 100644
--- a/pkg/query/vectorized/measure/plan/dispatch.go
+++ b/pkg/query/vectorized/measure/plan/dispatch.go
@@ -22,6 +22,8 @@ import (
"fmt"
"sync/atomic"
+ "github.com/pkg/errors"
+
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"
@@ -146,9 +148,16 @@ func Dispatch(
return nil, "", false, nil
}
}
- if req.GetTimeRange() == nil {
- return nil, "", false, nil
- }
+ // G9c #9: a nil TimeRange is NOT a fall-through. The row path does not
+ // reject it — parseFields feeds criteria.GetTimeRange().GetBegin().AsTime()
+ // into the index scan, and a nil *timestamppb.Timestamp resolves to the
+ // Unix epoch (1970-01-01T00:00:00Z). The query then runs over the
+ // degenerate [epoch, epoch] window and yields an empty result. The
+ // scan.Params.TimeRange construction below mirrors that exactly
+ // (req.GetTimeRange().GetBegin().AsTime() == epoch when TimeRange is
+ // nil), so dispatch produces the row path's canonical empty response
+ // directly instead of borrowing it.
+
// Defensive nil guards on the runtime context. These should not fire
// in production paths — buildMeasureContext populates all of them —
// but a defensive fallthrough is safer than a nil dereference.
@@ -156,13 +165,14 @@ func Dispatch(
return nil, "", false, nil
}
- // Projection validation. The row path's Analyze rejects unknown
- // projection names via ValidateProjectionTags / ValidateProjectionFields
- // and surfaces a descriptive error. Dispatch falls through so the
- // row path produces that canonical error (test fixtures with
- // WantErr=true depend on it).
- if !projectionsExistInSchema(req, measureSchema) {
- return nil, "", false, nil
+ // G9c #11: projection validation. The row path's Analyze rejects
+ // unknown projection names via ValidateProjectionTags /
+ // ValidateProjectionFields and surfaces a descriptive error
+ // (test fixtures with WantErr=true assert it). Dispatch reproduces
+ // that canonical error byte-for-byte and returns handled=true so the
+ // caller surfaces it rather than falling through.
+ if projErr := validateProjectionParity(req, logicalSchema, measureSchema); projErr != nil {
+ return nil, "", true, projErr
}
// Hidden-tag detection: criteria may reference tags that are NOT in
@@ -250,10 +260,13 @@ func Dispatch(
return nil, "", true, fmt.Errorf("vec dispatch: query measure: %w", queryErr)
}
if result == nil {
- // Match the row path's typed-nil handling: an empty query result
- // flows through the row iterator as a no-op. Falling back lets
- // that machinery surface the empty response unchanged.
- return nil, "", false, nil
+ // G9c #13: a typed-nil result is the row path's canonical empty
+ // response. The row iterator (resultMIterator{result: nil}) reports
+ // Next()==false immediately and Close()==nil, so the client
+ // observes an empty []*measurev1.InternalDataPoint. Emit the same
+ // empty MIterator directly with handled=true instead of borrowing
+ // the row machinery.
+ return emptyMIterator{}, p.String(), true, nil
}
pool := vectorized.NewBatchPool(scan.BatchSchema, cfg.BatchSize)
@@ -329,54 +342,64 @@ func projectedNames(tp *modelv1.TagProjection) map[string]struct{} {
return out
}
-// projectionsExistInSchema returns false if any tag (in any requested tag
-// family) or field name in the request's projection is absent from the
-// Measure schema. Callers use the result as an eligibility gate: missing
-// names route through the row path, which surfaces a descriptive error
-// via logical_measure.Analyze.
-func projectionsExistInSchema(req *measurev1.QueryRequest, m *databasev1.Measure) bool {
+// validateProjectionParity reproduces, byte-for-byte, the projection
+// errors the row path's logical_measure.Analyze raises so dispatch can
+// surface the canonical WantErr=true message directly instead of falling
+// through. It mirrors the row path exactly:
+//
+// - Tags are validated before fields (measure_analyzer.go:110-119).
+// - Tag projection is checked only when non-empty; each projected tag
+// (families in order, tags in order) is looked up schema-wide via the
+// TagSpec registry — the logical.Schema equivalent of CommonSchema's
+// TagSpecMap. The first miss returns errors.Wrap(ErrTagNotDefined,
+// tagName), identical to CommonSchema.ValidateProjectionTags
+// (schema.go:175): "<tagName>: tag is not defined".
+// - Field projection is checked only when non-empty; the first name
+// absent from the Measure schema's fields returns errors.Errorf(
+// "field %s not found in schema", field), identical to
+// measure.schema.ValidateProjectionFields (measure/schema.go:77).
+//
+// A nil error means every projected name resolves, so dispatch proceeds.
+func validateProjectionParity(req *measurev1.QueryRequest, logicalSchema logical.Schema, m *databasev1.Measure) error {
if tp := req.GetTagProjection(); tp != nil {
for _, reqFamily := range tp.GetTagFamilies() {
- schemaFamily := findSchemaTagFamily(m, reqFamily.GetName())
- if schemaFamily == nil {
- return false
- }
- known := make(map[string]struct{}, len(schemaFamily.GetTags()))
- for _, ts := range schemaFamily.GetTags() {
- known[ts.GetName()] = struct{}{}
- }
for _, name := range reqFamily.GetTags() {
- if _, ok := known[name]; !ok {
- return false
+ if logicalSchema.FindTagSpecByName(name) == nil {
+ return errors.Wrap(logical.ErrTagNotDefined, name)
}
}
}
}
if fp := req.GetFieldProjection(); fp != nil && len(fp.GetNames()) > 0 {
+ // The row path's m.fieldMap is built from md.GetFields() in
+ // logical_measure.BuildSchema, so the Measure schema's field set
+ // is the authoritative lookup the row path's
+ // ValidateProjectionFields consults.
known := make(map[string]struct{}, len(m.GetFields()))
for _, fs := range m.GetFields() {
known[fs.GetName()] = struct{}{}
}
for _, name := range fp.GetNames() {
if _, ok := known[name]; !ok {
- return false
+ return errors.Errorf("field %s not found in schema", name)
}
}
}
- return true
-}
-
-// findSchemaTagFamily returns the schema-defined tag family with the
-// given name, or nil if no such family exists.
-func findSchemaTagFamily(m *databasev1.Measure, name string) *databasev1.TagFamilySpec {
- for _, tf := range m.GetTagFamilies() {
- if tf.GetName() == name {
- return tf
- }
- }
return nil
}
+// emptyMIterator is the vec equivalent of the row path's
+// resultMIterator{result: nil}: Next reports no rows, Current is never
+// reached, and Close is a no-op error. Dispatch returns it for the
+// canonical empty response (G9c #13).
+type emptyMIterator struct{}
+
+func (emptyMIterator) Next() bool { return false }
+
+func (emptyMIterator) Current() []*measurev1.InternalDataPoint { return nil }
+
+func (emptyMIterator) Close() error { return nil }
+
// aggProjectionCoverage reports whether the request's GroupBy keys and
// Agg field are all present in the request's projections. Required by
// the dispatch eligibility gate: the BatchAggregation operator locates
diff --git a/pkg/query/vectorized/measure/plan/dispatch_test.go b/pkg/query/vectorized/measure/plan/dispatch_test.go
index d83a29191..08f9dcb6c 100644
--- a/pkg/query/vectorized/measure/plan/dispatch_test.go
+++ b/pkg/query/vectorized/measure/plan/dispatch_test.go
@@ -246,11 +246,12 @@ func TestDispatch_OrderBy_UnknownIndexRule_BubblesUpError(t *testing.T) {
}
}
-// TestDispatch_UnknownTagProjection_FallsThrough covers the parity gap
-// for WantErr=true fixtures: the row path rejects unknown tags via
-// ValidateProjectionTags and returns a descriptive error. Dispatch
-// falls through so the row path surfaces that canonical error.
-func TestDispatch_UnknownTagProjection_FallsThrough(t *testing.T) {
+// TestDispatch_UnknownTagProjection_SurfacesCanonicalError covers G9c
+// #11: the row path rejects unknown tags via ValidateProjectionTags with
+// errors.Wrap(ErrTagNotDefined, tagName). Dispatch reproduces that exact
+// message and returns handled=true so the caller surfaces it (the
+// WantErr=true fixtures depend on it) instead of borrowing the row path.
+func TestDispatch_UnknownTagProjection_SurfacesCanonicalError(t *testing.T) {
measureSchema := testMeasureSchema()
// nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
@@ -266,20 +267,27 @@ func TestDispatch_UnknownTagProjection_FallsThrough(t *testing.T) {
}}
_, _, handled, err := Dispatch(context.Background(),
req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
- if err != nil {
- t.Fatalf("unknown tag fallthrough must not error: %v", err)
+ if err == nil {
+ t.Fatal("unknown tag in projection must surface the canonical row-path error")
}
- if handled {
- t.Fatal("unknown tag in projection must fall through (row path returns WantErr)")
+ // Byte-identical to logical.CommonSchema.ValidateProjectionTags
+ // (pkg/query/logical/schema.go:175): errors.Wrap(ErrTagNotDefined, "ghost").
+ const wantMsg = "ghost: tag is not defined"
+ if err.Error() != wantMsg {
+ t.Fatalf("error message parity: want %q, got %q", wantMsg, err.Error())
+ }
+ if !handled {
+ t.Fatal("projection error must report handled=true so caller surfaces it (no row-path retry)")
}
if ec.called {
t.Fatal("ec.Query must not be invoked when projection is invalid")
}
}
-// TestDispatch_UnknownFieldProjection_FallsThrough is the field-side
-// counterpart of UnknownTagProjection.
-func TestDispatch_UnknownFieldProjection_FallsThrough(t *testing.T) {
+// TestDispatch_UnknownFieldProjection_SurfacesCanonicalError is the
+// field-side counterpart: byte-identical to
+// measure.schema.ValidateProjectionFields (measure/schema.go:77).
+func TestDispatch_UnknownFieldProjection_SurfacesCanonicalError(t *testing.T) {
measureSchema := testMeasureSchema()
// nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
@@ -293,29 +301,84 @@ func TestDispatch_UnknownFieldProjection_FallsThrough(t *testing.T) {
req.FieldProjection = &measurev1.QueryRequest_FieldProjection{Names: []string{"ghost"}}
_, _, handled, err := Dispatch(context.Background(),
req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
- if err != nil {
- t.Fatalf("unknown field fallthrough must not error: %v", err)
+ if err == nil {
+ t.Fatal("unknown field in projection must surface the canonical row-path error")
}
- if handled {
- t.Fatal("unknown field in projection must fall through (row path returns WantErr)")
+ const wantMsg = "field ghost not found in schema"
+ if err.Error() != wantMsg {
+ t.Fatalf("error message parity: want %q, got %q", wantMsg, err.Error())
+ }
+ if !handled {
+ t.Fatal("projection error must report handled=true so caller surfaces it (no row-path retry)")
}
if ec.called {
t.Fatal("ec.Query must not be invoked when projection is invalid")
}
}
-// TestDispatch_NoTimeRange_FallsThrough covers the bounded-window
-// requirement.
-func TestDispatch_NoTimeRange_FallsThrough(t *testing.T) {
+// TestDispatch_TagValidatedBeforeField mirrors the row path's ordering
+// (measure_analyzer.go validates tags before fields): when both
+// projections are unknown, the tag error wins.
+func TestDispatch_TagValidatedBeforeField(t *testing.T) {
+ measureSchema := testMeasureSchema()
+ // nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
+ logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ metadata := &commonv1.Metadata{Name: "demo", Group: "default"}
+ ec := &fakeEC{}
+
+ req := bareReq()
+ req.TagProjection = &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{
+ {Name: "default", Tags: []string{"ghost"}},
+ }}
+ req.FieldProjection = &measurev1.QueryRequest_FieldProjection{Names: []string{"phantom"}}
+ _, _, _, err := Dispatch(context.Background(),
+ req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
+ if err == nil || err.Error() != "ghost: tag is not defined" {
+ t.Fatalf("tag error must take precedence over field error; got %v", err)
+ }
+}
+
+// TestDispatch_NoTimeRange_EmptyResultParity covers G9c #9: a nil
+// TimeRange is NOT rejected by the row path. parseFields feeds
+// criteria.GetTimeRange().GetBegin().AsTime() (nil → Unix epoch) into the
+// scan, the query runs over [epoch, epoch], and the client observes an
+// empty response. Dispatch reproduces that exact behavior directly:
+// ec.Query is invoked (over the epoch window) and the canonical empty
+// MIterator is emitted with handled=true.
+func TestDispatch_NoTimeRange_EmptyResultParity(t *testing.T) {
+ measureSchema := testMeasureSchema()
+ // nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
+ logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ metadata := &commonv1.Metadata{Name: "demo", Group: "default"}
+ ec := &fakeEC{wantResult: nil, wantErr: nil}
+
req := bareReq()
req.TimeRange = nil
- _, _, handled, err := Dispatch(context.Background(),
- req, nil, nil, nil, nil, dispatchCfg(true))
+ iter, _, handled, err := Dispatch(context.Background(),
+ req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
if err != nil {
- t.Fatalf("no-TimeRange fallthrough must not error: %v", err)
+ t.Fatalf("nil TimeRange must not error (row path does not reject it): %v", err)
}
- if handled {
- t.Fatal("missing TimeRange must fall through")
+ if !handled {
+ t.Fatal("nil TimeRange must be handled (row path produces an empty result, not a fall-through)")
+ }
+ if !ec.called {
+ t.Fatal("ec.Query must be invoked over the epoch window (row-path parity)")
+ }
+ if iter == nil {
+ t.Fatal("expected an empty MIterator, got nil")
+ }
+ if iter.Next() {
+ t.Fatal("empty-result MIterator must report Next()==false")
+ }
+ if closeErr := iter.Close(); closeErr != nil {
+ t.Fatalf("empty-result MIterator Close must be nil (row-path parity): %v", closeErr)
}
}
@@ -348,13 +411,15 @@ func (f *fakeEC) Query(_ context.Context, opts model.MeasureQueryOptions) (model
return f.wantResult, f.wantErr
}
-// TestDispatch_EmptyResult_FallsThrough exercises the full eligibility
-// path: an eligible request reaches ec.Query, ec returns (nil, nil)
-// (empty range), Dispatch reports fallthrough so the row path can surface
-// the empty response. This also confirms the index.Query construction
-// and Analyze invocation complete without error against a real
-// logical.Schema.
-func TestDispatch_EmptyResult_FallsThrough(t *testing.T) {
+// TestDispatch_EmptyResult_CanonicalEmptyIterator covers G9c #13: an
+// eligible request reaches ec.Query, ec returns (nil, nil) (the row
+// path's typed-nil empty result). Dispatch emits the canonical empty
+// MIterator (Next()==false, Close()==nil) with handled=true — the same
+// empty []*measurev1.InternalDataPoint the row iterator
+// (resultMIterator{result: nil}) would surface. This also confirms the
+// index.Query construction and Analyze invocation complete without error
+// against a real logical.Schema.
+func TestDispatch_EmptyResult_CanonicalEmptyIterator(t *testing.T) {
measureSchema := testMeasureSchema()
// nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
@@ -364,19 +429,25 @@ func TestDispatch_EmptyResult_FallsThrough(t *testing.T) {
metadata := &commonv1.Metadata{Name: "demo", Group: "default"}
ec := &fakeEC{wantResult: nil, wantErr: nil}
- iter, planStr, handled, err := Dispatch(context.Background(),
+ iter, _, handled, err := Dispatch(context.Background(),
bareReq(), metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
if err != nil {
t.Fatalf("dispatch must not error on empty result: %v", err)
}
- if handled {
- t.Fatal("empty result must fall through to row path")
+ if !handled {
+ t.Fatal("empty result must be handled (canonical empty response, not a fall-through)")
}
- if iter != nil || planStr != "" {
- t.Fatalf("expect zero outputs on fallthrough, got iter=%v planStr=%q", iter, planStr)
+ if iter == nil {
+ t.Fatal("expected an empty MIterator, got nil")
+ }
+ if iter.Next() {
+ t.Fatal("empty-result MIterator must report Next()==false")
+ }
+ if closeErr := iter.Close(); closeErr != nil {
+ t.Fatalf("empty-result MIterator Close must be nil (row-path parity): %v", closeErr)
}
if !ec.called {
- t.Fatal("ec.Query must be invoked before fallthrough decision")
+ t.Fatal("ec.Query must be invoked before the empty-result decision")
}
if ec.lastOpts.Name != "demo" {
t.Fatalf("opts.Name: want demo, got %q", ec.lastOpts.Name)
@@ -395,8 +466,11 @@ func TestDispatch_Counters_TrackFellThroughCalls(t *testing.T) {
startHandled := HandledCount()
startFellThrough := FellThroughCount()
- // Three fallthroughs of distinct shapes, each tripping a different
- // gate so the counter is exercised across the eligibility branches.
+ // Three fallthroughs of distinct shapes. topReq trips the Top gate;
+ // the other two trip the PERMANENT nil-runtime-context guard (all-nil
+ // schema/ec/metadata). Note: post-G9c a nil TimeRange is no longer a
+ // fall-through on its own — noTimeReq falls through here only because
+ // the runtime context is nil.
topReq := bareReq()
topReq.Top = &measurev1.QueryRequest_Top{Number: 5, FieldName: "value"}
noTimeReq := bareReq()
diff --git a/test/cases/measure/data/input/empty_result.ql b/test/cases/measure/data/input/empty_result.ql
new file mode 100644
index 000000000..1ba070c8c
--- /dev/null
+++ b/test/cases/measure/data/input/empty_result.ql
@@ -0,0 +1,21 @@
+# 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.
+
+
+SELECT id, total::field, value::field FROM MEASURE service_cpm_minute IN sw_metric
+TIME > '-15m'
+WHERE id = 'no_such_id_xyz'
diff --git a/test/cases/measure/data/input/empty_result.yaml b/test/cases/measure/data/input/empty_result.yaml
new file mode 100644
index 000000000..48b91206e
--- /dev/null
+++ b/test/cases/measure/data/input/empty_result.yaml
@@ -0,0 +1,32 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: "service_cpm_minute"
+groups: ["sw_metric"]
+tagProjection:
+ tagFamilies:
+ - name: "default"
+ tags: ["id"]
+fieldProjection:
+ names: ["total", "value"]
+criteria:
+ condition:
+ name: "id"
+ op: "BINARY_OP_EQ"
+ value:
+ str:
+ value: "no_such_id_xyz"
diff --git a/test/cases/measure/data/input/project_non_existent_field.ql b/test/cases/measure/data/input/project_non_existent_field.ql
new file mode 100644
index 000000000..0a567dc5f
--- /dev/null
+++ b/test/cases/measure/data/input/project_non_existent_field.ql
@@ -0,0 +1,19 @@
+# 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.
+
+SELECT id, total, nonexistent_field FROM service_cpm_minute IN sw_metric
+TIME > '-15m'
diff --git a/test/cases/measure/data/input/project_non_existent_field.yaml b/test/cases/measure/data/input/project_non_existent_field.yaml
new file mode 100644
index 000000000..64c7da9e4
--- /dev/null
+++ b/test/cases/measure/data/input/project_non_existent_field.yaml
@@ -0,0 +1,25 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: "service_cpm_minute"
+groups: ["sw_metric"]
+tagProjection:
+ tagFamilies:
+ - name: "default"
+ tags: ["id"]
+fieldProjection:
+ names: ["total", "nonexistent_field"]
diff --git a/test/cases/measure/measure.go b/test/cases/measure/measure.go
index 81127e221..a240092e4 100644
--- a/test/cases/measure/measure.go
+++ b/test/cases/measure/measure.go
@@ -97,6 +97,7 @@ var measureEntries = []any{
g.Entry("multi groups: new tag and fields", helpers.Args{Input: "multi_group_new_tag_field", Duration: 35 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("filter by non-existent tag", helpers.Args{Input: "filter_non_existent_tag", Duration: 25 * time.Minute, Offset: -20 * time.Minute, WantErr: true}),
g.Entry("project non-existent tag", helpers.Args{Input: "project_non_existent_tag", Duration: 25 * time.Minute, Offset: -20 * time.Minute, WantErr: true}),
+ g.Entry("project non-existent field", helpers.Args{Input: "project_non_existent_field", Duration: 25 * time.Minute, Offset: -20 * time.Minute, WantErr: true}),
g.Entry("write mixed", helpers.Args{Input: "write_mixed", Duration: 15 * time.Minute, Offset: 25 * time.Minute, DisOrder: true}),
g.Entry("filter by tag with NE", helpers.Args{Input: "tag_filter_ne", Duration: 25 * time.Minute, Offset: -20 * time.Minute, DisOrder: true}),
g.Entry("filter by tag with NOT IN", helpers.Args{Input: "tag_filter_not_in", Duration: 25 * time.Minute, Offset: -20 * time.Minute, DisOrder: true}),
@@ -107,6 +108,7 @@ var measureEntries = []any{
g.Entry("index mode filter by NE", helpers.Args{Input: "index_mode_ne", Duration: 25 * time.Minute, Offset: -20 * time.Minute, DisOrder: true}),
g.Entry("index mode filter by LE on int", helpers.Args{Input: "index_mode_le", Duration: 25 * time.Minute, Offset: -20 * time.Minute, DisOrder: true}),
g.Entry("offset beyond results", helpers.Args{Input: "offset_empty", Duration: 25 * time.Minute, Offset: -20 * time.Minute, WantEmpty: true}),
+ g.Entry("empty result for unmatched filter", helpers.Args{Input: "empty_result", Duration: 25 * time.Minute, Offset: -20 * time.Minute, WantEmpty: true}),
// Generated test cases (Layer 1: leaf conditions)
g.Entry("gen: leaf EQ string", helpers.Args{Input: "gen_leaf_eq_str", Want: "gen_leaf_eq_str", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
(skywalking-banyandb) 04/08: feat(query/vectorized/measure/plan): G9b GroupBy/Agg completeness (scalar reduce, raw GroupBy, projection auto-coverage)
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 cadfec0bf5ef48169fc491cae309d2aaeb96976a
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 10:08:28 2026 +0000
feat(query/vectorized/measure/plan): G9b GroupBy/Agg completeness (scalar reduce, raw GroupBy, projection auto-coverage)
---
pkg/query/vectorized/measure/groupby.go | 33 +++++-
pkg/query/vectorized/measure/plan.go | 46 +++++---
pkg/query/vectorized/measure/plan/analyzer.go | 129 ++++++++++++++++-----
pkg/query/vectorized/measure/plan/analyzer_test.go | 49 ++++++--
pkg/query/vectorized/measure/plan/dispatch.go | 90 ++------------
pkg/query/vectorized/measure/plan/executor_test.go | 115 ++++++++++++++++++
pkg/query/vectorized/measure/plan/groupby_agg.go | 39 ++++---
pkg/query/vectorized/measure/plan_test.go | 44 +++++--
8 files changed, 383 insertions(+), 162 deletions(-)
diff --git a/pkg/query/vectorized/measure/groupby.go b/pkg/query/vectorized/measure/groupby.go
index 9a8101131..7904da051 100644
--- a/pkg/query/vectorized/measure/groupby.go
+++ b/pkg/query/vectorized/measure/groupby.go
@@ -35,6 +35,13 @@ import (
// emitted in group-insertion order, with all rows of the first group flushed
// before the next group begins.
//
+// When firstOnly is set the operator keeps only the first-seen row of each
+// group and drops subsequent rows. This reproduces the row path's raw
+// GroupBy egress: the row aggregator wraps every group in a single
+// InternalDataPoint slice and banyand/query/processor.go reads only
+// current[0] per Next, so a raw GroupBy surfaces exactly one row (the
+// first-seen) per group in group-insertion order.
+//
// Memory accounting follows a pessimistic-reserve, refund-unused pattern:
//
// - entrySize is the per-new-group bucket overhead.
@@ -58,6 +65,7 @@ type BatchGroupBy struct {
cursorBucket int
cursorRow int
closed bool
+ firstOnly bool
}
type groupBucket struct {
@@ -88,6 +96,22 @@ func NewBatchGroupBy(
}
}
+// NewBatchGroupByFirst constructs a BatchGroupBy that emits only the
+// first-seen row of each group (raw GroupBy without aggregation). It
+// matches the row path: groupBy.Execute produces a groupIterator whose
+// Current() returns the whole group, and processor.go keeps only
+// current[0], so each group surfaces a single row in group-insertion
+// order with the input schema unchanged.
+func NewBatchGroupByFirst(
+ schema *vectorized.BatchSchema, keyIndices []int,
+ pool *vectorized.BatchPool, batchSize int,
+ tracker *vectorized.MemoryTracker, entrySize int64,
+) *BatchGroupBy {
+ g := NewBatchGroupBy(schema, keyIndices, pool, batchSize, tracker, entrySize, 0)
+ g.firstOnly = true
+ return g
+}
+
// Init prepares the group map.
func (g *BatchGroupBy) Init(_ context.Context) error {
g.groups = make(map[uint64][]*groupBucket)
@@ -120,16 +144,23 @@ func (g *BatchGroupBy) Consume(_ context.Context, b *vectorized.RecordBatch) err
func (g *BatchGroupBy) consumeRows(b *vectorized.RecordBatch, active []uint16) int64 {
var newGroups int64
+ var copiedRows int64
for _, rowIdx := range active {
bucket, isNew := g.findOrCreate(b, int(rowIdx))
if isNew {
newGroups++
}
+ if g.firstOnly && !isNew {
+ // Raw GroupBy keeps only the first-seen row per group;
+ // drop every later row in the same group.
+ continue
+ }
for colIdx, srcCol := range b.Columns {
copyOneValue(bucket.cols[colIdx], srcCol, int(rowIdx))
}
+ copiedRows++
}
- return newGroups*g.entrySize + int64(len(active))*g.rowSize
+ return newGroups*g.entrySize + copiedRows*g.rowSize
}
// findOrCreate locates the bucket for the row's key, creating it if absent.
diff --git a/pkg/query/vectorized/measure/plan.go b/pkg/query/vectorized/measure/plan.go
index 747b96079..38235b1e8 100644
--- a/pkg/query/vectorized/measure/plan.go
+++ b/pkg/query/vectorized/measure/plan.go
@@ -33,15 +33,18 @@ const aggEntrySize int64 = 512
// into a list of BreakerOperators that, chained after the scan source, form
// the vectorized aggregation pipeline.
//
-// Routing rules (v1):
+// Routing rules:
// - GroupBy + Agg both set → emit a single BatchAggregation. The operator's
// own keyIndex map produces per-group buckets and folds the agg slot;
// a separate BatchGroupBy would double the row materialization.
-// - GroupBy set without Agg → unsupported. The proto field is paired with
-// `agg` for the existing TopN path; without Agg the result semantic is
-// ambiguous (selection? deduplication?). Return an error.
-// - Agg set without GroupBy → unsupported. Scalar reduce → single output
-// row is deferred to a future increment.
+// - Agg set without GroupBy → scalar reduce: a BatchAggregation with no
+// key columns. Every row maps to one group, so a single output row is
+// emitted carrying the first-seen projected tags plus the agg result,
+// matching the row path's aggAllIterator.
+// - GroupBy set without Agg → raw GroupBy: a first-seen-row-per-group
+// BatchGroupBy. The output preserves the input schema; one row per
+// group is emitted in group-insertion order, matching the row path's
+// groupIterator + processor.go's current[0] read.
// - Neither set → empty operator list; caller emits raw rows.
//
// tracker is the per-pipeline MemoryTracker (G7a); it must be non-nil when
@@ -57,12 +60,6 @@ func BuildOperators(
if !hasGroupBy && !hasAgg {
return nil, nil
}
- if hasGroupBy && !hasAgg {
- return nil, fmt.Errorf("vectorized.measure: GroupBy without Agg is not supported in v1")
- }
- if hasAgg && !hasGroupBy {
- return nil, fmt.Errorf("vectorized.measure: Agg without GroupBy (scalar reduce) is not supported in v1")
- }
if tracker == nil {
return nil, fmt.Errorf("vectorized.measure: BuildOperators requires a non-nil shared MemoryTracker")
}
@@ -70,9 +67,22 @@ func BuildOperators(
return nil, fmt.Errorf("vectorized.measure: batchSize must be > 0, got %d", batchSize)
}
- keyIndices, keyErr := lookupGroupByKeyIndices(schema, opts.GroupBy)
- if keyErr != nil {
- return nil, keyErr
+ var keyIndices []int
+ if hasGroupBy {
+ var keyErr error
+ keyIndices, keyErr = lookupGroupByKeyIndices(schema, opts.GroupBy)
+ if keyErr != nil {
+ return nil, keyErr
+ }
+ }
+
+ if !hasAgg {
+ // Raw GroupBy: emit the first-seen row of each group with the
+ // input schema unchanged. entrySize accounts for the per-group
+ // bucket; rowSize is zero because at most one row is retained.
+ pool := vectorized.NewBatchPool(schema, batchSize)
+ gb := NewBatchGroupByFirst(schema, keyIndices, pool, batchSize, tracker, aggEntrySize)
+ return []vectorized.BreakerOperator{gb}, nil
}
fieldIdx, fieldErr := lookupFieldColumnIndex(schema, opts.Agg.FieldName)
@@ -93,6 +103,12 @@ func BuildOperators(
// like "value_sum" — the suffix would break proto.Equal parity in the
// integration suite. The aggregation function lives on the operator
// spec, not in the column name.
+ //
+ // keyIndices is empty for scalar reduce (Agg without GroupBy):
+ // BatchAggregation.computeKey then returns the same key for every
+ // row, so all rows collapse into a single output row carrying the
+ // first-seen projected tags plus the agg result — the columnar
+ // equivalent of the row path's aggAllIterator.
spec := AggSpec{
Func: aggFn,
InputCol: fieldIdx,
diff --git a/pkg/query/vectorized/measure/plan/analyzer.go b/pkg/query/vectorized/measure/plan/analyzer.go
index 620c9dcc5..6eee444fb 100644
--- a/pkg/query/vectorized/measure/plan/analyzer.go
+++ b/pkg/query/vectorized/measure/plan/analyzer.go
@@ -48,8 +48,11 @@ const defaultLimit uint32 = 100
// - tag/field projection naming columns not in the schema
// - GroupBy referencing a tag absent from the schema
// - Agg referencing a field absent from the schema
-// - GroupBy set without Agg (raw groupby not supported in v1)
-// - Agg set without GroupBy (scalar reduce not supported in v1)
+//
+// GroupBy and Agg may travel together (group + aggregate), or either
+// alone: Agg without GroupBy is a scalar reduce (single output row);
+// GroupBy without Agg is a raw GroupBy (first-seen row per group). Both
+// mirror the row path (measure_plan_aggregation.go / measure_plan_groupby.go).
func Analyze(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (VecPlan, error) {
if req == nil {
return nil, fmt.Errorf("plan.Analyze: nil request")
@@ -72,13 +75,30 @@ func Analyze(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (Ve
hasAgg := req.GetAgg() != nil
var gbModel *model.MeasureGroupBy
var aggModel *model.MeasureAgg
- if hasGroupBy && hasAgg {
- var translateErr error
- gbModel, aggModel, translateErr = translateGroupByAgg(req, measureSchema)
- if translateErr != nil {
- return nil, translateErr
+ if hasGroupBy {
+ var gbErr error
+ gbModel, gbErr = translateGroupBy(req, measureSchema)
+ if gbErr != nil {
+ return nil, gbErr
+ }
+ }
+ if hasAgg {
+ var aggErr error
+ aggModel, aggErr = translateAgg(req, measureSchema)
+ if aggErr != nil {
+ return nil, aggErr
}
}
+
+ // Projection auto-coverage: BatchAggregation / BatchGroupBy locate
+ // their key and value columns by name inside the BatchSchema, which
+ // BuildBatchSchema derives from TagProjection + FieldProjection.
+ // Extend the projection implicitly so the GroupBy keys and the Agg
+ // field always materialize a column, instead of falling through to
+ // the row path when the request omitted them from its projection.
+ tagProjection = ensureGroupByProjected(tagProjection, gbModel)
+ fieldProjection = ensureAggFieldProjected(fieldProjection, aggModel)
+
opts := model.MeasureQueryOptions{
TagProjection: tagProjection,
FieldProjection: fieldProjection,
@@ -105,17 +125,15 @@ func Analyze(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (Ve
Agg: aggModel,
})
- switch {
- case hasGroupBy && hasAgg:
+ if hasGroupBy || hasAgg {
+ // One BatchAggregation / BatchGroupBy node covers all three
+ // shapes (group+agg, scalar reduce, raw groupby); BuildOperators
+ // routes on which of gbModel/aggModel is set.
gba, gbaErr := NewGroupByAgg(plan, gbModel, aggModel)
if gbaErr != nil {
return nil, gbaErr
}
plan = gba
- case hasGroupBy:
- return nil, fmt.Errorf("plan.Analyze: GroupBy without Agg is not supported in v1")
- case hasAgg:
- return nil, fmt.Errorf("plan.Analyze: Agg without GroupBy (scalar reduce) is not supported in v1")
}
if t := req.GetTop(); t != nil {
@@ -150,44 +168,101 @@ func buildTagProjection(req *measurev1.QueryRequest) []model.TagProjection {
return out
}
-// translateGroupByAgg builds the model GroupBy/Agg structs from the proto,
+// translateGroupBy builds the model GroupBy struct from the proto,
// validating that:
// - the GroupBy tag_projection names exactly one family with non-empty tags
// (v1 single-family limitation)
// - the named GroupBy tags exist in the Measure schema
-// - the Agg field exists in the Measure schema
-func translateGroupByAgg(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (
- *model.MeasureGroupBy, *model.MeasureAgg, error,
-) {
- gbProto := req.GetGroupBy()
- families := gbProto.GetTagProjection().GetTagFamilies()
+func translateGroupBy(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (*model.MeasureGroupBy, error) {
+ families := req.GetGroupBy().GetTagProjection().GetTagFamilies()
if len(families) == 0 {
- return nil, nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection must list at least one tag family")
+ return nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection must list at least one tag family")
}
if len(families) > 1 {
- return nil, nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection v1 supports a single tag family, got %d", len(families))
+ return nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection v1 supports a single tag family, got %d", len(families))
}
family := families[0]
if len(family.GetTags()) == 0 {
- return nil, nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection family %q has no tags", family.GetName())
+ return nil, fmt.Errorf("plan.Analyze: GroupBy.tag_projection family %q has no tags", family.GetName())
}
gb := &model.MeasureGroupBy{
TagFamily: family.GetName(),
TagNames: append([]string(nil), family.GetTags()...),
}
if validateErr := validateGroupByTags(measureSchema, gb); validateErr != nil {
- return nil, nil, validateErr
+ return nil, validateErr
}
+ return gb, nil
+}
+// translateAgg builds the model Agg struct from the proto, validating
+// that the Agg field exists in the Measure schema.
+func translateAgg(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (*model.MeasureAgg, error) {
aggProto := req.GetAgg()
if validateErr := validateAggField(measureSchema, aggProto.GetFieldName()); validateErr != nil {
- return nil, nil, validateErr
+ return nil, validateErr
}
- agg := &model.MeasureAgg{
+ return &model.MeasureAgg{
FieldName: aggProto.GetFieldName(),
Func: aggProto.GetFunction(),
+ }, nil
+}
+
+// ensureGroupByProjected returns a TagProjection slice guaranteed to
+// include every GroupBy key tag. When the request already projects them
+// the input slice is returned unchanged; otherwise the missing key tags
+// are appended to (or create) the GroupBy family. Mirrors the row path,
+// whose GroupBy resolves its key tag refs against the schema regardless
+// of the request projection.
+func ensureGroupByProjected(tp []model.TagProjection, gb *model.MeasureGroupBy) []model.TagProjection {
+ if gb == nil || gb.TagFamily == "" || len(gb.TagNames) == 0 {
+ return tp
+ }
+ out := append([]model.TagProjection(nil), tp...)
+ familyIdx := -1
+ for i := range out {
+ if out[i].Family == gb.TagFamily {
+ familyIdx = i
+ break
+ }
+ }
+ if familyIdx < 0 {
+ out = append(out, model.TagProjection{
+ Family: gb.TagFamily,
+ Names: append([]string(nil), gb.TagNames...),
+ })
+ return out
+ }
+ present := make(map[string]struct{}, len(out[familyIdx].Names))
+ for _, n := range out[familyIdx].Names {
+ present[n] = struct{}{}
+ }
+ names := append([]string(nil), out[familyIdx].Names...)
+ for _, n := range gb.TagNames {
+ if _, ok := present[n]; !ok {
+ names = append(names, n)
+ }
+ }
+ out[familyIdx].Names = names
+ return out
+}
+
+// ensureAggFieldProjected returns a FieldProjection slice guaranteed to
+// include the Agg field. When already present the input is returned
+// unchanged; otherwise the field is appended. Mirrors the row path,
+// whose aggregation resolves its field ref against the schema regardless
+// of the request projection.
+func ensureAggFieldProjected(fp []string, agg *model.MeasureAgg) []string {
+ if agg == nil || agg.FieldName == "" {
+ return fp
+ }
+ for _, n := range fp {
+ if n == agg.FieldName {
+ return fp
+ }
}
- return gb, agg, nil
+ out := append([]string(nil), fp...)
+ return append(out, agg.FieldName)
}
// validateGroupByTags ensures every name in gb.TagNames exists within the
diff --git a/pkg/query/vectorized/measure/plan/analyzer_test.go b/pkg/query/vectorized/measure/plan/analyzer_test.go
index 1605a3399..f7041dd87 100644
--- a/pkg/query/vectorized/measure/plan/analyzer_test.go
+++ b/pkg/query/vectorized/measure/plan/analyzer_test.go
@@ -144,7 +144,9 @@ func TestAnalyze_TopBetweenGroupByAggAndLimit(t *testing.T) {
}
}
-func TestAnalyze_GroupByWithoutAgg_Errors(t *testing.T) {
+// TestAnalyze_GroupByWithoutAgg_BuildsRawGroupBy verifies the raw
+// GroupBy shape: a GroupByAgg node (Agg nil) below Limit.
+func TestAnalyze_GroupByWithoutAgg_BuildsRawGroupBy(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
@@ -154,16 +156,29 @@ func TestAnalyze_GroupByWithoutAgg_Errors(t *testing.T) {
FieldName: "value",
},
}
- _, err := Analyze(req, testMeasureSchema())
- if err == nil {
- t.Fatal("GroupBy without Agg must error")
+ p, err := Analyze(req, testMeasureSchema())
+ if err != nil {
+ t.Fatalf("GroupBy without Agg (raw groupby) must not error: %v", err)
+ }
+ limit, ok := p.(*Limit)
+ if !ok {
+ t.Fatalf("root should be *Limit, got %T", p)
}
- if !strings.Contains(err.Error(), "Agg") {
- t.Fatalf("error should mention Agg, got %v", err)
+ gba, ok := limit.Child.(*GroupByAgg)
+ if !ok {
+ t.Fatalf("Limit child should be *GroupByAgg, got %T", limit.Child)
+ }
+ if gba.Agg != nil {
+ t.Fatalf("raw GroupBy must have nil Agg, got %+v", gba.Agg)
+ }
+ if gba.GroupBy == nil || gba.GroupBy.TagNames[0] != "svc" {
+ t.Fatalf("GroupBy.TagNames: want [svc], got %+v", gba.GroupBy)
}
}
-func TestAnalyze_AggWithoutGroupBy_Errors(t *testing.T) {
+// TestAnalyze_AggWithoutGroupBy_BuildsScalarReduce verifies the scalar
+// reduce shape: a GroupByAgg node (GroupBy nil) below Limit.
+func TestAnalyze_AggWithoutGroupBy_BuildsScalarReduce(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
@@ -173,9 +188,23 @@ func TestAnalyze_AggWithoutGroupBy_Errors(t *testing.T) {
FieldName: "value",
},
}
- _, err := Analyze(req, testMeasureSchema())
- if err == nil {
- t.Fatal("Agg without GroupBy must error (scalar reduce not supported)")
+ p, err := Analyze(req, testMeasureSchema())
+ if err != nil {
+ t.Fatalf("Agg without GroupBy (scalar reduce) must not error: %v", err)
+ }
+ limit, ok := p.(*Limit)
+ if !ok {
+ t.Fatalf("root should be *Limit, got %T", p)
+ }
+ gba, ok := limit.Child.(*GroupByAgg)
+ if !ok {
+ t.Fatalf("Limit child should be *GroupByAgg, got %T", limit.Child)
+ }
+ if gba.GroupBy != nil {
+ t.Fatalf("scalar reduce must have nil GroupBy, got %+v", gba.GroupBy)
+ }
+ if gba.Agg == nil || gba.Agg.FieldName != "value" {
+ t.Fatalf("Agg.FieldName: want value, got %+v", gba.Agg)
}
}
diff --git a/pkg/query/vectorized/measure/plan/dispatch.go b/pkg/query/vectorized/measure/plan/dispatch.go
index 8a2a7b573..bafceb089 100644
--- a/pkg/query/vectorized/measure/plan/dispatch.go
+++ b/pkg/query/vectorized/measure/plan/dispatch.go
@@ -88,8 +88,9 @@ func FellThroughCount() int64 { return fellThroughCount.Load() }
//
// Eligibility gate (v1):
// - cfg.Enabled must be true
-// - request may carry GroupBy+Agg as a pair (scalar reduce + raw
-// GroupBy are deferred); aggProjectionCoverage must also hold
+// - request may carry GroupBy and/or Agg in any combination (group+agg,
+// scalar reduce, raw GroupBy); plan.Analyze auto-extends the
+// projection so the keys / agg field always resolve
// - request must NOT carry Top (BatchTop's single-heap semantic differs
// from the row path's per-timestamp top-N)
// - request must carry TimeRange (storage requires a bounded window)
@@ -131,24 +132,11 @@ func Dispatch(
// today.
return nil, "", false, nil
}
- hasGroupBy := req.GetGroupBy() != nil
- hasAgg := req.GetAgg() != nil
- if hasGroupBy || hasAgg {
- // GroupBy and Agg must travel as a pair (scalar reduce + raw
- // groupby are deferred). Either alone falls through.
- if hasGroupBy != hasAgg {
- return nil, "", false, nil
- }
- // Projection coverage: BatchAggregation locates its key + value
- // columns by name inside the BatchSchema, which is built from
- // TagProjection + FieldProjection. Missing coverage means the
- // operator would fail at construction; fall through so the row
- // path can extend projection implicitly or surface its
- // canonical error.
- if !aggProjectionCoverage(req) {
- return nil, "", false, nil
- }
- }
+ // GroupBy and Agg are handled by the vec subsystem in all three
+ // shapes — group+agg, scalar reduce (Agg only), raw GroupBy (GroupBy
+ // only). plan.Analyze auto-extends the projection so the GroupBy keys
+ // and Agg field always materialize a column, so there is no
+ // projection-coverage gate here anymore (G9b).
// G9c #9: a nil TimeRange is NOT a fall-through. The row path does not
// reject it — parseFields feeds criteria.GetTimeRange().GetBegin().AsTime()
// into the index scan, and a nil *timestamppb.Timestamp resolves to the
@@ -409,65 +397,3 @@ func (emptyMIterator) Next() bool { return false }
func (emptyMIterator) Current() []*measurev1.InternalDataPoint { return nil }
func (emptyMIterator) Close() error { return nil }
-
-// aggProjectionCoverage reports whether the request's GroupBy keys and
-// Agg field are all present in the request's projections. Required by
-// the dispatch eligibility gate: the BatchAggregation operator locates
-// its key + value columns by name inside the BatchSchema, and the
-// BatchSchema is built from TagProjection + FieldProjection. Missing
-// coverage means the operator would fail at construction; dispatch
-// instead falls through so the row path can handle the request.
-//
-// v1 requires GroupBy.tag_projection to name a single family; that
-// family must appear in req.TagProjection with every tag in GroupBy
-// present. Agg.field_name must appear in req.FieldProjection. Non-key
-// projected tags are allowed; BatchAggregation carries them forward as
-// first-seen-per-group, matching the row path.
-func aggProjectionCoverage(req *measurev1.QueryRequest) bool {
- gb := req.GetGroupBy()
- if gb == nil {
- return false
- }
- gbFamilies := gb.GetTagProjection().GetTagFamilies()
- if len(gbFamilies) != 1 {
- return false
- }
- gbFamily := gbFamilies[0]
- projected := projectedTagsByFamily(req.GetTagProjection())
- present, ok := projected[gbFamily.GetName()]
- if !ok {
- return false
- }
- for _, name := range gbFamily.GetTags() {
- if _, hit := present[name]; !hit {
- return false
- }
- }
- aggField := req.GetAgg().GetFieldName()
- if aggField == "" {
- return false
- }
- for _, name := range req.GetFieldProjection().GetNames() {
- if name == aggField {
- return true
- }
- }
- return false
-}
-
-// projectedTagsByFamily flattens a TagProjection into family → name-set.
-func projectedTagsByFamily(tp *modelv1.TagProjection) map[string]map[string]struct{} {
- out := make(map[string]map[string]struct{})
- if tp == nil {
- return out
- }
- for _, tf := range tp.GetTagFamilies() {
- family := tf.GetName()
- names := make(map[string]struct{}, len(tf.GetTags()))
- for _, n := range tf.GetTags() {
- names[n] = struct{}{}
- }
- out[family] = names
- }
- return out
-}
diff --git a/pkg/query/vectorized/measure/plan/executor_test.go b/pkg/query/vectorized/measure/plan/executor_test.go
index f2148874e..cfce60e0f 100644
--- a/pkg/query/vectorized/measure/plan/executor_test.go
+++ b/pkg/query/vectorized/measure/plan/executor_test.go
@@ -123,6 +123,121 @@ func TestExecute_GroupByAgg_EmitsAggregatedRowsWithNilTimestamp(t *testing.T) {
}
}
+// TestExecute_ScalarReduce_EmitsSingleRow drives Agg without GroupBy.
+// The whole result collapses into one row carrying the first-seen
+// projected tag plus the agg result, with nil Timestamp — matching the
+// row path's aggAllIterator. Fixture: value column {1,4,2,3,5} → SUM 15.
+func TestExecute_ScalarReduce_EmitsSingleRow(t *testing.T) {
+ schema, batch := buildScanInput(t)
+ src := &fakePullSource{schema: schema, batches: []*vectorized.RecordBatch{batch}}
+ scan := NewScan(schema, ScanParams{})
+ scan.Source = src
+ gba, err := NewGroupByAgg(scan, nil,
+ &model.MeasureAgg{FieldName: "value", Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ )
+ if err != nil {
+ t.Fatalf("NewGroupByAgg: %v", err)
+ }
+ root := NewLimit(gba, 0, 10)
+
+ iter, err := Execute(context.Background(), root, execCfg())
+ if err != nil {
+ t.Fatalf("Execute: %v", err)
+ }
+ defer iter.Close()
+
+ rows := 0
+ var sum int64
+ var svc string
+ for iter.Next() {
+ dps := iter.Current()
+ if len(dps) != 1 {
+ t.Fatalf("Current must yield 1 InternalDataPoint per Next, got %d", len(dps))
+ }
+ idp := dps[0]
+ if idp.DataPoint.Timestamp != nil {
+ t.Fatalf("scalar reduce row must have nil Timestamp; got %v", idp.DataPoint.Timestamp)
+ }
+ tags := idp.DataPoint.TagFamilies[0].Tags
+ svc = tags[0].Value.GetStr().GetValue()
+ if len(idp.DataPoint.Fields) != 1 || idp.DataPoint.Fields[0].Name != "value" {
+ t.Fatalf("want one Field 'value', got %+v", idp.DataPoint.Fields)
+ }
+ sum = idp.DataPoint.Fields[0].Value.GetInt().GetValue()
+ rows++
+ }
+ if rows != 1 {
+ t.Fatalf("scalar reduce must emit exactly 1 row, got %d", rows)
+ }
+ if sum != 15 {
+ t.Fatalf("scalar SUM(value): want 15, got %d", sum)
+ }
+ if svc != "a" {
+ t.Fatalf("scalar reduce carries first-seen tag: want svc=a, got %q", svc)
+ }
+}
+
+// TestExecute_RawGroupBy_EmitsFirstRowPerGroup drives GroupBy without
+// Agg. One row per group is emitted in group-insertion order with the
+// input schema preserved (timestamp + field present), matching the row
+// path's groupIterator + processor.go's current[0] read. Fixture groups:
+// a → first row (ts=1,value=1); b → first row (ts=2,value=4).
+func TestExecute_RawGroupBy_EmitsFirstRowPerGroup(t *testing.T) {
+ schema, batch := buildScanInput(t)
+ src := &fakePullSource{schema: schema, batches: []*vectorized.RecordBatch{batch}}
+ scan := NewScan(schema, ScanParams{})
+ scan.Source = src
+ gba, err := NewGroupByAgg(scan,
+ &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{"svc"}}, nil,
+ )
+ if err != nil {
+ t.Fatalf("NewGroupByAgg: %v", err)
+ }
+ root := NewLimit(gba, 0, 10)
+
+ iter, err := Execute(context.Background(), root, execCfg())
+ if err != nil {
+ t.Fatalf("Execute: %v", err)
+ }
+ defer iter.Close()
+
+ type row struct {
+ value int64
+ first bool
+ }
+ got := map[string]row{}
+ order := make([]string, 0, 2)
+ for iter.Next() {
+ dps := iter.Current()
+ if len(dps) != 1 {
+ t.Fatalf("Current must yield 1 InternalDataPoint per Next, got %d", len(dps))
+ }
+ idp := dps[0]
+ // Raw GroupBy preserves the input schema, so the field column
+ // survives (unlike the aggregation shapes).
+ if len(idp.DataPoint.Fields) != 1 || idp.DataPoint.Fields[0].Name != "value" {
+ t.Fatalf("raw GroupBy must preserve the field column, got %+v", idp.DataPoint.Fields)
+ }
+ svc := idp.DataPoint.TagFamilies[0].Tags[0].Value.GetStr().GetValue()
+ if _, seen := got[svc]; !seen {
+ order = append(order, svc)
+ }
+ got[svc] = row{value: idp.DataPoint.Fields[0].Value.GetInt().GetValue(), first: true}
+ }
+ if len(got) != 2 {
+ t.Fatalf("raw GroupBy must emit one row per group (2), got %d", len(got))
+ }
+ if got["a"].value != 1 {
+ t.Fatalf("group a first-seen value: want 1, got %d", got["a"].value)
+ }
+ if got["b"].value != 4 {
+ t.Fatalf("group b first-seen value: want 4, got %d", got["b"].value)
+ }
+ if len(order) != 2 || order[0] != "a" || order[1] != "b" {
+ t.Fatalf("groups must emit in insertion order [a b], got %v", order)
+ }
+}
+
func TestExecute_BuildError_SurfacesAsExecuteError(t *testing.T) {
schema, _ := buildScanInput(t)
scan := NewScan(schema, ScanParams{}) // Source intentionally unset
diff --git a/pkg/query/vectorized/measure/plan/groupby_agg.go b/pkg/query/vectorized/measure/plan/groupby_agg.go
index df678440b..326235b3b 100644
--- a/pkg/query/vectorized/measure/plan/groupby_agg.go
+++ b/pkg/query/vectorized/measure/plan/groupby_agg.go
@@ -27,14 +27,19 @@ import (
vmeasure "github.com/apache/skywalking-banyandb/pkg/query/vectorized/measure"
)
-// GroupByAgg is the v1 vec aggregation node: GroupBy + single Aggregation
-// fused into one BatchAggregation operator (see G7d planner). The
-// operator does its own grouping via keyIndices and folding via aggs.
+// GroupByAgg is the vec aggregation/grouping node. It fuses GroupBy and a
+// single Aggregation into one BatchAggregation operator (see G7d
+// planner), and also covers the two single-sided shapes:
//
-// Schema-rewriting: output is key columns + one agg result column. The
-// timestamp column is dropped, so serializeBatchToProto emits
-// DataPoint.Timestamp == nil for every output row (per G7 design
-// decision D2).
+// - GroupBy + Agg → BatchAggregation: key columns + one agg result
+// column (schema-rewriting; timestamp dropped, per G7 decision D2).
+// - Agg without GroupBy → scalar reduce: BatchAggregation with no key
+// columns, a single output row (first-seen tags + agg result).
+// - GroupBy without Agg → raw GroupBy: a first-seen-row-per-group
+// BatchGroupBy, schema-preserving.
+//
+// The concrete operator (and therefore the output schema) is chosen by
+// vmeasure.BuildOperators from which of GroupBy/Agg is set.
type GroupByAgg struct {
Child VecPlan
GroupBy *model.MeasureGroupBy
@@ -42,15 +47,12 @@ type GroupByAgg struct {
outputCache *vectorized.BatchSchema
}
-// NewGroupByAgg constructs a GroupByAgg node wrapping child. Returns an
-// error if GroupBy or Agg are nil (v1 requires both: scalar reduce and
-// raw groupby are not supported).
+// NewGroupByAgg constructs a GroupByAgg node wrapping child. At least one
+// of groupBy/agg must be set (BuildOperators routes on which); child must
+// not be nil.
func NewGroupByAgg(child VecPlan, groupBy *model.MeasureGroupBy, agg *model.MeasureAgg) (*GroupByAgg, error) {
- if groupBy == nil {
- return nil, fmt.Errorf("plan.GroupByAgg: GroupBy must not be nil (scalar reduce not supported in v1)")
- }
- if agg == nil {
- return nil, fmt.Errorf("plan.GroupByAgg: Agg must not be nil (raw groupby not supported in v1)")
+ if groupBy == nil && agg == nil {
+ return nil, fmt.Errorf("plan.GroupByAgg: GroupBy and Agg must not both be nil")
}
if child == nil {
return nil, fmt.Errorf("plan.GroupByAgg: Child must not be nil")
@@ -116,9 +118,8 @@ func (g *GroupByAgg) String() string {
if g.GroupBy != nil {
tagNames = strings.Join(g.GroupBy.TagNames, ",")
}
- field := ""
- if g.Agg != nil {
- field = g.Agg.FieldName
+ if g.Agg == nil {
+ return fmt.Sprintf("GroupByAgg(keys=%s, raw)", tagNames)
}
- return fmt.Sprintf("GroupByAgg(keys=%s, fn=%v, field=%s)", tagNames, g.Agg.Func, field)
+ return fmt.Sprintf("GroupByAgg(keys=%s, fn=%v, field=%s)", tagNames, g.Agg.Func, g.Agg.FieldName)
}
diff --git a/pkg/query/vectorized/measure/plan_test.go b/pkg/query/vectorized/measure/plan_test.go
index ed4b7e4ea..eade8311f 100644
--- a/pkg/query/vectorized/measure/plan_test.go
+++ b/pkg/query/vectorized/measure/plan_test.go
@@ -96,23 +96,51 @@ func TestBuildOperators_AggOutputName_InheritsInputFieldName(t *testing.T) {
}
}
-func TestBuildOperators_GroupByWithoutAgg_Errors(t *testing.T) {
+// TestBuildOperators_GroupByWithoutAgg_EmitsFirstOnlyGroupBy pins the
+// raw-GroupBy shape: a first-seen-row-per-group BatchGroupBy whose output
+// preserves the input schema. It matches the row path's groupIterator
+// combined with processor.go's current[0] read.
+func TestBuildOperators_GroupByWithoutAgg_EmitsFirstOnlyGroupBy(t *testing.T) {
opts := model.MeasureQueryOptions{
GroupBy: &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{"svc"}},
}
- _, err := BuildOperators(opts, planSchema(), vectorized.NewMemoryTracker(1<<20), 1024)
- if err == nil {
- t.Fatal("GroupBy without Agg must error in v1")
+ ops, err := BuildOperators(opts, planSchema(), vectorized.NewMemoryTracker(1<<20), 1024)
+ if err != nil {
+ t.Fatalf("GroupBy without Agg (raw groupby) must not error: %v", err)
+ }
+ if len(ops) != 1 {
+ t.Fatalf("raw GroupBy should emit 1 operator, got %d", len(ops))
+ }
+ gb, ok := ops[0].(*BatchGroupBy)
+ if !ok {
+ t.Fatalf("operator must be *BatchGroupBy, got %T", ops[0])
+ }
+ if !gb.firstOnly {
+ t.Fatal("raw GroupBy must be first-only (one row per group)")
}
}
-func TestBuildOperators_AggWithoutGroupBy_Errors(t *testing.T) {
+// TestBuildOperators_AggWithoutGroupBy_EmitsBatchAggregation pins the
+// scalar-reduce shape: a BatchAggregation with no key columns, so every
+// row collapses into a single output row carrying the first-seen tags
+// plus the agg result, matching the row path's aggAllIterator.
+func TestBuildOperators_AggWithoutGroupBy_EmitsBatchAggregation(t *testing.T) {
opts := model.MeasureQueryOptions{
Agg: &model.MeasureAgg{FieldName: "value", Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
}
- _, err := BuildOperators(opts, planSchema(), vectorized.NewMemoryTracker(1<<20), 1024)
- if err == nil {
- t.Fatal("Agg without GroupBy (scalar reduce) must error in v1")
+ ops, err := BuildOperators(opts, planSchema(), vectorized.NewMemoryTracker(1<<20), 1024)
+ if err != nil {
+ t.Fatalf("Agg without GroupBy (scalar reduce) must not error: %v", err)
+ }
+ if len(ops) != 1 {
+ t.Fatalf("scalar reduce should emit 1 operator, got %d", len(ops))
+ }
+ agg, ok := ops[0].(*BatchAggregation)
+ if !ok {
+ t.Fatalf("operator must be *BatchAggregation, got %T", ops[0])
+ }
+ if len(agg.keyIndices) != 0 {
+ t.Fatalf("scalar reduce must have no key columns, got %d", len(agg.keyIndices))
}
}
(skywalking-banyandb) 06/08: fix(query/vectorized/measure): COUNT preserves input type for row-path parity (float_top_count)
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 2cb0ddcfe6bc1c7449c08036e9d7b0efa7c02494
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 10:12:35 2026 +0000
fix(query/vectorized/measure): COUNT preserves input type for row-path parity (float_top_count)
---
pkg/query/vectorized/measure/aggregation.go | 36 ++++++++++++-----------------
1 file changed, 15 insertions(+), 21 deletions(-)
diff --git a/pkg/query/vectorized/measure/aggregation.go b/pkg/query/vectorized/measure/aggregation.go
index c3cbba065..a03e5cbc4 100644
--- a/pkg/query/vectorized/measure/aggregation.go
+++ b/pkg/query/vectorized/measure/aggregation.go
@@ -345,18 +345,14 @@ func newAggSlot(fn AggFunc, inputIsFloat bool) (aggSlot, error) {
return aggSlot{}, modelErr
}
slot := aggSlot{fn: fn, inputIsFloat: inputIsFloat}
- useFloat := false
- switch fn {
- case AggCount:
- useFloat = false
- default:
- // MEAN follows input type to match the row path's
- // aggregation.NewMap[int64] / [float64] dispatch in
- // pkg/query/logical/measure/measure_plan_aggregation.go
- // (FIELD_TYPE_INT → int64 truncation; FIELD_TYPE_FLOAT →
- // float64). Same handling for SUM/MIN/MAX.
- useFloat = inputIsFloat
- }
+ // All functions follow the input type to match the row path, whose
+ // aggregation.NewMap[int64] / [float64] is dispatched on the field's
+ // declared type in pkg/query/logical/measure/measure_plan_aggregation.go
+ // (FIELD_TYPE_INT → int64; FIELD_TYPE_FLOAT → float64). COUNT is
+ // included: the row path's countFunc[N] is parameterized by N and
+ // ToFieldValue[N] emits FieldValue_Int / FieldValue_Float by N, so
+ // COUNT on a float field must emit a float (e.g. float_top_count).
+ useFloat := inputIsFloat
if useFloat {
m, mapErr := aggregation.NewMap[float64](af)
if mapErr != nil {
@@ -417,14 +413,12 @@ func buildAggOutputSchema(
}
// aggOutputType maps (input type, agg func) to the output column type.
-// - COUNT is always int64.
-// - MEAN, SUM, MIN, MAX preserve the input type so vec egress emits the
-// same FieldValue oneof variant the row path uses (its accumulator is
-// dispatched on FIELD_TYPE_INT → Map[int64] / FIELD_TYPE_FLOAT →
-// Map[float64]; see measure_plan_aggregation.go).
-func aggOutputType(in vectorized.ColumnType, fn AggFunc) vectorized.ColumnType {
- if fn == AggCount {
- return vectorized.ColumnTypeInt64
- }
+// Every function (COUNT included) preserves the input type so vec egress
+// emits the same FieldValue oneof variant the row path uses: the row
+// path's accumulator and ToFieldValue[N] are dispatched on the field's
+// declared type (FIELD_TYPE_INT → int64 → FieldValue_Int;
+// FIELD_TYPE_FLOAT → float64 → FieldValue_Float; see
+// measure_plan_aggregation.go and pkg/query/aggregation).
+func aggOutputType(in vectorized.ColumnType, _ AggFunc) vectorized.ColumnType {
return in
}
(skywalking-banyandb) 07/08: chore(query/vectorized/measure/plan): fix goconst/dogsled lint in G9 test files
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 08762454659162fe76bf6d278e756c15b0b9e5ce
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 10:25:43 2026 +0000
chore(query/vectorized/measure/plan): fix goconst/dogsled lint in G9 test files
---
pkg/query/vectorized/measure/plan/analyzer_test.go | 67 ++++++++++++----------
pkg/query/vectorized/measure/plan/build_test.go | 16 +++---
pkg/query/vectorized/measure/plan/dispatch_test.go | 39 +++++++------
pkg/query/vectorized/measure/plan/executor_test.go | 16 +++---
4 files changed, 73 insertions(+), 65 deletions(-)
diff --git a/pkg/query/vectorized/measure/plan/analyzer_test.go b/pkg/query/vectorized/measure/plan/analyzer_test.go
index 544b6fc5b..94197fc54 100644
--- a/pkg/query/vectorized/measure/plan/analyzer_test.go
+++ b/pkg/query/vectorized/measure/plan/analyzer_test.go
@@ -27,8 +27,13 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
)
-// testMeasureSchema builds a minimal Measure schema with one "default" tag
-// family containing "svc" + "region" tag specs and one "value" field.
+const (
+ tagSvc = "svc"
+ fieldValue = "value"
+)
+
+// testMeasureSchema builds a minimal Measure schema with one default tag
+// family containing the svc + region tag specs and one value field.
func testMeasureSchema() *databasev1.Measure {
return &databasev1.Measure{
Metadata: &commonv1.Metadata{Name: "demo", Group: "default"},
@@ -36,20 +41,20 @@ func testMeasureSchema() *databasev1.Measure {
{
Name: "default",
Tags: []*databasev1.TagSpec{
- {Name: "svc", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: tagSvc, Type: databasev1.TagType_TAG_TYPE_STRING},
{Name: "region", Type: databasev1.TagType_TAG_TYPE_STRING},
},
},
},
Fields: []*databasev1.FieldSpec{
- {Name: "value", FieldType: databasev1.FieldType_FIELD_TYPE_INT},
+ {Name: fieldValue, FieldType: databasev1.FieldType_FIELD_TYPE_INT},
},
}
}
func projTagProj() *modelv1.TagProjection {
return &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{
- {Name: "default", Tags: []string{"svc"}},
+ {Name: "default", Tags: []string{tagSvc}},
}}
}
@@ -57,7 +62,7 @@ func TestAnalyze_BareRequest_BuildsScanWrappedInLimit(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
}
p, err := Analyze(req, testMeasureSchema())
if err != nil {
@@ -75,14 +80,14 @@ func TestAnalyze_GroupByAgg_BuildsGroupByAggBelowLimit(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
GroupBy: &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
- FieldName: "value",
+ FieldName: fieldValue,
},
Agg: &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
},
}
p, err := Analyze(req, testMeasureSchema())
@@ -100,10 +105,10 @@ func TestAnalyze_GroupByAgg_BuildsGroupByAggBelowLimit(t *testing.T) {
if _, ok := gba.Children()[0].(*Scan); !ok {
t.Fatalf("GroupByAgg child should be *Scan, got %T", gba.Children()[0])
}
- if gba.GroupBy.TagNames[0] != "svc" {
+ if gba.GroupBy.TagNames[0] != tagSvc {
t.Fatalf("GroupBy.TagNames: want [svc], got %v", gba.GroupBy.TagNames)
}
- if gba.Agg.FieldName != "value" {
+ if gba.Agg.FieldName != fieldValue {
t.Fatalf("Agg.FieldName: want value, got %s", gba.Agg.FieldName)
}
}
@@ -112,14 +117,14 @@ func TestAnalyze_TopBetweenGroupByAggAndLimit(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
GroupBy: &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
- FieldName: "value",
+ FieldName: fieldValue,
},
Agg: &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
},
Top: &measurev1.QueryRequest_Top{
Number: 5,
@@ -155,10 +160,10 @@ func TestAnalyze_TopSortAsc_MapsToAscending(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
Top: &measurev1.QueryRequest_Top{
Number: 3,
- FieldName: "value",
+ FieldName: fieldValue,
FieldValueSort: modelv1.Sort_SORT_ASC,
},
}
@@ -185,10 +190,10 @@ func TestAnalyze_GroupByWithoutAgg_BuildsRawGroupBy(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
GroupBy: &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
- FieldName: "value",
+ FieldName: fieldValue,
},
}
p, err := Analyze(req, testMeasureSchema())
@@ -206,7 +211,7 @@ func TestAnalyze_GroupByWithoutAgg_BuildsRawGroupBy(t *testing.T) {
if gba.Agg != nil {
t.Fatalf("raw GroupBy must have nil Agg, got %+v", gba.Agg)
}
- if gba.GroupBy == nil || gba.GroupBy.TagNames[0] != "svc" {
+ if gba.GroupBy == nil || gba.GroupBy.TagNames[0] != tagSvc {
t.Fatalf("GroupBy.TagNames: want [svc], got %+v", gba.GroupBy)
}
}
@@ -217,10 +222,10 @@ func TestAnalyze_AggWithoutGroupBy_BuildsScalarReduce(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
Agg: &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
},
}
p, err := Analyze(req, testMeasureSchema())
@@ -238,7 +243,7 @@ func TestAnalyze_AggWithoutGroupBy_BuildsScalarReduce(t *testing.T) {
if gba.GroupBy != nil {
t.Fatalf("scalar reduce must have nil GroupBy, got %+v", gba.GroupBy)
}
- if gba.Agg == nil || gba.Agg.FieldName != "value" {
+ if gba.Agg == nil || gba.Agg.FieldName != fieldValue {
t.Fatalf("Agg.FieldName: want value, got %+v", gba.Agg)
}
}
@@ -247,16 +252,16 @@ func TestAnalyze_UnknownGroupByTag_Errors(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
GroupBy: &measurev1.QueryRequest_GroupBy{
TagProjection: &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{
{Name: "default", Tags: []string{"missing"}},
}},
- FieldName: "value",
+ FieldName: fieldValue,
},
Agg: &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
},
}
_, err := Analyze(req, testMeasureSchema())
@@ -272,10 +277,10 @@ func TestAnalyze_UnknownAggField_Errors(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
GroupBy: &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
- FieldName: "value",
+ FieldName: fieldValue,
},
Agg: &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
@@ -310,7 +315,7 @@ func TestAnalyze_DefaultLimit_AppliedWhenZero(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
// Limit unset (0) → default 100 per defaultLimit constant
}
p, err := Analyze(req, testMeasureSchema())
@@ -327,14 +332,14 @@ func TestPrintTree_RendersHierarchy(t *testing.T) {
req := &measurev1.QueryRequest{
Name: "demo",
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
GroupBy: &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
- FieldName: "value",
+ FieldName: fieldValue,
},
Agg: &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
},
}
p, err := Analyze(req, testMeasureSchema())
diff --git a/pkg/query/vectorized/measure/plan/build_test.go b/pkg/query/vectorized/measure/plan/build_test.go
index 88fc05a11..644873046 100644
--- a/pkg/query/vectorized/measure/plan/build_test.go
+++ b/pkg/query/vectorized/measure/plan/build_test.go
@@ -56,8 +56,8 @@ func buildScanInput(t *testing.T) (*vectorized.BatchSchema, *vectorized.RecordBa
{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.RoleField, Name: "value", Type: vectorized.ColumnTypeInt64},
+ {Role: vectorized.RoleTag, TagFamily: "default", Name: tagSvc, Type: vectorized.ColumnTypeString},
+ {Role: vectorized.RoleField, Name: fieldValue, Type: vectorized.ColumnTypeInt64},
})
b := vectorized.NewRecordBatch(schema, 5)
pushRow := func(ts int64, svc string, v int64) {
@@ -135,8 +135,8 @@ func TestScanGroupByAggLimit_Build_AggregatesByKey(t *testing.T) {
scan := NewScan(schema, ScanParams{})
scan.Source = src
gba, err := NewGroupByAgg(scan,
- &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{"svc"}},
- &model.MeasureAgg{FieldName: "value", Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{tagSvc}},
+ &model.MeasureAgg{FieldName: fieldValue, Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
)
if err != nil {
t.Fatalf("NewGroupByAgg: %v", err)
@@ -205,8 +205,8 @@ func TestGroupByAgg_Schema_DropsTimestampAddsAggField(t *testing.T) {
scan := NewScan(schema, ScanParams{})
scan.Source = src
gba, err := NewGroupByAgg(scan,
- &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{"svc"}},
- &model.MeasureAgg{FieldName: "value", Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{tagSvc}},
+ &model.MeasureAgg{FieldName: fieldValue, Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
)
if err != nil {
t.Fatalf("NewGroupByAgg: %v", err)
@@ -222,10 +222,10 @@ func TestGroupByAgg_Schema_DropsTimestampAddsAggField(t *testing.T) {
if len(out.Columns) != 2 {
t.Fatalf("want 2 output columns (key + agg result), got %d", len(out.Columns))
}
- if out.Columns[0].Name != "svc" {
+ if out.Columns[0].Name != tagSvc {
t.Fatalf("col 0 should be the svc key, got %s", out.Columns[0].Name)
}
- if out.Columns[1].Name != "value" {
+ if out.Columns[1].Name != fieldValue {
t.Fatalf("col 1 should be 'value' (row-path parity), got %s", out.Columns[1].Name)
}
}
diff --git a/pkg/query/vectorized/measure/plan/dispatch_test.go b/pkg/query/vectorized/measure/plan/dispatch_test.go
index 2e2a979a8..3b9fcf0ec 100644
--- a/pkg/query/vectorized/measure/plan/dispatch_test.go
+++ b/pkg/query/vectorized/measure/plan/dispatch_test.go
@@ -42,7 +42,7 @@ func bareReq() *measurev1.QueryRequest {
Name: "demo",
Groups: []string{"default"},
TagProjection: projTagProj(),
- FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{fieldValue}},
TimeRange: &modelv1.TimeRange{
Begin: timestamppb.New(time.Unix(0, 0)),
End: timestamppb.New(time.Unix(0, 1_000_000)),
@@ -72,7 +72,7 @@ func TestDispatch_GroupByWithoutAgg_FallsThrough(t *testing.T) {
req := bareReq()
req.GroupBy = &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
- FieldName: "value",
+ FieldName: fieldValue,
}
_, _, handled, err := Dispatch(context.Background(),
req, nil, nil, nil, nil, dispatchCfg(true))
@@ -90,7 +90,7 @@ func TestDispatch_AggWithoutGroupBy_FallsThrough(t *testing.T) {
req := bareReq()
req.Agg = &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
}
_, _, handled, err := Dispatch(context.Background(),
req, nil, nil, nil, nil, dispatchCfg(true))
@@ -118,16 +118,16 @@ func TestDispatch_GroupByAggUncoveredProjection_FallsThrough(t *testing.T) {
{
name: "groupby_tag_not_in_projection",
mutate: func(req *measurev1.QueryRequest) {
- // GroupBy references "region" but TagProjection only carries "svc".
+ // GroupBy references region but TagProjection only carries svc.
req.GroupBy = &measurev1.QueryRequest_GroupBy{
TagProjection: &modelv1.TagProjection{TagFamilies: []*modelv1.TagProjection_TagFamily{
{Name: "default", Tags: []string{"region"}},
}},
- FieldName: "value",
+ FieldName: fieldValue,
}
req.Agg = &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
}
},
},
@@ -136,13 +136,13 @@ func TestDispatch_GroupByAggUncoveredProjection_FallsThrough(t *testing.T) {
mutate: func(req *measurev1.QueryRequest) {
req.GroupBy = &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
- FieldName: "value",
+ FieldName: fieldValue,
}
req.Agg = &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
}
- // Strip "value" from FieldProjection so the Agg field is uncovered.
+ // Strip the value field from FieldProjection so the Agg field is uncovered.
req.FieldProjection = nil
},
},
@@ -182,7 +182,7 @@ func TestDispatch_Top_ReachesEcQuery(t *testing.T) {
req := bareReq()
req.Top = &measurev1.QueryRequest_Top{
Number: 5,
- FieldName: "value",
+ FieldName: fieldValue,
FieldValueSort: modelv1.Sort_SORT_DESC,
}
@@ -351,11 +351,14 @@ func TestDispatch_TagValidatedBeforeField(t *testing.T) {
{Name: "default", Tags: []string{"ghost"}},
}}
req.FieldProjection = &measurev1.QueryRequest_FieldProjection{Names: []string{"phantom"}}
- _, _, _, err := Dispatch(context.Background(),
+ _, _, handled, err := Dispatch(context.Background(),
req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
if err == nil || err.Error() != "ghost: tag is not defined" {
t.Fatalf("tag error must take precedence over field error; got %v", err)
}
+ if !handled {
+ t.Fatal("projection parity error must report handled=true (caller must not retry row path)")
+ }
}
// TestDispatch_NoTimeRange_EmptyResultParity covers G9c #9: a nil
@@ -489,7 +492,7 @@ func TestDispatch_Counters_TrackFellThroughCalls(t *testing.T) {
// schema/ec/metadata in the loop below). The counter is still
// exercised on the clean (non-error) fall-through path.
gbReq := bareReq()
- gbReq.GroupBy = &measurev1.QueryRequest_GroupBy{TagProjection: projTagProj(), FieldName: "value"}
+ gbReq.GroupBy = &measurev1.QueryRequest_GroupBy{TagProjection: projTagProj(), FieldName: fieldValue}
noTimeReq := bareReq()
noTimeReq.TimeRange = nil
@@ -531,11 +534,11 @@ func TestDispatch_GroupByAggCovered_ReachesEcQuery(t *testing.T) {
req := bareReq()
req.GroupBy = &measurev1.QueryRequest_GroupBy{
TagProjection: projTagProj(),
- FieldName: "value",
+ FieldName: fieldValue,
}
req.Agg = &measurev1.QueryRequest_Aggregation{
Function: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM,
- FieldName: "value",
+ FieldName: fieldValue,
}
iter, planStr, handled, err := Dispatch(context.Background(),
@@ -568,14 +571,14 @@ func TestAugmentRequestWithHiddenTags_AppendsFamiliesAfterVisible(t *testing.T)
t.Fatal("augment must return a clone when extras are present")
}
if len(req.GetTagProjection().GetTagFamilies()) != 1 ||
- req.GetTagProjection().GetTagFamilies()[0].GetTags()[0] != "svc" {
+ req.GetTagProjection().GetTagFamilies()[0].GetTags()[0] != tagSvc {
t.Fatalf("caller's req.TagProjection must be untouched, got %+v", req.GetTagProjection())
}
fams := got.GetTagProjection().GetTagFamilies()
if len(fams) != 3 {
t.Fatalf("want 3 families (1 visible + 2 hidden), got %d: %+v", len(fams), fams)
}
- if fams[0].GetName() != "default" || fams[0].GetTags()[0] != "svc" {
+ if fams[0].GetName() != "default" || fams[0].GetTags()[0] != tagSvc {
t.Fatalf("visible family must stay first, got %+v", fams[0])
}
if fams[1].GetName() != "default" || fams[1].GetTags()[0] != "region" {
@@ -600,7 +603,7 @@ func TestHiddenTagsMIterator_StripsHiddenTagsFromCurrent(t *testing.T) {
TagFamilies: []*modelv1.TagFamily{{
Name: "default",
Tags: []*modelv1.Tag{
- {Key: "svc", Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "a"}}}},
+ {Key: tagSvc, Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "a"}}}},
{Key: "region", Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "us"}}}},
},
}},
@@ -618,7 +621,7 @@ func TestHiddenTagsMIterator_StripsHiddenTagsFromCurrent(t *testing.T) {
t.Fatalf("expected one family after strip, got %+v", cur)
}
tags := cur[0].DataPoint.TagFamilies[0].Tags
- if len(tags) != 1 || tags[0].Key != "svc" {
+ if len(tags) != 1 || tags[0].Key != tagSvc {
t.Fatalf("only visible tag 'svc' must remain, got %+v", tags)
}
if it.Close() != nil {
diff --git a/pkg/query/vectorized/measure/plan/executor_test.go b/pkg/query/vectorized/measure/plan/executor_test.go
index cfce60e0f..dbedf6ea1 100644
--- a/pkg/query/vectorized/measure/plan/executor_test.go
+++ b/pkg/query/vectorized/measure/plan/executor_test.go
@@ -78,8 +78,8 @@ func TestExecute_GroupByAgg_EmitsAggregatedRowsWithNilTimestamp(t *testing.T) {
scan := NewScan(schema, ScanParams{})
scan.Source = src
gba, err := NewGroupByAgg(scan,
- &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{"svc"}},
- &model.MeasureAgg{FieldName: "value", Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{tagSvc}},
+ &model.MeasureAgg{FieldName: fieldValue, Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
)
if err != nil {
t.Fatalf("NewGroupByAgg: %v", err)
@@ -106,11 +106,11 @@ func TestExecute_GroupByAgg_EmitsAggregatedRowsWithNilTimestamp(t *testing.T) {
t.Fatalf("want one TagFamily 'default', got %+v", idp.DataPoint.TagFamilies)
}
tags := idp.DataPoint.TagFamilies[0].Tags
- if len(tags) != 1 || tags[0].Key != "svc" {
+ if len(tags) != 1 || tags[0].Key != tagSvc {
t.Fatalf("want one Tag 'svc', got %+v", tags)
}
svc := tags[0].Value.GetStr().GetValue()
- if len(idp.DataPoint.Fields) != 1 || idp.DataPoint.Fields[0].Name != "value" {
+ if len(idp.DataPoint.Fields) != 1 || idp.DataPoint.Fields[0].Name != fieldValue {
t.Fatalf("want one Field 'value' (row-path parity), got %+v", idp.DataPoint.Fields)
}
bySvc[svc] = idp.DataPoint.Fields[0].Value.GetInt().GetValue()
@@ -133,7 +133,7 @@ func TestExecute_ScalarReduce_EmitsSingleRow(t *testing.T) {
scan := NewScan(schema, ScanParams{})
scan.Source = src
gba, err := NewGroupByAgg(scan, nil,
- &model.MeasureAgg{FieldName: "value", Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
+ &model.MeasureAgg{FieldName: fieldValue, Func: modelv1.AggregationFunction_AGGREGATION_FUNCTION_SUM},
)
if err != nil {
t.Fatalf("NewGroupByAgg: %v", err)
@@ -160,7 +160,7 @@ func TestExecute_ScalarReduce_EmitsSingleRow(t *testing.T) {
}
tags := idp.DataPoint.TagFamilies[0].Tags
svc = tags[0].Value.GetStr().GetValue()
- if len(idp.DataPoint.Fields) != 1 || idp.DataPoint.Fields[0].Name != "value" {
+ if len(idp.DataPoint.Fields) != 1 || idp.DataPoint.Fields[0].Name != fieldValue {
t.Fatalf("want one Field 'value', got %+v", idp.DataPoint.Fields)
}
sum = idp.DataPoint.Fields[0].Value.GetInt().GetValue()
@@ -188,7 +188,7 @@ func TestExecute_RawGroupBy_EmitsFirstRowPerGroup(t *testing.T) {
scan := NewScan(schema, ScanParams{})
scan.Source = src
gba, err := NewGroupByAgg(scan,
- &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{"svc"}}, nil,
+ &model.MeasureGroupBy{TagFamily: "default", TagNames: []string{tagSvc}}, nil,
)
if err != nil {
t.Fatalf("NewGroupByAgg: %v", err)
@@ -215,7 +215,7 @@ func TestExecute_RawGroupBy_EmitsFirstRowPerGroup(t *testing.T) {
idp := dps[0]
// Raw GroupBy preserves the input schema, so the field column
// survives (unlike the aggregation shapes).
- if len(idp.DataPoint.Fields) != 1 || idp.DataPoint.Fields[0].Name != "value" {
+ if len(idp.DataPoint.Fields) != 1 || idp.DataPoint.Fields[0].Name != fieldValue {
t.Fatalf("raw GroupBy must preserve the field column, got %+v", idp.DataPoint.Fields)
}
svc := idp.DataPoint.TagFamilies[0].Tags[0].Value.GetStr().GetValue()
(skywalking-banyandb) 05/08: feat(query/vectorized/measure/plan): G9a single-node Top dispatch + analyzer sort-direction fix
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 9128d9fc2ac0c3c42af458133c1d75899f14adca
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 10:10:49 2026 +0000
feat(query/vectorized/measure/plan): G9a single-node Top dispatch + analyzer sort-direction fix
---
pkg/query/vectorized/measure/plan/analyzer.go | 6 ++-
pkg/query/vectorized/measure/plan/analyzer_test.go | 35 ++++++++++++++
pkg/query/vectorized/measure/plan/dispatch.go | 14 +++---
pkg/query/vectorized/measure/plan/dispatch_test.go | 46 ++++++++++++------
pkg/query/vectorized/measure/plan/top.go | 8 ++--
pkg/query/vectorized/measure/top.go | 54 +++++++++++++++++-----
pkg/query/vectorized/measure/top_test.go | 44 ++++++++++++++++++
7 files changed, 169 insertions(+), 38 deletions(-)
diff --git a/pkg/query/vectorized/measure/plan/analyzer.go b/pkg/query/vectorized/measure/plan/analyzer.go
index 6eee444fb..035ae3882 100644
--- a/pkg/query/vectorized/measure/plan/analyzer.go
+++ b/pkg/query/vectorized/measure/plan/analyzer.go
@@ -22,6 +22,7 @@ import (
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/model"
measure "github.com/apache/skywalking-banyandb/pkg/query/vectorized/measure"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
@@ -137,7 +138,10 @@ func Analyze(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (Ve
}
if t := req.GetTop(); t != nil {
- asc := t.GetFieldValueSort() == 1 // SORT_ASC == 1 in modelv1.Sort
+ // Match the row path (pkg/query/logical/measure.unresolvedTop):
+ // FieldValueSort==SORT_ASC keeps the lowest N (BatchTop asc=true);
+ // anything else (SORT_DESC / SORT_UNSPECIFIED) keeps the highest N.
+ asc := t.GetFieldValueSort() == modelv1.Sort_SORT_ASC
plan = NewTop(plan, t.GetFieldName(), int(t.GetNumber()), asc)
}
diff --git a/pkg/query/vectorized/measure/plan/analyzer_test.go b/pkg/query/vectorized/measure/plan/analyzer_test.go
index f7041dd87..544b6fc5b 100644
--- a/pkg/query/vectorized/measure/plan/analyzer_test.go
+++ b/pkg/query/vectorized/measure/plan/analyzer_test.go
@@ -142,6 +142,41 @@ func TestAnalyze_TopBetweenGroupByAggAndLimit(t *testing.T) {
if _, ok := top.Child.(*GroupByAgg); !ok {
t.Fatalf("Top child: want *GroupByAgg, got %T", top.Child)
}
+ // SORT_DESC must map to a descending top (asc=false) so the vec
+ // analyzer matches the row path's reverted semantics. Sort enum
+ // values are SORT_UNSPECIFIED=0, SORT_DESC=1, SORT_ASC=2 — a naive
+ // "==1 means asc" inverts every Top fixture.
+ if top.Asc {
+ t.Fatal("SORT_DESC must produce a descending Top (Asc=false)")
+ }
+}
+
+func TestAnalyze_TopSortAsc_MapsToAscending(t *testing.T) {
+ req := &measurev1.QueryRequest{
+ Name: "demo",
+ TagProjection: projTagProj(),
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{Names: []string{"value"}},
+ Top: &measurev1.QueryRequest_Top{
+ Number: 3,
+ FieldName: "value",
+ FieldValueSort: modelv1.Sort_SORT_ASC,
+ },
+ }
+ p, err := Analyze(req, testMeasureSchema())
+ if err != nil {
+ t.Fatalf("Analyze: %v", err)
+ }
+ limit, ok := p.(*Limit)
+ if !ok {
+ t.Fatalf("root: want *Limit, got %T", p)
+ }
+ top, ok := limit.Child.(*Top)
+ if !ok {
+ t.Fatalf("Limit child: want *Top, got %T", limit.Child)
+ }
+ if !top.Asc {
+ t.Fatal("SORT_ASC must produce an ascending Top (Asc=true)")
+ }
}
// TestAnalyze_GroupByWithoutAgg_BuildsRawGroupBy verifies the raw
diff --git a/pkg/query/vectorized/measure/plan/dispatch.go b/pkg/query/vectorized/measure/plan/dispatch.go
index bafceb089..c1864a21e 100644
--- a/pkg/query/vectorized/measure/plan/dispatch.go
+++ b/pkg/query/vectorized/measure/plan/dispatch.go
@@ -91,8 +91,9 @@ func FellThroughCount() int64 { return fellThroughCount.Load() }
// - request may carry GroupBy and/or Agg in any combination (group+agg,
// scalar reduce, raw GroupBy); plan.Analyze auto-extends the
// projection so the keys / agg field always resolve
-// - request must NOT carry Top (BatchTop's single-heap semantic differs
-// from the row path's per-timestamp top-N)
+// - request may carry Top: the analyzer emits Scan → Top → Limit
+// (or Scan → GroupByAgg → Top → Limit) and BatchTop reproduces the
+// row path's top-N (G9a)
// - request must carry TimeRange (storage requires a bounded window)
// - hidden criteria tags (criteria tags absent from the projection)
// are projected for storage-side filtering, then stripped at egress
@@ -126,12 +127,9 @@ func Dispatch(
if req == nil {
return nil, "", false, nil
}
- if req.GetTop() != nil {
- // Top awaits per-timestamp partitioning of BatchTop; the row
- // path's TopN semantic differs from BatchTop's single-heap
- // today.
- return nil, "", false, nil
- }
+ // Top is handled by the vec subsystem: plan.Analyze emits
+ // Scan → Top → Limit (or Scan → GroupByAgg → Top → Limit) and
+ // BatchTop reproduces the row path's top-N (G9a).
// GroupBy and Agg are handled by the vec subsystem in all three
// shapes — group+agg, scalar reduce (Agg only), raw GroupBy (GroupBy
// only). plan.Analyze auto-extends the projection so the GroupBy keys
diff --git a/pkg/query/vectorized/measure/plan/dispatch_test.go b/pkg/query/vectorized/measure/plan/dispatch_test.go
index a809995b2..2e2a979a8 100644
--- a/pkg/query/vectorized/measure/plan/dispatch_test.go
+++ b/pkg/query/vectorized/measure/plan/dispatch_test.go
@@ -163,21 +163,37 @@ func TestDispatch_GroupByAggUncoveredProjection_FallsThrough(t *testing.T) {
}
}
-// TestDispatch_Top_FallsThrough covers the per-timestamp top-N gap.
-func TestDispatch_Top_FallsThrough(t *testing.T) {
+// TestDispatch_Top_ReachesEcQuery confirms G9a removed the Top gate:
+// req.Top no longer triggers an eligibility fall-through. The request
+// proceeds to ec.Query, where the analyzer has emitted
+// Scan → Top → Limit. fakeEC returns nil so dispatch falls through after
+// ec.Query (the empty-result branch) — what matters is that ec.Query was
+// invoked at all, proving the Top gate no longer rejects the request.
+func TestDispatch_Top_ReachesEcQuery(t *testing.T) {
+ measureSchema := testMeasureSchema()
+ // nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
+ logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ metadata := &commonv1.Metadata{Name: "demo", Group: "default"}
+ ec := &fakeEC{wantResult: nil, wantErr: nil}
+
req := bareReq()
req.Top = &measurev1.QueryRequest_Top{
Number: 5,
FieldName: "value",
FieldValueSort: modelv1.Sort_SORT_DESC,
}
- _, _, handled, err := Dispatch(context.Background(),
- req, nil, nil, nil, nil, dispatchCfg(true))
+
+ iter, planStr, handled, err := Dispatch(context.Background(),
+ req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
if err != nil {
- t.Fatalf("Top fallthrough must not error: %v", err)
+ t.Fatalf("Top request must not error before ec.Query: %v", err)
}
- if handled {
- t.Fatal("Top must fall through (BatchTop semantics differ from row TopN)")
+ if !ec.called {
+ t.Fatalf("Top request must reach ec.Query (G9a removed the Top gate); "+
+ "got iter=%v planStr=%q handled=%v", iter, planStr, handled)
}
}
@@ -467,17 +483,17 @@ func TestDispatch_Counters_TrackFellThroughCalls(t *testing.T) {
startHandled := HandledCount()
startFellThrough := FellThroughCount()
- // Three fallthroughs of distinct shapes. topReq trips the Top gate;
- // the other two trip the PERMANENT nil-runtime-context guard (all-nil
- // schema/ec/metadata). Note: post-G9c a nil TimeRange is no longer a
- // fall-through on its own — noTimeReq falls through here only because
- // the runtime context is nil.
- topReq := bareReq()
- topReq.Top = &measurev1.QueryRequest_Top{Number: 5, FieldName: "value"}
+ // Post-G9 the Top / GroupBy / nil-TimeRange shapes are all handled by
+ // the vec subsystem, so these requests fall through only via the
+ // PERMANENT nil-runtime-context guard (Dispatch is called with all-nil
+ // schema/ec/metadata in the loop below). The counter is still
+ // exercised on the clean (non-error) fall-through path.
+ gbReq := bareReq()
+ gbReq.GroupBy = &measurev1.QueryRequest_GroupBy{TagProjection: projTagProj(), FieldName: "value"}
noTimeReq := bareReq()
noTimeReq.TimeRange = nil
- for _, req := range []*measurev1.QueryRequest{topReq, noTimeReq, bareReq() /* nil ec */} {
+ for _, req := range []*measurev1.QueryRequest{gbReq, noTimeReq, bareReq() /* nil ec */} {
_, _, handled, dispatchErr := Dispatch(context.Background(),
req, nil, nil, nil, nil, dispatchCfg(true))
if dispatchErr != nil {
diff --git a/pkg/query/vectorized/measure/plan/top.go b/pkg/query/vectorized/measure/plan/top.go
index 65dc62628..f80f1bfcc 100644
--- a/pkg/query/vectorized/measure/plan/top.go
+++ b/pkg/query/vectorized/measure/plan/top.go
@@ -26,9 +26,11 @@ import (
)
// Top selects the top-N (or bottom-N when Asc) rows by FieldName. Wraps
-// `measure.BatchTop`, which uses a single global heap — the row path's
-// per-timestamp TopN semantic is not yet reproduced here (BatchTop
-// extension is tracked separately; see G6 plan).
+// `measure.BatchTop`, a single global heap. This matches the row path's
+// req.Top handler (pkg/query/logical/measure.topOp), which also inserts
+// every data point into one TopQueue — req.Top is a whole-result top-N,
+// not the per-timestamp TopNQuery RPC (out of scope, see .omc/g9-plan.md
+// G9a).
//
// Schema-preserving.
type Top struct {
diff --git a/pkg/query/vectorized/measure/top.go b/pkg/query/vectorized/measure/top.go
index 4bf63e0ba..f58fc99b3 100644
--- a/pkg/query/vectorized/measure/top.go
+++ b/pkg/query/vectorized/measure/top.go
@@ -21,6 +21,7 @@ import (
"container/heap"
"context"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/pkg/query/vectorized"
)
@@ -147,9 +148,8 @@ func (t *BatchTop) Consume(_ context.Context, b *vectorized.RecordBatch) error {
return nil
}
active := activeIndices(b)
- isFloat := t.schema.Columns[t.fieldCol].Type == vectorized.ColumnTypeFloat64
for _, rowIdx := range active {
- candidate := t.materialize(b, int(rowIdx), isFloat)
+ candidate := t.materialize(b, int(rowIdx))
candidate.seq = t.inputCount
t.inputCount++
if t.heapState.Len() < t.n {
@@ -226,22 +226,54 @@ func (t *BatchTop) Close() error {
// materialize copies row rowIdx of b into a new topRow, reading the sort key
// from the configured field column.
-func (t *BatchTop) materialize(b *vectorized.RecordBatch, rowIdx int, isFloat bool) *topRow {
+//
+// The key column may be a native typed column (ColumnTypeInt64 /
+// ColumnTypeFloat64 — promoted when an Agg reduces over the field) or a
+// passthrough *modelv1.FieldValue column (ColumnTypeFieldValue — the
+// non-Agg Scan→Top→Limit path, where BuildBatchSchema leaves projected
+// 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, isFloat: isFloat}
+ row := &topRow{cols: cols}
keyCol := b.Columns[t.fieldCol]
- if keyCol.IsNull(rowIdx) {
+ switch c := keyCol.(type) {
+ case *vectorized.TypedColumn[float64]:
+ row.isFloat = true
+ if c.IsNull(rowIdx) {
+ row.isNull = true
+ return row
+ }
+ row.floatVal = c.Data()[rowIdx]
+ case *vectorized.TypedColumn[int64]:
+ if c.IsNull(rowIdx) {
+ row.isNull = true
+ return row
+ }
+ row.intVal = c.Data()[rowIdx]
+ case *vectorized.TypedColumn[*modelv1.FieldValue]:
+ fv := c.Data()[rowIdx]
+ switch v := fv.GetValue().(type) {
+ case *modelv1.FieldValue_Float:
+ row.isFloat = true
+ row.floatVal = v.Float.GetValue()
+ case *modelv1.FieldValue_Int:
+ row.intVal = v.Int.GetValue()
+ default:
+ // Null / unset / non-numeric field value: treat as lowest,
+ // matching the native-column null handling and the row path.
+ row.isNull = true
+ }
+ default:
+ // Unexpected key column shape (string / bytes / arrays): no
+ // numeric sort key. Treat as lowest so the query still completes
+ // rather than panicking.
row.isNull = true
- return row
- }
- if isFloat {
- row.floatVal = keyCol.(*vectorized.TypedColumn[float64]).Data()[rowIdx]
- } else {
- row.intVal = keyCol.(*vectorized.TypedColumn[int64]).Data()[rowIdx]
}
return row
}
diff --git a/pkg/query/vectorized/measure/top_test.go b/pkg/query/vectorized/measure/top_test.go
index 3515d2203..c68202d40 100644
--- a/pkg/query/vectorized/measure/top_test.go
+++ b/pkg/query/vectorized/measure/top_test.go
@@ -21,6 +21,7 @@ import (
"context"
"testing"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/pkg/query/vectorized"
)
@@ -205,6 +206,49 @@ func TestBatchTop_ZeroN_NoOp(t *testing.T) {
}
}
+// TestBatchTop_FieldValuePassthroughKey_Float pins the G9a regression: the
+// non-Agg Scan→Top→Limit path leaves the projected field as a passthrough
+// ColumnTypeFieldValue column (BuildBatchSchema only promotes Agg fields to
+// native typed columns). BatchTop must read the float sort key out of the
+// *modelv1.FieldValue rather than panicking on a TypedColumn[int64] cast.
+func TestBatchTop_FieldValuePassthroughKey_Float(t *testing.T) {
+ s := vectorized.NewBatchSchema([]vectorized.ColumnDef{
+ {Role: vectorized.RoleField, Name: "value", Type: vectorized.ColumnTypeFieldValue},
+ })
+ top := NewBatchTop(s, 0, 3, false, 8) // desc top-3
+ _ = top.Init(context.Background())
+ defer top.Close()
+
+ vals := []float64{12.5, 3.0, 99.25, 7.1, 42.0, 88.0}
+ b := vectorized.NewRecordBatch(s, len(vals))
+ col := b.Columns[0].(*vectorized.TypedColumn[*modelv1.FieldValue])
+ for _, v := range vals {
+ col.Append(&modelv1.FieldValue{Value: &modelv1.FieldValue_Float{Float: &modelv1.Float{Value: v}}})
+ }
+ b.Len = len(vals)
+
+ if err := top.Consume(context.Background(), b); err != nil {
+ t.Fatalf("Consume must not panic on passthrough FieldValue key: %v", err)
+ }
+ if err := top.Finalize(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ out, err := top.NextBatch(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if out == nil || out.Len != 3 {
+ t.Fatalf("want 3 rows, got %v", out)
+ }
+ got := out.Columns[0].(*vectorized.TypedColumn[*modelv1.FieldValue]).Data()
+ want := []float64{99.25, 88.0, 42.0} // highest 3, descending
+ for i := range want {
+ if g := got[i].GetFloat().GetValue(); g != want[i] {
+ t.Fatalf("row %d: want %v, got %v", i, want[i], g)
+ }
+ }
+}
+
func TestBatchTop_Close_AfterPartialConsume_ReleasesHeapAllocation(t *testing.T) {
s := topTestSchema()
top := NewBatchTop(s, 0, 3, true, 8)
(skywalking-banyandb) 03/08: feat(query/vectorized/measure/plan): G9d hidden criteria tags egress-strip wrapper
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 02d0c194018a71527eee37d2f8df887c6c8815b2
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 10:06:04 2026 +0000
feat(query/vectorized/measure/plan): G9d hidden criteria tags egress-strip wrapper
---
pkg/query/vectorized/measure/plan/dispatch.go | 32 +++++---
pkg/query/vectorized/measure/plan/dispatch_test.go | 95 ++++++++++++++++++++++
pkg/query/vectorized/measure/plan/hidden_tags.go | 94 +++++++++++++++++++++
.../standalone/query/vectorized_test.go | 10 ++-
4 files changed, 216 insertions(+), 15 deletions(-)
diff --git a/pkg/query/vectorized/measure/plan/dispatch.go b/pkg/query/vectorized/measure/plan/dispatch.go
index baa870a2b..8a2a7b573 100644
--- a/pkg/query/vectorized/measure/plan/dispatch.go
+++ b/pkg/query/vectorized/measure/plan/dispatch.go
@@ -93,8 +93,9 @@ func FellThroughCount() int64 { return fellThroughCount.Load() }
// - request must NOT carry Top (BatchTop's single-heap semantic differs
// from the row path's per-timestamp top-N)
// - request must carry TimeRange (storage requires a bounded window)
-// - request must NOT have hidden criteria tags (those need an egress
-// strip wrapper that v1 dispatch does not implement)
+// - hidden criteria tags (criteria tags absent from the projection)
+// are projected for storage-side filtering, then stripped at egress
+// by hiddenTagsMIterator so the wire format is byte-identical
// - measureSchema and logicalSchema must be non-nil
func Dispatch(
ctx context.Context,
@@ -176,9 +177,11 @@ func Dispatch(
}
// Hidden-tag detection: criteria may reference tags that are NOT in
- // the projection (they're needed only as filter inputs). The row
- // path strips them at egress via hiddenTagsMIterator. v1 dispatch
- // does not implement that strip yet, so fall through when present.
+ // the projection (they're needed only as filter inputs). Such tags
+ // are projected so storage can evaluate the criteria, then stripped
+ // at egress by hiddenTagsMIterator so the wire bytes stay identical
+ // to a query without hidden criteria tags (the row path does the
+ // same in unresolvedIndexScan.Analyze + resultMIterator).
projectedTagNames := projectedNames(req.GetTagProjection())
entityList := logicalSchema.EntityList()
entityMap := make(map[string]int, len(entityList))
@@ -191,13 +194,14 @@ func Dispatch(
for _, tf := range measureSchema.GetTagFamilies() {
familyNames = append(familyNames, tf.GetName())
}
- hidden, _ := logical.CollectHiddenCriteriaTags(
+ hidden, hiddenExtras := logical.CollectHiddenCriteriaTags(
req.GetCriteria(), projectedTagNames, entityMap, logicalSchema,
func() []string { return familyNames },
)
- if !hidden.IsEmpty() {
- return nil, "", false, nil
- }
+ // analyzeReq carries the hidden tags in its TagProjection so the Scan
+ // materializes them for storage-side filtering; req itself is left
+ // unchanged for the index.Query / opts wiring below.
+ analyzeReq := augmentRequestWithHiddenTags(req, hiddenExtras)
indexOrder, orderErr := resolveOrderBy(req.GetOrderBy(), logicalSchema)
if orderErr != nil {
@@ -218,8 +222,9 @@ func Dispatch(
return nil, "", true, fmt.Errorf("vec dispatch: build query: %w", qErr)
}
- // Build the structural plan tree.
- p, analyzeErr := Analyze(req, measureSchema)
+ // Build the structural plan tree from analyzeReq so the Scan's
+ // BatchSchema + opts.TagProjection carry the hidden criteria tags.
+ p, analyzeErr := Analyze(analyzeReq, measureSchema)
if analyzeErr != nil {
return nil, "", true, fmt.Errorf("vec dispatch: analyze: %w", analyzeErr)
}
@@ -285,6 +290,11 @@ func Dispatch(
// here.
return nil, "", true, fmt.Errorf("vec dispatch: execute: %w", execErr)
}
+ if !hidden.IsEmpty() {
+ // Strip the projected-for-filtering hidden tags before
+ // serialization so the wire bytes match a query without them.
+ iter = &hiddenTagsMIterator{inner: iter, hiddenTags: hidden}
+ }
return iter, p.String(), true, nil
}
diff --git a/pkg/query/vectorized/measure/plan/dispatch_test.go b/pkg/query/vectorized/measure/plan/dispatch_test.go
index 08f9dcb6c..a809995b2 100644
--- a/pkg/query/vectorized/measure/plan/dispatch_test.go
+++ b/pkg/query/vectorized/measure/plan/dispatch_test.go
@@ -27,6 +27,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/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"
measure "github.com/apache/skywalking-banyandb/pkg/query/vectorized/measure"
@@ -535,6 +536,100 @@ func TestDispatch_GroupByAggCovered_ReachesEcQuery(t *testing.T) {
}
}
+// TestAugmentRequestWithHiddenTags_AppendsFamiliesAfterVisible covers the
+// G9d projection-extension mechanism: hidden criteria tags (grouped by
+// family) are appended AFTER the visible projection so the visible tags
+// keep their projected order, the caller's request is never mutated, and
+// an empty extras slice returns the request unchanged (no clone).
+func TestAugmentRequestWithHiddenTags_AppendsFamiliesAfterVisible(t *testing.T) {
+ req := bareReq() // TagProjection: default=[svc]
+ extras := [][]*logical.Tag{
+ {logical.NewTag("default", "region")},
+ {logical.NewTag("extra", "zone")},
+ }
+ got := augmentRequestWithHiddenTags(req, extras)
+ if got == req {
+ t.Fatal("augment must return a clone when extras are present")
+ }
+ if len(req.GetTagProjection().GetTagFamilies()) != 1 ||
+ req.GetTagProjection().GetTagFamilies()[0].GetTags()[0] != "svc" {
+ t.Fatalf("caller's req.TagProjection must be untouched, got %+v", req.GetTagProjection())
+ }
+ fams := got.GetTagProjection().GetTagFamilies()
+ if len(fams) != 3 {
+ t.Fatalf("want 3 families (1 visible + 2 hidden), got %d: %+v", len(fams), fams)
+ }
+ if fams[0].GetName() != "default" || fams[0].GetTags()[0] != "svc" {
+ t.Fatalf("visible family must stay first, got %+v", fams[0])
+ }
+ if fams[1].GetName() != "default" || fams[1].GetTags()[0] != "region" {
+ t.Fatalf("first hidden family wrong, got %+v", fams[1])
+ }
+ if fams[2].GetName() != "extra" || fams[2].GetTags()[0] != "zone" {
+ t.Fatalf("second hidden family wrong, got %+v", fams[2])
+ }
+ if same := augmentRequestWithHiddenTags(req, nil); same != req {
+ t.Fatal("augment with no extras must return the original request")
+ }
+}
+
+// TestHiddenTagsMIterator_StripsHiddenTagsFromCurrent proves the egress
+// wrapper removes exactly the hidden tags from each Current() DataPoint,
+// leaving visible tags untouched — the same contract as the row path's
+// hiddenTagsMIterator, which is what keeps the wire bytes identical.
+func TestHiddenTagsMIterator_StripsHiddenTagsFromCurrent(t *testing.T) {
+ build := func() []*measurev1.InternalDataPoint {
+ return []*measurev1.InternalDataPoint{{
+ DataPoint: &measurev1.DataPoint{
+ TagFamilies: []*modelv1.TagFamily{{
+ Name: "default",
+ Tags: []*modelv1.Tag{
+ {Key: "svc", Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "a"}}}},
+ {Key: "region", Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "us"}}}},
+ },
+ }},
+ },
+ }}
+ }
+ hidden := logical.NewHiddenTagSet()
+ hidden.Add("region")
+ it := &hiddenTagsMIterator{inner: &stubMIterator{rows: build()}, hiddenTags: hidden}
+ if !it.Next() {
+ t.Fatal("Next must advance once")
+ }
+ cur := it.Current()
+ if len(cur) != 1 || len(cur[0].DataPoint.TagFamilies) != 1 {
+ t.Fatalf("expected one family after strip, got %+v", cur)
+ }
+ tags := cur[0].DataPoint.TagFamilies[0].Tags
+ if len(tags) != 1 || tags[0].Key != "svc" {
+ t.Fatalf("only visible tag 'svc' must remain, got %+v", tags)
+ }
+ if it.Close() != nil {
+ t.Fatal("Close must propagate inner Close (nil)")
+ }
+}
+
+// stubMIterator is a one-shot executor.MIterator over a fixed row slice.
+type stubMIterator struct {
+ rows []*measurev1.InternalDataPoint
+ pos int
+}
+
+func (s *stubMIterator) Next() bool {
+ s.pos++
+ return s.pos <= len(s.rows)
+}
+
+func (s *stubMIterator) Current() []*measurev1.InternalDataPoint {
+ if s.pos < 1 || s.pos > len(s.rows) {
+ return nil
+ }
+ return s.rows[s.pos-1 : s.pos]
+}
+
+func (s *stubMIterator) Close() error { return nil }
+
// TestDispatch_QueryError_BubblesUp covers the error propagation when
// the storage query itself fails. Dispatch must report (nil, "", true,
// err) so the caller surfaces the error rather than re-trying the row
diff --git a/pkg/query/vectorized/measure/plan/hidden_tags.go b/pkg/query/vectorized/measure/plan/hidden_tags.go
new file mode 100644
index 000000000..bb0726a06
--- /dev/null
+++ b/pkg/query/vectorized/measure/plan/hidden_tags.go
@@ -0,0 +1,94 @@
+// 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 (
+ "google.golang.org/protobuf/proto"
+
+ 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"
+)
+
+// augmentRequestWithHiddenTags returns a request whose TagProjection also
+// carries the hidden criteria tags (grouped by family) so the vec Scan
+// materializes them as columns. Storage needs the hidden tag values to
+// evaluate criteria conditions on non-indexed tags; the row path performs
+// the same projection extension in unresolvedIndexScan.Analyze before
+// stripping the hidden tags at egress.
+//
+// The original request is left untouched: a shallow proto clone is taken
+// and only its TagProjection is replaced. Hidden families are appended
+// AFTER the visible projection families so the visible tags retain their
+// projected order. BatchSchema coalesces same-named families into one
+// TagFamilyGroup, so a hidden tag whose family is already projected lands
+// in that family's group after the visible columns — and is removed again
+// by the egress strip, leaving the wire bytes identical to a query with
+// no hidden tags.
+func augmentRequestWithHiddenTags(req *measurev1.QueryRequest, extras [][]*logical.Tag) *measurev1.QueryRequest {
+ if len(extras) == 0 {
+ return req
+ }
+ families := make([]*modelv1.TagProjection_TagFamily, 0)
+ if tp := req.GetTagProjection(); tp != nil {
+ families = append(families, tp.GetTagFamilies()...)
+ }
+ for _, group := range extras {
+ if len(group) == 0 {
+ continue
+ }
+ names := make([]string, 0, len(group))
+ for _, t := range group {
+ names = append(names, t.GetTagName())
+ }
+ families = append(families, &modelv1.TagProjection_TagFamily{
+ Name: group[0].GetFamilyName(),
+ Tags: names,
+ })
+ }
+ cloned := proto.Clone(req).(*measurev1.QueryRequest)
+ cloned.TagProjection = &modelv1.TagProjection{TagFamilies: families}
+ return cloned
+}
+
+// hiddenTagsMIterator wraps an MIterator and strips hidden criteria tags
+// from each Current() result. It is the vec counterpart of the row path's
+// hiddenTagsMIterator (pkg/query/logical/measure): the hidden tags were
+// projected so storage could filter on them, and are removed here so the
+// serialized DataPoints carry only the requested projection — keeping the
+// wire bytes byte-identical to a query without hidden criteria tags.
+type hiddenTagsMIterator struct {
+ inner executor.MIterator
+ hiddenTags logical.HiddenTagSet
+}
+
+func (h *hiddenTagsMIterator) Next() bool { return h.inner.Next() }
+
+func (h *hiddenTagsMIterator) Current() []*measurev1.InternalDataPoint {
+ dps := h.inner.Current()
+ for _, dp := range dps {
+ if dp == nil || dp.DataPoint == nil {
+ continue
+ }
+ dp.DataPoint.TagFamilies = h.hiddenTags.StripHiddenTags(dp.DataPoint.TagFamilies)
+ }
+ return dps
+}
+
+func (h *hiddenTagsMIterator) Close() error { return h.inner.Close() }
diff --git a/test/integration/standalone/query/vectorized_test.go b/test/integration/standalone/query/vectorized_test.go
index abd037352..6be0a7432 100644
--- a/test/integration/standalone/query/vectorized_test.go
+++ b/test/integration/standalone/query/vectorized_test.go
@@ -47,10 +47,12 @@ import (
//
// With G8d (top-level vec dispatch) wired, plain measure queries take the
// new pkg/query/vectorized/measure/plan.Dispatch path instead of the legacy
-// leaf-substitution at localIndexScan.maybeVectorized. GroupBy/Agg/Top and
-// queries with hidden criteria tags continue through the row plan with leaf
-// substitution. The AfterAll assertion below uses vecplan.HandledCount to
-// confirm dispatch actually fires for at least one query — protecting
+// leaf-substitution at localIndexScan.maybeVectorized. Queries with hidden
+// criteria tags (G9d) also route through Dispatch: the hidden tags are
+// projected for storage filtering and stripped at egress so the wire bytes
+// match the row path. GroupBy/Agg/Top continue through the row plan with
+// leaf substitution. The AfterAll assertion below uses vecplan.HandledCount
+// to confirm dispatch actually fires for at least one query — protecting
// against a silent regression where the eligibility gate excludes
// everything.
//
(skywalking-banyandb) 01/08: feat(query/vectorized/measure/plan): resolve order_by in vec dispatch
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 8f21383d9b10d0a9fe2dc93d9dab3b4b3694d5d3
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri May 15 01:49:20 2026 +0000
feat(query/vectorized/measure/plan): resolve order_by in vec dispatch
Lift the order_by bail-out and thread req.OrderBy through
logical.ParseOrderBy into MeasureQueryOptions.Order, mirroring the
row path's PushDownOrder optimizer rule. Empty index rule +
UNSPECIFIED sort yields no Order (matching the row path's no-order
default); unknown index rules surface as a dispatch error with
handled=true so the caller does not silently retry the row path.
Single-node order_by queries now clear the G8e parity gate
(518/518 specs pass with proto.Equal) instead of falling through.
Replace TestDispatch_OrderBy_FallsThrough — which passed only because
bareReq's nil schemas caused an earlier bail-out — with two tests
that exercise the new code: TestDispatch_OrderBy_ReachesEcQuery
covers the time-only ordering path;
TestDispatch_OrderBy_UnknownIndexRule_BubblesUpError covers the
parse-error branch.
---
CHANGES.md | 2 +-
pkg/query/vectorized/measure/plan/dispatch.go | 39 ++++++++++---
pkg/query/vectorized/measure/plan/dispatch_test.go | 67 ++++++++++++++++++----
3 files changed, 89 insertions(+), 19 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 6d9c3f311..49b87c729 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -5,7 +5,7 @@ Release Notes.
## 0.11.0
### Features
-- Vectorized measure query path is now enabled by default. The columnar pipeline replaces per-row protobuf serialization in `NewMIterator`, cutting allocations and ns/op for scan-heavy measure queries; gRPC wire format (`*measurev1.InternalDataPoint`) is byte-identical. Coverage extends to single-node GroupBy+Agg via `BatchAggregation` (SUM/COUNT/MIN/MAX/MEAN with row-path-equivalent type semantics and first-seen carry-forward of non-key projected tags). Distributed Map-mode GroupBy+Agg, [...]
+- Vectorized measure query path is now enabled by default. The columnar pipeline replaces per-row protobuf serialization in `NewMIterator`, cutting allocations and ns/op for scan-heavy measure queries; gRPC wire format (`*measurev1.InternalDataPoint`) is byte-identical. Coverage extends to single-node GroupBy+Agg via `BatchAggregation` (SUM/COUNT/MIN/MAX/MEAN with row-path-equivalent type semantics and first-seen carry-forward of non-key projected tags). Distributed Map-mode GroupBy+Agg, [...]
- Add validation to ensure Measure's ShardingKey contains all Entity tags to guarantee entity locality.
- Organize access logs under a dedicated "accesslog" subdirectory to improve log organization and separation from other application data.
- Collect BanyanDB data on e2e test failure for CI debugging.
diff --git a/pkg/query/vectorized/measure/plan/dispatch.go b/pkg/query/vectorized/measure/plan/dispatch.go
index d1d07fc2f..3ada53ef1 100644
--- a/pkg/query/vectorized/measure/plan/dispatch.go
+++ b/pkg/query/vectorized/measure/plan/dispatch.go
@@ -146,14 +146,6 @@ func Dispatch(
return nil, "", false, nil
}
}
- if req.GetOrderBy() != nil {
- // The row path resolves order_by via the PushDownOrder optimizer
- // rule (logical.NewPushDownOrder applied after Analyze). The vec
- // dispatch does not invoke those rules, so it would silently
- // drop OrderBy and return unsorted rows. Fall through until
- // dispatch threads order_by into model.MeasureQueryOptions.Order.
- return nil, "", false, nil
- }
if req.GetTimeRange() == nil {
return nil, "", false, nil
}
@@ -197,6 +189,11 @@ func Dispatch(
return nil, "", false, nil
}
+ indexOrder, orderErr := resolveOrderBy(req.GetOrderBy(), logicalSchema)
+ if orderErr != nil {
+ return nil, "", true, orderErr
+ }
+
// Resolve the index.Query + entities the same way the row path does
// in unresolvedIndexScan.Analyze.
var query index.Query
@@ -242,6 +239,7 @@ func Dispatch(
TimeRange: scan.Params.TimeRange,
Entities: entities,
Query: query,
+ Order: indexOrder,
GroupBy: scan.Params.GroupBy,
Agg: scan.Params.Agg,
TagProjection: scan.Params.TagProjection,
@@ -277,6 +275,31 @@ func Dispatch(
return iter, p.String(), true, nil
}
+// resolveOrderBy mirrors the row path's PushDownOrder optimizer rule.
+// Empty index rule + UNSPECIFIED sort yields (nil, nil) so dispatch
+// leaves opts.Order unset, matching the row path's no-order default.
+// ParseOrderBy errors on an unknown index rule or one with NoSort=true
+// — surface that error so dispatch reports handled=true rather than
+// silently retrying the row path, which would produce the same error
+// downstream.
+func resolveOrderBy(reqOrder *modelv1.QueryOrder, schema logical.Schema) (*index.OrderBy, error) {
+ if reqOrder == nil {
+ return nil, nil
+ }
+ parsed, err := logical.ParseOrderBy(schema, reqOrder.GetIndexRuleName(), reqOrder.GetSort())
+ if err != nil {
+ return nil, fmt.Errorf("vec dispatch: parse order_by: %w", err)
+ }
+ if parsed == nil {
+ return nil, nil
+ }
+ out := &index.OrderBy{Sort: parsed.Sort, Index: parsed.Index, Type: index.OrderByTypeIndex}
+ if parsed.Index == nil {
+ out.Type = index.OrderByTypeTime
+ }
+ return out, nil
+}
+
// locateScan walks a vec plan tree to find the leaf Scan node. Today there
// is exactly one Scan per plan (multi-measure merge is a G8 follow-up).
func locateScan(p VecPlan) *Scan {
diff --git a/pkg/query/vectorized/measure/plan/dispatch_test.go b/pkg/query/vectorized/measure/plan/dispatch_test.go
index d0c923e4b..d83a29191 100644
--- a/pkg/query/vectorized/measure/plan/dispatch_test.go
+++ b/pkg/query/vectorized/measure/plan/dispatch_test.go
@@ -180,22 +180,69 @@ func TestDispatch_Top_FallsThrough(t *testing.T) {
}
}
-// TestDispatch_OrderBy_FallsThrough covers the order_by gap. The row
-// path resolves OrderBy via the PushDownOrder optimizer rule which the
-// vec dispatch does not invoke. Until dispatch threads OrderBy into
-// MeasureQueryOptions.Order, requests with OrderBy must fall through.
-func TestDispatch_OrderBy_FallsThrough(t *testing.T) {
+// TestDispatch_OrderBy_ReachesEcQuery confirms dispatch resolves
+// req.OrderBy via logical.ParseOrderBy and threads it into
+// MeasureQueryOptions.Order, instead of falling through to the row
+// path. fakeEC returns nil so dispatch falls through after ec.Query
+// — what matters is that ec.Query was reached at all, proving the
+// OrderBy gate no longer rejects the request.
+func TestDispatch_OrderBy_ReachesEcQuery(t *testing.T) {
+ measureSchema := testMeasureSchema()
+ // nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
+ logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ metadata := &commonv1.Metadata{Name: "demo", Group: "default"}
+ ec := &fakeEC{wantResult: nil, wantErr: nil}
+
req := bareReq()
req.OrderBy = &modelv1.QueryOrder{
Sort: modelv1.Sort_SORT_DESC,
}
- _, _, handled, err := Dispatch(context.Background(),
- req, nil, nil, nil, nil, dispatchCfg(true))
+
+ iter, planStr, handled, err := Dispatch(context.Background(),
+ req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
if err != nil {
- t.Fatalf("OrderBy fallthrough must not error: %v", err)
+ t.Fatalf("OrderBy must not error before ec.Query: %v", err)
}
- if handled {
- t.Fatal("OrderBy must fall through (row path applies it via PushDownOrder)")
+ if !ec.called {
+ t.Fatalf("OrderBy must reach ec.Query (no longer falls through); "+
+ "got iter=%v planStr=%q handled=%v", iter, planStr, handled)
+ }
+}
+
+// TestDispatch_OrderBy_UnknownIndexRule_BubblesUpError covers the
+// error branch dispatch added when threading OrderBy through
+// logical.ParseOrderBy: an unknown index rule name must surface as a
+// dispatch error with handled=true so the caller does not silently
+// retry the row path (which would produce the same canonical error).
+func TestDispatch_OrderBy_UnknownIndexRule_BubblesUpError(t *testing.T) {
+ measureSchema := testMeasureSchema()
+ // nolint:staticcheck // SA1019 — row-path BuildSchema is the only schema builder until G8 replaces it.
+ logicalSchema, schemaErr := logicalmeasure.BuildSchema(measureSchema, nil)
+ if schemaErr != nil {
+ t.Fatalf("BuildSchema: %v", schemaErr)
+ }
+ metadata := &commonv1.Metadata{Name: "demo", Group: "default"}
+ ec := &fakeEC{wantResult: nil, wantErr: nil}
+
+ req := bareReq()
+ req.OrderBy = &modelv1.QueryOrder{
+ IndexRuleName: "no_such_index_rule",
+ Sort: modelv1.Sort_SORT_ASC,
+ }
+
+ _, _, handled, err := Dispatch(context.Background(),
+ req, metadata, measureSchema, logicalSchema, ec, dispatchCfg(true))
+ if err == nil {
+ t.Fatal("unknown OrderBy index rule must surface as a dispatch error")
+ }
+ if !handled {
+ t.Fatal("unknown OrderBy index rule must report handled=true so caller does not re-try row path")
+ }
+ if ec.called {
+ t.Fatal("unknown OrderBy index rule must error before ec.Query is invoked")
}
}