lsmkv

package
v1.25.25 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 13, 2024 License: BSD-3-Clause Imports: 46 Imported by: 0

Documentation

Overview

LSMKV (= Log-structured Merge-Tree Key-Value Store)

This package contains Weaviate's custom LSM store. While modeled after the usecases that are required for Weaviate to be fast, reliable, and scalable, it is technically completely independent. You could build your own database on top of this key-value store.

Covering the architecture of LSM Stores in general goes beyond the scope of this documentation. Therefore things that are specific to this implementation are highlighted.

Strategies

To understand the different type of buckets in this store, you need to familiarize yourself with the following strategies. A strategy defines a different usecase for a Bucket.

  • "Replace"

    Replace resembles the classical key-value store. Each key has exactly one value. A subsequent PUT on an an existing key, replaces the value (hence the name "replace"). Once replaced a former value can no longer be retrieved, and will eventually be removed in compactions.

  • "Set" (aka "SetCollection")

    A set behaves like an unordered collection of independent values. In other words a single key has multiple values. For example, for key "foo", you could have values "bar1", "bar2", "bazzinga". A bucket of this type is optimized for cheap writes to add new set additions. For example adding another set element has a fixed cost independent of the number of the existing set length. This makes it very well suited for building an inverted index.

    Retrieving a Set has a slight cost to it if a set is spread across multiple segments. This cost will eventually reduce as more and more compactions happen. In the ideal case (fully compacted DB), retrieving a Set requires just a single disk read.

  • "Map" (aka "MapCollection")

    Maps are similar to Sets in the sense that for a single key there are multiple values. However, each value is in itself a key-value pair. This makes this type very similar to a dict or hashmap type. For example for key "foo", you could have value pairs: "bar":17, "baz":19.

    This makes a map a great use case for an inverted index that needs to store additional info beyond just the docid-pointer, such as in the case of a BM25 index where the term frequency needs to be stored.

    The same performance-considerations as for sets apply.

Navigate around these docs

Good entrypoints to learn more about how this package works include Store with New and Store.CreateOrLoadBucket, as well as Bucket with Bucket.Get, Bucket.GetBySecondary, Bucket.Put, etc.

Each strategy also supports cursor types: CursorReplace can be created using Bucket.Cursor, CursorSet can be created with Bucket.SetCursor , and CursorMap can be created with Bucket.MapCursor.

Index

Constants

View Source 
const (
	// StrategyReplace allows for idem-potent PUT where the latest takes presence
	StrategyReplace       = "replace"
	StrategySetCollection = "setcollection"
	StrategyMapCollection = "mapcollection"
	StrategyRoaringSet    = "roaringset"
)
View Source 
const CurrentVersion uint8 = 1
View Source 
const DeleteMarkerSuffix = ".deleteme"
View Source 
const FlushAfterDirtyDefault = 60 * time.Second

Variables

View Source 
var ErrAlreadyClosed = errors.New("store already closed")
View Source 
var ErrBucketAlreadyRegistered = errors.New("bucket already registered")
View Source 
var ErrInvalidChecksum = errors.New("invalid checksum")

ErrInvalidChecksum indicates that the read file should not be trusted. For any pre-computed data this is a recoverable issue, as the data can simply be re-computed at read-time.

View Source 
var GlobalBucketRegistry *globalBucketRegistry

Functions

func CheckExpectedStrategy added in v1.18.0 

func CheckExpectedStrategy(strategy string, expectedStrategies ...string)

func IsExpectedStrategy added in v1.18.0 

func IsExpectedStrategy(strategy string, expectedStrategies ...string) bool

func ParseCollectionNode 

func ParseCollectionNode(r io.Reader) (segmentCollectionNode, error)

ParseCollectionNode reads from r and parses the collection values into a segmentCollectionNode

When only given an offset, r is constructed as a *bufio.Reader to avoid first reading the entire segment (could be GBs). Each consecutive read will be buffered to avoid excessive syscalls.

When we already have a finite and manageable []byte (i.e. when we have already seeked to an lsmkv node and have start+end offset), r should be constructed as a *bytes.Reader, since the contents have already been `pread` from the segment contentFile.

func ParseCollectionNodeInto 

func ParseCollectionNodeInto(r io.Reader, node *segmentCollectionNode) error

ParseCollectionNodeInto takes the []byte slice and parses it into the specified node. It does not perform any copies and the caller must be aware that memory may be shared between the two. As a result, the caller must make sure that they do not modify "in" while "node" is still in use. A safer alternative is to use ParseCollectionNode.

The primary intention of this function is to provide a way to reuse buffers when the lifetime is controlled tightly, for example in cursors used within compactions. Use at your own risk!

