Merge property buckets

adapters/handlers/rest/configure_api.go

@@ -42,6 +42,7 @@ import (
  "github.com/weaviate/weaviate/adapters/repos/classifications"
  "github.com/weaviate/weaviate/adapters/repos/db"
  "github.com/weaviate/weaviate/adapters/repos/db/inverted"
+ "github.com/weaviate/weaviate/adapters/repos/db/lsmkv"
  modulestorage "github.com/weaviate/weaviate/adapters/repos/modules"
  schemarepo "github.com/weaviate/weaviate/adapters/repos/schema"
  rCluster "github.com/weaviate/weaviate/cluster"
@@ -137,9 +138,18 @@ func getCores() (int, error) {
 }
 
 func MakeAppState(ctx context.Context, options *swag.CommandLineOptionsGroup) *state.State {
+ if os.Getenv("ENABLE_MERGED_PROPERTY_BUCKETS") != "" && strings.ToLower(os.Getenv("ENABLE_MERGED_PROPERTY_BUCKETS")) != "false" {
+ lsmkv.FeatureUseMergedBuckets = true
+ } else {
+ lsmkv.FeatureUseMergedBuckets = false
+ }
  appState := startupRoutine(ctx, options)
  setupGoProfiling(appState.ServerConfig.Config, appState.Logger)
 
+ if lsmkv.FeatureUseMergedBuckets {
+ appState.Logger.WithField("action", "startup").Warnf("Merged property buckets are enabled. This is an experimental feature and may be removed in the future. Please report any issues you encounter.")
+ }
+
  if appState.ServerConfig.Config.Monitoring.Enabled {
  appState.TenantActivity = tenantactivity.NewHandler()
 

adapters/repos/db/aggregations_integration_test.go

@@ -24,13 +24,25 @@ import (
  "github.com/sirupsen/logrus"
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
+ "github.com/weaviate/weaviate/adapters/repos/db/helpers"
  "github.com/weaviate/weaviate/entities/aggregation"
  "github.com/weaviate/weaviate/entities/filters"
  "github.com/weaviate/weaviate/entities/models"
  "github.com/weaviate/weaviate/entities/schema"
  "github.com/weaviate/weaviate/usecases/memwatch"
 )
 
+func Test_KeyHelpers(t *testing.T) {
+ prefix := []byte("testprefix")
+ data := []byte("testdata")
+
+ composite_key := helpers.MakePropertyKey(prefix, data)
+ assert.True(t, helpers.MatchesPropertyKey(prefix, composite_key))
+ recovered_key := helpers.UnMakePropertyKey(prefix, composite_key)
+
+ assert.Equal(t, data, recovered_key)
+}
+
 func Test_Aggregations(t *testing.T) {
  dirName := t.TempDir()
 

adapters/repos/db/aggregator/aggregator.go

@@ -20,6 +20,7 @@ import (
  "github.com/weaviate/weaviate/adapters/repos/db/helpers"
  "github.com/weaviate/weaviate/adapters/repos/db/inverted"
  "github.com/weaviate/weaviate/adapters/repos/db/inverted/stopwords"
+ "github.com/weaviate/weaviate/adapters/repos/db/inverted/tracker"
  "github.com/weaviate/weaviate/adapters/repos/db/lsmkv"
  "github.com/weaviate/weaviate/adapters/repos/db/roaringset"
  "github.com/weaviate/weaviate/entities/aggregation"
@@ -43,10 +44,13 @@ type Aggregator struct {
  vectorIndex vectorIndex
  stopwords stopwords.StopwordDetector
  shardVersion uint16
- propLenTracker *inverted.JsonPropertyLengthTracker
+ propLengths *inverted.JsonPropertyLengthTracker
+ propertyIds *tracker.JsonPropertyIdTracker
  isFallbackToSearchable inverted.IsFallbackToSearchable
  tenant string
  nestedCrossRefLimit int64
+ PropertyName string
+ propLenTracker *inverted.JsonPropertyLengthTracker
  bitmapFactory *roaringset.BitmapFactory
  modules *modules.Provider
 }
@@ -57,7 +61,7 @@ func New(store *lsmkv.Store, params aggregation.Params,
  vectorIndex vectorIndex, logger logrus.FieldLogger,
  propLenTracker *inverted.JsonPropertyLengthTracker,
  isFallbackToSearchable inverted.IsFallbackToSearchable,
- tenant string, nestedCrossRefLimit int64,
+ tenant string, propertyIds *tracker.JsonPropertyIdTracker, nestedCrossRefLimit int64,
  bitmapFactory *roaringset.BitmapFactory,
  modules *modules.Provider,
 ) *Aggregator {
@@ -73,6 +77,7 @@ func New(store *lsmkv.Store, params aggregation.Params,
  propLenTracker: propLenTracker,
  isFallbackToSearchable: isFallbackToSearchable,
  tenant: tenant,
+ propertyIds: propertyIds,
  nestedCrossRefLimit: nestedCrossRefLimit,
  bitmapFactory: bitmapFactory,
  modules: modules,

adapters/repos/db/aggregator/filtered.go

@@ -129,7 +129,7 @@ func (fa *filteredAggregator) bm25Objects(ctx context.Context, kw *searchparams.
 
  objs, scores, err := inverted.NewBM25Searcher(cfg.BM25, fa.store, fa.getSchema.ReadOnlyClass,
  propertyspecific.Indices{}, fa.classSearcher,
- fa.GetPropertyLengthTracker(), fa.logger, fa.shardVersion,
+ fa.GetPropertyLengthTracker(), fa.logger, fa.shardVersion, fa.propertyIds,
  ).BM25F(ctx, nil, fa.params.ClassName, *fa.params.ObjectLimit, *kw)
  if err != nil {
  return nil, nil, fmt.Errorf("bm25 objects: %w", err)

adapters/repos/db/aggregator/grouper.go

@@ -154,12 +154,16 @@ func (g *grouper) hybrid(ctx context.Context, allowList helpers.AllowList, modul
  return ids, nil
 }
 
-func (g *grouper) addElementById(s *models.PropertySchema, docID uint64) error {
- if s == nil {
+func (g *grouper) addElementById(propertySchemaP *models.PropertySchema, docID uint64) error {
+ if propertySchemaP == nil {
  return nil
  }
 
- item, ok := (*s).(map[string]interface{})[g.params.GroupBy.Property.String()]
+ key := g.params.GroupBy.Property.String()
+ S := *propertySchemaP
+ m := S.(map[string]interface{})
+ item, ok := m[key]
+
  if !ok {
  return nil
  }

adapters/repos/db/aggregator/hybrid.go

@@ -52,8 +52,7 @@ func (a *Aggregator) bm25Objects(ctx context.Context, kw *searchparams.KeywordRa
 
  objs, dists, err := inverted.NewBM25Searcher(cfg.BM25, a.store, a.getSchema.ReadOnlyClass,
  propertyspecific.Indices{}, a.classSearcher,
- a.GetPropertyLengthTracker(), a.logger, a.shardVersion,
- ).BM25F(ctx, nil, a.params.ClassName, *a.params.ObjectLimit, *kw)
+ a.GetPropertyLengthTracker(), a.logger, a.shardVersion, a.propertyIds).BM25F(ctx, nil, a.params.ClassName, *a.params.ObjectLimit, *kw)
  if err != nil {
  return nil, nil, fmt.Errorf("bm25 objects: %w", err)
  }

adapters/repos/db/aggregator/unfiltered_type_specific.go

@@ -31,9 +31,9 @@ func (ua unfilteredAggregator) boolProperty(ctx context.Context,
  Type: aggregation.PropertyTypeBoolean,
  }
 
- b := ua.store.Bucket(helpers.BucketFromPropNameLSM(prop.Name.String()))
- if b == nil {
- return nil, errors.Errorf("could not find bucket for prop %s", prop.Name)
+ b, err := lsmkv.FetchMeABucket(ua.store, "filterable_properties", helpers.BucketFromPropertyNameLSM(prop.Name.String()), prop.Name.String(), ua.propertyIds)
+ if err != nil {
+ return nil, errors.Errorf("could not create proxy bucket for prop %s: %v", prop.Name, err)
  }
 
  agg := newBoolAggregator()
@@ -44,6 +44,10 @@ func (ua unfilteredAggregator) boolProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+ k = helpers.UnMakePropertyKey(b.PropertyPrefix(), k)
  err := ua.parseAndAddBoolRowRoaringSet(agg, k, v)
  if err != nil {
  return nil, err
@@ -54,6 +58,10 @@ func (ua unfilteredAggregator) boolProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+ k = helpers.UnMakePropertyKey(b.PropertyPrefix(), k)
  err := ua.parseAndAddBoolRowSet(agg, k, v)
  if err != nil {
  return nil, err
@@ -84,6 +92,10 @@ func (ua unfilteredAggregator) boolArrayProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+
  err := ua.parseAndAddBoolArrayRow(agg, v, prop.Name)
  if err != nil {
  return nil, err
@@ -98,8 +110,7 @@ func (ua unfilteredAggregator) boolArrayProperty(ctx context.Context,
 func (ua unfilteredAggregator) parseAndAddBoolRowSet(agg *boolAggregator, k []byte, v [][]byte) error {
  if len(k) != 1 {
  // we expect to see a single byte for a marshalled bool
- return fmt.Errorf("unexpected key length on inverted index, "+
- "expected 1: got %d", len(k))
+ return fmt.Errorf("parseAndAddBoolRowSet: unexpected key length on inverted index, expected 1: got %d", len(k))
  }
 
  if err := agg.AddBoolRow(k, uint64(len(v))); err != nil {
@@ -112,8 +123,7 @@ func (ua unfilteredAggregator) parseAndAddBoolRowSet(agg *boolAggregator, k []by
 func (ua unfilteredAggregator) parseAndAddBoolRowRoaringSet(agg *boolAggregator, k []byte, v *sroar.Bitmap) error {
  if len(k) != 1 {
  // we expect to see a single byte for a marshalled bool
- return fmt.Errorf("unexpected key length on inverted index, "+
- "expected 1: got %d", len(k))
+ return fmt.Errorf("parseAndAddBoolRowRoaringSet: unexpected key length on inverted index, expected 1: got %d", len(k))
  }
 
  if err := agg.AddBoolRow(k, uint64(v.GetCardinality())); err != nil {
@@ -152,9 +162,9 @@ func (ua unfilteredAggregator) floatProperty(ctx context.Context,
  NumericalAggregations: map[string]interface{}{},
  }
 
- b := ua.store.Bucket(helpers.BucketFromPropNameLSM(prop.Name.String()))
- if b == nil {
- return nil, errors.Errorf("could not find bucket for prop %s", prop.Name)
+ b, err := lsmkv.FetchMeABucket(ua.store, "filterable_properties", helpers.BucketFromPropertyNameLSM(prop.Name.String()), prop.Name.String(), ua.propertyIds)
+ if err != nil {
+ return nil, errors.Errorf("could not create proxy bucket for prop %s: %v", prop.Name, err)
  }
 
  agg := newNumericalAggregator()
@@ -165,6 +175,10 @@ func (ua unfilteredAggregator) floatProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+ k = helpers.UnMakePropertyKey(b.PropertyPrefix(), k)
  if err := ua.parseAndAddFloatRowRoaringSet(agg, k, v); err != nil {
  return nil, err
  }
@@ -174,6 +188,10 @@ func (ua unfilteredAggregator) floatProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+ k = helpers.UnMakePropertyKey(b.PropertyPrefix(), k)
  if err := ua.parseAndAddFloatRowSet(agg, k, v); err != nil {
  return nil, err
  }
@@ -193,9 +211,9 @@ func (ua unfilteredAggregator) intProperty(ctx context.Context,
  NumericalAggregations: map[string]interface{}{},
  }
 
- b := ua.store.Bucket(helpers.BucketFromPropNameLSM(prop.Name.String()))
- if b == nil {
- return nil, errors.Errorf("could not find bucket for prop %s", prop.Name)
+ b, err := lsmkv.FetchMeABucket(ua.store, "filterable_properties", helpers.BucketFromPropertyNameLSM(prop.Name.String()), prop.Name.String(), ua.propertyIds)
+ if err != nil {
+ return nil, errors.Errorf("could not create proxy bucket for prop %s: %v", prop.Name, err)
  }
 
  agg := newNumericalAggregator()
@@ -206,6 +224,10 @@ func (ua unfilteredAggregator) intProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+ k = helpers.UnMakePropertyKey(b.PropertyPrefix(), k)
  if err := ua.parseAndAddIntRowRoaringSet(agg, k, v); err != nil {
  return nil, err
  }
@@ -216,6 +238,10 @@ func (ua unfilteredAggregator) intProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+ k = helpers.UnMakePropertyKey(b.PropertyPrefix(), k)
  if err := ua.parseAndAddIntRowSet(agg, k, v); err != nil {
  return nil, err
  }
@@ -235,9 +261,9 @@ func (ua unfilteredAggregator) dateProperty(ctx context.Context,
  DateAggregations: map[string]interface{}{},
  }
 
- b := ua.store.Bucket(helpers.BucketFromPropNameLSM(prop.Name.String()))
- if b == nil {
- return nil, errors.Errorf("could not find bucket for prop %s", prop.Name)
+ b, err := lsmkv.FetchMeABucket(ua.store, "filterable_properties", helpers.BucketFromPropertyNameLSM(prop.Name.String()), prop.Name.String(), ua.propertyIds)
+ if err != nil {
+ return nil, errors.Errorf("could not create proxy bucket for prop %s: %v", prop.Name, err)
  }
 
  agg := newDateAggregator()
@@ -247,7 +273,11 @@ func (ua unfilteredAggregator) dateProperty(ctx context.Context,
  c := b.CursorRoaringSet()
  defer c.Close()
 
- for k, v := c.First(); k != nil; k, v = c.Next() {
+ for key, v := c.First(); key != nil; key, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), key) {
+ continue
+ }
+ k := helpers.UnMakePropertyKey(b.PropertyPrefix(), key)
  if err := ua.parseAndAddDateRowRoaringSet(agg, k, v); err != nil {
  return nil, err
  }
@@ -257,6 +287,10 @@ func (ua unfilteredAggregator) dateProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+ k = helpers.UnMakePropertyKey(b.PropertyPrefix(), k)
  if err := ua.parseAndAddDateRowSet(agg, k, v); err != nil {
  return nil, err
  }
@@ -273,8 +307,7 @@ func (ua unfilteredAggregator) parseAndAddDateRowSet(agg *dateAggregator, k []by
 ) error {
  if len(k) != 8 {
  // dates are stored as epoch nanoseconds, we expect to see an int64
- return fmt.Errorf("unexpected key length on inverted index, "+
- "expected 8: got %d", len(k))
+ return fmt.Errorf("parseAndAddDateRowSet: unexpected key length on inverted index, expected 8: got %d", len(k))
  }
 
  if err := agg.AddTimestampRow(k, uint64(len(v))); err != nil {
@@ -289,8 +322,7 @@ func (ua unfilteredAggregator) parseAndAddDateRowRoaringSet(agg *dateAggregator,
 ) error {
  if len(k) != 8 {
  // dates are stored as epoch nanoseconds, we expect to see an int64
- return fmt.Errorf("unexpected key length on inverted index, "+
- "expected 8: got %d", len(k))
+ return fmt.Errorf("parseAndAddDateRowRoaringSet: unexpected key length on inverted index, expected 8: got %d (%v,%v)", len(k), k, string(k))
  }
 
  if err := agg.AddTimestampRow(k, uint64(v.GetCardinality())); err != nil {
@@ -319,6 +351,9 @@ func (ua unfilteredAggregator) dateArrayProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
  if err := ua.parseAndAddDateArrayRow(agg, v, prop.Name); err != nil {
  return nil, err
  }
@@ -356,8 +391,7 @@ func (ua unfilteredAggregator) parseAndAddFloatRowSet(agg *numericalAggregator,
  if len(k) != 8 {
  // we expect to see either an int64 or a float64, so any non-8 length
  // is unexpected
- return fmt.Errorf("unexpected key length on inverted index, "+
- "expected 8: got %d", len(k))
+ return fmt.Errorf("parseAndAddFloatRowSet: unexpected key length on inverted index, expected 8: got %d", len(k))
  }
 
  if err := agg.AddFloat64Row(k, uint64(len(v))); err != nil {
@@ -373,8 +407,7 @@ func (ua unfilteredAggregator) parseAndAddFloatRowRoaringSet(agg *numericalAggre
  if len(k) != 8 {
  // we expect to see either an int64 or a float64, so any non-8 length
  // is unexpected
- return fmt.Errorf("unexpected key length on inverted index, "+
- "expected 8: got %d", len(k))
+ return fmt.Errorf("parseAndAddFloatRowRoaringSet: unexpected key length on inverted index, expected 8: got %d", len(k))
  }
 
  if err := agg.AddFloat64Row(k, uint64(v.GetCardinality())); err != nil {
@@ -390,8 +423,7 @@ func (ua unfilteredAggregator) parseAndAddIntRowSet(agg *numericalAggregator, k
  if len(k) != 8 {
  // we expect to see either an int64 or a float64, so any non-8 length
  // is unexpected
- return fmt.Errorf("unexpected key length on inverted index, "+
- "expected 8: got %d", len(k))
+ return fmt.Errorf("parseAndAddIntRowSet: unexpected key length on inverted index, expected 8: got %d", len(k))
  }
 
  if err := agg.AddInt64Row(k, uint64(len(v))); err != nil {
@@ -407,8 +439,7 @@ func (ua unfilteredAggregator) parseAndAddIntRowRoaringSet(agg *numericalAggrega
  if len(k) != 8 {
  // we expect to see either an int64 or a float64, so any non-8 length
  // is unexpected
- return fmt.Errorf("unexpected key length on inverted index, "+
- "expected 8: got %d", len(k))
+ return fmt.Errorf("parseAndAddIntRowRoaringSet: unexpected key length on inverted index, expected 8: got %d", len(k))
  }
 
  if err := agg.AddInt64Row(k, uint64(v.GetCardinality())); err != nil {
@@ -463,6 +494,10 @@ func (ua unfilteredAggregator) textProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+
  if err := ua.parseAndAddTextRow(agg, v, prop.Name); err != nil {
  return nil, err
  }
@@ -492,6 +527,10 @@ func (ua unfilteredAggregator) numberArrayProperty(ctx context.Context,
  defer c.Close()
 
  for k, v := c.First(); k != nil; k, v = c.Next() {
+ if !helpers.MatchesPropertyKey(b.PropertyPrefix(), k) {
+ continue
+ }
+
  if err := ua.parseAndAddNumberArrayRow(agg, v, prop.Name); err != nil {
  return nil, err
  }