diff --git a/pkg/local_object_storage/metabase/metadata.go b/pkg/local_object_storage/metabase/metadata.go index e2ecea1b49..869f777054 100644 --- a/pkg/local_object_storage/metabase/metadata.go +++ b/pkg/local_object_storage/metabase/metadata.go @@ -34,6 +34,8 @@ const ( attrIDFixedLen = 1 + oid.Size + utf8DelimiterLen // prefix first ) +const binPropMarker = "1" // ROOT, PHY, etc. + var ( maxUint256 = new(big.Int).SetBytes([]byte{255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255}) @@ -45,10 +47,10 @@ func invalidMetaBucketKeyErr(key []byte, cause error) error { } // TODO: fill on migration. -// TODO: ROOT and PHY props. // TODO: cleaning on obj removal. func putMetadata(tx *bbolt.Tx, cnr cid.ID, id oid.ID, ver version.Version, owner user.ID, typ object.Type, creationEpoch uint64, - payloadLen uint64, pldHash, pldHmmHash, splitID []byte, parentID, firstID oid.ID, attrs []object.Attribute) error { + payloadLen uint64, pldHash, pldHmmHash, splitID []byte, parentID, firstID oid.ID, attrs []object.Attribute, + root, phy bool) error { metaBkt, err := tx.CreateBucketIfNotExists(metaBucketKey(cnr)) if err != nil { return fmt.Errorf("create meta bucket for container: %w", err) @@ -96,6 +98,16 @@ func putMetadata(tx *bbolt.Tx, cnr cid.ID, id oid.ID, ver version.Version, owner return err } } + if root { + if err = putPlainAttribute(metaBkt, &keyBuf, id, object.FilterRoot, binPropMarker); err != nil { + return err + } + } + if phy { + if err = putPlainAttribute(metaBkt, &keyBuf, id, object.FilterPhysical, binPropMarker); err != nil { + return err + } + } for i := range attrs { ak, av := attrs[i].Key(), attrs[i].Value() if n, isInt := parseInt(av); isInt && n.Cmp(maxUint256Neg) >= 0 && n.Cmp(maxUint256) <= 0 { @@ -195,7 +207,7 @@ func (db *DB) search(cnr cid.ID, fs object.SearchFilters, attrs []string, cursor func (db *DB) searchInBucket(metaBkt *bbolt.Bucket, fs object.SearchFilters, attrs []string, cursor *SearchCursor, count uint16) ([]client.SearchResultItem, *SearchCursor, error) { // TODO: make as much as possible outside the Bolt tx - primMatcher := fs[0].Operation() + primMatcher, primVal := convertFilterValue(fs[0]) intPrimMatcher := isNumericOp(primMatcher) notPresentPrimMatcher := primMatcher == object.MatchNotPresent primAttr := fs[0].Header() // attribute emptiness already prevented @@ -223,7 +235,7 @@ func (db *DB) searchInBucket(metaBkt *bbolt.Bucket, fs object.SearchFilters, att if primMatcher == object.MatchStringEqual || primMatcher == object.MatchCommonPrefix || primMatcher == object.MatchNumGT || primMatcher == object.MatchNumGE { var err error - if primSeekKey, primSeekPrefix, err = seekKeyForAttribute(primAttr, fs[0].Value()); err != nil { + if primSeekKey, primSeekPrefix, err = seekKeyForAttribute(primAttr, primVal); err != nil { return nil, nil, fmt.Errorf("invalid primary filter value: %w", err) } } else { @@ -279,11 +291,12 @@ nextPrimKey: if i > 0 && attr != primAttr { continue } - checkedDBVal, fltVal, err := combineValues(attr, dbVal, fs[i].Value()) // TODO: deduplicate DB value preparation + mch, val := convertFilterValue(fs[i]) + checkedDBVal, fltVal, err := combineValues(attr, dbVal, val) // TODO: deduplicate DB value preparation if err != nil { return nil, nil, fmt.Errorf("invalid key in meta bucket: invalid attribute %s value: %w", attr, err) } - if !matchValues(checkedDBVal, fs[i].Operation(), fltVal) { + if !matchValues(checkedDBVal, mch, fltVal) { continue nextPrimKey } // TODO: attribute value can be requested, it can be collected here, or we can @@ -311,7 +324,7 @@ nextPrimKey: if j > 0 && fs[j].Header() != attr { continue } - m := fs[j].Operation() + m, val := convertFilterValue(fs[j]) if dbVal == nil { if m == object.MatchNotPresent { continue @@ -347,7 +360,7 @@ nextPrimKey: } else { checkedDBVal = dbVal } - checkedDBVal, fltVal, err := combineValues(attr, checkedDBVal, fs[j].Value()) // TODO: deduplicate DB value preparation + checkedDBVal, fltVal, err := combineValues(attr, checkedDBVal, val) // TODO: deduplicate DB value preparation if err != nil { return nil, nil, invalidMetaBucketKeyErr(primKey, fmt.Errorf("invalid attribute %s value: %w", attr, err)) } @@ -507,7 +520,7 @@ func seekKeyForAttribute(attr, fltVal string) ([]byte, []byte, error) { return nil, nil, fmt.Errorf("decode %q UUID attribute: %w", attr, err) } dbVal = uid[:] - case object.FilterVersion, object.FilterType: + case object.FilterVersion, object.FilterType, object.FilterRoot, object.FilterPhysical: } key := make([]byte, 1+len(attr)+utf8DelimiterLen+len(dbVal)) // prefix 1st key[0] = metaPrefixAttrIDPlain @@ -794,3 +807,10 @@ func (x *metaAttributeSeeker) restoreVal(id []byte, attr string, stored []byte) } return string(stored), nil } + +func convertFilterValue(f object.SearchFilter) (object.SearchMatchType, string) { + if attr := f.Header(); attr == object.FilterRoot || attr == object.FilterPhysical { + return object.MatchStringEqual, binPropMarker + } + return f.Operation(), f.Value() +} diff --git a/pkg/local_object_storage/metabase/metadata_test.go b/pkg/local_object_storage/metabase/metadata_test.go index f70b94eb3d..ef860792bc 100644 --- a/pkg/local_object_storage/metabase/metadata_test.go +++ b/pkg/local_object_storage/metabase/metadata_test.go @@ -128,6 +128,8 @@ func TestPutMetadata(t *testing.T) { 0, 0, 0, 0, 101, 118, 30, 154, 145, 227, 159, 231}) assertIntAttr(t, mb, id, "$Object:payloadLength", []byte{1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 29, 7, 76, 78, 96, 175, 200, 130}) + assertAttr(t, mb, id, "$Object:ROOT", "1") + assertAttr(t, mb, id, "$Object:PHY", "1") assertAttr(t, mb, id, "attr_1", "val_1") assertAttr(t, mb, id, "attr_2", "val_2") assertAttr(t, mb, id, "num_negative_overflow", "-115792089237316195423570985008687907853269984665640564039457584007913129639936") @@ -670,24 +672,22 @@ func TestDB_SearchObjects(t *testing.T) { } }) t.Run("ROOT", func(t *testing.T) { - t.Skip("not supported yet") - check("$Object:ROOT", 0, "", []uint{0, 1}) for _, matcher := range []object.SearchMatchType{ - object.MatchStringEqual, object.MatchStringNotEqual, object.MatchNotPresent, object.MatchCommonPrefix, + object.MatchUnspecified, object.MatchStringEqual, object.MatchStringNotEqual, object.MatchNumGT, object.MatchNumGE, object.MatchNumLT, object.MatchNumLE, } { - check("$Object:ROOT", matcher, "", nil) + check("$Object:ROOT", matcher, "", []uint{0, 1}) } + check("$Object:ROOT", object.MatchNotPresent, "", nil) }) t.Run("PHY", func(t *testing.T) { - t.Skip("not supported yet") - check("$Object;PHY", 0, "", []uint{0, 1, 2, 3}) for _, matcher := range []object.SearchMatchType{ - object.MatchStringEqual, object.MatchStringNotEqual, object.MatchNotPresent, object.MatchCommonPrefix, + object.MatchUnspecified, object.MatchStringEqual, object.MatchStringNotEqual, object.MatchNumGT, object.MatchNumGE, object.MatchNumLT, object.MatchNumLE, } { - check("$Object:PHY", matcher, "", nil) + check("$Object:PHY", matcher, "", []uint{2, 3, 4, 5}) } + check("$Object:PHY", object.MatchNotPresent, "", nil) }) t.Run("version", func(t *testing.T) { check := func(m object.SearchMatchType, v string, matchInds []uint) { diff --git a/pkg/local_object_storage/metabase/put.go b/pkg/local_object_storage/metabase/put.go index 3f8ad2adfb..ce1c1e4d88 100644 --- a/pkg/local_object_storage/metabase/put.go +++ b/pkg/local_object_storage/metabase/put.go @@ -111,7 +111,8 @@ func (db *DB) put( return nil } - if par := obj.Parent(); par != nil && !isParent { // limit depth by two + par := obj.Parent() + if par != nil && !isParent { // limit depth by two if parID := par.GetID(); !parID.IsZero() { // skip the first object without useful info parentSI, err := splitInfoFromObject(obj) if err != nil { @@ -171,7 +172,7 @@ func (db *DB) put( pldHmmHash = h.Value() } if err := putMetadata(tx, cnr, obj.GetID(), ver, *owner, obj.Type(), obj.CreationEpoch(), obj.PayloadSize(), - pldHash.Value(), pldHmmHash, obj.SplitID().ToV2(), obj.GetParentID(), obj.GetFirstID(), obj.Attributes()); err != nil { + pldHash.Value(), pldHmmHash, obj.SplitID().ToV2(), obj.GetParentID(), obj.GetFirstID(), obj.Attributes(), par == nil, !isParent); err != nil { return fmt.Errorf("put metadata: %w", err) }