If the buffers of the provided node have enough capacity they will be reused. Only if the capacity is not enough, will an allocation occur. This allocation uses 25% overhead to avoid future allocations for nodes of similar size.

As a result calling this method only makes sense if you plan on calling it multiple times. Calling it just once on an uninitialized node does not have major advantages over calling ParseCollectionNode.

func ParseReplaceNode 

func ParseReplaceNode(r io.Reader, secondaryIndexCount uint16) (segmentReplaceNode, error)

func ParseReplaceNodeIntoMMAP added in v1.23.0 

func ParseReplaceNodeIntoMMAP(r *byteops.ReadWriter, secondaryIndexCount uint16, out *segmentReplaceNode) error

func ParseReplaceNodeIntoPread added in v1.23.0 

func ParseReplaceNodeIntoPread(r io.Reader, secondaryIndexCount uint16, out *segmentReplaceNode) (err error)

func SegmentStrategyFromString 

func SegmentStrategyFromString(in string) segmentindex.Strategy

Types

type Bucket 

type Bucket struct {
	// contains filtered or unexported fields
}

func NewBucketCreator added in v1.25.0 

func NewBucketCreator() *Bucket

func (*Bucket) Count 

func (b *Bucket) Count() int

func (*Bucket) CountAsync added in v1.23.10 

func (b *Bucket) CountAsync() int

CountAsync ignores the current memtable, that makes it async because it only reflects what has been already flushed. This in turn makes it very cheap to call, so it can be used for observability purposes where eventual consistency on the count is fine, but a large cost is not.

func (*Bucket) Cursor 

func (b *Bucket) Cursor() *CursorReplace

Cursor holds a RLock for the flushing state. It needs to be closed using the .Close() methods or otherwise the lock will never be released

func (*Bucket) CursorRoaringSet added in v1.18.0 

func (b *Bucket) CursorRoaringSet() CursorRoaringSet

func (*Bucket) CursorRoaringSetKeyOnly added in v1.18.0 

func (b *Bucket) CursorRoaringSetKeyOnly() CursorRoaringSet

func (*Bucket) Delete 

func (b *Bucket) Delete(key []byte, opts ...SecondaryKeyOption) error

Delete removes the given row. Note that LSM stores are append only, thus internally this action appends a tombstone. The entry will not be removed until a compaction has run, and even then a compaction does not guarantee the removal of the data right away. This is because an entry could have been created in an older segment than those present in the compaction. This can be seen as an implementation detail, unless the caller expects to free disk space by calling this method. Such freeing is not guaranteed.

Delete is specific to the Replace Strategy. For Maps, you can use Bucket.MapDeleteKey to delete a single key-value pair, for Sets use Bucket.SetDeleteSingle to delete a single set element.

func (*Bucket) DesiredStrategy added in v1.18.0 

func (b *Bucket) DesiredStrategy() string

func (*Bucket) DocPointerWithScoreList added in v1.24.25 

func (b *Bucket) DocPointerWithScoreList(ctx context.Context, key []byte, propBoost float32, cfgs ...MapListOption) ([]terms.DocPointerWithScore, error)

func (*Bucket) FlushAndSwitch 

func (b *Bucket) FlushAndSwitch() error

FlushAndSwitch is the main way to flush a memtable, replace it with a new one, and make sure that the flushed segment gets added to the segment group.

Flushing and adding a segment can take considerable time, which is why the whole process is designed to be non-blocking.

