From 2e074d3846cc41c234b0165c21221b6aadf98e5b Mon Sep 17 00:00:00 2001 From: Evgenii Stratonikov Date: Wed, 5 Jun 2024 11:14:06 +0300 Subject: [PATCH] [#1163] metabase: Properly save EC parent split ID Search by SplitID should return all parts of a complex object. Signed-off-by: Evgenii Stratonikov --- pkg/local_object_storage/metabase/put.go | 12 ++ .../metabase/select_test.go | 103 ++++++++++++++++++ 2 files changed, 115 insertions(+) diff --git a/pkg/local_object_storage/metabase/put.go b/pkg/local_object_storage/metabase/put.go index 4b6359ebf..05922d1bd 100644 --- a/pkg/local_object_storage/metabase/put.go +++ b/pkg/local_object_storage/metabase/put.go @@ -347,6 +347,18 @@ func updateListIndexes(tx *bbolt.Tx, obj *objectSDK.Object, f updateIndexItemFun if err != nil { return err } + + if ech.ParentSplitID() != nil { + objKey := objectKey(ech.Parent(), make([]byte, objectKeySize)) + err := f(tx, namedBucketItem{ + name: splitBucketName(cnr, bucketName), + key: ech.ParentSplitID().ToV2(), + val: objKey, + }) + if err != nil { + return err + } + } } return nil diff --git a/pkg/local_object_storage/metabase/select_test.go b/pkg/local_object_storage/metabase/select_test.go index 9ff2fd65e..8951a2dbe 100644 --- a/pkg/local_object_storage/metabase/select_test.go +++ b/pkg/local_object_storage/metabase/select_test.go @@ -3,6 +3,7 @@ package meta_test import ( "context" "encoding/hex" + "math/rand" "strconv" "testing" @@ -13,9 +14,13 @@ import ( cidSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test" objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" + "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/erasurecode" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test" + "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/transformer" + usertest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user/test" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/version" + "github.com/nspcc-dev/neo-go/pkg/crypto/keys" "github.com/stretchr/testify/require" ) @@ -628,6 +633,104 @@ func TestDB_SelectObjectID(t *testing.T) { }) } +type testTarget struct { + objects []*objectSDK.Object +} + +func (tt *testTarget) WriteObject(_ context.Context, obj *objectSDK.Object) error { + tt.objects = append(tt.objects, obj) + return nil +} + +func cutObject(t *testing.T, p transformer.ChunkedObjectWriter, hdr *objectSDK.Object, size int) { + ctx := context.Background() + require.NoError(t, p.WriteHeader(ctx, hdr)) + + payload := make([]byte, size) + rand.New(rand.NewSource(0)).Read(payload) + + _, err := p.Write(ctx, payload) + require.NoError(t, err) + + _, err = p.Close(ctx) + require.NoError(t, err) +} + +func TestDB_SelectSplitID_EC(t *testing.T) { + t.Parallel() + + const ( + partSize = 10 + partCount = 2 + dataCount = 2 + parityCount = 1 + ) + + db := newDB(t) + defer func() { require.NoError(t, db.Close()) }() + + cnr := cidtest.ID() + + pk, err := keys.NewPrivateKey() + require.NoError(t, err) + + tt := new(testTarget) + p := transformer.NewPayloadSizeLimiter(transformer.Params{ + Key: &pk.PrivateKey, + NextTargetInit: func() transformer.ObjectWriter { return tt }, + NetworkState: epochState{e: 1}, + MaxSize: partSize, + }) + + hdr := objectSDK.New() + hdr.SetContainerID(cnr) + hdr.SetOwnerID(usertest.ID()) + cutObject(t, p, hdr, partSize*partCount) + require.Equal(t, len(tt.objects), partCount+1) + + split := tt.objects[0].SplitID() + require.NotNil(t, split) + + ec, err := erasurecode.NewConstructor(dataCount, parityCount) + require.NoError(t, err) + + for i := 0; i < partCount; i++ { + cs, err := ec.Split(tt.objects[i], &pk.PrivateKey) + require.NoError(t, err) + + require.NoError(t, putBig(db, cs[0])) + } + + t.Run("not present", func(t *testing.T) { + fs := objectSDK.SearchFilters{} + fs.AddFilter(v2object.FilterHeaderSplitID, "", objectSDK.MatchNotPresent) + testSelect(t, db, cnr, fs) + }) + + t.Run("split id", func(t *testing.T) { + fs := objectSDK.SearchFilters{} + fs.AddFilter(v2object.FilterHeaderSplitID, split.String(), objectSDK.MatchStringEqual) + testSelect(t, db, cnr, fs, + object.AddressOf(tt.objects[0]), + object.AddressOf(tt.objects[1]), + ) + }) + + t.Run("empty split", func(t *testing.T) { + fs := objectSDK.SearchFilters{} + fs.AddFilter(v2object.FilterHeaderSplitID, "", objectSDK.MatchStringEqual) + testSelect(t, db, cnr, fs) + }) + + t.Run("unknown split id", func(t *testing.T) { + fs := objectSDK.SearchFilters{} + fs.AddFilter(v2object.FilterHeaderSplitID, + objectSDK.NewSplitID().String(), + objectSDK.MatchStringEqual) + testSelect(t, db, cnr, fs) + }) +} + func TestDB_SelectSplitID(t *testing.T) { t.Parallel()