Home Original page

frostdb package - github.com/polarsignals/frostdb - Go Packages

View Source

const (
	B   = 1
	KiB = 1024 * B
	MiB = 1024 * KiB
	GiB = 1024 * MiB
	TiB = 1024 * GiB
)

DefaultBlockReaderLimit is the concurrency limit for reading blocks.

DefaultIndexConfig returns the default level configs used. This is a function So that any modifications to the result will not affect the default config. 

type CloseOption func(*closeOptions)
func WithClearStorage() CloseOption
type Closer interface {
	Close(cleanup bool) error
}

type ColumnStore 

type ColumnStore struct {
	
}
func New(
	options ...Option,
) (*ColumnStore, error)

func (*ColumnStore) Close

Close persists all data from the columnstore to storage. It is no longer valid to use the coumnstore for reads or writes, and the object should not longer be reused.

func (*ColumnStore) DB

DB gets or creates a database on the given ColumnStore with the given options. Note that if the database already exists, the options will be applied cumulatively to the database.

func (*ColumnStore) DBs

DBs returns all the DB names of this column store.

func (*ColumnStore) GetDB 

func (db *DB) Close(options ...CloseOption) error
func (db *DB) HighWatermark() uint64

HighWatermark returns the current high watermark.

Snapshot performs a database snapshot and writes it to the database snapshots directory, as is done by automatic snapshots.

Table will get or create a new table with the given name and config. If a table already exists with the given name, it will have it's configuration updated. 

func (db *DB) TableNames() []string

TableNames returns the names of all the db's tables. 

func (db *DB) TableProvider() *DBTableProvider

Wait is a blocking function that returns once the high watermark has equaled or exceeded the transaction id. Wait makes no differentiation between completed and aborted transactions. 

type DBOption func(*DB) error
func WithCompactionAfterOpen(compact bool, tableNames []string) DBOption
type DBTableProvider struct {
	
}
func NewDBTableProvider(db *DB) *DBTableProvider

DataSink is a remote destination for data. 

type DataSinkSource interface {
	DataSink
	DataSource
}

DataSinkSource is a convenience interface for a data source and sink.

DataSource is remote source of data that can be queried.

DefaultObjstoreBucket is the default implementation of the DataSource and DataSink interface.

ProcessFile will process a bucket block parquet file. 

type DefaultObjstoreBucketOption func(*DefaultObjstoreBucket)
func StorageWithBlockReaderLimit(limit int) DefaultObjstoreBucketOption
type ErrCreateSchemaWriter struct {
	
}
type ErrReadRow struct {
	
}
type ErrTableNotFound struct {
	TableName string
}
type ErrWriteRow struct {
	
}
type GenericTable[T any] struct {
	*Table
	
}

GenericTable is a wrapper around *Table that writes structs of type T. It consist of a generic arrow.Record builder that ingests structs of type T. The generated record is then passed to (*Table).InsertRecord.

Struct tag `frostdb` is used to pass options for the schema for T.

This api is opinionated.

  • Nested Columns are not supported

Tags

Use `frostdb` to define tags that customizes field values. You can express everything needed to construct schema v1alpha1.

Tags are defined as a comma separated list. The first item is the column name. Column name is optional, when omitted it is derived from the field name (snake_cased)

Supported Tags 

    delta_binary_packed | Delta binary packed encoding.
                 brotli | Brotli compression.
                    asc | Sorts in ascending order.Use asc(n) where n is an integer for sorting order.
                   gzip | GZIP compression.
                 snappy | Snappy compression.
delta_length_byte_array | Delta Length Byte Array encoding.
       delta_byte_array | Delta Byte Array encoding.
                   desc | Sorts in descending order.Use desc(n) where n is an integer for sorting order
                lz4_raw | LZ4_RAW compression.
               pre_hash | Prehash the column before storing it.
             null_first | When used wit asc nulls are smallest and with des nulls are largest.
                   zstd | ZSTD compression.
               rle_dict | Dictionary run-length encoding.
                  plain | Plain encoding.

Example tagged Sample struct 

type Sample struct {
	ExampleType string      `frostdb:"example_type,rle_dict,asc(0)"`
	Labels      []Label     `frostdb:"labels,rle_dict,null,dyn,asc(1),null_first"`
	Stacktrace  []uuid.UUID `frostdb:"stacktrace,rle_dict,asc(3),null_first"`
	Timestamp   int64       `frostdb:"timestamp,asc(2)"`
	Value       int64       `frostdb:"value"`
}

Dynamic columns

Field of type map<string, T> is a dynamic column by default. 

type Example struct {
	// Use supported tags to customize the column value
	Labels map[string]string `frostdb:"labels"`
}

Repeated columns

Fields of type []int64, []float64, []bool, and []string are supported. These are represented as arrow.LIST.

