collections

package module
v0.16.4 Latest Latest
Warning

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

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

Documentation

Overview

Package collections is an in-memory, sharded, memory-dense store of ClassAds with a compiled query engine. Ads are serialized to the compact wire form and packed into append-only arena segments; a per-shard directory maps a stable key hash to a record location. Updates are MVCC-stamped so table scans see every ad exactly once even while background compaction moves bytes.

Index

Constants

View Source 
const DefaultDictSize = 112 * 1024

DefaultDictSize is the target size in bytes for a trained dictionary's content.

View Source 
const DefaultRetrainInterval = 15 * time.Minute

DefaultRetrainInterval is the suggested cadence for background dictionary retraining. A dictionary trained on a representative sample stays effective as long as the ad population's shape is stable, so retraining need not be frequent.

Variables

View Source 
var ErrTimeTravelDisabled = errors.New("collections: time travel is not enabled on this collection")

ErrTimeTravelDisabled is returned by the AS OF query paths when the collection has no time-travel configuration.

Functions

func ConflictChecks added in v0.6.0 

func ConflictChecks() int64

ConflictChecks returns the cumulative number of per-key conflict checks committed transactions have performed -- zero while a single writer runs (the fast path).

func IsSystemKey added in v0.9.0 

func IsSystemKey(key string) bool

IsSystemKey reports whether key is a reserved internal system key (begins with a NUL byte). Exported so callers (and the db/dbrpc layers) can classify a key without duplicating the sentinel.

func LatencyBucketBoundsNanos added in v0.12.1 

func LatencyBucketBoundsNanos() []int64

LatencyBucketBoundsNanos returns a copy of the histogram's inclusive upper bounds (in nanoseconds). An OpStat's Buckets slice has one more entry than this: buckets[i] counts observations <= bound[i], and the final entry counts observations above the last bound.

func ProjectionChecksum added in v0.5.2 

func ProjectionChecksum(ad *classad.ClassAd, attrs []string) uint64

ProjectionChecksum returns a 64-bit FNV-1a checksum over the given attributes' literal expression text in ad, taken in the order given and delimited so two distinct projections cannot alias. A missing attribute contributes a fixed marker, so "absent" is distinguished from every present value.

This is the standalone, caller-supplied-attrs form of an ordered index's cluster Signature (OrderSpec.Cluster). Unlike that index-time signature -- whose attribute set is fixed when the index is opened -- ProjectionChecksum takes the projection at call time, for callers whose significant attributes vary per query (e.g. a schedd whose negotiator sends the significant-attribute set in each negotiation header).

It hashes each attribute's *expression text* rather than its evaluated value, matching HTCondor autocluster semantics: two job ads whose significant attributes are textually identical (same Requirements expression, same RequestMemory literal, ...) hash equal, so a run-length fold over a priority-ordered stream folds them into one resource request with a stored-value compare. A 64-bit collision could merge two adjacent runs; over the projected value bytes that is negligible.

func RenderRawAdInline added in v0.16.4 

func RenderRawAdInline(w []byte, buf []byte, offs []int) (outBuf []byte, outOffs []int, myType, targetType string, ok bool)

