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/08/20 03:37:08 UTC
(skywalking-banyandb) 02/05: Add mock components for SIDX testing
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch sidx/mock
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit b638fcd5f9f205350cf5cd6899fb4b2bbcf0c292
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Aug 20 09:00:43 2025 +0700
Add mock components for SIDX testing
- Introduced `mock_components.go` and `mock_components_test.go` to implement mock versions of Writer, Querier, Flusher, and Merger for testing purposes.
- Implemented core functionalities including element accumulation, range filtering, and error injection.
---
banyand/internal/sidx/TODO.md | 20 +-
banyand/internal/sidx/mock_components.go | 636 +++++++++++++++++++++++++
banyand/internal/sidx/mock_components_test.go | 658 ++++++++++++++++++++++++++
3 files changed, 1304 insertions(+), 10 deletions(-)
diff --git a/banyand/internal/sidx/TODO.md b/banyand/internal/sidx/TODO.md
index ff200293..f0425b89 100644
--- a/banyand/internal/sidx/TODO.md
+++ b/banyand/internal/sidx/TODO.md
@@ -167,16 +167,16 @@ This document tracks the implementation progress of the Secondary Index File Sys
- [x] Query filtering produces correct results
- [x] Error injection works as expected
-### 3.2 Mock Component Implementations (`mock_components.go`)
-- [ ] Create mock Writer with element accumulation
-- [ ] Create mock Querier with range filtering
-- [ ] Create mock Flusher with no-op operations
-- [ ] Create mock Merger with simple consolidation
-- [ ] **Test Cases**:
- - [ ] All mock components integrate correctly
- - [ ] Mock behavior is configurable and predictable
- - [ ] Component interactions work as designed
- - [ ] Performance characteristics are documented
+### 3.2 Mock Component Implementations (`mock_components.go`) ✅
+- [x] Create mock Writer with element accumulation
+- [x] Create mock Querier with range filtering
+- [x] Create mock Flusher with no-op operations
+- [x] Create mock Merger with simple consolidation
+- [x] **Test Cases**:
+ - [x] All mock components integrate correctly
+ - [x] Mock behavior is configurable and predictable
+ - [x] Component interactions work as designed
+ - [x] Performance characteristics are documented
### 3.3 Integration Test Framework (`integration_test_framework.go`)
- [ ] Create test harness using mock implementations
diff --git a/banyand/internal/sidx/mock_components.go b/banyand/internal/sidx/mock_components.go
new file mode 100644
index 00000000..dd79c371
--- /dev/null
+++ b/banyand/internal/sidx/mock_components.go
@@ -0,0 +1,636 @@
+// 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 sidx
+
+import (
+ "context"
+ "fmt"
+ "sort"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+)
+
+// MockWriter provides a mock implementation of the Writer interface
+// for testing and early integration development.
+type MockWriter struct {
+ mu sync.RWMutex
+ elements []WriteRequest
+ config MockWriterConfig
+ writeCount atomic.Int64
+ elementCount atomic.Int64
+ lastWriteTime int64
+}
+
+// MockWriterConfig provides configuration options for the mock writer.
+type MockWriterConfig struct {
+ // DelayMs introduces artificial delay for write operations
+ DelayMs int
+ // ErrorRate controls the percentage of operations that should fail (0-100)
+ ErrorRate int
+ // MaxElements limits the number of elements to prevent OOM in tests
+ MaxElements int
+ // EnableValidation enables strict request validation
+ EnableValidation bool
+ // SortElements automatically sorts elements by SeriesID then Key
+ SortElements bool
+}
+
+// DefaultMockWriterConfig returns a sensible default configuration.
+func DefaultMockWriterConfig() MockWriterConfig {
+ return MockWriterConfig{
+ DelayMs: 0,
+ ErrorRate: 0,
+ MaxElements: 50000,
+ EnableValidation: true,
+ SortElements: true,
+ }
+}
+
+// NewMockWriter creates a new mock writer with the given configuration.
+func NewMockWriter(config MockWriterConfig) *MockWriter {
+ return &MockWriter{
+ elements: make([]WriteRequest, 0),
+ config: config,
+ }
+}
+
+// Write implements the Writer interface with element accumulation.
+func (mw *MockWriter) Write(ctx context.Context, reqs []WriteRequest) error {
+ if len(reqs) == 0 {
+ return nil
+ }
+
+ // Simulate artificial delay if configured
+ if mw.config.DelayMs > 0 {
+ time.Sleep(time.Duration(mw.config.DelayMs) * time.Millisecond)
+ }
+
+ // Simulate error injection if configured
+ if mw.config.ErrorRate > 0 && (int(time.Now().UnixNano())%100) < mw.config.ErrorRate {
+ return fmt.Errorf("mock writer error injection: write failed")
+ }
+
+ // Validate requests if enabled
+ if mw.config.EnableValidation {
+ for i, req := range reqs {
+ if err := req.Validate(); err != nil {
+ return fmt.Errorf("invalid write request[%d]: %w", i, err)
+ }
+ }
+ }
+
+ mw.mu.Lock()
+ defer mw.mu.Unlock()
+
+ // Check element limit
+ if mw.config.MaxElements > 0 && len(mw.elements)+len(reqs) > mw.config.MaxElements {
+ return fmt.Errorf("element limit exceeded: current=%d, adding=%d, limit=%d",
+ len(mw.elements), len(reqs), mw.config.MaxElements)
+ }
+
+ // Accumulate elements
+ mw.elements = append(mw.elements, reqs...)
+
+ // Sort elements if configured
+ if mw.config.SortElements {
+ sort.Slice(mw.elements, func(i, j int) bool {
+ if mw.elements[i].SeriesID != mw.elements[j].SeriesID {
+ return mw.elements[i].SeriesID < mw.elements[j].SeriesID
+ }
+ return mw.elements[i].Key < mw.elements[j].Key
+ })
+ }
+
+ // Update stats
+ mw.writeCount.Add(1)
+ mw.elementCount.Add(int64(len(reqs)))
+ atomic.StoreInt64(&mw.lastWriteTime, time.Now().UnixNano())
+
+ return nil
+}
+
+// GetElements returns a copy of all accumulated elements (for testing).
+func (mw *MockWriter) GetElements() []WriteRequest {
+ mw.mu.RLock()
+ defer mw.mu.RUnlock()
+
+ result := make([]WriteRequest, len(mw.elements))
+ copy(result, mw.elements)
+ return result
+}
+
+// GetStats returns writer statistics.
+func (mw *MockWriter) GetStats() (writeCount, elementCount int64, lastWriteTime time.Time) {
+ lastWriteNanos := atomic.LoadInt64(&mw.lastWriteTime)
+ if lastWriteNanos == 0 {
+ return mw.writeCount.Load(), mw.elementCount.Load(), time.Time{}
+ }
+ return mw.writeCount.Load(), mw.elementCount.Load(), time.Unix(0, lastWriteNanos)
+}
+
+// Clear removes all accumulated elements (for testing).
+func (mw *MockWriter) Clear() {
+ mw.mu.Lock()
+ defer mw.mu.Unlock()
+
+ mw.elements = mw.elements[:0]
+ mw.elementCount.Store(0)
+}
+
+// SetErrorRate allows dynamic configuration of error injection rate.
+func (mw *MockWriter) SetErrorRate(rate int) {
+ mw.mu.Lock()
+ defer mw.mu.Unlock()
+ if rate >= 0 && rate <= 100 {
+ mw.config.ErrorRate = rate
+ }
+}
+
+// MockQuerier provides a mock implementation of the Querier interface
+// for testing and early integration development.
+type MockQuerier struct {
+ mu sync.RWMutex
+ elements []WriteRequest
+ config MockQuerierConfig
+ queryCount atomic.Int64
+ lastQueryTime int64
+}
+
+// MockQuerierConfig provides configuration options for the mock querier.
+type MockQuerierConfig struct {
+ // DelayMs introduces artificial delay for query operations
+ DelayMs int
+ // ErrorRate controls the percentage of operations that should fail (0-100)
+ ErrorRate int
+ // EnableValidation enables strict request validation
+ EnableValidation bool
+ // MaxResultSize limits the maximum number of results returned
+ MaxResultSize int
+}
+
+// DefaultMockQuerierConfig returns a sensible default configuration.
+func DefaultMockQuerierConfig() MockQuerierConfig {
+ return MockQuerierConfig{
+ DelayMs: 0,
+ ErrorRate: 0,
+ EnableValidation: true,
+ MaxResultSize: 10000,
+ }
+}
+
+// NewMockQuerier creates a new mock querier with the given configuration.
+func NewMockQuerier(config MockQuerierConfig) *MockQuerier {
+ return &MockQuerier{
+ elements: make([]WriteRequest, 0),
+ config: config,
+ }
+}
+
+// SetElements sets the elements available for querying (for testing).
+func (mq *MockQuerier) SetElements(elements []WriteRequest) {
+ mq.mu.Lock()
+ defer mq.mu.Unlock()
+
+ mq.elements = make([]WriteRequest, len(elements))
+ copy(mq.elements, elements)
+
+ // Ensure elements are sorted for range filtering
+ sort.Slice(mq.elements, func(i, j int) bool {
+ if mq.elements[i].SeriesID != mq.elements[j].SeriesID {
+ return mq.elements[i].SeriesID < mq.elements[j].SeriesID
+ }
+ return mq.elements[i].Key < mq.elements[j].Key
+ })
+}
+
+// Query implements the Querier interface with range filtering.
+func (mq *MockQuerier) Query(ctx context.Context, req QueryRequest) (QueryResult, error) {
+ // Simulate artificial delay if configured
+ if mq.config.DelayMs > 0 {
+ time.Sleep(time.Duration(mq.config.DelayMs) * time.Millisecond)
+ }
+
+ // Simulate error injection if configured
+ if mq.config.ErrorRate > 0 && (int(time.Now().UnixNano())%100) < mq.config.ErrorRate {
+ return nil, fmt.Errorf("mock querier error injection: query failed")
+ }
+
+ // Validate request if enabled
+ if mq.config.EnableValidation {
+ if err := req.Validate(); err != nil {
+ return nil, fmt.Errorf("invalid query request: %w", err)
+ }
+ }
+
+ mq.mu.RLock()
+ defer mq.mu.RUnlock()
+
+ // Update stats
+ mq.queryCount.Add(1)
+ atomic.StoreInt64(&mq.lastQueryTime, time.Now().UnixNano())
+
+ // Apply basic range filtering (simplified for mock)
+ var filteredElements []WriteRequest
+ for _, elem := range mq.elements {
+ // Simple filtering based on request parameters
+ // In a real implementation, this would use proper index filtering
+ if mq.matchesQuery(elem, req) {
+ filteredElements = append(filteredElements, elem)
+ if mq.config.MaxResultSize > 0 && len(filteredElements) >= mq.config.MaxResultSize {
+ break
+ }
+ }
+ }
+
+ // Create and return query result
+ return &mockQuerierResult{
+ elements: filteredElements,
+ request: req,
+ }, nil
+}
+
+// matchesQuery performs basic filtering to determine if an element matches the query.
+func (mq *MockQuerier) matchesQuery(elem WriteRequest, req QueryRequest) bool {
+ // Basic range filtering - in reality this would be more sophisticated
+ // For the mock, we'll match all elements unless specific filters are applied
+
+ // Apply entity filtering if specified
+ if len(req.Entities) > 0 {
+ // Simplified entity matching for the mock
+ // In practice, this would properly evaluate entity constraints
+ _ = elem // Use elem to avoid unused parameter warning
+ }
+
+ // For the mock, we'll accept all elements that pass basic checks
+ return true
+}
+
+// GetStats returns querier statistics.
+func (mq *MockQuerier) GetStats() (queryCount int64, lastQueryTime time.Time) {
+ lastQueryNanos := atomic.LoadInt64(&mq.lastQueryTime)
+ if lastQueryNanos == 0 {
+ return mq.queryCount.Load(), time.Time{}
+ }
+ return mq.queryCount.Load(), time.Unix(0, lastQueryNanos)
+}
+
+// SetErrorRate allows dynamic configuration of error injection rate.
+func (mq *MockQuerier) SetErrorRate(rate int) {
+ mq.mu.Lock()
+ defer mq.mu.Unlock()
+ if rate >= 0 && rate <= 100 {
+ mq.config.ErrorRate = rate
+ }
+}
+
+// mockQuerierResult implements QueryResult for the mock querier.
+type mockQuerierResult struct {
+ elements []WriteRequest
+ request QueryRequest
+ position int
+ finished bool
+}
+
+// Pull returns the next batch of query results.
+func (mqr *mockQuerierResult) Pull() *QueryResponse {
+ if mqr.finished || mqr.position >= len(mqr.elements) {
+ return nil
+ }
+
+ startTime := time.Now()
+ batchSize := mqr.request.MaxElementSize
+ if batchSize <= 0 {
+ batchSize = len(mqr.elements) - mqr.position // Return all remaining
+ }
+
+ endPos := mqr.position + batchSize
+ if endPos > len(mqr.elements) {
+ endPos = len(mqr.elements)
+ }
+
+ batch := mqr.elements[mqr.position:endPos]
+ mqr.position = endPos
+
+ if mqr.position >= len(mqr.elements) {
+ mqr.finished = true
+ }
+
+ // Convert batch to response format
+ response := &QueryResponse{
+ Keys: make([]int64, len(batch)),
+ Data: make([][]byte, len(batch)),
+ Tags: make([][]tag, len(batch)),
+ SIDs: make([]common.SeriesID, len(batch)),
+ }
+
+ for i, elem := range batch {
+ response.Keys[i] = elem.Key
+ response.Data[i] = make([]byte, len(elem.Data))
+ copy(response.Data[i], elem.Data)
+ response.Tags[i] = make([]tag, len(elem.Tags))
+ copy(response.Tags[i], elem.Tags)
+ response.SIDs[i] = elem.SeriesID
+ }
+
+ // Fill metadata
+ response.Metadata = ResponseMetadata{
+ ExecutionTimeMs: time.Since(startTime).Milliseconds(),
+ ElementsScanned: int64(len(batch)),
+ ElementsFiltered: int64(len(batch)),
+ PartsAccessed: 1, // Mock has 1 virtual part
+ BlocksScanned: 1, // Mock scans 1 virtual block
+ CacheHitRatio: 1.0, // Everything is in "memory cache"
+ TruncatedResults: mqr.request.MaxElementSize > 0 && len(batch) == batchSize && !mqr.finished,
+ }
+
+ return response
+}
+
+// Release releases resources associated with the query result.
+func (mqr *mockQuerierResult) Release() {
+ mqr.elements = nil
+ mqr.finished = true
+}
+
+// MockFlusher provides a mock implementation of the Flusher interface
+// for testing and early integration development.
+type MockFlusher struct {
+ mu sync.RWMutex
+ config MockFlusherConfig
+ flushCount atomic.Int64
+ lastFlushTime int64
+}
+
+// MockFlusherConfig provides configuration options for the mock flusher.
+type MockFlusherConfig struct {
+ // DelayMs introduces artificial delay for flush operations
+ DelayMs int
+ // ErrorRate controls the percentage of operations that should fail (0-100)
+ ErrorRate int
+ // SimulateWork enables simulation of actual flush work
+ SimulateWork bool
+}
+
+// DefaultMockFlusherConfig returns a sensible default configuration.
+func DefaultMockFlusherConfig() MockFlusherConfig {
+ return MockFlusherConfig{
+ DelayMs: 50, // Simulate some disk I/O time
+ ErrorRate: 0,
+ SimulateWork: true,
+ }
+}
+
+// NewMockFlusher creates a new mock flusher with the given configuration.
+func NewMockFlusher(config MockFlusherConfig) *MockFlusher {
+ return &MockFlusher{
+ config: config,
+ }
+}
+
+// Flush implements the Flusher interface with no-op operations.
+func (mf *MockFlusher) Flush() error {
+ // Simulate artificial delay if configured
+ if mf.config.DelayMs > 0 {
+ time.Sleep(time.Duration(mf.config.DelayMs) * time.Millisecond)
+ }
+
+ // Simulate error injection if configured
+ if mf.config.ErrorRate > 0 && (int(time.Now().UnixNano())%100) < mf.config.ErrorRate {
+ return fmt.Errorf("mock flusher error injection: flush failed")
+ }
+
+ mf.mu.Lock()
+ defer mf.mu.Unlock()
+
+ // Simulate flush work if enabled
+ if mf.config.SimulateWork {
+ // Simulate some processing time
+ time.Sleep(10 * time.Millisecond)
+ }
+
+ // Update stats
+ mf.flushCount.Add(1)
+ atomic.StoreInt64(&mf.lastFlushTime, time.Now().UnixNano())
+
+ return nil
+}
+
+// GetStats returns flusher statistics.
+func (mf *MockFlusher) GetStats() (flushCount int64, lastFlushTime time.Time) {
+ lastFlushNanos := atomic.LoadInt64(&mf.lastFlushTime)
+ if lastFlushNanos == 0 {
+ return mf.flushCount.Load(), time.Time{}
+ }
+ return mf.flushCount.Load(), time.Unix(0, lastFlushNanos)
+}
+
+// SetErrorRate allows dynamic configuration of error injection rate.
+func (mf *MockFlusher) SetErrorRate(rate int) {
+ mf.mu.Lock()
+ defer mf.mu.Unlock()
+ if rate >= 0 && rate <= 100 {
+ mf.config.ErrorRate = rate
+ }
+}
+
+// MockMerger provides a mock implementation of the Merger interface
+// for testing and early integration development.
+type MockMerger struct {
+ mu sync.RWMutex
+ config MockMergerConfig
+ mergeCount atomic.Int64
+ lastMergeTime int64
+ consolidations []MergeConsolidation
+}
+
+// MergeConsolidation represents a simulated merge operation result.
+type MergeConsolidation struct {
+ Timestamp time.Time
+ PartsInput int
+ PartsOutput int
+ ElementsProcessed int64
+ Duration time.Duration
+}
+
+// MockMergerConfig provides configuration options for the mock merger.
+type MockMergerConfig struct {
+ // DelayMs introduces artificial delay for merge operations
+ DelayMs int
+ // ErrorRate controls the percentage of operations that should fail (0-100)
+ ErrorRate int
+ // SimulateWork enables simulation of actual merge work
+ SimulateWork bool
+ // ConsolidationRatio simulates the reduction ratio during merge (0.1 to 1.0)
+ ConsolidationRatio float64
+}
+
+// DefaultMockMergerConfig returns a sensible default configuration.
+func DefaultMockMergerConfig() MockMergerConfig {
+ return MockMergerConfig{
+ DelayMs: 100, // Simulate merge processing time
+ ErrorRate: 0,
+ SimulateWork: true,
+ ConsolidationRatio: 0.7, // 30% reduction typical
+ }
+}
+
+// NewMockMerger creates a new mock merger with the given configuration.
+func NewMockMerger(config MockMergerConfig) *MockMerger {
+ return &MockMerger{
+ config: config,
+ consolidations: make([]MergeConsolidation, 0),
+ }
+}
+
+// Merge implements the Merger interface with simple consolidation.
+func (mm *MockMerger) Merge() error {
+ startTime := time.Now()
+
+ // Simulate artificial delay if configured
+ if mm.config.DelayMs > 0 {
+ time.Sleep(time.Duration(mm.config.DelayMs) * time.Millisecond)
+ }
+
+ // Simulate error injection if configured
+ if mm.config.ErrorRate > 0 && (int(time.Now().UnixNano())%100) < mm.config.ErrorRate {
+ return fmt.Errorf("mock merger error injection: merge failed")
+ }
+
+ mm.mu.Lock()
+ defer mm.mu.Unlock()
+
+ // Simulate merge work if enabled
+ if mm.config.SimulateWork {
+ // Simulate processing time
+ time.Sleep(20 * time.Millisecond)
+ }
+
+ // Simulate consolidation
+ inputParts := 3 + (int(time.Now().UnixNano()) % 5) // 3-7 input parts
+ outputParts := int(float64(inputParts) * mm.config.ConsolidationRatio)
+ if outputParts < 1 {
+ outputParts = 1
+ }
+ elementsProcessed := int64(1000 + (time.Now().UnixNano() % 5000)) // 1000-6000 elements
+
+ consolidation := MergeConsolidation{
+ Timestamp: startTime,
+ PartsInput: inputParts,
+ PartsOutput: outputParts,
+ ElementsProcessed: elementsProcessed,
+ Duration: time.Since(startTime),
+ }
+
+ mm.consolidations = append(mm.consolidations, consolidation)
+
+ // Keep only last 10 consolidations for memory efficiency
+ if len(mm.consolidations) > 10 {
+ mm.consolidations = mm.consolidations[len(mm.consolidations)-10:]
+ }
+
+ // Update stats
+ mm.mergeCount.Add(1)
+ atomic.StoreInt64(&mm.lastMergeTime, time.Now().UnixNano())
+
+ return nil
+}
+
+// GetStats returns merger statistics.
+func (mm *MockMerger) GetStats() (mergeCount int64, lastMergeTime time.Time) {
+ lastMergeNanos := atomic.LoadInt64(&mm.lastMergeTime)
+ if lastMergeNanos == 0 {
+ return mm.mergeCount.Load(), time.Time{}
+ }
+ return mm.mergeCount.Load(), time.Unix(0, lastMergeNanos)
+}
+
+// GetConsolidations returns recent merge consolidation history (for testing).
+func (mm *MockMerger) GetConsolidations() []MergeConsolidation {
+ mm.mu.RLock()
+ defer mm.mu.RUnlock()
+
+ result := make([]MergeConsolidation, len(mm.consolidations))
+ copy(result, mm.consolidations)
+ return result
+}
+
+// SetErrorRate allows dynamic configuration of error injection rate.
+func (mm *MockMerger) SetErrorRate(rate int) {
+ mm.mu.Lock()
+ defer mm.mu.Unlock()
+ if rate >= 0 && rate <= 100 {
+ mm.config.ErrorRate = rate
+ }
+}
+
+// MockComponentSuite provides a convenient way to create and manage
+// all mock components together for integration testing.
+type MockComponentSuite struct {
+ Writer *MockWriter
+ Querier *MockQuerier
+ Flusher *MockFlusher
+ Merger *MockMerger
+}
+
+// NewMockComponentSuite creates a new suite of mock components with default configurations.
+func NewMockComponentSuite() *MockComponentSuite {
+ return &MockComponentSuite{
+ Writer: NewMockWriter(DefaultMockWriterConfig()),
+ Querier: NewMockQuerier(DefaultMockQuerierConfig()),
+ Flusher: NewMockFlusher(DefaultMockFlusherConfig()),
+ Merger: NewMockMerger(DefaultMockMergerConfig()),
+ }
+}
+
+// NewMockComponentSuiteWithConfigs creates a new suite with custom configurations.
+func NewMockComponentSuiteWithConfigs(
+ writerConfig MockWriterConfig,
+ querierConfig MockQuerierConfig,
+ flusherConfig MockFlusherConfig,
+ mergerConfig MockMergerConfig,
+) *MockComponentSuite {
+ return &MockComponentSuite{
+ Writer: NewMockWriter(writerConfig),
+ Querier: NewMockQuerier(querierConfig),
+ Flusher: NewMockFlusher(flusherConfig),
+ Merger: NewMockMerger(mergerConfig),
+ }
+}
+
+// SyncElements synchronizes elements from writer to querier (for testing workflows).
+func (mcs *MockComponentSuite) SyncElements() {
+ elements := mcs.Writer.GetElements()
+ mcs.Querier.SetElements(elements)
+}
+
+// SetGlobalErrorRate sets error rate for all components.
+func (mcs *MockComponentSuite) SetGlobalErrorRate(rate int) {
+ mcs.Writer.SetErrorRate(rate)
+ mcs.Querier.SetErrorRate(rate)
+ mcs.Flusher.SetErrorRate(rate)
+ mcs.Merger.SetErrorRate(rate)
+}
+
+// Reset clears all components to initial state (for testing).
+func (mcs *MockComponentSuite) Reset() {
+ mcs.Writer.Clear()
+ mcs.Querier.SetElements(nil)
+}
\ No newline at end of file
diff --git a/banyand/internal/sidx/mock_components_test.go b/banyand/internal/sidx/mock_components_test.go
new file mode 100644
index 00000000..44006150
--- /dev/null
+++ b/banyand/internal/sidx/mock_components_test.go
@@ -0,0 +1,658 @@
+// 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 sidx
+
+import (
+ "context"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+func TestMockWriter_BasicOperations(t *testing.T) {
+ config := DefaultMockWriterConfig()
+ writer := NewMockWriter(config)
+ ctx := context.Background()
+
+ // Test initial state
+ elements := writer.GetElements()
+ assert.Empty(t, elements)
+
+ writeCount, elementCount, lastWriteTime := writer.GetStats()
+ assert.Equal(t, int64(0), writeCount)
+ assert.Equal(t, int64(0), elementCount)
+ assert.True(t, lastWriteTime.IsZero())
+
+ // Test write operations
+ writeReqs := []WriteRequest{
+ {
+ SeriesID: 2,
+ Key: 200,
+ Data: []byte("data2"),
+ Tags: []tag{
+ {
+ name: "service",
+ value: []byte("test-service"),
+ valueType: pbv1.ValueTypeStr,
+ indexed: true,
+ },
+ },
+ },
+ {
+ SeriesID: 1,
+ Key: 100,
+ Data: []byte("data1"),
+ Tags: []tag{
+ {
+ name: "service",
+ value: []byte("test-service"),
+ valueType: pbv1.ValueTypeStr,
+ indexed: true,
+ },
+ },
+ },
+ }
+
+ err := writer.Write(ctx, writeReqs)
+ require.NoError(t, err)
+
+ // Verify elements are stored and sorted
+ elements = writer.GetElements()
+ require.Len(t, elements, 2)
+ assert.Equal(t, common.SeriesID(1), elements[0].SeriesID) // Should be sorted
+ assert.Equal(t, int64(100), elements[0].Key)
+ assert.Equal(t, common.SeriesID(2), elements[1].SeriesID)
+ assert.Equal(t, int64(200), elements[1].Key)
+
+ // Verify stats
+ writeCount, elementCount, lastWriteTime = writer.GetStats()
+ assert.Equal(t, int64(1), writeCount)
+ assert.Equal(t, int64(2), elementCount)
+ assert.False(t, lastWriteTime.IsZero())
+}
+
+func TestMockWriter_Validation(t *testing.T) {
+ config := DefaultMockWriterConfig()
+ config.EnableValidation = true
+ writer := NewMockWriter(config)
+ ctx := context.Background()
+
+ tests := []struct {
+ name string
+ writeReqs []WriteRequest
+ expectError bool
+ }{
+ {
+ name: "empty requests",
+ writeReqs: []WriteRequest{},
+ expectError: false,
+ },
+ {
+ name: "valid request",
+ writeReqs: []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data")},
+ },
+ expectError: false,
+ },
+ {
+ name: "invalid SeriesID",
+ writeReqs: []WriteRequest{
+ {SeriesID: 0, Key: 100, Data: []byte("data")},
+ },
+ expectError: true,
+ },
+ {
+ name: "nil data",
+ writeReqs: []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: nil},
+ },
+ expectError: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ writer.Clear() // Reset for each test
+ err := writer.Write(ctx, tt.writeReqs)
+ if tt.expectError {
+ assert.Error(t, err)
+ } else {
+ assert.NoError(t, err)
+ }
+ })
+ }
+}
+
+func TestMockWriter_ElementLimits(t *testing.T) {
+ config := DefaultMockWriterConfig()
+ config.MaxElements = 3
+ writer := NewMockWriter(config)
+ ctx := context.Background()
+
+ // Write up to limit
+ writeReqs := []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data1")},
+ {SeriesID: 1, Key: 101, Data: []byte("data2")},
+ {SeriesID: 1, Key: 102, Data: []byte("data3")},
+ }
+ err := writer.Write(ctx, writeReqs)
+ require.NoError(t, err)
+
+ // Try to exceed limit
+ writeReqs = []WriteRequest{
+ {SeriesID: 1, Key: 103, Data: []byte("data4")},
+ }
+ err = writer.Write(ctx, writeReqs)
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "element limit exceeded")
+}
+
+func TestMockWriter_ErrorInjection(t *testing.T) {
+ config := DefaultMockWriterConfig()
+ config.ErrorRate = 100 // Always error
+ writer := NewMockWriter(config)
+ ctx := context.Background()
+
+ writeReqs := []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data")},
+ }
+ err := writer.Write(ctx, writeReqs)
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "mock writer error injection")
+}
+
+func TestMockWriter_DelaySimulation(t *testing.T) {
+ config := DefaultMockWriterConfig()
+ config.DelayMs = 50
+ writer := NewMockWriter(config)
+ ctx := context.Background()
+
+ start := time.Now()
+ writeReqs := []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data")},
+ }
+ err := writer.Write(ctx, writeReqs)
+ require.NoError(t, err)
+ elapsed := time.Since(start)
+
+ assert.GreaterOrEqual(t, elapsed, 50*time.Millisecond)
+}
+
+func TestMockQuerier_BasicOperations(t *testing.T) {
+ config := DefaultMockQuerierConfig()
+ querier := NewMockQuerier(config)
+ ctx := context.Background()
+
+ // Test initial state
+ queryCount, lastQueryTime := querier.GetStats()
+ assert.Equal(t, int64(0), queryCount)
+ assert.True(t, lastQueryTime.IsZero())
+
+ // Set up test elements
+ elements := []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data1")},
+ {SeriesID: 1, Key: 101, Data: []byte("data2")},
+ {SeriesID: 2, Key: 200, Data: []byte("data3")},
+ }
+ querier.SetElements(elements)
+
+ // Test query operations
+ queryReq := QueryRequest{
+ Name: "test-query",
+ MaxElementSize: 10,
+ }
+
+ result, err := querier.Query(ctx, queryReq)
+ require.NoError(t, err)
+ require.NotNil(t, result)
+ defer result.Release()
+
+ // Pull results
+ response := result.Pull()
+ require.NotNil(t, response)
+ assert.NoError(t, response.Error)
+ assert.Equal(t, 3, response.Len())
+ assert.Equal(t, []int64{100, 101, 200}, response.Keys)
+
+ // Verify no more results
+ response = result.Pull()
+ assert.Nil(t, response)
+
+ // Verify stats
+ queryCount, lastQueryTime = querier.GetStats()
+ assert.Equal(t, int64(1), queryCount)
+ assert.False(t, lastQueryTime.IsZero())
+}
+
+func TestMockQuerier_QueryBatching(t *testing.T) {
+ config := DefaultMockQuerierConfig()
+ querier := NewMockQuerier(config)
+ ctx := context.Background()
+
+ // Set up many elements
+ var elements []WriteRequest
+ for i := 0; i < 10; i++ {
+ elements = append(elements, WriteRequest{
+ SeriesID: 1,
+ Key: int64(i),
+ Data: []byte(fmt.Sprintf("data%d", i)),
+ })
+ }
+ querier.SetElements(elements)
+
+ // Query with small batch size
+ queryReq := QueryRequest{
+ Name: "batch-test",
+ MaxElementSize: 3,
+ }
+
+ result, err := querier.Query(ctx, queryReq)
+ require.NoError(t, err)
+ defer result.Release()
+
+ // First batch
+ response := result.Pull()
+ require.NotNil(t, response)
+ assert.Equal(t, 3, response.Len())
+ assert.True(t, response.Metadata.TruncatedResults)
+
+ // Continue pulling until done
+ totalElements := 3
+ for {
+ response = result.Pull()
+ if response == nil {
+ break
+ }
+ totalElements += response.Len()
+ }
+ assert.Equal(t, 10, totalElements)
+}
+
+func TestMockQuerier_ErrorInjection(t *testing.T) {
+ config := DefaultMockQuerierConfig()
+ config.ErrorRate = 100 // Always error
+ querier := NewMockQuerier(config)
+ ctx := context.Background()
+
+ queryReq := QueryRequest{Name: "test"}
+ _, err := querier.Query(ctx, queryReq)
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "mock querier error injection")
+}
+
+func TestMockFlusher_BasicOperations(t *testing.T) {
+ config := DefaultMockFlusherConfig()
+ flusher := NewMockFlusher(config)
+
+ // Test initial state
+ flushCount, lastFlushTime := flusher.GetStats()
+ assert.Equal(t, int64(0), flushCount)
+ assert.True(t, lastFlushTime.IsZero())
+
+ // Test flush operation
+ err := flusher.Flush()
+ require.NoError(t, err)
+
+ // Verify stats
+ flushCount, lastFlushTime = flusher.GetStats()
+ assert.Equal(t, int64(1), flushCount)
+ assert.False(t, lastFlushTime.IsZero())
+}
+
+func TestMockFlusher_DelaySimulation(t *testing.T) {
+ config := DefaultMockFlusherConfig()
+ config.DelayMs = 30
+ flusher := NewMockFlusher(config)
+
+ start := time.Now()
+ err := flusher.Flush()
+ require.NoError(t, err)
+ elapsed := time.Since(start)
+
+ assert.GreaterOrEqual(t, elapsed, 30*time.Millisecond)
+}
+
+func TestMockFlusher_ErrorInjection(t *testing.T) {
+ config := DefaultMockFlusherConfig()
+ config.ErrorRate = 100 // Always error
+ flusher := NewMockFlusher(config)
+
+ err := flusher.Flush()
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "mock flusher error injection")
+}
+
+func TestMockMerger_BasicOperations(t *testing.T) {
+ config := DefaultMockMergerConfig()
+ merger := NewMockMerger(config)
+
+ // Test initial state
+ mergeCount, lastMergeTime := merger.GetStats()
+ assert.Equal(t, int64(0), mergeCount)
+ assert.True(t, lastMergeTime.IsZero())
+
+ consolidations := merger.GetConsolidations()
+ assert.Empty(t, consolidations)
+
+ // Test merge operation
+ err := merger.Merge()
+ require.NoError(t, err)
+
+ // Verify stats
+ mergeCount, lastMergeTime = merger.GetStats()
+ assert.Equal(t, int64(1), mergeCount)
+ assert.False(t, lastMergeTime.IsZero())
+
+ // Verify consolidations
+ consolidations = merger.GetConsolidations()
+ require.Len(t, consolidations, 1)
+ consolidation := consolidations[0]
+ assert.Greater(t, consolidation.PartsInput, 0)
+ assert.Greater(t, consolidation.PartsOutput, 0)
+ assert.Greater(t, consolidation.ElementsProcessed, int64(0))
+ assert.Greater(t, consolidation.Duration, time.Duration(0))
+}
+
+func TestMockMerger_ConsolidationRatio(t *testing.T) {
+ config := DefaultMockMergerConfig()
+ config.ConsolidationRatio = 0.5 // 50% reduction
+ merger := NewMockMerger(config)
+
+ err := merger.Merge()
+ require.NoError(t, err)
+
+ consolidations := merger.GetConsolidations()
+ require.Len(t, consolidations, 1)
+ consolidation := consolidations[0]
+
+ // Output should be roughly half of input
+ expectedOutput := int(float64(consolidation.PartsInput) * 0.5)
+ if expectedOutput < 1 {
+ expectedOutput = 1
+ }
+ assert.Equal(t, expectedOutput, consolidation.PartsOutput)
+}
+
+func TestMockMerger_ErrorInjection(t *testing.T) {
+ config := DefaultMockMergerConfig()
+ config.ErrorRate = 100 // Always error
+ merger := NewMockMerger(config)
+
+ err := merger.Merge()
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "mock merger error injection")
+}
+
+func TestMockComponentSuite_Integration(t *testing.T) {
+ suite := NewMockComponentSuite()
+ ctx := context.Background()
+
+ // Test write-query workflow
+ writeReqs := []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data1")},
+ {SeriesID: 1, Key: 101, Data: []byte("data2")},
+ {SeriesID: 2, Key: 200, Data: []byte("data3")},
+ }
+
+ // Write elements
+ err := suite.Writer.Write(ctx, writeReqs)
+ require.NoError(t, err)
+
+ // Sync elements to querier
+ suite.SyncElements()
+
+ // Query elements
+ queryReq := QueryRequest{
+ Name: "integration-test",
+ MaxElementSize: 10,
+ }
+
+ result, err := suite.Querier.Query(ctx, queryReq)
+ require.NoError(t, err)
+ defer result.Release()
+
+ response := result.Pull()
+ require.NotNil(t, response)
+ assert.Equal(t, 3, response.Len())
+
+ // Test flush operation
+ err = suite.Flusher.Flush()
+ require.NoError(t, err)
+
+ // Test merge operation
+ err = suite.Merger.Merge()
+ require.NoError(t, err)
+}
+
+func TestMockComponentSuite_ErrorInjection(t *testing.T) {
+ suite := NewMockComponentSuite()
+ ctx := context.Background()
+
+ // Set global error rate
+ suite.SetGlobalErrorRate(100)
+
+ // All operations should fail
+ writeReqs := []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data")},
+ }
+ err := suite.Writer.Write(ctx, writeReqs)
+ assert.Error(t, err)
+
+ queryReq := QueryRequest{Name: "test"}
+ _, err = suite.Querier.Query(ctx, queryReq)
+ assert.Error(t, err)
+
+ err = suite.Flusher.Flush()
+ assert.Error(t, err)
+
+ err = suite.Merger.Merge()
+ assert.Error(t, err)
+}
+
+func TestMockComponentSuite_CustomConfigs(t *testing.T) {
+ writerConfig := MockWriterConfig{
+ DelayMs: 10,
+ ErrorRate: 0,
+ MaxElements: 100,
+ EnableValidation: true,
+ SortElements: true,
+ }
+ querierConfig := MockQuerierConfig{
+ DelayMs: 5,
+ ErrorRate: 0,
+ EnableValidation: true,
+ MaxResultSize: 50,
+ }
+ flusherConfig := MockFlusherConfig{
+ DelayMs: 20,
+ ErrorRate: 0,
+ SimulateWork: true,
+ }
+ mergerConfig := MockMergerConfig{
+ DelayMs: 30,
+ ErrorRate: 0,
+ SimulateWork: true,
+ ConsolidationRatio: 0.8,
+ }
+
+ suite := NewMockComponentSuiteWithConfigs(
+ writerConfig, querierConfig, flusherConfig, mergerConfig)
+
+ ctx := context.Background()
+
+ // Test that custom configurations work
+ start := time.Now()
+ writeReqs := []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data")},
+ }
+ err := suite.Writer.Write(ctx, writeReqs)
+ require.NoError(t, err)
+ writerElapsed := time.Since(start)
+
+ start = time.Now()
+ err = suite.Flusher.Flush()
+ require.NoError(t, err)
+ flusherElapsed := time.Since(start)
+
+ start = time.Now()
+ err = suite.Merger.Merge()
+ require.NoError(t, err)
+ mergerElapsed := time.Since(start)
+
+ // Verify delays are applied
+ assert.GreaterOrEqual(t, writerElapsed, 10*time.Millisecond)
+ assert.GreaterOrEqual(t, flusherElapsed, 20*time.Millisecond)
+ assert.GreaterOrEqual(t, mergerElapsed, 30*time.Millisecond)
+}
+
+func TestMockComponentSuite_Reset(t *testing.T) {
+ suite := NewMockComponentSuite()
+ ctx := context.Background()
+
+ // Add some data
+ writeReqs := []WriteRequest{
+ {SeriesID: 1, Key: 100, Data: []byte("data")},
+ }
+ err := suite.Writer.Write(ctx, writeReqs)
+ require.NoError(t, err)
+
+ suite.SyncElements()
+
+ // Verify data exists
+ elements := suite.Writer.GetElements()
+ assert.Len(t, elements, 1)
+
+ // Reset
+ suite.Reset()
+
+ // Verify data is cleared
+ elements = suite.Writer.GetElements()
+ assert.Empty(t, elements)
+}
+
+func TestMockComponents_ConcurrentOperations(t *testing.T) {
+ suite := NewMockComponentSuite()
+ ctx := context.Background()
+
+ const numGoroutines = 10
+ const operationsPerGoroutine = 10
+
+ // Concurrent writes
+ done := make(chan bool, numGoroutines)
+ for i := 0; i < numGoroutines; i++ {
+ go func(id int) {
+ defer func() { done <- true }()
+ for j := 0; j < operationsPerGoroutine; j++ {
+ writeReqs := []WriteRequest{
+ {
+ SeriesID: common.SeriesID(id + 1), // Start from 1, not 0
+ Key: int64(j),
+ Data: []byte(fmt.Sprintf("data_%d_%d", id, j)),
+ },
+ }
+ err := suite.Writer.Write(ctx, writeReqs)
+ assert.NoError(t, err)
+ }
+ }(i)
+ }
+
+ // Wait for all writes to complete
+ for i := 0; i < numGoroutines; i++ {
+ <-done
+ }
+
+ // Verify total elements
+ elements := suite.Writer.GetElements()
+ assert.Len(t, elements, numGoroutines*operationsPerGoroutine)
+
+ // Verify elements are properly sorted
+ for i := 1; i < len(elements); i++ {
+ prev := elements[i-1]
+ curr := elements[i]
+ if prev.SeriesID == curr.SeriesID {
+ assert.LessOrEqual(t, prev.Key, curr.Key)
+ } else {
+ assert.Less(t, prev.SeriesID, curr.SeriesID)
+ }
+ }
+}
+
+func TestMockComponents_PerformanceCharacteristics(t *testing.T) {
+ // This test documents the performance characteristics of mock components
+
+ suite := NewMockComponentSuite()
+ ctx := context.Background()
+
+ // Measure write performance
+ const numElements = 1000
+ var writeReqs []WriteRequest
+ for i := 0; i < numElements; i++ {
+ writeReqs = append(writeReqs, WriteRequest{
+ SeriesID: common.SeriesID(i%10 + 1),
+ Key: int64(i),
+ Data: []byte(fmt.Sprintf("data_%d", i)),
+ })
+ }
+
+ start := time.Now()
+ err := suite.Writer.Write(ctx, writeReqs)
+ require.NoError(t, err)
+ writeDuration := time.Since(start)
+
+ writeRate := float64(numElements) / writeDuration.Seconds()
+ t.Logf("Write performance: %d elements in %v (%.0f elem/sec)",
+ numElements, writeDuration, writeRate)
+
+ // Sync and measure query performance
+ suite.SyncElements()
+
+ queryReq := QueryRequest{
+ Name: "performance-test",
+ MaxElementSize: 100,
+ }
+
+ start = time.Now()
+ result, err := suite.Querier.Query(ctx, queryReq)
+ require.NoError(t, err)
+ defer result.Release()
+
+ resultCount := 0
+ for {
+ response := result.Pull()
+ if response == nil {
+ break
+ }
+ resultCount += response.Len()
+ }
+ queryDuration := time.Since(start)
+
+ queryRate := float64(resultCount) / queryDuration.Seconds()
+ t.Logf("Query performance: %d results in %v (%.0f results/sec)",
+ resultCount, queryDuration, queryRate)
+
+ // Document expected performance characteristics
+ assert.Greater(t, writeRate, 1000.0, "Write rate should be > 1000 elem/sec")
+ assert.Greater(t, queryRate, 5000.0, "Query rate should be > 5000 results/sec")
+ assert.Equal(t, numElements, resultCount, "All elements should be returned")
+}
\ No newline at end of file