catalog

package
v0.18.13 Latest Latest
Warning

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

 Go to latest Published: Sep 3, 2026 License: AGPL-3.0 Imports: 25 Imported by: 0

Documentation

Overview

Package catalog manages table schema and partition metadata. Metadata is stored in a MetaKV (NATS KV in production, MemKV in tests). Data files remain in object storage (S3/MinIO).

All KV keys are prefixed with a cluster ID to support federation: 

<clusterID>.meta              → CatalogMeta JSON
<clusterID>.table.<name>      → TableMeta JSON
<clusterID>.manifest.<name>   → PartitionManifest JSON

Index

Constants

View Source 
const DefaultDropTableGrace = 30 * time.Minute

DefaultDropTableGrace bounds how long a dropped table's data files stay physically present after DropTable returns, mirroring compaction.DefaultDeleteGrace's reasoning exactly: a query dispatched against the table's last manifest resolved its file list at dispatch time and keeps reading those exact paths until it finishes, so deleting the bytes the instant the manifest disappears races every such query. No NEW query can be racing — the table is already gone from meta.Tables — so the grace only has to outlive work already in flight.

Nothing ENFORCES that it does, and an operator enabling reclaim has to know it: wadjet's --query-timeout defaults to 0 (unlimited), so a long analytical query can outlive any grace. The rule is to keep the query timeout at or below the drop grace, or raise the grace above the longest query allowed. The failure mode if you don't is a query failing on a missing object, not a wrong answer — but it is still a failure the operator chose. See docs/adr/0020-drop-table-reclaim-is-opt-in.md.

View Source 
const (
	HistDefaultBuckets = 64
)

Histogram is an equi-depth histogram over a column's values. Each bucket holds boundary values and a count. For numeric columns, the boundaries are int64-encoded; the planner converts to/from float64 for range comparisons.

Equi-depth means each bucket holds approximately the same number of values (1/K of the total). Bucket boundaries adapt to the data distribution: dense regions get narrow buckets, sparse regions wide. This gives accurate selectivity estimates for both common and rare values.

Used in stats.estimatePredSelectivity for range/equality filters when the column has a histogram in the catalog. Replaces hardcoded 0.33 / 0.1 fractions with data-driven estimates.

Wire format (binary, version-1): 

[1]   version (1)
[1]   bucket count K (≤ 255)
[1]   value type code (0=int64, 1=float64, 2=bytes)
[1]   reserved
[8]   total values
[K+1] boundary values (K buckets → K+1 boundaries)
[K*8] per-bucket counts (uint64 LE)

Boundary encoding depends on type code:

  • int64: 8 bytes LE per value
  • float64: 8 bytes LE (math.Float64bits) per value
  • bytes: uint16 length prefix + raw bytes per value
View Source 
const (
	SampleDefaultSize = 256
)

ColumnSample is a fixed-size random sample of a column's values, persisted in FileColumnStats so the catalog can build aggregate histograms across files at query time. The sample is stored sorted so cross-file merge is a sorted-merge of K-way streams; the histogram is built once from the merged sample on AggregateColumnStats.

Storage: ~256 values × 8 bytes for numerics = 2 KB per column per file. Comparable to the per-file Histogram size but mergeable without distribution-assumption hacks.

Type-discriminated wire format mirrors Histogram's encodeValue:

  • int64: 8 bytes LE per value
  • float64: 8 bytes LE (math.Float64bits) per value
  • bytes: uint16 length prefix + raw bytes per value

Variables

View Source 
var ErrKeyNotFound = errors.New("key not found")

ErrKeyNotFound is returned when a key does not exist in the KV store.

View Source 
var ErrRevisionMismatch = errors.New("revision mismatch")