RenderRawAdInline renders a wire-form row (a self-contained inline-names ad, e.g. one shipped by ScanRawWire over dbrpc) to old-ClassAd "Name = Value" expression text: each expression is buf[offs[i]:offs[i+1]], with MyType/TargetType lifted out exactly as the collection scans do -- the shape message.PutClassAdRawBytes consumes. This is the LAST-MINUTE conversion at the client edge; everything upstream of it stays in wire form. buf/offs are reset and reused (pass the previous call's returns to avoid allocation).

func SystemKey added in v0.9.0 

func SystemKey(name string) string

SystemKey builds a system key from a name by prefixing the reserved NUL byte. The name is otherwise opaque; callers namespace it however they like.

func TrainDict 

func TrainDict(samples [][]byte) ([]byte, error)

TrainDict builds a ZSTD compression dictionary from sample records (the wire-encoded bytes of a representative set of ads; see CollectSamples). The resulting dictionary can be handed to NewZSTDCodec. It uses DefaultDictSize.

func TrainDictSize 

func TrainDictSize(samples [][]byte, dictSize int) (dict []byte, err error)

TrainDictSize is TrainDict with an explicit dictionary content size.

The pure-Go zstd.BuildDict does not perform ZDICT-style content *selection* (the cover algorithm); the dictionary content is whatever we supply as the builder's History. We therefore assemble the content ourselves by concatenating distinct samples up to dictSize — for a pool of similar ClassAds, this captures the shared attribute names and values that later ads back- reference. BuildDict then trains the entropy tables from the full sample set.

func WatchFilter added in v0.4.0 

func WatchFilter(seq iter.Seq[WatchEvent], match func(*classad.ClassAd) bool) iter.Seq[WatchEvent]

WatchFilter wraps a Watch event stream to deliver only events for keys whose ad satisfies match. It keeps a set of the keys currently matching so a filtered view stays correct as ads change:

  • an Upsert whose ad matches is delivered (the key is marked matching);
  • an Upsert whose ad no longer matches, for a key that was matching, is converted to a Delete so the client drops it from its filtered view;
  • a Delete is delivered for a key known to be matching, and -- during catch-up (before Synced), where the prior match state of a resumed key is unknown -- forwarded conservatively (the client no-ops an unknown key);
  • Reset clears the matched set; Synced and Resync pass through.

A nil match returns seq unchanged (no filtering). match is called on each Upsert's ad and must be safe for concurrent-free sequential use.

Types

type AdUpdate 

type AdUpdate struct {
	Key []byte
	Ad  *classad.ClassAd
}

AdUpdate is one insert-or-update in a batch.

type Archive 

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

Archive is an append-only, larger-than-RAM, rotated store of ClassAds — the "condor history file" use case. Ads are appended once and never updated; old data is dropped in bulk by rotation. Unlike Collection it has no MVCC, no compaction, no per-key delete, and no in-RAM key directory: a segment is sealed once, its index persisted as an immutable sidecar, and dropped as a unit on rotation.

See docs/history-archive.md for the design (H1–H5): sealed segments with persisted sidecar indexes, a segment catalog with zone maps for whole-segment pruning, O(segments) recovery, newest-first + LIMIT queries, and age/size/count rotation. Indexes are never held in RAM: each sealed segment's index is an immutable v2 sidecar, mmap'd and queried in place (binary search / range scan over sorted key runs), loaded lazily on first query and demand-paged (see mmapSegIndex).

func CreateArchive 

func CreateArchive(opts ArchiveOptions) (*Archive, error)

CreateArchive creates a new, empty Archive. The directory must not already hold a catalog (use OpenArchive to reopen one).

func OpenArchive 

func OpenArchive(opts ArchiveOptions) (*Archive, error)

OpenArchive reopens an existing Archive from opts.Dir, recovering its segments and indexes. The index/hot/zone options should match those the archive was created with. A segment left un-sealed by a crash (present on disk but absent from the catalog) is recovered by scanning its CRC-framed records and sealing it.

func (*Archive) Append 

func (a *Archive) Append(ad *classad.ClassAd) error

Append adds one ad to the archive. It is appended to the active segment; when the ad will not fit, the active segment is sealed (its index + zone maps built and persisted) and a new one started. Safe for use by a single writer concurrently with queries.

func (*Archive) Close 

func (a *Archive) Close() error

func (*Archive) Count added in v0.7.0 

func (a *Archive) Count() int

Close seals the active segment and unmaps every segment file. The Archive must not be used afterward. Count returns the number of records currently retained (sealed segments plus the active one). Rotation reduces it in whole-segment steps.

func (*Archive) Flush 

func (a *Archive) Flush() error

Flush seals the active segment (if any), making all appended ads durable and indexed. Useful to make recent ads queryable-by-index and recoverable without closing.

func (*Archive) Query 

func (a *Archive) Query(q *vm.Query) iter.Seq[*classad.ClassAd]

Query returns an iterator over the archived ads matching q, newest first. Each segment is pruned by its zone maps, then its sidecar index narrows the candidate records, and every candidate is re-verified against the full predicate — so the result is identical to a full scan. The active (unsealed) segment is full-scanned.

func (*Archive) QueryLimit 

func (a *Archive) QueryLimit(q *vm.Query, limit int) iter.Seq[*classad.ClassAd]

QueryLimit is Query capped at limit matches (limit <= 0 means unlimited). Because iteration is newest-first, QueryLimit(q, K) yields the K most recent matches and stops — the common condor_history "last K" pattern, which lets the scan terminate after touching only the newest satisfying segments.

func (*Archive) Rotate 

func (a *Archive) Rotate(now float64) (int, error)

Rotate drops the oldest sealed segments until every configured Retention bound is met, and returns the number dropped. now (e.g. the current time as unix seconds) is used only to evaluate an age bound, so the Archive needs no clock of its own. A dropped segment's data file and sidecar index are unmapped and unlinked; a query currently scanning a dropped segment holds a pin, so the data file's unlink is deferred until that query finishes (no read-through-a-torn-mapping).

func (*Archive) SidecarSizes added in v0.8.0 

func (a *Archive) SidecarSizes() SidecarSizes

SidecarSizes sums each sealed segment's sidecar size and its MPH/bloom portions. It maps each sidecar briefly and closes it immediately, so it is an operator diagnostic, not a hot path. The active (unsealed) segment has no sidecar and is skipped.

func (*Archive) Watch added in v0.4.0 

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

Watch replays every archived ad after cursor (nil ⇒ from the oldest retained), then streams new appends until ctx is cancelled or the consumer stops. A cursor from a different archive, or older than what rotation still retains, yields a WatchReset and resumes from the current floor. See WatchEvent and docs/WATCH.md.

type ArchiveOptions 

type ArchiveOptions struct {
	// Dir is the directory holding segment/index/catalog files. Required.
	Dir string
	// SegmentSize is the sealed-segment (mmap file) size in bytes; a segment rolls
	// over when the next ad will not fit. Default 8 MiB.
	SegmentSize int
	// Codec compresses stored ad bytes. Default identity. For recovery the same codec
	// must be supplied (the archive does not yet persist codec identity per segment).
	Codec Codec
	// HotAttrs front-loads these attributes in each ad's hot header (see Collection).
	HotAttrs []string
	// CategoricalAttrs / ValueAttrs configure the per-segment indexes (see Collection).
	CategoricalAttrs []string
	ValueAttrs       []string
	// ZoneAttrs names numeric attributes to keep per-segment min/max on, so a query
	// with a range/equality constraint on one can skip whole segments without opening
	// them. ValueAttrs are automatically included.
	ZoneAttrs []string
	// Retention bounds what rotation keeps. Zero ⇒ keep everything.
	Retention Retention
}

ArchiveOptions configures an Archive.

type AutoTuneChange 

type AutoTuneChange struct {
	Attr, Kind, Action, Reason string // Action: "add" | "drop"
}

AutoTuneChange records one add/drop AutoTune applied.

type AutoTuneOptions 

type AutoTuneOptions struct {
	// SampleMax caps the ads profiled. Default maxDistinctSample.
	SampleMax int
	// MinDemand is the minimum equality+range probe count for an attribute to be
	// added as an index. Default 1 (any observed demand).
	MinDemand int64
	// DropUnused, if set, drops AUTO-created indexes that no query has filtered on
	// ("unused"). Off by default so AutoTune never removes an index a workload has
	// simply not exercised yet. Human-created indexes and low-cardinality indexes are
	// never auto-dropped (SuggestDrops reports them for manual review).
	DropUnused bool
	// BudgetHighFrac / BudgetLowFrac, when BudgetHighFrac > 0, bound index memory as a
	// fraction of the live data bytes: AutoTune stops adding demand-driven indexes once
	// index bytes reach the high mark, and trims the least-used AUTO indexes (never
	// human-created ones) until index bytes fall below BudgetLowFrac. This is a high/low-
	// watermark with hysteresis: grow until over the high mark, trim back to the low
	// mark. 0 disables the budget (unbounded). BudgetLowFrac defaults to 0.7*BudgetHighFrac.
	BudgetHighFrac float64
	BudgetLowFrac  float64
	// BudgetSlackBytes is absolute leeway on top of the high watermark: index bytes may
	// exceed BudgetHighFrac by up to this many bytes before any trim triggers, so a small
	// overage (or a small database, where the percentage is tiny in absolute terms) does
	// not cause churn. 0 uses defaultBudgetSlackBytes (10 MiB); set a negative value for
	// no slack (trim exactly at the fraction).
	BudgetSlackBytes int64
	// Reindex, if set, calls Reindex after applying changes so the new/removed
	// indexes take effect immediately instead of at the caller's next Reindex.
	Reindex bool
}

AutoTuneOptions configures AutoTune.

type AutoTuneResult 

type AutoTuneResult struct {
	Changes []AutoTuneChange
	Changed bool
}

AutoTuneResult reports what AutoTune changed.

type Codec 

type Codec interface {
	// Compress appends the compressed form of src to dst and returns it.
	Compress(dst, src []byte) []byte
	// Decompress appends the decompressed form of src to dst and returns it.
	Decompress(dst, src []byte) ([]byte, error)
	// Name identifies the codec (for diagnostics).
	Name() string
}

Codec compresses and decompresses encoded ad bytes. It is a seam so the store can run with no compression (identityCodec, the default) in tests and benchmarks, and with ZSTD — optionally with a pre-trained shared dictionary — in production.

A codec's dictionary is fixed for the life of a collection: stored records are opaque compressed bytes that compaction copies verbatim, so every record must be decodable by the same codec. Re-training the dictionary between compactions (which requires versioning dictionaries per record and recompressing) is a future extension; a dictionary trained once from a representative sample (see TrainDict) already captures the cross-ad redundancy that dominates a pool of similar ClassAds.

func NewZSTDCodec 

func NewZSTDCodec(dict []byte) (Codec, error)

NewZSTDCodec returns a ZSTD codec. If dict is non-empty it is used as a shared compression dictionary (see TrainDict). Pass nil for dictionary-less ZSTD.

type CodecStats added in v0.6.0 

type CodecStats struct {
	// Codec is the current codec name: "identity" (no compression), "zstd", or "zstd+dict".
	Codec string `json:"codec"`
	// DictBytes is the trained ZSTD dictionary size (0 if none). Set by RetrainDict.
	DictBytes int64 `json:"dictBytes"`
	// LastRetrain is when RetrainDict last succeeded in this process (zero if never; note
	// a persistent collection recovers its codec but not this in-process timestamp).
	LastRetrain time.Time `json:"lastRetrain,omitempty"`
	// SampleRecords is how many live records were sampled for the ratio below.
	SampleRecords int `json:"sampleRecords"`
	// CompressedBytes / UncompressedBytes are the sampled records' stored (compressed) and
	// decompressed sizes; Ratio is UncompressedBytes/CompressedBytes (1.0 = no gain).
	CompressedBytes   int64   `json:"compressedBytes"`
	UncompressedBytes int64   `json:"uncompressedBytes"`
	Ratio             float64 `json:"ratio"`
}

CodecStats reports the storage codec's state and effectiveness, for the diagnostic question "is compression on, when was the dictionary last (re)trained, and how well is it compressing?". A collection defaults to the identity codec (no compression) until Options.Codec or RetrainDict switches it, so a Ratio near 1.0 with Codec "identity" means compression was never enabled.

type Collection 

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

Collection is an in-memory, sharded, memory-dense store of ClassAds. It is safe for concurrent use.

func New 

func New(opts Options) *Collection

New creates an empty Collection.

func Open 

func Open(opts Options) (*Collection, error)

Open opens a persistent collection under opts.Dir, whose arenas are memory-mapped files. Committed data is flushed to disk on Close (per-commit msync durability is added in a later milestone). If opts.Dir is empty, Open is equivalent to New (an in-memory collection). Persistence is unix-only.

NOTE (P2): this creates a fresh persistent collection; recovering an existing directory (rebuilding the directory + index from the segment files) is the next milestone.

func (*Collection) AddAutoIndex added in v0.6.0 

func (c *Collection) AddAutoIndex(categorical, value []string) bool

AddAutoIndex adds indexes marked as auto-created (provenance auto), like AddIndex but eligible for automatic trimming by AutoTune's memory budget. Used by the persistence layer to restore auto provenance on restart.

func (*Collection) AddHotAttrs added in v0.6.0 

func (c *Collection) AddHotAttrs(names ...string) []string

AddHotAttrs pins the named attributes into the hot set (front-loaded in future writes' hot headers), merging them with the current set, and returns the resulting hot attribute names. Unlike RefreshHotSet, which recomputes the set from sampled frequency, this forces specific attributes in regardless of how often they appear. Works in both RAM (interned) and persistent (inline) modes.

func (*Collection) AddIndex 

func (c *Collection) AddIndex(categorical, value []string) bool

AddIndex adds categorical (string equality / membership) and value (numeric equality + range) indexes at runtime and returns whether the configuration changed. Names already indexed are ignored; a name given as both categorical and value, or already indexed as the other kind, is placed as categorical (equality on strings is the common case and categorical also serves it). The new indexes take effect for existing segments on the next Reindex; new segments pick them up when they are first indexed.

func (*Collection) AutoIndexNames added in v0.6.0 

func (c *Collection) AutoIndexNames() []string

AutoIndexNames returns the names of auto-created indexes (provenance auto), so the human/auto distinction can be persisted and survive a restart.

func (*Collection) AutoTune 

func (c *Collection) AutoTune(opts AutoTuneOptions) AutoTuneResult

AutoTune applies the advisory signals: it adds the demand-driven indexes from SuggestIndexes (those at or above MinDemand) and, if DropUnused is set, drops the unused ones from SuggestDrops. It is the "make it so" companion to the SuggestX methods — call it on whatever schedule you like (e.g. alongside Reindex). All suggestions are computed from one consistent snapshot before any change is applied.

func (*Collection) Begin added in v0.6.0 

func (c *Collection) Begin() *Txn

Begin starts an optimistic transaction. Its snapshot for a shard is captured the first time the transaction reads or writes a key in that shard.

func (*Collection) Close 

func (c *Collection) Close() error

Close flushes all committed data to disk and unmaps the collection's segment files. It is a no-op for an in-memory collection. The collection must not be used after Close.

func (*Collection) CodecStats added in v0.6.0 

func (c *Collection) CodecStats(sampleMax int) CodecStats

CodecStats measures the storage codec's effectiveness over a sample of up to sampleMax live records: it decompresses each and compares the stored (compressed) size to the decompressed size. It takes each shard's read lock briefly, so it is safe alongside readers and writers.

func (*Collection) CollectSamples 

func (c *Collection) CollectSamples(max int) [][]byte

CollectSamples returns the decompressed wire bytes of up to max ads, for training a ZSTD dictionary (see TrainDict). It samples across shards, decoding each record with the codec it was stored under.

func (*Collection) Compact 

func (c *Collection) Compact() int

Compact reclaims space from superseded/deleted records. For each shard it first unlinks any fully-dead segment cheaply (no rewrite), then, if the shard's dead-byte ratio still warrants it, recompresses the live records into fresh segments. It is safe to call concurrently with reads and writes. Returns the number of shards where space was reclaimed (by either mechanism).

func (*Collection) Delete 

func (c *Collection) Delete(key []byte) bool

Delete removes key, returning whether it was present. The removal is an MVCC tombstone: scans already in progress still see the pre-delete version.

func (*Collection) DropIndex 

func (c *Collection) DropIndex(names ...string) bool

DropIndex removes the given attributes from the configured indexes (categorical or value, whichever they were) and returns whether the configuration changed. The attributes stop being used by queries immediately; their postings are reclaimed from existing segments at the next Reindex. Names that were never indexed are ignored.

func (*Collection) EncryptedAttrNames added in v0.7.0 

func (c *Collection) EncryptedAttrNames() []string

EncryptedAttrNames returns the explicit encrypted-attribute set (case-folded, sorted) -- the human-toggled attributes, not the always-on private ones. Used for persistence and diagnostics.

func (*Collection) EncryptionEnabled added in v0.7.0 

func (c *Collection) EncryptionEnabled() bool

EncryptionEnabled reports whether this collection has a data key (encryption at rest is active). Without it, EncryptedAttrs is inert.

func (*Collection) ExplainMatch added in v0.6.0 

func (c *Collection) ExplainMatch(job *classad.ClassAd, targetConstraint string) MatchExplain

ExplainMatch reports how matchmaking job against this collection would execute: it rewrites the job's Requirements over the slot (baking the job's attribute values to constants), extracts the index-satisfiable probes, and reports which are covered by a configured index and the resulting access path. targetConstraint, if non-empty, is the MATCH's resource-side filter (WHERE TARGET, including NOPREEMPT's `State =!= "Claimed"`) -- already slot-scoped -- which is melded into the shown predicate, probes, and evaluation order so the explanation reflects the full effective slot filter. No I/O beyond reading the spec.

