[#1347] metabase: Fix EC search

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>

pkg/local_object_storage/metabase/select.go

@@ -150,10 +150,10 @@ func (db *DB) selectObjects(tx *bbolt.Tx, cnr cid.ID, fs objectSDK.SearchFilters
 			continue // ignore removed objects
 		}
 
-		if !db.matchSlowFilters(tx, addr, group.slowFilters, currEpoch) {
+		addr, match := db.matchSlowFilters(tx, addr, group.slowFilters, currEpoch)
+		if !match {
 			continue // ignore objects with unmatched slow filters
 		}
-
 		res = append(res, addr)
 	}
 
@@ -382,15 +382,15 @@ func (db *DB) selectObjectID(
 }
 
 // matchSlowFilters return true if object header is matched by all slow filters.
-func (db *DB) matchSlowFilters(tx *bbolt.Tx, addr oid.Address, f objectSDK.SearchFilters, currEpoch uint64) bool {
+func (db *DB) matchSlowFilters(tx *bbolt.Tx, addr oid.Address, f objectSDK.SearchFilters, currEpoch uint64) (oid.Address, bool) {
+	result := addr
 	if len(f) == 0 {
-		return true
+		return result, true
 	}
-
 	buf := make([]byte, addressKeySize)
 	obj, err := db.get(tx, addr, buf, true, false, currEpoch)
 	if err != nil {
-		return false
+		return result, false
 	}
 
 	for i := range f {
@@ -415,23 +415,26 @@ func (db *DB) matchSlowFilters(tx *bbolt.Tx, addr oid.Address, f objectSDK.Searc
 		default: // user attribute
 			v, ok := attributeValue(obj, f[i].Header())
 			if ok {
+				if ech := obj.ECHeader(); ech != nil {
+					result.SetObject(ech.Parent())
+				}
 				data = []byte(v)
 			} else {
-				return f[i].Operation() == objectSDK.MatchNotPresent
+				return result, f[i].Operation() == objectSDK.MatchNotPresent
 			}
 		}
 
 		matchFunc, ok := db.matchers[f[i].Operation()]
 		if !ok {
-			return false
+			return result, false
 		}
 
 		if !matchFunc.matchSlow(f[i].Header(), data, f[i].Value()) {
-			return false
+			return result, false
 		}
 	}
 
-	return true
+	return result, true
 }
 
 func attributeValue(obj *objectSDK.Object, attribute string) (string, bool) {

pkg/local_object_storage/metabase/select_test.go

@@ -70,6 +70,22 @@ func TestDB_SelectUserAttributes(t *testing.T) {
 	err = putBig(db, raw6)
 	require.NoError(t, err)
 
+	raw7 := testutil.GenerateObjectWithCID(cnr)
+	var attr objectSDK.Attribute
+	attr.SetKey("path")
+	attr.SetValue("test/3/4")
+	attrs := raw7.Attributes()
+	attrs = append(attrs, attr)
+	ech := objectSDK.NewECHeader(objectSDK.ECParentInfo{
+		ID:         oidtest.ID(),
+		Attributes: attrs,
+	}, 0, 3, []byte{}, 0)
+	raw7.SetECHeader(ech)
+	require.NoError(t, putBig(db, raw7))
+	var raw7Parent oid.Address
+	raw7Parent.SetContainer(cnr)
+	raw7Parent.SetObject(ech.Parent())
+
 	fs := objectSDK.SearchFilters{}
 	fs.AddFilter("foo", "bar", objectSDK.MatchStringEqual)
 	testSelect(t, db, cnr, fs,
@@ -100,6 +116,7 @@ func TestDB_SelectUserAttributes(t *testing.T) {
 		object.AddressOf(raw4),
 		object.AddressOf(raw5),
 		object.AddressOf(raw6),
+		object.AddressOf(raw7),
 	)
 
 	fs = objectSDK.SearchFilters{}
@@ -110,6 +127,7 @@ func TestDB_SelectUserAttributes(t *testing.T) {
 		object.AddressOf(raw4),
 		object.AddressOf(raw5),
 		object.AddressOf(raw6),
+		object.AddressOf(raw7),
 	)
 
 	fs = objectSDK.SearchFilters{}
@@ -120,6 +138,7 @@ func TestDB_SelectUserAttributes(t *testing.T) {
 		object.AddressOf(raw4),
 		object.AddressOf(raw5),
 		object.AddressOf(raw6),
+		object.AddressOf(raw7),
 	)
 
 	fs = objectSDK.SearchFilters{}
@@ -131,6 +150,7 @@ func TestDB_SelectUserAttributes(t *testing.T) {
 		object.AddressOf(raw4),
 		object.AddressOf(raw5),
 		object.AddressOf(raw6),
+		object.AddressOf(raw7),
 	)
 
 	fs = objectSDK.SearchFilters{}
@@ -139,6 +159,7 @@ func TestDB_SelectUserAttributes(t *testing.T) {
 		object.AddressOf(raw4),
 		object.AddressOf(raw5),
 		object.AddressOf(raw6),
+		raw7Parent,
 	)
 
 	fs = objectSDK.SearchFilters{}
@@ -147,6 +168,12 @@ func TestDB_SelectUserAttributes(t *testing.T) {
 		object.AddressOf(raw4),
 		object.AddressOf(raw5),
 	)
+
+	fs = objectSDK.SearchFilters{}
+	fs.AddFilter("path", "test/3/4", objectSDK.MatchStringEqual)
+	testSelect(t, db, cnr, fs,
+		raw7Parent,
+	)
 }
 
 func TestDB_SelectRootPhyParent(t *testing.T) {