ErrRevisionMismatch is returned when a CAS update fails due to a concurrent modification (the key's current revision != expected).

View Source 
var ErrTableNotFound = errors.New("not found")

ErrTableNotFound marks a GetTable miss: the catalog was reachable and the table is definitely absent. Callers distinguish it (errors.Is) from a transport failure, where the table's existence is unknown — the planner rejects a query on the former (42P01) and stays conservative on the latter.

Functions

func AddValueToHLL 

func AddValueToHLL(h *HLL, v any, t parquet.TypeID)

AddValueToHLL hashes a value according to its declared parquet type and inserts the hash into the sketch.

Shared between the ingest path (which receives raw map[string]any rows pre-write) and the ANALYZE path (which decodes parquet rows post-write). Both produce hash-compatible HLLs that the catalog can merge.

Canonical encodings:

  • integer/temporal types → 8-byte LE of int64 representation
  • float32/float64 → 8-byte LE of math.Float64bits
  • bool → single byte 0/1
  • string/bytes/network-id types → raw bytes
  • timestamp → UnixMilli (time.Time) or int64 (raw)
  • duration → nanoseconds

The encoding intentionally normalizes int32 and int64 to the same byte stream so a column widened post-write hashes identically.

func DecodeSample 

func DecodeSample(r io.Reader) (values []any, totalSeen int64, typeCode uint8, err error)

DecodeSample reads a sample written by EncodeSample.

func DecodeTableRGMeta 

func DecodeTableRGMeta(r io.Reader) (map[string][]parquet.RowGroupStats, error)

DecodeTableRGMeta parses a v1 RG-metadata blob into a by-path map, the shape buildRGUnits consumes.

func DeletedRowsByFile added in v0.18.5 

func DeletedRowsByFile(markers []DeleteMarker) map[string]map[int64]bool

DeletedRowsByFile indexes a manifest's delete markers by file path, as the set of row positions WITHIN that file which no longer exist.

Every reader of a table owes this filter. The scanner applies it (its own deleteMarkers map is the same thing, built at Init) and so the SELECT path has always been right; the DML match scans did not, so an UPDATE matched rows in files its own earlier UPDATEs had already superseded, re-emitted them, and DOUBLED the row on every re-update — 1, 2, 4 (#674). A merge-on- read table has exactly one definition of which rows exist, and it is this.

func EncodeFileSketches 

func EncodeFileSketches(entries []FileSketchesEntry) []byte

EncodeFileSketches serializes a per-column sketch map to the v1 wire format. Empty input returns nil (no blob to upload).

func EncodeSample 

func EncodeSample(w io.Writer, values []any, totalSeen int64, typeCode uint8) error

EncodeSample writes a sorted sample to w in version-1 format. 

[1]   version (1)
[1]   type code
[2]   value count K (uint16 LE)
[8]   total observed count (uint64 LE, before reservoir downsampling)
[K*?] K values, type-discriminated encoding

func EncodeTableRGMeta 

func EncodeTableRGMeta(files []FileRGMeta) []byte

EncodeTableRGMeta serializes all files' row-group metadata to the v1 wire format. Empty input returns nil (no blob to upload).

func IsHLLSupportedType 

func IsHLLSupportedType(t parquet.TypeID) bool

IsHLLSupportedType reports whether the column type is one we collect HLL sketches for. Returns false for nested types whose "value" isn't a single scalar.

func MergeSamples 

func MergeSamples(files [][]byte) (values []any, totalSeen int64, typeCode uint8)

MergeSamples combines multiple per-file samples into a single sorted sample, weighted by each file's totalSeen so larger files contribute more samples to the merged distribution. Returns combined values, total observed across all files, and the shared type code.

If samples have mixed type codes (shouldn't happen for a valid column), the first sample's type wins; mismatched entries are skipped.

func SampleBytes 

func SampleBytes(values []any, totalSeen int64, typeCode uint8) []byte

SampleBytes serializes a snapshot for catalog persistence.

func SampleFromBytes 

func SampleFromBytes(b []byte) (values []any, totalSeen int64, typeCode uint8, ok bool)

SampleFromBytes parses a sample blob. Returns nil values on any error.

Types

type AlertMeta 

type AlertMeta struct {
	Name            string            `json:"name"`
	QueryText       string            `json:"query"`
	IntervalSeconds int64             `json:"interval_seconds"`
	WebhookURL      string            `json:"webhook_url,omitempty"`
	WebhookHeaders  map[string]string `json:"webhook_headers,omitempty"`
	InsertIntoTable string            `json:"insert_into_table,omitempty"`
	Enabled         bool              `json:"enabled"`
	CreatedAt       time.Time         `json:"created_at"`
	CreatedBy       string            `json:"created_by,omitempty"`
	// Creator identity snapshot for definer's-rights scheduled execution: the
	// alert query runs under this identity's ABAC subject on every tick (see
	// auth.IdentitySnapshot). Empty on alerts created before this existed —
	// the scheduler treats those fail-closed under enabled auth.
	CreatedByRole   string            `json:"created_by_role,omitempty"`
	CreatedByMethod string            `json:"created_by_method,omitempty"`
	CreatedByAttrs  map[string]string `json:"created_by_attrs,omitempty"`
	LastEvaluatedAt time.Time         `json:"last_evaluated_at,omitempty"`
	Version         int64             `json:"version"`
}

AlertMeta is the catalog entry for a CREATE ALERT definition. Stored at key "<clusterID>.alert.<name>" via MetaKV CAS.

type Catalog 

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

Catalog manages table metadata via KV and data via object storage.

func New 

func New(kv MetaKV, store objstore.Store, bucket string) *Catalog

New creates a new Catalog backed by the given KV store and object store.

func NewWithCluster 

func NewWithCluster(kv MetaKV, store objstore.Store, bucket string, clusterID string) *Catalog

NewWithCluster creates a Catalog with a specific cluster identity.

func NewWithStore 

func NewWithStore(store objstore.Store, bucket string) *Catalog

NewWithStore creates a Catalog using an in-memory KV (for tests/embedded use).

func (*Catalog) AddDeleteMarkers 

func (c *Catalog) AddDeleteMarkers(_ context.Context, tableName string, markers []DeleteMarker) error

AddDeleteMarkers adds delete markers to a table's manifest using CAS. Merges new markers with existing ones for the same file.

func (*Catalog) AddFiles 

func (c *Catalog) AddFiles(_ context.Context, tableName string, partValues map[string]string, partPath string, files []FileEntry) error

AddFiles adds file entries to the manifest for a given partition. Uses compare-and-swap to prevent concurrent flushes from losing updates. Idempotent per file path (mergeFileEntries): duplicate adds replace rather than append. For a writer minting brand-new paths, prefer AddNewFiles, which refuses a collision instead of masking it.

func (*Catalog) AddNewFiles added in v0.18.2 

func (c *Catalog) AddNewFiles(_ context.Context, tableName string, partValues map[string]string, partPath string, files []FileEntry) error

AddNewFiles adds newly-created file entries to the manifest for a given partition — the production write path for ingest, compaction, and delete-marker GC, none of which ever legitimately re-register a path (#494). Uses the same CAS retry as AddFiles, but a Path collision with an existing entry is refused with an error rather than silently replaced; see mergeNewFileEntries.

This is also where the ownership marker is stamped: every entry landing through here names an object wadjet itself just wrote, which is exactly the condition FileEntry.EngineWritten records and DropTable's physical reclaim requires. The caller's slice is copied rather than mutated — stamping in place would edit a FileEntry the caller still holds (and, for a caller that reuses a backing array, entries it has already handed elsewhere).

func (*Catalog) AggregateColumnStats 

func (c *Catalog) AggregateColumnStats(_ context.Context, tableName string) (map[string]TableColumnStats, error)

AggregateColumnStats computes table-level column statistics by merging per-file stats across all partitions. Returns nil for columns without stats.

func (*Catalog) AnalyzeTable 

func (c *Catalog) AnalyzeTable(ctx context.Context, name string) (int, error)

AnalyzeTable computes HyperLogLog sketches over every column of every file in the named table and writes them back into the manifest's FileColumnStats.HLL field. Idempotent — re-running ANALYZE replaces existing HLLs with freshly computed ones.

Used when a table's data was pre-staged (e.g., the SF10/SF100 EC2 deploy buckets) without going through the ingest path, so HLL never got collected at write time. The planner's NDV estimator then has real distinct-count data instead of falling back to min/max-range heuristics or FK-naming.

Strategy: for each file, download the parquet bytes, decode row groups via the existing parquet.Reader API, hash every column value into a per-(file, column) HLL. After all files of one table are processed, persist the augmented manifest.

Cost: one full table scan, decompressed but not joined. SF10 lineitem (60 chunks × 1M rows × 16 cols) takes 1-2 minutes serial. Cheap relative to a single query at the same scale; expected to run once per data load.

Returns the count of files analyzed and any error from the first failed file. Files that fail (corrupt, missing) are logged and skipped — partial coverage is better than total failure.

func (*Catalog) Bucket 

func (c *Catalog) Bucket() string

Bucket returns the bucket name.

func (*Catalog) ClusterID 

func (c *Catalog) ClusterID() string

ClusterID returns this catalog's cluster identifier.

func (*Catalog) CreateAlert 

func (c *Catalog) CreateAlert(_ context.Context, m AlertMeta) error

CreateAlert writes a new alert entry; fails if an alert with the same name exists.

func (*Catalog) CreateTable 

func (c *Catalog) CreateTable(_ context.Context, name string, schema parquet.Schema, partitionKeys []string) error

CreateTable creates a new table with the given schema and partition keys.

func (*Catalog) DropAlert 

func (c *Catalog) DropAlert(_ context.Context, name string) error

DropAlert removes the alert entry; no error if missing.

func (*Catalog) DropTable 

func (c *Catalog) DropTable(ctx context.Context, name string) error

DropTable removes a table from the catalog.

Metadata only: the table's name and manifest KV keys go away here, which is what makes it immediately invisible to every NEW query (GetTable, GetManifest, and ListTables all answer from this same metadata, and #483 keys the manifest cache by KV revision so a stale in-process copy can't serve a resurrected name's old files either). The table's DATA FILES are deliberately NOT deleted here — see FlushDroppedTableFiles for why, when, and under what guard they go.

Tombstone-then-grace-delete, not a prefix delete under tables/<name>/, and not "leave it forever" either (#494 asked for a decision between those). A live prefix delete is the wrong shape regardless of timing: a CREATE TABLE of the same name during the grace window gets an entirely new, unrelated set of files at that same prefix (chunk/compacted names are per-file random, not derived from the table name), and a prefix delete run after the fact cannot tell that incarnation's files from the dropped one's — it would eat the new table's data. Recording the exact paths this incarnation OWNED (engine-written only — see the snapshot below), once, right here, and checking each one against every CURRENT manifest before ever deleting it (FlushDroppedTableFiles) has no such blast radius. It doesn't reach RGMetaKey/SketchesKey blobs under stats/<name>/ — those are named by table+column, not by a birthday-collision-prone short ID, so they sit outside #494's collision hazard; leaking them is a separate, lower- severity storage-hygiene gap.

Ordering matters twice. The metadata put that removes the name from meta.Tables goes FIRST — it is the write that constitutes the drop, and putting it first is what makes a failed DROP a clean no-op rather than a table that is listed but unreadable (see the comment at the put). And the pending-drop record is appended only AFTER that put succeeds: a failed DROP must leave the table exactly as recoverable as it was before the call — nothing scheduled for physical deletion — not half-gone with its files already timed for reclaim.

func (*Catalog) EnableDropReclaim added in v0.18.3 

func (c *Catalog) EnableDropReclaim()

EnableDropReclaim declares that something in this process will call FlushDroppedTableFiles, and is what allows DropTable to record anything at all.

Reclaim is opt-in (compaction.BackgroundConfig.ReclaimDroppedTables, default off) and a *Catalog is not unique per process — an embedded wadjet.DB and a standalone pgwire DB each hold their own. On a catalog nobody sweeps, a pending-drop list is pure cost: nothing will ever consume it, so every DROP would grow it until the cap started evicting. Recording only where a flusher exists makes the default configuration (reclaim off) cost exactly nothing, and makes "which catalogs reclaim" a structural fact rather than a comment.

Call it before the DROPs whose files should be reclaimed — compaction.NewBackgroundCompactor does, at construction, when ReclaimDroppedTables is set. Idempotent; there is no disable, since a flusher that stops running just leaves entries pending.

func (*Catalog) FlushDroppedTableFiles added in v0.18.3 

func (c *Catalog) FlushDroppedTableFiles(ctx context.Context, grace time.Duration) int

FlushDroppedTableFiles physically deletes the data files of tables DropTable removed at least grace ago (zero or negative flushes everything pending, for tests). Three independent safety layers stand between a pending path and the Delete call below; the first alone bounds the blast radius to bytes wadjet wrote, and either of the next two alone blocks the #494 review's reproduced data loss:

  1. Ownership (DropTable, upstream of this list at all): a path is only ever in pendingDrops if its FileEntry was EngineWritten — stamped by AddNewFiles and SwapFileForGC, never by the AddFiles registration path. Nothing an operator staged and merely registered can reach this function, whatever shape its path takes.
  2. The live-manifest guard, RE-OBSERVED per pending entry immediately before that entry's deletes (liveCatalogState, and only when something is actually DUE): a path referenced by ANY current table's manifest is never deleted, no matter how long its OLD incarnation has been gone. This is the load-bearing layer — it is what makes drop-then-re-register-the-same-files (#278's workflow) and Iceberg's RefreshTable (drop+recreate over the same warehouse files, every refresh) safe. Building the set ONCE up front and deleting against it was the review's second reproduced data loss: a re-registration landing after the set was built and before the Delete fired was invisible to it. Re-observation narrows that window from "the whole flush" to "one entry's delete batch"; it does not close it (see the residual note below).
  3. Defense in depth: a path is only ever a delete candidate if it falls under its OWN table's partition.TablePrefix(name) — "tables/<name>/..." — and only via this catalog's own configured store and bucket. This is a CONVENTION, not an impossibility: iceberg/reader.go's resolvePath strips the scheme AND the bucket off an absolute data-file URI, so a warehouse at s3://somebucket/tables/events/... resolves into exactly the guarded shape. It is a cheap second opinion on paths that are already owned, not the thing standing between an Iceberg warehouse and a delete — layer 0 is (everything Iceberg registers goes through AddFiles, so none of it is ever marked).

On top of those, this mirrors compaction.Compactor's own deleteFromStore/FlushDeferredDeletes recreated-object guard: a path whose object was modified after the drop was recorded is skipped, since something has legitimately written there since.

RESIDUAL, stated plainly: pendingDrops is in-process, and the re-observation is a read; nothing serializes it against a write. dropMu guards only pendingDrops itself, not the Head/Delete calls below, so this is NOT scoped to a DIFFERENT *Catalog instance — a second goroutine calling AddFiles on THIS SAME *Catalog while the delete loop is mid-entry is just as invisible, and was reproduced directly against one instance. The window is one pending entry's WHOLE delete batch (every Head+Delete pair over that entry's paths), not a single call. cmd/wadjet's standalone mode has no in-process AddFiles caller sharing a *Catalog with its BackgroundCompactor (its pgwire server opens a separate wadjet.DB), so this is unreachable through that binary today; an embedder calling db.Catalog().AddFiles beside its own BackgroundCompactor reaches it. Layer 0 — ownership — is the layer that does not depend on timing at all, which is why it, not this one, is what bounds the blast radius. See docs/adr/0020-drop-table-reclaim-is-opt-in.md.

Not called from within this package on any timer, and — unlike compaction's own deferred-delete flush — not called unconditionally by the production background sweep either: see compaction.BackgroundConfig.ReclaimDroppedTables (opt-in, default off). Not every process that can DROP a table runs that sweep against the same *Catalog (an embedded wadjet.DB and a standalone pgwire DB each hold their own), so leaving this off by default means "not reclaimed yet" rather than "reclaimed here but not there" is the honest default everywhere; a leaked object is an ops cleanup problem, where an incorrectly deleted one is data loss. Like the compactor's own pendingDeletes, this list is process-local — a crash before the grace elapses leaves the files in place rather than losing track of them destructively, the same trade compaction already makes. Returns the number of files deleted.

func (*Catalog) GCDeleteMarkers 

func (c *Catalog) GCDeleteMarkers(_ context.Context, tableName string, minAge time.Duration) (rewriteTargets map[string][]int64, orphanPaths []string, err error)

GCDeleteMarkers identifies delete markers older than minAge. Returns file paths that need a forced rewrite (marker aged, file still exists) and orphan paths (marker aged, file already gone — orphan markers are removed from the manifest). Rewrite markers are left in the manifest so ForceCompactFile can apply them during the file rewrite; SwapFileForGC removes only the applied markers atomically.

Zero-value CreatedAt markers (pre-existing before the GC feature) are backfilled with the current time so they become eligible for GC in the next cycle rather than being immortal.

func (*Catalog) GCSnapshots 

func (c *Catalog) GCSnapshots(ctx context.Context, opts SnapshotOptions, keep int, minAge time.Duration) error

GCSnapshots deletes snapshot timestamps that are older than minAge AND not in the `keep` newest. Never deletes the snapshot currently pointed at by the latest pointer.

func (*Catalog) GetAlert 

func (c *Catalog) GetAlert(_ context.Context, name string) (*AlertMeta, error)

GetAlert returns the AlertMeta for name; returns an error if missing.

func (*Catalog) GetManifest 

func (c *Catalog) GetManifest(_ context.Context, tableName string) (*PartitionManifest, error)

GetManifest returns the partition manifest for a table.

Freshness is decided by the manifest key's KV REVISION, on every call. The cache only ever skips re-decoding a revision this process already decoded; it is a decode memo, never a staleness window.

It used to be one, and that was #483. A 2-second wall-clock TTL, invalidated only by writes made through the same *Catalog value, is sound only while a process holds exactly one of them. Standalone holds three over the same KV — the coordinator's, the pgwire DB's, and a fresh one per worker pipeline task — and pgwire routes SELECT through the coordinator's catalog while INSERT/UPDATE/DELETE and DDL go through the DB's. Every write therefore invalidated a cache no reader was consulting, and reads answered from a manifest up to two seconds old. Statements issued back to back (a psql script, a SQLancer round, any client driving a session) all land inside that window: writes looked lost, and DROP TABLE + CREATE TABLE of the same name answered out of the previous incarnation's files — silently when the two schemas were encoding-compatible, and as a decode-time type refusal when they were not. A revision is the catalog's own notion of "which version is this", so validating against it cannot drift from what the catalog holds; a clock can.

The returned manifest is SHARED with every other caller holding this revision. Treat it as immutable — mutators inside this package take loadManifest instead.

func (*Catalog) GetRemoteManifest 

func (c *Catalog) GetRemoteManifest(clusterID, tableName string) (*PartitionManifest, error)

GetRemoteManifest reads the partition manifest from a remote cluster's catalog.

func (*Catalog) GetRemoteTable 

func (c *Catalog) GetRemoteTable(clusterID, tableName string) (*TableMeta, error)

GetRemoteTable reads table metadata from a remote cluster's catalog.

func (*Catalog) GetTable 

func (c *Catalog) GetTable(_ context.Context, name string) (*TableMeta, error)

GetTable returns the metadata for a table.

func (*Catalog) Init 

func (c *Catalog) Init(ctx context.Context) error

Init initializes the catalog. Creates the S3 bucket and seed metadata if needed.

func (*Catalog) IsKVEmpty 

func (c *Catalog) IsKVEmpty(_ context.Context) (bool, error)

IsKVEmpty returns true if no <clusterID>.meta key exists. Used by the coordinator to decide whether startup should restore from a snapshot.

func (*Catalog) KV 

func (c *Catalog) KV() MetaKV

KV returns the underlying MetaKV store.

func (*Catalog) ListAlerts 

func (c *Catalog) ListAlerts(_ context.Context) ([]AlertMeta, error)

ListAlerts returns all alert entries, sorted by name.

func (*Catalog) ListClusters 

func (c *Catalog) ListClusters() ([]RemoteClusterInfo, error)

ListClusters discovers all clusters that have registered in the shared KV. Returns cluster IDs and their table lists.

func (*Catalog) ListTables 

func (c *Catalog) ListTables(_ context.Context) ([]string, error)

ListTables returns the names of all tables in the local catalog.

func (*Catalog) LoadUDFs 

func (c *Catalog) LoadUDFs() ([]UDFDef, error)

LoadUDFs reads persisted UDF definitions from the catalog KV. Returns nil (not error) if no UDFs have been saved.

func (*Catalog) PendingDropCount added in v0.18.3 

func (c *Catalog) PendingDropCount() int

PendingDropCount reports how many dropped-table entries are currently queued in pendingDrops, awaiting FlushDroppedTableFiles. Exported so a regression test outside this package (internal/iceberg's #494 repros) can pin layer 0 — ownership marking — directly: zero here means nothing was ever scheduled, which a test that only checks what a later flush deletes cannot distinguish from "scheduled, then caught by a later guard".

func (*Catalog) PutTableRGMeta 

func (c *Catalog) PutTableRGMeta(ctx context.Context, table string, files []FileRGMeta) (string, error)

PutTableRGMeta uploads the table's RG-metadata blob and returns its object-store key. Empty input returns "" (nothing uploaded).

func (*Catalog) ReadFile 

func (c *Catalog) ReadFile(ctx context.Context, key string) (io.ReadCloser, objstore.ObjectInfo, error)

ReadFile reads a file from the catalog's bucket. Convenience helper.

func (*Catalog) RemoveFiles 

func (c *Catalog) RemoveFiles(_ context.Context, tableName string, filePaths []string) error

RemoveFiles removes data files and their delete markers from the manifest. Used after compaction to clean up rewritten files.

func (*Catalog) Restore 

func (c *Catalog) Restore(ctx context.Context, opts RestoreOptions) (string, error)

Restore reads a snapshot from S3 and populates every KV key it contains. Returns the timestamp restored, or "" if there was nothing to restore (latest pointer absent).

Restore does NOT check whether the KV is already populated — that is the caller's responsibility (see coordinator's startup hook).

func (*Catalog) SaveUDFs 

func (c *Catalog) SaveUDFs(defs []UDFDef) error

SaveUDFs persists user-defined function definitions to the catalog KV.

func (*Catalog) SetAlertEnabled 

func (c *Catalog) SetAlertEnabled(ctx context.Context, name string, enabled bool) error

SetAlertEnabled toggles the enabled flag via CAS. Retries on revision mismatch.

func (*Catalog) Snapshot 

func (c *Catalog) Snapshot(ctx context.Context, opts SnapshotOptions) (string, error)

Snapshot writes every <clusterID>.* key to <prefix>/snapshots/<ts>/ and atomically updates <prefix>/latest. Returns the timestamp written.

func (*Catalog) Store 

func (c *Catalog) Store() objstore.Store

Store returns the underlying object store (for data file access).

func (*Catalog) SwapFileForGC 

func (c *Catalog) SwapFileForGC(_ context.Context, tableName string, oldPath string, newFile *FileEntry, partValues map[string]string, partPath string, appliedIndices map[int64]bool) error

SwapFileForGC atomically replaces an old file with a rewritten file in the manifest. In a single CAS operation it: (1) removes the old file from the partition, (2) adds the new file entry, and (3) removes only the specific delete marker row indices that were applied during the rewrite. Any row indices added concurrently (by a DELETE after GC started) are preserved.

NOTE: Surviving concurrent markers still reference the old file path after the swap. These become dangling markers since the old file no longer exists. This is by design — the next GC sweep detects them as orphans and cleans them up. The deleted rows they reference will be visible in query results for at most one GC cycle (~5 min default). Remapping marker paths and row indices inside the CAS loop was rejected due to complexity and increased CAS conflict surface (see security review, 2026-04-05).

If newFile is nil, the old file is simply removed (all rows were deleted).

func (*Catalog) TableRGMeta 

func (c *Catalog) TableRGMeta(ctx context.Context, tableName string) (map[string][]parquet.RowGroupStats, error)

TableRGMeta returns the table's persisted row-group metadata as a by-path map, or nil when the table has no blob (never analyzed). Best-effort: fetch/decode failures return nil, nil so scans degrade to per-file footer reads instead of failing.

The decoded blob is memoized per table, keyed by the manifest's KV revision — the same invalidation contract as AggregateColumnStats. In the 22-query benchmark process the blob is fetched from the store once per table, not once per query.

func (*Catalog) TouchAlertEvaluated 

func (c *Catalog) TouchAlertEvaluated(ctx context.Context, name string, at time.Time) error

TouchAlertEvaluated updates LastEvaluatedAt; retries on CAS conflict. Failure to update is non-fatal for the scheduler; callers log and move on.

func (*Catalog) UploadFileSketches 

func (c *Catalog) UploadFileSketches(ctx context.Context, table, parquetPath string, entries []FileSketchesEntry) (string, error)

UploadFileSketches is the exported entry point for the ingest path. It bundles per-column sketches into a single object-store blob and returns the canonical key.

type CatalogMeta 

type CatalogMeta struct {
	Version   int       `json:"version"`
	Tables    []string  `json:"tables"`
	CreatedAt time.Time `json:"created_at"`
	UpdatedAt time.Time `json:"updated_at"`
}

CatalogMeta is the top-level catalog metadata.

type DeleteMarker 

type DeleteMarker struct {
	FilePath   string    `json:"file_path"`   // path of the data file containing deleted rows
	RowIndices []int64   `json:"row_indices"` // 0-based row indices to skip
	CreatedAt  time.Time `json:"created_at"`  // when this marker was created
}

DeleteMarker records rows to skip during scan (merge-on-read). Each marker identifies deleted rows within a specific data file.

type FileColumnStats 

type FileColumnStats struct {
	MinValue  any   `json:"min_value,omitempty"`
	MaxValue  any   `json:"max_value,omitempty"`
	NullCount int64 `json:"null_count"`

	// HLL is a HyperLogLog sketch over the column's distinct values in this
	// file. Persisted as 1 byte version + 16384 register bytes (~16 KB).
	// Empty if HLL was not collected at write time (e.g., legacy files,
	// or for columns where NDV isn't useful — strings of comments). Read
	// at plan time via AggregateColumnStats which merges sketches across
	// files to estimate table-level NDV.
	HLL []byte `json:"hll,omitempty"`

	// Sample is a reservoir-sampled snapshot of the column's values in
	// this file, encoded by EncodeSample (typically ~256 values, ~2 KB).
	// AggregateColumnStats merges samples across files (weighted by
	// file row count) and builds an equi-depth Histogram at query time.
	// Used by stats.estimatePredSelectivity for range/equality filters.
	Sample []byte `json:"sample,omitempty"`
}

FileColumnStats contains per-column min/max/null statistics for a single file. Extracted from Parquet row group metadata at write time.

type FileEntry 

type FileEntry struct {
	Path        string                     `json:"path"`
	SizeBytes   int64                      `json:"size_bytes"`
	NumRows     int64                      `json:"num_rows"`
	CreatedAt   time.Time                  `json:"created_at"`
	ColumnStats map[string]FileColumnStats `json:"column_stats,omitempty"`
	// SketchesKey points to a bundled per-column HLL+Sample blob in the
	// object store (see sketches.go). Externalizing the sketches keeps
	// the manifest small enough to fit in the NATS KV per-message
	// payload cap (~1 MB) — without it, SF100 lineitem manifests
	// (63 files × 16 cols × ~18 KB per sketch) blew past the limit and
	// ANALYZE failed to persist the manifest. Empty string when no
	// sketches are externalized (legacy inline path still supported via
	// FileColumnStats.HLL / .Sample).
	SketchesKey string `json:"sketches_key,omitempty"`
	// EngineWritten marks an object WADJET ITSELF wrote: ingest's
	// chunk_<uuid>, compaction's compacted_<uuid>, delete-marker GC's
	// rewrite_<uuid>. It is the ownership marker DropTable's physical
	// reclaim keys off — only a marked entry is ever scheduled for
	// deletion (#494), so reclaim can only ever delete bytes this engine
	// created.
	//
	// Set in exactly two places, both of which mint the path themselves:
	// AddNewFiles (ingest, compaction) and SwapFileForGC's rewrite output.
	// AddFiles — the REGISTRATION path — deliberately leaves it alone,
	// because its callers point the catalog at objects somebody else
	// staged: cmd/tpch-bench (--data-prefix "tables/"), cmd/clickbench-
	// bench (--s3-prefix "tables/hits/"), internal/harness's s3_catalog,
	// and iceberg.CatalogIntegration all register pre-existing operator
	// data, and a bench bucket's reference dataset is not wadjet's to
	// delete on a DROP.
	//
	// Absent means NOT owned, which is the safe default in both
	// directions that matter: `omitempty` keeps it out of every manifest
	// that has no engine-written files, and every manifest written before
	// this field existed decodes with it false — so no pre-existing
	// object can be reclaimed by a newer binary. Unmarked entries leak on
	// DROP by design; see docs/adr/0020-drop-table-reclaim-is-opt-in.md.
	EngineWritten bool `json:"engine_written,omitempty"`
}

FileEntry describes a single Parquet file within a partition.

type FileRGMeta 

type FileRGMeta struct {
	Path   string
	Groups []parquet.RowGroupStats
}

FileRGMeta is the per-file unit of the table RG-metadata blob.

type FileSketchesEntry 

type FileSketchesEntry struct {
	Column string
	HLL    []byte
	Sample []byte
}

FileSketchesEntry is the in-memory form of one column's sketches.

func DecodeFileSketches 

func DecodeFileSketches(r io.Reader) ([]FileSketchesEntry, error)

DecodeFileSketches parses a v1 sketches blob.

type HLL 

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

HLL is a fixed-size HyperLogLog++ sketch. Zero value is a valid empty sketch; use Add to insert hashed values.

func DecodeHLL 

func DecodeHLL(r io.Reader) (*HLL, error)

DecodeHLL reads a sketch in version-1 format.

func HLLFromBytes 

func HLLFromBytes(b []byte) *HLL

HLLFromBytes parses a sketch from its serialized form. Returns nil on any error (corrupt bytes, wrong version) — callers fall back to the heuristic NDV path.

func (*HLL) Add 

func (h *HLL) Add(hash uint64)

Add inserts a value's hash into the sketch.

func (*HLL) AddBytes 

func (h *HLL) AddBytes(b []byte)

AddBytes hashes b and inserts it.

func (*HLL) AddInt64 

func (h *HLL) AddInt64(v int64)

AddInt64 hashes a uint64 representation of v and inserts it.

func (*HLL) Bytes 

func (h *HLL) Bytes() []byte

HLLBytes returns the serialized form, suitable for catalog persistence.

func (*HLL) Encode 

func (h *HLL) Encode(w io.Writer) error

Encode writes the sketch to w in version-1 format.

func (*HLL) Estimate 

func (h *HLL) Estimate() int64

Estimate returns the approximate cardinality.

func (*HLL) Merge 

func (h *HLL) Merge(o *HLL)

Merge takes the byte-wise max of the two sketches. Equivalent to computing a sketch over the union of inserted values.

type Histogram 

type Histogram struct {
	TypeCode    uint8
	TotalValues int64
	// Boundaries has K+1 entries; bucket[i] covers [Boundaries[i], Boundaries[i+1]).
	// The last bucket's upper bound is inclusive of the max value.
	Boundaries []any
	Counts     []int64
}

Histogram is the catalog's per-column histogram.

func BuildHistogramFromSamples 

func BuildHistogramFromSamples(sample []any, k int) *Histogram

BuildHistogramFromSamples constructs an equi-depth histogram from a pre-collected sample of values. Picks K bucket boundaries at sorted 1/K positions of the sample, assigns each sample to a bucket. K caps at 255 to fit the bucket count in one wire byte.

Caller is responsible for typing — pass a []int64, []float64, or [][]byte. Mixed types in the sample slice are rejected (returns nil).

func DecodeHistogram 

func DecodeHistogram(r io.Reader) (*Histogram, error)

DecodeHistogram reads a version-1 histogram from r.

func HistogramFromBytes 

func HistogramFromBytes(b []byte) *Histogram

HistogramFromBytes parses a histogram from its serialized form. Returns nil on any error (corrupt bytes, wrong version).

func HistogramFromMergedSample 

func HistogramFromMergedSample(values []any, totalRows int64, typeCode uint8, k int) *Histogram

HistogramFromMergedSample builds a histogram from the merged sample and the actual total row count. The total drives the histogram's TotalValues so selectivities are reported as fractions of the real table size, not of the sample size.

func (*Histogram) Bytes 

func (h *Histogram) Bytes() []byte

HistogramBytes returns the serialized form for catalog persistence.

func (*Histogram) Encode 

func (h *Histogram) Encode(w io.Writer) error

Encode writes the histogram to w in version-1 format.

func (*Histogram) SelectivityEQ 

func (h *Histogram) SelectivityEQ(v any) float64

SelectivityEQ returns the estimated fraction of values exactly equal to v. Uniform-distribution assumption within the containing bucket: 1 / (count in that bucket / count of distinct values in bucket). Without per-bucket NDV, falls back to 1 / TotalValues.

func (*Histogram) SelectivityLE 

func (h *Histogram) SelectivityLE(v any) float64

SelectivityLE returns the estimated fraction of values ≤ v.

Cumulates the counts of all buckets fully below v, plus a linear- interpolation fraction of the bucket containing v. For string columns, the partial-bucket fraction defaults to 0.5 (uniform).

Returns 0 (nothing matches) when v < all values, 1 (all match) when v ≥ all values. Clamped to [0, 1].

func (*Histogram) SelectivityLT 

func (h *Histogram) SelectivityLT(v any) float64

SelectivityLT returns the estimated fraction of values < v.

func (*Histogram) SelectivityRange 

func (h *Histogram) SelectivityRange(lo, hi any) float64

SelectivityRange returns the estimated fraction of values in [lo, hi]. Inclusive on both ends.

type Lock 

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

Lock represents a held distributed lock.

func (*Lock) Release 

func (l *Lock) Release() error

Release releases the lock.

type LockManager 

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

LockManager provides distributed read-write locks via NATS KV.

func NewLockManager 

func NewLockManager(js jetstream.JetStream) (*LockManager, error)

NewLockManager creates a lock manager backed by a NATS KV bucket.

func (*LockManager) AcquireReadLock 

func (lm *LockManager) AcquireReadLock(ctx context.Context, space, table, readerID string) (*Lock, error)

AcquireReadLock acquires a shared read lock on a table. Multiple readers can hold locks concurrently. Each reader gets a unique key.

func (*LockManager) AcquireWriteLock 

func (lm *LockManager) AcquireWriteLock(ctx context.Context, space, table string) (*Lock, error)

AcquireWriteLock acquires an exclusive write lock on a table. Blocks until the lock is acquired or the context is cancelled.

func (*LockManager) HasWriteLock 

func (lm *LockManager) HasWriteLock(ctx context.Context, space, table string) (bool, error)

HasWriteLock checks if a write lock is currently held on a table.

type MemKV 

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

MemKV is an in-memory MetaKV implementation for tests and embedded use.

func NewMemKV 

func NewMemKV() *MemKV

NewMemKV creates a new in-memory KV store.

func (*MemKV) Delete 

func (m *MemKV) Delete(key string) error

func (*MemKV) Get 

func (m *MemKV) Get(key string) ([]byte, uint64, error)

func (*MemKV) List 

func (m *MemKV) List(prefix string) ([]string, error)

func (*MemKV) Put 

func (m *MemKV) Put(key string, value []byte) (uint64, error)

func (*MemKV) Revision added in v0.18.1 

func (m *MemKV) Revision(key string) (uint64, error)

Revision implements RevisionReader: the key's revision without copying its value, so a catalog cache can revalidate for the price of a map lookup.

func (*MemKV) Update 

func (m *MemKV) Update(key string, value []byte, expectedRev uint64) (uint64, error)

type MetaKV 

type MetaKV interface {
	// Get returns the value and revision for a key.
	// Returns ErrKeyNotFound if the key does not exist.
	Get(key string) (value []byte, revision uint64, err error)

	// Put creates or updates a key, returning the new revision.
	Put(key string, value []byte) (revision uint64, err error)

	// Update performs a compare-and-swap: writes value only if the key's
	// current revision matches expectedRev. Returns ErrRevisionMismatch
	// if a concurrent write changed the key since it was read.
	Update(key string, value []byte, expectedRev uint64) (revision uint64, err error)

	// Delete removes a key. No error if the key does not exist.
	Delete(key string) error

	// List returns all keys matching the given prefix.
	// An empty prefix returns all keys.
	List(prefix string) ([]string, error)
}

MetaKV abstracts key-value storage for catalog metadata. Production uses NATSKVAdapter; tests/embedded use MemKV.

type NATSKVAdapter 

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

NATSKVAdapter wraps a NATS JetStream KeyValue as a MetaKV.

func NewNATSKV 

func NewNATSKV(js jetstream.JetStream) (*NATSKVAdapter, error)

NewNATSKV creates a NATS KV-backed MetaKV. Creates or opens the wadjet_catalog KV bucket.

func (*NATSKVAdapter) Delete 

func (n *NATSKVAdapter) Delete(key string) error

func (*NATSKVAdapter) Get 

func (n *NATSKVAdapter) Get(key string) ([]byte, uint64, error)

func (*NATSKVAdapter) List 

func (n *NATSKVAdapter) List(prefix string) ([]string, error)

func (*NATSKVAdapter) Put 

func (n *NATSKVAdapter) Put(key string, value []byte) (uint64, error)

func (*NATSKVAdapter) Update 

func (n *NATSKVAdapter) Update(key string, value []byte, expectedRev uint64) (uint64, error)

type PartitionEntry 

type PartitionEntry struct {
	Path   string            `json:"path"`
	Values map[string]string `json:"values"`
	Files  []FileEntry       `json:"files"`
}

PartitionEntry describes a single partition.

type PartitionManifest 

type PartitionManifest struct {
	Table         string           `json:"table"`
	Partitions    []PartitionEntry `json:"partitions"`
	DeleteMarkers []DeleteMarker   `json:"delete_markers,omitempty"` // merge-on-read deletes
	UpdatedAt     time.Time        `json:"updated_at"`
	// RGMetaKey points to the table's row-group-metadata blob in the
	// object store (see rgmeta.go), written by AnalyzeTable. Scans use
	// it to enumerate and prune row groups without reading any parquet
	// footers. Files added after the blob was written simply aren't in
	// it and fall back to footer reads — the key stays valid across
	// ingest. Empty until the table is first analyzed.
	RGMetaKey string `json:"rg_meta_key,omitempty"`
}

PartitionManifest tracks all partitions and their files for a table.

type RemoteClusterInfo 

type RemoteClusterInfo struct {
	ClusterID string
	Tables    []string
}

RemoteClusterInfo describes a remote cluster's catalog.

type ReservoirSampler 

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

ReservoirSampler holds an in-memory sample using Algorithm L reservoir sampling: O(N) producer cost, uniformly distributed sample of size K from a stream of unknown length. Each Add either replaces a random existing entry or is dropped, weighted to keep all input values equally probable.

func NewReservoirSampler 

func NewReservoirSampler(k int) *ReservoirSampler

NewReservoirSampler creates a sampler of the given capacity. K ≤ 0 uses the default (SampleDefaultSize).

func (*ReservoirSampler) Add 

func (rs *ReservoirSampler) Add(v any)

Add inserts a value. The sample is updated in O(1) amortized.

func (*ReservoirSampler) Snapshot 

func (rs *ReservoirSampler) Snapshot() (sorted []any, totalSeen int64, typeCode uint8)

Snapshot returns the sampled values along with the total observed count. The sample slice is sorted in-place.

type RestoreOptions 

type RestoreOptions struct {
	SnapshotOptions
	// ForceTS, when non-empty, overrides the latest pointer. Use "latest"
	// as a sentinel to mean "read the pointer" (equivalent to empty).
	ForceTS string
}

RestoreOptions configures where to read a snapshot from.

type RevisionReader added in v0.18.1 

type RevisionReader interface {
	// Revision returns the key's current revision, or ErrKeyNotFound.
	Revision(key string) (uint64, error)
}

RevisionReader is an OPTIONAL MetaKV capability: report a key's current revision without transferring its value.

Catalog validates every cached manifest against the KV revision on every read (a wall-clock TTL is not a correctness mechanism — see Catalog.GetManifest), so the validation happens on the hottest planner path there is. A store that can answer "what revision is this key at" without shipping back a manifest that is megabytes of JSON at SF100 turns that validation into an O(1) probe. A store that cannot (NATS KV has no value-free get) simply omits the method and pays the full read, which is what the pre-cache code did anyway.

type SnapshotKeyEntry 

type SnapshotKeyEntry struct {
	KVKey  string `json:"kv_key"`  // e.g. "local.table.orders"
	S3Path string `json:"s3_path"` // relative to <prefix>/snapshots/<ts>/, e.g. "table/orders.json"
	SHA256 string `json:"sha256"`  // hex-encoded
}

SnapshotKeyEntry records one KV key's location and integrity hash.

type SnapshotManifest 

type SnapshotManifest struct {
	Version   int                `json:"version"`
	Timestamp string             `json:"timestamp"`
	ClusterID string             `json:"cluster_id"`
	KeyCount  int                `json:"key_count"`
	Keys      []SnapshotKeyEntry `json:"keys"`
}

SnapshotManifest is the JSON body of <ts>/manifest.json. Lists every KV key included in the snapshot and its SHA256 for integrity.

type SnapshotOptions 

type SnapshotOptions struct {
	Store  objstore.Store
	Bucket string
	Prefix string // path within bucket, e.g. "wadjet/catalog/". Must end in "/".
}

SnapshotOptions configures where catalog snapshots are written.

type TableColumnStats 

type TableColumnStats struct {
	MinValue  any
	MaxValue  any
	NullCount int64
	TotalRows int64
	// NDV is the merged HLL estimate of distinct values across all files,
	// or 0 when no file had an HLL sketch. When >0 it's preferred over the
	// min/max-range heuristic in the optimizer's NDV estimator.
	NDV int64
	// Histogram is the equi-depth histogram built from merging per-file
	// reservoir samples, scaled to TotalRows. Nil when no file had a
	// Sample. Used by stats.estimatePredSelectivity for range/equality
	// filters — replaces the hardcoded 0.33/0.1 fractions with
	// data-driven selectivity.
	Histogram *Histogram
}

TableColumnStats holds aggregated per-column statistics across all files. Used by the optimizer for selectivity estimation.

type TableMeta 

type TableMeta struct {
	Name          string         `json:"name"`
	Schema        parquet.Schema `json:"schema"`
	PartitionKeys []string       `json:"partition_keys"`
	CreatedAt     time.Time      `json:"created_at"`
	UpdatedAt     time.Time      `json:"updated_at"`
	Version       int            `json:"version"`
}

TableMeta contains metadata for a single table.

type UDFDef 

type UDFDef struct {
	Name   string   `json:"name"`
	Params []string `json:"params"`
	Body   string   `json:"body"`
	Owner  string   `json:"owner"`
	Locked bool     `json:"locked"`
}

UDFDef mirrors expr.UDFDef for persistence without import cycles.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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