func (*Collection) ExplainQuery added in v0.6.0 

func (c *Collection) ExplainQuery(q *vm.Query) QueryExplain

ExplainQuery reports how the store would execute q: which of its conjuncts are index-usable, and the resulting access path (indexed / parallel scan / serial scan). It performs no I/O beyond reading the current index spec.

func (*Collection) Flatten added in v0.4.0 

func (c *Collection) Flatten(key []byte) (*classad.ClassAd, bool)

Flatten returns a standalone, self-contained copy of the ad at key with every inherited parent attribute materialized into it (the child's own values winning, parent-private attributes excluded) and no residual parent link. It is the form to archive as a history record: like condor_history, which stores each completed job flattened rather than as a live cluster/proc pair, so the record stays readable after its structural parent is gone. For a collection without parent chaining this is identical to Get.

func (*Collection) ForEachAd added in v0.7.0 

func (c *Collection) ForEachAd(fn func(key string, ad *classad.ClassAd) bool)

ForEachAd calls fn with every stored (non-system) ad and its key, including structural (parent-only) ads -- a client-facing image, unlike Scan/Keys which hide structural ads. It is the driver for db.DeleteWhere's key matching, so it must NOT expose internal system records (they are neither client-visible nor client-deletable); system records are enumerated separately via ForEachSystemAd. Each ad is fully decoded (decompressed and, for encrypted attributes, decrypted). Iteration stops early if fn returns false. Per-shard consistent snapshot, like Scan.

func (*Collection) ForEachSystemAd added in v0.9.0 

func (c *Collection) ForEachSystemAd(fn func(key string, ad *classad.ClassAd) bool)

ForEachSystemAd calls fn with every internal system-keyed ad and its key -- the mirror image of ForEachAd, which hides them. It is the enumeration path a TTL reaper uses to find and expire durable bookkeeping records. Each ad is fully decoded; iteration stops early if fn returns false. Per-shard consistent snapshot, like Scan.

func (*Collection) Get 

func (c *Collection) Get(key []byte) (*classad.ClassAd, bool)

Get returns the current ad for key, decoded, or (nil, false).

func (*Collection) HotAttrNames 

func (c *Collection) HotAttrNames() []string

HotAttrNames returns the current hot attributes by name (for diagnostics).

func (*Collection) IndexSizes added in v0.6.0 

func (c *Collection) IndexSizes() IndexSizes

IndexSizes measures each configured index's resident bytes across all live segments, tagged with provenance (human vs auto), against the live data bytes. It takes each shard's read lock briefly, so it is safe alongside readers and writers.

func (*Collection) IndexedAttrs 

func (c *Collection) IndexedAttrs() (categorical, value []string)

IndexedAttrs returns the currently-indexed attribute names, split by kind, in the collection's canonical (interned) casing.

func (*Collection) Keys added in v0.6.0 

func (c *Collection) Keys() []string

Keys returns every visible key in the collection at a consistent per-shard snapshot, in no particular order. Structural (parent-only) keys of a chained collection are excluded, matching Scan's output. Each key is a fresh copy the caller owns. Useful for administrative enumeration and for a replica that must clear its keyspace before a full re-sync.

func (*Collection) Len 

func (c *Collection) Len() int

Len returns the number of live keys across all shards.

func (*Collection) Match added in v0.5.2 

func (c *Collection) Match(job *classad.ClassAd) iter.Seq[*classad.ClassAd]

Match returns every ad in the collection that symmetrically matches job, in no particular order. Use MatchSorted for a Rank-ordered result. job is not modified.

func (*Collection) MatchSorted added in v0.5.2 

func (c *Collection) MatchSorted(job *classad.ClassAd, limit int) []*classad.ClassAd

MatchSorted returns the matching ads ranked by job's Rank expression, best (highest Rank) first. limit <= 0 returns all matches; limit > 0 returns at most the top limit. Ads whose Rank does not evaluate to a number sort after ranked ones. Ties in Rank are broken in an unspecified order (it depends on scan/fan-out order); a caller needing a deterministic tiebreak should apply its own. job is not modified.

func (*Collection) MatchSortedRanked added in v0.6.0 

func (c *Collection) MatchSortedRanked(job *classad.ClassAd, limit int) []RankedMatch

MatchSortedRanked is MatchSorted that also returns each match's Rank, best (highest Rank) first, up to limit (<= 0 = all). Ads whose Rank is not numeric sort after ranked ones. Like MatchSorted, the job's Requirements is used to visit only index-candidate slots when an index covers it (the matchmaking pushdown), rather than bilaterally evaluating every slot.

func (*Collection) MatchSortedRankedFiltered added in v0.6.0 

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

MatchSortedRankedFiltered is MatchSortedRanked restricted to slots that also satisfy targetConstraint (a constraint over the resource ad, e.g. `State =!= "Claimed"`). The constraint's index-satisfiable probes narrow the candidate scan (pushdown); the full constraint is then re-checked on each matched slot. An empty constraint is exactly MatchSortedRanked. Returns an error only if targetConstraint fails to parse.

func (*Collection) OpStats added in v0.11.0 

func (c *Collection) OpStats() OpStats

OpStats returns a snapshot of the collection's operational timing counters -- where its callers spent time blocked in, or holding, each of the store's stall points (shard write lock, segment allocation, durability sync, and the maintenance passes). Per-shard counters are summed across all shards; every value is a monotonic cumulative total, so a scraper derives rate and mean latency from deltas. It reads atomics locklessly, so it never contends with readers or writers.

func (*Collection) Ordered added in v0.5.2 

func (c *Collection) Ordered(index int, partition classad.Value, resume OrderCursor) iter.Seq[OrderedAd]

Ordered iterates one partition of the index-th configured ordered index in sort order, yielding each member ad with its resume cursor and cluster signature. Iteration is over an O(1) snapshot taken at the call, so it is stable even as the index churns; each ad is fetched live by key, so a member deleted since the snapshot is skipped. For an index configured without a Partition, the partition argument is ignored (there is a single global run). resume's zero value starts at the beginning.

func (*Collection) Put 

func (c *Collection) Put(key []byte, ad *classad.ClassAd) error

Put inserts or updates a single ad. It is the hot path for the common one-ad-at-a-time daemon pattern, so unlike Update it takes no per-call slice or map allocations: it encodes, compresses, and commits directly to the one shard.

func (*Collection) Query 

func (c *Collection) Query(q *vm.Query) iter.Seq[*classad.ClassAd]

Query returns an iterator over the ads matching q, with the same scan-exactly-once guarantee as Scan.

Each ad is match-tested cheaply and only matching ads are fully decoded to be yielded. The match test uses, in order of preference: wire-native evaluation (the query reads scalar-literal attributes directly from the encoded ad, building no ClassAd); partial decode (decode only the attributes the query reads, transitively) when an attribute is a non-literal expression; or full decode for queries that read attributes by a runtime-computed name (eval()).

func (*Collection) QueryAsOf added in v0.12.0 

func (c *Collection) QueryAsOf(q *vm.Query, t time.Time) (iter.Seq[*classad.ClassAd], error)

QueryAsOf runs q against the point-in-time snapshot at time t: it resolves t to a per-shard commit sequence and scans each shard at that sequence, so the result is exactly the ads that were current at t. It errors if time travel is disabled or t predates the retained window. v1 uses a serial full scan per shard (the index, parallel, and chained fast paths are current-time only); the query constraint is still applied to every candidate.

func (*Collection) QueryProject added in v0.6.0 

func (c *Collection) QueryProject(q *vm.Query, attrs []string) iter.Seq[[]classad.Value]

QueryProject is Query specialized for aggregation and projection: for each ad matching q it yields just the named attributes' values, read wire-native (straight from the encoded ad, no *classad.ClassAd built) whenever they are scalar literals -- the common case for the numeric/string attributes an aggregate groups or sums by. Only an ad that stores one of the requested attributes as a non-literal expression falls back to a full decode, and only that ad.

This avoids the dominant cost of a GROUP BY / aggregate over Query, which fully decodes every matching ad (tens of allocations each) just to read one or two attributes.

The yielded slice is reused across iterations and aligned with attrs; the consumer must copy any value it needs to retain past the next step (the aggregate reads each value into its group state immediately).

func (*Collection) QueryRaw 

func (c *Collection) QueryRaw(q *vm.Query) iter.Seq[RawAd]

QueryRaw is Query, but yields each matching ad as RawAd (AST-free) instead of a *classad.ClassAd. It uses the same index/scan machinery -- an indexed lookup still visits only candidate ads -- so it does not regress selective queries. Inline-name (persistent) collections have no intern table, so QueryRaw yields nothing for them; callers that must support those should use Query.

func (*Collection) QueryRawProjected added in v0.16.3 

func (c *Collection) QueryRawProjected(q *vm.Query, projection []string, chaseRefs, redact bool) iter.Seq[RawAd]

QueryRawProjected is QueryRaw with the same in-walk projection as ScanRawProjected (see there for chaseRefs/redact).

func (*Collection) QueryRawRedacted added in v0.16.3 

func (c *Collection) QueryRawRedacted(q *vm.Query) iter.Seq[RawAd]

QueryRawRedacted is QueryRaw with private (secret) attributes stripped from every yielded ad, using the intern table's precomputed per-id private flags (see wire.NewInternTableWithPrivacy): the redaction costs one bool check per attribute and private nodes are never rendered at all. Use it to serve a public query; QueryRaw preserves private attributes for trusted consumers (e.g. the negotiator's private-ad query).

