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 2025/07/13 02:56:06 UTC
(skywalking-banyandb) branch topn created (now ceb91374)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch topn
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at ceb91374 Add stream write queue
This branch includes the following new commits:
new 09c558b9 Fix topN parsing panic when the criteria is set
new 03f338e2 Fix errors
new bc1cb31e Merge remote-tracking branch 'origin/main' into topn
new ceb91374 Add stream write queue
The 4 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.
(skywalking-banyandb) 02/04: Fix errors
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch topn
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 03f338e232251d21895cf3d9f0cb9b562b872db0
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Jun 25 21:29:27 2025 +0800
Fix errors
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/metadata.go | 2 +-
.../topn_aggregations/not_in_svc2_svc4 copy.json | 28 ----------------------
2 files changed, 1 insertion(+), 29 deletions(-)
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 0edad3ac..a53930af 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -344,7 +344,7 @@ func (sr *schemaRepo) stopAllProcessorsWithGroupPrefix(groupName string) {
var keysToDelete []string
groupPrefix := groupName + "/"
- sr.topNProcessorMap.Range(func(key, val any) bool {
+ sr.topNProcessorMap.Range(func(key, _ any) bool {
keyStr := key.(string)
if strings.HasPrefix(keyStr, groupPrefix) {
keysToDelete = append(keysToDelete, keyStr)
diff --git a/pkg/test/measure/testdata/topn_aggregations/not_in_svc2_svc4 copy.json b/pkg/test/measure/testdata/topn_aggregations/not_in_svc2_svc4 copy.json
deleted file mode 100644
index d95e3c7e..00000000
--- a/pkg/test/measure/testdata/topn_aggregations/not_in_svc2_svc4 copy.json
+++ /dev/null
@@ -1,28 +0,0 @@
-{
- "metadata": {
- "name": "not_in_svc2_svc4",
- "group": "sw_metric"
- },
- "source_measure": {
- "name": "service_instance_endpoint_cpm_minute",
- "group": "sw_metric"
- },
- "field_name": "total",
- "field_value_sort": 0,
- "group_by_tag_names": [
- "http.uri"
- ],
- "criteria": {
- "condition":{
- "name": "service_id",
- "op": "BINARY_OP_NOT_IN",
- "value": {
- "str_array": {
- "value": ["svc_2", "svc_4"]
- }
- }
- }
- },
- "counters_number": 1000,
- "lru_size": 10
-}
\ No newline at end of file
(skywalking-banyandb) 04/04: Add stream write queue
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch topn
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit ceb91374ea5ae7e510cd0478bc7764246dc81316
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun Jul 13 10:52:02 2025 +0800
Add stream write queue
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
.cursorrules | 313 +++++++++
api/data/data.go | 30 +-
api/data/stream.go | 27 +
banyand/internal/storage/rotation.go | 2 +-
banyand/internal/storage/segment.go | 6 +-
banyand/internal/storage/storage.go | 6 +-
banyand/internal/wqueue/shard.go | 42 ++
banyand/internal/wqueue/wqueue.go | 222 ++++++
banyand/internal/wqueue/wqueue_test.go | 139 ++++
banyand/liaison/grpc/registry_test.go | 1 -
banyand/liaison/grpc/server.go | 14 +-
banyand/liaison/grpc/stream.go | 78 ---
banyand/measure/introducer.go | 12 +-
banyand/measure/metrics.go | 8 +-
banyand/queue/pub/pub.go | 19 +-
banyand/queue/sub/sub.go | 12 +-
banyand/stream/block_scanner.go | 3 +
banyand/stream/elements.go | 29 +-
banyand/stream/flusher.go | 43 ++
banyand/stream/introducer.go | 90 ++-
banyand/stream/metadata.go | 129 +++-
banyand/stream/metrics.go | 83 ++-
banyand/stream/part.go | 293 ++++++++
banyand/stream/part_test.go | 319 +++++++++
banyand/stream/snapshot.go | 5 +-
banyand/stream/stream.go | 4 +-
banyand/stream/{service.go => svc_liaison.go} | 136 +---
banyand/stream/{service.go => svc_standalone.go} | 55 +-
banyand/stream/syncer.go | 160 +++++
banyand/stream/tstable.go | 77 ++-
banyand/stream/wqueue.go | 47 ++
banyand/stream/wqueue_test.go | 90 +++
banyand/stream/write.go | 46 +-
banyand/stream/write_liaison.go | 244 +++++++
banyand/stream/write_standalone.go | 283 ++++++++
pkg/cmdsetup/liaison.go | 12 +-
pkg/cmdsetup/standalone.go | 1 -
pkg/index/document_test.go | 817 +++++++++++++++++++++++
pkg/index/field_key_test.go | 283 ++++++++
pkg/index/field_test.go | 158 +++++
pkg/index/index.go | 497 +++++++++++++-
pkg/index/range_opts_test.go | 142 ++++
pkg/index/term_value_test.go | 263 ++++++++
pkg/run/run.go | 4 +-
test/cases/stream/stream.go | 8 +-
45 files changed, 4908 insertions(+), 344 deletions(-)
diff --git a/.cursorrules b/.cursorrules
new file mode 100644
index 00000000..6fd656d8
--- /dev/null
+++ b/.cursorrules
@@ -0,0 +1,313 @@
+# 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.
+
+# Cursor Rules for SkyWalking BanyanDB Go Project
+# This file configures Cursor to generate Go code that follows the project's formatting and linting standards
+
+# Language-specific rules for Go
+[go]
+description = """
+When generating Go code for the SkyWalking BanyanDB project, follow these strict rules:
+
+## FORMATTING RULES (from .golangci-format.yml):
+1. Use gofumpt for code formatting (stricter than gofmt)
+2. Use gci for import organization with specific sections:
+ - standard library imports
+ - default imports
+ - github.com/apache/skywalking-banyandb/ prefix imports
+3. Maximum line length: 170 characters (from .golangci.yml lll setting)
+
+## LINTING RULES (from .golangci.yml):
+1. Variable shadowing prevention (govet shadow enabled)
+2. Import aliases for specific packages (importas settings)
+3. Error handling patterns (errcheck, errorlint, errname)
+4. Code style and conventions (gosimple, staticcheck, stylecheck)
+5. Security considerations (gosec)
+6. Documentation standards (godot scope: toplevel)
+
+## IMPORT ORGANIZATION:
+Follow the gci sections order:
+1. Standard library imports
+2. Default imports
+3. github.com/apache/skywalking-banyandb/ prefix imports
+
+## IMPORT ALIASES (from importas settings):
+- github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1 → commonv1
+- github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1 → databasev1
+- github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1 → modelv1
+- github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1 → propertyv1
+- github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1 → measurev1
+- github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1 → streamv1
+- github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1 → clusterv1
+- github.com/apache/skywalking-banyandb/pkg/pb/v1 → pbv1
+
+## ERROR HANDLING PATTERNS:
+1. Always check errors immediately after function calls
+2. Use descriptive error variable names to avoid shadowing
+3. Follow the errname convention for error types
+4. Use errorlint for consistent error handling
+
+## VARIABLE SHADOWING PREVENTION:
+1. NEVER shadow variables in the same scope
+2. Use unique variable names in nested scopes
+3. Be especially careful with common variable names like 'err', 'ctx', 'v', 'i', 'n', etc.
+4. When using range loops, use descriptive variable names instead of single letters
+5. In if statements with variable declarations, ensure the variable name doesn't conflict with outer scope
+
+## CODE STYLE:
+1. Use gocritic enabled checks for code quality
+2. Follow exhaustive switch/map patterns
+3. Use proper field alignment (govet fieldalignment)
+4. Avoid unnecessary conversions (unconvert)
+5. Use standard library variables when available (usestdlibvars)
+6. Follow whitespace rules (whitespace linter)
+
+## SECURITY:
+1. Follow gosec security guidelines
+2. Be aware of integer overflow conversions (G115 excluded in config)
+
+## DOCUMENTATION:
+1. Add top-level comments for exported functions and types (godot scope: toplevel)
+2. Follow docStub patterns for documentation
+
+## CONSTANTS AND MAGIC NUMBERS:
+1. Use goconst for repeated string literals (min-occurrences: 4)
+2. Avoid magic numbers, use named constants
+
+## COMPLEXITY:
+1. Keep cyclomatic complexity low (gocyclo)
+2. Avoid deeply nested structures
+3. Use proper abstraction levels
+
+## NAMING CONVENTIONS:
+1. Use descriptive names for variables, especially in nested scopes
+2. Follow Go naming conventions
+3. Use proper package naming
+4. Follow the project's specific naming patterns
+
+## TESTING:
+1. Follow proper test naming conventions
+2. Use appropriate test helpers
+3. Follow the project's testing patterns
+
+## EXCLUDED PATTERNS:
+The following file patterns are excluded from linting:
+- *.pb.go (protobuf generated files)
+- *.pb.gw.go (protobuf gateway files)
+- *.gen.go (generated files)
+- *_mock.go (mock files)
+- *_test.go files have relaxed errcheck rules
+"""
+
+# Specific patterns to avoid
+avoid_patterns = [
+ # Variable shadowing patterns
+ "if err := someFunc(); err != nil {",
+ "if v := someValue(); v != nil {",
+ "for _, v := range items {",
+ "for i, v := range items {",
+ "switch v := someValue(); v {",
+ "defer func() { if err := cleanup(); err != nil {",
+
+ # Import organization violations
+ "import (\n\t\"fmt\"\n\t\"github.com/apache/skywalking-banyandb/pkg/something\"\n\t\"os\"\n)",
+
+ # Error handling violations
+ "someFunc() // ignoring error",
+ "if err != nil {\n\treturn\n}",
+
+ # Style violations
+ "if x == true {",
+ "if x == false {",
+ "if x != nil && x.y == true {",
+
+ # Documentation violations
+ "// Function does something",
+ "// This function does something",
+]
+
+# Preferred patterns to use
+preferred_patterns = [
+ # Proper variable naming to avoid shadowing
+ "if resultErr := someFunc(); resultErr != nil {",
+ "if item := someValue(); item != nil {",
+ "for _, item := range items {",
+ "for idx, item := range items {",
+ "switch value := someValue(); value {",
+ "defer func() { if cleanupErr := cleanup(); cleanupErr != nil {",
+
+ # Proper import organization
+ "import (\n\t\"fmt\"\n\t\"os\"\n\n\t\"github.com/apache/skywalking-banyandb/pkg/something\"\n)",
+
+ # Proper error handling
+ "if err := someFunc(); err != nil {\n\treturn fmt.Errorf(\"failed to do something: %w\", err)\n}",
+
+ # Proper style
+ "if x {",
+ "if !x {",
+ "if x != nil && x.y {",
+
+ # Proper documentation
+ "// DoSomething performs a specific action.",
+ "// DoSomething performs a specific action and returns an error if it fails.",
+]
+
+# Code generation guidelines
+guidelines = """
+When generating Go code for this project:
+
+## IMPORTANT: Always follow these steps:
+1. Check for existing variable names in the current scope before declaring new ones
+2. Use descriptive variable names to avoid shadowing
+3. Organize imports according to gci sections
+4. Use proper import aliases for the project's protobuf packages
+5. Follow error handling patterns with proper error wrapping
+6. Add appropriate documentation for exported functions and types
+7. Keep line length under 170 characters
+8. Use gofumpt formatting style
+
+## Example of good code structure:
+```go
+package example
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// ProcessData processes the given data and returns the result.
+func ProcessData(ctx context.Context, data []byte) ([]byte, error) {
+ if len(data) == 0 {
+ return nil, fmt.Errorf("data cannot be empty")
+ }
+
+ // Use descriptive variable names to avoid shadowing
+ for idx, item := range data {
+ if processErr := processItem(ctx, item); processErr != nil {
+ return nil, fmt.Errorf("failed to process item at index %d: %w", idx, processErr)
+ }
+ }
+
+ result, err := finalizeProcessing(ctx, data)
+ if err != nil {
+ return nil, fmt.Errorf("failed to finalize processing: %w", err)
+ }
+
+ return result, nil
+}
+```
+
+## Example of bad code (violates multiple rules):
+```go
+package example
+
+import (
+ "fmt"
+ "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1" // wrong alias
+ "os"
+)
+
+func ProcessData(ctx context.Context, data []byte) ([]byte, error) {
+ for _, v := range data {
+ if err := processItem(ctx, v); err != nil { // 'err' shadows outer scope
+ return err
+ }
+
+ for i, v := range v.SubItems { // 'v' shadows outer 'v'
+ if err := processSubItem(v); err != nil { // 'err' shadows outer 'err'
+ return err
+ }
+ }
+ }
+ return nil
+}
+```
+"""
+
+# Linting integration
+linting = """
+This project uses golangci-lint with strict configuration.
+Always run 'golangci-lint run' to verify generated code.
+Key linters enabled:
+- govet (including shadow and fieldalignment)
+- gci (import organization)
+- gofumpt (formatting)
+- errcheck, errorlint, errname (error handling)
+- gosimple, staticcheck, stylecheck (code quality)
+- gosec (security)
+- godot (documentation)
+"""
+
+# Common variable names to be extra careful with
+sensitive_variables = [
+ "err", "error", "ctx", "context", "v", "value", "i", "idx", "index",
+ "n", "num", "count", "size", "len", "length", "k", "key", "val",
+ "result", "res", "data", "item", "obj", "object", "msg", "message",
+ "resp", "response", "req", "request", "client", "conn", "connection"
+]
+
+# Naming conventions for nested scopes
+naming_conventions = """
+When dealing with nested scopes, use these naming patterns:
+- Outer scope: 'err', 'ctx', 'v', 'i'
+- Inner scope: 'innerErr', 'innerCtx', 'item', 'idx'
+- Deep nested: 'deepErr', 'deepCtx', 'subItem', 'subIdx'
+- Or use descriptive prefixes: 'processErr', 'validateErr', 'parseErr'
+"""
+
+# Import organization rules
+import_organization = """
+Follow this exact import order:
+1. Standard library imports
+2. Default imports (third-party)
+3. github.com/apache/skywalking-banyandb/ prefix imports
+
+Use these specific aliases:
+- commonv1 for github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1
+- databasev1 for github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1
+- modelv1 for github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1
+- propertyv1 for github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1
+- measurev1 for github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1
+- streamv1 for github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1
+- clusterv1 for github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1
+- pbv1 for github.com/apache/skywalking-banyandb/pkg/pb/v1
+"""
+
+# Error handling patterns
+error_handling = """
+Follow these error handling patterns:
+1. Always check errors immediately after function calls
+2. Use descriptive error variable names to avoid shadowing
+3. Wrap errors with context using fmt.Errorf and %w verb
+4. Follow the errname convention for error types
+5. Use proper error return patterns
+6. Avoid ignoring errors unless explicitly documented
+"""
+
+# Documentation standards
+documentation = """
+Follow these documentation standards:
+1. Add top-level comments for all exported functions and types
+2. Use proper sentence structure and punctuation
+3. Describe what the function does, not how it does it
+4. Include parameter and return value descriptions for complex functions
+5. Follow the godot scope: toplevel setting
+"""
\ No newline at end of file
diff --git a/api/data/data.go b/api/data/data.go
index 046f1c6e..065d7949 100644
--- a/api/data/data.go
+++ b/api/data/data.go
@@ -30,15 +30,18 @@ import (
var (
// TopicMap is the map of topic name to topic.
TopicMap = map[string]bus.Topic{
- TopicStreamWrite.String(): TopicStreamWrite,
- TopicStreamQuery.String(): TopicStreamQuery,
- TopicMeasureWrite.String(): TopicMeasureWrite,
- TopicMeasureQuery.String(): TopicMeasureQuery,
- TopicTopNQuery.String(): TopicTopNQuery,
- TopicPropertyDelete.String(): TopicPropertyDelete,
- TopicPropertyQuery.String(): TopicPropertyQuery,
- TopicPropertyUpdate.String(): TopicPropertyUpdate,
- TopicPropertyRepair.String(): TopicPropertyRepair,
+ TopicStreamWrite.String(): TopicStreamWrite,
+ TopicStreamQuery.String(): TopicStreamQuery,
+ TopicMeasureWrite.String(): TopicMeasureWrite,
+ TopicMeasureQuery.String(): TopicMeasureQuery,
+ TopicTopNQuery.String(): TopicTopNQuery,
+ TopicPropertyDelete.String(): TopicPropertyDelete,
+ TopicPropertyQuery.String(): TopicPropertyQuery,
+ TopicPropertyUpdate.String(): TopicPropertyUpdate,
+ TopicStreamPartSync.String(): TopicStreamPartSync,
+ TopicPropertyRepair.String(): TopicPropertyRepair,
+ TopicStreamSeriesIndexWrite.String(): TopicStreamSeriesIndexWrite,
+ TopicStreamLocalIndexWrite.String(): TopicStreamLocalIndexWrite,
}
// TopicRequestMap is the map of topic name to request message.
@@ -68,9 +71,18 @@ var (
TopicPropertyDelete: func() proto.Message {
return &propertyv1.InternalDeleteRequest{}
},
+ TopicStreamPartSync: func() proto.Message {
+ return nil
+ },
TopicPropertyRepair: func() proto.Message {
return &propertyv1.InternalRepairRequest{}
},
+ TopicStreamSeriesIndexWrite: func() proto.Message {
+ return nil
+ },
+ TopicStreamLocalIndexWrite: func() proto.Message {
+ return nil
+ },
}
// TopicResponseMap is the map of topic name to response message.
diff --git a/api/data/stream.go b/api/data/stream.go
index 08877d7c..dd43c974 100644
--- a/api/data/stream.go
+++ b/api/data/stream.go
@@ -48,3 +48,30 @@ var StreamDeleteExpiredSegmentsKindVersion = common.KindVersion{
// TopicDeleteExpiredStreamSegments is the delete stream segments topic.
var TopicDeleteExpiredStreamSegments = bus.BiTopic(StreamDeleteExpiredSegmentsKindVersion.String())
+
+// StreamPartSyncKindVersion is the version tag of part sync kind.
+var StreamPartSyncKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "part-sync",
+}
+
+// TopicStreamPartSync is the part sync topic.
+var TopicStreamPartSync = bus.BiTopic(StreamPartSyncKindVersion.String())
+
+// StreamSeriesIndexWriteKindVersion is the version tag of stream series index write kind.
+var StreamSeriesIndexWriteKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "stream-series-index-write",
+}
+
+// TopicStreamSeriesIndexWrite is the stream series index write topic.
+var TopicStreamSeriesIndexWrite = bus.BiTopic(StreamSeriesIndexWriteKindVersion.String())
+
+// StreamLocalIndexWriteKindVersion is the version tag of stream local index write kind.
+var StreamLocalIndexWriteKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "stream-local-index-write",
+}
+
+// TopicStreamLocalIndexWrite is the stream local index write topic.
+var TopicStreamLocalIndexWrite = bus.BiTopic(StreamLocalIndexWriteKindVersion.String())
diff --git a/banyand/internal/storage/rotation.go b/banyand/internal/storage/rotation.go
index 9d77b339..2a0ab9ea 100644
--- a/banyand/internal/storage/rotation.go
+++ b/banyand/internal/storage/rotation.go
@@ -105,7 +105,7 @@ func (d *database[T, O]) startRotationTask() error {
}
d.incTotalRotationStarted(1)
defer d.incTotalRotationFinished(1)
- start := options.SegmentInterval.nextTime(t)
+ start := options.SegmentInterval.NextTime(t)
d.logger.Info().Time("segment_start", start).Time("event_time", t).Msg("create new segment")
_, err = d.segmentController.create(start)
if err != nil {
diff --git a/banyand/internal/storage/segment.go b/banyand/internal/storage/segment.go
index e7b686ed..fe54b9b8 100644
--- a/banyand/internal/storage/segment.go
+++ b/banyand/internal/storage/segment.go
@@ -501,7 +501,7 @@ func (sc *segmentController[T, O]) create(start time.Time) (*segment[T, O], erro
}
}
options := sc.getOptions()
- start = options.SegmentInterval.Unit.standard(start)
+ start = options.SegmentInterval.Unit.Standard(start)
var next *segment[T, O]
for _, s := range sc.lst {
if s.Contains(start.UnixNano()) {
@@ -511,7 +511,7 @@ func (sc *segmentController[T, O]) create(start time.Time) (*segment[T, O], erro
next = s
}
}
- stdEnd := options.SegmentInterval.nextTime(start)
+ stdEnd := options.SegmentInterval.NextTime(start)
var end time.Time
if next != nil && next.Start.Before(stdEnd) {
end = next.Start
@@ -657,7 +657,7 @@ func loadSegments[T TSTable, O any](root, prefix string, parser *segmentControll
if i < len(startTimeLst)-1 {
end = startTimeLst[i+1]
} else {
- end = intervalRule.nextTime(start)
+ end = intervalRule.NextTime(start)
}
if err := loadFn(start, end); err != nil {
return err
diff --git a/banyand/internal/storage/storage.go b/banyand/internal/storage/storage.go
index a7784b76..2c6521b5 100644
--- a/banyand/internal/storage/storage.go
+++ b/banyand/internal/storage/storage.go
@@ -161,7 +161,8 @@ func (iu IntervalUnit) String() string {
panic("invalid interval unit")
}
-func (iu IntervalUnit) standard(t time.Time) time.Time {
+// Standard returns a standardized time based on the interval unit.
+func (iu IntervalUnit) Standard(t time.Time) time.Time {
switch iu {
case HOUR:
return time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), 0, 0, 0, t.Location())
@@ -177,7 +178,8 @@ type IntervalRule struct {
Num int
}
-func (ir IntervalRule) nextTime(current time.Time) time.Time {
+// NextTime returns the next time point based on the current time and interval rule.
+func (ir IntervalRule) NextTime(current time.Time) time.Time {
switch ir.Unit {
case HOUR:
return current.Add(time.Hour * time.Duration(ir.Num))
diff --git a/banyand/internal/wqueue/shard.go b/banyand/internal/wqueue/shard.go
new file mode 100644
index 00000000..44ffce4d
--- /dev/null
+++ b/banyand/internal/wqueue/shard.go
@@ -0,0 +1,42 @@
+// 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 wqueue provides a write queue implementation for managing data shards.
+package wqueue
+
+import (
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// Shard represents a data shard containing a sub-queue and associated metadata.
+type Shard[S SubQueue] struct {
+ sq S
+ l *logger.Logger
+ location string
+ id common.ShardID
+}
+
+// SubQueue returns the underlying sub-queue of the shard.
+func (s Shard[S]) SubQueue() S {
+ return s.sq
+}
+
+// Close closes the shard and its underlying sub-queue.
+func (s Shard[S]) Close() error {
+ return s.sq.Close()
+}
diff --git a/banyand/internal/wqueue/wqueue.go b/banyand/internal/wqueue/wqueue.go
new file mode 100644
index 00000000..099664ae
--- /dev/null
+++ b/banyand/internal/wqueue/wqueue.go
@@ -0,0 +1,222 @@
+// 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 wqueue
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+const (
+ shardPathPrefix = "shard"
+ shardTemplate = shardPathPrefix + "-%d"
+ lockFilename = "lock"
+)
+
+// ErrUnknownShard indicates that the shard is not found.
+var ErrUnknownShard = errors.New("unknown shard")
+
+// Metrics is the interface of metrics.
+type Metrics interface {
+ // DeleteAll deletes all metrics.
+ DeleteAll()
+}
+
+// SubQueue represents a sub-queue interface that can be closed.
+type SubQueue interface {
+ io.Closer
+}
+
+// SubQueueCreator is a function type that creates a sub-queue with the given parameters.
+type SubQueueCreator[S SubQueue, O any] func(fileSystem fs.FileSystem, root string, position common.Position,
+ l *logger.Logger, option O, metrics any, group string, shardID common.ShardID, getNodes func() []string) (S, error)
+
+// Opts contains configuration options for creating a queue.
+type Opts[S SubQueue, O any] struct {
+ SubQueueCreator SubQueueCreator[S, O]
+ GetNodes func(common.ShardID) []string
+ Metrics Metrics
+ Option O
+ Group string
+ Location string
+ SegmentInterval storage.IntervalRule
+ ShardNum uint32
+}
+
+// Queue represents a write queue that manages multiple shards.
+type Queue[S SubQueue, O any] struct {
+ lfs fs.FileSystem
+ lock fs.File
+ logger *logger.Logger
+ p common.Position
+ location string
+ sLst []*Shard[S]
+ opts Opts[S, O]
+ sync.RWMutex
+ closed atomic.Bool
+}
+
+// UpdateOptions updates the queue options with new resource options.
+func (q *Queue[S, O]) UpdateOptions(resourceOpts *commonv1.ResourceOpts) {
+ if q.closed.Load() {
+ return
+ }
+ q.Lock()
+ defer q.Unlock()
+ si := storage.MustToIntervalRule(resourceOpts.SegmentInterval)
+ if q.opts.SegmentInterval.Unit != si.Unit {
+ q.logger.Panic().Msg("segment interval unit cannot be changed")
+ return
+ }
+ q.opts.SegmentInterval = si
+ q.opts.ShardNum = resourceOpts.ShardNum
+}
+
+func (q *Queue[S, O]) getOpts() Opts[S, O] {
+ q.RLock()
+ defer q.RUnlock()
+ return q.opts
+}
+
+// Close closes the queue and all its shards.
+func (q *Queue[S, O]) Close() error {
+ if q.closed.Load() {
+ return nil
+ }
+ q.closed.Store(true)
+ q.Lock()
+ defer q.Unlock()
+ for _, shard := range q.sLst {
+ shard.Close()
+ }
+ q.lock.Close()
+ if err := q.lfs.DeleteFile(q.lock.Path()); err != nil {
+ logger.Panicf("cannot delete lock file %s: %s", q.lock.Path(), err)
+ }
+ return nil
+}
+
+// Open creates and initializes a new queue with the given options.
+func Open[S SubQueue, O any](ctx context.Context, opts Opts[S, O], _ string) (*Queue[S, O], error) {
+ p := common.GetPosition(ctx)
+ location := filepath.Clean(opts.Location)
+ lfs := fs.NewLocalFileSystemWithLogger(logger.GetLogger("wqueue"))
+ lfs.MkdirIfNotExist(location, storage.DirPerm)
+ l := logger.Fetch(ctx, p.Database)
+ q := &Queue[S, O]{
+ location: location,
+ logger: l,
+ p: p,
+ lfs: lfs,
+ opts: opts,
+ }
+ q.logger.Info().Str("path", opts.Location).Msg("initialized")
+ lockPath := filepath.Join(opts.Location, lockFilename)
+ lock, err := lfs.CreateLockFile(lockPath, storage.FilePerm)
+ if err != nil {
+ logger.Panicf("cannot create lock file %s: %s", lockPath, err)
+ }
+ q.lock = lock
+ return q, nil
+}
+
+// GetOrCreateShard gets or creates a shard with the given ShardID.
+// If the shard already exists, it returns it without locking.
+// If the shard doesn't exist, it creates a new one with proper locking.
+func (q *Queue[S, O]) GetOrCreateShard(shardID common.ShardID) (*Shard[S], error) {
+ // First check if shard exists without locking
+ if shard := q.getShard(shardID); shard != nil {
+ return shard, nil
+ }
+
+ // Shard doesn't exist, need to create it with locking
+ q.Lock()
+ defer q.Unlock()
+
+ // Double-check after acquiring lock
+ if shard := q.getShard(shardID); shard != nil {
+ return shard, nil
+ }
+
+ // Create the shard directory path
+ shardPath := filepath.Join(q.location, fmt.Sprintf(shardTemplate, int(shardID)))
+
+ // Create the shard directory if it doesn't exist
+ q.lfs.MkdirIfNotExist(shardPath, storage.DirPerm)
+
+ // Create the sub-queue using the provided creator
+ subQueue, err := q.opts.SubQueueCreator(q.lfs, shardPath, q.p, q.logger, q.opts.Option, q.opts.Metrics, q.opts.Group, shardID, func() []string {
+ return q.opts.GetNodes(shardID)
+ })
+ if err != nil {
+ return nil, errors.Wrapf(err, "failed to create sub-queue for shard %d", shardID)
+ }
+
+ // Create the shard
+ shard := &Shard[S]{
+ sq: subQueue,
+ l: q.logger,
+ location: shardPath,
+ id: shardID,
+ }
+
+ // Add the shard to the list
+ q.sLst = append(q.sLst, shard)
+
+ q.logger.Info().Uint32("shard_id", uint32(shardID)).Str("path", shardPath).Msg("created new shard")
+ return shard, nil
+}
+
+// getShard returns the shard with the given ID if it exists, nil otherwise.
+// This method is not thread-safe and should only be called when the caller holds the appropriate lock.
+func (q *Queue[S, O]) getShard(shardID common.ShardID) *Shard[S] {
+ for _, shard := range q.sLst {
+ if shard.id == shardID {
+ return shard
+ }
+ }
+ return nil
+}
+
+// GetTimeRange returns a valid time range based on an input timestamp.
+// It uses the Queue's SegmentInterval to generate the time range.
+func (q *Queue[S, O]) GetTimeRange(ts time.Time) timestamp.TimeRange {
+ opts := q.getOpts()
+ start := opts.SegmentInterval.Unit.Standard(ts)
+ end := opts.SegmentInterval.NextTime(start)
+ return timestamp.NewSectionTimeRange(start, end)
+}
+
+// GetNodes returns the nodes for the given shard ID.
+func (q *Queue[S, O]) GetNodes(shardID common.ShardID) []string {
+ return q.opts.GetNodes(shardID)
+}
diff --git a/banyand/internal/wqueue/wqueue_test.go b/banyand/internal/wqueue/wqueue_test.go
new file mode 100644
index 00000000..91c14ff4
--- /dev/null
+++ b/banyand/internal/wqueue/wqueue_test.go
@@ -0,0 +1,139 @@
+// 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 wqueue
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+type mockSubQueue struct{}
+
+func (m *mockSubQueue) Close() error {
+ return nil
+}
+
+type mockSubQueueOption struct{}
+
+func TestQueue_GetTimeRange(t *testing.T) {
+ tests := []struct {
+ inputTime time.Time
+ expectedStart time.Time
+ expectedEnd time.Time
+ name string
+ segmentInterval storage.IntervalRule
+ }{
+ {
+ name: "hour interval",
+ segmentInterval: storage.IntervalRule{
+ Unit: storage.HOUR,
+ Num: 1,
+ },
+ inputTime: time.Date(2023, 12, 15, 14, 30, 45, 123456789, time.UTC),
+ expectedStart: time.Date(2023, 12, 15, 14, 0, 0, 0, time.UTC),
+ expectedEnd: time.Date(2023, 12, 15, 15, 0, 0, 0, time.UTC),
+ },
+ {
+ name: "day interval",
+ segmentInterval: storage.IntervalRule{
+ Unit: storage.DAY,
+ Num: 1,
+ },
+ inputTime: time.Date(2023, 12, 15, 14, 30, 45, 123456789, time.UTC),
+ expectedStart: time.Date(2023, 12, 15, 0, 0, 0, 0, time.UTC),
+ expectedEnd: time.Date(2023, 12, 16, 0, 0, 0, 0, time.UTC),
+ },
+ {
+ name: "multiple hours interval",
+ segmentInterval: storage.IntervalRule{
+ Unit: storage.HOUR,
+ Num: 2,
+ },
+ inputTime: time.Date(2023, 12, 15, 14, 30, 45, 123456789, time.UTC),
+ expectedStart: time.Date(2023, 12, 15, 14, 0, 0, 0, time.UTC),
+ expectedEnd: time.Date(2023, 12, 15, 16, 0, 0, 0, time.UTC),
+ },
+ {
+ name: "multiple days interval",
+ segmentInterval: storage.IntervalRule{
+ Unit: storage.DAY,
+ Num: 3,
+ },
+ inputTime: time.Date(2023, 12, 15, 14, 30, 45, 123456789, time.UTC),
+ expectedStart: time.Date(2023, 12, 15, 0, 0, 0, 0, time.UTC),
+ expectedEnd: time.Date(2023, 12, 18, 0, 0, 0, 0, time.UTC),
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Create a temporary directory for the test
+ tempDir := t.TempDir()
+
+ // Create queue options
+ opts := Opts[*mockSubQueue, mockSubQueueOption]{
+ Option: mockSubQueueOption{},
+ Metrics: nil,
+ SubQueueCreator: createMockSubQueue,
+ Location: tempDir,
+ ShardNum: 1,
+ SegmentInterval: tt.segmentInterval,
+ GetNodes: func(_ common.ShardID) []string { return []string{"node1"} },
+ }
+
+ // Create context with position
+ ctx := context.Background()
+ ctx = common.SetPosition(ctx, func(_ common.Position) common.Position {
+ return common.Position{
+ Database: "test-db",
+ Stage: "test-stage",
+ }
+ })
+ ctx = context.WithValue(ctx, logger.ContextKey, logger.GetLogger("test"))
+
+ // Create queue
+ queue, err := Open[*mockSubQueue, mockSubQueueOption](ctx, opts, "test-group")
+ require.NoError(t, err)
+ defer queue.Close()
+
+ // Test GetTimeRange
+ timeRange := queue.GetTimeRange(tt.inputTime)
+
+ // Verify results
+ assert.Equal(t, tt.expectedStart, timeRange.Start)
+ assert.Equal(t, tt.expectedEnd, timeRange.End)
+ assert.True(t, timeRange.IncludeStart)
+ assert.False(t, timeRange.IncludeEnd)
+ })
+ }
+}
+
+func createMockSubQueue(_ fs.FileSystem, _ string, _ common.Position, _ *logger.Logger,
+ _ mockSubQueueOption, _ any, _ string, _ common.ShardID, _ func() []string,
+) (*mockSubQueue, error) {
+ return &mockSubQueue{}, nil
+}
diff --git a/banyand/liaison/grpc/registry_test.go b/banyand/liaison/grpc/registry_test.go
index 3b3f58a6..76bf9048 100644
--- a/banyand/liaison/grpc/registry_test.go
+++ b/banyand/liaison/grpc/registry_test.go
@@ -184,7 +184,6 @@ func setupForRegistry() func() {
nr := grpc.NewLocalNodeRegistry()
tcp := grpc.NewServer(context.TODO(), pipeline, pipeline, pipeline, nil, metaSvc, grpc.NodeRegistries{
MeasureLiaisonNodeRegistry: nr,
- StreamDataNodeRegistry: nr,
PropertyNodeRegistry: nr,
}, metricSvc, nil, pipeline)
preloadStreamSvc := &preloadStreamService{metaSvc: metaSvc}
diff --git a/banyand/liaison/grpc/server.go b/banyand/liaison/grpc/server.go
index c523c834..0ee0965a 100644
--- a/banyand/liaison/grpc/server.go
+++ b/banyand/liaison/grpc/server.go
@@ -73,7 +73,6 @@ type Server interface {
// NodeRegistries contains the node registries.
type NodeRegistries struct {
StreamLiaisonNodeRegistry NodeRegistry
- StreamDataNodeRegistry NodeRegistry
MeasureLiaisonNodeRegistry NodeRegistry
MeasureDataNodeRegistry NodeRegistry
PropertyNodeRegistry NodeRegistry
@@ -92,8 +91,7 @@ type server struct {
stopCh chan struct{}
*indexRuleRegistryServer
*measureRegistryServer
- streamSVC *streamService
- streamCallback *streamRedirectWriteCallback
+ streamSVC *streamService
*streamRegistryServer
measureSVC *measureService
log *logger.Logger
@@ -140,11 +138,6 @@ func NewServer(_ context.Context, tir1Client, tir2Client, broadcaster queue.Clie
measureSVC: measureSVC,
groupRepo: gr,
tire2Server: tire2Server,
- streamCallback: &streamRedirectWriteCallback{
- pipeline: tir2Client,
- groupRepo: gr,
- nodeRegistry: nr.StreamDataNodeRegistry,
- },
measureCallback: &measureRedirectWriteCallback{
pipeline: tir2Client,
groupRepo: gr,
@@ -190,7 +183,6 @@ func NewServer(_ context.Context, tir1Client, tir2Client, broadcaster queue.Clie
func (s *server) PreRun(_ context.Context) error {
s.log = logger.GetLogger("liaison-grpc")
s.streamSVC.setLogger(s.log.Named("stream-t1"))
- s.streamCallback.l = s.log.Named("stream-t2")
s.measureSVC.setLogger(s.log)
s.propertyServer.SetLogger(s.log)
s.measureCallback.l = s.log.Named("measure-t2")
@@ -206,9 +198,6 @@ func (s *server) PreRun(_ context.Context) error {
return err
}
}
- if err := s.tire2Server.Subscribe(data.TopicStreamWrite, s.streamCallback); err != nil {
- return err
- }
if err := s.tire2Server.Subscribe(data.TopicMeasureWrite, s.measureCallback); err != nil {
return err
}
@@ -281,7 +270,6 @@ func (s *server) FlagSet() *run.FlagSet {
fs.BoolVar(&s.enableIngestionAccessLog, "enable-ingestion-access-log", false, "enable ingestion access log")
fs.StringVar(&s.accessLogRootPath, "access-log-root-path", "", "access log root path")
fs.DurationVar(&s.streamSVC.writeTimeout, "stream-write-timeout", 15*time.Second, "timeout for writing stream among liaison nodes")
- fs.DurationVar(&s.streamCallback.writeTimeout, "stream-write-data-timeout", 15*time.Second, "timeout for writing stream data to the data nodes")
fs.DurationVar(&s.measureCallback.writeTimeout, "measure-write-data-timeout", 15*time.Second, "timeout for writing measure data to the data nodes")
fs.DurationVar(&s.measureSVC.writeTimeout, "measure-write-timeout", 15*time.Second, "timeout for writing measure among liaison nodes")
fs.DurationVar(&s.measureSVC.maxWaitDuration, "measure-metadata-cache-wait-duration", 0,
diff --git a/banyand/liaison/grpc/stream.go b/banyand/liaison/grpc/stream.go
index cf9baada..8dc9ff5f 100644
--- a/banyand/liaison/grpc/stream.go
+++ b/banyand/liaison/grpc/stream.go
@@ -25,7 +25,6 @@ import (
"github.com/pkg/errors"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
- "google.golang.org/protobuf/proto"
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/api/data"
@@ -316,80 +315,3 @@ func (s *streamService) Close() error {
}
return nil
}
-
-type streamRedirectWriteCallback struct {
- *bus.UnImplementedHealthyListener
- l *logger.Logger
- pipeline queue.Client
- groupRepo *groupRepo
- nodeRegistry NodeRegistry
- writeTimeout time.Duration
-}
-
-func (r *streamRedirectWriteCallback) Rev(ctx context.Context, message bus.Message) (resp bus.Message) {
- events, ok := message.Data().([]any)
- if !ok {
- r.l.Warn().Msg("invalid event data type")
- return
- }
- if len(events) < 1 {
- r.l.Warn().Msg("empty event")
- return
- }
-
- publisher := r.pipeline.NewBatchPublisher(r.writeTimeout)
- defer func() {
- _, err := publisher.Close()
- if err != nil {
- r.l.Error().Err(err).Msg("failed to close publisher")
- }
- }()
-
- for i := range events {
- var writeEvent *streamv1.InternalWriteRequest
- switch e := events[i].(type) {
- case *streamv1.InternalWriteRequest:
- writeEvent = e
- case []byte:
- writeEvent = &streamv1.InternalWriteRequest{}
- if err := proto.Unmarshal(e, writeEvent); err != nil {
- r.l.Error().Err(err).RawJSON("written", e).Msg("fail to unmarshal event")
- continue
- }
- default:
- r.l.Warn().Msg("invalid event data type")
- continue
- }
-
- metadata := writeEvent.Request.GetMetadata()
- if metadata == nil {
- r.l.Warn().Msg("metadata is nil in InternalWriteRequest")
- continue
- }
-
- group := metadata.GetGroup()
- streamName := metadata.GetName()
- shardID := writeEvent.GetShardId()
-
- copies, ok := r.groupRepo.copies(group)
- if !ok {
- r.l.Error().Str("group", group).Msg("failed to get group copies")
- continue
- }
-
- for copyIdx := range copies {
- nodeID, err := r.nodeRegistry.Locate(group, streamName, shardID, copyIdx)
- if err != nil {
- r.l.Error().Err(err).Str("group", group).Str("stream", streamName).Uint32("shard", shardID).Uint32("copy", copyIdx).Msg("failed to locate node")
- continue
- }
-
- msg := bus.NewBatchMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeID, writeEvent)
- if _, err := publisher.Publish(ctx, data.TopicStreamWrite, msg); err != nil {
- r.l.Error().Err(err).Str("node", nodeID).Msg("failed to publish message")
- }
- }
- }
-
- return
-}
diff --git a/banyand/measure/introducer.go b/banyand/measure/introducer.go
index 6364809c..ab3a9771 100644
--- a/banyand/measure/introducer.go
+++ b/banyand/measure/introducer.go
@@ -118,20 +118,20 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
case <-tst.loopCloser.CloseNotify():
return
case next := <-tst.introductions:
- tst.incTotalIntroduceLoopStarted(1, "mem")
+ tst.incTotalIntroduceLoopStarted("mem")
tst.introduceMemPart(next, epoch)
- tst.incTotalIntroduceLoopFinished(1, "mem")
+ tst.incTotalIntroduceLoopFinished("mem")
epoch++
case next := <-flushCh:
- tst.incTotalIntroduceLoopStarted(1, "flush")
+ tst.incTotalIntroduceLoopStarted("flush")
tst.introduceFlushed(next, epoch)
- tst.incTotalIntroduceLoopFinished(1, "flush")
+ tst.incTotalIntroduceLoopFinished("flush")
tst.gc.clean()
epoch++
case next := <-mergeCh:
- tst.incTotalIntroduceLoopStarted(1, "merge")
+ tst.incTotalIntroduceLoopStarted("merge")
tst.introduceMerged(next, epoch)
- tst.incTotalIntroduceLoopFinished(1, "merge")
+ tst.incTotalIntroduceLoopFinished("merge")
tst.gc.clean()
epoch++
case epochWatcher := <-watcherCh:
diff --git a/banyand/measure/metrics.go b/banyand/measure/metrics.go
index 6ab3cb47..1b85934c 100644
--- a/banyand/measure/metrics.go
+++ b/banyand/measure/metrics.go
@@ -84,18 +84,18 @@ func (tst *tsTable) incTotalBatchIntroLatency(delta float64) {
tst.metrics.totalBatchIntroLatency.Inc(delta)
}
-func (tst *tsTable) incTotalIntroduceLoopStarted(delta int, phase string) {
+func (tst *tsTable) incTotalIntroduceLoopStarted(phase string) {
if tst == nil || tst.metrics == nil {
return
}
- tst.metrics.totalIntroduceLoopStarted.Inc(float64(delta), phase)
+ tst.metrics.totalIntroduceLoopStarted.Inc(1, phase)
}
-func (tst *tsTable) incTotalIntroduceLoopFinished(delta int, phase string) {
+func (tst *tsTable) incTotalIntroduceLoopFinished(phase string) {
if tst == nil || tst.metrics == nil {
return
}
- tst.metrics.totalIntroduceLoopFinished.Inc(float64(delta), phase)
+ tst.metrics.totalIntroduceLoopFinished.Inc(1, phase)
}
func (tst *tsTable) incTotalFlushLoopStarted(delta int) {
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 18efbef8..40f79043 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -309,15 +309,20 @@ func messageToRequest(topic bus.Topic, m bus.Message) (*clusterv1.SendRequest, e
MessageId: uint64(m.ID()),
BatchMod: m.BatchModeEnabled(),
}
- message, ok := m.Data().(proto.Message)
- if !ok {
+
+ switch data := m.Data().(type) {
+ case proto.Message:
+ messageData, err := proto.Marshal(data)
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal message %T: %w", m, err)
+ }
+ r.Body = messageData
+ case []byte:
+ r.Body = data
+ default:
return nil, fmt.Errorf("invalid message type %T", m.Data())
}
- data, err := proto.Marshal(message)
- if err != nil {
- return nil, fmt.Errorf("failed to marshal message %T: %w", m, err)
- }
- r.Body = data
+
return r, nil
}
diff --git a/banyand/queue/sub/sub.go b/banyand/queue/sub/sub.go
index cd646883..1e7023ce 100644
--- a/banyand/queue/sub/sub.go
+++ b/banyand/queue/sub/sub.go
@@ -75,11 +75,15 @@ func (s *server) Send(stream clusterv1.Service_SendServer) error {
if reqSupplier, ok := data.TopicRequestMap[*topic]; ok {
req := reqSupplier()
- if errUnmarshal := proto.Unmarshal(writeEntity.Body, req); errUnmarshal != nil {
- s.reply(stream, writeEntity, errUnmarshal, "failed to unmarshal message")
- continue
+ if req == nil {
+ m = bus.NewMessage(bus.MessageID(writeEntity.MessageId), writeEntity.Body)
+ } else {
+ if errUnmarshal := proto.Unmarshal(writeEntity.Body, req); errUnmarshal != nil {
+ s.reply(stream, writeEntity, errUnmarshal, "failed to unmarshal message")
+ continue
+ }
+ m = bus.NewMessage(bus.MessageID(writeEntity.MessageId), req)
}
- m = bus.NewMessage(bus.MessageID(writeEntity.MessageId), req)
} else {
s.reply(stream, writeEntity, err, "unknown topic")
continue
diff --git a/banyand/stream/block_scanner.go b/banyand/stream/block_scanner.go
index 601b3af0..736f8293 100644
--- a/banyand/stream/block_scanner.go
+++ b/banyand/stream/block_scanner.go
@@ -130,6 +130,9 @@ func getBlockScanner(ctx context.Context, segment storage.Segment[*tsTable, *opt
}
minTimestamp, maxTimestamp := updateTimeRange(filterTS, qo.minTimestamp, qo.maxTimestamp)
snp := tabs[i].currentSnapshot()
+ if snp == nil {
+ continue
+ }
parts, size = snp.getParts(parts, minTimestamp, maxTimestamp)
if size < 1 {
snp.decRef()
diff --git a/banyand/stream/elements.go b/banyand/stream/elements.go
index c040e23f..70a023da 100644
--- a/banyand/stream/elements.go
+++ b/banyand/stream/elements.go
@@ -24,6 +24,7 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/internal/wqueue"
"github.com/apache/skywalking-banyandb/pkg/index"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/pool"
@@ -200,19 +201,29 @@ func releaseElements(e *elements) {
var elementsPool = pool.Register[*elements]("stream-elements")
type elementsInTable struct {
- timeRange timestamp.TimeRange
- tsTable *tsTable
+ seriesDocs seriesDoc
+ segment storage.Segment[*tsTable, option]
+ tsTable *tsTable
+ elements *elements
+ timeRange timestamp.TimeRange
+ docs index.Documents
+ shardID common.ShardID
+}
- elements *elements
+type elementsInGroup struct {
+ tsdb storage.TSDB[*tsTable, option]
+ tables []*elementsInTable
+ segments []storage.Segment[*tsTable, option]
+ latestTS int64
+}
- docs index.Documents
+type elementsInQueue struct {
+ name string
+ queue *wqueue.Queue[*tsTable, option]
+ tables []*elementsInTable
}
-type elementsInGroup struct {
- tsdb storage.TSDB[*tsTable, option]
+type seriesDoc struct {
docIDsAdded map[uint64]struct{}
docs index.Documents
- tables []*elementsInTable
- segments []storage.Segment[*tsTable, option]
- latestTS int64
}
diff --git a/banyand/stream/flusher.go b/banyand/stream/flusher.go
index ae7010b3..63796385 100644
--- a/banyand/stream/flusher.go
+++ b/banyand/stream/flusher.go
@@ -89,6 +89,49 @@ func (tst *tsTable) flusherLoop(flushCh chan *flusherIntroduction, mergeCh chan
}
}
+func (tst *tsTable) flusherLoopNoMerger(flushCh chan *flusherIntroduction, introducerWatcher, flusherWatcher watcher.Channel, epoch uint64) {
+ defer tst.loopCloser.Done()
+ epochWatcher := introducerWatcher.Add(epoch, tst.loopCloser.CloseNotify())
+ if epochWatcher == nil {
+ return
+ }
+ var flusherWatchers watcher.Epochs
+
+ for {
+ select {
+ case <-tst.loopCloser.CloseNotify():
+ return
+ case e := <-flusherWatcher:
+ flusherWatchers.Add(e)
+ case <-epochWatcher.Watch():
+ if func() bool {
+ tst.incTotalFlushLoopStarted(1)
+ start := time.Now()
+ defer func() {
+ tst.incTotalFlushLoopFinished(1)
+ tst.incTotalFlushLatency(time.Since(start).Seconds())
+ }()
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot != nil {
+ defer curSnapshot.decRef()
+ tst.flush(curSnapshot, flushCh)
+ epoch = curSnapshot.epoch
+ flusherWatchers.Notify(epoch)
+ flusherWatchers = nil
+ if tst.currentEpoch() != epoch {
+ tst.incTotalFlushLoopProgress(1)
+ return false
+ }
+ }
+ epochWatcher = introducerWatcher.Add(epoch, tst.loopCloser.CloseNotify())
+ return epochWatcher == nil
+ }() {
+ return
+ }
+ }
+ }
+}
+
// pauseFlusherToPileupMemParts takes a pause to wait for in-memory parts to pile up.
// If there is no in-memory part, we can skip the pause.
// When a merging is finished, we can skip the pause.
diff --git a/banyand/stream/introducer.go b/banyand/stream/introducer.go
index d0dc0911..cdc25f21 100644
--- a/banyand/stream/introducer.go
+++ b/banyand/stream/introducer.go
@@ -110,6 +110,36 @@ func releaseMergerIntroduction(i *mergerIntroduction) {
mergerIntroductionPool.Put(i)
}
+type syncIntroduction struct {
+ synced map[uint64]struct{}
+ applied chan struct{}
+}
+
+func (i *syncIntroduction) reset() {
+ for k := range i.synced {
+ delete(i.synced, k)
+ }
+ i.applied = nil
+}
+
+var syncIntroductionPool = pool.Register[*syncIntroduction]("stream-sync-introduction")
+
+func generateSyncIntroduction() *syncIntroduction {
+ v := syncIntroductionPool.Get()
+ if v == nil {
+ return &syncIntroduction{
+ synced: make(map[uint64]struct{}),
+ }
+ }
+ si := v
+ si.reset()
+ return si
+}
+
+func releaseSyncIntroduction(i *syncIntroduction) {
+ syncIntroductionPool.Put(i)
+}
+
func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh chan *mergerIntroduction, watcherCh watcher.Channel, epoch uint64) {
var introducerWatchers watcher.Epochs
defer tst.loopCloser.Done()
@@ -118,20 +148,52 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
case <-tst.loopCloser.CloseNotify():
return
case next := <-tst.introductions:
- tst.incTotalIntroduceLoopStarted(1, "mem")
+ tst.incTotalIntroduceLoopStarted("mem")
tst.introduceMemPart(next, epoch)
- tst.incTotalIntroduceLoopFinished(1, "mem")
+ tst.incTotalIntroduceLoopFinished("mem")
epoch++
case next := <-flushCh:
- tst.incTotalIntroduceLoopStarted(1, "flush")
+ tst.incTotalIntroduceLoopStarted("flush")
tst.introduceFlushed(next, epoch)
- tst.incTotalIntroduceLoopFinished(1, "flush")
+ tst.incTotalIntroduceLoopFinished("flush")
tst.gc.clean()
epoch++
case next := <-mergeCh:
- tst.incTotalIntroduceLoopStarted(1, "merge")
+ tst.incTotalIntroduceLoopStarted("merge")
tst.introduceMerged(next, epoch)
- tst.incTotalIntroduceLoopFinished(1, "merge")
+ tst.incTotalIntroduceLoopFinished("merge")
+ tst.gc.clean()
+ epoch++
+ case epochWatcher := <-watcherCh:
+ introducerWatchers.Add(epochWatcher)
+ }
+ curEpoch := tst.currentEpoch()
+ introducerWatchers.Notify(curEpoch)
+ }
+}
+
+func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, syncCh chan *syncIntroduction, watcherCh watcher.Channel, epoch uint64) {
+ var introducerWatchers watcher.Epochs
+ defer tst.loopCloser.Done()
+ for {
+ select {
+ case <-tst.loopCloser.CloseNotify():
+ return
+ case next := <-tst.introductions:
+ tst.incTotalIntroduceLoopStarted("mem")
+ tst.introduceMemPart(next, epoch)
+ tst.incTotalIntroduceLoopFinished("mem")
+ epoch++
+ case next := <-flushCh:
+ tst.incTotalIntroduceLoopStarted("flush")
+ tst.introduceFlushed(next, epoch)
+ tst.incTotalIntroduceLoopFinished("flush")
+ tst.gc.clean()
+ epoch++
+ case next := <-syncCh:
+ tst.incTotalIntroduceLoopStarted("sync")
+ tst.introduceSync(next, epoch)
+ tst.incTotalIntroduceLoopFinished("sync")
tst.gc.clean()
epoch++
case epochWatcher := <-watcherCh:
@@ -192,6 +254,22 @@ func (tst *tsTable) introduceMerged(nextIntroduction *mergerIntroduction, epoch
}
}
+func (tst *tsTable) introduceSync(nextIntroduction *syncIntroduction, epoch uint64) {
+ cur := tst.currentSnapshot()
+ if cur == nil {
+ tst.l.Panic().Msg("current snapshot is nil")
+ return
+ }
+ defer cur.decRef()
+ nextSnp := cur.remove(epoch, nextIntroduction.synced)
+ nextSnp.creator = snapshotCreatorSyncer
+ tst.replaceSnapshot(&nextSnp)
+ tst.persistSnapshot(&nextSnp)
+ if nextIntroduction.applied != nil {
+ close(nextIntroduction.applied)
+ }
+}
+
func (tst *tsTable) replaceSnapshot(next *snapshot) {
tst.Lock()
defer tst.Unlock()
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index 6d789408..eb7d3b0e 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -31,11 +31,12 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/api/validate"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/internal/wqueue"
+ "github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/protector"
- "github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/queue/pub"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/meter"
@@ -57,7 +58,7 @@ type schemaRepo struct {
path string
}
-func newSchemaRepo(path string, svc *service, nodeLabels map[string]string) schemaRepo {
+func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string) schemaRepo {
sr := schemaRepo{
l: svc.l,
path: path,
@@ -73,6 +74,22 @@ func newSchemaRepo(path string, svc *service, nodeLabels map[string]string) sche
return sr
}
+func newLiaisonSchemaRepo(path string, svc *liaison, streamDataNodeRegistry grpc.NodeRegistry) schemaRepo {
+ sr := schemaRepo{
+ l: svc.l,
+ path: path,
+ metadata: svc.metadata,
+ Repository: resourceSchema.NewRepository(
+ svc.metadata,
+ svc.l,
+ newQueueSupplier(path, svc, streamDataNodeRegistry),
+ resourceSchema.NewMetrics(svc.omr.With(metadataScope)),
+ ),
+ }
+ sr.start()
+ return sr
+}
+
// NewPortableRepository creates a new portable repository.
func NewPortableRepository(metadata metadata.Repo, l *logger.Logger, metrics *resourceSchema.Metrics) SchemaService {
r := &schemaRepo{
@@ -242,21 +259,32 @@ func (sr *schemaRepo) loadTSDB(groupName string) (storage.TSDB[*tsTable, option]
return db.(storage.TSDB[*tsTable, option]), nil
}
+func (sr *schemaRepo) loadQueue(groupName string) (*wqueue.Queue[*tsTable, option], error) {
+ g, ok := sr.LoadGroup(groupName)
+ if !ok {
+ return nil, fmt.Errorf("group %s not found", groupName)
+ }
+ db := g.SupplyTSDB()
+ if db == nil {
+ return nil, fmt.Errorf("queue for group %s not found", groupName)
+ }
+ return db.(*wqueue.Queue[*tsTable, option]), nil
+}
+
var _ resourceSchema.ResourceSupplier = (*supplier)(nil)
type supplier struct {
metadata metadata.Repo
- pipeline queue.Queue
omr observability.MetricsRegistry
- l *logger.Logger
pm protector.Memory
+ l *logger.Logger
schemaRepo *schemaRepo
nodeLabels map[string]string
path string
option option
}
-func newSupplier(path string, svc *service, nodeLabels map[string]string) *supplier {
+func newSupplier(path string, svc *standalone, nodeLabels map[string]string) *supplier {
if svc.pm == nil {
svc.l.Panic().Msg("CRITICAL: svc.pm is nil in newSupplier")
}
@@ -270,7 +298,6 @@ func newSupplier(path string, svc *service, nodeLabels map[string]string) *suppl
return &supplier{
metadata: svc.metadata,
l: svc.l,
- pipeline: svc.localPipeline,
option: opt,
omr: svc.omr,
pm: svc.pm,
@@ -381,3 +408,93 @@ func (s *portableSupplier) OpenResource(spec resourceSchema.Resource) (resourceS
schema: streamSchema,
}, s.l, nil, nil), nil
}
+
+var _ resourceSchema.ResourceSupplier = (*queueSupplier)(nil)
+
+type queueSupplier struct {
+ metadata metadata.Repo
+ omr observability.MetricsRegistry
+ pm protector.Memory
+ streamDataNodeRegistry grpc.NodeRegistry
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ path string
+ option option
+}
+
+func newQueueSupplier(path string, svc *liaison, streamDataNodeRegistry grpc.NodeRegistry) *queueSupplier {
+ if svc.pm == nil {
+ svc.l.Panic().Msg("CRITICAL: svc.pm is nil in newSupplier")
+ }
+ opt := svc.option
+ opt.protector = svc.pm
+
+ if opt.protector == nil {
+ svc.l.Panic().Msg("CRITICAL: opt.protector is still nil after assignment")
+ }
+
+ return &queueSupplier{
+ metadata: svc.metadata,
+ l: svc.l,
+ option: opt,
+ omr: svc.omr,
+ pm: svc.pm,
+ path: path,
+ schemaRepo: &svc.schemaRepo,
+ streamDataNodeRegistry: streamDataNodeRegistry,
+ }
+}
+
+func (s *queueSupplier) OpenResource(spec resourceSchema.Resource) (resourceSchema.IndexListener, error) {
+ streamSchema := spec.Schema().(*databasev1.Stream)
+ return openStream(streamSpec{
+ schema: streamSchema,
+ }, s.l, s.pm, s.schemaRepo), nil
+}
+
+func (s *queueSupplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ return s.metadata.StreamRegistry().GetStream(ctx, md)
+}
+
+func (s *queueSupplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error) {
+ name := groupSchema.Metadata.Name
+ p := common.Position{
+ Module: "stream",
+ Database: name,
+ }
+ ro := groupSchema.ResourceOpts
+ if ro == nil {
+ return nil, fmt.Errorf("no resource opts in group %s", name)
+ }
+ shardNum := ro.ShardNum
+ group := groupSchema.Metadata.Name
+ opts := wqueue.Opts[*tsTable, option]{
+ Group: group,
+ ShardNum: shardNum,
+ Location: path.Join(s.path, group),
+ Option: s.option,
+ Metrics: s.newMetrics(p),
+ SubQueueCreator: newWriteQueue,
+ GetNodes: func(shardID common.ShardID) []string {
+ copies := ro.Replicas + 1
+ nodes := make([]string, 0, copies)
+ for i := uint32(0); i < copies; i++ {
+ nodeID, err := s.streamDataNodeRegistry.Locate(group, "", uint32(shardID), i)
+ if err != nil {
+ s.l.Error().Err(err).Str("group", group).Uint32("shard", uint32(shardID)).Uint32("copy", i).Msg("failed to locate node")
+ return nil
+ }
+ nodes = append(nodes, nodeID)
+ }
+ return nodes
+ },
+ }
+ return wqueue.Open(
+ common.SetPosition(context.Background(), func(_ common.Position) common.Position {
+ return p
+ }),
+ opts, group,
+ )
+}
diff --git a/banyand/stream/metrics.go b/banyand/stream/metrics.go
index 339f3a2c..5a63c8e8 100644
--- a/banyand/stream/metrics.go
+++ b/banyand/stream/metrics.go
@@ -49,6 +49,10 @@ type metrics struct {
totalMergeLoopFinished meter.Counter
totalMergeLoopErr meter.Counter
+ totalSyncLoopStarted meter.Counter
+ totalSyncLoopFinished meter.Counter
+ totalSyncLoopErr meter.Counter
+
totalFlushLoopProgress meter.Counter
totalFlushed meter.Counter
totalFlushedMemParts meter.Counter
@@ -83,18 +87,18 @@ func (tst *tsTable) incTotalBatchIntroLatency(delta float64) {
tst.metrics.totalBatchIntroLatency.Inc(delta)
}
-func (tst *tsTable) incTotalIntroduceLoopStarted(delta int, phase string) {
+func (tst *tsTable) incTotalIntroduceLoopStarted(phase string) {
if tst == nil || tst.metrics == nil {
return
}
- tst.metrics.totalIntroduceLoopStarted.Inc(float64(delta), phase)
+ tst.metrics.totalIntroduceLoopStarted.Inc(1, phase)
}
-func (tst *tsTable) incTotalIntroduceLoopFinished(delta int, phase string) {
+func (tst *tsTable) incTotalIntroduceLoopFinished(phase string) {
if tst == nil || tst.metrics == nil {
return
}
- tst.metrics.totalIntroduceLoopFinished.Inc(float64(delta), phase)
+ tst.metrics.totalIntroduceLoopFinished.Inc(1, phase)
}
func (tst *tsTable) incTotalFlushLoopStarted(delta int) {
@@ -139,6 +143,27 @@ func (tst *tsTable) incTotalMergeLoopErr(delta int) {
tst.metrics.totalMergeLoopErr.Inc(float64(delta))
}
+func (tst *tsTable) incTotalSyncLoopStarted(delta int) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalSyncLoopStarted.Inc(float64(delta))
+}
+
+func (tst *tsTable) incTotalSyncLoopFinished(delta int) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalSyncLoopFinished.Inc(float64(delta))
+}
+
+func (tst *tsTable) incTotalSyncLoopErr(delta int) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalSyncLoopErr.Inc(float64(delta))
+}
+
func (tst *tsTable) incTotalFlushLoopProgress(delta int) {
if tst == nil || tst.metrics == nil {
return
@@ -232,6 +257,10 @@ func (m *metrics) DeleteAll() {
m.totalMergeLoopFinished.Delete()
m.totalMergeLoopErr.Delete()
+ m.totalSyncLoopStarted.Delete()
+ m.totalSyncLoopFinished.Delete()
+ m.totalSyncLoopErr.Delete()
+
m.totalFlushLoopProgress.Delete()
m.totalFlushed.Delete()
m.totalFlushedMemParts.Delete()
@@ -261,6 +290,52 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalMergeLoopStarted: factory.NewCounter("total_merge_loop_started"),
totalMergeLoopFinished: factory.NewCounter("total_merge_loop_finished"),
totalMergeLoopErr: factory.NewCounter("total_merge_loop_err"),
+ totalSyncLoopStarted: factory.NewCounter("total_sync_loop_started"),
+ totalSyncLoopFinished: factory.NewCounter("total_sync_loop_finished"),
+ totalSyncLoopErr: factory.NewCounter("total_sync_loop_err"),
+ totalFlushLoopProgress: factory.NewCounter("total_flush_loop_progress"),
+ totalFlushed: factory.NewCounter("total_flushed"),
+ totalFlushedMemParts: factory.NewCounter("total_flushed_mem_parts"),
+ totalFlushPauseCompleted: factory.NewCounter("total_flush_pause_completed"),
+ totalFlushPauseBreak: factory.NewCounter("total_flush_pause_break"),
+ totalFlushIntroLatency: factory.NewCounter("total_flush_intro_latency"),
+ totalFlushLatency: factory.NewCounter("total_flush_latency"),
+ totalMergedParts: factory.NewCounter("total_merged_parts", "type"),
+ totalMergeLatency: factory.NewCounter("total_merge_latency", "type"),
+ totalMerged: factory.NewCounter("total_merged", "type"),
+ tbMetrics: tbMetrics{
+ totalMemParts: factory.NewGauge("total_mem_part", common.ShardLabelNames()...),
+ totalMemElements: factory.NewGauge("total_mem_elements", common.ShardLabelNames()...),
+ totalMemBlocks: factory.NewGauge("total_mem_blocks", common.ShardLabelNames()...),
+ totalMemPartBytes: factory.NewGauge("total_mem_part_bytes", common.ShardLabelNames()...),
+ totalMemPartUncompressedBytes: factory.NewGauge("total_mem_part_uncompressed_bytes", common.ShardLabelNames()...),
+ totalFileParts: factory.NewGauge("total_file_parts", common.ShardLabelNames()...),
+ totalFileElements: factory.NewGauge("total_file_elements", common.ShardLabelNames()...),
+ totalFileBlocks: factory.NewGauge("total_file_blocks", common.ShardLabelNames()...),
+ totalFilePartBytes: factory.NewGauge("total_file_part_bytes", common.ShardLabelNames()...),
+ totalFilePartUncompressedBytes: factory.NewGauge("total_file_part_uncompressed_bytes", common.ShardLabelNames()...),
+ },
+ indexMetrics: inverted.NewMetrics(factory, common.SegLabelNames()...),
+ }
+}
+
+func (s *queueSupplier) newMetrics(p common.Position) storage.Metrics {
+ factory := s.omr.With(tbScope.ConstLabels(meter.ToLabelPairs(common.DBLabelNames(), p.DBLabelValues())))
+ return &metrics{
+ totalWritten: factory.NewCounter("total_written"),
+ totalBatch: factory.NewCounter("total_batch"),
+ totalBatchIntroLatency: factory.NewCounter("total_batch_intro_time"),
+ totalIntroduceLoopStarted: factory.NewCounter("total_introduce_loop_started", "phase"),
+ totalIntroduceLoopFinished: factory.NewCounter("total_introduce_loop_finished", "phase"),
+ totalFlushLoopStarted: factory.NewCounter("total_flush_loop_started"),
+ totalFlushLoopFinished: factory.NewCounter("total_flush_loop_finished"),
+ totalFlushLoopErr: factory.NewCounter("total_flush_loop_err"),
+ totalMergeLoopStarted: factory.NewCounter("total_merge_loop_started"),
+ totalMergeLoopFinished: factory.NewCounter("total_merge_loop_finished"),
+ totalMergeLoopErr: factory.NewCounter("total_merge_loop_err"),
+ totalSyncLoopStarted: factory.NewCounter("total_sync_loop_started"),
+ totalSyncLoopFinished: factory.NewCounter("total_sync_loop_finished"),
+ totalSyncLoopErr: factory.NewCounter("total_sync_loop_err"),
totalFlushLoopProgress: factory.NewCounter("total_flush_loop_progress"),
totalFlushed: factory.NewCounter("total_flushed"),
totalFlushedMemParts: factory.NewCounter("total_flushed_mem_parts"),
diff --git a/banyand/stream/part.go b/banyand/stream/part.go
index eaacae09..54d598b9 100644
--- a/banyand/stream/part.go
+++ b/banyand/stream/part.go
@@ -18,7 +18,9 @@
package stream
import (
+ "encoding/json"
"fmt"
+ "io"
"path"
"path/filepath"
"sort"
@@ -27,6 +29,8 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
@@ -148,6 +152,295 @@ func (mp *memPart) reset() {
}
}
+// Marshal serializes the memPart to []byte for network transmission.
+// The format is:
+// - partMetadata (JSON)
+// - meta buffer length + data
+// - primary buffer length + data
+// - timestamps buffer length + data
+// - tagFamilies count + (name + buffer length + data) for each
+// - tagFamilyMetadata count + (name + buffer length + data) for each
+// - tagFamilyFilter count + (name + buffer length + data) for each.
+func (mp *memPart) Marshal() ([]byte, error) {
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal partMetadata as JSON
+ metadataBytes, err := json.Marshal(mp.partMetadata)
+ if err != nil {
+ return nil, fmt.Errorf("cannot marshal partMetadata: %w", err)
+ }
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(metadataBytes)))
+ bb.Buf = append(bb.Buf, metadataBytes...)
+
+ // Marshal meta buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.meta.Buf)))
+ bb.Buf = append(bb.Buf, mp.meta.Buf...)
+
+ // Marshal primary buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.primary.Buf)))
+ bb.Buf = append(bb.Buf, mp.primary.Buf...)
+
+ // Marshal timestamps buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.timestamps.Buf)))
+ bb.Buf = append(bb.Buf, mp.timestamps.Buf...)
+
+ // Marshal tagFamilies
+ if mp.tagFamilies == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFamilies)))
+ for name, tf := range mp.tagFamilies {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tf.Buf)))
+ bb.Buf = append(bb.Buf, tf.Buf...)
+ }
+ }
+
+ // Marshal tagFamilyMetadata
+ if mp.tagFamilyMetadata == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFamilyMetadata)))
+ for name, tfh := range mp.tagFamilyMetadata {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tfh.Buf)))
+ bb.Buf = append(bb.Buf, tfh.Buf...)
+ }
+ }
+
+ // Marshal tagFamilyFilter
+ if mp.tagFamilyFilter == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFamilyFilter)))
+ for name, tff := range mp.tagFamilyFilter {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tff.Buf)))
+ bb.Buf = append(bb.Buf, tff.Buf...)
+ }
+ }
+
+ result := make([]byte, len(bb.Buf))
+ copy(result, bb.Buf)
+ return result, nil
+}
+
+// Unmarshal deserializes the memPart from []byte.
+func (mp *memPart) Unmarshal(data []byte) error {
+ mp.reset()
+
+ // Unmarshal partMetadata
+ tail, metadataLen := encoding.BytesToVarUint64(data)
+ if uint64(len(tail)) < metadataLen {
+ return fmt.Errorf("insufficient data for partMetadata: need %d bytes, have %d", metadataLen, len(tail))
+ }
+ metadataBytes := tail[:metadataLen]
+ tail = tail[metadataLen:]
+ if err := json.Unmarshal(metadataBytes, &mp.partMetadata); err != nil {
+ return fmt.Errorf("cannot unmarshal partMetadata: %w", err)
+ }
+
+ // Unmarshal meta buffer
+ tail, metaLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < metaLen {
+ return fmt.Errorf("insufficient data for meta buffer: need %d bytes, have %d", metaLen, len(tail))
+ }
+ mp.meta.Buf = append(mp.meta.Buf[:0], tail[:metaLen]...)
+ tail = tail[metaLen:]
+
+ // Unmarshal primary buffer
+ tail, primaryLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < primaryLen {
+ return fmt.Errorf("insufficient data for primary buffer: need %d bytes, have %d", primaryLen, len(tail))
+ }
+ mp.primary.Buf = append(mp.primary.Buf[:0], tail[:primaryLen]...)
+ tail = tail[primaryLen:]
+
+ // Unmarshal timestamps buffer
+ tail, timestampsLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < timestampsLen {
+ return fmt.Errorf("insufficient data for timestamps buffer: need %d bytes, have %d", timestampsLen, len(tail))
+ }
+ mp.timestamps.Buf = append(mp.timestamps.Buf[:0], tail[:timestampsLen]...)
+ tail = tail[timestampsLen:]
+
+ var nameLen, bufLen uint64
+ // Unmarshal tagFamilies
+ tail, tagFamiliesCount := encoding.BytesToVarUint64(tail)
+ if tagFamiliesCount > 0 {
+ mp.tagFamilies = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagFamiliesCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag family name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag family buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tf := &bytes.Buffer{}
+ tf.Buf = append(tf.Buf[:0], tail[:bufLen]...)
+ mp.tagFamilies[name] = tf
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagFamilyMetadata
+ tail, tagFamilyMetadataCount := encoding.BytesToVarUint64(tail)
+ if tagFamilyMetadataCount > 0 {
+ mp.tagFamilyMetadata = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagFamilyMetadataCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag family metadata name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag family metadata buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tfh := &bytes.Buffer{}
+ tfh.Buf = append(tfh.Buf[:0], tail[:bufLen]...)
+ mp.tagFamilyMetadata[name] = tfh
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagFamilyFilter
+ tail, tagFamilyFilterCount := encoding.BytesToVarUint64(tail)
+ if tagFamilyFilterCount > 0 {
+ mp.tagFamilyFilter = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagFamilyFilterCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag family filter name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag family filter buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tff := &bytes.Buffer{}
+ tff.Buf = append(tff.Buf[:0], tail[:bufLen]...)
+ mp.tagFamilyFilter[name] = tff
+ tail = tail[bufLen:]
+ }
+ }
+
+ if len(tail) > 0 {
+ return fmt.Errorf("unexpected trailing data: %d bytes", len(tail))
+ }
+
+ return nil
+}
+
+func (mp *memPart) mustInitFromPart(p *part) {
+ mp.reset()
+
+ // Copy part metadata
+ mp.partMetadata = p.partMetadata
+
+ // Read primary data
+ sr := p.primary.SequentialRead()
+ data, err := io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read primary data from %s: %s", p.primary.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.primary.Buf = append(mp.primary.Buf[:0], data...)
+
+ // Read timestamps data
+ sr = p.timestamps.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read timestamps data from %s: %s", p.timestamps.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.timestamps.Buf = append(mp.timestamps.Buf[:0], data...)
+
+ // Read tag families data
+ if p.tagFamilies != nil {
+ mp.tagFamilies = make(map[string]*bytes.Buffer)
+ mp.tagFamilyMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFamilyFilter = make(map[string]*bytes.Buffer)
+
+ for name, reader := range p.tagFamilies {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag family data from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagFamilies[name] = &bytes.Buffer{}
+ mp.tagFamilies[name].Buf = append(mp.tagFamilies[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagFamilyMetadata {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag family metadata from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagFamilyMetadata[name] = &bytes.Buffer{}
+ mp.tagFamilyMetadata[name].Buf = append(mp.tagFamilyMetadata[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagFamilyFilter {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag family filter from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagFamilyFilter[name] = &bytes.Buffer{}
+ mp.tagFamilyFilter[name].Buf = append(mp.tagFamilyFilter[name].Buf[:0], data...)
+ }
+ }
+
+ // Encode primaryBlockMetadata to memPart.meta
+ // This is the reverse process of mustReadPrimaryBlockMetadata
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal all primaryBlockMetadata entries
+ for _, pbm := range p.primaryBlockMetadata {
+ bb.Buf = pbm.marshal(bb.Buf)
+ }
+
+ // Compress the marshaled data
+ mp.meta.Buf = zstd.Compress(mp.meta.Buf[:0], bb.Buf, 1)
+}
+
func (mp *memPart) mustInitFromElements(es *elements) {
mp.reset()
diff --git a/banyand/stream/part_test.go b/banyand/stream/part_test.go
index f4443998..926041a2 100644
--- a/banyand/stream/part_test.go
+++ b/banyand/stream/part_test.go
@@ -24,6 +24,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/fs"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
@@ -180,3 +181,321 @@ var es = &elements{
{}, // empty tagFamilies for seriesID 3
},
}
+
+func TestMustInitFromPart(t *testing.T) {
+ // Step 1: Load the global variable "es" to a memPart
+ originalMemPart := &memPart{}
+ originalMemPart.mustInitFromElements(es)
+
+ // Step 2: Flush the memPart
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ epoch := uint64(1)
+ path := partPath(tmpPath, epoch)
+ fileSystem := fs.NewLocalFileSystem()
+ originalMemPart.mustFlush(fileSystem, path)
+
+ // Step 3: Init a part from the files the memPart flushed
+ part := mustOpenFilePart(epoch, tmpPath, fileSystem)
+ defer part.close()
+
+ // Step 4: Convert the part to a new memPart
+ newMemPart := &memPart{}
+ newMemPart.mustInitFromPart(part)
+
+ // Step 5: Compare the new memPart with the original memPart
+ // Compare part metadata
+ assert.NotEqual(t, originalMemPart.partMetadata.ID, newMemPart.partMetadata.ID)
+ assert.Equal(t, originalMemPart.partMetadata.BlocksCount, newMemPart.partMetadata.BlocksCount)
+ assert.Equal(t, originalMemPart.partMetadata.MinTimestamp, newMemPart.partMetadata.MinTimestamp)
+ assert.Equal(t, originalMemPart.partMetadata.MaxTimestamp, newMemPart.partMetadata.MaxTimestamp)
+ assert.Equal(t, originalMemPart.partMetadata.TotalCount, newMemPart.partMetadata.TotalCount)
+
+ // Compare primary data
+ assert.Equal(t, originalMemPart.primary.Buf, newMemPart.primary.Buf)
+
+ // Compare timestamps data
+ assert.Equal(t, originalMemPart.timestamps.Buf, newMemPart.timestamps.Buf)
+
+ // Compare meta data (primaryBlockMetadata)
+ assert.Equal(t, originalMemPart.meta.Buf, newMemPart.meta.Buf)
+
+ // Compare tag families
+ assert.Equal(t, len(originalMemPart.tagFamilies), len(newMemPart.tagFamilies))
+ for name, originalBuffer := range originalMemPart.tagFamilies {
+ newBuffer, exists := newMemPart.tagFamilies[name]
+ assert.True(t, exists, "Tag family %s not found in new memPart", name)
+ assert.Equal(t, originalBuffer.Buf, newBuffer.Buf)
+ }
+
+ // Compare tag family metadata
+ assert.Equal(t, len(originalMemPart.tagFamilyMetadata), len(newMemPart.tagFamilyMetadata))
+ for name, originalBuffer := range originalMemPart.tagFamilyMetadata {
+ newBuffer, exists := newMemPart.tagFamilyMetadata[name]
+ assert.True(t, exists, "Tag family metadata %s not found in new memPart", name)
+ assert.Equal(t, originalBuffer.Buf, newBuffer.Buf)
+ }
+
+ // Compare tag family filters
+ assert.Equal(t, len(originalMemPart.tagFamilyFilter), len(newMemPart.tagFamilyFilter))
+ for name, originalBuffer := range originalMemPart.tagFamilyFilter {
+ newBuffer, exists := newMemPart.tagFamilyFilter[name]
+ assert.True(t, exists, "Tag family filter %s not found in new memPart", name)
+ assert.Equal(t, originalBuffer.Buf, newBuffer.Buf)
+ }
+}
+
+func Test_memPart_Marshal_Unmarshal(t *testing.T) {
+ // Create a test memPart with some data
+ mp := generateMemPart()
+ defer releaseMemPart(mp)
+
+ // Initialize with test elements
+ esData := &elements{
+ timestamps: []int64{1, 2, 3},
+ elementIDs: []uint64{100, 200, 300},
+ seriesIDs: []common.SeriesID{1, 1, 2},
+ tagFamilies: [][]tagValues{
+ {
+ {
+ tag: "service",
+ values: []*tagValue{
+ {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("test-service")},
+ },
+ },
+ },
+ {
+ {
+ tag: "service",
+ values: []*tagValue{
+ {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("test-service-2")},
+ },
+ },
+ },
+ {
+ {
+ tag: "instance",
+ values: []*tagValue{
+ {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("test-instance")},
+ },
+ },
+ },
+ },
+ }
+
+ mp.mustInitFromElements(esData)
+
+ // Test marshal
+ marshaled, err := mp.Marshal()
+ require.NoError(t, err)
+ require.NotEmpty(t, marshaled)
+
+ // Test unmarshal
+ mp2 := generateMemPart()
+ defer releaseMemPart(mp2)
+
+ err = mp2.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Verify the unmarshaled data matches the original
+ require.Equal(t, mp.partMetadata.ID, mp2.partMetadata.ID)
+ require.Equal(t, mp.partMetadata.TotalCount, mp2.partMetadata.TotalCount)
+ require.Equal(t, mp.partMetadata.BlocksCount, mp2.partMetadata.BlocksCount)
+ require.Equal(t, mp.partMetadata.MinTimestamp, mp2.partMetadata.MinTimestamp)
+ require.Equal(t, mp.partMetadata.MaxTimestamp, mp2.partMetadata.MaxTimestamp)
+
+ // Verify buffers
+ require.Equal(t, mp.meta.Buf, mp2.meta.Buf)
+ require.Equal(t, mp.primary.Buf, mp2.primary.Buf)
+ require.Equal(t, mp.timestamps.Buf, mp2.timestamps.Buf)
+
+ // Verify tag families
+ require.Equal(t, len(mp.tagFamilies), len(mp2.tagFamilies))
+ for name, tf := range mp.tagFamilies {
+ tf2, exists := mp2.tagFamilies[name]
+ require.True(t, exists)
+ require.Equal(t, tf.Buf, tf2.Buf)
+ }
+
+ // Verify tag family metadata
+ require.Equal(t, len(mp.tagFamilyMetadata), len(mp2.tagFamilyMetadata))
+ for name, tfh := range mp.tagFamilyMetadata {
+ tfh2, exists := mp2.tagFamilyMetadata[name]
+ require.True(t, exists)
+ require.Equal(t, tfh.Buf, tfh2.Buf)
+ }
+
+ // Verify tag family filter
+ require.Equal(t, len(mp.tagFamilyFilter), len(mp2.tagFamilyFilter))
+ for name, tff := range mp.tagFamilyFilter {
+ tff2, exists := mp2.tagFamilyFilter[name]
+ require.True(t, exists)
+ require.Equal(t, tff.Buf, tff2.Buf)
+ }
+}
+
+func Test_memPart_Marshal_Unmarshal_Empty(t *testing.T) {
+ // Test with empty memPart
+ mp := generateMemPart()
+ defer releaseMemPart(mp)
+
+ // Test marshal
+ marshaled, err := mp.Marshal()
+ require.NoError(t, err)
+ require.NotEmpty(t, marshaled)
+
+ // Test unmarshal
+ mp2 := generateMemPart()
+ defer releaseMemPart(mp2)
+
+ err = mp2.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Verify the unmarshaled data matches the original
+ // partMetadata.ID is not serialized, so always expect 0 after unmarshal
+ require.Equal(t, uint64(0), mp2.partMetadata.ID)
+ require.Equal(t, mp.partMetadata.TotalCount, mp2.partMetadata.TotalCount)
+ require.Equal(t, mp.partMetadata.BlocksCount, mp2.partMetadata.BlocksCount)
+ require.Equal(t, mp.partMetadata.MinTimestamp, mp2.partMetadata.MinTimestamp)
+ require.Equal(t, mp.partMetadata.MaxTimestamp, mp2.partMetadata.MaxTimestamp)
+
+ // Verify buffers are empty
+ require.Empty(t, mp2.meta.Buf)
+ require.Empty(t, mp2.primary.Buf)
+ require.Empty(t, mp2.timestamps.Buf)
+ require.Empty(t, mp2.tagFamilies)
+ require.Empty(t, mp2.tagFamilyMetadata)
+ require.Empty(t, mp2.tagFamilyFilter)
+}
+
+func Test_memPart_Marshal_Unmarshal_Simple(t *testing.T) {
+ // Create a test memPart manually without using mustInitFromElements
+ mp := generateMemPart()
+ defer releaseMemPart(mp)
+
+ // Set some basic metadata
+ mp.partMetadata.ID = 123
+ mp.partMetadata.TotalCount = 10
+ mp.partMetadata.BlocksCount = 2
+ mp.partMetadata.MinTimestamp = 1000
+ mp.partMetadata.MaxTimestamp = 2000
+
+ // Set some buffer data
+ mp.meta.Buf = []byte("test meta data")
+ mp.primary.Buf = []byte("test primary data")
+ mp.timestamps.Buf = []byte("test timestamps data")
+
+ // Create some tag families
+ mp.tagFamilies = make(map[string]*bytes.Buffer)
+ mp.tagFamilies["service"] = &bytes.Buffer{Buf: []byte("service data")}
+ mp.tagFamilies["instance"] = &bytes.Buffer{Buf: []byte("instance data")}
+
+ // Create some tag family metadata
+ mp.tagFamilyMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFamilyMetadata["service"] = &bytes.Buffer{Buf: []byte("service metadata")}
+ mp.tagFamilyMetadata["instance"] = &bytes.Buffer{Buf: []byte("instance metadata")}
+
+ // Create some tag family filters
+ mp.tagFamilyFilter = make(map[string]*bytes.Buffer)
+ mp.tagFamilyFilter["service"] = &bytes.Buffer{Buf: []byte("service filter")}
+ mp.tagFamilyFilter["instance"] = &bytes.Buffer{Buf: []byte("instance filter")}
+
+ // Test marshal
+ marshaled, err := mp.Marshal()
+ require.NoError(t, err)
+ require.NotEmpty(t, marshaled)
+
+ // Test unmarshal
+ mp2 := generateMemPart()
+ defer releaseMemPart(mp2)
+
+ err = mp2.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Verify the unmarshaled data matches the original
+ require.NotEqual(t, mp.partMetadata.ID, mp2.partMetadata.ID)
+ require.Equal(t, mp.partMetadata.TotalCount, mp2.partMetadata.TotalCount)
+ require.Equal(t, mp.partMetadata.BlocksCount, mp2.partMetadata.BlocksCount)
+ require.Equal(t, mp.partMetadata.MinTimestamp, mp2.partMetadata.MinTimestamp)
+ require.Equal(t, mp.partMetadata.MaxTimestamp, mp2.partMetadata.MaxTimestamp)
+
+ // Verify buffers
+ require.Equal(t, mp.meta.Buf, mp2.meta.Buf)
+ require.Equal(t, mp.primary.Buf, mp2.primary.Buf)
+ require.Equal(t, mp.timestamps.Buf, mp2.timestamps.Buf)
+
+ // Verify tag families
+ require.Equal(t, len(mp.tagFamilies), len(mp2.tagFamilies))
+ for name, tf := range mp.tagFamilies {
+ tf2, exists := mp2.tagFamilies[name]
+ require.True(t, exists)
+ require.Equal(t, tf.Buf, tf2.Buf)
+ }
+
+ // Verify tag family metadata
+ require.Equal(t, len(mp.tagFamilyMetadata), len(mp2.tagFamilyMetadata))
+ for name, tfh := range mp.tagFamilyMetadata {
+ tfh2, exists := mp2.tagFamilyMetadata[name]
+ require.True(t, exists)
+ require.Equal(t, tfh.Buf, tfh2.Buf)
+ }
+
+ // Verify tag family filter
+ require.Equal(t, len(mp.tagFamilyFilter), len(mp2.tagFamilyFilter))
+ for name, tff := range mp.tagFamilyFilter {
+ tff2, exists := mp2.tagFamilyFilter[name]
+ require.True(t, exists)
+ require.Equal(t, tff.Buf, tff2.Buf)
+ }
+}
+
+func Test_memPart_Marshal_Unmarshal_Minimal(t *testing.T) {
+ // Create a minimal test memPart
+ mp := generateMemPart()
+ defer releaseMemPart(mp)
+
+ // Set only basic metadata
+ mp.partMetadata.ID = 123
+ mp.partMetadata.TotalCount = 10
+
+ // Set only basic buffer data
+ mp.meta.Buf = []byte("test")
+ mp.primary.Buf = []byte("test")
+ mp.timestamps.Buf = []byte("test")
+
+ // No tag families, metadata, or filters
+
+ // Test marshal
+ marshaled, err := mp.Marshal()
+ require.NoError(t, err)
+ require.NotEmpty(t, marshaled)
+
+ // Debug: Check what was marshaled
+ t.Logf("Marshaled data length: %d", len(marshaled))
+ t.Logf("Original partMetadata ID: %d", mp.partMetadata.ID)
+
+ // Test unmarshal
+ mp2 := generateMemPart()
+ defer releaseMemPart(mp2)
+
+ err = mp2.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Debug: Check what was unmarshaled
+ t.Logf("Unmarshaled partMetadata ID: %d", mp2.partMetadata.ID)
+
+ // Verify the unmarshaled data matches the original
+ // partMetadata.ID is not serialized, so always expect 0 after unmarshal
+ require.Equal(t, uint64(0), mp2.partMetadata.ID)
+ require.Equal(t, mp.partMetadata.TotalCount, mp2.partMetadata.TotalCount)
+
+ // Verify buffers
+ require.Equal(t, mp.meta.Buf, mp2.meta.Buf)
+ require.Equal(t, mp.primary.Buf, mp2.primary.Buf)
+ require.Equal(t, mp.timestamps.Buf, mp2.timestamps.Buf)
+
+ // Verify maps are empty (not nil, but empty)
+ require.Empty(t, mp2.tagFamilies)
+ require.Empty(t, mp2.tagFamilyMetadata)
+ require.Empty(t, mp2.tagFamilyFilter)
+}
diff --git a/banyand/stream/snapshot.go b/banyand/stream/snapshot.go
index bb4b5d3b..c41a63cf 100644
--- a/banyand/stream/snapshot.go
+++ b/banyand/stream/snapshot.go
@@ -56,6 +56,7 @@ const (
snapshotCreatorFlusher
snapshotCreatorMerger
snapshotCreatorMergedFlusher
+ snapshotCreatorSyncer
)
type snapshot struct {
@@ -247,7 +248,7 @@ func (tst *tsTable) createMetadata(dst string, snapshot *snapshot) {
}
}
-func (s *service) takeGroupSnapshot(dstDir string, groupName string) error {
+func (s *standalone) takeGroupSnapshot(dstDir string, groupName string) error {
group, ok := s.schemaRepo.LoadGroup(groupName)
if !ok {
return errors.Errorf("group %s not found", groupName)
@@ -265,7 +266,7 @@ func (s *service) takeGroupSnapshot(dstDir string, groupName string) error {
type snapshotListener struct {
*bus.UnImplementedHealthyListener
- s *service
+ s *standalone
snapshotSeq uint64
snapshotMux sync.Mutex
}
diff --git a/banyand/stream/stream.go b/banyand/stream/stream.go
index 5c2d2b29..51e698a6 100644
--- a/banyand/stream/stream.go
+++ b/banyand/stream/stream.go
@@ -27,6 +27,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/partition"
"github.com/apache/skywalking-banyandb/pkg/query/model"
@@ -48,6 +49,7 @@ const (
type option struct {
mergePolicy *mergePolicy
protector protector.Memory
+ tire2Client queue.Client
seriesCacheMaxSize run.Bytes
flushTimeout time.Duration
elementIndexFlushTimeout time.Duration
@@ -86,11 +88,11 @@ func (i *indexSchema) parse(schema *databasev1.Stream) {
var _ Stream = (*stream)(nil)
type stream struct {
+ pm protector.Memory
indexSchema atomic.Value
tsdb atomic.Value
l *logger.Logger
schema *databasev1.Stream
- pm protector.Memory
schemaRepo *schemaRepo
name string
group string
diff --git a/banyand/stream/service.go b/banyand/stream/svc_liaison.go
similarity index 51%
copy from banyand/stream/service.go
copy to banyand/stream/svc_liaison.go
index 367dea43..0239077d 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/svc_liaison.go
@@ -22,7 +22,6 @@ import (
"path"
"path/filepath"
"strings"
- "time"
"github.com/pkg/errors"
@@ -30,8 +29,8 @@ import (
"github.com/apache/skywalking-banyandb/api/data"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/protector"
@@ -39,46 +38,29 @@ import (
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
"github.com/apache/skywalking-banyandb/pkg/run"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
-var (
- errEmptyRootPath = errors.New("root path is empty")
- // ErrStreamNotExist denotes a stream doesn't exist in the metadata repo.
- ErrStreamNotExist = errors.New("stream doesn't exist")
-)
-
-// Service allows inspecting the stream elements.
-type Service interface {
- run.PreRunner
- run.Config
- run.Service
- Query
-}
-
-var _ Service = (*service)(nil)
-
-type service struct {
- writeListener bus.MessageListener
+type liaison struct {
+ pm protector.Memory
metadata metadata.Repo
pipeline queue.Server
- localPipeline queue.Queue
omr observability.MetricsRegistry
lfs fs.FileSystem
- pm protector.Memory
+ writeListener bus.MessageListener
+ dataNodeSelector node.Selector
l *logger.Logger
schemaRepo schemaRepo
- root string
- snapshotDir string
dataPath string
+ root string
option option
maxDiskUsagePercent int
- maxFileSnapshotNum int
}
-func (s *service) Stream(metadata *commonv1.Metadata) (Stream, error) {
+func (s *liaison) Stream(metadata *commonv1.Metadata) (Stream, error) {
sm, ok := s.schemaRepo.loadStream(metadata)
if !ok {
return nil, errors.WithStack(ErrStreamNotExist)
@@ -86,30 +68,24 @@ func (s *service) Stream(metadata *commonv1.Metadata) (Stream, error) {
return sm, nil
}
-func (s *service) LoadGroup(name string) (resourceSchema.Group, bool) {
+func (s *liaison) LoadGroup(name string) (resourceSchema.Group, bool) {
return s.schemaRepo.LoadGroup(name)
}
-func (s *service) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+func (s *liaison) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
return s.schemaRepo.GetRemovalSegmentsTimeRange(group)
}
-func (s *service) FlagSet() *run.FlagSet {
+func (s *liaison) FlagSet() *run.FlagSet {
flagS := run.NewFlagSet("storage")
flagS.StringVar(&s.root, "stream-root-path", "/tmp", "the root path of stream")
flagS.StringVar(&s.dataPath, "stream-data-path", "", "the data directory path of stream. If not set, <stream-root-path>/stream/data will be used")
flagS.DurationVar(&s.option.flushTimeout, "stream-flush-timeout", defaultFlushTimeout, "the memory data timeout of stream")
- flagS.DurationVar(&s.option.elementIndexFlushTimeout, "element-index-flush-timeout", defaultFlushTimeout, "the elementIndex timeout of stream")
- s.option.mergePolicy = newDefaultMergePolicy()
- flagS.VarP(&s.option.mergePolicy.maxFanOutSize, "stream-max-fan-out-size", "", "the upper bound of a single file size after merge of stream")
- s.option.seriesCacheMaxSize = run.Bytes(32 << 20)
- flagS.VarP(&s.option.seriesCacheMaxSize, "stream-series-cache-max-size", "", "the max size of series cache in each group")
flagS.IntVar(&s.maxDiskUsagePercent, "stream-max-disk-usage-percent", 95, "the maximum disk usage percentage allowed")
- flagS.IntVar(&s.maxFileSnapshotNum, "stream-max-file-snapshot-num", 2, "the maximum number of file snapshots allowed")
return flagS
}
-func (s *service) Validate() error {
+func (s *liaison) Validate() error {
if s.root == "" {
return errEmptyRootPath
}
@@ -122,98 +98,56 @@ func (s *service) Validate() error {
return nil
}
-func (s *service) Name() string {
+func (s *liaison) Name() string {
return "stream"
}
-func (s *service) Role() databasev1.Role {
- return databasev1.Role_ROLE_DATA
+func (s *liaison) Role() databasev1.Role {
+ return databasev1.Role_ROLE_LIAISON
}
-func (s *service) PreRun(ctx context.Context) error {
+func (s *liaison) PreRun(ctx context.Context) error {
s.l = logger.GetLogger(s.Name())
s.l.Info().Msg("memory protector is initialized in PreRun")
s.lfs = fs.NewLocalFileSystemWithLoggerAndLimit(s.l, s.pm.GetLimit())
path := path.Join(s.root, s.Name())
- s.snapshotDir = filepath.Join(path, storage.SnapshotsDir)
observability.UpdatePath(path)
val := ctx.Value(common.ContextNodeKey)
if val == nil {
return errors.New("node id is empty")
}
- node := val.(common.Node)
if s.dataPath == "" {
s.dataPath = filepath.Join(path, storage.DataDir)
}
if !strings.HasPrefix(filepath.VolumeName(s.dataPath), filepath.VolumeName(path)) {
observability.UpdatePath(s.dataPath)
}
- s.schemaRepo = newSchemaRepo(s.dataPath, s, node.Labels)
- if s.pipeline == nil {
- return nil
- }
-
- s.localPipeline = queue.Local()
- if err := s.pipeline.Subscribe(data.TopicSnapshot, &snapshotListener{s: s}); err != nil {
- return err
- }
- if err := s.pipeline.Subscribe(data.TopicDeleteExpiredStreamSegments, &deleteStreamSegmentsListener{s: s}); err != nil {
- return err
- }
- s.writeListener = setUpWriteCallback(s.l, &s.schemaRepo, s.maxDiskUsagePercent)
- err := s.pipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
- if err != nil {
- return err
- }
- return s.localPipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
+ streamDataNodeRegistry := grpc.NewClusterNodeRegistry(data.TopicStreamPartSync, s.option.tire2Client, s.dataNodeSelector)
+ s.schemaRepo = newLiaisonSchemaRepo(s.dataPath, s, streamDataNodeRegistry)
+ s.writeListener = setUpWriteQueueCallback(s.l, &s.schemaRepo, s.maxDiskUsagePercent, s.option.tire2Client)
+ return s.pipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
}
-func (s *service) Serve() run.StopNotify {
+func (s *liaison) Serve() run.StopNotify {
return s.schemaRepo.StopCh()
}
-func (s *service) GracefulStop() {
+func (s *liaison) GracefulStop() {
s.schemaRepo.Close()
- if s.localPipeline != nil {
- s.localPipeline.GracefulStop()
- }
}
-// NewService returns a new service.
-func NewService(metadata metadata.Repo, pipeline queue.Server, omr observability.MetricsRegistry, pm protector.Memory) (Service, error) {
- return &service{
- metadata: metadata,
- pipeline: pipeline,
- omr: omr,
- pm: pm,
+// NewLiaison creates a new stream liaison service with the given dependencies.
+func NewLiaison(metadata metadata.Repo, pipeline queue.Server, omr observability.MetricsRegistry, pm protector.Memory,
+ dataNodeSelector node.Selector, tire2Client queue.Client,
+) (Service, error) {
+ return &liaison{
+ metadata: metadata,
+ pipeline: pipeline,
+ omr: omr,
+ pm: pm,
+ dataNodeSelector: dataNodeSelector,
+ option: option{
+ tire2Client: tire2Client,
+ },
}, nil
}
-
-// NewReadonlyService returns a new readonly service.
-func NewReadonlyService(metadata metadata.Repo, omr observability.MetricsRegistry, pm protector.Memory) (Service, error) {
- return &service{
- metadata: metadata,
- omr: omr,
- pm: pm,
- }, nil
-}
-
-type deleteStreamSegmentsListener struct {
- *bus.UnImplementedHealthyListener
- s *service
-}
-
-func (d *deleteStreamSegmentsListener) Rev(_ context.Context, message bus.Message) bus.Message {
- req := message.Data().(*streamv1.DeleteExpiredSegmentsRequest)
- if req == nil {
- return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), int64(0))
- }
-
- db, err := d.s.schemaRepo.loadTSDB(req.Group)
- if err != nil {
- d.s.l.Error().Err(err).Str("group", req.Group).Msg("failed to load tsdb")
- return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), int64(0))
- }
- deleted := db.DeleteExpiredSegments(timestamp.NewSectionTimeRange(req.TimeRange.Begin.AsTime(), req.TimeRange.End.AsTime()))
- return bus.NewMessage(bus.MessageID(time.Now().UnixNano()), deleted)
-}
diff --git a/banyand/stream/service.go b/banyand/stream/svc_standalone.go
similarity index 84%
rename from banyand/stream/service.go
rename to banyand/stream/svc_standalone.go
index 367dea43..f9f6a9fa 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/svc_standalone.go
@@ -58,27 +58,26 @@ type Service interface {
Query
}
-var _ Service = (*service)(nil)
+var _ Service = (*standalone)(nil)
-type service struct {
- writeListener bus.MessageListener
- metadata metadata.Repo
+type standalone struct {
+ pm protector.Memory
pipeline queue.Server
localPipeline queue.Queue
omr observability.MetricsRegistry
lfs fs.FileSystem
- pm protector.Memory
+ metadata metadata.Repo
l *logger.Logger
schemaRepo schemaRepo
- root string
snapshotDir string
+ root string
dataPath string
option option
maxDiskUsagePercent int
maxFileSnapshotNum int
}
-func (s *service) Stream(metadata *commonv1.Metadata) (Stream, error) {
+func (s *standalone) Stream(metadata *commonv1.Metadata) (Stream, error) {
sm, ok := s.schemaRepo.loadStream(metadata)
if !ok {
return nil, errors.WithStack(ErrStreamNotExist)
@@ -86,15 +85,15 @@ func (s *service) Stream(metadata *commonv1.Metadata) (Stream, error) {
return sm, nil
}
-func (s *service) LoadGroup(name string) (resourceSchema.Group, bool) {
+func (s *standalone) LoadGroup(name string) (resourceSchema.Group, bool) {
return s.schemaRepo.LoadGroup(name)
}
-func (s *service) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
+func (s *standalone) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
return s.schemaRepo.GetRemovalSegmentsTimeRange(group)
}
-func (s *service) FlagSet() *run.FlagSet {
+func (s *standalone) FlagSet() *run.FlagSet {
flagS := run.NewFlagSet("storage")
flagS.StringVar(&s.root, "stream-root-path", "/tmp", "the root path of stream")
flagS.StringVar(&s.dataPath, "stream-data-path", "", "the data directory path of stream. If not set, <stream-root-path>/stream/data will be used")
@@ -109,7 +108,7 @@ func (s *service) FlagSet() *run.FlagSet {
return flagS
}
-func (s *service) Validate() error {
+func (s *standalone) Validate() error {
if s.root == "" {
return errEmptyRootPath
}
@@ -122,15 +121,15 @@ func (s *service) Validate() error {
return nil
}
-func (s *service) Name() string {
+func (s *standalone) Name() string {
return "stream"
}
-func (s *service) Role() databasev1.Role {
+func (s *standalone) Role() databasev1.Role {
return databasev1.Role_ROLE_DATA
}
-func (s *service) PreRun(ctx context.Context) error {
+func (s *standalone) PreRun(ctx context.Context) error {
s.l = logger.GetLogger(s.Name())
s.l.Info().Msg("memory protector is initialized in PreRun")
s.lfs = fs.NewLocalFileSystemWithLoggerAndLimit(s.l, s.pm.GetLimit())
@@ -160,19 +159,31 @@ func (s *service) PreRun(ctx context.Context) error {
if err := s.pipeline.Subscribe(data.TopicDeleteExpiredStreamSegments, &deleteStreamSegmentsListener{s: s}); err != nil {
return err
}
- s.writeListener = setUpWriteCallback(s.l, &s.schemaRepo, s.maxDiskUsagePercent)
- err := s.pipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
+ writeListener := setUpWriteCallback(s.l, &s.schemaRepo, s.maxDiskUsagePercent)
+ err := s.pipeline.Subscribe(data.TopicStreamWrite, writeListener)
+ if err != nil {
+ return err
+ }
+ err = s.pipeline.Subscribe(data.TopicStreamPartSync, setUpSyncCallback(s.l, &s.schemaRepo))
+ if err != nil {
+ return err
+ }
+ err = s.pipeline.Subscribe(data.TopicStreamSeriesIndexWrite, setUpSeriesIndexCallback(s.l, &s.schemaRepo))
+ if err != nil {
+ return err
+ }
+ err = s.pipeline.Subscribe(data.TopicStreamLocalIndexWrite, setUpLocalIndexCallback(s.l, &s.schemaRepo))
if err != nil {
return err
}
- return s.localPipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
+ return s.localPipeline.Subscribe(data.TopicStreamWrite, writeListener)
}
-func (s *service) Serve() run.StopNotify {
+func (s *standalone) Serve() run.StopNotify {
return s.schemaRepo.StopCh()
}
-func (s *service) GracefulStop() {
+func (s *standalone) GracefulStop() {
s.schemaRepo.Close()
if s.localPipeline != nil {
s.localPipeline.GracefulStop()
@@ -181,7 +192,7 @@ func (s *service) GracefulStop() {
// NewService returns a new service.
func NewService(metadata metadata.Repo, pipeline queue.Server, omr observability.MetricsRegistry, pm protector.Memory) (Service, error) {
- return &service{
+ return &standalone{
metadata: metadata,
pipeline: pipeline,
omr: omr,
@@ -191,7 +202,7 @@ func NewService(metadata metadata.Repo, pipeline queue.Server, omr observability
// NewReadonlyService returns a new readonly service.
func NewReadonlyService(metadata metadata.Repo, omr observability.MetricsRegistry, pm protector.Memory) (Service, error) {
- return &service{
+ return &standalone{
metadata: metadata,
omr: omr,
pm: pm,
@@ -200,7 +211,7 @@ func NewReadonlyService(metadata metadata.Repo, omr observability.MetricsRegistr
type deleteStreamSegmentsListener struct {
*bus.UnImplementedHealthyListener
- s *service
+ s *standalone
}
func (d *deleteStreamSegmentsListener) Rev(_ context.Context, message bus.Message) bus.Message {
diff --git a/banyand/stream/syncer.go b/banyand/stream/syncer.go
new file mode 100644
index 00000000..0883340e
--- /dev/null
+++ b/banyand/stream/syncer.go
@@ -0,0 +1,160 @@
+// 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 stream
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/watcher"
+)
+
+func (tst *tsTable) syncLoop(syncCh chan *syncIntroduction, flusherNotifier watcher.Channel) {
+ defer tst.loopCloser.Done()
+
+ var epoch uint64
+
+ ew := flusherNotifier.Add(0, tst.loopCloser.CloseNotify())
+ if ew == nil {
+ return
+ }
+
+ for {
+ select {
+ case <-tst.loopCloser.CloseNotify():
+ return
+ case <-ew.Watch():
+ if func() bool {
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot == nil {
+ return false
+ }
+ defer curSnapshot.decRef()
+ if curSnapshot.epoch != epoch {
+ tst.incTotalSyncLoopStarted(1)
+ defer tst.incTotalSyncLoopFinished(1)
+ var err error
+ if err = tst.syncSnapshot(curSnapshot, syncCh); err != nil {
+ if errors.Is(err, errClosed) {
+ return true
+ }
+ tst.l.Logger.Warn().Err(err).Msgf("cannot sync snapshot: %d", curSnapshot.epoch)
+ tst.incTotalSyncLoopErr(1)
+ return false
+ }
+ epoch = curSnapshot.epoch
+ }
+ ew = flusherNotifier.Add(epoch, tst.loopCloser.CloseNotify())
+ return ew == nil
+ }() {
+ return
+ }
+ }
+ }
+}
+
+func (tst *tsTable) syncSnapshot(curSnapshot *snapshot, syncCh chan *syncIntroduction) error {
+ // Get all parts from the current snapshot
+ var partsToSync []*part
+ for _, pw := range curSnapshot.parts {
+ if pw.p.partMetadata.TotalCount > 0 {
+ partsToSync = append(partsToSync, pw.p)
+ }
+ }
+
+ if len(partsToSync) == 0 {
+ return nil
+ }
+ nodes := tst.getNodes()
+ if len(nodes) == 0 {
+ return fmt.Errorf("no nodes to sync parts")
+ }
+
+ // Sort parts from old to new (by part ID)
+ // Parts with lower IDs are older
+ for i := 0; i < len(partsToSync); i++ {
+ for j := i + 1; j < len(partsToSync); j++ {
+ if partsToSync[i].partMetadata.ID > partsToSync[j].partMetadata.ID {
+ partsToSync[i], partsToSync[j] = partsToSync[j], partsToSync[i]
+ }
+ }
+ }
+
+ // Send parts through pipeline
+ ctx := context.Background()
+
+ for _, node := range nodes {
+ for _, part := range partsToSync {
+ // Convert part to memPart
+ memPart := generateMemPart()
+ memPart.mustInitFromPart(part)
+
+ // Marshal memPart to []byte
+ partData, err := memPart.Marshal()
+ if err != nil {
+ releaseMemPart(memPart)
+ return fmt.Errorf("failed to marshal part %d: %w", part.partMetadata.ID, err)
+ }
+
+ // Encode group, shardID and prepend to partData
+ combinedData := make([]byte, 0, len(partData)+len(tst.group)+4)
+ combinedData = encoding.EncodeBytes(combinedData, convert.StringToBytes(tst.group))
+ combinedData = encoding.Uint32ToBytes(combinedData, uint32(tst.shardID))
+ combinedData = append(combinedData, partData...)
+
+ // Create message with combined data (group + shardID + partData)
+ message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), node, combinedData)
+ _, err = tst.option.tire2Client.Publish(ctx, data.TopicStreamPartSync, message)
+
+ releaseMemPart(memPart)
+
+ if err != nil {
+ return fmt.Errorf("failed to publish part %d to pipeline: %w", part.partMetadata.ID, err)
+ }
+ }
+ }
+
+ // Construct syncIntroduction to remove synced parts from snapshot
+ si := generateSyncIntroduction()
+ defer releaseSyncIntroduction(si)
+ si.applied = make(chan struct{})
+
+ // Mark all synced parts for removal
+ for _, part := range partsToSync {
+ si.synced[part.partMetadata.ID] = struct{}{}
+ }
+
+ select {
+ case syncCh <- si:
+ case <-tst.loopCloser.CloseNotify():
+ return errClosed
+ }
+ select {
+ case <-si.applied:
+ case <-tst.loopCloser.CloseNotify():
+ return errClosed
+ }
+
+ return nil
+}
diff --git a/banyand/stream/tstable.go b/banyand/stream/tstable.go
index 493444c4..93163649 100644
--- a/banyand/stream/tstable.go
+++ b/banyand/stream/tstable.go
@@ -49,20 +49,23 @@ const (
)
type tsTable struct {
- fileSystem fs.FileSystem
l *logger.Logger
- snapshot *snapshot
- introductions chan *introduction
- loopCloser *run.Closer
+ fileSystem fs.FileSystem
+ pm protector.Memory
metrics *metrics
index *elementIndex
- p common.Position
+ snapshot *snapshot
+ loopCloser *run.Closer
+ getNodes func() []string
option option
- pm protector.Memory
+ introductions chan *introduction
+ p common.Position
root string
+ group string
gc garbageCleaner
curPartID uint64
sync.RWMutex
+ shardID common.ShardID
}
func (tst *tsTable) loadSnapshot(epoch uint64, loadedParts []uint64) {
@@ -121,6 +124,18 @@ func (tst *tsTable) startLoop(cur uint64) {
go tst.mergeLoop(mergeCh, flusherWatcher)
}
+func (tst *tsTable) startLoopNoMerge(cur uint64) {
+ tst.loopCloser = run.NewCloser(1 + 3)
+ tst.introductions = make(chan *introduction)
+ flushCh := make(chan *flusherIntroduction)
+ syncCh := make(chan *syncIntroduction)
+ introducerWatcher := make(watcher.Channel, 1)
+ flusherWatcher := make(watcher.Channel, 1)
+ go tst.introducerLoopWithSync(flushCh, syncCh, introducerWatcher, cur+1)
+ go tst.flusherLoopNoMerger(flushCh, introducerWatcher, flusherWatcher, cur)
+ go tst.syncLoop(syncCh, flusherWatcher)
+}
+
func parseEpoch(epochStr string) (uint64, error) {
p, err := strconv.ParseUint(epochStr, 16, 64)
if err != nil {
@@ -169,9 +184,10 @@ func (tst *tsTable) mustReadSnapshot(snapshot uint64) []uint64 {
return result
}
-func newTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
- l *logger.Logger, _ timestamp.TimeRange, option option, m any,
-) (*tsTable, error) {
+// initTSTable initializes a tsTable and loads parts/snapshots, but does not start any background loops.
+func initTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
+ l *logger.Logger, option option, m any,
+) (*tsTable, uint64, error) {
if option.protector == nil {
logger.GetLogger("stream").
Panic().
@@ -192,15 +208,13 @@ func newTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
}
index, err := newElementIndex(context.TODO(), rootPath, option.elementIndexFlushTimeout.Nanoseconds()/int64(time.Second), indexMetrics)
if err != nil {
- return nil, err
+ return nil, 0, err
}
tst.index = index
tst.gc.init(&tst)
ee := fileSystem.ReadDir(rootPath)
if len(ee) == 0 {
- t := &tst
- t.startLoop(uint64(time.Now().UnixNano()))
- return t, nil
+ return &tst, uint64(time.Now().UnixNano()), nil
}
var loadedParts []uint64
var loadedSnapshots []uint64
@@ -241,16 +255,23 @@ func newTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
fileSystem.MustRMAll(filepath.Join(rootPath, needToDelete[i]))
}
if len(loadedParts) == 0 || len(loadedSnapshots) == 0 {
- t := &tst
- t.startLoop(uint64(time.Now().UnixNano()))
- return t, nil
+ return &tst, uint64(time.Now().UnixNano()), nil
}
sort.Slice(loadedSnapshots, func(i, j int) bool {
return loadedSnapshots[i] > loadedSnapshots[j]
})
epoch := loadedSnapshots[0]
- t := &tst
- t.loadSnapshot(epoch, loadedParts)
+ tst.loadSnapshot(epoch, loadedParts)
+ return &tst, epoch, nil
+}
+
+func newTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
+ l *logger.Logger, _ timestamp.TimeRange, option option, m any,
+) (*tsTable, error) {
+ t, epoch, err := initTSTable(fileSystem, rootPath, p, l, option, m)
+ if err != nil {
+ return nil, err
+ }
t.startLoop(epoch)
return t, nil
}
@@ -275,13 +296,7 @@ func (tst *tsTable) Close() error {
return tst.index.Close()
}
-func (tst *tsTable) mustAddElements(es *elements) {
- if len(es.seriesIDs) == 0 {
- return
- }
-
- mp := generateMemPart()
- mp.mustInitFromElements(es)
+func (tst *tsTable) mustAddMemPart(mp *memPart) {
p := openMemPart(mp)
ind := generateIntroduction()
@@ -299,11 +314,21 @@ func (tst *tsTable) mustAddElements(es *elements) {
case <-ind.applied:
case <-tst.loopCloser.CloseNotify():
}
- tst.incTotalWritten(len(es.timestamps))
+ tst.incTotalWritten(int(mp.partMetadata.TotalCount))
tst.incTotalBatch(1)
tst.incTotalBatchIntroLatency(time.Since(startTime).Seconds())
}
+func (tst *tsTable) mustAddElements(es *elements) {
+ if len(es.seriesIDs) == 0 {
+ return
+ }
+
+ mp := generateMemPart()
+ mp.mustInitFromElements(es)
+ tst.mustAddMemPart(mp)
+}
+
type tstIter struct {
err error
parts []*part
diff --git a/banyand/stream/wqueue.go b/banyand/stream/wqueue.go
new file mode 100644
index 00000000..e0852463
--- /dev/null
+++ b/banyand/stream/wqueue.go
@@ -0,0 +1,47 @@
+// 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 stream
+
+import (
+ "fmt"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// NodeSelector provides functionality to locate nodes for data distribution.
+type NodeSelector interface {
+ Locate(group, name string, shardID, replicaID uint32) (string, error)
+ fmt.Stringer
+}
+
+// newWriteQueue is like newTSTable but does not start the merge loop (or any background loops).
+func newWriteQueue(fileSystem fs.FileSystem, rootPath string, p common.Position,
+ l *logger.Logger, option option, m any, group string, shardID common.ShardID, getNodes func() []string,
+) (*tsTable, error) {
+ t, epoch, err := initTSTable(fileSystem, rootPath, p, l, option, m)
+ if err != nil {
+ return nil, err
+ }
+ t.getNodes = getNodes
+ t.group = group
+ t.shardID = shardID
+ t.startLoopNoMerge(epoch)
+ return t, nil
+}
diff --git a/banyand/stream/wqueue_test.go b/banyand/stream/wqueue_test.go
new file mode 100644
index 00000000..49f94fc1
--- /dev/null
+++ b/banyand/stream/wqueue_test.go
@@ -0,0 +1,90 @@
+// 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 stream
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "go.uber.org/mock/gomock"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+)
+
+func Test_newWriteQueue(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+
+ // Create mock tire2Client
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ mockTire2Client := queue.NewMockClient(ctrl)
+ mockTire2Client.EXPECT().
+ Publish(gomock.Any(), gomock.Any(), gomock.Any()).
+ Return(bus.Future(nil), nil).
+ AnyTimes()
+
+ // Create tsTable using newWriteQueue
+ opts := option{
+ tire2Client: mockTire2Client,
+ protector: protector.Nop{},
+ }
+
+ tst, err := newWriteQueue(
+ fs.NewLocalFileSystem(),
+ tmpPath,
+ common.Position{},
+ logger.GetLogger("test"),
+ opts,
+ nil,
+ "test-group",
+ common.ShardID(1),
+ func() []string { return []string{"node1", "node2"} },
+ )
+ require.NoError(t, err)
+ require.NotNil(t, tst)
+ defer tst.Close()
+
+ // Wait for the creator to be set to snapshotCreatorSyncer
+ assert.Eventually(t, func() bool {
+ tst.mustAddElements(es)
+ time.Sleep(100 * time.Millisecond)
+ s := tst.currentSnapshot()
+ if s == nil {
+ return false
+ }
+ defer s.decRef()
+ return s.creator == snapshotCreatorSyncer
+ }, flags.EventuallyTimeout, 500*time.Millisecond, "wait for snapshot creator to be set to syncer")
+
+ // Verify that the tsTable has the expected group and shardID
+ assert.Equal(t, "test-group", tst.group)
+ assert.Equal(t, common.ShardID(1), tst.shardID)
+
+ // Verify that the tire2Client was properly set
+ assert.Equal(t, mockTire2Client, tst.option.tire2Client)
+}
diff --git a/banyand/stream/write.go b/banyand/stream/write.go
index 1b310c18..09756c45 100644
--- a/banyand/stream/write.go
+++ b/banyand/stream/write.go
@@ -85,7 +85,7 @@ func (w *writeCallback) handle(dst map[string]*elementsInGroup, writeEvent *stre
if err != nil {
return nil, err
}
- err = w.processElements(et, eg, writeEvent, docIDBuilder, ts)
+ err = processElements(w.schemaRepo, et.elements, writeEvent, docIDBuilder, ts, &et.docs, &et.seriesDocs)
if err != nil {
return nil, err
}
@@ -102,10 +102,9 @@ func (w *writeCallback) prepareElementsInGroup(dst map[string]*elementsInGroup,
eg, ok := dst[gn]
if !ok {
eg = &elementsInGroup{
- tsdb: tsdb,
- tables: make([]*elementsInTable, 0),
- segments: make([]storage.Segment[*tsTable, option], 0),
- docIDsAdded: make(map[uint64]struct{}), // Initialize the map
+ tsdb: tsdb,
+ tables: make([]*elementsInTable, 0),
+ segments: make([]storage.Segment[*tsTable, option], 0),
}
dst[gn] = eg
}
@@ -151,6 +150,11 @@ func (w *writeCallback) prepareElementsInTable(eg *elementsInGroup, writeEvent *
timeRange: segment.GetTimeRange(),
tsTable: tstb,
elements: generateElements(),
+ segment: segment,
+ seriesDocs: seriesDoc{
+ docs: make(index.Documents, 0),
+ docIDsAdded: make(map[uint64]struct{}),
+ },
}
et.elements.reset()
eg.tables = append(eg.tables, et)
@@ -158,20 +162,20 @@ func (w *writeCallback) prepareElementsInTable(eg *elementsInGroup, writeEvent *
return et, nil
}
-func (w *writeCallback) processElements(et *elementsInTable, eg *elementsInGroup, writeEvent *streamv1.InternalWriteRequest,
- docIDBuilder *strings.Builder, ts int64,
+func processElements(schemaRepo *schemaRepo, elements *elements, writeEvent *streamv1.InternalWriteRequest,
+ docIDBuilder *strings.Builder, ts int64, tableDocs *index.Documents, seriesDocs *seriesDoc,
) error {
req := writeEvent.Request
- et.elements.timestamps = append(et.elements.timestamps, ts)
+ elements.timestamps = append(elements.timestamps, ts)
docIDBuilder.Reset()
docIDBuilder.WriteString(req.Metadata.Name)
docIDBuilder.WriteByte('|')
docIDBuilder.WriteString(req.Element.ElementId)
eID := convert.HashStr(docIDBuilder.String())
- et.elements.elementIDs = append(et.elements.elementIDs, eID)
+ elements.elementIDs = append(elements.elementIDs, eID)
- stm, ok := w.schemaRepo.loadStream(writeEvent.GetRequest().GetMetadata())
+ stm, ok := schemaRepo.loadStream(writeEvent.GetRequest().GetMetadata())
if !ok {
return fmt.Errorf("cannot find stream definition: %s", writeEvent.GetRequest().GetMetadata())
}
@@ -191,7 +195,7 @@ func (w *writeCallback) processElements(et *elementsInTable, eg *elementsInGroup
if err := series.Marshal(); err != nil {
return fmt.Errorf("cannot marshal series: %w", err)
}
- et.elements.seriesIDs = append(et.elements.seriesIDs, series.ID)
+ elements.seriesIDs = append(elements.seriesIDs, series.ID)
is := stm.indexSchema.Load().(indexSchema)
tagFamilies := make([]tagValues, 0, len(stm.schema.TagFamilies))
@@ -250,21 +254,21 @@ func (w *writeCallback) processElements(et *elementsInTable, eg *elementsInGroup
tagFamilies = append(tagFamilies, tf)
}
}
- et.elements.tagFamilies = append(et.elements.tagFamilies, tagFamilies)
+ elements.tagFamilies = append(elements.tagFamilies, tagFamilies)
- et.docs = append(et.docs, index.Document{
+ *tableDocs = append(*tableDocs, index.Document{
DocID: eID,
Fields: fields,
Timestamp: ts,
})
docID := uint64(series.ID)
- if _, exists := eg.docIDsAdded[docID]; !exists {
- eg.docs = append(eg.docs, index.Document{
+ if _, existed := seriesDocs.docIDsAdded[docID]; !existed {
+ seriesDocs.docs = append(seriesDocs.docs, index.Document{
DocID: docID,
EntityValues: series.Buffer,
})
- eg.docIDsAdded[docID] = struct{}{}
+ seriesDocs.docIDsAdded[docID] = struct{}{}
}
return nil
@@ -316,12 +320,14 @@ func (w *writeCallback) Rev(_ context.Context, message bus.Message) (resp bus.Me
w.l.Error().Err(err).Msg("cannot write element index")
}
}
+ if len(es.seriesDocs.docs) > 0 {
+ if err := es.segment.IndexDB().Insert(es.seriesDocs.docs); err != nil {
+ w.l.Error().Err(err).Msg("cannot write series index")
+ }
+ }
}
- if len(g.docs) > 0 {
+ if len(g.segments) > 0 {
for _, segment := range g.segments {
- if err := segment.IndexDB().Insert(g.docs); err != nil {
- w.l.Error().Err(err).Msg("cannot write index")
- }
segment.DecRef()
}
}
diff --git a/banyand/stream/write_liaison.go b/banyand/stream/write_liaison.go
new file mode 100644
index 00000000..0007d15b
--- /dev/null
+++ b/banyand/stream/write_liaison.go
@@ -0,0 +1,244 @@
+// 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 stream
+
+import (
+ "context"
+ "fmt"
+ "strings"
+ "time"
+
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+type writeQueueCallback struct {
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ tire2Client queue.Client
+ maxDiskUsagePercent int
+}
+
+func setUpWriteQueueCallback(l *logger.Logger, schemaRepo *schemaRepo, maxDiskUsagePercent int, tire2Client queue.Client) bus.MessageListener {
+ if maxDiskUsagePercent > 100 {
+ maxDiskUsagePercent = 100
+ }
+ return &writeQueueCallback{
+ l: l,
+ schemaRepo: schemaRepo,
+ maxDiskUsagePercent: maxDiskUsagePercent,
+ tire2Client: tire2Client,
+ }
+}
+
+func (w *writeQueueCallback) CheckHealth() *common.Error {
+ if w.maxDiskUsagePercent < 1 {
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "stream is readonly because \"stream-max-disk-usage-percent\" is 0")
+ }
+ diskPercent := observability.GetPathUsedPercent(w.schemaRepo.path)
+ if diskPercent < w.maxDiskUsagePercent {
+ return nil
+ }
+ w.l.Warn().Int("maxPercent", w.maxDiskUsagePercent).Int("diskPercent", diskPercent).Msg("disk usage is too high, stop writing")
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "disk usage is too high, stop writing")
+}
+
+func (w *writeQueueCallback) handle(dst map[string]*elementsInQueue, writeEvent *streamv1.InternalWriteRequest,
+ docIDBuilder *strings.Builder,
+) (map[string]*elementsInQueue, error) {
+ t := writeEvent.Request.Element.Timestamp.AsTime().Local()
+ if err := timestamp.Check(t); err != nil {
+ return nil, fmt.Errorf("invalid timestamp: %w", err)
+ }
+ ts := t.UnixNano()
+ eq, err := w.prepareElementsInQueue(dst, writeEvent)
+ if err != nil {
+ return nil, err
+ }
+ et, err := w.prepareElementsInTable(eq, writeEvent, ts)
+ if err != nil {
+ return nil, err
+ }
+ err = processElements(w.schemaRepo, et.elements, writeEvent, docIDBuilder, ts, &et.docs, &et.seriesDocs)
+ if err != nil {
+ return nil, err
+ }
+ return dst, nil
+}
+
+func (w *writeQueueCallback) prepareElementsInQueue(dst map[string]*elementsInQueue, writeEvent *streamv1.InternalWriteRequest) (*elementsInQueue, error) {
+ gn := writeEvent.Request.Metadata.Group
+ queue, err := w.schemaRepo.loadQueue(gn)
+ if err != nil {
+ return nil, fmt.Errorf("cannot load queue for group %s: %w", gn, err)
+ }
+
+ eq, ok := dst[gn]
+ if !ok {
+ eq = &elementsInQueue{
+ name: gn,
+ queue: queue,
+ tables: make([]*elementsInTable, 0),
+ }
+ dst[gn] = eq
+ }
+ return eq, nil
+}
+
+func (w *writeQueueCallback) prepareElementsInTable(eq *elementsInQueue, writeEvent *streamv1.InternalWriteRequest, ts int64) (*elementsInTable, error) {
+ var et *elementsInTable
+ for i := range eq.tables {
+ if eq.tables[i].timeRange.Contains(ts) {
+ et = eq.tables[i]
+ break
+ }
+ }
+
+ if et == nil {
+ shardID := common.ShardID(writeEvent.ShardId)
+ shard, err := eq.queue.GetOrCreateShard(shardID)
+ if err != nil {
+ return nil, fmt.Errorf("cannot create shard: %w", err)
+ }
+
+ tstb := shard.SubQueue()
+ timeRange := eq.queue.GetTimeRange(time.Unix(0, ts))
+
+ et = &elementsInTable{
+ shardID: shardID,
+ timeRange: timeRange,
+ tsTable: tstb,
+ elements: generateElements(),
+ seriesDocs: seriesDoc{
+ docs: make(index.Documents, 0),
+ docIDsAdded: make(map[uint64]struct{}),
+ },
+ }
+ et.elements.reset()
+ eq.tables = append(eq.tables, et)
+ }
+ return et, nil
+}
+
+func (w *writeQueueCallback) Rev(ctx context.Context, message bus.Message) (resp bus.Message) {
+ events, ok := message.Data().([]any)
+ if !ok {
+ w.l.Warn().Msg("invalid event data type")
+ return
+ }
+ if len(events) < 1 {
+ w.l.Warn().Msg("empty event")
+ return
+ }
+ groups := make(map[string]*elementsInQueue)
+ var builder strings.Builder
+ for i := range events {
+ var writeEvent *streamv1.InternalWriteRequest
+ switch e := events[i].(type) {
+ case *streamv1.InternalWriteRequest:
+ writeEvent = e
+ case []byte:
+ writeEvent = &streamv1.InternalWriteRequest{}
+ if err := proto.Unmarshal(e, writeEvent); err != nil {
+ w.l.Error().Err(err).RawJSON("written", e).Msg("fail to unmarshal event")
+ continue
+ }
+ default:
+ w.l.Warn().Msg("invalid event data type")
+ continue
+ }
+ var err error
+ if groups, err = w.handle(groups, writeEvent, &builder); err != nil {
+ w.l.Error().Err(err).Msg("cannot handle write event")
+ groups = make(map[string]*elementsInQueue)
+ continue
+ }
+ }
+ for i := range groups {
+ g := groups[i]
+ for j := range g.tables {
+ es := g.tables[j]
+ es.tsTable.mustAddElements(es.elements)
+ releaseElements(es.elements)
+ // Get nodes for this shard
+ nodes := g.queue.GetNodes(es.shardID)
+ if len(nodes) == 0 {
+ w.l.Warn().Uint32("shardID", uint32(es.shardID)).Msg("no nodes found for shard")
+ continue
+ }
+ // Process series documents independently
+ if len(es.seriesDocs.docs) > 0 {
+ seriesDocData, marshalErr := es.seriesDocs.docs.Marshal()
+ if marshalErr != nil {
+ w.l.Error().Err(marshalErr).Uint32("shardID", uint32(es.shardID)).Msg("failed to marshal series documents")
+ } else {
+ // Encode group name, start timestamp from timeRange, and prepend to docData
+ combinedData := make([]byte, 0, len(seriesDocData)+len(g.name)+8)
+ combinedData = encoding.EncodeBytes(combinedData, convert.StringToBytes(g.name))
+ combinedData = encoding.Int64ToBytes(combinedData, es.timeRange.Start.UnixNano())
+ combinedData = append(combinedData, seriesDocData...)
+
+ // Send to all nodes for this shard
+ for _, node := range nodes {
+ message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), node, combinedData)
+ _, publishErr := w.tire2Client.Publish(ctx, data.TopicStreamSeriesIndexWrite, message)
+ if publishErr != nil {
+ w.l.Error().Err(publishErr).Str("node", node).Uint32("shardID", uint32(es.shardID)).Msg("failed to publish series index to node")
+ }
+ }
+ }
+ }
+
+ // Process documents independently
+ if len(es.docs) > 0 {
+ docData, marshalErr := es.docs.Marshal()
+ if marshalErr != nil {
+ w.l.Error().Err(marshalErr).Uint32("shardID", uint32(es.shardID)).Msg("failed to marshal documents")
+ } else {
+ // Encode group name, shardID and prepend to docData
+ combinedData := make([]byte, 0, len(docData)+len(g.name)+4)
+ combinedData = encoding.EncodeBytes(combinedData, convert.StringToBytes(g.name))
+ combinedData = encoding.Uint32ToBytes(combinedData, uint32(es.shardID))
+ combinedData = append(combinedData, docData...)
+
+ // Send to all nodes for this shard
+ for _, node := range nodes {
+ message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), node, combinedData)
+ _, publishErr := w.tire2Client.Publish(ctx, data.TopicStreamLocalIndexWrite, message)
+ if publishErr != nil {
+ w.l.Error().Err(publishErr).Str("node", node).Uint32("shardID", uint32(es.shardID)).Msg("failed to publish local index to node")
+ }
+ }
+ }
+ }
+ }
+ }
+ return
+}
diff --git a/banyand/stream/write_standalone.go b/banyand/stream/write_standalone.go
new file mode 100644
index 00000000..6d927b7e
--- /dev/null
+++ b/banyand/stream/write_standalone.go
@@ -0,0 +1,283 @@
+// 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 stream
+
+import (
+ "context"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+type syncCallback struct {
+ l *logger.Logger
+ schemaRepo *schemaRepo
+}
+
+func setUpSyncCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
+ return &syncCallback{
+ l: l,
+ schemaRepo: schemaRepo,
+ }
+}
+
+func (s *syncCallback) CheckHealth() *common.Error {
+ return nil
+}
+
+func (s *syncCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
+ data, ok := message.Data().([]byte)
+ if !ok {
+ s.l.Warn().Msg("invalid sync message data type")
+ return
+ }
+
+ // Decode group name
+ tail, groupBytes, err := encoding.DecodeBytes(data)
+ if err != nil {
+ s.l.Error().Err(err).Msg("failed to decode group name from sync message")
+ return
+ }
+ group := string(groupBytes)
+
+ // Decode shardID (uint32)
+ if len(tail) < 4 {
+ s.l.Error().Msg("insufficient data for shardID in sync message")
+ return
+ }
+ shardID := encoding.BytesToUint32(tail[:4])
+ tail = tail[4:]
+
+ // Decode memPart
+ memPart := generateMemPart()
+
+ if err = memPart.Unmarshal(tail); err != nil {
+ s.l.Error().Err(err).Msg("failed to unmarshal memPart from sync message")
+ return
+ }
+
+ // Get the group schema from schemaRepo by the string group
+ tsdb, err := s.schemaRepo.loadTSDB(group)
+ if err != nil {
+ s.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
+ return
+ }
+
+ // Get the segment using memPart.partMetadata.MinTimestamp
+ segmentTime := time.Unix(0, memPart.partMetadata.MinTimestamp)
+ segment, err := tsdb.CreateSegmentIfNotExist(segmentTime)
+ if err != nil {
+ s.l.Error().Err(err).Str("group", group).Time("segmentTime", segmentTime).Msg("failed to create segment")
+ return
+ }
+ defer segment.DecRef()
+
+ // Get Shard from a segment using shardID
+ tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(shardID))
+ if err != nil {
+ s.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Msg("failed to create ts table")
+ return
+ }
+
+ // Use tsTable.mustAddMemPart to introduce memPart to tsTable
+ tsTable.mustAddMemPart(memPart)
+ tsdb.Tick(memPart.partMetadata.MaxTimestamp)
+
+ s.l.Info().
+ Str("group", group).
+ Uint32("shardID", shardID).
+ Uint64("partID", memPart.partMetadata.ID).
+ Msg("successfully stored memPart to tsTable")
+
+ return
+}
+
+type seriesIndexCallback struct {
+ l *logger.Logger
+ schemaRepo *schemaRepo
+}
+
+func setUpSeriesIndexCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
+ return &seriesIndexCallback{
+ l: l,
+ schemaRepo: schemaRepo,
+ }
+}
+
+func (s *seriesIndexCallback) CheckHealth() *common.Error {
+ return nil
+}
+
+func (s *seriesIndexCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
+ data, ok := message.Data().([]byte)
+ if !ok {
+ s.l.Warn().Msg("invalid series index message data type")
+ return
+ }
+
+ // Decode group name (first part of the message)
+ tail, groupBytes, err := encoding.DecodeBytes(data)
+ if err != nil {
+ s.l.Error().Err(err).Msg("failed to decode group name from series index message")
+ return
+ }
+ group := string(groupBytes)
+
+ // Decode timestamp (uint64) - second part of the message
+ if len(tail) < 8 {
+ s.l.Error().Msg("insufficient data for timestamp in series index message")
+ return
+ }
+ timestamp := encoding.BytesToInt64(tail[:8])
+ tail = tail[8:]
+
+ // Unmarshal Documents from the remaining data
+ var documents index.Documents
+ if err = documents.Unmarshal(tail); err != nil {
+ s.l.Error().Err(err).Msg("failed to unmarshal documents from series index message")
+ return
+ }
+
+ if len(documents) == 0 {
+ s.l.Warn().Msg("empty documents in series index message")
+ return
+ }
+
+ // Get TSDB by group name using schemaRepo
+ tsdb, err := s.schemaRepo.loadTSDB(group)
+ if err != nil {
+ s.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
+ return
+ }
+
+ // Use the timestamp to get the segment directly
+ segmentTime := time.Unix(0, timestamp)
+ segment, err := tsdb.CreateSegmentIfNotExist(segmentTime)
+ if err != nil {
+ s.l.Error().Err(err).Str("group", group).Time("segmentTime", segmentTime).Msg("failed to create segment")
+ return
+ }
+ defer segment.DecRef()
+
+ // Insert all documents into the segment's index
+ if err := segment.IndexDB().Insert(documents); err != nil {
+ s.l.Error().Err(err).Str("group", group).Int("documentCount", len(documents)).Msg("failed to insert documents to index")
+ return
+ }
+
+ s.l.Info().
+ Str("group", group).
+ Int("documentCount", len(documents)).
+ Time("segmentTime", segmentTime).
+ Msg("successfully inserted documents to series index")
+
+ return
+}
+
+type localIndexCallback struct {
+ l *logger.Logger
+ schemaRepo *schemaRepo
+}
+
+func setUpLocalIndexCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
+ return &localIndexCallback{
+ l: l,
+ schemaRepo: schemaRepo,
+ }
+}
+
+func (l *localIndexCallback) CheckHealth() *common.Error {
+ return nil
+}
+
+func (l *localIndexCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
+ data, ok := message.Data().([]byte)
+ if !ok {
+ l.l.Warn().Msg("invalid local index message data type")
+ return
+ }
+
+ // Decode group name (first part of the message)
+ tail, groupBytes, err := encoding.DecodeBytes(data)
+ if err != nil {
+ l.l.Error().Err(err).Msg("failed to decode group name from local index message")
+ return
+ }
+ group := string(groupBytes)
+
+ // Decode shardID (uint32)
+ if len(tail) < 4 {
+ l.l.Error().Msg("insufficient data for shardID in local index message")
+ return
+ }
+ shardID := encoding.BytesToUint32(tail[:4])
+ tail = tail[4:]
+
+ // Unmarshal Documents from the remaining data
+ var documents index.Documents
+ if err = documents.Unmarshal(tail); err != nil {
+ l.l.Error().Err(err).Msg("failed to unmarshal documents from local index message")
+ return
+ }
+
+ if len(documents) == 0 {
+ l.l.Warn().Msg("empty documents in local index message")
+ return
+ }
+
+ // Get TSDB by group name using schemaRepo
+ tsdb, err := l.schemaRepo.loadTSDB(group)
+ if err != nil {
+ l.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
+ return
+ }
+
+ // Use the first document's timestamp to get the segment
+ firstDocTime := time.Unix(0, documents[0].Timestamp)
+ segment, err := tsdb.CreateSegmentIfNotExist(firstDocTime)
+ if err != nil {
+ l.l.Error().Err(err).Str("group", group).Time("docTime", firstDocTime).Msg("failed to create segment for documents")
+ return
+ }
+ defer segment.DecRef()
+
+ // Get the tsTable using shardID
+ tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(shardID))
+ if err != nil {
+ l.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Msg("failed to create ts table")
+ return
+ }
+
+ // Insert documents into tsTable.index
+ if err := tsTable.Index().Write(documents); err != nil {
+ l.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Int("documentCount", len(documents)).Msg("failed to insert documents to tsTable index")
+ return
+ }
+
+ l.l.Info().
+ Str("group", group).
+ Uint32("shardID", shardID).
+ Int("documentCount", len(documents)).
+ Msg("successfully inserted documents to tsTable index")
+
+ return
+}
diff --git a/pkg/cmdsetup/liaison.go b/pkg/cmdsetup/liaison.go
index 7f2efcb5..fb209d6d 100644
--- a/pkg/cmdsetup/liaison.go
+++ b/pkg/cmdsetup/liaison.go
@@ -31,9 +31,11 @@ import (
"github.com/apache/skywalking-banyandb/banyand/liaison/http"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/queue/pub"
"github.com/apache/skywalking-banyandb/banyand/queue/sub"
+ "github.com/apache/skywalking-banyandb/banyand/stream"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/node"
"github.com/apache/skywalking-banyandb/pkg/run"
@@ -50,13 +52,19 @@ func newLiaisonCmd(runners ...run.Unit) *cobra.Command {
tire1Client := pub.New(metaSvc, databasev1.Role_ROLE_LIAISON)
tire2Client := pub.New(metaSvc, databasev1.Role_ROLE_DATA)
localPipeline := queue.Local()
+
measureLiaisonNodeSel := node.NewRoundRobinSelector(data.TopicMeasureWrite.String(), metaSvc)
measureLiaisonNodeRegistry := grpc.NewClusterNodeRegistry(data.TopicMeasureWrite, tire1Client, measureLiaisonNodeSel)
measureDataNodeSel := node.NewRoundRobinSelector(data.TopicMeasureWrite.String(), metaSvc)
metricSvc := observability.NewMetricService(metaSvc, tire1Client, "liaison", measureLiaisonNodeRegistry)
+ pm := protector.NewMemory(metricSvc)
internalPipeline := sub.NewServerWithPorts(metricSvc, "liaison-server", 18912, 18913)
+ streamDataNodeSel := node.NewRoundRobinSelector(data.TopicStreamPartSync.String(), metaSvc)
+ streamSVC, err := stream.NewLiaison(metaSvc, internalPipeline, metricSvc, pm, streamDataNodeSel, tire2Client)
+ if err != nil {
+ l.Fatal().Err(err).Msg("failed to initiate stream liaison service")
+ }
streamLiaisonNodeSel := node.NewRoundRobinSelector(data.TopicStreamWrite.String(), metaSvc)
- streamDataNodeSel := node.NewRoundRobinSelector(data.TopicStreamWrite.String(), metaSvc)
propertyNodeSel := node.NewRoundRobinSelector(data.TopicPropertyUpdate.String(), metaSvc)
topNPipeline := queue.Local()
dQuery, err := dquery.NewService(metaSvc, localPipeline, tire2Client, topNPipeline, metricSvc)
@@ -67,7 +75,6 @@ func newLiaisonCmd(runners ...run.Unit) *cobra.Command {
MeasureLiaisonNodeRegistry: measureLiaisonNodeRegistry,
MeasureDataNodeRegistry: grpc.NewClusterNodeRegistry(data.TopicMeasureWrite, tire2Client, measureDataNodeSel),
StreamLiaisonNodeRegistry: grpc.NewClusterNodeRegistry(data.TopicStreamWrite, tire1Client, streamLiaisonNodeSel),
- StreamDataNodeRegistry: grpc.NewClusterNodeRegistry(data.TopicStreamWrite, tire2Client, streamDataNodeSel),
PropertyNodeRegistry: grpc.NewClusterNodeRegistry(data.TopicPropertyUpdate, tire2Client, propertyNodeSel),
}, metricSvc, dQuery, internalPipeline)
profSvc := observability.NewProfService()
@@ -86,6 +93,7 @@ func newLiaisonCmd(runners ...run.Unit) *cobra.Command {
streamLiaisonNodeSel,
streamDataNodeSel,
propertyNodeSel,
+ streamSVC,
dQuery,
grpcServer,
httpServer,
diff --git a/pkg/cmdsetup/standalone.go b/pkg/cmdsetup/standalone.go
index 9562e706..c5f0682f 100644
--- a/pkg/cmdsetup/standalone.go
+++ b/pkg/cmdsetup/standalone.go
@@ -74,7 +74,6 @@ func newStandaloneCmd(runners ...run.Unit) *cobra.Command {
grpcServer := grpc.NewServer(ctx, liaisonPipeline, dataPipeline, dataPipeline, nil, metaSvc, grpc.NodeRegistries{
MeasureLiaisonNodeRegistry: nr,
MeasureDataNodeRegistry: nr,
- StreamDataNodeRegistry: nr,
StreamLiaisonNodeRegistry: nr,
PropertyNodeRegistry: nr,
}, metricSvc, measureSvc, liaisonPipeline)
diff --git a/pkg/index/document_test.go b/pkg/index/document_test.go
new file mode 100644
index 00000000..f158b582
--- /dev/null
+++ b/pkg/index/document_test.go
@@ -0,0 +1,817 @@
+// 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 index
+
+import (
+ "bytes"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+)
+
+func TestDocument_Marshal_Unmarshal(t *testing.T) {
+ tests := []struct {
+ name string
+ document Document
+ }{
+ {
+ name: "Document with multiple fields",
+ document: Document{
+ EntityValues: []byte("entity-data"),
+ Timestamp: 1640995200000, // 2022-01-01 00:00:00 UTC
+ DocID: 12345,
+ Version: 1,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("2024-12-31")},
+ Lower: &BytesTermValue{Value: []byte("2024-01-01")},
+ IncludesUpper: true,
+ IncludesLower: false,
+ },
+ Analyzer: "keyword",
+ TagName: "service.name",
+ SeriesID: common.SeriesID(123456789),
+ IndexRuleID: 42,
+ },
+ term: &BytesTermValue{Value: []byte("test-service")},
+ NoSort: false,
+ Store: true,
+ Index: true,
+ },
+ {
+ Key: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &FloatTermValue{Value: 100.0},
+ Lower: &FloatTermValue{Value: 0.0},
+ IncludesUpper: true,
+ IncludesLower: true,
+ },
+ Analyzer: "standard",
+ TagName: "response.time",
+ SeriesID: common.SeriesID(987654321),
+ IndexRuleID: 99,
+ },
+ term: &FloatTermValue{Value: 45.67},
+ NoSort: true,
+ Store: false,
+ Index: true,
+ },
+ },
+ },
+ },
+ {
+ name: "Document with empty fields",
+ document: Document{
+ EntityValues: []byte{},
+ Timestamp: 0,
+ DocID: 0,
+ Version: 0,
+ Fields: []Field{},
+ },
+ },
+ {
+ name: "Document with single field",
+ document: Document{
+ EntityValues: []byte("single-field-entity"),
+ Timestamp: 1640995200000,
+ DocID: 999,
+ Version: 5,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "simple",
+ TagName: "status",
+ SeriesID: common.SeriesID(555),
+ IndexRuleID: 1,
+ },
+ term: &BytesTermValue{Value: []byte("success")},
+ NoSort: false,
+ Store: true,
+ Index: false,
+ },
+ },
+ },
+ },
+ {
+ name: "Document with mixed field types",
+ document: Document{
+ EntityValues: []byte("mixed-types-entity"),
+ Timestamp: 1640995200000,
+ DocID: 777,
+ Version: 3,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "keyword",
+ TagName: "string_field",
+ SeriesID: common.SeriesID(111),
+ IndexRuleID: 10,
+ },
+ term: &BytesTermValue{Value: []byte("string value")},
+ NoSort: false,
+ Store: true,
+ Index: true,
+ },
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "standard",
+ TagName: "float_field",
+ SeriesID: common.SeriesID(222),
+ IndexRuleID: 20,
+ },
+ term: &FloatTermValue{Value: 3.14159},
+ NoSort: true,
+ Store: false,
+ Index: true,
+ },
+ },
+ },
+ },
+ {
+ name: "Document with large values",
+ document: Document{
+ EntityValues: bytes.Repeat([]byte("large-entity-data-"), 100),
+ Timestamp: 1640995200000,
+ DocID: 123456789,
+ Version: 1000,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "keyword",
+ TagName: "large_field",
+ SeriesID: common.SeriesID(999999999),
+ IndexRuleID: 999,
+ },
+ term: &BytesTermValue{Value: bytes.Repeat([]byte("large-field-value-"), 50)},
+ NoSort: false,
+ Store: true,
+ Index: true,
+ },
+ },
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Marshal
+ marshaled, err := tt.document.Marshal()
+ require.NoError(t, err)
+ assert.NotNil(t, marshaled)
+ assert.Greater(t, len(marshaled), 0)
+
+ // Unmarshal
+ var unmarshaled Document
+ err = unmarshaled.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Verify basic fields
+ assert.Equal(t, tt.document.EntityValues, unmarshaled.EntityValues)
+ assert.Equal(t, tt.document.Timestamp, unmarshaled.Timestamp)
+ assert.Equal(t, tt.document.DocID, unmarshaled.DocID)
+ assert.Equal(t, tt.document.Version, unmarshaled.Version)
+
+ // Verify fields count
+ assert.Equal(t, len(tt.document.Fields), len(unmarshaled.Fields))
+
+ // Verify each field
+ for i, expectedField := range tt.document.Fields {
+ require.Less(t, i, len(unmarshaled.Fields))
+ actualField := unmarshaled.Fields[i]
+
+ // Check FieldKey
+ assert.Equal(t, expectedField.Key.Analyzer, actualField.Key.Analyzer)
+ assert.Equal(t, expectedField.Key.TagName, actualField.Key.TagName)
+ assert.Equal(t, expectedField.Key.SeriesID, actualField.Key.SeriesID)
+ assert.Equal(t, expectedField.Key.IndexRuleID, actualField.Key.IndexRuleID)
+
+ // Check TimeRange
+ if expectedField.Key.TimeRange != nil {
+ assert.NotNil(t, actualField.Key.TimeRange)
+ assert.Equal(t, expectedField.Key.TimeRange.IncludesUpper, actualField.Key.TimeRange.IncludesUpper)
+ assert.Equal(t, expectedField.Key.TimeRange.IncludesLower, actualField.Key.TimeRange.IncludesLower)
+
+ // Compare Upper bound
+ switch expected := expectedField.Key.TimeRange.Upper.(type) {
+ case *BytesTermValue:
+ actual, ok := actualField.Key.TimeRange.Upper.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := actualField.Key.TimeRange.Upper.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+
+ // Compare Lower bound
+ switch expected := expectedField.Key.TimeRange.Lower.(type) {
+ case *BytesTermValue:
+ actual, ok := actualField.Key.TimeRange.Lower.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := actualField.Key.TimeRange.Lower.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ } else {
+ assert.Nil(t, actualField.Key.TimeRange)
+ }
+
+ // Check term value
+ switch expected := expectedField.term.(type) {
+ case *BytesTermValue:
+ actual, ok := actualField.term.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := actualField.term.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+
+ // Check flags
+ assert.Equal(t, expectedField.NoSort, actualField.NoSort)
+ assert.Equal(t, expectedField.Store, actualField.Store)
+ assert.Equal(t, expectedField.Index, actualField.Index)
+ }
+ })
+ }
+}
+
+func TestDocument_Unmarshal_InvalidData(t *testing.T) {
+ tests := []struct {
+ name string
+ expectedErr string
+ data []byte
+ }{
+ {
+ name: "empty data",
+ data: []byte{},
+ expectedErr: "empty data for document",
+ },
+ {
+ name: "insufficient data for entity values",
+ data: []byte{0x01}, // Just a single byte, insufficient for varint length
+ expectedErr: "failed to decode entity values",
+ },
+ {
+ name: "insufficient data for timestamp",
+ data: []byte{0x00}, // Empty entity values, but no timestamp data
+ expectedErr: "failed to decode timestamp",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var document Document
+ err := document.Unmarshal(tt.data)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tt.expectedErr)
+ })
+ }
+}
+
+func TestDocument_Marshal_Unmarshal_RoundTrip(t *testing.T) {
+ // Test round-trip marshaling/unmarshaling with a complex document
+ original := Document{
+ EntityValues: []byte("round-trip-test-entity"),
+ Timestamp: 1640995200000,
+ DocID: 987654321,
+ Version: 42,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("2024-12-31T23:59:59Z")},
+ Lower: &FloatTermValue{Value: 1672531200.0},
+ IncludesUpper: true,
+ IncludesLower: false,
+ },
+ Analyzer: "standard",
+ TagName: "service.instance.name",
+ SeriesID: common.SeriesID(1234567890123456789),
+ IndexRuleID: 999,
+ },
+ term: &BytesTermValue{Value: []byte("test-instance")},
+ NoSort: true,
+ Store: true,
+ Index: false,
+ },
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "keyword",
+ TagName: "status",
+ SeriesID: common.SeriesID(9876543210987654321),
+ IndexRuleID: 888,
+ },
+ term: &FloatTermValue{Value: 200.0},
+ NoSort: false,
+ Store: false,
+ Index: true,
+ },
+ },
+ }
+
+ // First round-trip
+ marshaled1, err := original.Marshal()
+ require.NoError(t, err)
+
+ var unmarshaled1 Document
+ err = unmarshaled1.Unmarshal(marshaled1)
+ require.NoError(t, err)
+
+ // Second round-trip
+ marshaled2, err := unmarshaled1.Marshal()
+ require.NoError(t, err)
+
+ var unmarshaled2 Document
+ err = unmarshaled2.Unmarshal(marshaled2)
+ require.NoError(t, err)
+
+ // Verify all fields are preserved through multiple round-trips
+ assert.Equal(t, original.EntityValues, unmarshaled2.EntityValues)
+ assert.Equal(t, original.Timestamp, unmarshaled2.Timestamp)
+ assert.Equal(t, original.DocID, unmarshaled2.DocID)
+ assert.Equal(t, original.Version, unmarshaled2.Version)
+ assert.Equal(t, len(original.Fields), len(unmarshaled2.Fields))
+
+ // Verify each field is preserved
+ for i, expectedField := range original.Fields {
+ actualField := unmarshaled2.Fields[i]
+ assert.Equal(t, expectedField.Key.Analyzer, actualField.Key.Analyzer)
+ assert.Equal(t, expectedField.Key.TagName, actualField.Key.TagName)
+ assert.Equal(t, expectedField.Key.SeriesID, actualField.Key.SeriesID)
+ assert.Equal(t, expectedField.Key.IndexRuleID, actualField.Key.IndexRuleID)
+ assert.Equal(t, expectedField.NoSort, actualField.NoSort)
+ assert.Equal(t, expectedField.Store, actualField.Store)
+ assert.Equal(t, expectedField.Index, actualField.Index)
+
+ // Verify term values
+ switch expected := expectedField.term.(type) {
+ case *BytesTermValue:
+ actual, ok := actualField.term.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := actualField.term.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ }
+}
+
+func TestDocuments_Marshal_Unmarshal(t *testing.T) {
+ tests := []struct {
+ name string
+ documents Documents
+ }{
+ {
+ name: "Empty documents collection",
+ documents: Documents{},
+ },
+ {
+ name: "Single document",
+ documents: Documents{
+ {
+ EntityValues: []byte("single-doc-entity"),
+ Timestamp: 1640995200000,
+ DocID: 12345,
+ Version: 1,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "keyword",
+ TagName: "service.name",
+ SeriesID: common.SeriesID(123456789),
+ IndexRuleID: 42,
+ },
+ term: &BytesTermValue{Value: []byte("test-service")},
+ NoSort: false,
+ Store: true,
+ Index: true,
+ },
+ },
+ },
+ },
+ },
+ {
+ name: "Multiple documents with different field types",
+ documents: Documents{
+ {
+ EntityValues: []byte("doc1-entity"),
+ Timestamp: 1640995200000,
+ DocID: 1,
+ Version: 1,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("2024-12-31")},
+ Lower: &BytesTermValue{Value: []byte("2024-01-01")},
+ IncludesUpper: true,
+ IncludesLower: false,
+ },
+ Analyzer: "keyword",
+ TagName: "service.name",
+ SeriesID: common.SeriesID(111),
+ IndexRuleID: 10,
+ },
+ term: &BytesTermValue{Value: []byte("service-1")},
+ NoSort: false,
+ Store: true,
+ Index: true,
+ },
+ },
+ },
+ {
+ EntityValues: []byte("doc2-entity"),
+ Timestamp: 1640995260000,
+ DocID: 2,
+ Version: 2,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &FloatTermValue{Value: 100.0},
+ Lower: &FloatTermValue{Value: 0.0},
+ IncludesUpper: true,
+ IncludesLower: true,
+ },
+ Analyzer: "standard",
+ TagName: "response.time",
+ SeriesID: common.SeriesID(222),
+ IndexRuleID: 20,
+ },
+ term: &FloatTermValue{Value: 45.67},
+ NoSort: true,
+ Store: false,
+ Index: true,
+ },
+ },
+ },
+ {
+ EntityValues: []byte("doc3-entity"),
+ Timestamp: 1640995320000,
+ DocID: 3,
+ Version: 3,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "simple",
+ TagName: "status",
+ SeriesID: common.SeriesID(333),
+ IndexRuleID: 30,
+ },
+ term: &BytesTermValue{Value: []byte("success")},
+ NoSort: false,
+ Store: true,
+ Index: false,
+ },
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "standard",
+ TagName: "error.count",
+ SeriesID: common.SeriesID(444),
+ IndexRuleID: 40,
+ },
+ term: &FloatTermValue{Value: 0.0},
+ NoSort: true,
+ Store: true,
+ Index: true,
+ },
+ },
+ },
+ },
+ },
+ {
+ name: "Documents with empty fields",
+ documents: Documents{
+ {
+ EntityValues: []byte{},
+ Timestamp: 0,
+ DocID: 0,
+ Version: 0,
+ Fields: []Field{},
+ },
+ {
+ EntityValues: []byte("empty-fields-entity"),
+ Timestamp: 1640995200000,
+ DocID: 999,
+ Version: 1,
+ Fields: []Field{},
+ },
+ },
+ },
+ {
+ name: "Large documents collection",
+ documents: Documents{
+ {
+ EntityValues: bytes.Repeat([]byte("large-entity-1-"), 50),
+ Timestamp: 1640995200000,
+ DocID: 1001,
+ Version: 1,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "keyword",
+ TagName: "large_field_1",
+ SeriesID: common.SeriesID(1001),
+ IndexRuleID: 1001,
+ },
+ term: &BytesTermValue{Value: bytes.Repeat([]byte("large-value-1-"), 25)},
+ NoSort: false,
+ Store: true,
+ Index: true,
+ },
+ },
+ },
+ {
+ EntityValues: bytes.Repeat([]byte("large-entity-2-"), 50),
+ Timestamp: 1640995260000,
+ DocID: 1002,
+ Version: 2,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "standard",
+ TagName: "large_field_2",
+ SeriesID: common.SeriesID(1002),
+ IndexRuleID: 1002,
+ },
+ term: &FloatTermValue{Value: 999999.999},
+ NoSort: true,
+ Store: false,
+ Index: true,
+ },
+ },
+ },
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Marshal
+ marshaled, err := tt.documents.Marshal()
+ require.NoError(t, err)
+ assert.NotNil(t, marshaled)
+
+ // For empty documents, the marshaled data should be minimal
+ if len(tt.documents) == 0 {
+ assert.Equal(t, 1, len(marshaled)) // Just the varint for count (0)
+ } else {
+ assert.Greater(t, len(marshaled), 0)
+ }
+
+ // Unmarshal
+ var unmarshaled Documents
+ err = unmarshaled.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Verify documents count
+ assert.Equal(t, len(tt.documents), len(unmarshaled))
+
+ // Verify each document
+ for i, expectedDoc := range tt.documents {
+ require.Less(t, i, len(unmarshaled))
+ actualDoc := unmarshaled[i]
+
+ // Verify basic fields
+ assert.Equal(t, expectedDoc.EntityValues, actualDoc.EntityValues)
+ assert.Equal(t, expectedDoc.Timestamp, actualDoc.Timestamp)
+ assert.Equal(t, expectedDoc.DocID, actualDoc.DocID)
+ assert.Equal(t, expectedDoc.Version, actualDoc.Version)
+
+ // Verify fields count
+ assert.Equal(t, len(expectedDoc.Fields), len(actualDoc.Fields))
+
+ // Verify each field
+ for j, expectedField := range expectedDoc.Fields {
+ require.Less(t, j, len(actualDoc.Fields))
+ actualField := actualDoc.Fields[j]
+
+ // Check FieldKey
+ assert.Equal(t, expectedField.Key.Analyzer, actualField.Key.Analyzer)
+ assert.Equal(t, expectedField.Key.TagName, actualField.Key.TagName)
+ assert.Equal(t, expectedField.Key.SeriesID, actualField.Key.SeriesID)
+ assert.Equal(t, expectedField.Key.IndexRuleID, actualField.Key.IndexRuleID)
+
+ // Check TimeRange
+ if expectedField.Key.TimeRange != nil {
+ assert.NotNil(t, actualField.Key.TimeRange)
+ assert.Equal(t, expectedField.Key.TimeRange.IncludesUpper, actualField.Key.TimeRange.IncludesUpper)
+ assert.Equal(t, expectedField.Key.TimeRange.IncludesLower, actualField.Key.TimeRange.IncludesLower)
+
+ // Compare Upper bound
+ switch expected := expectedField.Key.TimeRange.Upper.(type) {
+ case *BytesTermValue:
+ actual, ok := actualField.Key.TimeRange.Upper.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := actualField.Key.TimeRange.Upper.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+
+ // Compare Lower bound
+ switch expected := expectedField.Key.TimeRange.Lower.(type) {
+ case *BytesTermValue:
+ actual, ok := actualField.Key.TimeRange.Lower.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := actualField.Key.TimeRange.Lower.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ } else {
+ assert.Nil(t, actualField.Key.TimeRange)
+ }
+
+ // Check term value
+ switch expected := expectedField.term.(type) {
+ case *BytesTermValue:
+ actual, ok := actualField.term.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := actualField.term.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+
+ // Check flags
+ assert.Equal(t, expectedField.NoSort, actualField.NoSort)
+ assert.Equal(t, expectedField.Store, actualField.Store)
+ assert.Equal(t, expectedField.Index, actualField.Index)
+ }
+ }
+ })
+ }
+}
+
+func TestDocuments_Unmarshal_InvalidData(t *testing.T) {
+ tests := []struct {
+ name string
+ expectedErr string
+ data []byte
+ }{
+ {
+ name: "insufficient data for documents",
+ data: []byte{0x02}, // Claim 2 documents but no actual data
+ expectedErr: "insufficient data for documents",
+ },
+ {
+ name: "invalid document data",
+ data: []byte{0x01, 0x01, 0x00}, // 1 document, but invalid document data
+ expectedErr: "failed to unmarshal document",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var documents Documents
+ err := documents.Unmarshal(tt.data)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tt.expectedErr)
+ })
+ }
+}
+
+func TestDocuments_Marshal_Unmarshal_RoundTrip(t *testing.T) {
+ // Test round-trip marshaling/unmarshaling with a complex documents collection
+ original := Documents{
+ {
+ EntityValues: []byte("round-trip-doc1-entity"),
+ Timestamp: 1640995200000,
+ DocID: 987654321,
+ Version: 42,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("2024-12-31T23:59:59Z")},
+ Lower: &FloatTermValue{Value: 1672531200.0},
+ IncludesUpper: true,
+ IncludesLower: false,
+ },
+ Analyzer: "standard",
+ TagName: "service.instance.name",
+ SeriesID: common.SeriesID(1234567890123456789),
+ IndexRuleID: 999,
+ },
+ term: &BytesTermValue{Value: []byte("test-instance")},
+ NoSort: true,
+ Store: true,
+ Index: false,
+ },
+ },
+ },
+ {
+ EntityValues: []byte("round-trip-doc2-entity"),
+ Timestamp: 1640995260000,
+ DocID: 987654322,
+ Version: 43,
+ Fields: []Field{
+ {
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "keyword",
+ TagName: "status",
+ SeriesID: common.SeriesID(9876543210987654321),
+ IndexRuleID: 888,
+ },
+ term: &FloatTermValue{Value: 200.0},
+ NoSort: false,
+ Store: false,
+ Index: true,
+ },
+ },
+ },
+ }
+
+ // First round-trip
+ marshaled1, err := original.Marshal()
+ require.NoError(t, err)
+
+ var unmarshaled1 Documents
+ err = unmarshaled1.Unmarshal(marshaled1)
+ require.NoError(t, err)
+
+ // Second round-trip
+ marshaled2, err := unmarshaled1.Marshal()
+ require.NoError(t, err)
+
+ var unmarshaled2 Documents
+ err = unmarshaled2.Unmarshal(marshaled2)
+ require.NoError(t, err)
+
+ // Verify all documents are preserved through multiple round-trips
+ assert.Equal(t, len(original), len(unmarshaled2))
+
+ // Verify each document is preserved
+ for i, expectedDoc := range original {
+ actualDoc := unmarshaled2[i]
+ assert.Equal(t, expectedDoc.EntityValues, actualDoc.EntityValues)
+ assert.Equal(t, expectedDoc.Timestamp, actualDoc.Timestamp)
+ assert.Equal(t, expectedDoc.DocID, actualDoc.DocID)
+ assert.Equal(t, expectedDoc.Version, actualDoc.Version)
+ assert.Equal(t, len(expectedDoc.Fields), len(actualDoc.Fields))
+
+ // Verify each field is preserved
+ for j, expectedField := range expectedDoc.Fields {
+ actualField := actualDoc.Fields[j]
+ assert.Equal(t, expectedField.Key.Analyzer, actualField.Key.Analyzer)
+ assert.Equal(t, expectedField.Key.TagName, actualField.Key.TagName)
+ assert.Equal(t, expectedField.Key.SeriesID, actualField.Key.SeriesID)
+ assert.Equal(t, expectedField.Key.IndexRuleID, actualField.Key.IndexRuleID)
+ assert.Equal(t, expectedField.NoSort, actualField.NoSort)
+ assert.Equal(t, expectedField.Store, actualField.Store)
+ assert.Equal(t, expectedField.Index, actualField.Index)
+
+ // Verify term values
+ switch expected := expectedField.term.(type) {
+ case *BytesTermValue:
+ actual, ok := actualField.term.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := actualField.term.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ }
+ }
+}
diff --git a/pkg/index/field_key_test.go b/pkg/index/field_key_test.go
new file mode 100644
index 00000000..da789afb
--- /dev/null
+++ b/pkg/index/field_key_test.go
@@ -0,0 +1,283 @@
+// 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 index
+
+import (
+ "math"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestFieldKey_MarshalAll_UnmarshalAll(t *testing.T) {
+ tests := []struct {
+ name string
+ original FieldKey
+ }{
+ {
+ name: "complete field key with time range",
+ original: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("2024-01-01")},
+ Lower: &BytesTermValue{Value: []byte("2023-01-01")},
+ IncludesUpper: true,
+ IncludesLower: false,
+ },
+ Analyzer: "standard",
+ TagName: "service_name",
+ SeriesID: 123456789,
+ IndexRuleID: 42,
+ },
+ },
+ {
+ name: "field key with float time range",
+ original: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &FloatTermValue{Value: 1704067200.0}, // 2024-01-01 timestamp
+ Lower: &FloatTermValue{Value: 1672531200.0}, // 2023-01-01 timestamp
+ IncludesUpper: false,
+ IncludesLower: true,
+ },
+ Analyzer: "keyword",
+ TagName: "instance_id",
+ SeriesID: 987654321,
+ IndexRuleID: 100,
+ },
+ },
+ {
+ name: "field key without time range",
+ original: FieldKey{
+ TimeRange: nil,
+ Analyzer: "simple",
+ TagName: "endpoint",
+ SeriesID: 555666777,
+ IndexRuleID: 200,
+ },
+ },
+ {
+ name: "field key with empty analyzer and tag name",
+ original: FieldKey{
+ TimeRange: nil,
+ Analyzer: "",
+ TagName: "",
+ SeriesID: 111222333,
+ IndexRuleID: 300,
+ },
+ },
+ {
+ name: "field key with unicode tag name",
+ original: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("测试上限")},
+ Lower: &BytesTermValue{Value: []byte("测试下限")},
+ IncludesUpper: true,
+ IncludesLower: true,
+ },
+ Analyzer: "url",
+ TagName: "测试标签",
+ SeriesID: 444555666,
+ IndexRuleID: 400,
+ },
+ },
+ {
+ name: "field key with maximum values",
+ original: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &FloatTermValue{Value: math.MaxFloat64},
+ Lower: &FloatTermValue{Value: -math.MaxFloat64},
+ IncludesUpper: true,
+ IncludesLower: true,
+ },
+ Analyzer: "standard",
+ TagName: "max_tag",
+ SeriesID: math.MaxUint64,
+ IndexRuleID: math.MaxUint32,
+ },
+ },
+ {
+ name: "field key with zero values",
+ original: FieldKey{
+ TimeRange: nil,
+ Analyzer: "",
+ TagName: "",
+ SeriesID: 0,
+ IndexRuleID: 0,
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Marshal
+ marshaled, err := tt.original.MarshalAll()
+ require.NoError(t, err)
+ assert.NotNil(t, marshaled)
+ assert.Greater(t, len(marshaled), 0)
+
+ // Unmarshal
+ unmarshaled := &FieldKey{}
+ err = unmarshaled.UnmarshalAll(marshaled)
+ require.NoError(t, err)
+
+ // Verify TimeRange
+ if tt.original.TimeRange != nil {
+ assert.NotNil(t, unmarshaled.TimeRange)
+ assert.Equal(t, tt.original.TimeRange.IncludesUpper, unmarshaled.TimeRange.IncludesUpper)
+ assert.Equal(t, tt.original.TimeRange.IncludesLower, unmarshaled.TimeRange.IncludesLower)
+
+ // Verify Upper bound
+ if tt.original.TimeRange.Upper != nil {
+ assert.NotNil(t, unmarshaled.TimeRange.Upper)
+ switch expected := tt.original.TimeRange.Upper.(type) {
+ case *BytesTermValue:
+ actual, ok := unmarshaled.TimeRange.Upper.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := unmarshaled.TimeRange.Upper.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ } else {
+ assert.Nil(t, unmarshaled.TimeRange.Upper)
+ }
+
+ // Verify Lower bound
+ if tt.original.TimeRange.Lower != nil {
+ assert.NotNil(t, unmarshaled.TimeRange.Lower)
+ switch expected := tt.original.TimeRange.Lower.(type) {
+ case *BytesTermValue:
+ actual, ok := unmarshaled.TimeRange.Lower.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := unmarshaled.TimeRange.Lower.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ } else {
+ assert.Nil(t, unmarshaled.TimeRange.Lower)
+ }
+ } else {
+ assert.Nil(t, unmarshaled.TimeRange)
+ }
+
+ // Verify other fields
+ assert.Equal(t, tt.original.Analyzer, unmarshaled.Analyzer)
+ assert.Equal(t, tt.original.TagName, unmarshaled.TagName)
+ assert.Equal(t, tt.original.SeriesID, unmarshaled.SeriesID)
+ assert.Equal(t, tt.original.IndexRuleID, unmarshaled.IndexRuleID)
+ })
+ }
+}
+
+func TestFieldKey_UnmarshalAll_InvalidData(t *testing.T) {
+ tests := []struct {
+ name string
+ expectedErr string
+ data []byte
+ }{
+ {
+ name: "empty data",
+ data: []byte{},
+ expectedErr: "empty data for field key",
+ },
+ {
+ name: "insufficient data for time range flag",
+ data: []byte{},
+ expectedErr: "empty data for field key",
+ },
+ {
+ name: "insufficient data for series ID",
+ data: []byte{0, 0, 0, 0, 0, 0, 0}, // only 7 bytes for series ID
+ expectedErr: "insufficient data for series ID",
+ },
+ {
+ name: "insufficient data for IndexRuleID",
+ data: []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, // only 8 bytes, missing IndexRuleID
+ expectedErr: "insufficient data for IndexRuleID",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ unmarshaled := &FieldKey{}
+ err := unmarshaled.UnmarshalAll(tt.data)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tt.expectedErr)
+ })
+ }
+}
+
+func TestFieldKey_MarshalAll_UnmarshalAll_RoundTrip(t *testing.T) {
+ // Test round-trip marshaling/unmarshaling with a complex field key
+ original := FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("2024-12-31T23:59:59Z")},
+ Lower: &FloatTermValue{Value: 1672531200.0},
+ IncludesUpper: true,
+ IncludesLower: false,
+ },
+ Analyzer: "standard",
+ TagName: "service.instance.name",
+ SeriesID: 1234567890123456789,
+ IndexRuleID: 999,
+ }
+
+ // First round-trip
+ marshaled1, err := original.MarshalAll()
+ require.NoError(t, err)
+
+ unmarshaled1 := &FieldKey{}
+ err = unmarshaled1.UnmarshalAll(marshaled1)
+ require.NoError(t, err)
+
+ // Second round-trip
+ marshaled2, err := unmarshaled1.MarshalAll()
+ require.NoError(t, err)
+
+ unmarshaled2 := &FieldKey{}
+ err = unmarshaled2.UnmarshalAll(marshaled2)
+ require.NoError(t, err)
+
+ // Verify all fields are preserved through multiple round-trips
+ assert.Equal(t, original.Analyzer, unmarshaled2.Analyzer)
+ assert.Equal(t, original.TagName, unmarshaled2.TagName)
+ assert.Equal(t, original.SeriesID, unmarshaled2.SeriesID)
+ assert.Equal(t, original.IndexRuleID, unmarshaled2.IndexRuleID)
+
+ // Verify TimeRange is preserved
+ assert.NotNil(t, unmarshaled2.TimeRange)
+ assert.Equal(t, original.TimeRange.IncludesUpper, unmarshaled2.TimeRange.IncludesUpper)
+ assert.Equal(t, original.TimeRange.IncludesLower, unmarshaled2.TimeRange.IncludesLower)
+
+ // Verify Upper bound
+ originalUpper, ok := original.TimeRange.Upper.(*BytesTermValue)
+ require.True(t, ok)
+ unmarshaledUpper, ok := unmarshaled2.TimeRange.Upper.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, originalUpper.Value, unmarshaledUpper.Value)
+
+ // Verify Lower bound
+ originalLower, ok := original.TimeRange.Lower.(*FloatTermValue)
+ require.True(t, ok)
+ unmarshaledLower, ok := unmarshaled2.TimeRange.Lower.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, originalLower.Value, unmarshaledLower.Value)
+}
diff --git a/pkg/index/field_test.go b/pkg/index/field_test.go
new file mode 100644
index 00000000..467ae287
--- /dev/null
+++ b/pkg/index/field_test.go
@@ -0,0 +1,158 @@
+// 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 index
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+)
+
+func TestField_Marshal_Unmarshal(t *testing.T) {
+ tests := []struct {
+ name string
+ field Field
+ }{
+ {
+ name: "BytesTermValue, all flags true",
+ field: Field{
+ Key: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("upper")},
+ Lower: &BytesTermValue{Value: []byte("lower")},
+ IncludesUpper: true,
+ IncludesLower: false,
+ },
+ Analyzer: "keyword",
+ TagName: "tag1",
+ SeriesID: common.SeriesID(12345),
+ IndexRuleID: 42,
+ },
+ term: &BytesTermValue{Value: []byte("test-value")},
+ NoSort: true,
+ Store: true,
+ Index: true,
+ },
+ },
+ {
+ name: "FloatTermValue, all flags false",
+ field: Field{
+ Key: FieldKey{
+ TimeRange: &RangeOpts{
+ Upper: &FloatTermValue{Value: 100.5},
+ Lower: &FloatTermValue{Value: -100.5},
+ IncludesUpper: false,
+ IncludesLower: true,
+ },
+ Analyzer: "standard",
+ TagName: "tag2",
+ SeriesID: common.SeriesID(67890),
+ IndexRuleID: 99,
+ },
+ term: &FloatTermValue{Value: 3.14159},
+ NoSort: false,
+ Store: false,
+ Index: false,
+ },
+ },
+ {
+ name: "Minimal field (no TimeRange, empty analyzer/tag)",
+ field: Field{
+ Key: FieldKey{
+ TimeRange: nil,
+ Analyzer: "",
+ TagName: "",
+ SeriesID: 0,
+ IndexRuleID: 0,
+ },
+ term: &BytesTermValue{Value: []byte{}},
+ NoSort: false,
+ Store: true,
+ Index: false,
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ marshaled, err := tt.field.Marshal()
+ require.NoError(t, err)
+ assert.NotNil(t, marshaled)
+ assert.Greater(t, len(marshaled), 0)
+
+ var unmarshaled Field
+ err = unmarshaled.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Check FieldKey
+ assert.Equal(t, tt.field.Key.Analyzer, unmarshaled.Key.Analyzer)
+ assert.Equal(t, tt.field.Key.TagName, unmarshaled.Key.TagName)
+ assert.Equal(t, tt.field.Key.SeriesID, unmarshaled.Key.SeriesID)
+ assert.Equal(t, tt.field.Key.IndexRuleID, unmarshaled.Key.IndexRuleID)
+ if tt.field.Key.TimeRange != nil {
+ assert.NotNil(t, unmarshaled.Key.TimeRange)
+ assert.Equal(t, tt.field.Key.TimeRange.IncludesUpper, unmarshaled.Key.TimeRange.IncludesUpper)
+ assert.Equal(t, tt.field.Key.TimeRange.IncludesLower, unmarshaled.Key.TimeRange.IncludesLower)
+ // Compare Upper
+ switch expected := tt.field.Key.TimeRange.Upper.(type) {
+ case *BytesTermValue:
+ actual, ok := unmarshaled.Key.TimeRange.Upper.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := unmarshaled.Key.TimeRange.Upper.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ // Compare Lower
+ switch expected := tt.field.Key.TimeRange.Lower.(type) {
+ case *BytesTermValue:
+ actual, ok := unmarshaled.Key.TimeRange.Lower.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := unmarshaled.Key.TimeRange.Lower.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ } else {
+ assert.Nil(t, unmarshaled.Key.TimeRange)
+ }
+
+ // Check term value
+ switch expected := tt.field.term.(type) {
+ case *BytesTermValue:
+ actual, ok := unmarshaled.term.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := unmarshaled.term.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+
+ // Check flags
+ assert.Equal(t, tt.field.NoSort, unmarshaled.NoSort)
+ assert.Equal(t, tt.field.Store, unmarshaled.Store)
+ assert.Equal(t, tt.field.Index, unmarshaled.Index)
+ })
+ }
+}
diff --git a/pkg/index/index.go b/pkg/index/index.go
index 9568bba3..fbf69095 100644
--- a/pkg/index/index.go
+++ b/pkg/index/index.go
@@ -32,6 +32,7 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/index/posting"
"github.com/apache/skywalking-banyandb/pkg/iter/sort"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
@@ -69,11 +70,11 @@ type FieldKey struct {
}
// Marshal encodes f to string.
-func (f FieldKey) Marshal() string {
- if len(f.TagName) > 0 {
- return f.TagName
+func (fk FieldKey) Marshal() string {
+ if len(fk.TagName) > 0 {
+ return fk.TagName
}
- return string(convert.Uint32ToBytes(f.IndexRuleID))
+ return string(convert.Uint32ToBytes(fk.IndexRuleID))
}
// NewStringField creates a new string field.
@@ -144,6 +145,8 @@ func (f *Field) MarshalJSON() ([]byte, error) {
type IsTermValue interface {
isTermValue()
String() string
+ Marshal() ([]byte, error)
+ Unmarshal(data []byte) error
}
// BytesTermValue represents a byte term value.
@@ -153,8 +156,37 @@ type BytesTermValue struct {
func (BytesTermValue) isTermValue() {}
-func (b BytesTermValue) String() string {
- return convert.BytesToString(b.Value)
+func (bv BytesTermValue) String() string {
+ return convert.BytesToString(bv.Value)
+}
+
+// Marshal encodes BytesTermValue to bytes.
+func (bv BytesTermValue) Marshal() ([]byte, error) {
+ var result []byte
+ result = append(result, 0) // type: bytes
+ result = encoding.EncodeBytes(result, bv.Value)
+ return result, nil
+}
+
+// Unmarshal decodes BytesTermValue from bytes.
+func (bv *BytesTermValue) Unmarshal(data []byte) error {
+ if len(data) < 1 {
+ return fmt.Errorf("insufficient data for term value type")
+ }
+
+ if data[0] != 0 {
+ return fmt.Errorf("invalid term value type, expected bytes (0), got %d", data[0])
+ }
+
+ var valueBytes []byte
+ var err error
+ _, valueBytes, err = encoding.DecodeBytes(data[1:])
+ if err != nil {
+ return fmt.Errorf("failed to decode bytes value: %w", err)
+ }
+
+ bv.Value = valueBytes
+ return nil
}
// FloatTermValue represents a float term value.
@@ -164,8 +196,41 @@ type FloatTermValue struct {
func (FloatTermValue) isTermValue() {}
-func (f FloatTermValue) String() string {
- return strconv.FormatInt(numeric.Float64ToInt64(f.Value), 10)
+func (fv FloatTermValue) String() string {
+ return strconv.FormatInt(numeric.Float64ToInt64(fv.Value), 10)
+}
+
+// Marshal encodes FloatTermValue to bytes.
+func (fv FloatTermValue) Marshal() ([]byte, error) {
+ var result []byte
+ result = append(result, 1) // type: float
+ result = encoding.EncodeBytes(result, convert.Float64ToBytes(fv.Value))
+ return result, nil
+}
+
+// Unmarshal decodes FloatTermValue from bytes.
+func (fv *FloatTermValue) Unmarshal(data []byte) error {
+ if len(data) < 1 {
+ return fmt.Errorf("insufficient data for term value type")
+ }
+
+ if data[0] != 1 {
+ return fmt.Errorf("invalid term value type, expected float (1), got %d", data[0])
+ }
+
+ var valueBytes []byte
+ var err error
+ _, valueBytes, err = encoding.DecodeBytes(data[1:])
+ if err != nil {
+ return fmt.Errorf("failed to decode float value: %w", err)
+ }
+
+ if len(valueBytes) != 8 {
+ return fmt.Errorf("invalid float value length: %d", len(valueBytes))
+ }
+
+ fv.Value = convert.BytesToFloat64(valueBytes)
+ return nil
}
// RangeOpts contains options to performance a continuous scan.
@@ -201,6 +266,123 @@ func (r RangeOpts) Valid() bool {
return true
}
+// Marshal encodes RangeOpts to bytes.
+func (r RangeOpts) Marshal() ([]byte, error) {
+ var result []byte
+
+ // Marshal Upper
+ if r.Upper != nil {
+ result = append(result, 1) // has upper
+ upperBytes, err := r.Upper.Marshal()
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal upper: %w", err)
+ }
+ result = encoding.EncodeBytes(result, upperBytes)
+ } else {
+ result = append(result, 0) // no upper
+ }
+
+ // Marshal Lower
+ if r.Lower != nil {
+ result = append(result, 1) // has lower
+ lowerBytes, err := r.Lower.Marshal()
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal lower: %w", err)
+ }
+ result = encoding.EncodeBytes(result, lowerBytes)
+ } else {
+ result = append(result, 0) // no lower
+ }
+
+ // Marshal boolean flags
+ flags := byte(0)
+ if r.IncludesUpper {
+ flags |= 1
+ }
+ if r.IncludesLower {
+ flags |= 2
+ }
+ result = append(result, flags)
+
+ return result, nil
+}
+
+// Unmarshal decodes RangeOpts from bytes.
+func (r *RangeOpts) Unmarshal(data []byte) error {
+ if len(data) < 3 {
+ return fmt.Errorf("insufficient data for range options")
+ }
+
+ // Unmarshal Upper
+ hasUpper := data[0] == 1
+ tail := data[1:]
+
+ if hasUpper {
+ var upperBytes []byte
+ var err error
+ tail, upperBytes, err = encoding.DecodeBytes(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode upper: %w", err)
+ }
+ // Create the appropriate type based on the first byte
+ if len(upperBytes) > 0 {
+ switch upperBytes[0] {
+ case 0:
+ r.Upper = &BytesTermValue{}
+ case 1:
+ r.Upper = &FloatTermValue{}
+ default:
+ return fmt.Errorf("unknown term value type for upper: %d", upperBytes[0])
+ }
+ } else {
+ return fmt.Errorf("empty upperBytes for upper term")
+ }
+ if err := r.Upper.Unmarshal(upperBytes); err != nil {
+ return fmt.Errorf("failed to unmarshal upper: %w", err)
+ }
+ } else {
+ r.Upper = nil
+ }
+
+ // Unmarshal Lower
+ hasLower := tail[0] == 1
+ tail = tail[1:]
+
+ if hasLower {
+ var lowerBytes []byte
+ var err error
+ tail, lowerBytes, err = encoding.DecodeBytes(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode lower: %w", err)
+ }
+ // Create the appropriate type based on the first byte
+ if len(lowerBytes) > 0 {
+ switch lowerBytes[0] {
+ case 0:
+ r.Lower = &BytesTermValue{}
+ case 1:
+ r.Lower = &FloatTermValue{}
+ default:
+ return fmt.Errorf("unknown term value type for lower: %d", lowerBytes[0])
+ }
+ } else {
+ return fmt.Errorf("empty lowerBytes for lower term")
+ }
+ if err := r.Lower.Unmarshal(lowerBytes); err != nil {
+ return fmt.Errorf("failed to unmarshal lower: %w", err)
+ }
+ } else {
+ r.Lower = nil
+ }
+
+ // Unmarshal boolean flags
+ flags := tail[0]
+ r.IncludesUpper = (flags & 1) != 0
+ r.IncludesLower = (flags & 2) != 0
+
+ return nil
+}
+
// NewStringRangeOpts creates a new string range option.
func NewStringRangeOpts(lower, upper string, includesLower, includesUpper bool) RangeOpts {
var upperBytes, lowerBytes []byte
@@ -215,8 +397,8 @@ func NewStringRangeOpts(lower, upper string, includesLower, includesUpper bool)
lowerBytes = convert.StringToBytes(lower)
}
return RangeOpts{
- Lower: &BytesTermValue{Value: upperBytes},
- Upper: &BytesTermValue{Value: lowerBytes},
+ Lower: &BytesTermValue{Value: lowerBytes},
+ Upper: &BytesTermValue{Value: upperBytes},
IncludesLower: includesLower,
IncludesUpper: includesUpper,
}
@@ -305,6 +487,301 @@ type Document struct {
// Documents is a collection of documents.
type Documents []Document
+// Marshal encodes Documents to bytes.
+func (docs Documents) Marshal() ([]byte, error) {
+ var result []byte
+
+ // Write the number of documents
+ result = encoding.VarUint64ToBytes(result, uint64(len(docs)))
+
+ // Marshal each document
+ for _, doc := range docs {
+ docBytes, err := doc.Marshal()
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal document: %w", err)
+ }
+ result = encoding.EncodeBytes(result, docBytes)
+ }
+
+ return result, nil
+}
+
+// Unmarshal decodes Documents from bytes.
+func (docs *Documents) Unmarshal(data []byte) error {
+ if len(data) == 0 {
+ *docs = nil
+ return nil
+ }
+
+ // Read the number of documents
+ tail, docCount := encoding.BytesToVarUint64(data)
+ if uint64(len(tail)) < docCount {
+ return fmt.Errorf("insufficient data for documents: need %d documents, have %d bytes", docCount, len(tail))
+ }
+
+ // Initialize the documents slice
+ *docs = make(Documents, 0, docCount)
+
+ // Unmarshal each document
+ for i := uint64(0); i < docCount; i++ {
+ var docBytes []byte
+ var err error
+ tail, docBytes, err = encoding.DecodeBytes(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode document %d: %w", i, err)
+ }
+
+ var doc Document
+ if err := doc.Unmarshal(docBytes); err != nil {
+ return fmt.Errorf("failed to unmarshal document %d: %w", i, err)
+ }
+
+ *docs = append(*docs, doc)
+ }
+
+ return nil
+}
+
+// Marshal encodes Document to bytes.
+func (doc Document) Marshal() ([]byte, error) {
+ var result []byte
+
+ // Marshal EntityValues
+ result = encoding.EncodeBytes(result, doc.EntityValues)
+
+ // Marshal Timestamp
+ result = encoding.VarInt64ToBytes(result, doc.Timestamp)
+
+ // Marshal DocID
+ result = encoding.VarUint64ToBytes(result, doc.DocID)
+
+ // Marshal Version
+ result = encoding.VarInt64ToBytes(result, doc.Version)
+
+ // Marshal Fields
+ result = encoding.VarUint64ToBytes(result, uint64(len(doc.Fields)))
+ for _, field := range doc.Fields {
+ fieldBytes, err := field.Marshal()
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal field: %w", err)
+ }
+ result = encoding.EncodeBytes(result, fieldBytes)
+ }
+
+ return result, nil
+}
+
+// Unmarshal decodes Document from bytes.
+func (doc *Document) Unmarshal(data []byte) error {
+ if len(data) == 0 {
+ return fmt.Errorf("empty data for document")
+ }
+ var (
+ tail []byte
+ entityValues []byte
+ err error
+ )
+ tail, entityValues, err = encoding.DecodeBytes(data)
+ if err != nil {
+ return fmt.Errorf("failed to decode entity values: %w", err)
+ }
+ doc.EntityValues = entityValues
+ var timestamp int64
+ tail, timestamp, err = encoding.BytesToVarInt64(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode timestamp: %w", err)
+ }
+ doc.Timestamp = timestamp
+ tail, doc.DocID = encoding.BytesToVarUint64(tail)
+ var version int64
+ tail, version, err = encoding.BytesToVarInt64(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode version: %w", err)
+ }
+ doc.Version = version
+ tail, fieldCount := encoding.BytesToVarUint64(tail)
+ doc.Fields = make([]Field, 0, fieldCount)
+ for i := uint64(0); i < fieldCount; i++ {
+ var fieldBytes []byte
+ tail, fieldBytes, err = encoding.DecodeBytes(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode field %d: %w", i, err)
+ }
+ var field Field
+ if err := field.Unmarshal(fieldBytes); err != nil {
+ return fmt.Errorf("failed to unmarshal field %d: %w", i, err)
+ }
+ doc.Fields = append(doc.Fields, field)
+ }
+ return nil
+}
+
+// Marshal encodes Field to bytes.
+func (f Field) Marshal() ([]byte, error) {
+ var result []byte
+ // Marshal Key
+ keyBytes, err := f.Key.MarshalAll()
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal field key: %w", err)
+ }
+ result = encoding.EncodeBytes(result, keyBytes)
+ // Marshal term
+ termBytes, err := f.term.Marshal()
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal field term: %w", err)
+ }
+ result = encoding.EncodeBytes(result, termBytes)
+ // Marshal boolean flags
+ flags := byte(0)
+ if f.NoSort {
+ flags |= 1
+ }
+ if f.Store {
+ flags |= 2
+ }
+ if f.Index {
+ flags |= 4
+ }
+ result = append(result, flags)
+ return result, nil
+}
+
+// Unmarshal decodes Field from bytes.
+func (f *Field) Unmarshal(data []byte) error {
+ if len(data) == 0 {
+ return fmt.Errorf("empty data for field")
+ }
+ var (
+ tail []byte
+ keyBytes []byte
+ err error
+ )
+ tail, keyBytes, err = encoding.DecodeBytes(data)
+ if err != nil {
+ return fmt.Errorf("failed to decode field key: %w", err)
+ }
+ if unmarshalErr := f.Key.UnmarshalAll(keyBytes); unmarshalErr != nil {
+ return fmt.Errorf("failed to unmarshal field key: %w", unmarshalErr)
+ }
+ var termBytes []byte
+ tail, termBytes, err = encoding.DecodeBytes(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode field term: %w", err)
+ }
+ if f.term == nil {
+ // Try to detect type from first byte
+ if len(termBytes) > 0 {
+ switch termBytes[0] {
+ case 0:
+ f.term = &BytesTermValue{}
+ case 1:
+ f.term = &FloatTermValue{}
+ default:
+ return fmt.Errorf("unknown term value type: %d", termBytes[0])
+ }
+ } else {
+ return fmt.Errorf("empty termBytes for field term")
+ }
+ }
+ if err := f.term.Unmarshal(termBytes); err != nil {
+ return fmt.Errorf("failed to unmarshal field term: %w", err)
+ }
+ if len(tail) < 1 {
+ return fmt.Errorf("insufficient data for field flags")
+ }
+ flags := tail[0]
+ f.NoSort = (flags & 1) != 0
+ f.Store = (flags & 2) != 0
+ f.Index = (flags & 4) != 0
+ return nil
+}
+
+// MarshalAll encodes FieldKey to bytes.
+func (fk FieldKey) MarshalAll() ([]byte, error) {
+ var result []byte
+
+ // Marshal TimeRange
+ if fk.TimeRange != nil {
+ result = append(result, 1) // has time range
+ timeRangeBytes, err := fk.TimeRange.Marshal()
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal time range: %w", err)
+ }
+ result = encoding.EncodeBytes(result, timeRangeBytes)
+ } else {
+ result = append(result, 0) // no time range
+ }
+
+ // Marshal Analyzer
+ result = encoding.EncodeBytes(result, convert.StringToBytes(fk.Analyzer))
+
+ // Marshal TagName
+ result = encoding.EncodeBytes(result, convert.StringToBytes(fk.TagName))
+
+ // Marshal SeriesID
+ result = fk.SeriesID.AppendToBytes(result)
+
+ // Marshal IndexRuleID
+ result = encoding.Uint32ToBytes(result, fk.IndexRuleID)
+
+ return result, nil
+}
+
+// UnmarshalAll decodes FieldKey from bytes.
+func (fk *FieldKey) UnmarshalAll(data []byte) error {
+ if len(data) == 0 {
+ return fmt.Errorf("empty data for field key")
+ }
+ // Unmarshal TimeRange
+ if len(data) < 1 {
+ return fmt.Errorf("insufficient data for time range flag")
+ }
+ hasTimeRange := data[0] == 1
+ tail := data[1:]
+ if hasTimeRange {
+ var timeRangeBytes []byte
+ var err error
+ tail, timeRangeBytes, err = encoding.DecodeBytes(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode time range: %w", err)
+ }
+ fk.TimeRange = &RangeOpts{}
+ if err := fk.TimeRange.Unmarshal(timeRangeBytes); err != nil {
+ return fmt.Errorf("failed to unmarshal time range: %w", err)
+ }
+ } else {
+ fk.TimeRange = nil
+ }
+ // Unmarshal Analyzer
+ var analyzerBytes []byte
+ var err error
+ tail, analyzerBytes, err = encoding.DecodeBytes(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode analyzer: %w", err)
+ }
+ fk.Analyzer = convert.BytesToString(analyzerBytes)
+ // Unmarshal TagName
+ var tagNameBytes []byte
+ tail, tagNameBytes, err = encoding.DecodeBytes(tail)
+ if err != nil {
+ return fmt.Errorf("failed to decode tag name: %w", err)
+ }
+ fk.TagName = convert.BytesToString(tagNameBytes)
+ // Unmarshal SeriesID
+ if len(tail) < 8 {
+ return fmt.Errorf("insufficient data for series ID")
+ }
+ fk.SeriesID = common.SeriesID(encoding.BytesToUint64(tail))
+ tail = tail[8:]
+ // Unmarshal IndexRuleID (fixed 4 bytes)
+ if len(tail) < 4 {
+ return fmt.Errorf("insufficient data for IndexRuleID")
+ }
+ fk.IndexRuleID = encoding.BytesToUint32(tail)
+ // No need to advance tail further as this is the last field
+ return nil
+}
+
// Batch is a collection of documents.
type Batch struct {
PersistentCallback func(error)
diff --git a/pkg/index/range_opts_test.go b/pkg/index/range_opts_test.go
new file mode 100644
index 00000000..61afb0c5
--- /dev/null
+++ b/pkg/index/range_opts_test.go
@@ -0,0 +1,142 @@
+// 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 index
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestRangeOpts_Marshal_Unmarshal(t *testing.T) {
+ tests := []struct {
+ expectedUpper IsTermValue
+ expectedLower IsTermValue
+ original RangeOpts
+ name string
+ includesUpper bool
+ includesLower bool
+ }{
+ {
+ name: "string range with both bounds",
+ original: RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("zebra")},
+ Lower: &BytesTermValue{Value: []byte("apple")},
+ IncludesUpper: true,
+ IncludesLower: false,
+ },
+ expectedUpper: &BytesTermValue{Value: []byte("zebra")},
+ expectedLower: &BytesTermValue{Value: []byte("apple")},
+ includesUpper: true,
+ includesLower: false,
+ },
+ {
+ name: "int range with both bounds",
+ original: RangeOpts{
+ Upper: &FloatTermValue{Value: 100.0},
+ Lower: &FloatTermValue{Value: 1.0},
+ IncludesUpper: false,
+ IncludesLower: true,
+ },
+ expectedUpper: &FloatTermValue{Value: 100.0},
+ expectedLower: &FloatTermValue{Value: 1.0},
+ includesUpper: false,
+ includesLower: true,
+ },
+ {
+ name: "string range with both bounds inclusive",
+ original: RangeOpts{
+ Upper: &BytesTermValue{Value: []byte("zebra")},
+ Lower: &BytesTermValue{Value: []byte("apple")},
+ IncludesUpper: true,
+ IncludesLower: true,
+ },
+ expectedUpper: &BytesTermValue{Value: []byte("zebra")},
+ expectedLower: &BytesTermValue{Value: []byte("apple")},
+ includesUpper: true,
+ includesLower: true,
+ },
+ {
+ name: "int range with both bounds exclusive",
+ original: RangeOpts{
+ Upper: &FloatTermValue{Value: 100.0},
+ Lower: &FloatTermValue{Value: 1.0},
+ IncludesUpper: false,
+ IncludesLower: false,
+ },
+ expectedUpper: &FloatTermValue{Value: 100.0},
+ expectedLower: &FloatTermValue{Value: 1.0},
+ includesUpper: false,
+ includesLower: false,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Marshal
+ marshaled, err := tt.original.Marshal()
+ require.NoError(t, err)
+ assert.NotNil(t, marshaled)
+ assert.Greater(t, len(marshaled), 0)
+
+ // Unmarshal
+ unmarshaled := &RangeOpts{}
+ err = unmarshaled.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Verify upper bound
+ if tt.expectedUpper != nil {
+ assert.NotNil(t, unmarshaled.Upper)
+ switch expected := tt.expectedUpper.(type) {
+ case *BytesTermValue:
+ actual, ok := unmarshaled.Upper.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := unmarshaled.Upper.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ } else {
+ assert.Nil(t, unmarshaled.Upper)
+ }
+
+ // Verify lower bound
+ if tt.expectedLower != nil {
+ assert.NotNil(t, unmarshaled.Lower)
+ switch expected := tt.expectedLower.(type) {
+ case *BytesTermValue:
+ actual, ok := unmarshaled.Lower.(*BytesTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ case *FloatTermValue:
+ actual, ok := unmarshaled.Lower.(*FloatTermValue)
+ require.True(t, ok)
+ assert.Equal(t, expected.Value, actual.Value)
+ }
+ } else {
+ assert.Nil(t, unmarshaled.Lower)
+ }
+
+ // Verify inclusion flags
+ assert.Equal(t, tt.includesUpper, unmarshaled.IncludesUpper)
+ assert.Equal(t, tt.includesLower, unmarshaled.IncludesLower)
+ })
+ }
+}
diff --git a/pkg/index/term_value_test.go b/pkg/index/term_value_test.go
new file mode 100644
index 00000000..69df5ee7
--- /dev/null
+++ b/pkg/index/term_value_test.go
@@ -0,0 +1,263 @@
+// 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 index
+
+import (
+ "bytes"
+ "math"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestBytesTermValue_Marshal_Unmarshal(t *testing.T) {
+ tests := []struct {
+ name string
+ value []byte
+ expected []byte
+ }{
+ {
+ name: "empty bytes",
+ value: []byte{},
+ expected: []byte{},
+ },
+ {
+ name: "simple string",
+ value: []byte("hello world"),
+ expected: []byte("hello world"),
+ },
+ {
+ name: "binary data",
+ value: []byte{0x00, 0x01, 0x02, 0xFF, 0xFE, 0xFD},
+ expected: []byte{0x00, 0x01, 0x02, 0xFF, 0xFE, 0xFD},
+ },
+ {
+ name: "unicode string",
+ value: []byte("测试字符串"),
+ expected: []byte("测试字符串"),
+ },
+ {
+ name: "large data",
+ value: bytes.Repeat([]byte{0x42}, 1000),
+ expected: bytes.Repeat([]byte{0x42}, 1000),
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Create original BytesTermValue
+ original := &BytesTermValue{Value: tt.value}
+
+ // Marshal
+ marshaled, err := original.Marshal()
+ require.NoError(t, err)
+ assert.NotNil(t, marshaled)
+ assert.Greater(t, len(marshaled), 0)
+
+ // Verify type byte is correct (0 for bytes)
+ assert.Equal(t, byte(0), marshaled[0])
+
+ // Unmarshal
+ unmarshaled := &BytesTermValue{}
+ err = unmarshaled.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Verify the unmarshaled value matches the original
+ assert.Equal(t, tt.expected, unmarshaled.Value)
+ assert.Equal(t, original.Value, unmarshaled.Value)
+ })
+ }
+}
+
+func TestBytesTermValue_Unmarshal_InvalidData(t *testing.T) {
+ tests := []struct {
+ name string
+ expectedErr string
+ data []byte
+ }{
+ {
+ name: "empty data",
+ data: []byte{},
+ expectedErr: "insufficient data for term value type",
+ },
+ {
+ name: "wrong type byte",
+ data: []byte{1}, // float type instead of bytes type
+ expectedErr: "invalid term value type, expected bytes (0), got 1",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ unmarshaled := &BytesTermValue{}
+ err := unmarshaled.Unmarshal(tt.data)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tt.expectedErr)
+ })
+ }
+}
+
+func TestFloatTermValue_Marshal_Unmarshal(t *testing.T) {
+ tests := []struct {
+ name string
+ value float64
+ expected float64
+ }{
+ {
+ name: "zero",
+ value: 0.0,
+ expected: 0.0,
+ },
+ {
+ name: "positive integer",
+ value: 42.0,
+ expected: 42.0,
+ },
+ {
+ name: "negative integer",
+ value: -42.0,
+ expected: -42.0,
+ },
+ {
+ name: "positive decimal",
+ value: 3.14159,
+ expected: 3.14159,
+ },
+ {
+ name: "negative decimal",
+ value: -2.71828,
+ expected: -2.71828,
+ },
+ {
+ name: "large positive number",
+ value: 1e10,
+ expected: 1e10,
+ },
+ {
+ name: "large negative number",
+ value: -1e10,
+ expected: -1e10,
+ },
+ {
+ name: "small positive number",
+ value: 1e-10,
+ expected: 1e-10,
+ },
+ {
+ name: "small negative number",
+ value: -1e-10,
+ expected: -1e-10,
+ },
+ {
+ name: "max float64",
+ value: math.MaxFloat64,
+ expected: math.MaxFloat64,
+ },
+ {
+ name: "min float64",
+ value: -math.MaxFloat64,
+ expected: -math.MaxFloat64,
+ },
+ {
+ name: "infinity",
+ value: math.Inf(1),
+ expected: math.Inf(1),
+ },
+ {
+ name: "negative infinity",
+ value: math.Inf(-1),
+ expected: math.Inf(-1),
+ },
+ {
+ name: "NaN",
+ value: math.NaN(),
+ expected: math.NaN(),
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Create original FloatTermValue
+ original := &FloatTermValue{Value: tt.value}
+
+ // Marshal
+ marshaled, err := original.Marshal()
+ require.NoError(t, err)
+ assert.NotNil(t, marshaled)
+ assert.Greater(t, len(marshaled), 0)
+
+ // Verify type byte is correct (1 for float)
+ assert.Equal(t, byte(1), marshaled[0])
+
+ // Unmarshal
+ unmarshaled := &FloatTermValue{}
+ err = unmarshaled.Unmarshal(marshaled)
+ require.NoError(t, err)
+
+ // Verify the unmarshaled value matches the original
+ if math.IsNaN(tt.expected) {
+ assert.True(t, math.IsNaN(unmarshaled.Value))
+ } else {
+ assert.Equal(t, tt.expected, unmarshaled.Value)
+ }
+ // For NaN, we can't use direct equality comparison
+ if !math.IsNaN(original.Value) {
+ assert.Equal(t, original.Value, unmarshaled.Value)
+ }
+ })
+ }
+}
+
+func TestFloatTermValue_Unmarshal_InvalidData(t *testing.T) {
+ tests := []struct {
+ name string
+ expectedErr string
+ data []byte
+ }{
+ {
+ name: "empty data",
+ data: []byte{},
+ expectedErr: "insufficient data for term value type",
+ },
+ {
+ name: "wrong type byte",
+ data: []byte{0}, // bytes type instead of float type
+ expectedErr: "invalid term value type, expected float (1), got 0",
+ },
+ {
+ name: "incomplete data after type byte",
+ data: []byte{1},
+ expectedErr: "invalid float value length: 0",
+ },
+ {
+ name: "wrong float value length",
+ data: []byte{1, 0, 0, 0, 0, 0, 0}, // 7 bytes instead of 8
+ expectedErr: "invalid float value length: 0",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ unmarshaled := &FloatTermValue{}
+ err := unmarshaled.Unmarshal(tt.data)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tt.expectedErr)
+ })
+ }
+}
diff --git a/pkg/run/run.go b/pkg/run/run.go
index eca91b61..7c0bd958 100644
--- a/pkg/run/run.go
+++ b/pkg/run/run.go
@@ -421,10 +421,10 @@ func (g *Group) Run(ctx context.Context) (err error) {
<-notify
return nil
}, func(_ error) {
- g.log.Debug().Uint32("total", uint32(len(g.s))).Uint32("ran", uint32(idx+1)).Str("name", s.Name()).Msg("stopping")
+ g.log.Warn().Uint32("total", uint32(len(g.s))).Uint32("ran", uint32(idx+1)).Str("name", s.Name()).Msg("stopping")
startTime := time.Now()
s.GracefulStop()
- g.log.Info().Dur("elapsed", time.Since(startTime)).Str("name", s.Name()).Msg("stopped")
+ g.log.Warn().Dur("elapsed", time.Since(startTime)).Str("name", s.Name()).Msg("stopped")
})
}
diff --git a/test/cases/stream/stream.go b/test/cases/stream/stream.go
index 69cbfd4f..a78255b9 100644
--- a/test/cases/stream/stream.go
+++ b/test/cases/stream/stream.go
@@ -41,12 +41,12 @@ var (
}
)
-var _ = g.DescribeTable("Scanning Streams", func(args helpers.Args) {
+var _ = g.FDescribeTable("Scanning Streams", func(args helpers.Args) {
gm.Eventually(func(innerGm gm.Gomega) {
verify(innerGm, args)
}, flags.EventuallyTimeout).Should(gm.Succeed())
},
- g.Entry("all elements", helpers.Args{Input: "all", Duration: 1 * time.Hour}),
+ g.FEntry("all elements", helpers.Args{Input: "all", Duration: 1 * time.Hour}),
g.Entry("limit", helpers.Args{Input: "limit", Duration: 1 * time.Hour}),
g.Entry("max limit", helpers.Args{Input: "all_max_limit", Want: "all", Duration: 1 * time.Hour}),
g.Entry("offset", helpers.Args{Input: "offset", Duration: 1 * time.Hour}),
@@ -83,8 +83,8 @@ var _ = g.DescribeTable("Scanning Streams", func(args helpers.Args) {
g.Entry("filter by non-indexed tag with or", helpers.Args{Input: "filter_no_indexed_or", Duration: 1 * time.Hour}),
g.Entry("filter with desc order", helpers.Args{Input: "filter_order_desc", Duration: 1 * time.Hour}),
g.Entry("duplicated all elements", helpers.Args{Input: "duplicated_all", Duration: 1 * time.Hour, DisOrder: true}),
- g.Entry("duplicated entity filter", helpers.Args{Input: "duplicated_entity_filter", Duration: 1 * time.Hour}),
- g.Entry("duplicated index filter", helpers.Args{Input: "duplicated_index_filter", Duration: 1 * time.Hour}),
+ g.Entry("duplicated entity filter", helpers.Args{Input: "duplicated_entity_filter", Duration: 1 * time.Hour, DisOrder: true}),
+ g.Entry("duplicated index filter", helpers.Args{Input: "duplicated_index_filter", Duration: 1 * time.Hour, DisOrder: true}),
g.Entry("duplicated order by index", helpers.Args{Input: "duplicated_order_by_index", Duration: 1 * time.Hour}),
g.Entry("duplicated order by index with the index filter", helpers.Args{Input: "duplicated_order_by_filter", Duration: 1 * time.Hour}),
g.Entry("multi-groups: unchanged tags", helpers.Args{Input: "multi_group_unchanged", Duration: 1 * time.Hour, IgnoreElementID: true}),
(skywalking-banyandb) 01/04: Fix topN parsing panic when the criteria is set
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch topn
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 09c558b97b853dfcfa6e2b82d6b5740b29bdf576
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Jun 25 21:04:52 2025 +0800
Fix topN parsing panic when the criteria is set
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/metadata.go | 29 ++++++++++++++++++++++
.../topn_aggregations/not_in_svc2_svc4 copy.json | 28 +++++++++++++++++++++
2 files changed, 57 insertions(+)
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index a9d87e88..0edad3ac 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -22,6 +22,7 @@ import (
"fmt"
"io"
"path"
+ "strings"
"sync"
"time"
@@ -236,6 +237,7 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
Kind: resourceSchema.EventKindGroup,
Metadata: g,
})
+ sr.stopAllProcessorsWithGroupPrefix(g.Metadata.Name)
case schema.KindMeasure:
m := metadata.Spec.(*databasev1.Measure)
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
@@ -338,6 +340,33 @@ func (sr *schemaRepo) createTopNResultMeasure(ctx context.Context, measureSchema
}
}
+func (sr *schemaRepo) stopAllProcessorsWithGroupPrefix(groupName string) {
+ var keysToDelete []string
+ groupPrefix := groupName + "/"
+
+ sr.topNProcessorMap.Range(func(key, val any) bool {
+ keyStr := key.(string)
+ if strings.HasPrefix(keyStr, groupPrefix) {
+ keysToDelete = append(keysToDelete, keyStr)
+ }
+ return true
+ })
+
+ for _, key := range keysToDelete {
+ if v, ok := sr.topNProcessorMap.Load(key); ok {
+ manager := v.(*topNProcessorManager)
+ if err := manager.Close(); err != nil {
+ sr.l.Error().Err(err).Str("key", key).Msg("failed to close topN processor manager")
+ }
+ sr.topNProcessorMap.Delete(key)
+ }
+ }
+
+ if len(keysToDelete) > 0 {
+ sr.l.Info().Str("groupName", groupName).Int("count", len(keysToDelete)).Msg("stopped topN processors for group")
+ }
+}
+
var _ resourceSchema.ResourceSupplier = (*supplier)(nil)
type supplier struct {
diff --git a/pkg/test/measure/testdata/topn_aggregations/not_in_svc2_svc4 copy.json b/pkg/test/measure/testdata/topn_aggregations/not_in_svc2_svc4 copy.json
new file mode 100644
index 00000000..d95e3c7e
--- /dev/null
+++ b/pkg/test/measure/testdata/topn_aggregations/not_in_svc2_svc4 copy.json
@@ -0,0 +1,28 @@
+{
+ "metadata": {
+ "name": "not_in_svc2_svc4",
+ "group": "sw_metric"
+ },
+ "source_measure": {
+ "name": "service_instance_endpoint_cpm_minute",
+ "group": "sw_metric"
+ },
+ "field_name": "total",
+ "field_value_sort": 0,
+ "group_by_tag_names": [
+ "http.uri"
+ ],
+ "criteria": {
+ "condition":{
+ "name": "service_id",
+ "op": "BINARY_OP_NOT_IN",
+ "value": {
+ "str_array": {
+ "value": ["svc_2", "svc_4"]
+ }
+ }
+ }
+ },
+ "counters_number": 1000,
+ "lru_size": 10
+}
\ No newline at end of file
(skywalking-banyandb) 03/04: Merge remote-tracking branch 'origin/main' into topn
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch topn
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit bc1cb31e2799421e695e7d4b6816ae0ed00bd5a3
Merge: 03f338e2 164b4917
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Jul 11 18:13:36 2025 +0800
Merge remote-tracking branch 'origin/main' into topn
CHANGES.md | 1 +
api/proto/banyandb/measure/v1/query.proto | 2 +
banyand/internal/storage/storage.go | 2 +
banyand/property/db.go | 55 +-
banyand/property/repair.go | 1007 ++++++++++++++++++++
banyand/property/repair_test.go | 646 +++++++++++++
banyand/property/service.go | 59 +-
banyand/property/shard.go | 39 +-
banyand/property/shard_test.go | 24 +-
banyand/query/processor.go | 150 +++
banyand/stream/block.go | 3 +-
docs/api-reference.md | 1 +
pkg/encoding/float.go | 2 -
.../logical/measure/measure_plan_distributed.go | 7 +
pkg/timestamp/scheduler.go | 16 +
15 files changed, 1962 insertions(+), 52 deletions(-)