Spaces:
Running
Running
| // _ _ | |
| // __ _____ __ ___ ___ __ _| |_ ___ | |
| // \ \ /\ / / _ \/ _` \ \ / / |/ _` | __/ _ \ | |
| // \ V V / __/ (_| |\ V /| | (_| | || __/ | |
| // \_/\_/ \___|\__,_| \_/ |_|\__,_|\__\___| | |
| // | |
| // Copyright © 2016 - 2024 Weaviate B.V. All rights reserved. | |
| // | |
| // CONTACT: [email protected] | |
| // | |
| //go:build integrationTest | |
| // +build integrationTest | |
| package lsmkv | |
| import ( | |
| "context" | |
| "fmt" | |
| "math/rand" | |
| "testing" | |
| "github.com/stretchr/testify/assert" | |
| "github.com/stretchr/testify/require" | |
| "github.com/weaviate/weaviate/entities/cyclemanager" | |
| ) | |
| func compactionSetStrategy(ctx context.Context, t *testing.T, opts []BucketOption, | |
| expectedMinSize, expectedMaxSize int64, | |
| ) { | |
| size := 100 | |
| type kv struct { | |
| key []byte | |
| values [][]byte | |
| delete bool | |
| } | |
| // this segment is not part of the merge, but might still play a role in | |
| // overall results. For example if one of the later segments has a tombstone | |
| // for it | |
| var previous1 []kv | |
| var previous2 []kv | |
| var segment1 []kv | |
| var segment2 []kv | |
| var expected []kv | |
| var bucket *Bucket | |
| dirName := t.TempDir() | |
| t.Run("create test data", func(t *testing.T) { | |
| // The test data is split into 4 scenarios evenly: | |
| // | |
| // 0.) created in the first segment, never touched again | |
| // 1.) created in the first segment, appended to it in the second | |
| // 2.) created in the first segment, first element deleted in the second | |
| // 3.) created in the first segment, second element deleted in the second | |
| // 4.) not present in the first segment, created in the second | |
| // 5.) present in an unrelated previous segment, deleted in the first | |
| // 6.) present in an unrelated previous segment, deleted in the second | |
| // 7.) present in an unrelated previous segment, never touched again | |
| for i := 0; i < size; i++ { | |
| key := []byte(fmt.Sprintf("key-%02d", i)) | |
| value1 := []byte(fmt.Sprintf("value-%02d-01", i)) | |
| value2 := []byte(fmt.Sprintf("value-%02d-02", i)) | |
| values := [][]byte{value1, value2} | |
| switch i % 8 { | |
| case 0: | |
| // add to segment 1 | |
| segment1 = append(segment1, kv{ | |
| key: key, | |
| values: values[:1], | |
| }) | |
| // leave this element untouched in the second segment | |
| expected = append(expected, kv{ | |
| key: key, | |
| values: values[:1], | |
| }) | |
| case 1: | |
| // add to segment 1 | |
| segment1 = append(segment1, kv{ | |
| key: key, | |
| values: values[:1], | |
| }) | |
| // update in the second segment | |
| segment2 = append(segment2, kv{ | |
| key: key, | |
| values: values[1:2], | |
| }) | |
| expected = append(expected, kv{ | |
| key: key, | |
| values: values, | |
| }) | |
| case 2: | |
| // add both to segment 1, delete the first | |
| segment1 = append(segment1, kv{ | |
| key: key, | |
| values: values, | |
| }) | |
| // delete first element in the second segment | |
| segment2 = append(segment2, kv{ | |
| key: key, | |
| values: values[:1], | |
| delete: true, | |
| }) | |
| // only the 2nd element should be left in the expected | |
| expected = append(expected, kv{ | |
| key: key, | |
| values: values[1:2], | |
| }) | |
| case 3: | |
| // add both to segment 1, delete the second | |
| segment1 = append(segment1, kv{ | |
| key: key, | |
| values: values, | |
| }) | |
| // delete second element in the second segment | |
| segment2 = append(segment2, kv{ | |
| key: key, | |
| values: values[1:], | |
| delete: true, | |
| }) | |
| // only the 1st element should be left in the expected | |
| expected = append(expected, kv{ | |
| key: key, | |
| values: values[:1], | |
| }) | |
| case 4: | |
| // do not add to segment 1 | |
| // only add to segment 2 (first entry) | |
| segment2 = append(segment2, kv{ | |
| key: key, | |
| values: values, | |
| }) | |
| expected = append(expected, kv{ | |
| key: key, | |
| values: values, | |
| }) | |
| case 5: | |
| // only part of a previous segment, which is not part of the merge | |
| previous1 = append(previous1, kv{ | |
| key: key, | |
| values: values[:1], | |
| }) | |
| previous2 = append(previous2, kv{ | |
| key: key, | |
| values: values[1:], | |
| }) | |
| // delete in segment 1 | |
| segment1 = append(segment1, kv{ | |
| key: key, | |
| values: values[:1], | |
| delete: true, | |
| }) | |
| segment1 = append(segment1, kv{ | |
| key: key, | |
| values: values[1:], | |
| delete: true, | |
| }) | |
| // should not have any values in expected at all, not even a key | |
| case 6: | |
| // only part of a previous segment, which is not part of the merge | |
| previous1 = append(previous1, kv{ | |
| key: key, | |
| values: values[:1], | |
| }) | |
| previous2 = append(previous2, kv{ | |
| key: key, | |
| values: values[1:], | |
| }) | |
| // delete in segment 2 | |
| segment2 = append(segment2, kv{ | |
| key: key, | |
| values: values[:1], | |
| delete: true, | |
| }) | |
| segment2 = append(segment2, kv{ | |
| key: key, | |
| values: values[1:], | |
| delete: true, | |
| }) | |
| // should not have any values in expected at all, not even a key | |
| case 7: | |
| // part of a previous segment | |
| previous1 = append(previous1, kv{ | |
| key: key, | |
| values: values[:1], | |
| }) | |
| previous2 = append(previous2, kv{ | |
| key: key, | |
| values: values[1:], | |
| }) | |
| expected = append(expected, kv{ | |
| key: key, | |
| values: values, | |
| }) | |
| } | |
| } | |
| }) | |
| t.Run("shuffle the import order for each segment", func(t *testing.T) { | |
| // this is to make sure we don't accidentally rely on the import order | |
| rand.Shuffle(len(segment1), func(i, j int) { | |
| segment1[i], segment1[j] = segment1[j], segment1[i] | |
| }) | |
| rand.Shuffle(len(segment2), func(i, j int) { | |
| segment2[i], segment2[j] = segment2[j], segment2[i] | |
| }) | |
| }) | |
| t.Run("init bucket", func(t *testing.T) { | |
| b, err := NewBucket(ctx, dirName, dirName, nullLogger(), nil, | |
| cyclemanager.NewCallbackGroupNoop(), cyclemanager.NewCallbackGroupNoop(), opts...) | |
| require.Nil(t, err) | |
| // so big it effectively never triggers as part of this test | |
| b.SetMemtableThreshold(1e9) | |
| bucket = b | |
| }) | |
| t.Run("import and flush previous segments", func(t *testing.T) { | |
| for _, pair := range previous1 { | |
| err := bucket.SetAdd(pair.key, pair.values) | |
| require.Nil(t, err) | |
| } | |
| require.Nil(t, bucket.FlushAndSwitch()) | |
| for _, pair := range previous2 { | |
| err := bucket.SetAdd(pair.key, pair.values) | |
| require.Nil(t, err) | |
| } | |
| require.Nil(t, bucket.FlushAndSwitch()) | |
| }) | |
| t.Run("import segment 1", func(t *testing.T) { | |
| for _, pair := range segment1 { | |
| if !pair.delete { | |
| err := bucket.SetAdd(pair.key, pair.values) | |
| require.Nil(t, err) | |
| } else { | |
| err := bucket.SetDeleteSingle(pair.key, pair.values[0]) | |
| require.Nil(t, err) | |
| } | |
| } | |
| }) | |
| t.Run("flush to disk", func(t *testing.T) { | |
| require.Nil(t, bucket.FlushAndSwitch()) | |
| }) | |
| t.Run("import segment 2", func(t *testing.T) { | |
| for _, pair := range segment2 { | |
| if !pair.delete { | |
| err := bucket.SetAdd(pair.key, pair.values) | |
| require.Nil(t, err) | |
| } else { | |
| err := bucket.SetDeleteSingle(pair.key, pair.values[0]) | |
| require.Nil(t, err) | |
| } | |
| } | |
| }) | |
| t.Run("flush to disk", func(t *testing.T) { | |
| require.Nil(t, bucket.FlushAndSwitch()) | |
| }) | |
| t.Run("verify control before compaction", func(t *testing.T) { | |
| var retrieved []kv | |
| c := bucket.SetCursor() | |
| defer c.Close() | |
| for k, v := c.First(); k != nil; k, v = c.Next() { | |
| retrieved = append(retrieved, kv{ | |
| key: k, | |
| values: v, | |
| }) | |
| } | |
| assert.Equal(t, expected, retrieved) | |
| }) | |
| t.Run("compact until no longer eligible", func(t *testing.T) { | |
| i := 0 | |
| var compacted bool | |
| var err error | |
| for compacted, err = bucket.disk.compactOnce(); err == nil && compacted; compacted, err = bucket.disk.compactOnce() { | |
| if i == 1 { | |
| // segment1 and segment2 merged | |
| // none of them is root segment, so tombstones | |
| // will not be removed regardless of keepTombstones setting | |
| assertSecondSegmentOfSize(t, bucket, 8556, 8556) | |
| } | |
| i++ | |
| } | |
| require.Nil(t, err) | |
| }) | |
| t.Run("verify control after compaction", func(t *testing.T) { | |
| var retrieved []kv | |
| c := bucket.SetCursor() | |
| defer c.Close() | |
| for k, v := c.First(); k != nil; k, v = c.Next() { | |
| retrieved = append(retrieved, kv{ | |
| key: k, | |
| values: v, | |
| }) | |
| } | |
| assert.Equal(t, expected, retrieved) | |
| assertSingleSegmentOfSize(t, bucket, expectedMinSize, expectedMaxSize) | |
| }) | |
| } | |
| func compactionSetStrategy_RemoveUnnecessary(ctx context.Context, t *testing.T, opts []BucketOption) { | |
| // in this test each segment reverses the action of the previous segment so | |
| // that in the end a lot of information is present in the individual segments | |
| // which is no longer needed. We then verify that after all compaction this | |
| // information is gone, thus freeing up disk space | |
| size := 100 | |
| type kv struct { | |
| key []byte | |
| values [][]byte | |
| } | |
| key := []byte("my-key") | |
| var bucket *Bucket | |
| dirName := t.TempDir() | |
| t.Run("init bucket", func(t *testing.T) { | |
| b, err := NewBucket(ctx, dirName, dirName, nullLogger(), nil, | |
| cyclemanager.NewCallbackGroupNoop(), cyclemanager.NewCallbackGroupNoop(), opts...) | |
| require.Nil(t, err) | |
| // so big it effectively never triggers as part of this test | |
| b.SetMemtableThreshold(1e9) | |
| bucket = b | |
| }) | |
| t.Run("write segments", func(t *testing.T) { | |
| for i := 0; i < size; i++ { | |
| if i != 0 { | |
| // we can only delete an existing value if this isn't the first write | |
| value := []byte(fmt.Sprintf("value-%05d", i-1)) | |
| err := bucket.SetDeleteSingle(key, value) | |
| require.Nil(t, err) | |
| } | |
| value := []byte(fmt.Sprintf("value-%05d", i)) | |
| err := bucket.SetAdd(key, [][]byte{value}) | |
| require.Nil(t, err) | |
| require.Nil(t, bucket.FlushAndSwitch()) | |
| } | |
| }) | |
| t.Run("verify control before compaction", func(t *testing.T) { | |
| var retrieved []kv | |
| expected := []kv{ | |
| { | |
| key: key, | |
| values: [][]byte{[]byte(fmt.Sprintf("value-%05d", size-1))}, | |
| }, | |
| } | |
| c := bucket.SetCursor() | |
| defer c.Close() | |
| for k, v := c.First(); k != nil; k, v = c.Next() { | |
| retrieved = append(retrieved, kv{ | |
| key: k, | |
| values: v, | |
| }) | |
| } | |
| assert.Equal(t, expected, retrieved) | |
| }) | |
| t.Run("compact until no longer eligible", func(t *testing.T) { | |
| var compacted bool | |
| var err error | |
| for compacted, err = bucket.disk.compactOnce(); err == nil && compacted; compacted, err = bucket.disk.compactOnce() { | |
| } | |
| require.Nil(t, err) | |
| }) | |
| t.Run("verify control before compaction", func(t *testing.T) { | |
| var retrieved []kv | |
| expected := []kv{ | |
| { | |
| key: key, | |
| values: [][]byte{[]byte(fmt.Sprintf("value-%05d", size-1))}, | |
| }, | |
| } | |
| c := bucket.SetCursor() | |
| defer c.Close() | |
| for k, v := c.First(); k != nil; k, v = c.Next() { | |
| retrieved = append(retrieved, kv{ | |
| key: k, | |
| values: v, | |
| }) | |
| } | |
| assert.Equal(t, expected, retrieved) | |
| }) | |
| } | |
| func compactionSetStrategy_FrequentPutDeleteOperations(ctx context.Context, t *testing.T, opts []BucketOption) { | |
| // In this test we are testing that the compaction works well for set collection | |
| maxSize := 10 | |
| for size := 4; size < maxSize; size++ { | |
| t.Run(fmt.Sprintf("compact %v segments", size), func(t *testing.T) { | |
| var bucket *Bucket | |
| key := []byte("key-original") | |
| value1 := []byte("value-01") | |
| value2 := []byte("value-02") | |
| values := [][]byte{value1, value2} | |
| dirName := t.TempDir() | |
| t.Run("init bucket", func(t *testing.T) { | |
| b, err := NewBucket(ctx, dirName, dirName, nullLogger(), nil, | |
| cyclemanager.NewCallbackGroupNoop(), cyclemanager.NewCallbackGroupNoop(), opts...) | |
| require.Nil(t, err) | |
| // so big it effectively never triggers as part of this test | |
| b.SetMemtableThreshold(1e9) | |
| bucket = b | |
| }) | |
| t.Run("import and flush segments", func(t *testing.T) { | |
| for i := 0; i < size; i++ { | |
| err := bucket.SetAdd(key, values) | |
| require.Nil(t, err) | |
| if size == 5 { | |
| // delete all | |
| err := bucket.SetDeleteSingle(key, values[0]) | |
| require.Nil(t, err) | |
| err = bucket.SetDeleteSingle(key, values[1]) | |
| require.Nil(t, err) | |
| } else if size == 6 { | |
| // delete only one value | |
| err := bucket.SetDeleteSingle(key, values[0]) | |
| require.Nil(t, err) | |
| } else if i != size-1 { | |
| // don't delete from the last segment | |
| err := bucket.SetDeleteSingle(key, values[0]) | |
| require.Nil(t, err) | |
| err = bucket.SetDeleteSingle(key, values[1]) | |
| require.Nil(t, err) | |
| } | |
| require.Nil(t, bucket.FlushAndSwitch()) | |
| } | |
| }) | |
| t.Run("verify that objects exist before compaction", func(t *testing.T) { | |
| res, err := bucket.SetList(key) | |
| assert.Nil(t, err) | |
| if size == 5 { | |
| assert.Len(t, res, 0) | |
| } else if size == 6 { | |
| assert.Len(t, res, 1) | |
| } else { | |
| assert.Len(t, res, 2) | |
| } | |
| }) | |
| t.Run("compact until no longer eligible", func(t *testing.T) { | |
| var compacted bool | |
| var err error | |
| for compacted, err = bucket.disk.compactOnce(); err == nil && compacted; compacted, err = bucket.disk.compactOnce() { | |
| } | |
| require.Nil(t, err) | |
| }) | |
| t.Run("verify that objects exist after compaction", func(t *testing.T) { | |
| res, err := bucket.SetList(key) | |
| assert.Nil(t, err) | |
| if size == 5 { | |
| assert.Len(t, res, 0) | |
| } else if size == 6 { | |
| assert.Len(t, res, 1) | |
| } else { | |
| assert.Len(t, res, 2) | |
| } | |
| }) | |
| }) | |
| } | |
| } | |