func (*Collection) QueryRawWire added in v0.16.4 

func (c *Collection) QueryRawWire(q *vm.Query, projection []string, redact bool) iter.Seq[[]byte]

QueryRawWire is ScanRawWire restricted to ads matching q.

func (*Collection) RecordDemand added in v0.6.0 

func (c *Collection) RecordDemand(probes []vm.Probe)

RecordDemand notes the attributes a constraint filters on (for SuggestIndexes) without running a scan. Query records demand automatically; this is for callers that filter outside the normal scan path -- e.g. cross-table MATCH applying a resource-side (WHERE TARGET) constraint to already-matched candidates -- so those attributes still surface as index suggestions.

func (*Collection) RefreshHotSet 

func (c *Collection) RefreshHotSet(sampleMax, topN int) int

RefreshHotSet samples up to sampleMax live ads, tallies how often each attribute appears, and installs the topN most common attributes as the hot set used to front-load future writes' hot headers. It counts attribute ids directly from the wire form (no full decode). Existing ads keep the hot header they were written with; because daemons rewrite ads periodically, the population's hot headers converge on the refreshed set over time.

Returns the number of attributes chosen. A no-op (returns 0) when there are no ads yet.

func (*Collection) Reindex 

func (c *Collection) Reindex()

Reindex (re)builds the per-segment value/categorical indexes for every live segment, covering all records written so far. It reads only immutable segment bytes, so it runs off the write path and does not block writers or compaction. Call it on whatever schedule you like: queries use whatever coverage exists and full-scan the rest, so Reindex only affects query speed, never results.

Reindex also reconciles segments with the current index configuration: a segment indexed before an AddIndex is rebuilt so the new attribute is backfilled, and one indexed before a DropIndex is rebuilt (or, if nothing is indexed anymore, its index is dropped) so the removed attribute's postings are reclaimed. A whole span of segments therefore evolves toward the current spec at whatever cadence the caller reindexes — no write-path or compaction coupling.

func (*Collection) RetrainDict 

func (c *Collection) RetrainDict(sampleMax int) (int, error)

func (*Collection) Rewrite added in v0.6.0 

func (c *Collection) Rewrite() int

Rewrite re-encodes every live ad with the current hot set (and match closure) so a changed hot set takes effect on existing ads, not just future writes, then force-compacts every shard to reclaim the superseded pre-rewrite records. Returns the number of ads rewritten.

It re-Puts ads on the normal write path, so it is a maintenance operation: an update to a key that races the rewrite may be overwritten by the pre-rewrite value. Run it during low write activity (or, in an HA deployment, on the sole writer).

func (*Collection) Scan 

func (c *Collection) Scan() iter.Seq[*classad.ClassAd]

Scan returns an iterator over every ad in the collection. It is scan-exactly- once: each key present at the moment a shard's scan begins is yielded exactly once (never duplicated, never skipped), even while concurrent updates and compaction run. An ad updated mid-scan is seen at whichever version was current when that shard's scan started.

func (*Collection) ScanRaw 

func (c *Collection) ScanRaw() iter.Seq[RawAd]

ScanRaw is Scan, yielding every ad as RawAd (AST-free).

func (*Collection) ScanRawProjected added in v0.16.3 

func (c *Collection) ScanRawProjected(projection []string, chaseRefs, redact bool) iter.Seq[RawAd]

ScanRawProjected is ScanRaw restricted to the projected attribute names, applied INSIDE the wire walk: the projection is resolved to interned ids once per scan, and each ad is walked as raw (id, node) pairs -- a non-projected attribute costs one id comparison and a TLV length hop, and is never name- resolved or rendered. For a typical monitoring projection (10-20 attributes of a several-hundred-attribute ad) this skips the overwhelming majority of the per-ad decode work that a decode-everything-then-filter projection pays.

chaseRefs additionally resolves each emitted expression's attribute references against the same ad (transitively, per ad -- an "elevator" of linear passes that repeats only while new references surface), so a projected ad evaluates self-contained. HTCondor's query protocol sends exactly the requested attributes, so a protocol-compatible server passes false.

redact strips private attributes exactly as ScanRawRedacted does. An empty projection means no attribute filter (the whole ad, matching QueryRawProject semantics upstream). Inline-name collections yield nothing, as with ScanRaw.

func (*Collection) ScanRawRedacted added in v0.16.3 

func (c *Collection) ScanRawRedacted() iter.Seq[RawAd]

ScanRawRedacted is ScanRaw with private attributes stripped (see QueryRawRedacted).

func (*Collection) ScanRawWire added in v0.16.4 

func (c *Collection) ScanRawWire(projection []string, redact bool) iter.Seq[[]byte]

ScanRawWire yields each ad as a WIRE-FORM ROW: a self-contained inline-names ad holding only the selected entries, assembled by slice copies from the stored bytes (see wire.AppendAdSubsetInline) -- nothing is decoded or rendered here. It is the relay scan for shipping a persistent table's ads across a process boundary (dbrpc) with the old-ClassAd render deferred to the far edge (RenderRawAdInline); the consumer needs no intern table and no data key (at-rest-encrypted values are opened during assembly).

