db

package module
v0.20.3 Latest Latest
Warning

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

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

Documentation

Overview

Package db is an embedded ClassAd log: a persistent key->ClassAd store with optimistic multi-writer transactions, mirroring HTCondor's ClassAdLog (src/condor_utils/classad_log.h). It is the Go core that the cgo layer (package capi) exposes as C symbols for a C++ interface to sit on top of, and that the client/server module serves over CEDAR.

It maps directly onto the collections store: the key->ClassAd table is a Collection, and each transaction is a collections.Txn (snapshot isolation, write-write conflicts, per-ad commit). Unlike classad_log.h -- which allows only one active transaction -- this supports any number of independent concurrent transactions, each a distinct *Txn.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AggProjection added in v0.16.8 

func AggProjection(groupCols []GroupCol, aggs []AggSpec) (attrs []string, groupCol, aggCol []int)

AggProjection builds the deduplicated list of attributes the aggregation reads (group columns then non-"*" aggregate arguments) and the index of each group column / aggregate argument within that list. An aggregate whose argument is "*" (COUNT(*)) gets index -1. A caller projects a scan to attrs and feeds the resulting value rows, groupCol, and aggCol to AggregateValues.

func IsSystemKey added in v0.9.0 

func IsSystemKey(key string) bool

IsSystemKey and SystemKey re-export the collections helpers so callers holding only a *db.DB (e.g. dbrpc building marker keys) can classify or construct a reserved system key without importing collections directly. A system key begins with a NUL byte and names a record hidden from client reads but retrievable by explicit LookupClassAd.

func MatchSignature 

func MatchSignature(ad *classad.ClassAd, significantAttrs []string) uint64

MatchSignature is HTCondor's autocluster key: a 64-bit checksum over the given significant attributes' expression text in ad. Two ads with textually identical significant attributes (same Requirements, same RequestCpus literal, ...) hash equal, so a matchmaker can compute one candidate list per distinct signature and reuse it for every identical request.

func SystemKey added in v0.9.0 

func SystemKey(name string) string

SystemKey builds a reserved system key from name (prefixing the NUL sentinel).

func ValidTableName 

func ValidTableName(name string) bool

ValidTableName reports whether name is usable as a table (and a directory): it must start with a letter or underscore and contain only letters, digits, underscores, and hyphens.

func ValueText added in v0.16.8 

func ValueText(v classad.Value) string

ValueText renders a value as a group-key/display string.

Types

type AggFunc added in v0.16.8 

type AggFunc uint8

AggFunc is a SQL aggregate function. 

const (
	AggCount AggFunc = iota // COUNT(*) or COUNT(col)
	AggSum
	AggAvg
	AggMin
	AggMax
)

type AggRow added in v0.16.8 

type AggRow struct {
	Group  []string
	Values []string
}

AggRow is one group's result: the group-by column values followed by the aggregate values, all rendered as strings (aligned with the request's group columns and aggregate specs).

func AggregateValues added in v0.16.8 

func AggregateValues(seq iter.Seq[[]classad.Value], groupCols []GroupCol, aggs []AggSpec, groupCol, aggCol []int, stop func() bool) []AggRow

AggregateValues is the shared GROUP BY core: it buckets a sequence of projected value rows (each aligned to the attribute list AggProjection returned) by the (possibly time-bucketed) group tuple and reduces each group with the COUNT/SUM/ AVG/MIN/MAX accumulators, returning one AggRow per group in first-seen order. groupCol[i]/aggCol[i] index into each value row (aggCol[i] < 0 means COUNT(*)). With no group columns it returns a single row aggregating the whole sequence, still yielding one row over an empty sequence (SQL semantics: COUNT is 0, others undefined). If stop is non-nil it is polled per row and, when it returns true, the scan halts early with the groups accumulated so far -- used to abandon a scan whose client has gone away.

type AggSpec added in v0.16.8 

type AggSpec struct {
	Func AggFunc
	Arg  string
}

AggSpec is one aggregate in a query: a function over an argument attribute. Arg "*" (only meaningful for COUNT) counts every row in the group; otherwise Arg is an attribute name evaluated per ad.

type ArchiveConfig added in v0.7.0 

type ArchiveConfig struct {
	// SegmentSize is the sealed-segment file size in bytes (default 8 MiB).
	SegmentSize int
	// HotAttrs / CategoricalAttrs / ValueAttrs tune the per-segment hot header and
	// indexes; ZoneAttrs names numeric attributes to keep per-segment min/max on for
	// whole-segment query pruning (value-indexed attributes are included automatically).
	HotAttrs                     []string
	CategoricalAttrs, ValueAttrs []string
	ZoneAttrs                    []string
	// Retention bounds what rotation keeps (max segments / bytes / age). Zero keeps all.
	Retention collections.Retention
}

ArchiveConfig configures an archive table. Dir is set by the catalog.

type ArchiveTable added in v0.7.0 

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

An ArchiveTable is an append-only, rotated history store (the "condor history file" use case), exposed as a catalog table type alongside the mutable tables. Ads are appended once and never updated or deleted individually; old data is dropped in bulk by rotation. Queries are newest-first with an optional limit -- condor_history's "last K" -- with whole-segment pruning via zone maps. See collections.Archive.

func (*ArchiveTable) AddIndex added in v0.18.0 

func (t *ArchiveTable) AddIndex(categorical, value []string) bool

AddIndex adds per-segment indexes on the named categorical and/or value attributes and rebuilds them over existing segments. Returns false if the index set was unchanged.

func (*ArchiveTable) Aggregate added in v0.16.8 

func (t *ArchiveTable) Aggregate(constraint string, groupBy []string, aggs []AggSpec) ([]AggRow, error)

Aggregate runs a server-side GROUP BY over the archive's matches: it applies the constraint (using the archive's zone-map pruning, so segments no matching record can fall in are never scanned), groups by the raw group columns, and reduces each group with COUNT/SUM/AVG/MIN/MAX. It shares the exact grouping/reduce engine (AggregateValues) the mutable-table aggregate uses, so an archive aggregate behaves identically to the same aggregate over a live table -- only the (small) grouped result is produced, not every matched ad. With no group columns it returns a single row over the whole match.

func (*ArchiveTable) Append added in v0.7.0 

func (t *ArchiveTable) Append(ad *classad.ClassAd) error

Append adds one ad to the archive (append-only; there is no update or per-key delete).

func (*ArchiveTable) AppendOld added in v0.7.0 

func (t *ArchiveTable) AppendOld(text string) error

AppendOld appends an ad parsed from old-ClassAd text (the qmgmt/history line format).

func (*ArchiveTable) Close added in v0.7.0 

func (t *ArchiveTable) Close() error

Close flushes and closes the archive.

func (*ArchiveTable) CodecStats added in v0.19.0 

func (t *ArchiveTable) CodecStats(sampleMax int) CodecStats

CodecStats reports the archive's compression (codec, dict size, last retrain, sampled ratio).

func (*ArchiveTable) Count added in v0.7.0 

func (t *ArchiveTable) Count() int

Count is the number of records currently retained (reduced by rotation).

func (*ArchiveTable) DropIndex added in v0.18.0 

func (t *ArchiveTable) DropIndex(names ...string) bool

DropIndex removes the named per-segment indexes. Returns false if none matched.

func (*ArchiveTable) IndexSizes added in v0.19.0 

func (t *ArchiveTable) IndexSizes() IndexSizes

IndexSizes reports the per-attribute index byte footprint.

func (*ArchiveTable) IndexedAttrs added in v0.19.0 

