Merge stable/v1.25 into main

adapters/handlers/grpc/v1/parse_search_request.go

@@ -368,12 +368,6 @@ func extractTargetVectors(req *pb.SearchRequest, class *models.Class) (*[]string
  var targetVectors *[]string
  if hs := req.HybridSearch; hs != nil {
  targetVectors = &hs.TargetVectors
- if hs.NearText != nil {
- targetVectors = &hs.NearText.TargetVectors
- }
- if hs.NearVector != nil {
- targetVectors = &hs.NearVector.TargetVectors
- }
  }
  if na := req.NearAudio; na != nil {
  targetVectors = &na.TargetVectors

adapters/handlers/grpc/v1/parse_search_request_test.go

@@ -204,11 +204,11 @@ func TestGRPCRequest(t *testing.T) {
  Alpha: 1.0,
  Query: "nearvecquery",
  NearVector: &pb.NearVector{
- VectorBytes: byteops.Float32ToByteVector([]float32{1, 2, 3}),
- TargetVectors: []string{"custom"},
- Certainty: &one,
- Distance: &one,
+ VectorBytes: byteops.Float32ToByteVector([]float32{1, 2, 3}),
+ Certainty: &one,
+ Distance: &one,
  },
+ TargetVectors: []string{"custom"},
  },
  },
  out: dto.GetParams{
@@ -221,12 +221,12 @@ func TestGRPCRequest(t *testing.T) {
  Query: "nearvecquery",
  FusionAlgorithm: 1,
  NearVectorParams: &searchparams.NearVector{
- Vector: []float32{1, 2, 3},
- TargetVectors: []string{"custom"},
- Certainty: 1.0,
- Distance: 1.0,
- WithDistance: true,
+ Vector: []float32{1, 2, 3},
+ Certainty: 1.0,
+ Distance: 1.0,
+ WithDistance: true,
  },
+ TargetVectors: []string{"custom"},
  },
  },
  error: false,

adapters/handlers/grpc/v1/tenants.go

@@ -27,7 +27,7 @@ func (s *Service) tenantsGet(ctx context.Context, principal *models.Principal, r
  var err error
  var tenants []*models.Tenant
  if req.Params == nil {
- tenants, err = s.schemaManager.GetConsistentTenants(ctx, principal, req.Collection, req.IsConsistent, []string{})
+ tenants, err = s.schemaManager.GetConsistentTenants(ctx, principal, req.Collection, true, []string{})
  if err != nil {
  return nil, err
  }
@@ -38,7 +38,7 @@ func (s *Service) tenantsGet(ctx context.Context, principal *models.Principal, r
  if len(requestedNames) == 0 {
  return nil, fmt.Errorf("must specify at least one tenant name")
  }
- tenants, err = s.schemaManager.GetConsistentTenants(ctx, principal, req.Collection, req.IsConsistent, requestedNames)
+ tenants, err = s.schemaManager.GetConsistentTenants(ctx, principal, req.Collection, true, requestedNames)
  if err != nil {
  return nil, err
  }

adapters/handlers/rest/configure_api.go

@@ -193,6 +193,8 @@ func MakeAppState(ctx context.Context, options *swag.CommandLineOptionsGroup) *s
  MemtablesMaxSizeMB: appState.ServerConfig.Config.Persistence.MemtablesMaxSizeMB,
  MemtablesMinActiveSeconds: appState.ServerConfig.Config.Persistence.MemtablesMinActiveDurationSeconds,
  MemtablesMaxActiveSeconds: appState.ServerConfig.Config.Persistence.MemtablesMaxActiveDurationSeconds,
+ MaxSegmentSize: appState.ServerConfig.Config.Persistence.LSMMaxSegmentSize,
+ HNSWMaxLogSize: appState.ServerConfig.Config.Persistence.HNSWMaxLogSize,
  RootPath: appState.ServerConfig.Config.Persistence.DataPath,
  QueryLimit: appState.ServerConfig.Config.QueryDefaults.Limit,
  QueryMaximumResults: appState.ServerConfig.Config.QueryMaximumResults,
@@ -251,8 +253,6 @@ func MakeAppState(ctx context.Context, options *swag.CommandLineOptionsGroup) *s
  remoteIndexClient, appState.Logger, appState.ServerConfig.Config.Persistence.DataPath)
  appState.Scaler = scaler
 
- /// TODO-RAFT START
- //
  server2port, err := parseNode2Port(appState)
  if len(server2port) == 0 || err != nil {
  appState.Logger.
@@ -268,31 +268,31 @@ func MakeAppState(ctx context.Context, options *swag.CommandLineOptionsGroup) *s
  dataPath := appState.ServerConfig.Config.Persistence.DataPath
 
  rConfig := rStore.Config{
- WorkDir: filepath.Join(dataPath, "raft"),
- NodeID: nodeName,
- Host: addrs[0],
- RaftPort: appState.ServerConfig.Config.Raft.Port,
- RPCPort: appState.ServerConfig.Config.Raft.InternalRPCPort,
- RaftRPCMessageMaxSize: appState.ServerConfig.Config.Raft.RPCMessageMaxSize,
- ServerName2PortMap: server2port,
- BootstrapTimeout: appState.ServerConfig.Config.Raft.BootstrapTimeout,
- BootstrapExpect: appState.ServerConfig.Config.Raft.BootstrapExpect,
- HeartbeatTimeout: appState.ServerConfig.Config.Raft.HeartbeatTimeout,
- RecoveryTimeout: appState.ServerConfig.Config.Raft.RecoveryTimeout,
- ElectionTimeout: appState.ServerConfig.Config.Raft.ElectionTimeout,
- SnapshotInterval: appState.ServerConfig.Config.Raft.SnapshotInterval,
- SnapshotThreshold: appState.ServerConfig.Config.Raft.SnapshotThreshold,
- UpdateWaitTimeout: time.Second * 10, // TODO-RAFT read from the flag
- MetadataOnlyVoters: appState.ServerConfig.Config.Raft.MetadataOnlyVoters,
- DB: nil,
- Parser: schema.NewParser(appState.Cluster, vectorIndex.ParseAndValidateConfig, migrator),
- AddrResolver: appState.Cluster,
- Logger: appState.Logger,
- LogLevel: logLevel(),
- LogJSONFormat: !logTextFormat(),
- IsLocalHost: appState.ServerConfig.Config.Cluster.Localhost,
- LoadLegacySchema: schemaRepo.LoadLegacySchema,
- SaveLegacySchema: schemaRepo.SaveLegacySchema,
+ WorkDir:  filepath.Join(dataPath, config.DefaultRaftDir),
+ NodeID:  nodeName,
+ Host:  addrs[0],
+ RaftPort:  appState.ServerConfig.Config.Raft.Port,
+ RPCPort:  appState.ServerConfig.Config.Raft.InternalRPCPort,
+ RaftRPCMessageMaxSize:  appState.ServerConfig.Config.Raft.RPCMessageMaxSize,
+ ServerName2PortMap:  server2port,
+ BootstrapTimeout:  appState.ServerConfig.Config.Raft.BootstrapTimeout,
+ BootstrapExpect:  appState.ServerConfig.Config.Raft.BootstrapExpect,
+ HeartbeatTimeout:  appState.ServerConfig.Config.Raft.HeartbeatTimeout,
+ RecoveryTimeout:  appState.ServerConfig.Config.Raft.RecoveryTimeout,
+ ElectionTimeout:  appState.ServerConfig.Config.Raft.ElectionTimeout,
+ SnapshotInterval:  appState.ServerConfig.Config.Raft.SnapshotInterval,
+ SnapshotThreshold:  appState.ServerConfig.Config.Raft.SnapshotThreshold,
+ ConsistencyWaitTimeout: appState.ServerConfig.Config.Raft.ConsistencyWaitTimeout,
+ MetadataOnlyVoters:  appState.ServerConfig.Config.Raft.MetadataOnlyVoters,
+ DB:  nil,
+ Parser:  schema.NewParser(appState.Cluster, vectorIndex.ParseAndValidateConfig, migrator),
+ AddrResolver:  appState.Cluster,
+ Logger:  appState.Logger,
+ LogLevel:  logLevel(),
+ LogJSONFormat:  !logTextFormat(),
+ IsLocalHost:  appState.ServerConfig.Config.Cluster.Localhost,
+ LoadLegacySchema:  schemaRepo.LoadLegacySchema,
+ SaveLegacySchema:  schemaRepo.SaveLegacySchema,
  }
  for _, name := range appState.ServerConfig.Config.Raft.Join[:rConfig.BootstrapExpect] {
  if strings.Contains(name, rConfig.NodeID) {

adapters/repos/db/helper_for_test.go

@@ -294,7 +294,7 @@ func testObject(className string) *storobj.Object {
  }
 }
 
-func createRandomObjects(r *rand.Rand, className string, numObj int) []*storobj.Object {
+func createRandomObjects(r *rand.Rand, className string, numObj int, vectorDim int) []*storobj.Object {
  obj := make([]*storobj.Object, numObj)
 
  for i := 0; i < numObj; i++ {
@@ -304,7 +304,11 @@ func createRandomObjects(r *rand.Rand, className string, numObj int) []*storobj.
  ID: strfmt.UUID(uuid.NewString()),
  Class: className,
  },
- Vector: []float32{r.Float32(), r.Float32(), r.Float32(), r.Float32()},
+ Vector: make([]float32, vectorDim),
+ }
+
+ for d := 0; d < vectorDim; d++ {
+ obj[i].Vector[d] = r.Float32()
  }
  }
  return obj

adapters/repos/db/index.go

@@ -555,6 +555,8 @@ type IndexConfig struct {
  MemtablesMaxSizeMB int
  MemtablesMinActiveSeconds int
  MemtablesMaxActiveSeconds int
+ MaxSegmentSize int64
+ HNSWMaxLogSize int64
  ReplicationFactor int64
  AvoidMMap bool
  DisableLazyLoadShards bool

adapters/repos/db/init.go

@@ -89,6 +89,8 @@ func (db *DB) init(ctx context.Context) error {
  MemtablesMaxSizeMB: db.config.MemtablesMaxSizeMB,
  MemtablesMinActiveSeconds: db.config.MemtablesMinActiveSeconds,
  MemtablesMaxActiveSeconds: db.config.MemtablesMaxActiveSeconds,
+ MaxSegmentSize: db.config.MaxSegmentSize,
+ HNSWMaxLogSize: db.config.HNSWMaxLogSize,
  TrackVectorDimensions: db.config.TrackVectorDimensions,
  AvoidMMap: db.config.AvoidMMap,
  DisableLazyLoadShards: db.config.DisableLazyLoadShards,

adapters/repos/db/lsmkv/bucket.go

@@ -118,6 +118,10 @@ type Bucket struct {
  // optionally supplied to prevent starting memory-intensive
  // processes when memory pressure is high
  allocChecker memwatch.AllocChecker
+
+ // optional segment size limit. If set, a compaction will skip segments that
+ // sum to more than the specified value.
+ maxSegmentSize int64
 }
 
 func NewBucketCreator() *Bucket { return &Bucket{} }
@@ -178,6 +182,7 @@ func (*Bucket) NewBucket(ctx context.Context, dir, rootDir string, logger logrus
  forceCompaction: b.forceCompaction,
  useBloomFilter: b.useBloomFilter,
  calcCountNetAdditions: b.calcCountNetAdditions,
+ maxSegmentSize: b.maxSegmentSize,
  }, b.allocChecker)
  if err != nil {
  return nil, fmt.Errorf("init disk segments: %w", err)

adapters/repos/db/lsmkv/bucket_options.go

@@ -148,6 +148,13 @@ func WithCalcCountNetAdditions(calcCountNetAdditions bool) BucketOption {
  }
 }
 
+func WithMaxSegmentSize(maxSegmentSize int64) BucketOption {
+ return func(b *Bucket) error {
+ b.maxSegmentSize = maxSegmentSize
+ return nil
+ }
+}
+
 /*
 Background for this option:
 

adapters/repos/db/lsmkv/segment_group.go

@@ -62,7 +62,8 @@ type SegmentGroup struct {
  calcCountNetAdditions bool // see bucket for more datails
  compactLeftOverSegments bool // see bucket for more datails
 
- allocChecker memwatch.AllocChecker
+ allocChecker memwatch.AllocChecker
+ maxSegmentSize int64
 }
 
 type sgConfig struct {
@@ -75,6 +76,7 @@ type sgConfig struct {
  useBloomFilter bool
  calcCountNetAdditions bool
  forceCompaction bool
+ maxSegmentSize int64
 }
 
 func newSegmentGroup(logger logrus.FieldLogger, metrics *Metrics,
@@ -99,6 +101,7 @@ func newSegmentGroup(logger logrus.FieldLogger, metrics *Metrics,
  useBloomFilter: cfg.useBloomFilter,
  calcCountNetAdditions: cfg.calcCountNetAdditions,
  compactLeftOverSegments: cfg.forceCompaction,
+ maxSegmentSize: cfg.maxSegmentSize,
  allocChecker: allocChecker,
  }
 

adapters/repos/db/lsmkv/segment_group_compaction.go

@@ -146,6 +146,12 @@ func (sg *SegmentGroup) compactOnce() (bool, error) {
  leftSegment := sg.segmentAtPos(pair[0])
  rightSegment := sg.segmentAtPos(pair[1])
 
+ if !sg.compactionFitsSizeLimit(leftSegment, rightSegment) {
+ // nothing to do this round, let's wait for the next round in the hopes
+ // that we'll find smaller (lower-level) segments that can still fit.
+ return false, nil
+ }
+
  path := filepath.Join(sg.dir, "segment-"+segmentID(leftSegment.path)+"_"+segmentID(rightSegment.path)+".db.tmp")
 
  f, err := os.Create(path)
@@ -470,3 +476,13 @@ func (s *segmentLevelStats) report(metrics *Metrics,
  }).Set(float64(count))
  }
 }
+
+func (sg *SegmentGroup) compactionFitsSizeLimit(left, right *segment) bool {
+ if sg.maxSegmentSize == 0 {
+ // no limit is set, always return true
+ return true
+ }
+
+ totalSize := left.size + right.size
+ return totalSize <= sg.maxSegmentSize
+}

adapters/repos/db/lsmkv/segment_group_compaction_test.go

@@ -0,0 +1,101 @@
+// _ _
+// __ _____ __ ___ ___ __ _| |_ ___
+// \ \ /\ / / _ \/ _` \ \ / / |/ _` | __/ _ \
+// \ V V / __/ (_| |\ V /| | (_| | || __/
+// \_/\_/ \___|\__,_| \_/ |_|\__,_|\__\___|
+//
+// Copyright © 2016 - 2024 Weaviate B.V. All rights reserved.
+//
+// CONTACT: [email protected]
+//
+
+package lsmkv
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestSegmentGroup_BestCompactionPair(t *testing.T) {
+ var maxSegmentSize int64 = 10000
+
+ tests := []struct {
+ name string
+ segments []*segment
+ expectedPair []string
+ }{
+ {
+ name: "single segment",
+ segments: []*segment{
+ {size: 1000, path: "segment0", level: 0},
+ },
+ expectedPair: nil,
+ },
+ {
+ name: "two segments, same level",
+ segments: []*segment{
+ {size: 1000, path: "segment0", level: 0},
+ {size: 1000, path: "segment1", level: 0},
+ },
+ expectedPair: []string{"segment0", "segment1"},
+ },
+ {
+ name: "multiple segments, multiple levels, lowest level is picked",
+ segments: []*segment{
+ {size: 4000, path: "segment0", level: 2},
+ {size: 4000, path: "segment1", level: 2},
+ {size: 2000, path: "segment2", level: 1},
+ {size: 2000, path: "segment3", level: 1},
+ {size: 1000, path: "segment4", level: 0},
+ {size: 1000, path: "segment5", level: 0},
+ },
+ expectedPair: []string{"segment4", "segment5"},
+ },
+ {
+ name: "two segments that don't fit the max size, but eliglbe segments of a lower level are present",
+ segments: []*segment{
+ {size: 8000, path: "segment0", level: 3},
+ {size: 8000, path: "segment1", level: 3},
+ {size: 4000, path: "segment2", level: 2},
+ {size: 4000, path: "segment3", level: 2},
+ },
+ expectedPair: []string{"segment2", "segment3"},
+ },
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ sg := &SegmentGroup{
+ segments: test.segments,
+ maxSegmentSize: maxSegmentSize,
+ }
+ pair := sg.bestCompactionCandidatePair()
+ if test.expectedPair == nil {
+ assert.Nil(t, pair)
+ } else {
+ leftPath := test.segments[pair[0]].path
+ rightPath := test.segments[pair[1]].path
+ assert.Equal(t, test.expectedPair, []string{leftPath, rightPath})
+ }
+ })
+ }
+}
+
+func TestSegmenGroup_CompactionLargerThanMaxSize(t *testing.T) {
+ maxSegmentSize := int64(10000)
+ // this test only tests the unhappy path which has an early exist condition,
+ // meaning we don't need real segments, it is only metadata that is evaluated
+ // here.
+ sg := &SegmentGroup{
+ segments: []*segment{
+ {size: 8000, path: "segment0", level: 3},
+ {size: 8000, path: "segment1", level: 3},
+ },
+ maxSegmentSize: maxSegmentSize,
+ }
+
+ ok, err := sg.compactOnce()
+ assert.False(t, ok, "segments are too large to run")
+ assert.Nil(t, err)
+}