projection restricts the entries to the named attributes (case-insensitive; MyType/TargetType always ship so the ad stays typed); empty means every entry. redact drops private attributes -- pre-pruned from a projection, or tested per entry with a byte-gated predicate for the whole-ad case. The yielded row aliases a buffer reused across the iteration. Non-inline (RAM) collections yield nothing: their ads are not self-contained, and an in-process consumer reads them through the collection directly.

func (*Collection) SetEncryptedAttrs added in v0.7.0 

func (c *Collection) SetEncryptedAttrs(attrs []string) error

SetEncryptedAttrs replaces the explicit encrypted-attribute set at runtime (the toggle meta-command). It is a no-op-with-error if encryption is disabled, and errors if any named attribute is currently indexed (an encrypted value is opaque, so it cannot be indexed). Private attributes are always encrypted regardless of this set. New records use the new set immediately; existing records keep their prior form until rewritten (compaction/Rewrite re-encodes them under the current policy).

func (*Collection) SetTimeTravel added in v0.12.0 

func (c *Collection) SetTimeTravel(o *TimeTravelOptions)

SetTimeTravel enables, retunes, or (with a nil/zero option) disables point-in-time queries at runtime. Enabling starts recording checkpoints and retaining superseded versions from now on (it is not retroactive); disabling lets the next compaction reclaim the retained history. Safe to call concurrently with reads and writes.

func (*Collection) SidecarSizes added in v0.8.0 

func (c *Collection) SidecarSizes() SidecarSizes

SidecarSizes reports the live Collection's sealed-segment sidecar bytes: the mmap-backed index each sealed segment holds after the flip from the in-RAM segIndex -- a file mapping for a persistent collection (page-cache resident, evictable to disk) or an anonymous mapping for an in-memory one (off-heap, MADV_FREE-able to swap). Either way these bytes are NOT Go-heap memory, so they are reported apart from IndexSizes (which now measures only the active, still-in-RAM segments' postings). Together the two give the operator the full picture: heap postings on the hot active segment plus off-heap sidecar bytes on the sealed tail. It reads each segment's already-mapped bytes under the shard read lock -- no re-mapping -- so it is cheap enough for periodic sampling.

func (*Collection) StartAutoRetrain 

func (c *Collection) StartAutoRetrain(interval time.Duration, sampleMax, hotTopN int) (stop func())

StartAutoRetrain runs periodic maintenance in a background goroutine every interval, until the returned stop function is called (stop blocks until the goroutine exits). Each tick:

  • retrains the ZSTD dictionary from a sample of up to sampleMax ads (RetrainDict), and
  • if hotTopN > 0, refreshes the hot-attribute set to the topN most common attributes (RefreshHotSet), so the store self-tunes which attributes it front-loads for fast queries.

Retrain errors (e.g. the pure-Go BuildDict declining a small/homogeneous corpus) are ignored — the previous codec keeps working.

Cost note: retraining recompacts every shard. Compaction is concurrent (the recompression runs without the shard lock), so writers and scanners are not blocked for its duration; only two brief per-shard critical sections take the lock. The 15-minute default still keeps the work rare.

func (*Collection) Stats 

func (c *Collection) Stats() Stats

Stats returns a snapshot of the collection's storage. It takes each shard's read lock briefly, so it is safe to call concurrently with readers and writers.

func (*Collection) SuggestDrops 

func (c *Collection) SuggestDrops(sampleMax int) []DropSuggestion

SuggestDrops recommends configured indexes to remove, from observed query demand and a sample of up to sampleMax live ads. It flags two cases: an index no query has ever filtered on ("unused"), and one whose sampled values are effectively constant ("low-cardinality", ≤1 distinct value — the index cannot prune). It is advisory: apply via DropIndex. Demand is cumulative since the collection was created, so give a workload time to run before trusting "unused".

func (*Collection) SuggestIndexes 

func (c *Collection) SuggestIndexes(sampleMax int) []IndexSuggestion

SuggestIndexes recommends value/categorical indexes from observed query demand and a sample of up to sampleMax live ads. It is advisory: apply the returned CategoricalAttrs/ValueAttrs via New (or a future dynamic Reindex). Attributes already indexed are not re-suggested. Results are ordered by demand, most first.

func (*Collection) TimeTravelConfig added in v0.12.0 

func (c *Collection) TimeTravelConfig() (opts TimeTravelOptions, enabled bool)

TimeTravelConfig reports the collection's current time-travel settings and whether it is enabled (for persisting the runtime toggle; see db.saveIndexConfig).

func (*Collection) Truncate added in v0.7.0 

func (c *Collection) Truncate()

Truncate removes every ad from the collection, leaving it empty (an in-place reset used by a DB restore before it reloads a snapshot). It resets each shard's directory and count and retires its segments -- RAM segments are dropped for the GC; persistent segments are munmap'd + unlinked once no in-flight scan references them (the compaction pin/reap protocol). A concurrent scan that already captured its window reads the old data safely until it finishes; scans and writes that start after Truncate see an empty collection. Ordered indexes are cleared in place. Callers needing atomicity against concurrent writers must serialize Truncate with them (the db layer holds its DB-wide lock); at the shard level Truncate is itself consistent.

func (*Collection) Update 

func (c *Collection) Update(batch []AdUpdate) error

Update applies a batch of inserts/updates and returns only once every ad is committed and visible to new scans. Ads are encoded (and compressed) outside the shard locks; each affected shard then applies its updates under one lock acquisition at a single new commit sequence.

func (*Collection) UpdateOld 

func (c *Collection) UpdateOld(batch []OldAdUpdate) error

UpdateOld applies a batch of inserts/updates whose ads arrive in old-ClassAd form. It encodes each ad directly to the wire form, attribute by attribute, without materializing an intermediate ast.ClassAd: scalar-literal attributes (the common case) are written straight to wire, and only genuinely computed values are parsed with the expression parser. This is the efficient path for ads read from a socket. Commit semantics match Update.

func (*Collection) Watch added in v0.4.0 

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

Watch replays everything that may have changed since cursor (nil ⇒ a full replay from empty), then streams live changes until ctx is cancelled or the consumer stops. Requires Options.WatchHistory > 0. See docs/WATCH.md and WatchEvent.

func (*Collection) WatchCursor added in v0.7.0 

func (c *Collection) WatchCursor() ([]byte, error)

WatchCursor returns an opaque cursor pointing at the current head of the change log, so a subsequent Watch(ctx, cursor) streams only changes from now on -- no initial replay of the current contents. Requires Options.WatchHistory > 0. It only snapshots each shard's commit sequence (no registration, no replay), so it is cheap.

type CommitResult added in v0.6.0 

type CommitResult struct {
	Committed int
	Conflicts [][]byte
}

CommitResult reports a transaction's outcome. Conflicts holds the keys whose write lost a write-write race and were not applied; the caller may re-read and retry just those. The other buffered writes committed.

func (CommitResult) Conflicted added in v0.6.0 

func (r CommitResult) Conflicted() bool

Conflicted reports whether any buffered write lost a conflict.

type ConjunctExplain added in v0.6.0 

type ConjunctExplain struct {
	// Text is the conjunct rewritten over the slot (job refs baked, TARGET scope dropped).
	Text string `json:"text"`
	// Probed is true when the conjunct is covered by an active index probe -- it prunes
	// the candidate set before the bilateral re-verify.
	Probed bool `json:"probed"`
	// Indexed is true when the conjunct's selectivity is estimable from an index (a
	// superset of Probed: it also covers bare boolean flags, which are estimable via
	// `attr == true` but are not extracted as pushdown probes).
	Indexed bool `json:"indexed"`
	// HasSelectivity reports whether TrueFrac is populated.
	HasSelectivity bool `json:"hasSelectivity"`
	// TrueFrac is the estimated fraction of slots for which the conjunct holds.
	TrueFrac float64 `json:"trueFrac"`
	// ResourceSide is true for a conjunct that came from the MATCH's resource-side filter
	// (WHERE TARGET / NOPREEMPT) rather than the job's Requirements. It is applied to
	// matched candidates as a post-filter, so it re-checks rather than prunes today.
	ResourceSide bool `json:"resourceSide,omitempty"`
}

ConjunctExplain is one top-level && conjunct in the match's evaluation order.

type DropSuggestion 

type DropSuggestion struct {
	Attr   string // canonical attribute name
	Kind   string // "categorical" or "value" (how it is currently indexed)
	Reason string // "unused" (never queried) or "low-cardinality" (no pruning power)

	QueriesEq      int64
	QueriesRange   int64
	SampledPresent int
	DistinctValues int
	Capped         bool
}

DropSuggestion recommends removing one configured index, with the rationale.

type Hasher 

type Hasher interface {
	Hash(key []byte) uint64
}