To achieve a non-blocking flush, the process is split into four parts:

  1. atomicallySwitchMemtable: A new memtable is created, the previous memtable is moved from "active" to "flushing". This switch is blocking (holds b.flushLock.Lock()), but extremely fast, as we essentially just switch a pointer.

  2. flush: The previous memtable is flushed to disk. This may take considerable time as we are I/O-bound. This is done "in the background"meaning that it does not block any CRUD operations for the user. It only blocks the flush process itself, meaning only one flush per bucket can happen simultaneously. This is by design.

  3. initAndPrecomputeNewSegment: (Newly added in https://github.com/weaviate/weaviate/pull/5943, early October 2024). After the previous flush step the segment can now be initialized. However, to make it usable for real life, we still need to compute metadata, such as bloom filters (all types) and net count additions (only Replace type). Bloom filters can be calculated in isolation and are therefore fairly trivial. Net count additions on the other hand are more complex, as they depend on all previous segments. Calculating net count additions can take a considerable amount of time, especially as the buckets grow larger. As a result, we need to provide two guarantees: (1) the calculation is non-blocking from a user's POV and (2) for the duration of the calculation, the segment group is considered stable, i.e. no other segments are added, removed, or merged. We can achieve this by holding a `b.disk.maintenanceLock.RLock()` which prevents modification of the segments array, but does not block user operation (which are themselves RLock-holders on that same Lock).

  4. atomicallyAddDiskSegmentAndRemoveFlushing: The previous method returned a fully initialized segment that has not yet been added to the segment group. This last step is the counter part to the first step and again blocking, but fast. It adds the segment to the segment group  which at this point is just a simple array append. At the same time it removes the "flushing" memtable. It holds the `b.flushLock.Lock()` making this operation atomic, but blocking.

FlushAndSwitch is typically called periodically and does not require manual calling, but there are some situations where this might be intended, such as in test scenarios or when a force flush is desired.

func (*Bucket) FlushMemtable 

func (b *Bucket) FlushMemtable() error

FlushMemtable flushes any active memtable and returns only once the memtable has been fully flushed and a stable state on disk has been reached.

This is a preparatory stage for creating backups.

Method should be run only if flushCycle is not running (was not started, is stopped, or noop impl is provided)

func (*Bucket) Get 

func (b *Bucket) Get(key []byte) ([]byte, error)

Get retrieves the single value for the given key.

Get is specific to ReplaceStrategy and cannot be used with any of the other strategies. Use Bucket.SetList or Bucket.MapList instead.

Get uses the regular or "primary" key for an object. If a bucket has secondary indexes, use Bucket.GetBySecondary to retrieve an object using its secondary key

func (*Bucket) GetBySecondary 

func (b *Bucket) GetBySecondary(pos int, key []byte) ([]byte, error)

GetBySecondary retrieves an object using one of its secondary keys. A bucket can have an infinite number of secondary keys. Specify the secondary key position as the first argument.

A real-life example of secondary keys is the Weaviate object store. Objects are stored with the user-facing ID as their primary key and with the doc-id (an ever-increasing uint64) as the secondary key.

Similar to Bucket.Get, GetBySecondary is limited to ReplaceStrategy. No equivalent exists for Set and Map, as those do not support secondary indexes.

func (*Bucket) GetBySecondaryIntoMemory added in v1.20.0 

func (b *Bucket) GetBySecondaryIntoMemory(pos int, key []byte, buffer []byte) ([]byte, []byte, error)

GetBySecondaryIntoMemory copies into the specified memory, and retrieves an object using one of its secondary keys. A bucket can have an infinite number of secondary keys. Specify the secondary key position as the first argument.

A real-life example of secondary keys is the Weaviate object store. Objects are stored with the user-facing ID as their primary key and with the doc-id (an ever-increasing uint64) as the secondary key.

Similar to Bucket.Get, GetBySecondary is limited to ReplaceStrategy. No equivalent exists for Set and Map, as those do not support secondary indexes.

func (*Bucket) GetBySecondaryWithBuffer added in v1.22.6 

func (b *Bucket) GetBySecondaryWithBuffer(pos int, key []byte, buf []byte) ([]byte, []byte, error)

GetBySecondaryWithBuffer is like Bucket.GetBySecondary, but also takes a buffer. It's in the response of the caller to pool the buffer, since the bucket does not know when the caller is done using it. The return bytes will likely point to the same memory that's part of the buffer. However, if the buffer is to small, a larger buffer may also be returned (second arg).

func (*Bucket) GetDesiredStrategy added in v1.22.0 

func (b *Bucket) GetDesiredStrategy() string

func (*Bucket) GetDir added in v1.22.0 

func (b *Bucket) GetDir() string

func (*Bucket) GetErrDeleted added in v1.23.15 

func (b *Bucket) GetErrDeleted(key []byte) ([]byte, error)

func (*Bucket) GetFlushCallbackCtrl added in v1.22.0 

func (b *Bucket) GetFlushCallbackCtrl() cyclemanager.CycleCallbackCtrl

func (*Bucket) GetMemtableThreshold added in v1.22.0 

func (b *Bucket) GetMemtableThreshold() uint64

func (*Bucket) GetRootDir added in v1.22.0 

func (b *Bucket) GetRootDir() string

func (*Bucket) GetSecondaryIndices added in v1.22.0 

func (b *Bucket) GetSecondaryIndices() uint16

func (*Bucket) GetStatus added in v1.22.0 

func (b *Bucket) GetStatus() storagestate.Status

func (*Bucket) GetStrategy added in v1.22.0 

func (b *Bucket) GetStrategy() string

func (*Bucket) GetWalThreshold added in v1.22.0 

func (b *Bucket) GetWalThreshold() uint64

func (*Bucket) IterateMapObjects added in v1.22.0 

func (b *Bucket) IterateMapObjects(ctx context.Context, f func([]byte, []byte, []byte, bool) error) error

func (*Bucket) IterateObjects 

func (b *Bucket) IterateObjects(ctx context.Context, f func(object *storobj.Object) error) error

func (*Bucket) ListFiles 

func (b *Bucket) ListFiles(ctx context.Context, basePath string) ([]string, error)

ListFiles lists all files that currently exist in the Bucket. The files are only in a stable state if the memtable is empty, and if compactions are paused. If one of those conditions is not given, it errors

func (*Bucket) MapCursor 

func (b *Bucket) MapCursor(cfgs ...MapListOption) *CursorMap

func (*Bucket) MapCursorKeyOnly 

func (b *Bucket) MapCursorKeyOnly(cfgs ...MapListOption) *CursorMap

func (*Bucket) MapDeleteKey 

func (b *Bucket) MapDeleteKey(rowKey, mapKey []byte) error

MapDeleteKey removes one key-value pair from the given map row. Note that LSM stores are append only, thus internally this action appends a tombstone. The entry will not be removed until a compaction has run, and even then a compaction does not guarantee the removal of the data right away. This is because an entry could have been created in an older segment than those present in the compaction. This can be seen as an implementation detail, unless the caller expects to free disk space by calling this method. Such freeing is not guaranteed.

MapDeleteKey is specific to the Map Strategy. For Replace, you can use Bucket.Delete to delete the entire row, for Sets use Bucket.SetDeleteSingle to delete a single set element.

func (*Bucket) MapList 

func (b *Bucket) MapList(ctx context.Context, key []byte, cfgs ...MapListOption) ([]MapPair, error)

MapList returns all map entries for a given row key. The order of map pairs has no specific meaning. For efficient merge operations, pair entries are stored sorted on disk, however that is an implementation detail and not a caller-facing guarantee.

MapList is specific to the Map strategy, for Sets use Bucket.SetList, for Replace use Bucket.Get.

func (*Bucket) MapSet 

func (b *Bucket) MapSet(rowKey []byte, kv MapPair) error

MapSet writes one MapPair into the map for the given row key. It is agnostic of whether the row key already exists, as well as agnostic of whether the map key already exists. In both cases it will create the entry if it does not exist or override if it does.

Example to add a new MapPair: 

pair := MapPair{Key: []byte("Jane"), Value: []byte("Backend")}
err := bucket.MapSet([]byte("developers"), pair)
if err != nil {
	/* do something */
}

MapSet is specific to the Map Strategy, for Replace use Bucket.Put, and for Set use Bucket.SetAdd instead.

func (*Bucket) MapSetMulti 

func (b *Bucket) MapSetMulti(rowKey []byte, kvs []MapPair) error

MapSetMulti is the same as Bucket.MapSet, except that it takes in multiple MapPair objects at the same time.

func (*Bucket) NewBucket added in v1.25.0 

func (*Bucket) NewBucket(ctx context.Context, dir, rootDir string, logger logrus.FieldLogger,
	metrics *Metrics, compactionCallbacks, flushCallbacks cyclemanager.CycleCallbackGroup,
	opts ...BucketOption,
) (*Bucket, error)

NewBucket initializes a new bucket. It either loads the state from disk if it exists, or initializes new state.

You do not need to ever call NewBucket() yourself, if you are using a Store. In this case the Store can manage buckets for you, using methods such as CreateOrLoadBucket().

func (*Bucket) Put 

func (b *Bucket) Put(key, value []byte, opts ...SecondaryKeyOption) error

Put creates or replaces a single value for a given key. 

err := bucket.Put([]byte("my_key"), []byte("my_value"))
 if err != nil {
	/* do something */
}

If a bucket has a secondary index configured, you can also specify one or more secondary keys, like so: 

err := bucket.Put([]byte("my_key"), []byte("my_value"),
	WithSecondaryKey(0, []byte("my_alternative_key")),
)
 if err != nil {
	/* do something */
}

Put is limited to ReplaceStrategy, use Bucket.SetAdd for Set or Bucket.MapSet and Bucket.MapSetMulti.

func (*Bucket) QuantileKeys added in v1.25.11 

func (b *Bucket) QuantileKeys(q int) [][]byte

QuantileKeys returns an approximation of the keys that make up the specified quantiles. This can be used to start parallel cursors at fairly evenly distributed positions in the segment.

To understand the approximation, checkout [lsmkv.segmentindex.DiskTree.QuantileKeys] that runs on each segment.

Some things to keep in mind:

  1. It may return fewer keys than requested (including 0) if the segment contains fewer entries
  2. It may return keys that do not exist, for example because they are tombstoned. This is acceptable, as a key does not have to exist to be used as part of .Seek() in a cursor.
  3. It will never return duplicates, to make sure all parallel cursors return unique values.

func (*Bucket) RoaringSetAddBitmap added in v1.18.0 

func (b *Bucket) RoaringSetAddBitmap(key []byte, bm *sroar.Bitmap) error

func (*Bucket) RoaringSetAddList added in v1.18.0 

func (b *Bucket) RoaringSetAddList(key []byte, values []uint64) error

func (*Bucket) RoaringSetAddOne added in v1.18.0 

func (b *Bucket) RoaringSetAddOne(key []byte, value uint64) error

func (*Bucket) RoaringSetGet added in v1.18.0 

func (b *Bucket) RoaringSetGet(key []byte) (*sroar.Bitmap, error)

func (*Bucket) RoaringSetRemoveOne added in v1.18.0 

func (b *Bucket) RoaringSetRemoveOne(key []byte, value uint64) error

func (*Bucket) SetAdd 

func (b *Bucket) SetAdd(key []byte, values [][]byte) error

SetAdd adds one or more Set-Entries to a Set for the given key. SetAdd is entirely agnostic of existing entries, it acts as append-only. This also makes it agnostic of whether the key already exists or not.

Example to add two entries to a set: 

err := bucket.SetAdd([]byte("my_key"), [][]byte{
	[]byte("one-set-element"), []byte("another-set-element"),
})
if err != nil {
	/* do something */
}

SetAdd is specific to the Set strategy. For Replace, use Bucket.Put, for Map use either Bucket.MapSet or Bucket.MapSetMulti.

func (*Bucket) SetCursor 

func (b *Bucket) SetCursor() *CursorSet

SetCursor holds a RLock for the flushing state. It needs to be closed using the .Close() methods or otherwise the lock will never be released

func (*Bucket) SetCursorKeyOnly 

func (b *Bucket) SetCursorKeyOnly() *CursorSet

SetCursorKeyOnly returns nil for all values. It has no control over the underlying "inner" cursors which may still retrieve a value which is then discarded. It does however, omit any handling of values, such as decoding, making this considerably more efficient if only keys are required.

The same locking rules as for SetCursor apply.

func (*Bucket) SetDeleteSingle 

func (b *Bucket) SetDeleteSingle(key []byte, valueToDelete []byte) error

SetDeleteSingle removes one Set element from the given key. Note that LSM stores are append only, thus internally this action appends a tombstone. The entry will not be removed until a compaction has run, and even then a compaction does not guarantee the removal of the data right away. This is because an entry could have been created in an older segment than those present in the compaction. This can be seen as an implementation detail, unless the caller expects to free disk space by calling this method. Such freeing is not guaranteed.

SetDeleteSingle is specific to the Set Strategy. For Replace, you can use Bucket.Delete to delete the entire row, for Maps use Bucket.MapDeleteKey to delete a single map entry.

func (*Bucket) SetList 

func (b *Bucket) SetList(key []byte) ([][]byte, error)

SetList returns all Set entries for a given key.

SetList is specific to the Set Strategy, for Map use Bucket.MapList, and for Replace use Bucket.Get.

func (*Bucket) SetMemtableThreshold 

func (b *Bucket) SetMemtableThreshold(size uint64)

func (*Bucket) Shutdown 

func (b *Bucket) Shutdown(ctx context.Context) error

func (*Bucket) Strategy 

func (b *Bucket) Strategy() string

func (*Bucket) UpdateStatus 

func (b *Bucket) UpdateStatus(status storagestate.Status)

UpdateStatus is used by the parent shard to communicate to the bucket when the shard has been set to readonly, or when it is ready for writes.

func (*Bucket) WasDeleted added in v1.18.0 

func (b *Bucket) WasDeleted(key []byte) (bool, error)

WasDeleted determines if an object used to exist in the LSM store

There are 3 different locations that we need to check for the key in this order: active memtable, flushing memtable, and disk segment

func (*Bucket) WriteWAL 

func (b *Bucket) WriteWAL() error

the WAL uses a buffer and isn't written until the buffer size is crossed or this function explicitly called. This allows to avoid unnecessary disk writes in larger operations, such as batches. It is sufficient to call write on the WAL just once. This does not make a batch atomic, but it guarantees that the WAL is written before a successful response is returned to the user.

type BucketCreator added in v1.25.0 

type BucketCreator interface {
	NewBucket(ctx context.Context, dir, rootDir string, logger logrus.FieldLogger,
		metrics *Metrics, compactionCallbacks, flushCallbacks cyclemanager.CycleCallbackGroup,
		opts ...BucketOption,
	) (*Bucket, error)
}

type BucketOption 

type BucketOption func(b *Bucket) error

func WithAllocChecker added in v1.24.9 

func WithAllocChecker(mm memwatch.AllocChecker) BucketOption

func WithCalcCountNetAdditions added in v1.23.0 

func WithCalcCountNetAdditions(calcCountNetAdditions bool) BucketOption

func WithDirtyThreshold added in v1.24.2 

func WithDirtyThreshold(threshold time.Duration) BucketOption

func WithDynamicMemtableSizing 

func WithDynamicMemtableSizing(
	initialMB, maxMB, minActiveSeconds, maxActiveSeconds int,
) BucketOption

func WithForceCompation added in v1.23.0 

func WithForceCompation(opt bool) BucketOption

Background for this option:

We use the LSM store in two places: Our existing key/value and inverted buckets As part of the new brute-force based index (to be built this week).

Brute-force index This is a simple disk-index where we use a cursor to iterate over all objects. This is what we need the force-compaction for. The experimentation so far has shown that the cursor is much more performant on a single segment than it is on multiple segments. This is because with a single segment it’s essentially just one conitiguuous chunk of data on disk that we read through. But with multiple segments (and an unpredicatable order) it ends up being many tiny reads (inefficient). Existing uses of the LSM store For existing uses, e.g. the object store, we don’t want to force-compact. This is because they can grow massive. For example, you could have a 100GB segment, then a new write leads to a new segment that is just a few bytes. If we would force-compact those two we would write 100GB every time the user sends a few bytes to Weaviate. In this case, the existing tiered compaction strategy makes more sense. Configurability of buckets

func WithKeepTombstones added in v1.23.0 

func WithKeepTombstones(keepTombstones bool) BucketOption

func WithLegacyMapSorting 

func WithLegacyMapSorting() BucketOption

func WithMaxSegmentSize added in v1.24.12 

func WithMaxSegmentSize(maxSegmentSize int64) BucketOption

func WithMemtableThreshold 

func WithMemtableThreshold(threshold uint64) BucketOption

func WithMonitorCount 

func WithMonitorCount() BucketOption

func WithPread added in v1.21.0 

func WithPread(with bool) BucketOption

func WithSecondaryIndices 

func WithSecondaryIndices(count uint16) BucketOption

func WithSegmentsCleanupInterval added in v1.25.21 

func WithSegmentsCleanupInterval(interval time.Duration) BucketOption

func WithStrategy 

func WithStrategy(strategy string) BucketOption

func WithUseBloomFilter added in v1.23.0 

func WithUseBloomFilter(useBloomFilter bool) BucketOption

func WithWalThreshold 

func WithWalThreshold(threshold uint64) BucketOption

type CommitType 

type CommitType uint8
const (
	CommitTypeReplace CommitType = iota // replace strategy

	// collection strategy - this can handle all cases as updates and deletes are
	// only appends in a collection strategy
	CommitTypeCollection
	CommitTypeRoaringSet
	// new version of roaringset that stores data as a list of uint64 values,
	// instead of a roaring bitmap
	CommitTypeRoaringSetList
)

func (CommitType) Is added in v1.18.0 

func (ct CommitType) Is(checkedCommitType CommitType) bool

func (CommitType) String added in v1.18.0 

func (ct CommitType) String() string

type CursorMap 

type CursorMap struct {
	// contains filtered or unexported fields
}

func (*CursorMap) Close 

func (c *CursorMap) Close()

func (*CursorMap) First 

func (c *CursorMap) First(ctx context.Context) ([]byte, []MapPair)

func (*CursorMap) Next 

func (c *CursorMap) Next(ctx context.Context) ([]byte, []MapPair)

func (*CursorMap) Seek 

func (c *CursorMap) Seek(ctx context.Context, key []byte) ([]byte, []MapPair)

type CursorReplace 

type CursorReplace struct {
	// contains filtered or unexported fields
}

func (*CursorReplace) Close 

func (c *CursorReplace) Close()

func (*CursorReplace) First 

func (c *CursorReplace) First() ([]byte, []byte)

func (*CursorReplace) Next 

func (c *CursorReplace) Next() ([]byte, []byte)

func (*CursorReplace) Seek 

func (c *CursorReplace) Seek(key []byte) ([]byte, []byte)

type CursorRoaringSet added in v1.18.0 

type CursorRoaringSet interface {
	First() ([]byte, *sroar.Bitmap)
	Next() ([]byte, *sroar.Bitmap)
	Seek([]byte) ([]byte, *sroar.Bitmap)
	Close()
}

type CursorSet 

type CursorSet struct {
	// contains filtered or unexported fields
}

func (*CursorSet) Close 

func (c *CursorSet) Close()

func (*CursorSet) First 

func (c *CursorSet) First() ([]byte, [][]byte)

func (*CursorSet) Next 

func (c *CursorSet) Next() ([]byte, [][]byte)

func (*CursorSet) Seek 

func (c *CursorSet) Seek(key []byte) ([]byte, [][]byte)

type MapListOption 

type MapListOption func(c *MapListOptionConfig)

func MapListAcceptDuplicates 

func MapListAcceptDuplicates() MapListOption

func MapListLegacySortingRequired 

func MapListLegacySortingRequired() MapListOption

type MapListOptionConfig 

type MapListOptionConfig struct {
	// contains filtered or unexported fields
}

type MapPair 

type MapPair struct {
	Key       []byte
	Value     []byte
	Tombstone bool
}

func (MapPair) Bytes 

func (kv MapPair) Bytes() ([]byte, error)

func (MapPair) EncodeBytes 

func (kv MapPair) EncodeBytes(buf []byte) error

func (*MapPair) FromBytes 

func (kv *MapPair) FromBytes(in []byte, keyOnly bool) error

func (*MapPair) FromBytesReusable 

func (kv *MapPair) FromBytesReusable(in []byte, keyOnly bool) error

func (MapPair) Size 

func (kv MapPair) Size() int

Size() returns the exact size in bytes that will be used when Bytes() is called

type Memtable 

type Memtable struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*Memtable) ActiveDuration 

