store

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Jul 17, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Overview

Package store is the DynamoDB storage engine: tables and items in bbolt, GSI/LSI entries maintained in the same transaction as every write, range queries via order-preserving key encodings (keyenc), single-node transactions with real atomicity, and TTL enforcement.

bbolt layout: 

tables            name -> Table JSON
d:<table>         keyenc(pk[,sk]) -> item wire JSON
x:<table>/<index> keyenc(ipk[,isk]) ++ keyenc(pk[,sk]) -> primary key bytes
tx:               ClientRequestToken -> {hash, expiry} (idempotency)

Index entries are key references (no projected copies): reads chase the reference to the base item and apply the index projection — microseconds locally, and index consistency is free.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func RequestHash 

func RequestHash(body []byte) string

RequestHash fingerprints a transaction body for idempotency comparison.

Types

type CancellationReason 

type CancellationReason struct {
	Code    string          `json:"Code"`
	Message string          `json:"Message,omitempty"`
	Item    json.RawMessage `json:"Item,omitempty"`
}

CancellationReason mirrors DynamoDB's per-item transaction outcome.

type Cond 

type Cond struct {
	Expr *expr.Cond
	Env  *expr.Env
	// ReturnOld controls ReturnValuesOnConditionCheckFailure.
	ReturnOld bool
}

Cond wraps an optional parsed condition.

type ErrTransactionCanceled 

type ErrTransactionCanceled struct {
	Reasons []CancellationReason
}

ErrTransactionCanceled carries the reasons; the service layer renders it.

func (*ErrTransactionCanceled) Error 

func (e *ErrTransactionCanceled) Error() string

type Index 

type Index struct {
	Name        string   `json:"name"`
	Hash        KeyPart  `json:"hash"`
	Range       *KeyPart `json:"range,omitempty"`
	Projection  string   `json:"projection"` // ALL | KEYS_ONLY | INCLUDE
	NonKeyAttrs []string `json:"non_key_attrs,omitempty"`
	Local       bool     `json:"local"`
}

Index describes a GSI or LSI.

type KeyPart 

type KeyPart struct {
	Name string `json:"name"`
	Type string `json:"type"` // S | N | B
}

KeyPart names one key attribute.

type QueryInput 

type QueryInput struct {
	Table    string
	Index    string // optional
	KeyCond  *expr.KeyCondition
	Filter   *expr.Cond
	Forward  bool // ScanIndexForward (default true)
	Limit    int
	StartKey json.RawMessage // ExclusiveStartKey (item key attrs, incl. index keys)
}

QueryInput drives Query.

type QueryOutput 

type QueryOutput struct {
	Items            []item.Item
	Count            int
	ScannedCount     int
	LastEvaluatedKey item.Item // nil when the page completes the result set
}

QueryOutput is a page of results.

type ScanInput 

type ScanInput struct {
	Table         string
	Index         string
	Filter        *expr.Cond
	Limit         int
	StartKey      json.RawMessage
	Segment       int
	TotalSegments int
}

ScanInput drives Scan.

type Store 

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

Store is the bbolt-backed DynamoDB engine.

func New 

func New(db *bolt.DB) *Store

New wraps an open bbolt DB.

func (*Store) CountItems 

func (s *Store) CountItems(table string) int64

CountItems reports a table's item count (DescribeTable convenience).

func (*Store) CreateTable 

func (s *Store) CreateTable(t Table) (*Table, error)

CreateTable registers a table.

func (*Store) DB 

func (s *Store) DB() *bolt.DB

DB exposes the handle for Close.

func (*Store) DeleteItem 

func (s *Store) DeleteItem(table string, rawKey json.RawMessage, cond *Cond, _ string) (old item.Item, err error)

DeleteItem removes an item, returning the previous one.

func (*Store) DeleteTable 

func (s *Store) DeleteTable(name string) (*Table, error)

DeleteTable removes a table and its data.

func (*Store) GetItem 

func (s *Store) GetItem(table string, rawKey json.RawMessage) (item.Item, error)

GetItem fetches an item by key.

func (*Store) GetTable 

func (s *Store) GetTable(name string) (*Table, error)

GetTable loads a table definition.

func (*Store) KeyFromWire 

func (s *Store) KeyFromWire(t *Table, raw json.RawMessage) ([]byte, item.Item, *awshttp.APIError)

keyFromWire decodes a request's Key map and encodes it.

func (*Store) LatestStreamSeq added in v0.2.0 

func (s *Store) LatestStreamSeq(table string) uint64

LatestStreamSeq returns the highest stored sequence for a table's stream (0 if none) — the position a LATEST shard iterator starts after.

func (*Store) ListTables 

