(skywalking-banyandb) 01/08: feat(query/vectorized/measure/plan): resolve order_by in vec dispatch

[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")
 	}
 }