func (m *Memtable) ActiveDuration() time.Duration

func (*Memtable) DirtyDuration added in v1.24.2 

func (m *Memtable) DirtyDuration() time.Duration

returns time memtable got dirty (1st write occurred) (0 if clean)

func (*Memtable) Size 

func (m *Memtable) Size() uint64

type Metrics 

type Metrics struct {
	CompactionReplace    *prometheus.GaugeVec
	CompactionSet        *prometheus.GaugeVec
	CompactionMap        *prometheus.GaugeVec
	CompactionRoaringSet *prometheus.GaugeVec
	ActiveSegments       *prometheus.GaugeVec

	SegmentObjects *prometheus.GaugeVec
	SegmentSize    *prometheus.GaugeVec
	SegmentCount   *prometheus.GaugeVec

	DimensionSum *prometheus.GaugeVec
	// contains filtered or unexported fields
}

func NewMetrics 

func NewMetrics(promMetrics *monitoring.PrometheusMetrics, className,
	shardName string,
) *Metrics

func (*Metrics) BloomFilterObserver 

func (m *Metrics) BloomFilterObserver(strategy, operation string) TimeObserver

func (*Metrics) MemtableOpObserver 

func (m *Metrics) MemtableOpObserver(path, strategy, op string) NsObserver

func (*Metrics) MemtableSizeSetter 