func (t *ArchiveTable) IndexedAttrs() (categorical, value []string)

IndexedAttrs returns the archive's categorical and value index attributes.

func (*ArchiveTable) OpStats added in v0.19.0 

func (t *ArchiveTable) OpStats() OpStats

OpStats reports cumulative operational timings. An archive has no DB-level snapshot lock, so SnapshotLock is zero.

func (*ArchiveTable) Query added in v0.7.0 

func (t *ArchiveTable) Query(constraint string) (iter.Seq[*classad.ClassAd], error)

Query returns the archived ads matching constraint, newest first. QueryLimit caps the result at the newest limit matches (<= 0 = all) -- the scan stops after the newest satisfying segments, so "last K" is cheap.

func (*ArchiveTable) QueryLimit added in v0.7.0 

func (t *ArchiveTable) QueryLimit(constraint string, limit int) (iter.Seq[*classad.ClassAd], error)

func (*ArchiveTable) QueryProject added in v0.20.0 

func (t *ArchiveTable) QueryProject(constraint string, attrs []string) (iter.Seq[[]classad.Value], error)

QueryProject scans the matching ads and yields each projected to just attrs' values, read wire-native where possible -- so an aggregate reads only the attributes it needs instead of fully decoding every record. Errors only on a malformed constraint.

func (*ArchiveTable) QueryRawProjected added in v0.20.1 

func (t *ArchiveTable) QueryRawProjected(constraint string, projection []string, redact bool) (iter.Seq[collections.RawAd], error)

QueryRawProjected yields each matching ad as a raw projected subset (only the projection attributes, rendered from the stored representation), newest first — the archive-side of the server-side projection op. redact strips private attributes. It mirrors db.DB.QueryRawProjected so the same wire op serves archives and mutable tables uniformly. chaseRefs is false, matching HTCondor's projection protocol (exactly the requested attributes).

func (*ArchiveTable) Reindex added in v0.18.0 

func (t *ArchiveTable) Reindex()

Reindex rebuilds the per-segment indexes over all segments.

func (*ArchiveTable) Retention added in v0.19.0 

func (t *ArchiveTable) Retention() collections.Retention

Retention returns the archive's current retention bounds.

func (*ArchiveTable) RetrainDict added in v0.18.0 

func (t *ArchiveTable) RetrainDict(sampleMax int) (int, error)

RetrainDict trains a fresh ZSTD dictionary from up to sampleMax records and recompresses every segment in place under it, returning the new dictionary's size in bytes.

func (*ArchiveTable) Rewrite added in v0.18.0 

func (t *ArchiveTable) Rewrite() int

Rewrite recompresses and re-encodes every segment in place under the current codec and hot set, returning the number of records rewritten.

func (*ArchiveTable) Rotate added in v0.7.0 

func (t *ArchiveTable) Rotate(now float64) (int, error)

Rotate drops sealed segments that fall outside the retention policy, given the current time (unix seconds, for age-based retention). Returns how many segments were dropped.

func (*ArchiveTable) SetRetention added in v0.19.0 

func (t *ArchiveTable) SetRetention(r collections.Retention) error

SetRetention updates the retention bounds and persists them (archiveconfig.json), so they take effect on the next Rotate and survive a restart.

func (*ArchiveTable) SidecarSizes added in v0.19.0 

func (t *ArchiveTable) SidecarSizes() SidecarSizes

SidecarSizes reports the sealed-segment sidecar index bytes (mmap-backed, evictable).

func (*ArchiveTable) Stats added in v0.19.0 

func (t *ArchiveTable) Stats() Stats

Stats reports storage accounting (records, segments, arena/used/dead bytes).

func (*ArchiveTable) Truncate added in v0.20.1 

func (t *ArchiveTable) Truncate()

Truncate drops every record, resetting the archive to empty in place (see collections.Archive.Truncate). It is the destructive reset behind a from-scratch history re-sync: empty the table, then re-ingest from the source. The persisted config (indexes, zone maps, retention) is retained, so the archive keeps its shape.

func (*ArchiveTable) Watch added in v0.20.2 

func (t *ArchiveTable) Watch(ctx context.Context, cursor []byte) (iter.Seq[WatchEvent], error)

Watch streams the archive's change events (append-only: upserts and the catch-up/live markers, no deletes), converting collections.WatchEvent to the db WatchEvent used by the mutable-table watch so the two are wire-identical. An empty cursor replays the retained history, then goes live. See collections.Archive.Watch.

type Catalog 

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

A Catalog is a set of named tables, each an independent ClassAd store (its own keyspace, indexes, hot set, and persisted index config). Tables give the database more than one collection to work with -- e.g. separate machine and job ads -- without joins. A persistent catalog keeps each table in its own subdirectory (<dir>/tables/<name>), so tables are isolated on disk and the set is recovered by enumerating those subdirectories on open.

func OpenCatalog 

func OpenCatalog(dir string) (*Catalog, error)

OpenCatalog opens the catalog rooted at dir with no encryption. See OpenCatalogConfig to enable encryption at rest.

func OpenCatalogConfig added in v0.7.0 

func OpenCatalogConfig(cfg CatalogConfig) (*Catalog, error)

OpenCatalogConfig opens the catalog rooted at cfg.Dir, recovering any tables from <dir>/tables/ and applying cfg's encryption at rest to each. Dir == "" makes an in-memory catalog whose tables do not persist.

func (*Catalog) ArchiveTable added in v0.7.0 

func (cat *Catalog) ArchiveTable(name string) (*ArchiveTable, bool)

ArchiveTable returns the archive (history) table named name.

func (*Catalog) ArchiveTables added in v0.7.0 

func (cat *Catalog) ArchiveTables() []string

ArchiveTables returns the archive table names, sorted.

func (*Catalog) Close 

func (cat *Catalog) Close() error

Close closes every table.

func (*Catalog) ConvertTableToMemory added in v0.10.2 

func (cat *Catalog) ConvertTableToMemory(name string) error

ConvertTableToMemory drops the on-disk backing of an existing persistent table, keeping its current contents live in RAM only. It copies the table through a consistent snapshot into a fresh in-memory table (preserving ads, index configuration, hot set, codec, and encryption policy), swaps that in, then closes and deletes the on-disk original.

It is a no-op when the catalog itself is in-memory or the table is already RAM-only. Like Rewrite, it takes a consistent snapshot but does not globally quiesce writers, so a write that races the swap can be lost -- run it during low write activity. DAEMON-level: it changes a table's durability, so callers gate it above ordinary WRITE.

func (*Catalog) CreateArchiveTable added in v0.7.0 

func (cat *Catalog) CreateArchiveTable(name string, cfg ArchiveConfig) (*ArchiveTable, error)

CreateArchiveTable creates (or returns the existing) append-only archive table named name, persisted under <dir>/archives/<name>. cfg configures indexes / zone maps / retention on first creation and is ignored for an existing archive.

func (*Catalog) CreateExporter added in v0.14.0 

func (cat *Catalog) CreateExporter(def ExporterDef) error

CreateExporter registers a new exporter definition. The name must be a valid identifier and must not already be registered. Exporters occupy their own namespace, so an exporter may share a name with a table (an exporter named "jobs" that mirrors the "jobs" table is expected). On a persistent catalog the definition is written to disk before it is returned; on an in-memory catalog it lives only in memory.

func (*Catalog) CreateTable 

func (cat *Catalog) CreateTable(name string) (*DB, error)

CreateTable creates (or returns the existing) table named name. Its data persists under <dir>/tables/<name> for a persistent catalog.

