From fcb35d82cf588eee4a1e5ca338cf90df3c200cd4 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Mon, 9 Nov 2020 17:58:22 +0300
Subject: [PATCH] [#165] Revert "[#142] metabase: Store header value index in a
 tree leaf"

Revert commit 0faa40e4 to increase the disk space consumed by the
metabase in favor of the speed of index updates.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 pkg/local_object_storage/metabase/put.go    | 45 +++------------------
 pkg/local_object_storage/metabase/select.go | 19 ++++-----
 2 files changed, 14 insertions(+), 50 deletions(-)

diff --git a/pkg/local_object_storage/metabase/put.go b/pkg/local_object_storage/metabase/put.go
index 7aaa08922d..8a287e2fcd 100644
--- a/pkg/local_object_storage/metabase/put.go
+++ b/pkg/local_object_storage/metabase/put.go
@@ -1,9 +1,6 @@
 package meta
 
 import (
-	"bytes"
-	"encoding/gob"
-
 	objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object"
 	v2object "github.com/nspcc-dev/neofs-api-go/v2/object"
 	"github.com/nspcc-dev/neofs-node/pkg/core/object"
@@ -64,19 +61,16 @@ func (db *DB) Put(obj *object.Object) error {
 					return errors.Wrapf(err, "(%T) could not create bucket for header key", db)
 				}
 
-				v := nonEmptyKeyBytes([]byte(indices[i].val))
+				v := []byte(indices[i].val)
 
-				strs, err := decodeAddressList(keyBucket.Get(v))
+				// create address bucket for the value
+				valBucket, err := keyBucket.CreateBucketIfNotExists(nonEmptyKeyBytes(v))
 				if err != nil {
-					return errors.Wrapf(err, "(%T) could not decode address list", db)
+					return errors.Wrapf(err, "(%T) could not create bucket for header value", db)
 				}
 
-				data, err := encodeAddressList(append(strs, string(addrKey)))
-				if err != nil {
-					return errors.Wrapf(err, "(%T) could not encode address list", db)
-				}
-
-				if err := keyBucket.Put(v, data); err != nil {
+				// put object address to value bucket
+				if err := valBucket.Put(addrKey, nil); err != nil {
 					return errors.Wrapf(err, "(%T) could not put item to header bucket", db)
 				}
 			}
@@ -160,30 +154,3 @@ func objectIndices(obj *object.Object, parent bool) []bucketItem {
 
 	return res
 }
-
-// FIXME: gob is a temporary solution, use protobuf.
-func decodeAddressList(data []byte) ([]string, error) {
-	if len(data) == 0 {
-		return nil, nil
-	}
-
-	var strs []string
-
-	decoder := gob.NewDecoder(bytes.NewReader(data))
-	if err := decoder.Decode(&strs); err != nil {
-		return nil, err
-	}
-
-	return strs, nil
-}
-
-func encodeAddressList(l []string) ([]byte, error) {
-	buf := bytes.NewBuffer(nil)
-	encoder := gob.NewEncoder(buf)
-
-	if err := encoder.Encode(l); err != nil {
-		return nil, err
-	}
-
-	return buf.Bytes(), nil
-}
diff --git a/pkg/local_object_storage/metabase/select.go b/pkg/local_object_storage/metabase/select.go
index a285ed5375..3f18604ea3 100644
--- a/pkg/local_object_storage/metabase/select.go
+++ b/pkg/local_object_storage/metabase/select.go
@@ -53,18 +53,15 @@ func (db *DB) selectObjects(tx *bbolt.Tx, fs object.SearchFilters) ([]*object.Ad
 		if err := keyBucket.ForEach(func(k, v []byte) error {
 			include := matchFunc(string(key), string(cutKeyBytes(k)), fVal)
 
-			strs, err := decodeAddressList(v)
-			if err != nil {
-				return errors.Wrapf(err, "(%T) could not decode address list", db)
-			}
+			if include {
+				return keyBucket.Bucket(k).ForEach(func(k, _ []byte) error {
+					if num := mAddr[string(k)]; num == fNum {
+						// otherwise object does not match current or some previous filter
+						mAddr[string(k)] = fNum + 1
+					}
 
-			for j := range strs {
-				if num := mAddr[strs[j]]; num != fNum {
-					// than object does not match some previous filter
-					continue
-				} else if include {
-					mAddr[strs[j]] = fNum + 1
-				}
+					return nil
+				})
 			}
 
 			return nil