func (m *Metrics) MemtableSizeSetter(path, strategy string) Setter

func (*Metrics) ObjectCount 

func (m *Metrics) ObjectCount(count int)

func (*Metrics) TrackStartupBucket 

func (m *Metrics) TrackStartupBucket(start time.Time)

func (*Metrics) TrackStartupBucketRecovery 

func (m *Metrics) TrackStartupBucketRecovery(start time.Time)

func (*Metrics) TrackStartupReadWALDiskIO 

func (m *Metrics) TrackStartupReadWALDiskIO(read int64, nanoseconds int64)

type MockBucketCreator added in v1.25.0 

type MockBucketCreator struct {
	mock.Mock
}

MockBucketCreator is an autogenerated mock type for the BucketCreator type

func NewMockBucketCreator added in v1.25.0 

func NewMockBucketCreator(t interface {
	mock.TestingT
	Cleanup(func())
},
) *MockBucketCreator

NewMockBucketCreator creates a new instance of MockBucketCreator. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. The first argument is typically a *testing.T value.

func (*MockBucketCreator) NewBucket added in v1.25.0 

func (_m *MockBucketCreator) NewBucket(ctx context.Context, dir string, rootDir string, logger logrus.FieldLogger, metrics *Metrics, compactionCallbacks cyclemanager.CycleCallbackGroup, flushCallbacks cyclemanager.CycleCallbackGroup, opts ...BucketOption) (*Bucket, error)