Generated schema for the repeated columns applies all supported tags. By default repeated fields are nullable. You can safely pass nil slices for repeated columns. 

func (t *GenericTable[T]) Release()

Write builds arrow.Record directly from values and calls (*Table).InsertRecord. 

func WithActiveMemorySize(size int64) Option
func WithCompactionAfterRecovery(tableNames []string) Option
func WithIndexDegree(indexDegree int) Option
func WithManualBlockRotation() Option
func WithReadOnlyStorage(ds DataSource) Option
func WithReadWriteStorage(ds DataSinkSource) Option
func WithRecoveryConcurrency(concurrency int) Option

WithRecoveryConcurrency limits the number of databases that are recovered simultaneously when calling frostdb.New. This helps limit memory usage on recovery. 

func WithSnapshotTriggerSize(size int64) Option

WithSnapshotTriggerSize specifies a size in bytes of uncompressed inserts that will trigger a snapshot of the whole database. This can be larger than the active memory size given that the active memory size tracks the size of *compressed* data, while snapshots are triggered based on the *uncompressed* data inserted into the database. The reason this choice was made is that if a database instance crashes, it is forced to reread all uncompressed inserts since the last snapshot from the WAL, which could potentially lead to unrecoverable OOMs on startup. Defining the snapshot trigger in terms of uncompressed bytes limits the memory usage on recovery to at most the snapshot trigger size (as long as snapshots were successful). If 0, snapshots are disabled. Note that snapshots (if enabled) are also triggered on block rotation of any database table. Snapshots are complementary to the WAL and will also be disabled if the WAL is disabled. 

func WithSplitSize(size int) Option
func WithStoragePath(path string) Option
func WithTestingOptions(opts ...TestingOption) Option
func WithWriteOnlyStorage(ds DataSink) Option
type RotateBlockOption func(*rotateBlockOptions)
func WithRotateBlockSkipPersist() RotateBlockOption

WithRotateBlockSkipPersist instructs the block rotation operation to not persist the block to object storage.

WithRotateBlockWaitGroup provides a WaitGroup. The rotate block operation will call wg.Done once the block has been persisted. Otherwise, RotateBlock asynchronously persists the block. 

type Sync interface {
	Sync() error
}
func (t *Table) ActiveBlock() *TableBlock
func (t *Table) ActiveWriteBlock() (*TableBlock, func(), error)
func (t *Table) EnsureCompaction() error

IndexConfig returns the index configuration for the table. It makes a copy of the column store index config and injects it's compactParts method.

Iterator iterates in order over all granules in the table. It stops iterating when the iterator function returns false.

SchemaIterator iterates in order over all granules in the table and returns all the schemas seen across the table. 

type TableBlock struct {
	
}
func (t *TableBlock) EnsureCompaction() error

EnsureCompaction forces a TableBlock compaction.

Index provides atomic access to the table index.

Persist uploads the block to the underlying bucket.

Serialize the table block into a single Parquet file.

Size returns the cumulative size of all buffers in the table. This is roughly the size of the table in bytes.

FromConfig sets the table configuration from the given config. NOTE: that this does not override the schema even though that is included in the passed in config. 

func WithBlockReaderLimit(n int) TableOption

WithBlockReaderLimit sets the limit of go routines that will be used to read persisted block files. A negative number indicates no limit. 

func WithRowGroupSize(numRows int) TableOption

WithRowGroupSize sets the size in number of rows for each row group for parquet files. A <= 0 value indicates no limit. 

func WithUniquePrimaryIndex(unique bool) TableOption
func WithoutWAL() TableOption

WithoutWAL disables the WAL for this table. 

type TestingOption Option
func WithTestingNoDiskSpaceReclaimOnSnapshot() TestingOption

NewTxPool returns a new TxPool and starts the pool cleaner routine. The transaction pool is used to keep track of completed transactions. It does this by inserting completed transactions into an ordered linked list.

Ex: insert: 12 [9]->[10]->[13] => [9]->[10]->[12]->[13]

Inserting a new node triggers the pool cleaner routine to run. The pool cleaner's job is to increment a high-watermark counter when it encounters contiguous transactions in the list, and then remove those elements in the pool.

Ex: watermark: 7 insert: 8 [9]->[10]->[13] => [8]->[9]->[10]->[13] (cleaner notified)

[8]->[9]->[10]->[13] 

^ watermark++; delete 8

[9]->[10]->[13] 

^ watermark++; delete 9

[10]->[13] 

^ watermark++; delete 9

[13] watermark: 10

TxPool is a sorted lockless linked-list described in https://timharris.uk/papers/2001-disc.pdf

Insert performs an insertion sort of the given tx.

Stop stops the TxPool's cleaner goroutine.