catalog

package
v0.18.49 Latest Latest
Warning

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

 Go to latest Published: Sep 5, 2026 License: AGPL-3.0 Imports: 26 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 MaxNameBytes = 63

MaxNameBytes is PostgreSQL's own effective identifier length — NAMEDATALEN - 1, measured: a longer name is truncated to exactly this many bytes there. Holding wadjet to the same number is what makes "a name this engine ACCEPTS behaves the way PostgreSQL's does" true rather than nearly true: at or below it the two agree byte for byte, and above it PostgreSQL's own answer is already lossy.

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 (
	// ErrCompactionInputMoved: one of the files the output replaces is no
	// longer in the partition — another writer has already consumed it.
	ErrCompactionInputMoved = fmt.Errorf("%w: an input file is no longer in the partition", ErrCompactionConflict)

	// ErrCompactionDeletesAdvanced: the delete markers on an input are not
	// the ones the output was cut against. Publishing would republish a row
	// a committed DELETE removed, or drop a row nobody deleted.
	ErrCompactionDeletesAdvanced = fmt.Errorf("%w: an input's delete markers changed after the output was written", ErrCompactionConflict)
)
View Source 
var ErrCompactionConflict = errors.New("this compaction output was cut from a snapshot the table no longer has")

ErrCompactionConflict reports that a compaction output cannot be published because the snapshot it was cut from is no longer the table's state.