NewBucket provides a mock function with given fields: ctx, dir, rootDir, logger, metrics, compactionCallbacks, flushCallbacks, opts

type NsObserver 

type NsObserver func(ns int64)

type SecondaryKeyOption 

type SecondaryKeyOption func(s secondaryIndexKeys) error

func WithSecondaryKey 

func WithSecondaryKey(pos int, key []byte) SecondaryKeyOption

type SegmentGroup 

type SegmentGroup struct {
	// contains filtered or unexported fields
}

func (*SegmentGroup) Len 

func (sg *SegmentGroup) Len() int

func (*SegmentGroup) UpdateStatus added in v1.23.0 

func (sg *SegmentGroup) UpdateStatus(status storagestate.Status)

type Setter 

type Setter func(val uint64)

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store groups multiple buckets together, it "owns" one folder on the file system

func New 

func New(dir, rootDir string, logger logrus.FieldLogger, metrics *Metrics,
	shardCompactionCallbacks, shardFlushCallbacks cyclemanager.CycleCallbackGroup,
) (*Store, error)

New initializes a new Store based on the root dir. If state is present on disk, it is loaded, if the folder is empty a new store is initialized in there.

func (*Store) Bucket 

func (s *Store) Bucket(name string) *Bucket

func (*Store) CreateBucket added in v1.18.0 