Hasher maps a stable key to a 64-bit hash. The hash routes a key to a shard and indexes the per-shard directory; it is never used as the ad's identity (the full key is stored inline in each record and compared on lookup), so hash collisions are resolved exactly and never lose data.

type IndexSize added in v0.6.0 

type IndexSize struct {
	Attr  string `json:"attr"`
	Kind  string `json:"kind"`  // "categorical" | "value"
	Bytes int64  `json:"bytes"` // resident posting bytes (roaring bitmaps + keys) across all live segments
	// SketchBytes is the resident memory of this attribute's per-segment sketches --
	// the categorical bloom filter and the HyperLogLog distinct-count registers --
	// reported apart from Bytes so it is visible rather than hidden in the posting total.
	SketchBytes int64   `json:"sketchBytes"`
	Auto        bool    `json:"auto"` // created by the auto-tuner (vs human/Options)
	Frac        float64 `json:"frac"` // Bytes as a fraction of the live data bytes
}

IndexSize is the measured memory footprint of one attribute's index.

type IndexSizes added in v0.6.0 

type IndexSizes struct {
	PerIndex   []IndexSize `json:"perIndex"`
	TotalBytes int64       `json:"totalBytes"` // posting bytes (the auto-tuner's budget denominator)
	// TotalSketchBytes is the sum of every index's SketchBytes (bloom + HLL). It is
	// reported separately and is NOT folded into TotalBytes/Frac, so the watermark and
	// auto-tuner budget stay calibrated on posting bytes; sketch memory is bounded and
	// small (<=8 KiB bloom + 1 KiB HLL per categorical attr per segment).
	TotalSketchBytes int64   `json:"totalSketchBytes"`
	DataBytes        int64   `json:"dataBytes"` // live compressed record bytes
	Frac             float64 `json:"frac"`      // TotalBytes / DataBytes
}

IndexSizes is the collection's index memory, per attribute and in total, against the live data bytes -- the denominator for the "index is N% of data" watermark.

type IndexSuggestion 

type IndexSuggestion struct {
	Attr string // canonical (first-seen) attribute name
	Kind string // "categorical" (string equality/membership) or "value" (numeric + range)

	QueriesEq    int64 // equality/membership probes observed
	QueriesRange int64 // range probes observed

	SampledPresent int     // sampled ads with this attribute present
	StringFrac     float64 // fraction of present values that are string literals
	NumericFrac    float64 // fraction that are numeric literals
	DistinctValues int     // distinct values in the sample (a lower bound if Capped)
	Capped         bool
}

IndexSuggestion recommends indexing one attribute, with the rationale (the two signals) so the decision is explainable.

type MatchExplain added in v0.6.0 

type MatchExplain struct {
	// HasRequirements is false when the job has no Requirements (it then matches every
	// slot -- a full scan with no pruning).
	HasRequirements bool `json:"hasRequirements"`
	// SlotPredicate is the job's Requirements rewritten over the slot: TARGET.attr
	// becomes the slot's own attribute and every job reference is baked to its value,
	// e.g. `Memory >= 4096 && Arch == "X86_64"`. This is the predicate whose probes
	// drive candidate pruning (the bilateral match still re-verifies every candidate).
	SlotPredicate string `json:"slotPredicate"`
	// Probes are the rewritten predicate's index-satisfiable conjuncts and their index
	// status on the resource collection.
	Probes []ProbeExplain `json:"probes"`
	// IndexUsable is how many probes prune via an index.
	IndexUsable int `json:"indexUsable"`
	// Plan is the access path over the resource slots: "indexed" (visit only candidate
	// slots), "parallel-scan", or "serial-scan" (match every slot).
	Plan        string `json:"plan"`
	Parallelism int    `json:"parallelism"`
	Shards      int    `json:"shards"`
	// TotalResources is the resource (slot) count, the denominator for selectivity.
	TotalResources int `json:"totalResources"`
	// EvalOrder is the top-level && conjuncts in the order the bilateral match evaluates
	// them (after short-circuit reordering), each tagged with its role: an active pruning
	// probe, or a re-check whose selectivity (from the index, when available) drove where
	// it sorts. It makes the ordering transparent -- e.g. an always-true capability flag
	// appears last as a ~99%-true re-check, not a candidate filter.
	EvalOrder []ConjunctExplain `json:"evalOrder,omitempty"`
}

MatchExplain describes how matchmaking a specific job against this (resource) collection would execute: the job's Requirements rewritten over the slot (with the job's attributes baked to constants), the index-satisfiable probes that rewrite yields, and which of them prune candidates via a configured index.

type OldAdUpdate 

type OldAdUpdate struct {
	Key  []byte
	Text string
}

OldAdUpdate is one insert-or-update whose ad is supplied in "old ClassAd" serialization (newline-separated `Name = Value` lines, as sent over a TCP socket), rather than as a parsed *classad.ClassAd.

type OpStat added in v0.11.0 

type OpStat struct {
	Count int64 `json:"count"`
	Nanos int64 `json:"nanos"`
	// MaxNanos is the longest single occurrence seen, and Buckets is the latency
	// histogram (see LatencyBucketBoundsNanos: buckets[i] counts occurrences <= bound[i],
	// the last entry counts the overflow above the final bound). Both surface the tail
	// the mean (Nanos/Count) hides. Omitted from JSON when empty so an older peer that
	// never populated them stays byte-compatible.
	MaxNanos int64   `json:"maxNanos,omitempty"`
	Buckets  []int64 `json:"buckets,omitempty"`
}

OpStat is one operation's cumulative call count and total wall-nanoseconds. Both are monotonic; a scraper derives rate and mean latency (Nanos/Count) from deltas.

type OpStats added in v0.11.0 

type OpStats struct {
	ShardWriteWait OpStat `json:"shardWriteWait"`
	ShardWriteHold OpStat `json:"shardWriteHold"`
	SegmentAlloc   OpStat `json:"segmentAlloc"`
	Sync           OpStat `json:"sync"`
	// CommitSync is the per-commit durability wall time: how long a Txn.Commit blocked
	// flushing all of its shards to disk. Since those shard flushes now run in parallel,
	// this is the commit's critical path (≈ the slowest shard), which the per-shard Sync
	// total -- now the summed fsync WORK, not the latency -- no longer reveals. Count is
	// the number of durable commits; Nanos/Count is mean commit durability latency.
	CommitSync OpStat `json:"commitSync,omitempty"`
	Compact    OpStat `json:"compact"`
	Retrain    OpStat `json:"retrain"`
	Reindex    OpStat `json:"reindex"`
}

OpStats is a snapshot of a collection's operational timing counters -- the wall time its callers spent blocked in, or holding, each of the store's stall points. Shard counters are summed across all shards. See opMetrics for what each measures. OpStats reports, per stall point, the cumulative call count and total wall time. The collection-wide snapshot lock (Truncate/Restore) lives one layer up in the db package, which composes its own snapshot-lock counter onto this snapshot.

type Options 

