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/19 23:56:04 UTC
(skywalking-banyandb) 05/10: Add SIDX options and validation tests
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch sidx/interface
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 5dc99b3989d6e5b562cf16eb6efdccfd96f5cd6b
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Aug 19 09:32:16 2025 +0700
Add SIDX options and validation tests
- Introduced new `Options` and `MergePolicy` structs to manage configuration for the Secondary Index File System (SIDX).
- Implemented methods for creating default options and validating configurations.
---
banyand/internal/sidx/TODO.md | 20 +--
banyand/internal/sidx/interfaces.go | 2 +-
banyand/internal/sidx/options.go | 146 ++++++++++++++++++
banyand/internal/sidx/options_test.go | 271 ++++++++++++++++++++++++++++++++++
4 files changed, 428 insertions(+), 11 deletions(-)
diff --git a/banyand/internal/sidx/TODO.md b/banyand/internal/sidx/TODO.md
index f7019aee..aa5dcd03 100644
--- a/banyand/internal/sidx/TODO.md
+++ b/banyand/internal/sidx/TODO.md
@@ -5,7 +5,7 @@ This document tracks the implementation progress of the Secondary Index File Sys
## Implementation Progress Overview
- [x] **Phase 1**: Core Data Structures (6 tasks) - 6/6 completed ✅
-- [x] **Phase 2**: Interface Definitions (5 tasks) - 3/5 completed ✅ **CORE INTERFACES READY**
+- [x] **Phase 2**: Interface Definitions (5 tasks) - 4/5 completed ✅ **CORE INTERFACES READY**
- [ ] **Phase 3**: Mock Implementations (4 tasks) 🔥 **NEW - FOR EARLY TESTING**
- [ ] **Phase 4**: Memory Management (4 tasks)
- [ ] **Phase 5**: Snapshot Management (4 tasks)
@@ -131,15 +131,15 @@ This document tracks the implementation progress of the Secondary Index File Sys
- [x] Type safety is maintained across operations
- [x] Memory pooling integration is ready
-### 2.4 Configuration Interfaces (`options.go`)
-- [ ] Define Options struct for SIDX configuration
-- [ ] Add path where the files are put.
-- [ ] Add protector.Memory to control the resource limit.
-- [ ] Add *mergePolicy to control merge behaviour.
-- [ ] **Test Cases**:
- - [ ] Default configurations are sensible
- - [ ] Configuration validation works correctly
- - [ ] Options can be merged and overridden
+### 2.4 Configuration Interfaces (`options.go`) ✅
+- [x] Define Options struct for SIDX configuration
+- [x] Add path where the files are put.
+- [x] Add protector.Memory to control the resource limit.
+- [x] Add *mergePolicy to control merge behaviour.
+- [x] **Test Cases**:
+ - [x] Default configurations are sensible
+ - [x] Configuration validation works correctly
+ - [x] Options can be merged and overridden
### 2.5 Interface Documentation and Examples (`interfaces_examples.go`)
- [ ] Create comprehensive interface usage examples
diff --git a/banyand/internal/sidx/interfaces.go b/banyand/internal/sidx/interfaces.go
index ccfccfbe..ddd33c4d 100644
--- a/banyand/internal/sidx/interfaces.go
+++ b/banyand/internal/sidx/interfaces.go
@@ -521,7 +521,7 @@ func (qr *QueryRequest) CopyFrom(other *QueryRequest) {
// },
// }
-// Interface Design Principles
+// Interface Design Principles.
//
// 1. **Single Responsibility**: Each interface has a focused, well-defined purpose
// 2. **Minimal Surface Area**: Interfaces expose only essential methods
diff --git a/banyand/internal/sidx/options.go b/banyand/internal/sidx/options.go
new file mode 100644
index 00000000..c1daf36d
--- /dev/null
+++ b/banyand/internal/sidx/options.go
@@ -0,0 +1,146 @@
+// 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 (
+ "fmt"
+ "path/filepath"
+
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+)
+
+// Options contains configuration options for the Secondary Index File System (SIDX).
+type Options struct {
+ // MergePolicy controls the merge behavior for parts.
+ // It determines which parts should be merged together to optimize
+ // storage efficiency and query performance.
+ MergePolicy *MergePolicy
+
+ // Memory is the memory protector that controls resource limits.
+ // It monitors memory usage and prevents OOM by applying backpressure
+ // when memory consumption exceeds configured thresholds.
+ Memory protector.Memory
+
+ // Path is the directory where SIDX files are stored.
+ // This includes part directories, metadata files, and temporary files.
+ Path string
+}
+
+// MergePolicy defines the strategy for merging parts in SIDX.
+// It aims to choose an optimal combination that has the lowest write amplification.
+type MergePolicy struct {
+ // MaxParts is the maximum number of parts that can be merged in a single operation.
+ // Higher values allow more aggressive merging but consume more resources.
+ MaxParts int
+
+ // MinMergeMultiplier is the minimum ratio between output size and largest input part size
+ // required to proceed with a merge. This prevents merging parts with too small benefit.
+ MinMergeMultiplier float64
+
+ // MaxFanOutSize is the maximum total size of parts that can be merged together.
+ // This limits the memory and disk I/O requirements of merge operations.
+ MaxFanOutSize uint64
+}
+
+// NewDefaultOptions creates Options with sensible default values.
+func NewDefaultOptions() *Options {
+ return &Options{
+ Path: "/tmp/sidx",
+ Memory: nil, // Must be provided by caller
+ MergePolicy: NewDefaultMergePolicy(),
+ }
+}
+
+// NewDefaultMergePolicy creates a MergePolicy with default parameters optimized
+// for typical SIDX workloads.
+func NewDefaultMergePolicy() *MergePolicy {
+ return &MergePolicy{
+ MaxParts: 8,
+ MinMergeMultiplier: 1.7,
+ MaxFanOutSize: 1 << 30, // 1GB
+ }
+}
+
+// NewMergePolicy creates a MergePolicy with custom parameters.
+func NewMergePolicy(maxParts int, minMergeMultiplier float64, maxFanOutSize uint64) *MergePolicy {
+ return &MergePolicy{
+ MaxParts: maxParts,
+ MinMergeMultiplier: minMergeMultiplier,
+ MaxFanOutSize: maxFanOutSize,
+ }
+}
+
+// Validate validates the options and returns an error if any configuration is invalid.
+func (o *Options) Validate() error {
+ if o.Path == "" {
+ return fmt.Errorf("path must not be empty")
+ }
+
+ if !filepath.IsAbs(o.Path) {
+ return fmt.Errorf("path must be absolute, got: %s", o.Path)
+ }
+
+ if o.Memory == nil {
+ return fmt.Errorf("memory protector must not be nil")
+ }
+
+ if o.MergePolicy == nil {
+ return fmt.Errorf("merge policy must not be nil")
+ }
+
+ return o.MergePolicy.Validate()
+}
+
+// Validate validates the merge policy configuration.
+func (mp *MergePolicy) Validate() error {
+ if mp.MaxParts < 2 {
+ return fmt.Errorf("maxParts must be at least 2, got: %d", mp.MaxParts)
+ }
+
+ if mp.MinMergeMultiplier <= 1.0 {
+ return fmt.Errorf("minMergeMultiplier must be greater than 1.0, got: %f", mp.MinMergeMultiplier)
+ }
+
+ if mp.MaxFanOutSize == 0 {
+ return fmt.Errorf("maxFanOutSize must be greater than 0")
+ }
+
+ return nil
+}
+
+// WithPath returns a copy of the options with the specified path.
+func (o *Options) WithPath(path string) *Options {
+ opts := *o
+ opts.Path = path
+ return &opts
+}
+
+// WithMemory returns a copy of the options with the specified memory protector.
+func (o *Options) WithMemory(memory protector.Memory) *Options {
+ opts := *o
+ opts.Memory = memory
+ return &opts
+}
+
+// WithMergePolicy returns a copy of the options with the specified merge policy.
+func (o *Options) WithMergePolicy(policy *MergePolicy) *Options {
+ opts := *o
+ opts.MergePolicy = policy
+ return &opts
+}
+
diff --git a/banyand/internal/sidx/options_test.go b/banyand/internal/sidx/options_test.go
new file mode 100644
index 00000000..1967096f
--- /dev/null
+++ b/banyand/internal/sidx/options_test.go
@@ -0,0 +1,271 @@
+// 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 (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+)
+
+func TestNewDefaultOptions(t *testing.T) {
+ opts := NewDefaultOptions()
+
+ assert.Equal(t, "/tmp/sidx", opts.Path)
+ assert.Nil(t, opts.Memory)
+ assert.NotNil(t, opts.MergePolicy)
+
+ // Verify default merge policy values
+ assert.Equal(t, 8, opts.MergePolicy.MaxParts)
+ assert.Equal(t, 1.7, opts.MergePolicy.MinMergeMultiplier)
+ assert.Equal(t, uint64(1<<30), opts.MergePolicy.MaxFanOutSize)
+}
+
+func TestNewDefaultMergePolicy(t *testing.T) {
+ policy := NewDefaultMergePolicy()
+
+ assert.Equal(t, 8, policy.MaxParts)
+ assert.Equal(t, 1.7, policy.MinMergeMultiplier)
+ assert.Equal(t, uint64(1<<30), policy.MaxFanOutSize)
+
+ // Ensure it validates successfully
+ assert.NoError(t, policy.Validate())
+}
+
+func TestNewMergePolicy(t *testing.T) {
+ policy := NewMergePolicy(10, 2.0, 1<<20)
+
+ assert.Equal(t, 10, policy.MaxParts)
+ assert.Equal(t, 2.0, policy.MinMergeMultiplier)
+ assert.Equal(t, uint64(1<<20), policy.MaxFanOutSize)
+
+ assert.NoError(t, policy.Validate())
+}
+
+func TestOptionsValidation(t *testing.T) {
+ mockMemory := protector.NewMemory(observability.NewBypassRegistry())
+
+ tests := []struct {
+ name string
+ opts *Options
+ expectError bool
+ errorMsg string
+ }{
+ {
+ name: "valid options",
+ opts: &Options{
+ Path: "/tmp/sidx",
+ Memory: mockMemory,
+ MergePolicy: NewDefaultMergePolicy(),
+ },
+ expectError: false,
+ },
+ {
+ name: "empty path",
+ opts: &Options{
+ Path: "",
+ Memory: mockMemory,
+ MergePolicy: NewDefaultMergePolicy(),
+ },
+ expectError: true,
+ errorMsg: "path must not be empty",
+ },
+ {
+ name: "relative path",
+ opts: &Options{
+ Path: "relative/path",
+ Memory: mockMemory,
+ MergePolicy: NewDefaultMergePolicy(),
+ },
+ expectError: true,
+ errorMsg: "path must be absolute",
+ },
+ {
+ name: "nil memory protector",
+ opts: &Options{
+ Path: "/tmp/sidx",
+ Memory: nil,
+ MergePolicy: NewDefaultMergePolicy(),
+ },
+ expectError: true,
+ errorMsg: "memory protector must not be nil",
+ },
+ {
+ name: "nil merge policy",
+ opts: &Options{
+ Path: "/tmp/sidx",
+ Memory: mockMemory,
+ MergePolicy: nil,
+ },
+ expectError: true,
+ errorMsg: "merge policy must not be nil",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ err := tt.opts.Validate()
+ if tt.expectError {
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), tt.errorMsg)
+ } else {
+ assert.NoError(t, err)
+ }
+ })
+ }
+}
+
+func TestMergePolicyValidation(t *testing.T) {
+ tests := []struct {
+ name string
+ policy *MergePolicy
+ expectError bool
+ errorMsg string
+ }{
+ {
+ name: "valid merge policy",
+ policy: &MergePolicy{
+ MaxParts: 8,
+ MinMergeMultiplier: 1.7,
+ MaxFanOutSize: 1 << 30,
+ },
+ expectError: false,
+ },
+ {
+ name: "maxParts too small",
+ policy: &MergePolicy{
+ MaxParts: 1,
+ MinMergeMultiplier: 1.7,
+ MaxFanOutSize: 1 << 30,
+ },
+ expectError: true,
+ errorMsg: "maxParts must be at least 2",
+ },
+ {
+ name: "minMergeMultiplier too small",
+ policy: &MergePolicy{
+ MaxParts: 8,
+ MinMergeMultiplier: 1.0,
+ MaxFanOutSize: 1 << 30,
+ },
+ expectError: true,
+ errorMsg: "minMergeMultiplier must be greater than 1.0",
+ },
+ {
+ name: "zero maxFanOutSize",
+ policy: &MergePolicy{
+ MaxParts: 8,
+ MinMergeMultiplier: 1.7,
+ MaxFanOutSize: 0,
+ },
+ expectError: true,
+ errorMsg: "maxFanOutSize must be greater than 0",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ err := tt.policy.Validate()
+ if tt.expectError {
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), tt.errorMsg)
+ } else {
+ assert.NoError(t, err)
+ }
+ })
+ }
+}
+
+func TestOptionsWithMethods(t *testing.T) {
+ mockMemory := protector.NewMemory(observability.NewBypassRegistry())
+ originalOpts := NewDefaultOptions()
+
+ t.Run("WithPath", func(t *testing.T) {
+ newPath := "/custom/path"
+ newOpts := originalOpts.WithPath(newPath)
+
+ assert.Equal(t, newPath, newOpts.Path)
+ assert.Equal(t, originalOpts.Path, "/tmp/sidx") // Original unchanged
+ assert.Equal(t, originalOpts.MergePolicy, newOpts.MergePolicy) // Other fields copied
+ })
+
+ t.Run("WithMemory", func(t *testing.T) {
+ newOpts := originalOpts.WithMemory(mockMemory)
+
+ assert.Equal(t, mockMemory, newOpts.Memory)
+ assert.Nil(t, originalOpts.Memory) // Original unchanged
+ assert.Equal(t, originalOpts.Path, newOpts.Path) // Other fields copied
+ })
+
+ t.Run("WithMergePolicy", func(t *testing.T) {
+ customPolicy := NewMergePolicy(10, 2.0, 1<<20)
+ newOpts := originalOpts.WithMergePolicy(customPolicy)
+
+ assert.Equal(t, customPolicy, newOpts.MergePolicy)
+ assert.NotEqual(t, customPolicy, originalOpts.MergePolicy) // Original unchanged
+ assert.Equal(t, originalOpts.Path, newOpts.Path) // Other fields copied
+ })
+}
+
+func TestOptionsConfiguration(t *testing.T) {
+ mockMemory := protector.NewMemory(observability.NewBypassRegistry())
+
+ // Test default configurations are sensible
+ defaultOpts := NewDefaultOptions()
+ require.NotNil(t, defaultOpts.MergePolicy)
+
+ // Test configuration can be merged and overridden
+ customOpts := defaultOpts.
+ WithPath("/custom/sidx").
+ WithMemory(mockMemory).
+ WithMergePolicy(NewMergePolicy(5, 1.5, 512<<20))
+
+ assert.Equal(t, "/custom/sidx", customOpts.Path)
+ assert.Equal(t, mockMemory, customOpts.Memory)
+ assert.Equal(t, 5, customOpts.MergePolicy.MaxParts)
+ assert.Equal(t, 1.5, customOpts.MergePolicy.MinMergeMultiplier)
+ assert.Equal(t, uint64(512<<20), customOpts.MergePolicy.MaxFanOutSize)
+
+ // Ensure validation works with custom configuration
+ assert.NoError(t, customOpts.Validate())
+}
+
+func TestOptionsEdgeCases(t *testing.T) {
+ t.Run("minimum valid merge policy", func(t *testing.T) {
+ policy := &MergePolicy{
+ MaxParts: 2,
+ MinMergeMultiplier: 1.001,
+ MaxFanOutSize: 1,
+ }
+ assert.NoError(t, policy.Validate())
+ })
+
+ t.Run("large values", func(t *testing.T) {
+ policy := &MergePolicy{
+ MaxParts: 1000,
+ MinMergeMultiplier: 100.0,
+ MaxFanOutSize: 1 << 60,
+ }
+ assert.NoError(t, policy.Validate())
+ })
+}
+