func (s *Store) CreateBucket(ctx context.Context, bucketName string,
	opts ...BucketOption,
) error

Creates bucket, first removing any files if already exist Bucket can not be registered in bucketsByName before removal

func (*Store) CreateOrLoadBucket 

func (s *Store) CreateOrLoadBucket(ctx context.Context, bucketName string,
	opts ...BucketOption,
) (err error)

CreateOrLoadBucket registers a bucket with the given name. If state on disk exists for this bucket it is loaded, otherwise created. Pass [BucketOptions] to configure the strategy of a bucket. The strategy defaults to "replace". For example, to load or create a map-type bucket, do: 

ctx := context.Background()
err := store.CreateOrLoadBucket(ctx, "my_bucket_name", WithStrategy(StrategyReplace))
if err != nil { /* handle error */ }

// you can now access the bucket using store.Bucket()
b := store.Bucket("my_bucket_name")

func (*Store) FlushMemtables 

func (s *Store) FlushMemtables(ctx context.Context) error

FlushMemtable flushes any active memtable and returns only once the memtable has been fully flushed and a stable state on disk has been reached.

This is a preparatory stage for creating backups.

A timeout should be specified for the input context as some flushes are long-running, in which case it may be better to fail the backup attempt and retry later, than to block indefinitely.