func (*Catalog) CreateTableInMemory added in v0.10.2 

func (cat *Catalog) CreateTableInMemory(name string) (*DB, error)

CreateTableInMemory creates (or returns the existing) table named name as RAM-only, even in a persistent catalog -- shorthand for CreateTableOpts with InMemory set. If the table already exists its backing is unchanged (options are not re-applied); use ConvertTableToMemory to drop an existing persistent table's on-disk backing.

func (*Catalog) CreateTableOpts added in v0.10.1 

func (cat *Catalog) CreateTableOpts(name string, opts TableOptions) (*DB, error)

CreateTableOpts is CreateTable with options; see TableOptions. If the table already exists it is returned as-is (options are not re-applied).

func (*Catalog) CreateView added in v0.12.0 

func (cat *Catalog) CreateView(name string, spec ViewSpec) error

CreateView creates a materialized view named name over spec.BaseTable, materializing it synchronously (so a cardinality-limit overflow or an aggregation error fails the create and leaves nothing behind) and then maintaining it live from the base table's change stream. The definition is persisted for a persistent catalog.

func (*Catalog) DropArchiveTable added in v0.7.0 

func (cat *Catalog) DropArchiveTable(name string) error

DropArchiveTable closes and removes the archive named name, deleting its on-disk data.

func (*Catalog) DropExporter added in v0.14.0 

func (cat *Catalog) DropExporter(name string) error

DropExporter removes an exporter's definition and its resume state. It is not an error to drop an exporter that does not exist (idempotent teardown).

func (*Catalog) DropTable 

func (cat *Catalog) DropTable(name string) error

DropTable closes and removes the table named name, deleting its on-disk data.

func (*Catalog) DropView added in v0.12.0 

func (cat *Catalog) DropView(name string) error

DropView removes a view: it stops the updater, drops the in-memory data, and deletes the persisted definition.

func (*Catalog) EnsureTable 

func (cat *Catalog) EnsureTable(name string) (*DB, error)

EnsureTable creates the table if it does not exist, returning it.

func (*Catalog) Exporter added in v0.14.0 

func (cat *Catalog) Exporter(name string) (ExporterDef, bool)

Exporter returns a single exporter definition.

func (*Catalog) Exporters added in v0.14.0 

func (cat *Catalog) Exporters() []ExporterDef

Exporters returns all registered exporter definitions, sorted by name.

func (*Catalog) LoadExporterState added in v0.14.0 

func (cat *Catalog) LoadExporterState(name string) ([]byte, bool, error)

LoadExporterState returns an exporter's last checkpointed resume-state blob. The bool is false when the exporter has never checkpointed (a fresh exporter with no state yet), which callers treat as "start from the beginning", not as an error.

func (*Catalog) Restore added in v0.7.0 

func (cat *Catalog) Restore(r io.Reader) error

Restore replaces each table named in the catalog snapshot with its snapshotted contents (creating the table if absent), each under that table's DB-wide lock. Tables present in the catalog but ABSENT from the snapshot are left untouched. Encrypted sections are opened with the catalog's pool keys (the tables share them).

func (*Catalog) SaveExporterState added in v0.14.0 

func (cat *Catalog) SaveExporterState(name string, state []byte) error

SaveExporterState durably records an exporter's opaque resume-state blob, replacing any prior state. The exporter calls this after its data has been accepted downstream, so that on restart it resumes just past the last delivered change (at-least-once). The exporter must already exist.

func (*Catalog) Snapshot added in v0.7.0 

func (cat *Catalog) Snapshot(w io.Writer) error

Snapshot writes a backup of every table in the catalog to w. Each table is captured under its own DB-wide lock (consistent per table); the set of tables is the snapshot's membership at the moment each name is read.

func (*Catalog) Table 

func (cat *Catalog) Table(name string) (*DB, bool)

Table returns the table named name.

func (*Catalog) Tables 

func (cat *Catalog) Tables() []string

Tables returns the mutable table names, sorted.

func (*Catalog) View added in v0.12.0 

func (cat *Catalog) View(name string) (*View, bool)

View returns the named view.

func (*Catalog) ViewBacking added in v0.12.0 

func (cat *Catalog) ViewBacking(name string) (*DB, bool)

ViewBacking returns the in-memory backing table of a view, so reads (SELECT/query) resolve a view name to its materialized rows.

func (*Catalog) ViewSealed added in v0.16.0 

func (cat *Catalog) ViewSealed(name, constraint string) (seq iter.Seq[*classad.ClassAd], ok bool, err error)

ViewSealed streams a continuous aggregate's sealed (archived) rows matching the constraint, so a view read can union sealed history with the live backing. ok is false for a plain table or a gauge view (nothing sealed). The returned sequence is empty (not an error) for a continuous aggregate with no archive (in-memory catalog).

func (*Catalog) Views added in v0.12.0 

func (cat *Catalog) Views() []string

Views returns the view names, sorted.

type CatalogConfig added in v0.7.0 

type CatalogConfig struct {
	Dir string
	// PoolKeys enables encryption at rest for every table (each table's master key is
	// wrapped under these keys). EncryptedAttrs is the default explicit encrypted-attr
	// set for each table (private attributes are always encrypted). See db/encrypt.go.
	PoolKeys       []KEK
	EncryptedAttrs []string
}

CatalogConfig configures a catalog, including encryption at rest applied to every table. Dir empty is in-memory.

type CodecStats 

type CodecStats = collections.CodecStats

Diagnostic and management types, re-exported from collections for callers that only import db.

type Config 

type Config struct {
	Dir string
	// Ordered configures maintained, filtered, sorted indexes -- e.g. the negotiator's
	// resource-request lists (partition by Owner, sort by JobPrio then QDate), iterated
	// in order via Ordered. Optional.
	Ordered []OrderSpec
	// HotAttrs / CategoricalAttrs / ValueAttrs / MatchClosureRoots tune storage and
	// query/match push-down (see collections.Options). Optional.
	HotAttrs                     []string
	CategoricalAttrs, ValueAttrs []string
	MatchClosureRoots            []string

	// PoolKeys enables encryption at rest: the DB master key is wrapped under each of
	// these pool/signing keys (any one opens the DB; a rotated-in key is added on the
	// next open). Empty ⇒ encryption disabled. See db/encrypt.go.
	PoolKeys []KEK
	// EncryptedAttrs names the attributes whose values are sealed at rest (case-
	// insensitive). Only meaningful with PoolKeys set. An encrypted attribute may not
	// also be indexed. See collections.Options.EncryptedAttrs.
	EncryptedAttrs []string
}

Config opens a DB with indexing and ordered-index configuration. Dir empty is in-memory; a non-empty Dir is persistent.

type ConflictError 

type ConflictError struct{ Keys []string }

ConflictError reports the keys whose writes lost an optimistic write-write race at commit. The other writes in the transaction committed; the caller re-reads and retries the conflicted keys.

func (*ConflictError) Error 

func (e *ConflictError) Error() string

type Constraint 

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

Constraint is a compiled ClassAd boolean expression, for evaluating the same filter against many ads without re-parsing.

func ParseConstraint 

func ParseConstraint(expr string) (*Constraint, error)

ParseConstraint compiles a ClassAd boolean expression.

func (*Constraint) Matches 

func (c *Constraint) Matches(ad *classad.ClassAd) bool

Matches reports whether ad satisfies the constraint.

type DB 

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

DB is an embedded ClassAd log. Safe for concurrent use.

func Open 

func Open(dir string) (*DB, error)