It is the compaction half of the rule `ErrDMLTargetMoved` states for DML (ADR-0030), and it exists for the same reason: a writer that reads a manifest, spends time producing a replacement, and then commits, is running a transaction whose read set has to be validated at commit time or not at all. Compaction's read set is two things — WHICH FILES it consumed and WHICH ROWS OF THEM were already deleted — and until #893/#894/#895 neither was checked:

  • `RemoveFiles` treated an input that was already gone as success, so two compactors could each publish a replacement for the same originals and the table ended up holding both copies of every row (#895).
  • a DELETE that committed after the output was written but before it was published was undone by the publication, because the output still carried the row and `RemoveFiles` stripped the marker that named it (#894).

A conflict is not a failure: nothing was written, the previous snapshot is intact, and the losing writer replans from the manifest that replaced the one it read. It is detected BEFORE the CAS write is attempted, which is why the caller may safely delete the output object it uploaded — a publication ERROR (the KV refused, timed out, or is unreachable) says nothing about whether the write landed, and the bytes are kept in that case.

View Source 
var ErrDMLRowSuperseded = errors.New("a row this statement supersedes was already superseded by another statement")

ErrDMLRowSuperseded reports that a DML statement's manifest change cannot be committed because ANOTHER STATEMENT has already superseded a row this one is about to supersede.

It is the row-level half of the same rule ErrDMLTargetMoved states over files, and it is what #691 left open — ADR-0030 said so in its own words: "Two writers racing each other … both succeed, and the second one's markers are valid because the files did not move … Closing it needs a conflict rule over ROWS, which this record does not decide." This is that rule.

The window is the ordinary one: each statement reads the manifest, scans the files it names, records WHICH ROW OF WHICH FILE it affected, and commits at the end. Two statements over the same row both see it live, both write a replacement, and both mark the copy they read — so the manifest ends up naming BOTH replacements and the key is present twice. Measured on v0.18.22, `UPDATE … n = 111 WHERE id = 1` against `UPDATE … n = 222 WHERE id = 1`: 

table afterwards:  1:111:a  1:222:a  2:20:b  3:30:c
both statements:   UPDATE 1

The same window resurrects a deleted row (an UPDATE whose scan predates a concurrent DELETE re-publishes the row it read) and reports `DELETE 1` over a row that is still readable.

A statement that sees this redoes itself against the manifest that replaced the one it read, exactly as ErrDMLTargetMoved makes it redo; the outcome is then one of the two serial orders PostgreSQL could have produced.

View Source 
var ErrDMLTargetMoved = errors.New("the files this statement read are no longer in the table's manifest")

ErrDMLTargetMoved reports that a DML statement's manifest change cannot be committed because the files it read are no longer the files the table has.

It is the catalog's half of #691. A DELETE/UPDATE/MERGE reads a manifest, scans the files it names, and records which ROW OF WHICH FILE it affected. Between that read and the commit, compaction can rewrite those files: mergeGroup calls RemoveFiles (which strips the markers for the paths it removes) and then AddNewFiles, so the statement's markers arrive naming files the table no longer has. AddDeleteMarkers used to accept them — `dm.FilePath` was only ever a map key there — and the manifest gained a marker pointing at nothing: 

DELETE FROM u WHERE id = 1   →  "DELETE 1", and row 1 is still there
UPDATE u SET n = 99 …        →  "UPDATE 1", and the table holds 1:10 AND 1:99

both reported as success. Reproduced deterministically on all three doors.

The statement retries when it sees this; a statement that has exhausted its retries reports it, and the DML layer gives it PostgreSQL's 40001 (serialization_failure) — the class a client is expected to retry.

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 ErrPathRetiring = errors.New("this path is being retired by a cleanup sweep: retry the registration")

ErrPathRetiring reports that a file path cannot be registered into a manifest right now because a retirement sweep is in the middle of deciding whether to physically delete the object at that path.

It is the second half of #896's fix, and it is the half that does not depend on ordering luck. The first half — "no live manifest references this object" — is a READ, and a read cannot exclude a registration that lands after it. Marking the candidate paths before that read, and refusing a registration that names a marked path until the sweep is done with it, closes the window from the other side: a registration either completes before the mark (and the sweep's read sees it, and preserves the bytes) or arrives after it (and is refused, loudly, with nothing written).

It is retryable and brief: the mark lives only for the duration of one retirement batch. Callers that register operator-staged objects — the harness loaders, `iceberg.CatalogIntegration` — should retry.

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 AmbiguousTableError added in v0.18.35 

func AmbiguousTableError(name string, candidates []string) error

AmbiguousTableError is the refusal BOTH doors raise for a reference that matches two or more registered tables case-insensitively and none of them byte-exact — the case ResolveTableName declines to answer.

One builder, because the two doors are one statement class to a client and the reader had the useful message while the WRITER got the bare "does not exist": `wadjet/dml.go` called ResolveTableName and never asked which tables the reference matched, so the candidate list the SELECT door prints was dropped on INSERT, UPDATE, DELETE and MERGE (#858).

Still 42P01, and the reason is in physical.validate's own comment: what is true is that no unique relation has this name, and PostgreSQL has no ambiguity class for relations at all — it cannot reach this state, because it folds at the catalog.

func CheckStorableName added in v0.18.46 

func CheckStorableName(kind, name string) error

CheckStorableName refuses a relation or column name that cannot be one component of an object key.

A table's data lives at `tables/<name>/…` (partition.TablePrefix) and a partition key's column name becomes a directory component below it (`<col>=<value>/`), so these two names are not only identifiers: they are spelled into the object store's namespace. The LEXER takes a delimited identifier byte-exact, so `CREATE TABLE "../../../tmp/x"` handed the store a key that climbed out of its root, and on `storage.type: file` that was an arbitrary file write (CodeQL go/path-injection #23/#24/#25).

objstore.ValidateObjectKey closes that at the store, which is the layer that has to be right regardless of who the caller is. This is the layer where a PERSON can be told what is wrong, at CREATE, before a table exists whose every write would fail.

The rule is objstore.ValidateObjectKey's, narrowed to ONE component: no '/', no '\', no NUL, the name is not "." or "..", and it does not begin with '.'. A ".." INSIDE a component ("x..y") is accepted, because it names a real directory and the store accepts the key — the danger is a component that IS "..", not the two characters. The SQLSTATE is PostgreSQL's 42602 invalid_name.

This is a DELIBERATE DIVERGENCE and it is recorded in ADR-0012's list: PostgreSQL accepts any of these inside a double-quoted identifier, because its relations are rows in pg_class and never filenames. Wadjet's are objects in a store, and the alternative to refusing the name is a table whose data has no home — or, on a filesystem store, one whose data lands somewhere it was never meant to. It is name-only and LOUD: no query answers differently, and no name is silently rewritten.

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.

It does NOT check that the files its markers name are still in the manifest, and it is not the entry point a DML statement uses. `CommitDML` is: it validates every marker against the manifest it is committing into, and lands the statement's new files in the same CAS (#691). This one stays as the low-level primitive for callers that mint markers against a manifest they are holding right now — the GC and compaction tests, and any embedder managing markers directly.

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(ctx 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) AggregateColumnStatsFrom added in v0.18.17 

func (c *Catalog) AggregateColumnStatsFrom(_ context.Context, tableName string, manifest *PartitionManifest, rev uint64) (map[string]TableColumnStats, error)

AggregateColumnStatsFrom is AggregateColumnStats over a manifest the caller ALREADY HOLDS, with the revision it came from.

The revision is not decorative: it keys the memo, and passing the pair is what makes the statistics and the manifest ONE consistent view. The internal fetch this replaces was the first statement of the body, so a caller that had just read the manifest read it again and could receive a different one — stats describing files the pinned manifest does not list (an AddFiles landed between the two reads) or omitting files it does (RemoveFiles, compaction). The direction is fixed, because annotateScanColumns reads the manifest first and the stats second, so the stats are always the newer half.

Today's two consumers are cost-model only, so a torn view is a worse plan rather than a wrong answer. That is a property of the current code and not an invariant: the natural next optimizer feature — proving a predicate unsatisfiable from ScanColStats.MinValue/MaxValue — turns it into dropped rows on the day it lands (#540).

func (*Catalog) AmbiguousTableNames added in v0.18.30 

func (c *Catalog) AmbiguousTableNames(name string) []string

AmbiguousTableNames returns the registered tables a reference matches case-insensitively when there are TWO OR MORE of them and none is a byte-exact match — the case ResolveTableName declines to answer. Nil otherwise, including for an ordinary miss.

The caller turns it into a refusal that names the candidates. Without it the refusal is the plain "does not exist", which is true (no unique relation has that name) but tells the user nothing about the two tables that do.

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) CommitCompaction added in v0.18.45 

func (c *Catalog) CommitCompaction(_ context.Context, cc CompactionCommit) error

CommitCompaction publishes a compaction replacement in ONE conditional manifest transaction: the inputs leave the partition, their delete markers leave with them, and the replacement arrives — or none of it does.

Before #893 this was two CAS writes, `RemoveFiles` then `AddNewFiles`. Each was atomic and the PAIR was not, which cost three distinct properties:

  1. A failure between them left the table with the inputs gone and the replacement unpublished — zero visible rows, unrecoverable by retry because the compactor selects its inputs from the manifest it just emptied (#893).
  2. Even when both succeeded, a reader landing between them saw the intermediate manifest and answered from it.
  3. Neither call validated anything: `RemoveFiles` accepts inputs that are already gone (#895) and strips markers it never applied (#894).

The validation is the other half of the fix and does not follow from atomicity: a single atomic write of a stale plan is still wrong. Two predicates, both exact rather than conservative:

  • **Input identity.** Every path in Inputs is still in PartPath's file list. A losing compactor whose originals another compactor already consumed is refused with ErrCompactionInputMoved instead of adding a second copy of the same rows beside the winner's.
  • **The delete-marker snapshot.** The manifest's marker set for each input equals the set the output applied. A DELETE that committed while the output was being written moves the set, and the commit is refused with ErrCompactionDeletesAdvanced rather than republishing the row it removed.

Neither predicate fires on a write that did not touch this partition's files, so unrelated ingest, DML on other files, and compaction of other partitions all commit alongside it.

func (*Catalog) CommitDML added in v0.18.20 

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

CommitDML commits one DML statement's whole manifest change in a SINGLE CAS: the files it wrote and the delete markers that remove the rows they replace, or neither.

Two properties, and both are load-bearing:

  1. **Validation, over files and over rows.** Every marker names a file the manifest STILL HOLDS at commit time (`ErrDMLTargetMoved`), and no marker names a (file, row) the manifest ALREADY MARKS (`ErrDMLRowSuperseded`). The first is the check `AddDeleteMarkers` never had — it decodes the manifest and never looks at `Partitions`. The second is the one #691 left open, and it rests on an invariant the DML door keeps: a statement filters its scan through `DeletedRowsByFile` before it matches anything (`deleteOnce`, `updateOnce` and `readMergeTarget` all do, which is #674's rule), so it NEVER mints a marker for a row the manifest it read already marked. An incoming (file, row) that is marked here was therefore marked by another statement SINCE this one read, and that is exactly the conflict.

    Both predicates are exactly right rather than merely conservative. A concurrent write that did not touch this statement's files leaves its markers valid; two statements over DIFFERENT rows of the same file both commit, because their marker sets are disjoint. A blunt "the revision moved" test would be wrong in both directions — it fails on any unrelated write, and an UPDATE's own ingest moves the revision.

  2. **Atomicity.** An UPDATE or MERGE used to commit twice — the ingester's AddNewFiles per flushed file, then AddDeleteMarkers — so a refusal at the second commit left the replacement rows beside the originals they were supposed to replace. Here the replacement files ride in the same CAS as the markers, so a refused statement has written nothing to the manifest and can simply be retried.

What remains outside it: the parquet objects a refused attempt already wrote stay in the object store, unreferenced, until the orphan sweep reclaims them. That is a leak of bytes on a rare retry, never a wrong row — the manifest is the only thing that decides which rows exist.

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) GetManifestWithRevision added in v0.18.17 

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

GetManifestWithRevision is GetManifest with the KV revision the manifest came from, so a caller that must pin a CONSISTENT view of a table can hand both halves to AggregateColumnStatsFrom instead of letting it read the manifest a second time.

Without it, a statement that read the manifest and then asked for column statistics got TWO reads of the same key, and — because a writer can commit between them — a stats map describing rows the pinned manifest does not contain. Measured with a NATS-equivalent KV: a pinned 2-file manifest of 200 rows alongside stats reporting TotalRows=300, and the tear then pinned for the whole statement (#540).

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, unconditionally: it does not check that the paths are still there, and it strips EVERY marker naming them whether or not anything applied them.

It is no longer compaction's publication path, and both of those properties are why. Compaction commits through CommitCompaction, which validates the inputs (#895) and the marker snapshot (#894) and does the removal, the addition and the marker change in one write (#893). This stays as the low-level primitive for a caller that already knows what it is removing — the type-matrix gate, the ANALYZE and torn-view tests — the same role AddDeleteMarkers keeps beside CommitDML (ADR-0030).

func (*Catalog) ResolveTableName added in v0.18.30 

func (c *Catalog) ResolveTableName(name string) string

ResolveTableName is the READ-side spelling of a table name: the catalog's own, for a reference that named it in a different case.

It is `batch.ResolveColumnIndex`'s rule one level up (ADR-0012): byte-exact first, and only on a miss a UNIQUE ASCII-case-insensitive match among the registered tables. Two matches resolve to NOTHING and the caller reports the miss, exactly as two columns do — picking one would be a silent wrong table.

It exists because an unquoted reference FOLDS at the lexer (#731) while wadjet's table names come from parquet and ingest, where a user-chosen mixed-case name is ordinary. Without it `FROM MyTab` is 42P01 against a table this engine itself created, which is PostgreSQL's rule but breaks every catalog written before the fold.

READ ONLY. Every write door — CreateTable, the DML paths, DropTable — keys byte-exact, because creating or writing `MyTab` when `mytab` exists must land on the name the caller wrote and never on its case-twin.

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) RetireObjects added in v0.18.45 

func (c *Catalog) RetireObjects(ctx context.Context, reqs []RetireRequest) map[string]RetireOutcome

RetireObjects physically deletes objects that NO live manifest in this catalog references, and preserves the bytes of every object it cannot prove that about.

It is the one place an object is allowed to leave the bucket on a compaction schedule, and it exists because "this table stopped referencing the file" is not the same claim as "nothing references the file". #896 is the difference: compaction removed a source from `events`'s manifest and queued its bytes; a still-live `archive` registered the very same object through AddFiles during the grace; the queue's only guard was the object's LastModified, which registering unchanged bytes does not move. The queue deleted a file a live table's manifest still names.

Three things stand between a queued path and the Delete call, and the order they run in is the point:

  1. **The retirement mark, taken first.** Every candidate path is marked before anything is read. A registration naming a marked path is refused with ErrPathRetiring until the mark is released. A path with a registration already IN FLIGHT is not marked at all — it comes back RetireUnproven, and the caller tries again once that registration has landed and can be observed.
  2. **The live-manifest reference check**, over EVERY table in the catalog (`liveCatalogState`, shared with DROP reclaim). A path any current manifest names is RetireReferenced and is never deleted. Because the mark is already held, a registration that could invalidate this read cannot be running: it either finished before the mark (and this read sees it) or is refused.
  3. **The recreated-object guard**: an object written since the retirement was scheduled is not the object that was scheduled.

A catalog read that fails yields RetireUnproven for every path rather than a delete against an incomplete picture. Doubt preserves bytes.

The residual, stated plainly: the mark is IN-PROCESS. It excludes a registration through this same *Catalog — which is what #896 reproduced, and what an embedder running a BackgroundCompactor beside its own AddFiles calls reaches — and it does not exclude a DIFFERENT process registering the path into a shared catalog. Closing that needs a catalog-side lease, which the deferred-delete queue could not use anyway: the queue itself is process-local, so another process's compactor never sees these paths.

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(ctx context.Context, tableName string, oldPath string, newFile *FileEntry, partValues map[string]string, partPath string, appliedIndices map[int64]bool) error

SwapFileForGC publishes a delete-marker GC rewrite: the old file leaves the partition, the rewritten replacement arrives, and the markers the rewrite applied go away — all in the one conditional transaction CommitCompaction runs, validated against the same two preconditions every compaction publication is (input identity, and the delete-marker snapshot the output was cut from).

It used to be its own CAS loop with its own rule, and the rule was wrong in two directions #894 and #895 reproduced:

  • It appended the rewrite output without requiring that oldPath was still a member of the partition, so two GC sweeps over the same file each published a rewrite and the surviving rows appeared twice.
  • It removed only the row indices the rewrite APPLIED and left any that had arrived since, under the OLD file's path — where no reader can apply them, because that file is gone. The comment here used to say those rows stayed visible "for at most one GC cycle". They did not: the next sweep removes the dangling marker as an orphan, and the replacement carries the row forever. Removing a marker cannot remove a row from a file that already contains it.

So a rewrite now applies ALL of a file's current markers or none of them: if the marker set moved between the manifest read the rewrite was cut from and this commit, the swap is refused with ErrCompactionDeletesAdvanced and the caller re-reads and rewrites against the newer set. appliedIndices is therefore a PRECONDITION, not just a cleanup list — it says which markers the output reflects, and the commit checks it.

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 CompactionCommit added in v0.18.45 

type CompactionCommit struct {
	Table      string
	PartPath   string
	PartValues map[string]string

	// Inputs are the file paths the output replaces. Every one must still be
	// in PartPath's file list at commit time.
	Inputs []string

	// Output is the replacement file, or nil when every input row was
	// delete-filtered away and nothing was written. A nil Output is a
	// publication like any other — the inputs and their markers still go
	// away in the same CAS.
	Output *FileEntry

	// AppliedDeletes is the delete-marker snapshot the output was cut
	// against: per input path, the set of row indices the merge skipped. The
	// manifest's markers for those paths must be EXACTLY this at commit
	// time. A marker that arrived since names a row the output still
	// carries, and publishing would resurrect it; a marker that vanished
	// since means the output dropped a row the table still has.
	AppliedDeletes map[string]map[int64]bool
}

CompactionCommit is ONE compaction publication: the input files it consumes, the replacement it publishes, and the delete-marker snapshot the replacement was cut from.

Every field is part of the commit's precondition, not just of its effect. See Catalog.CommitCompaction.

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 PendingFile added in v0.18.20 

type PendingFile struct {
	PartValues map[string]string
	PartPath   string
	Entry      FileEntry
}

PendingFile is a data file already written to the object store but NOT yet in the manifest, waiting to be committed together with the delete markers that supersede what it replaces.

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 RetireOutcome added in v0.18.45 

type RetireOutcome int

RetireOutcome is what RetireObjects decided about one path. 

const (
	// Retired: the object was deleted. The caller drops its queue entry.
	Retired RetireOutcome = iota
	// RetireReferenced: a live manifest in this catalog references the
	// object, or its bytes were replaced since the retirement was
	// scheduled. It must never be deleted on this schedule; the caller
	// drops its queue entry, because the reference is not going to
	// disappear because we waited.
	RetireReferenced
	// RetireUnproven: eligibility could not be established — the catalog
	// could not be read, or a registration naming this path was in flight.
	// Nothing was deleted. The caller requeues and tries again later.
	RetireUnproven
)

func (RetireOutcome) String added in v0.18.45 

func (o RetireOutcome) String() string

type RetireRequest added in v0.18.45 

type RetireRequest struct {
	// Path is the object key, in this catalog's own bucket.
	Path string
	// NotModifiedAfter is the instant the retirement was scheduled. An
	// object written since then is not the object that was scheduled —
	// something recreated the path — and is preserved. Zero disables the
	// check.
	NotModifiedAfter time.Time
}

RetireRequest is one object proposed for physical retirement.

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.