func (*Store) GetBucketsByName added in v1.18.0 

func (s *Store) GetBucketsByName() map[string]*Bucket

func (*Store) ListFiles 

func (s *Store) ListFiles(ctx context.Context, basePath string) ([]string, error)

func (*Store) PauseCompaction 

func (s *Store) PauseCompaction(ctx context.Context) error

PauseCompaction waits for all ongoing compactions to finish, then makes sure that no new compaction can be started.

This is a preparatory stage for creating backups.

A timeout should be specified for the input context as some compactions are long-running, in which case it may be better to fail the backup attempt and retry later, than to block indefinitely.

func (*Store) RenameBucket added in v1.19.0 

func (s *Store) RenameBucket(ctx context.Context, bucketName, newBucketName string) error

func (*Store) ReplaceBuckets added in v1.18.0 

func (s *Store) ReplaceBuckets(ctx context.Context, bucketName, replacementBucketName string) error

Replaces 1st bucket with 2nd one. Both buckets have to registered in bucketsByName. 2nd bucket swaps the 1st one in bucketsByName using 1st one's name, 2nd one's name is deleted. Dir path of 2nd bucket is changed to dir of 1st bucket as well as all other related paths of bucket resources (segment group, memtables, commit log). Dir path of 1st bucket is temporarily suffixed with "___del", later on bucket is shutdown and its files deleted. 2nd bucket becomes 1st bucket

func (*Store) ResumeCompaction 

func (s *Store) ResumeCompaction(ctx context.Context) error

ResumeCompaction starts the compaction cycle again. It errors if compactions were not paused

func (*Store) Shutdown 

func (s *Store) Shutdown(ctx context.Context) error

func (*Store) UpdateBucketsStatus 

func (s *Store) UpdateBucketsStatus(targetStatus storagestate.Status) error

func (*Store) WriteWALs 

func (s *Store) WriteWALs() error

type TimeObserver 

type TimeObserver func(start time.Time)

Source Files

View all Source files

Directories

Path Synopsis
ent contains common types used throughout various lsmkv (sub-)packages
 ent contains common types used throughout various lsmkv (sub-)packages

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.