Spaces:
Running
Running
File size: 3,422 Bytes
b110593 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
// _ _
// __ _____ __ ___ ___ __ _| |_ ___
// \ \ /\ / / _ \/ _` \ \ / / |/ _` | __/ _ \
// \ V V / __/ (_| |\ V /| | (_| | || __/
// \_/\_/ \___|\__,_| \_/ |_|\__,_|\__\___|
//
// Copyright © 2016 - 2024 Weaviate B.V. All rights reserved.
//
// CONTACT: [email protected]
//
package docid
import (
"encoding/binary"
"github.com/weaviate/weaviate/entities/storobj"
"github.com/weaviate/weaviate/entities/models"
"github.com/pkg/errors"
"github.com/weaviate/weaviate/adapters/repos/db/helpers"
"github.com/weaviate/weaviate/adapters/repos/db/lsmkv"
)
// ObjectScanFn is called once per object, if false or an error is returned,
// the scanning will stop
type ObjectScanFn func(prop *models.PropertySchema, docID uint64) (bool, error)
// ScanObjectsLSM calls the provided scanFn on each object for the
// specified pointer. If a pointer does not resolve to an object-id, the item
// will be skipped. The number of times scanFn is called can therefore be
// smaller than the input length of pointers.
func ScanObjectsLSM(store *lsmkv.Store, pointers []uint64, scan ObjectScanFn, properties []string) error {
return newObjectScannerLSM(store, pointers, scan, properties).Do()
}
type objectScannerLSM struct {
store *lsmkv.Store
pointers []uint64
scanFn ObjectScanFn
objectsBucket *lsmkv.Bucket
properties []string
}
func newObjectScannerLSM(store *lsmkv.Store, pointers []uint64,
scan ObjectScanFn, properties []string,
) *objectScannerLSM {
return &objectScannerLSM{
store: store,
pointers: pointers,
scanFn: scan,
properties: properties,
}
}
func (os *objectScannerLSM) Do() error {
if err := os.init(); err != nil {
return errors.Wrap(err, "init object scanner")
}
if err := os.scan(); err != nil {
return errors.Wrap(err, "scan")
}
return nil
}
func (os *objectScannerLSM) init() error {
bucket := os.store.Bucket(helpers.ObjectsBucketLSM)
if bucket == nil {
return errors.Errorf("objects bucket not found")
}
os.objectsBucket = bucket
return nil
}
func (os *objectScannerLSM) scan() error {
// each object is scanned one after the other, so we can reuse the same memory allocations for all objects
docIDBytes := make([]byte, 8)
// Preallocate strings needed for json unmarshalling
propStrings := make([][]string, len(os.properties))
for i := range os.properties {
propStrings[i] = []string{os.properties[i]}
}
// The typed properties are needed for extraction from json
var properties models.PropertySchema
propertiesTyped := map[string]interface{}{}
for _, prop := range os.properties {
propertiesTyped[prop] = nil
}
for _, id := range os.pointers {
binary.LittleEndian.PutUint64(docIDBytes, id)
res, err := os.objectsBucket.GetBySecondary(0, docIDBytes)
if err != nil {
return err
}
if res == nil {
continue
}
if len(os.properties) > 0 {
err = storobj.UnmarshalPropertiesFromObject(res, &propertiesTyped, os.properties, propStrings)
if err != nil {
return errors.Wrapf(err, "unmarshal data object")
}
properties = propertiesTyped
}
continueScan, err := os.scanFn(&properties, id)
if err != nil {
return errors.Wrapf(err, "scan")
}
if !continueScan {
break
}
}
return nil
}
|