Open opens a ClassAd log with default configuration. A non-empty dir makes it persistent (memory-mapped arenas under dir, recovered on reopen); an empty dir is in-memory. See OpenConfig for indexing / ordered-index configuration.

func OpenConfig 

func OpenConfig(cfg Config) (*DB, error)

OpenConfig opens a ClassAd log with the given configuration. Every DB is stamped with a stable DB id (persisted alongside a persistent store, so it survives reopen) and a fresh instance id for this open. Until high-availability DB servers exist they are effectively the same identity, but a follower/replica shares the DB id while carrying its own instance id.

func (*DB) AddHotAttrs 

func (db *DB) AddHotAttrs(names ...string) []string

AddHotAttrs pins the named attributes into the hot set and returns the resulting hot attributes. The hot set is persisted.

func (*DB) AddIndex 

func (db *DB) AddIndex(categorical, value []string) bool

AddIndex adds categorical and/or value indexes at runtime, returning whether the configuration changed. The spec is updated immediately; call Reindex to build the new index over existing ads. A change is persisted so it survives a restart.

func (*DB) BackupKey added in v0.7.0 

func (db *DB) BackupKey() []byte

BackupKey returns a copy of the backup key -- the key that unwraps a snapshot's encryption, so an operator can escrow it and decrypt/restore backups without the pool keys. It is DISTINCT from the live-data key (it cannot decrypt the running store) and from the master. Returns nil when encryption is disabled.

func (*DB) Begin 

func (db *DB) Begin() *Txn

Begin starts a new independent transaction.

func (*DB) Close 

func (db *DB) Close() error

Close releases the log's resources.

func (*DB) CodecStats 

func (db *DB) CodecStats(sampleMax int) CodecStats

CodecStats reports the storage codec's state and effectiveness (name, dictionary size, last-retrain time, and a sampled compression ratio).

func (*DB) Compact 

func (db *DB) Compact() int

Compact reclaims dead space in shards whose dead-byte ratio warrants it, returning the number of shards compacted.

func (*DB) Delete added in v0.8.0 

func (db *DB) Delete(key string) (bool, error)

Delete removes the ad at key in its own optimistic transaction, retrying on a write-write conflict. It reports whether an ad was present to remove. Equivalent to Begin + DestroyClassAd + Commit with retry.

func (*DB) DeleteWhere added in v0.8.0 

func (db *DB) DeleteWhere(constraint string) (int, error)

DeleteWhere removes every ad matching constraint, in batched optimistic transactions, and returns the number removed. It is the server-side pushdown for a bulk invalidation or a time-based expiry sweep (express expiry as a constraint, e.g. "<now> > LastHeardFrom + Lifetime"), replacing a client's per-key query-then-delete loop with one call executed where the data lives.

Each round re-scans on a fresh snapshot and, inside the deleting transaction, re-checks the constraint against the ad as of that snapshot before removing it. Two properties fall out of that:

  • Self-healing under concurrency: an ad concurrently modified so that it no longer matches (a re-advertised ad whose expiry constraint no longer holds) is spared, not deleted; and a partial commit's conflicted keys are simply re-evaluated on the next round.
  • Idempotent and safe to run alongside writers or another sweep.

It errors on a malformed constraint, a non-conflict commit failure, or if the sweep fails to converge within maxDeleteRounds (only reachable under relentless churn of the match set); the returned count reflects what was removed so far.

func (*DB) DropIndex 

func (db *DB) DropIndex(names ...string) bool

DropIndex removes the named attributes from the configured indexes, returning whether the configuration changed. A change is persisted.

func (*DB) EncryptedAttrNames added in v0.7.0 

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

EncryptedAttrNames returns the explicit encrypted-attribute set (not the always-on private attributes). EncryptionEnabled reports whether encryption at rest is active.

func (*DB) EncryptionEnabled added in v0.7.0 

func (db *DB) EncryptionEnabled() bool

func (*DB) Explain 

func (db *DB) Explain(constraint string) (QueryExplain, error)

Explain reports how the store would execute a constraint query -- which conjuncts are index-usable and the resulting access path.

func (*DB) ExplainMatch 

func (db *DB) ExplainMatch(job *classad.ClassAd, targetConstraint string) MatchExplain

ExplainMatch reports how matchmaking job against this (resource) collection would execute: the job's Requirements rewritten over the slot with the job's attributes baked to constants, and which of the resulting probes prune via a configured index. targetConstraint, if non-empty, is the MATCH resource-side filter (WHERE TARGET / NOPREEMPT) melded into the explanation.

func (*DB) ForEach 

func (db *DB) ForEach(fn func(ad *classad.ClassAd) bool)

ForEach calls fn for every committed ad, in no particular order, until fn returns false. It reads a consistent snapshot (concurrent writers do not block it).

func (*DB) ForEachSystemAd added in v0.9.0 

func (db *DB) ForEachSystemAd(fn func(key string, ad *classad.ClassAd) bool)

ForEachSystemAd calls fn for every internal system-keyed ad and its key -- the reaper enumeration path. System records are durable bookkeeping (e.g. idempotency markers) stored in the same table as data but hidden from every client scan/query/DeleteWhere; this is the only way to enumerate them. Reads a consistent snapshot. See SystemKey.

func (*DB) HotAttrs 

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