type Options struct {
	// Shards is the number of independently-locked shards; rounded up to a power
	// of two. Default 16.
	Shards int
	// SegmentSize is the arena segment size in bytes. Default 8 MiB.
	SegmentSize int
	// Hasher routes keys to shards / directory buckets. Default 64-bit FNV-1a.
	Hasher Hasher
	// Codec compresses stored ad bytes. Default identity (no compression).
	Codec Codec
	// HotAttrs names the "popular" attributes to front-load in each ad's hot
	// header, so a query filtering on them resolves each in O(1) instead of
	// scanning the ad body. Typically the attributes common queries filter on
	// (e.g. Cpus, Memory, Arch, OpSys, State). Optional.
	HotAttrs []string
	// MatchClosureRoots names attributes (typically "Requirements") whose transitive
	// self-reference closure is front-loaded into each ad's hot header at encode time,
	// so matching a wide ad reads only the match-relevant attributes (via the hot
	// header) instead of decoding the whole ad. Optional; enables the hot-closure match
	// fast path. The frequency/HotAttrs hot set is unioned in, so queries are unaffected.
	MatchClosureRoots []string
	// CommitSync, if set, is called once per committed (possibly group-coalesced)
	// batch on a shard, after the writes are applied — the point at which a future
	// durable collection would fsync/serialize the batch. Group commit amortizes
	// this across all writers that commit together. It must be safe for concurrent
	// use across shards. Optional; default no-op (in-memory store).
	CommitSync func()
	// CategoricalAttrs names string-valued attributes to index for equality and
	// set-membership (`Attr == "x"`, `Attr == "x" || Attr == "y"`). ValueAttrs
	// names numeric attributes to index for equality and range (`Attr >= n`). A
	// query filtering on an indexed attribute visits only candidate ads instead of
	// scanning all of them. Optional.
	CategoricalAttrs []string
	ValueAttrs       []string
	// Ordered configures maintained, filtered, ordered indexes -- the schedd
	// priority-queue pattern (see docs/MATCH.md §3 and Collection.Ordered). Each
	// spec groups members into partitions and keeps them sorted on the write path,
	// so a negotiator can iterate a partition in order (and resume) without
	// re-sorting each cycle. A malformed expression in a spec panics New. Optional.
	Ordered []OrderSpec
	// Dir, if set, makes the collection persistent: arenas are memory-mapped files
	// under this directory and committed writes are flushed to disk (see Open).
	// Empty ⇒ in-memory (the default). Unix-only.
	Dir string
	// QueryParallelism controls cross-segment fan-out for large full-scan queries:
	//   0 (default) ⇒ auto — the library picks the policy (currently: up to 6 workers
	//                 per query, clamped to GOMAXPROCS). The meaning of auto may change
	//                 across releases.
	//   1           ⇒ serial (never fan out).
	//   N ≥ 2       ⇒ fan out with up to N workers per query.
	// In every non-serial case fan-out is still gated by a work-size threshold, never
	// exceeds the segment count, and draws from a machine-wide worker budget shared
	// across concurrent queries (so it degrades to serial under load rather than
	// oversubscribing). Small scans and indexed queries run serial regardless.
	// See parallel_scan.go / docs/PARALLEL_QUERY.md.
	QueryParallelism int
	// WatchHistory enables the Watch subscription verb (see docs/WATCH.md). It is the
	// per-shard capacity of the delete journal — how many recent deletes are retained
	// so a resuming watcher can be told precisely which keys were removed. 0 (default)
	// disables Watch entirely (no journal, no live notification, zero write-path cost).
	// A larger value lets clients resume incrementally from further back before
	// falling to a full replay.
	WatchHistory int
	// WatchBuffer is the per-watcher live event-channel capacity (default 1024). A
	// watcher that overflows it is told to resync rather than stalling writers.
	WatchBuffer int
	// WatchCoalesce, if > 0, batches live Watch events into windows of this
	// duration and emits only the newest event per key in each window (default 0 =
	// off, one event per change). It smooths bursty churn -- e.g. a freshly
	// submitted job that takes many SetAttribute updates in quick succession is
	// delivered as a single Upsert of its settled state. Only the last event of a
	// flushed window carries a cursor, so a mid-window consumer crash resumes from
	// the prior window and re-delivers (at-least-once is preserved). Catch-up is
	// never coalesced.
	WatchCoalesce time.Duration

	// ParentKeyFor, if set, gives an ad a parent: it maps a child's key to its
	// parent's key (return nil for a top-level ad with no parent). A child then
	// resolves attribute references it lacks by falling through to its parent --
	// the primitive "join" the job queue needs, where a proc ad chains to its
	// cluster ad so a query like `DAGManJobId == 42` sees the cluster's attribute.
	// A family (a root and all its descendants) is co-located in one shard -- the
	// collection routes every key to the shard of its ultimate root -- so chained
	// evaluation is consistent and lock-free. Parent bindings are immutable: a
	// key's parent must not change across updates. Default nil disables chaining
	// (every ad is standalone) and keeps the plain single-key routing/scan.
	ParentKeyFor func(key []byte) []byte

	// IsStructural, if set, marks a key as a structural (parent-only) ad that
	// exists solely to be chained to -- e.g. a job cluster ad. Structural ads are
	// stored and used as parents but are hidden from Query/Scan/Watch results by
	// default (like condor_q omitting cluster ads). Default nil treats every ad as
	// a normal, visible ad. Only meaningful together with ParentKeyFor.
	IsStructural func(key []byte) bool

	// ParentPrivateAttrs are parent attributes children do NOT inherit and that do
	// NOT trigger a watch fan-out when they change -- e.g. a job cluster ad's
	// factory bookkeeping (JobMaterializeNextProcId, ...), which mutates per proc
	// materialized but is meaningless to a proc. They are excluded from flattening
	// and from the parent-change diff, so high-frequency parent-internal churn does
	// not fan out to every child. Case-insensitive. Only meaningful with ParentKeyFor.
	ParentPrivateAttrs []string

	// DataKey enables encryption at rest: the named attributes' values are sealed with
	// AES-256-GCM under this key before being written to a segment. It must be the DB
	// master key's DataInfo subkey (crypt.Subkey(master, crypt.DataInfo)) -- distinct
	// from the master itself -- so a stolen database is useless without a pool key.
	// Empty ⇒ encryption disabled (EncryptedAttrs is then inert). See encrypt.go.
	DataKey []byte
	// EncryptedAttrs names the attributes to encrypt at rest (case-insensitive). Only
	// meaningful with DataKey set. An encrypted attribute may NOT also be indexed
	// (CategoricalAttrs/ValueAttrs) -- New panics on overlap -- since its value is
	// opaque at rest. The set can be changed at runtime via the toggle meta-command;
	// existing records keep their prior form until rewritten.
	EncryptedAttrs []string

	// TimeTravel, if non-nil, enables point-in-time ("AS OF") queries: superseded
	// record versions committed within MaxDistance of now are retained (not reclaimed
	// by compaction), and a sparse time->seq checkpoint is recorded so a wall-clock
	// time resolves to the commit sequence that was current then. Nil (default)
	// disables it with zero write-path cost -- retention collapses to the current seq,
	// exactly as before. See timeseq.go. It can be toggled at runtime.
	TimeTravel *TimeTravelOptions
}

Options configures a Collection.

type OrderCursor added in v0.5.2 

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

OrderCursor marks a position in an ordered scan. Its zero value starts at the beginning of a partition; the cursor yielded alongside an ad resumes strictly after that ad, so a negotiator can iterate a partition across several calls.

type OrderSpec added in v0.5.2 

type OrderSpec struct {
	Partition string    // attribute partitioning the index; empty = one global partition
	Where     string    // membership predicate; empty = every ad is a member
	Keys      []SortKey // sort order within a partition
	// Cluster names the attributes whose combined value forms a member's cluster
	// signature -- a hash surfaced alongside each ad by Ordered. The schedd's RRL
	// fold groups the ordered stream into runs of equal signature (a stored-value
	// compare instead of re-hashing the requirement attributes each cycle). Computed
	// once on the write path. Empty = no signature (Ordered reports 0). Optional.
	Cluster []string
}

OrderSpec configures a maintained, filtered, ordered index over a collection -- the schedd's priority-queue pattern (§3 of docs/MATCH.md). Members are grouped into independent runs by Partition (e.g. per Owner); within a partition they are ordered by Keys, with insertion order as the final tiebreaker ("the order the job entered the queue"). Where restricts membership to the ads that matter (e.g. idle jobs), so the index tracks only the churn-prone subset.

type OrderedAd added in v0.5.2 

type OrderedAd struct {
	Ad        *classad.ClassAd
	Cursor    OrderCursor
	Signature uint64
}

OrderedAd is one step of an ordered scan: the member ad, a cursor that resumes right after it, and the cluster signature (0 unless OrderSpec.Cluster is set). The signature lets an app run-length-fold the stream into RRLs with a stored-value compare instead of re-hashing each ad's clustering attributes.

type PrefixDecompressor added in v0.16.4 

type PrefixDecompressor interface {
	DecompressPrefix(dst, src []byte, want int) ([]byte, error)
}

PrefixDecompressor is an optional Codec capability: decompress only (approximately) the first want bytes of a record. With the hot region encoded as a physical prefix (see wire.EncodeInlineWithHotEnc), a hot-covered projected read decompresses a couple of KB instead of the whole record. The result may be shorter than want (the record ends first -- then it is the complete record) and is otherwise a TRUNCATED record: the caller must be prepared for parses to run off the end and fall back to a full Decompress.

type ProbeExplain added in v0.6.0 

type ProbeExplain struct {
	Attr    string `json:"attr"`
	Op      string `json:"op"`
	Indexed bool   `json:"indexed"`        // this probe can use an index to prune
	Kind    string `json:"kind,omitempty"` // "categorical" | "value" | "" (attr not indexed)

	// Selectivity and EstCandidates are the planner's estimate of how much this
	// probe prunes, present only when Indexed: EstCandidates is the estimated
	// number of ads the index would visit for this probe (summed over the segment
	// indexes' selectivity stats), and Selectivity is that as a fraction of the
	// total (lower is more selective). HasSelectivity distinguishes a real 0 from
	// "not estimated".
	HasSelectivity bool    `json:"hasSelectivity"`
	Selectivity    float64 `json:"selectivity,omitempty"`
	EstCandidates  int64   `json:"estCandidates,omitempty"`
}

planIndex matches the query's probes against the configured indexes. Empty means no index-usable constraint (the store full-scans). ProbeExplain describes how one index-satisfiable conjunct of a query relates to the configured indexes.

type QueryExplain added in v0.6.0 