func (s *Store) ListTables() ([]string, error)

ListTables returns table names in order.

func (*Store) PutItem 

func (s *Store) PutItem(table string, rawItem json.RawMessage, cond *Cond) (old item.Item, err error)

PutItem stores an item, returning the previous one (for ReturnValues).

func (*Store) Query 

func (s *Store) Query(in QueryInput) (*QueryOutput, error)

Query runs a key-condition query against the table or one of its indexes.

func (*Store) Scan 

func (s *Store) Scan(in ScanInput) (*QueryOutput, error)

Scan walks a table (or index) with optional segment partitioning.

func (*Store) SetClock 

func (s *Store) SetClock(fn func() time.Time)

SetClock overrides the clock (tests).

func (*Store) StreamRecords added in v0.2.0 

func (s *Store) StreamRecords(table string, afterSeq uint64, limit int) ([]StreamRecord, uint64, error)

StreamRecords returns records with sequence > afterSeq (afterSeq 0 = from the trim horizon), up to limit, plus the highest sequence currently stored.

func (*Store) StreamViewType added in v0.2.0 

func (s *Store) StreamViewType(table string) (string, bool)

StreamViewType (on Store) resolves a table by name first.

func (*Store) SweepTTL 

func (s *Store) SweepTTL()

SweepTTL removes expired items across every TTL-enabled table, through the normal delete path so indexes stay consistent.

func (*Store) TransactGet 

func (s *Store) TransactGet(keys []struct {
	Table string
	Key   json.RawMessage
}) ([]item.Item, error)

TransactGet reads up to 100 items with a consistent view (one bbolt View).

func (*Store) TransactWrite 

func (s *Store) TransactWrite(ops []TxWriteOp, token string, requestHash string) error

TransactWrite runs up to 100 write ops atomically.

func (*Store) UpdateItem 

func (s *Store) UpdateItem(table string, rawKey json.RawMessage, upd *expr.Update, cond *Cond) (old, new item.Item, err error)

UpdateItem applies an update expression, returning (old, new) items.

func (*Store) UpdateTable 

func (s *Store) UpdateTable(name string, fn func(*Table) error) (*Table, error)

UpdateTable applies fn to a table definition. Adding a GSI triggers a synchronous backfill (local data volumes make this instant-ish).

type StreamRecord added in v0.2.0 

type StreamRecord struct {
	Seq       uint64
	EventName string
	Keys      json.RawMessage
	Old       json.RawMessage
	New       json.RawMessage
	CreatedNs int64
	SizeBytes int
}

StreamRecord is one change record handed to the Streams API / poller.

type Table 

type Table struct {
	Name    string   `json:"name"`
	Hash    KeyPart  `json:"hash"`
	Range   *KeyPart `json:"range,omitempty"`
	Indexes []Index  `json:"indexes,omitempty"`
	Created int64    `json:"created"`

	TTLAttribute string `json:"ttl_attribute,omitempty"`
	TTLEnabled   bool   `json:"ttl_enabled,omitempty"`

	Tags map[string]string `json:"tags,omitempty"`

	// Cosmetic round-trips.
	BillingMode        string `json:"billing_mode,omitempty"`
	DeletionProtection bool   `json:"deletion_protection,omitempty"`
	StreamSpec         string `json:"stream_spec,omitempty"` // stored, inert (streams post-1.0)
	ItemCount          int64  `json:"item_count"`
}

Table is a table definition.

func (*Table) ARN 

func (t *Table) ARN() string

ARN returns the table ARN.

func (*Table) FindIndex 

func (t *Table) FindIndex(name string) *Index

FindIndex locates an index by name.

func (*Table) StreamARN added in v0.2.0 

func (t *Table) StreamARN() string

func (*Table) StreamLabel added in v0.2.0 

func (t *Table) StreamLabel() string

StreamLabel is the stable per-table stream label (derived from creation time), and StreamARN the corresponding ARN.

func (*Table) StreamViewType added in v0.2.0 

func (t *Table) StreamViewType() (string, bool)

StreamViewType (on Table) returns the stream view type and whether streaming is enabled.

type TxWriteOp 

type TxWriteOp struct {
	Table string
	// Put stores an item.
	Put json.RawMessage
	// UpdateKey + Update applies an update expression.
	UpdateKey json.RawMessage
	Update    *expr.Update
	// DeleteKey removes an item.
	DeleteKey json.RawMessage
	// CheckKey asserts a condition without writing.
	CheckKey json.RawMessage

	Cond      *Cond
	ReturnOld bool // ReturnValuesOnConditionCheckFailure
}

TxWriteOp is one TransactWriteItems operation (exactly one field set).

Source Files

View all Source files

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.