HotAttrs returns the current hot attributes (front-loaded in each ad's hot header for cheap access).

func (*DB) ID 

func (db *DB) ID() string

ID is the stable database identity (same across reopens of a persistent store).

func (*DB) InMemory added in v0.10.2 

func (db *DB) InMemory() bool

InMemory reports whether the table's data lives only in RAM (no on-disk backing), i.e. it was opened with an empty Dir. Such a table is not recovered across restarts.

func (*DB) IndexSizes 

func (db *DB) IndexSizes() IndexSizes

IndexSizes reports each configured index's resident bytes (with human/auto provenance) against the live data bytes -- the memory cost of indexing.

func (*DB) IndexedAttrs 

func (db *DB) IndexedAttrs() (categorical, value []string)

IndexedAttrs returns the currently-indexed attribute names, split into categorical (string equality/membership) and value (numeric + range) indexes.

func (*DB) InstanceID 

func (db *DB) InstanceID() string

InstanceID is this open's identity (fresh each Open). Equal in spirit to ID until HA replicas exist, when replicas of one DB share ID but differ by InstanceID.

func (*DB) Keys 

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

Keys returns every committed key at a consistent snapshot, in no particular order. Useful for administrative enumeration and for a replica that must clear its keyspace before a full re-sync (see the leader-follower replicator).

func (*DB) Len 

func (db *DB) Len() int

Len returns the number of committed ads.

func (*DB) LookupClassAd 

func (db *DB) LookupClassAd(key string) (*classad.ClassAd, bool)

LookupClassAd returns the committed ad for key (the hash table, outside any transaction), or (nil, false).

func (*DB) Maintain 

func (db *DB) Maintain(opts MaintainOptions)

Maintain runs one self-tuning pass: it auto-tunes indexes (adds demand-driven ones, trims auto indexes over the memory budget, never touches human-created ones), refreshes the hot-attribute set, and optionally retrains the compression dictionary. Index/hot changes are persisted. Synchronous; a server drives it on a schedule (see dbrpc.Server.StartMaintenance).

func (*DB) Match 

func (db *DB) Match(job *classad.ClassAd) iter.Seq[*classad.ClassAd]

Match returns the ads that symmetrically match job (bilateral Requirements), pushed down to the store. For the negotiator's pick-best pattern, prefer MatchSorted.

func (*DB) MatchSorted 

func (db *DB) MatchSorted(job *classad.ClassAd, limit int) []*classad.ClassAd

MatchSorted returns job's matches ranked by the job's Rank, best first, at most limit (<=0 = all) -- the negotiator resource-request path, with the store's deferred materialization so only the returned top-N are built.

func (*DB) MatchSortedRanked 

func (db *DB) MatchSortedRanked(job *classad.ClassAd, limit int) []RankedMatch

MatchSortedRanked is MatchSorted that also returns each match's Rank. The job's Requirements prunes candidate slots via any covering index (the matchmaking pushdown) rather than bilaterally evaluating every slot.

func (*DB) MatchSortedRankedFiltered 

func (db *DB) MatchSortedRankedFiltered(job *classad.ClassAd, targetConstraint string, limit int) ([]RankedMatch, error)

MatchSortedRankedFiltered is MatchSortedRanked restricted to slots that also satisfy targetConstraint (a MATCH's WHERE TARGET / NOPREEMPT filter over the resource ad). The constraint's index probes narrow the candidate scan (pushdown) and it is re-checked on each matched slot. An empty constraint is exactly MatchSortedRanked.

func (*DB) OpStats added in v0.11.0 

func (db *DB) OpStats() OpStats

OpStats returns the store's operational timing counters (see the OpStats type).

func (*DB) Ordered 

func (db *DB) Ordered(index int, partition string, resume OrderCursor) iter.Seq[OrderedAd]

Ordered iterates one partition of the index-th configured ordered index in sort order (Config.Ordered), yielding each member ad with a resume cursor and its cluster signature (for run-length folding into resource-request lists). partition selects the run (e.g. an Owner); it is ignored for an index with no Partition. A zero resume starts at the beginning. The snapshot is O(1) and stable under concurrent churn.

func (*DB) Put added in v0.8.0 

func (db *DB) Put(key string, ad *classad.ClassAd) error

Put inserts or replaces the ad at key in its own optimistic transaction, retrying on a write-write conflict (another committer touched key since this attempt's snapshot) until it lands or maxWriteAttempts is exhausted. A blind overwrite that loses the optimistic race would otherwise be silently dropped; Put is the retrying convenience for the common single-ad upsert so callers do not each reimplement the loop. Equivalent to Begin + NewClassAd + Commit.

func (*DB) Query 

func (db *DB) Query(constraint string) (iter.Seq[*classad.ClassAd], error)

Query returns the committed ads matching the constraint expression (an "old ClassAd" boolean expression over each ad, e.g. `JobStatus == 2 && Owner == "alice"`). The store pushes the filter down -- indexed constraints visit only candidates -- so this is far cheaper than ForEach + client-side filtering. Errors only on a malformed constraint.

func (*DB) QueryAsOf added in v0.12.0 

func (db *DB) QueryAsOf(constraint string, t time.Time) (iter.Seq[*classad.ClassAd], error)

QueryAsOf runs a point-in-time ("AS OF") query: it returns the ads matching the constraint as they were at time t. It errors on a malformed constraint, when time travel is not enabled on this table, or when t is older than the retained window.

func (*DB) QueryProject 

func (db *DB) QueryProject(constraint string, attrs []string) (iter.Seq[[]classad.Value], error)

QueryProject returns, for each ad matching the constraint, just the named attributes' values (aligned with attrs), read wire-native where possible so an aggregate or projection does not pay the full-ad decode Query costs. The yielded slice is reused across iterations; copy any value to retain it past the next step. Errors only on a malformed constraint.

func (*DB) QueryRaw added in v0.8.0 

func (db *DB) QueryRaw(constraint string) (iter.Seq[collections.RawAd], error)

QueryRaw yields each matching ad as a collections.RawAd -- the wire-form attribute strings decoded straight from the stored representation with no AST, for a persistent (inline) store as well as an in-memory one -- so a whole-ad result set can be relayed without materializing and re-encoding each ad. Errors only on a malformed constraint.

func (*DB) QueryRawProjected added in v0.16.3 

func (db *DB) QueryRawProjected(constraint string, projection []string, redact bool) (iter.Seq[collections.RawAd], error)

QueryRawProjected is QueryRaw restricted to the projected attribute names, applied inside the collection's decode walk: a non-projected attribute is skipped before any name resolution or value rendering, and a hot-header- covered projection is served from the hot header alone (see collections.ScanRawProjected). redact additionally strips private attributes. An empty projection means no attribute filter.

func (*DB) QueryRawRedacted added in v0.16.3 

func (db *DB) QueryRawRedacted(constraint string) (iter.Seq[collections.RawAd], error)

QueryRawRedacted is QueryRaw with private (secret) attributes stripped inside the collection's decode walk -- an unprivileged consumer's whole-ad query pays no per-attribute re-classification and never renders a private value (see collections.ScanRawRedacted).

func (*DB) QueryRawWire added in v0.16.4 

func (db *DB) QueryRawWire(constraint string, projection []string, redact bool) (iter.Seq[[]byte], error)

QueryRawWire yields each matching ad as a self-contained WIRE-FORM ROW (an inline-names subset ad assembled by slice copies -- see collections.ScanRawWire): the relay form for shipping ads to a remote consumer with the old-ClassAd render deferred to that consumer's client edge. projection restricts the entries (empty = whole ad); redact strips private attributes at the source. At-rest-encrypted values are opened during assembly (the consumer holds no data key). Only meaningful for persistent (inline) stores; an in-memory table yields nothing.

func (*DB) RecordDemand 

func (db *DB) RecordDemand(constraint string)

RecordDemand notes the attributes constraint filters on (for index suggestions) without scanning. Use it for filters applied outside the normal Query path -- e.g. a MATCH's resource-side WHERE TARGET constraint -- so those attributes still surface in SuggestIndexes. A malformed constraint is ignored.

func (*DB) RefreshHotSet 

func (db *DB) RefreshHotSet(sampleMax, topN int) int

RefreshHotSet recomputes the hot set as the topN most frequent attributes from a sample of up to sampleMax live ads, returning how many were chosen. The resulting hot set is persisted.

func (*DB) Reindex 

func (db *DB) Reindex()

Reindex rebuilds all configured indexes from the live ads.

func (*DB) Restore added in v0.7.0 

func (db *DB) Restore(r io.Reader) error

Restore replaces the entire DB with the contents of the snapshot in r. It holds the DB-wide lock exclusively: all writers are blocked and the truncate+reload is atomic. An encrypted snapshot is opened with this DB's pool keys against the snapshot's embedded master envelope, so a snapshot taken by any DB sharing a pool key can be restored.

func (*DB) RestoreWith added in v0.7.0 

func (db *DB) RestoreWith(r io.Reader, keys SnapshotKeys) error

RestoreWith is Restore using any level of the key hierarchy (see SnapshotKeys).

func (*DB) RestoreWithBackupKey added in v0.7.0 

func (db *DB) RestoreWithBackupKey(r io.Reader, backupKey []byte) error

RestoreWithBackupKey is Restore using an explicitly-provided backup key (from BackupKey / the backup.key command) to unwrap the snapshot key, instead of the DB's pool keys. This recovers an encrypted backup independently of the pool keys -- the escrow path for disaster recovery when the original pool keys are unavailable.

func (*DB) RetrainDict 

func (db *DB) RetrainDict(sampleMax int) (int, error)

RetrainDict trains a fresh ZSTD dictionary from a sample of up to sampleMax live ads, switches new writes to it, and recompresses existing records under it. It returns the new dictionary's size in bytes. This is what turns on (or refreshes) compression for a collection that started with the identity codec. See collections.Collection.RetrainDict.

func (*DB) Rewrite 

func (db *DB) Rewrite() int

Rewrite re-encodes every live ad with the current hot set (so a changed hot set applies to existing ads) and force-compacts, returning the number of ads rewritten. A maintenance operation -- see collections.Collection.Rewrite.

func (*DB) SetEncryptedAttrs added in v0.7.0 

func (db *DB) SetEncryptedAttrs(attrs []string) error

SetEncryptedAttrs replaces the explicit set of attributes encrypted at rest (the human-toggled set; private attributes are always encrypted). It errors if encryption is disabled or a named attribute is indexed. The policy is persisted so it survives a restart and, in HA, lets a follower converge on reload. New writes seal the new set; existing records re-seal when next rewritten (compaction/Rewrite).

func (*DB) SetTimeTravel added in v0.12.0 

func (db *DB) SetTimeTravel(maxDistance, checkpoint time.Duration)

SetTimeTravel enables (with a positive maxDistance), retunes, or disables (maxDistance <= 0) point-in-time queries for this table, and persists the setting so it survives a restart. A zero checkpoint interval uses the collections default (1 minute). Enabling is not retroactive -- only changes from now on are travelable.

func (*DB) Snapshot added in v0.7.0 

func (db *DB) Snapshot(w io.Writer) error

Snapshot writes a consistent backup of every ad to w. It holds the DB-wide lock shared, so it is consistent against writers without blocking other readers or snapshots.

func (*DB) SnapshotWithKey added in v0.7.0 

func (db *DB) SnapshotWithKey(w io.Writer) ([]byte, error)

SnapshotWithKey is Snapshot that also returns the per-backup (snapshot) key it minted -- the finest-grained escrow key, which decrypts THIS backup only (unlike the backup key, which decrypts all of them). Returns nil when encryption is disabled.

func (*DB) StartMaintenance 

func (db *DB) StartMaintenance(interval time.Duration, opts MaintainOptions) (stop func())

StartMaintenance starts a background goroutine that runs Maintain with the given options every interval, returning a stop function. Prefer the catalog-wide dbrpc.Server.StartMaintenance, which also covers tables created later.

func (*DB) Stats 

func (db *DB) Stats() Stats

Stats returns a snapshot of the store's storage (ad count, segment/arena/dead bytes) for observability.

func (*DB) SuggestDrops 

func (db *DB) SuggestDrops(sampleMax int) []DropSuggestion

SuggestDrops recommends indexes to drop (unused or low-cardinality) from observed demand and a sample of up to sampleMax live ads.

func (*DB) SuggestIndexes 

func (db *DB) SuggestIndexes(sampleMax int) []collections.IndexSuggestion

SuggestIndexes samples the store and returns attributes that queries filter on but are not yet indexed (advisory; a server may log or auto-apply them).

func (*DB) TimeTravel added in v0.12.0 

func (db *DB) TimeTravel() (maxDistance, checkpoint time.Duration, enabled bool)

TimeTravel reports the table's current point-in-time settings and whether enabled.

func (*DB) Truncate added in v0.7.0 

func (db *DB) Truncate()

Truncate removes every ad from the DB, atomically against all writers (it takes the DB-wide lock exclusively). A DAEMON-level operation -- the first half of a restore, or a deliberate wipe.

func (*DB) UpdateOld added in v0.8.0 

func (db *DB) UpdateOld(key, text string) error

UpdateOld ingests an ad from old-ClassAd wire text under key, skipping the AST build -- the wire-native ingest path (as collections.UpdateOld). It writes through the same shard storage, change log, and watch feed as a committed Put, but bypasses the optimistic-concurrency layer (last-writer-wins), which suits high-rate single-key upserts where per-key write-write races do not occur (a collector re-advertising its own ad).

When the database is encrypted it falls back to parse + Put: the wire-native encoder does not seal, so an encrypted store must take the AST path to keep data encrypted at rest.

func (*DB) UpdateOldBatch added in v0.8.0 

func (db *DB) UpdateOldBatch(items []OldAdText) error

UpdateOldBatch ingests many ads (key + old-ClassAd text) in one shard-commit batch -- the wire-native bulk ingest, so a burst of upserts costs one commit instead of one per ad. Bypasses the optimistic-concurrency layer (last-writer-wins) like UpdateOld. Falls back to per-ad Put on an encrypted store (the wire-native encoder does not seal).

func (*DB) Watch 

func (db *DB) Watch(ctx context.Context, cursor []byte) (iter.Seq[WatchEvent], error)

func (*DB) WatchCursor added in v0.7.0 

func (db *DB) WatchCursor() ([]byte, error)

Watch streams changes committed after the given cursor (nil = from now). Cancel via ctx. Events arrive in commit order per key; see collections.Watch for the delivery and coalescing guarantees. WatchCursor returns an opaque cursor at the current head of the change log, so Watch(ctx, cursor) then streams only subsequent changes (no replay of current contents). Cheap; requires the store was opened with watch history.

type DropSuggestion 

type DropSuggestion = collections.DropSuggestion

Diagnostic and management types, re-exported from collections for callers that only import db.

type ExporterDef added in v0.14.0 

type ExporterDef struct {
	Name string `json:"name"`
	// Kind selects the exporter implementation (e.g. "kafka"). The catalog does not
	// enumerate valid kinds; an unknown kind is simply ignored by any exporter that does
	// not recognize it.
	Kind string `json:"kind"`
	// Config is the exporter's own configuration, opaque to the catalog. It is stored and
	// returned verbatim.
	Config json.RawMessage `json:"config,omitempty"`
}

ExporterDef defines an external sink that mirrors this database's change stream elsewhere (for example, a Kafka change-data exporter that federates several instances through a broker). The catalog is a passive registry: it stores the definition and an opaque resume-state blob and never runs, interprets, or validates Config -- that is the job of the out-of-process exporter (which reads these over dbrpc). Keeping the DB free of any sink-specific dependency is deliberate.

type GroupCol added in v0.16.8 

type GroupCol struct {
	Attr        string
	BucketWidth int64
}

GroupCol is one GROUP BY column for a (possibly bucketed) aggregate: the attribute Attr, optionally floored into fixed-width buckets. BucketWidth == 0 groups by the raw attribute value; BucketWidth > 0 (seconds) groups by the epoch-aligned bucket floor(number(Attr)/BucketWidth)*BucketWidth, and a row whose Attr is not a finite number drops out of the result. This is the shared shape a bucketed aggregate and a bucketed materialized view can both group by.

type IndexSize 

type IndexSize = collections.IndexSize

Diagnostic and management types, re-exported from collections for callers that only import db.

type IndexSizes 

type IndexSizes = collections.IndexSizes

Diagnostic and management types, re-exported from collections for callers that only import db.

type IndexSuggestion 

type IndexSuggestion = collections.IndexSuggestion

Diagnostic and management types, re-exported from collections for callers that only import db.

type KEK added in v0.7.0 

type KEK = crypt.KEK

KEK is a pool / signing key that can wrap the DB master key. Re-exported so db callers need not import crypt.

type MaintainOptions 

type MaintainOptions struct {
	// SampleMax caps the ads sampled for index tuning, hot-set frequency, and dictionary
	// training. Default 4096.
	SampleMax int
	// HotTopN refreshes the hot set to the topN most common attributes; 0 disables it.
	HotTopN int
	// Retrain retrains the ZSTD dictionary (recompacting + reindexing). Expensive on a
	// large store, so a server may run it on a longer cadence than the rest.
	Retrain bool
	// CompactInterval is the cadence of a lightweight, standalone compaction pass
	// (dbrpc.Server.StartMaintenance) that reclaims dead space independently of the
	// expensive retrain-dominated Maintain pass. Compaction is cheap and self-limiting
	// (it unlinks fully-dead segments for free and only recompacts shards past the
	// dead-byte threshold), so it runs far more often than Maintain -- essential for a
	// high-churn table (e.g. a collector re-advertising every few minutes) where dead
	// space would otherwise grow unbounded between the throttled retrain passes. 0 uses
	// a default cadence; a negative value disables the standalone compaction pass.
	CompactInterval time.Duration
	// MinIndexDemand is the minimum observed query count for the auto-tuner to add an
	// index; 0 leaves index auto-tune off (no demand-driven adds).
	MinIndexDemand int64
	// IndexBudgetHighFrac / IndexBudgetLowFrac / IndexBudgetSlackBytes bound auto-created
	// index memory as a fraction of the live data bytes (see collections.AutoTuneOptions);
	// 0 high frac disables the budget (auto indexes grow unbounded).
	IndexBudgetHighFrac   float64
	IndexBudgetLowFrac    float64
	IndexBudgetSlackBytes int64
}

MaintainOptions configures one maintenance pass (DB.Maintain).

type MatchExplain 

type MatchExplain = collections.MatchExplain

Diagnostic and management types, re-exported from collections for callers that only import db.

type OldAdText added in v0.8.0 

type OldAdText struct {
	Key  string
	Text string
}

OldAdText is one keyed ad in old-ClassAd wire text, for UpdateOldBatch.

type OpStat added in v0.11.0 

type OpStat = collections.OpStat

OpStat is one operation's cumulative call count and total wall-nanoseconds.

type OpStats added in v0.11.0 

type OpStats struct {
	collections.OpStats
	SnapshotLock OpStat `json:"snapshotLock"`
}

OpStats is a snapshot of the store's operational timing counters -- where callers spent time blocked in, or holding, each stall point (see collections.OpStats) -- plus the DB-wide snapshot lock's exclusive-hold time (Truncate/Restore). Every value is a monotonic cumulative total; a scraper derives rate and mean latency from deltas.

type OrderCursor 

type OrderCursor = collections.OrderCursor

OrderSpec, SortKey, OrderedAd, OrderCursor configure and drive maintained ordered indexes (the schedd priority-queue / resource-request-list pattern). Re-exported from collections so callers of db need not import it.

type OrderSpec 

type OrderSpec = collections.OrderSpec

OrderSpec, SortKey, OrderedAd, OrderCursor configure and drive maintained ordered indexes (the schedd priority-queue / resource-request-list pattern). Re-exported from collections so callers of db need not import it.

type OrderedAd 

type OrderedAd = collections.OrderedAd

OrderSpec, SortKey, OrderedAd, OrderCursor configure and drive maintained ordered indexes (the schedd priority-queue / resource-request-list pattern). Re-exported from collections so callers of db need not import it.

type QueryExplain 

type QueryExplain = collections.QueryExplain

Diagnostic and management types, re-exported from collections for callers that only import db.

type RankedMatch 

type RankedMatch = collections.RankedMatch

RankedMatch is a matched ad with the job's Rank of it.

type Retention added in v0.19.0 

type Retention = collections.Retention

Diagnostic and management types, re-exported from collections for callers that only import db.

type SidecarSizes added in v0.19.0 

type SidecarSizes = collections.SidecarSizes

Diagnostic and management types, re-exported from collections for callers that only import db.

type SnapshotKeys added in v0.7.0 

type SnapshotKeys struct {
	SnapshotKey []byte
	BackupKey   []byte
	MasterKey   []byte
	PoolKeys    []KEK
}

SnapshotKeys carries any ONE level of the key hierarchy sufficient to decrypt a snapshot. A snapshot is decryptable with, in decreasing specificity: the per-backup SnapshotKey (seals the body directly), the BackupKey (unwraps the snapshot key), the MasterKey (derives the backup key), or PoolKeys (open the embedded master envelope). Restore uses the most specific field provided; all are optional (plain Restore uses the DB's own pool keys). This lets an operator escrow whichever level fits their recovery model and decrypt a backup even without the original pool keys or a running daemon.

type SortKey 

type SortKey = collections.SortKey

OrderSpec, SortKey, OrderedAd, OrderCursor configure and drive maintained ordered indexes (the schedd priority-queue / resource-request-list pattern). Re-exported from collections so callers of db need not import it.

type Stats 

type Stats = collections.Stats

Diagnostic and management types, re-exported from collections for callers that only import db.

type TableOptions added in v0.10.1 

type TableOptions struct {
	// InMemory makes the table's data live only in RAM even in a persistent
	// catalog: no <dir>/tables/<name> directory is created, so the table is not
	// recovered across restarts (it reappears empty). Useful for high-churn,
	// reconstructible data (e.g. frequently-replaced ads) that is not worth the
	// disk I/O of persistence. In an already-in-memory catalog it is a no-op.
	InMemory bool
}

TableOptions tunes how a table is created. The zero value creates a normal table (persistent when the catalog has a directory).

type Txn 

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

Txn is an independent optimistic transaction. Operations are buffered and applied at Commit under snapshot-isolation OCC. A *Txn is not safe for concurrent use by multiple goroutines; independent transactions are.

func (*Txn) Abort 

func (t *Txn) Abort()

Abort discards the transaction's buffered operations. Nothing is written.

func (*Txn) Commit 

func (t *Txn) Commit() error

Commit applies the buffered operations. It returns a *ConflictError if any key was modified by another committer since this transaction's snapshot (the non-conflicted operations still committed), or nil on full success.

func (*Txn) CommitNondurable 

func (t *Txn) CommitNondurable() error

CommitNondurable is Commit that defers the disk durability sync (classad_log.h CommitNondurableTransaction): the writes are visible immediately but their flush is batched to a later durable commit. On an in-memory DB it is identical to Commit.

func (*Txn) DeleteAttribute 

func (t *Txn) DeleteAttribute(key, name string)

DeleteAttribute removes one attribute of key (classad_log.h LogDeleteAttribute). A no-op if key or the attribute is absent.

func (*Txn) DestroyClassAd 

func (t *Txn) DestroyClassAd(key string)

DestroyClassAd removes key (classad_log.h LogDestroyClassAd).

func (*Txn) LookupAttr 

func (t *Txn) LookupAttr(key, name string) (string, bool)

LookupAttr returns the unparsed expression of one attribute as the transaction sees it (classad_log.h LookupInTransaction), or ("", false).

func (*Txn) LookupClassAd 

func (t *Txn) LookupClassAd(key string) (*classad.ClassAd, bool)

LookupClassAd returns key's ad as the transaction sees it: its own buffered writes (read-your-writes) merged over the snapshot (classad_log.h Lookup + the LookupInTransaction overlay in one call).

func (*Txn) NewClassAd 

func (t *Txn) NewClassAd(key string, ad *classad.ClassAd)

NewClassAd stores ad under key (classad_log.h LogNewClassAd). An existing ad at key is replaced.

func (*Txn) SetAttribute 

func (t *Txn) SetAttribute(key, name, expr string) error

SetAttribute sets one attribute of key to the expression parsed from expr (classad_log.h LogSetAttribute) -- a read-modify-write within the transaction, so it composes with the transaction's own earlier writes to key. The ad is created if absent.

type View added in v0.12.0 

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

View is a live materialized view: its spec, the per-key contribution store, per-group accumulators, and an in-memory backing DB holding one rendered ad per group (so the view is queried like a table). All mutable state is guarded by mu.

func (*View) Backing added in v0.12.0 

func (v *View) Backing() *DB

Backing returns the in-memory table holding the materialized rows (read-only in intent).

func (*View) Cursor added in v0.12.0 

func (v *View) Cursor() []byte

Cursor returns the last durable resume cursor (for persistence).

func (*View) LateDrops added in v0.15.0 

func (v *View) LateDrops() int64

LateDrops reports how many base rows were dropped because their time bucket was already sealed (out-of-window late data).

func (*View) Seal added in v0.16.0 

func (v *View) Seal(now int64)

Seal appends and evicts every bucket whose window has closed as of now (unix seconds) into the archive, advancing the watermark. It is what the periodic tick invokes; exported so an operator (or a test) can force sealing at a chosen instant.

func (*View) SealedQuery added in v0.16.0 

func (v *View) SealedQuery(constraint string) (iter.Seq[*classad.ClassAd], error)

SealedQuery streams this continuous aggregate's sealed (archived) rows matching the constraint. It returns an empty sequence for a gauge view or an in-memory continuous aggregate (no archive). Reads union this with the live backing so a view read returns the full series, not just the unsealed buckets.

func (*View) SeriesCount added in v0.12.0 

func (v *View) SeriesCount() int

SeriesCount returns the current number of groups (Prometheus series).

func (*View) Spec added in v0.12.0 

func (v *View) Spec() ViewSpec

Spec returns the view's definition.

func (*View) State added in v0.12.0 

func (v *View) State() (ViewState, error)

State reports the view's lifecycle state and any failure error.

type ViewAggFunc added in v0.12.0 

type ViewAggFunc string

ViewAggFunc is a delta-maintainable aggregate. 

const (
	ViewCount ViewAggFunc = "count"
	ViewSum   ViewAggFunc = "sum"
	ViewAvg   ViewAggFunc = "avg"
)

type ViewGroupCol added in v0.12.0 

type ViewGroupCol struct {
	Attr        string `json:"attr"`
	Alias       string `json:"alias"`
	BucketWidth int64  `json:"bucketWidth,omitempty"` // seconds; 0 = raw value
}

ViewGroupCol is one GROUP BY column: the base-table attribute and the alias it is stored under in the materialized rows (e.g. attr "Owner", alias "label_owner"). When BucketWidth > 0, the column is time-bucketed: the (numeric, unix-seconds) attribute is floored to epoch-aligned buckets of that width, turning the view into a time series -- each interval is a distinct, permanent group rather than an overwritten gauge. This is the same shape as dbrpc.GroupCol.

type ViewMetric added in v0.12.0 

type ViewMetric struct {
	Func  ViewAggFunc `json:"func"`
	Arg   string      `json:"arg"`
	Alias string      `json:"alias"`
}

ViewMetric is one aggregate output: the function, its argument attribute ("*" for COUNT(*)), and the alias it is stored under (e.g. "metric_jobs").

type ViewSpec added in v0.12.0 

type ViewSpec struct {
	BaseTable string         `json:"baseTable"`
	Groups    []ViewGroupCol `json:"groups"`
	Metrics   []ViewMetric   `json:"metrics"`
	// Cardinality is the hard maximum number of groups (output series). Exceeding it fails
	// the build (CreateView error) or, at runtime, moves the view to the failed state.
	Cardinality int `json:"cardinality"`
	// SelectText is the original SELECT for display (.views); not used for execution.
	SelectText string `json:"selectText"`

	// Grace is seconds to wait after a time bucket's window closes before sealing it
	// (so late-arriving base rows still land). Only meaningful for a continuous aggregate
	// (a spec with a time-bucketed group column). 0 seals as soon as the window closes.
	Grace int64 `json:"grace,omitempty"`
	// Retention is seconds of sealed history to keep in the continuous aggregate's archive;
	// 0 keeps all.
	Retention int64 `json:"retention,omitempty"`
}

ViewSpec is a materialized view's persisted definition.

func (ViewSpec) IsContinuous added in v0.15.0 

func (s ViewSpec) IsContinuous() bool

IsContinuous reports whether the view is a continuous aggregate (has a time bucket).

func (ViewSpec) Validate added in v0.12.0 

func (s ViewSpec) Validate() error

Validate checks a spec is well-formed and delta-maintainable.

type ViewState added in v0.12.0 

type ViewState uint8

ViewState is a view's lifecycle state. 

const (
	// ViewBuilding: the initial catch-up is in progress.
	ViewBuilding ViewState = iota
	// ViewActive: built and maintaining live.
	ViewActive
	// ViewStale: definition loaded but the base table is absent; will bind on the next
	// activation once the base table exists.
	ViewStale
	// ViewFailed: a runtime error (e.g. cardinality exceeded); the updater has stopped.
	ViewFailed
)

func (ViewState) String added in v0.12.0 

func (s ViewState) String() string

type WatchEvent 

type WatchEvent struct {
	Kind   WatchKind
	Key    string
	Ad     *classad.ClassAd // nil for a delete
	Cursor []byte
}

WatchEvent is one change to the log. Cursor resumes a watch just after this event (survives reconnect / restart, so no change is missed).

type WatchKind 

type WatchKind uint8

WatchKind classifies a watch event. 

const (
	// WatchUpsert: key was added or updated; Ad holds its new value.
	WatchUpsert WatchKind = iota
	// WatchDelete: key was removed; Ad is nil.
	WatchDelete
	// WatchReset: discard prior state and rebuild from the upserts that follow (the
	// initial full replay, or a re-sync after the cursor fell out of retention). Key
	// and Ad are empty.
	WatchReset
	// WatchSynced: end of the initial catch-up/replay; the watcher is now live. Cursor is
	// a durable resume point (and, after a Reset, the point at which a shadow rebuild goes
	// live). Key and Ad are empty.
	WatchSynced
	// WatchResync: the live stream fell behind; reconnect with the last persisted cursor
	// (which re-enters catch-up). Key and Ad are empty; no state is implied.
	WatchResync
)

type Watcher 

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

Watcher is a watch whose readiness is signalled on a file descriptor -- for the C (DaemonCore) side, which registers the fd in its poll loop and, on wakeup, drains events with Next. Go writes a single wakeup byte when the queue goes non-empty (coalesced: the reader drains fully per wakeup). The fd is a pipe or eventfd the caller created and owns; Watcher never closes it.

func NewWatcher 

func NewWatcher(db *DB, notifyFD int, cursor []byte) (*Watcher, error)

NewWatcher starts watching db, signalling notifyFD when events are queued. cursor nil starts from now.

func (*Watcher) Next 

func (w *Watcher) Next() (WatchEvent, bool)

Next dequeues one event without blocking. ok is false when the queue is empty; the caller (having been woken via the fd) drains until ok is false.

func (*Watcher) Stop 

func (w *Watcher) Stop()

Stop ends the watch. The notify fd is not closed (the caller owns it).

Source Files

View all Source files

Directories

Path Synopsis
Package main builds the embedded ClassAd log as a C archive: it exports C symbols (cadb_*) mirroring HTCondor's classad_log.h, so a C++ interface (built on libcondor_utils) can sit on top.
 Package main builds the embedded ClassAd log as a C archive: it exports C symbols (cadb_*) mirroring HTCondor's classad_log.h, so a C++ interface (built on libcondor_utils) can sit on top.

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.