type QueryExplain struct {
	// Native reports wire-native evaluation: the query reads scalar-literal
	// attributes directly from the encoded ads, building no ClassAd per ad.
	Native bool `json:"native"`
	// Probes are the query's index-satisfiable conjuncts and their index status.
	Probes []ProbeExplain `json:"probes"`
	// IndexUsable is how many probes can prune via an index.
	IndexUsable int `json:"indexUsable"`
	// Plan is the chosen access path: "indexed" (visit index candidates),
	// "parallel-scan", or "serial-scan" (full scan).
	Plan string `json:"plan"`
	// Parallelism is the configured per-query worker cap; Shards is the shard count.
	Parallelism int `json:"parallelism"`
	Shards      int `json:"shards"`
	// TotalAds is the live ad count, the denominator for probe selectivity.
	TotalAds int `json:"totalAds"`
}

QueryExplain is a description of how the store would execute a query, for the diagnostic ".explain" command -- what the planner sees and the access path it would choose.

type RankedMatch added in v0.6.0 

type RankedMatch struct {
	Ad      *classad.ClassAd
	Rank    float64
	HasRank bool
}

RankedMatch is a matched ad and the job's Rank of it (HasRank is false when the job's Rank does not evaluate to a number for that ad).

type RawAd 

type RawAd struct {
	Exprs      [][]byte
	MyType     string
	TargetType string
}

RawAd is a query result rendered as old-ClassAd wire parts -- the "Name = Value" expression byte slices plus MyType/TargetType -- decoded straight from the stored form with no ast/classad ClassAd. A collector can hand Exprs to message.PutClassAdRawBytes to stream a result set without ever building an AST.

Exprs (and their backing bytes) alias a buffer reused across the iteration, so a consumer must finish with one RawAd -- e.g. write it to the wire -- before advancing the iterator to the next.

type Retention 

type Retention struct {
	MaxSegments int   // keep at most this many sealed segments
	MaxBytes    int64 // keep at most this many bytes of sealed segment files
	// MaxAgeAttr / MaxAge: drop a segment whose max value of the given numeric attr
	// (e.g. "CompletionDate", unix seconds) is older than Now-MaxAge. Now is supplied
	// per Rotate call so the store needs no clock of its own.
	MaxAgeAttr string
	MaxAge     float64
}

Retention bounds how much history is kept. Rotation (Rotate, or automatically after a seal) drops the oldest sealed segments until every set bound is met. A zero field means "no bound on that axis".

type SidecarSizes added in v0.8.0 

type SidecarSizes struct {
	Segments    int   `json:"segments"`    // sealed segments with a sidecar
	MappedBytes int64 `json:"mappedBytes"` // total sidecar bytes (mmap-backed, evictable)
	MPHBytes    int64 `json:"mphBytes"`    // of MappedBytes, minimal-perfect-hash structures
	BloomBytes  int64 `json:"bloomBytes"`  // of MappedBytes, bloom filters
}

SidecarSizes reports the on-disk index bytes of an Archive's sealed sidecars, broken out so the minimal-perfect-hash and bloom overhead is visible. These are distinct from the live Collection's IndexSizes: those measure HEAP-resident postings + sketches, whereas sidecar bytes live in the page cache -- demand-paged and evictable under memory pressure. They are reported as a separate budget, never folded into a heap figure, so the "index is N% of data" watermark stays an honest measure of resident memory.

type SortKey added in v0.5.2 

type SortKey struct {
	Expr string // attribute name or expression, e.g. "JobPrio" or "QDate"
	Desc bool   // sort this component descending
}

SortKey is one component of an ordered index's sort order: an attribute name or expression evaluated against each member ad, ascending by default.

type Stats 

type Stats struct {
	// Ads is the number of live keys (== Len).
	Ads int
	// Segments is the number of arena segments across all shards.
	Segments int
	// ArenaBytes is the total capacity of those segments -- the memory the
	// collection has reserved for record storage, and the dominant term in its
	// resident footprint. (RAM segments back this on the Go heap; persistent
	// segments back it with an mmap.)
	ArenaBytes int64
	// UsedBytes is the bytes written into segments: live records plus superseded
	// (dead) records not yet reclaimed by compaction.
	UsedBytes int64
	// DeadBytes is the bytes belonging to superseded records -- reclaimable by
	// Compact. LiveBytes = UsedBytes - DeadBytes.
	DeadBytes int64
}

Stats is a point-in-time summary of a collection's storage, for observability and capacity planning (e.g. Prometheus metrics). All byte counts are of the dictionary-compressed on-arena form, not the ads' decompressed text.

func (Stats) LiveBytes 

func (s Stats) LiveBytes() int64

LiveBytes is the compressed size of the live records (UsedBytes - DeadBytes).

type TimeTravelOptions added in v0.12.0 

type TimeTravelOptions struct {
	// MaxDistance is how far back queries may travel: versions superseded more than
	// this long ago are eligible for reclamation. Larger keeps more history (more
	// retained dead bytes). Must be > 0.
	MaxDistance time.Duration
	// CheckpointInterval is the granularity of the time->seq map: a checkpoint is
	// recorded at most once per interval, on the first commit that crosses the
	// boundary, so an AS OF time resolves to within one interval. Default 1 minute;
	// lower for finer resolution at proportionally more checkpoints.
	CheckpointInterval time.Duration
}

TimeTravelOptions configures point-in-time queries for a collection (see the TimeTravel option and timeseq.go).

type Txn added in v0.6.0 

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

Txn is an optimistic, snapshot-isolation transaction over a Collection. Not safe for concurrent use by multiple goroutines; each goroutine uses its own Txn.

func (*Txn) Commit added in v0.6.0 

func (tx *Txn) Commit() CommitResult

Commit applies the buffered writes, each independently: a write whose key is unchanged since the transaction's snapshot commits; one whose key was modified by another committer is reported in CommitResult.Conflicts and not applied (the successful writes are not rolled back). The transaction must not be used after Commit.

func (*Txn) Delete added in v0.6.0 

func (tx *Txn) Delete(key []byte)

Delete buffers a delete of key. Nothing is written until Commit.

func (*Txn) Get added in v0.6.0 

func (tx *Txn) Get(key []byte) (*classad.ClassAd, bool)

Get returns the ad for key as the transaction sees it: its own buffered write if any (read-your-writes), else the version live at the transaction's snapshot. On a chained (parent/child) collection it resolves inherited attributes by merging the parent as of the same snapshot -- mirroring Collection.Get, transactionally.

func (*Txn) Put added in v0.6.0 

func (tx *Txn) Put(key []byte, ad *classad.ClassAd)

Put buffers an insert or update of key. Nothing is written until Commit.

func (*Txn) SetDurable added in v0.6.0 

func (tx *Txn) SetDurable(d bool)

SetDurable controls whether Commit runs the durability sync (default true). A nondurable commit is visible immediately (readers and watchers see it) but its disk flush is deferred to a later durable commit or flush -- the classad_log.h CommitNondurableTransaction batching. No effect on an in-memory collection, whose sync is already a no-op.

type WatchEvent added in v0.4.0 

type WatchEvent struct {
	Kind   WatchKind
	Key    []byte
	Ad     *classad.ClassAd
	Cursor []byte
}

WatchEvent is one item in a Watch stream. Cursor, when non-nil, is an opaque token the client persists after processing the event and passes back to Watch on resume. Catch-up data events carry no cursor (persist at WatchSynced); live events do.

type WatchKind added in v0.4.0 

type WatchKind uint8

WatchKind is the type of a WatchEvent. 

const (
	// WatchUpsert carries the full ad for an added or updated key (Ad is set).
	WatchUpsert WatchKind = iota
	// WatchDelete signals a key was removed (Ad is nil).
	WatchDelete
	// WatchReset tells the client to discard its state (build into a shadow): an
	// authoritative full snapshot of Upserts follows, ending at WatchSynced. Emitted
	// when a precise incremental resume is impossible (first subscribe, cursor older
	// than the delete-retention window, or a different store generation).
	WatchReset
	// WatchSynced marks the end of the initial catch-up/snapshot: the client is now
	// live. Its Cursor is a durable resume point (and, after a Reset, the point to
	// swap the shadow state live).
	WatchSynced
	// WatchResync tells the client the live stream fell behind and it must reconnect
	// with its last persisted cursor (which re-enters catch-up). No state is implied.
	WatchResync
)

Source Files

View all Source files

Directories

Path Synopsis
Package crypt is the encryption-at-rest key hierarchy for a ClassAd database.
 Package crypt is the encryption-at-rest key hierarchy for a ClassAd database.
Package vm compiles ClassAd expressions to a linear instruction stream and interprets them against a scope, replacing per-query AST walks.
 Package vm compiles ClassAd expressions to a linear instruction stream and interprets them against a scope, replacing per-query AST walks.
Package wire defines a compact TLV binary form of a ClassAd with attribute-key interning and a hot-attribute header, plus the encoder/decoder that convert to and from the fully-public ast.ClassAd representation.
 Package wire defines a compact TLV binary form of a ClassAd with attribute-key interning and a hot-attribute header, plus the encoder/decoder that convert to and from the fully-public ast.ClassAd representation.

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.