Help us shape the future of Go! Take the 2026 Go Developer Survey

reconstruct

package
v0.116.0 Latest Latest
Warning

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

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

Documentation

Overview

Package reconstruct implements row state reconstruction by combining baseline Parquet snapshots with indexed binlog change events.

The core idea: a baseline Parquet file captures the full row state at a point in time (produced by "bintrail baseline"). Binlog events captured after that snapshot each carry a full row_after image. Applying them in order up to a target timestamp yields the exact row state at that moment.

Index

Constants

View Source 
const (
	// OutputFormatMydumper writes a mydumper-compatible SQL dump directory —
	// the original full-table output (#187), and what the zero value resolves to.
	OutputFormatMydumper = "mydumper"
	// OutputFormatParquet writes a baseline snapshot in Parquet (#1169): the same
	// artifact `bintrail baseline` produces from a mydumper dump, so the output of
	// a reconstruct can itself anchor the next one.
	OutputFormatParquet = "parquet"
)

FullTableConfig drives ReconstructTables — the full-table merge-on-read entry point for #187. One Config instance covers one run; each table in Tables is reconstructed concurrently with at most Parallelism goroutines running at a time (a goroutine is spawned for every table, but a buffered- channel semaphore caps the number that actually do work).

Supported primary-key types (#212 + #214):

  • integer: int, smallint, tinyint, mediumint, bigint (+ unsigned)
  • string: char, varchar, text, tinytext, mediumtext, longtext
  • enum, set
  • datetime, timestamp — canonicalized from DuckDB time.Time to the go-mysql string format the indexer stores
  • date — canonicalized to "2006-01-02"
  • year
  • decimal, numeric — pass-through string (go-mysql delivers a pre-formatted string when useDecimal is false, which is the bintrail default; baseline stores DECIMAL as parquet.String, so both sides agree byte-for-byte)
  • binary, varbinary, blob family — bytes; BINARY(n) padding trimmed (#1155)

PK columns with any other type (FLOAT, DOUBLE, TIME, BIT, JSON, spatial types) are rejected at ReconstructTable entry with a hard error; supportedPKType in pk_canonicalize.go is the authoritative list. Adding a type means verifying its round-trip between event.BuildPKValues and the baseline Parquet end to end, so each is deferred behind its own follow-up issue.

UPDATE events that mutate the primary key itself cannot be folded under one key: the change map is keyed by the before-image PK, so a changed PK would emit the after-image row under the OLD key and resurrect a row a later DELETE removed, duplicate one another event already targets, or (when its old key is reused) silently drop the moved row (#782). The paged fold (foldPage, every reconstruct and refresh path on a MySQL/MariaDB source) folds such an UPDATE as the two events it is, a DELETE of the old key and an INSERT of the new one at the UPDATE's own binary log position, and the window's "last change per key" rule resolves each shape above (#2501, splitPKChange says what still refuses there: a PostgreSQL window and a key whose decoded spelling is not the stored one). Three paths still REFUSE the run with an actionable error, by design: the shim's `_snapshot` and verify (SnapshotFullTableImages, which fetches one change per key and cannot rebuild the chain; its check, pkChangingUpdateInEvents, scans the raw event slice it is handed, and pkChangingUpdate is its map backstop), the Iceberg export (icebergexport/fold.go, its own per-event guard, not changed by #2501) and single-row reconstruct. There, re-snapshot the baseline at or after the PK change and reconstruct from the new one. Output formats FullTableConfig.OutputFormat accepts.

View Source 
const (
	ParquetWriterCompression  = "zstd"
	ParquetWriterRowGroupSize = 500_000
)

ParquetWriterCompression / ParquetWriterRowGroupSize are the codec and row group size a reconstructed baseline snapshot is written with. They are pinned to `bintrail baseline`'s own defaults rather than exposed as flags: the whole point of #1169 is that the emitted snapshot is indistinguishable from a mydumper-sourced one, and a snapshot chain whose files alternate codecs run to run is a needless source of "why is this one different".

View Source 
const (
	MetaKeySnapshotProducer = baseline.MetaKeySnapshotProducer
	MetaKeyDerivedFrom      = baseline.MetaKeyDerivedFrom
	MetaKeyDerivedFromPath  = baseline.MetaKeyDerivedFromPath
	// SnapshotProducerReconstruct is the MetaKeySnapshotProducer value this
	// package stamps. A dump stamps baseline.ProducerDump.
	SnapshotProducerReconstruct = baseline.ProducerReconstruct
)

Parquet file-metadata keys unique to a RECONSTRUCTED snapshot.

The definitions moved to internal/baseline in #1545, where the rest of the footer vocabulary lives and where the dump writers and every reader can reach them. These aliases stay so existing call sites and tests keep compiling, and because this is the package whose writer stamps them.

Written since #1169 as pure provenance — nothing looked them up, so their presence could not change how a consumer treated the snapshot. #1545 gave them readers; the string values did not move, because every snapshot already on disk carries them.

View Source 
const (
	RefreshVerdictRefreshed  = "refreshed"
	RefreshVerdictUnchanged  = "unchanged"
	RefreshVerdictRefusedGap = "refused-gap"
	RefreshVerdictRefusedDDL = "refused-ddl"
	RefreshVerdictRefused    = "refused"
	// RefreshVerdictRefusedRenumbered: the source's binary log started again
	// from another numbering since the snapshot (#2160); a full snapshot fixes it.
	RefreshVerdictRefusedRenumbered = "refused-renumbered"
	RefreshVerdictSkipped           = "skipped"
)

The verdicts a refresh gives one table. The literals are what `bintrail baseline refresh` prints and what the web interface stores in its run history, so they are a format: add one, never reword one.

View Source 
const (
	NoticeOnceAttr  = "notice"
	NoticeOnceValue = "once"
)

NoticeOnceAttr and NoticeOnceValue mark a line a NoticeOnce wrote: meant once per process. A process that relays the logs of children, each a new process, keeps the once across them by this mark (sqlsandbox ChildJob.LogOnce), and repeats every other line.

View Source 
const KeyIndexEnv = "BINTRAIL_KEY_INDEX"

KeyIndexEnv is the switch for the key index path of the lookup (#2496): ON by default since slice 4, measured against the join on real refreshes (the lookup of a 61 M-row table went from 2.0 s to 0.22 s per refresh). Unset, empty, 1/true/yes/on resolve the window's keys against <table>.keyidx beside the base, building one there when none is bound to it; 0/false/no/off (any case, spaces around) keep the join and the scan and build nothing. Any other value is the default, on, with one warning.

View Source 
const OutputFormatPsql = "psql"

OutputFormatPsql writes each table of a full-table reconstruct of a PostgreSQL source as one SQL file psql loads (#1596 slice 2): the table's CREATE TABLE, built from the snapshot's recorded shape, then its rows as a COPY ... FROM stdin block, in one transaction.

View Source 
const PGDeltaMargin = time.Minute

PGDeltaMargin is how far before a PostgreSQL snapshot's time a reader is sure to look for the changes it may be missing when the snapshot records no bound (MetaKeyPGDeltaSince). The only change a snapshot from `bintrail-pg baseline` can miss is one whose commit raced the snapshot's start: its commit time is just before the snapshot's, by far less than this. A commit held longer than this across the snapshot's start (a synchronous standby that stopped answering) is not covered. The fetch itself reads further back (query.CoarseSinceFloor under a position); the margin is what the coverage check holds the index to, so it stays short: an hour reached before a new index's first partition and refused every snapshot taken in its first hour.

Variables

View Source 
var (
	// ErrCaptureGap: the fold window spans a permanent capture loss (or an
	// index that cannot rule one out). Remedy: --allow-gaps to accept a
	// knowingly-incomplete result, or a fresh dump.
	ErrCaptureGap = errors.New("capture gap in the reconstruction window")
	// ErrSchemaChanged: the table's shape moved between the baseline and the
	// target — an ALTER, a destructive DDL, or delta events disagreeing with
	// the baseline's columns. Remedy: a real re-dump; no flag helps.
	ErrSchemaChanged = errors.New("schema changed since the baseline")
	// ErrTouchedRowBudget: the window changes more distinct rows than the
	// caller's MaxTouchedRows lets one update rebuild (#1107): past it on the
	// binlog-only fallback, or past it in one on-disk group of a baseline merge.
	// Remedy: a full backup, which streams the table instead of holding its
	// changes.
	ErrTouchedRowBudget = errors.New("too many changed rows to build this from the recorded changes")
)

Refusal classes a caller can act on without parsing error text. Every full-table refusal that a REFRESH loop needs to report distinctly wraps one of these; anything else is an ordinary failure.

They exist for `bintrail baseline refresh`, which must tell an operator WHY a table could not be refreshed — "the events are gone" and "the table changed shape" have completely different remedies — without the summary code string-matching messages that are free to be reworded.

View Source 
var ErrBinlogRenumbered = errors.New("the changes since the snapshot cannot be found by binlog position")

ErrBinlogRenumbered: the source's binlog numbering started over since the snapshot a run is anchored on (a RESET MASTER, a failover to another server, a change of the log_bin base name), so the changes since cannot be found by binlog position. Remedy: a new full snapshot. Not bypassed by --allow-gaps: nothing was lost that a flag could accept, the window is unreadable.

View Source 
var ErrDestructiveDDL = errors.New("destructive DDL in reconstruction window")

ErrDestructiveDDL is wrapped into the error CheckDestructiveDDL returns when it finds a TRUNCATE/DROP/RENAME/CREATE OR REPLACE on the target table inside the reconstruction window (#764).

View Source 
var ErrGeneratedPK = errors.New("primary key contains a generated column")

ErrGeneratedPK is the errors.Is sentinel for refusals caused by a generated primary-key member — the MariaDB system-versioning shape (#1266/#1273). Machine callers (the cascade engine's caveat classifier) use it to tell this PERMANENT table property apart from transient lookup failures without string matching.

Who carries it: every ERROR-typed refusal from this shape — the cascadebaseline provider's BaselineChildren refusal and fullTableGeneratedPKRefusal below, both built via GeneratedPKRefusalError. The verify inconclusive detail and the shim wire message are STRINGS by construction and cannot carry a sentinel; do not branch on errors.Is for those surfaces.

View Source 
var ErrLocalDiskFull = errors.New("no space left on the local disk")

ErrLocalDiskFull marks a write to a LOCAL folder (the download of an s3:// previous snapshot, or a compaction's temporary merge) that found the disk full (#2212). DuckDB reports that as text with no errno to unwrap, so it is classified here, where the write is known to be local, and never by matching the words later: the index MySQL reports its own full tmp disk with the same words, and that failure has nothing to do with this host's disk.

View Source 
var ErrNoBaseline = errors.New("no baseline snapshot found")

ErrNoBaseline is returned by FindBaseline when no baseline snapshot exists for the requested table at or before the target time.

View Source 
var ErrNoIndexedCoordinates = errors.New("no indexed event carries a binlog file/position")

ErrNoIndexedCoordinates is returned by ResolveSnapshotCut when the index holds events but none of them carries a usable binlog coordinate. A reconstructed baseline cannot be anchored in that case, and an unanchored baseline is worse than none: the next fold would have to guess where deltas resume.

View Source 
var ErrPGFullTableMydumper = errors.New("a full-table reconstruct of a PostgreSQL table is written as a snapshot (`--output-format parquet`) " +
	"or as SQL files psql loads (`--output-format psql`); a mydumper dump is MySQL's. Use single-row `reconstruct` or the shim `_flashback` to read one row")

ErrPGFullTableMydumper refuses a full-table PostgreSQL reconstruct to a mydumper dump: the dump is MySQL's, and so is every consumer of it (drill restores it into a MySQL scratch). Parquet is the PostgreSQL output.

View Source 
var ErrPKColumnMissing = errors.New("PK column missing from row")

ErrPKColumnMissing is a sentinel matchable via errors.Is for the case where a PK column declared in the resolver was not found in the row map passed to canonicalizePKMap. Returned errors are always typed as *MissingPKColumnError so callers that need the offending column name can recover it via errors.As.

View Source 
var ErrPKTypeUnknown = errors.New("primary key type unknown")

ErrPKTypeUnknown marks a refused key lookup: the key looks like a MariaDB UUID/INET value, and the schema snapshot that says whether the column is one could not be read. Surfaces map it to a client error, not a server fault.

View Source 
var ErrPsqlNeedsPostgres = errors.New("`--output-format psql` writes PostgreSQL's SQL, and this index records no PostgreSQL source; use `--output-format mydumper` for a MySQL dump")

ErrPsqlNeedsPostgres refuses the psql output on an index that records no PostgreSQL source: it is PostgreSQL's SQL.

View Source 
var ErrSnapshotNotDiscardable = errors.New("not an unpublished snapshot")

ErrSnapshotNotDiscardable marks a DELIBERATE refusal by DiscardUnpublishedSnapshot: the directory is not an unpublished snapshot, so nothing was removed and nothing went wrong.

It exists so a caller can tell "I correctly declined" from "I tried and the filesystem refused". Those are the same return shape and completely different facts: the first is the guard working, the second means the reclaim cannot run for that directory and its disk is leaking. Reporting them at one level, under one message, is how the second one goes unnoticed.

View Source 
var ErrUnreadableSnapshot = errors.New("a backup folder could not be read")

ErrUnreadableSnapshot marks a decision refused because a backup folder the local walk could not read sorts at or after the snapshot the decision would have picked (#1639): the answer would come from an older backup while looking current. A folder older than the pick changes nothing and never produces it. It is deliberately NOT ErrNoBaseline, which callers answer with a binlog-only table or another location.

View Source 
var ErrUnsupportedPKType = errors.New("primary-key type unsupported by the baseline canonicalizer")

ErrUnsupportedPKType is the errors.Is sentinel for refusals caused by a primary-key column whose DATA_TYPE the baseline canonicalizer does not handle (FLOAT/DOUBLE, TIME, BIT, JSON, the spatial family; supportedPKType is the authority, and DECIMAL and the BINARY/VARBINARY/BLOB family are NOT in this set - they have been supported since #214 and #1155). Like ErrGeneratedPK above it marks a PERMANENT property of the table, so machine callers can tell it apart from a transient lookup failure without string matching.

The ONE behavioural difference from ErrGeneratedPK, which the cascade engine relies on: a generated PK member also makes the BINLOG candidates unsafe (a versioned table's history rows and row_end tombstones share the surviving key), so that refusal skips both cascade phases. An unsupported PK TYPE only blocks the baseline join; the binlog scan is unaffected and must still run. See cascade.BaselineProvider's contract.

Who carries it: every ERROR-typed refusal from this shape, built via PKTypeRefusalError. The verify inconclusive detail and the shim wire message are STRINGS by construction and cannot carry a sentinel; do not branch on errors.Is for those surfaces.

View Source 
var ErrWriterClosed = errors.New("MydumperWriter: already closed")

ErrWriterClosed is returned from MydumperWriter methods called after Close. Kept as a sentinel so tests and future callers can match it with errors.Is.

View Source 
var KeyIndexEnabled = newKeyIndexSwitch(os.Getenv)

KeyIndexEnabled is the switch, read once per process and logged once. Exported for the daemon, which calls it when the baseline supervisor is built (every daemon that refreshes passes there, whether the refresh is scheduled from the console or by --baseline-refresh-interval) so the line is at boot and not at the first refresh with table deltas (the lookup is the only other reader, through the seam).

Functions

func AnchorOf added in v0.91.0 

func AnchorOf(bmeta baseline.DumpMetadata) *query.BinlogPos

AnchorOf is the binlog coordinate a snapshot's footer recorded, as DDLWindow.Anchor takes it: nil when it recorded none.

func ApplyAt 

func ApplyAt(initialState map[string]any, events []query.ResultRow, at time.Time) (map[string]any, error)

ApplyAt applies binlog events to initialState up to and including at, returning the final row state. Returns a nil state if the row was deleted. Events are folded in the order given. A caller that wants the row's changes in binary log order, and not in the (event_timestamp, event_id) order a fetch returns, passes them through EventsInBinlogOrder first (#2156).

The `ev.EventTimestamp.After(at)` check below is a defensive backstop, not the transaction-boundary cut: callers are expected to have already fetched events via FetchEventsAtomic (or trimmed them with TrimPartialTailTransaction), which drops a trailing PARTIAL transaction — one whose statements straddle `at` — wholesale rather than letting it be half-applied here (#783). A caller that passes raw query.FetchMerged output gets the pre-#783 per-row cut, which can apply half of a multi-statement transaction.

It fails loud on a residual unchanged-TOAST marker (#592) in any folded event's row image: the marker means the capture invariant (REPLICA IDENTITY FULL resolves every unchanged TOAST value at decode time) was violated, so the reconstructed state would silently carry the marker instead of the real value. Events after `at` are never inspected — they are not folded.

func BinlogBaseName added in v0.100.0 

func BinlogBaseName(file string) string

BinlogBaseName is a binlog file name without its numeric extension: "binlog.000007" is "binlog". A name with no numeric extension is its own base name.

func CanonicalizePKMap added in v0.18.0 

func CanonicalizePKMap(row map[string]any, pkCols []metadata.ColumnMeta) (map[string]any, error)

CanonicalizePKMap is the exported form of canonicalizePKMap, for callers outside this package (cascade Phase-2) that must encode a baseline Parquet row's primary key to match binlog_events.pk_values for deduplication.

func CheckBaselineSchemaCurrent added in v0.71.0 

func CheckBaselineSchemaCurrent(createSQL string, tm *metadata.TableMeta, schema, table string) error

CheckBaselineSchemaCurrent refuses with ErrSchemaChanged when the table's current schema snapshot has columns the baseline's embedded CREATE TABLE lacks, or the reverse. Column names only: unlike the fold, the export does not publish the carried CREATE TABLE, and it checks the types it exports itself (sameTableTypes).

func CheckCaptureGap added in v0.29.2 

func CheckCaptureGap(ctx context.Context, db *sql.DB, schema, table string, since, until time.Time, allowGaps bool) error

CheckCaptureGap consults stream_state.gap_lost_at/gap_lost_detail (#765) — the durable record indexer/streamrun stamps when an unfillable binlog gap forces an auto-advance, permanently losing events (docs/rotation-and-status.md: "the index is valid only up to the gap"). This is a DIFFERENT check from GapDetected in gap.go (baseline-anchor-vs-first-event position gap) and from the query planner's coverage-gap check (query.GapError: an hour rotated out of MySQL with no archive, which is potentially fillable by re-archiving). gap_lost_at marks events that no longer exist ANYWHERE — not live MySQL, not an archive — so a reconstruction spanning it is silently incomplete regardless of archive coverage.

If gap_lost_at falls inside (since, until], the loss is in scope for this reconstruction. The same verdict is returned when gap state is NOT EVALUABLE (a legacy index missing the gap_lost_* columns): the check would otherwise be silently inert on exactly the un-migrated indexes it protects. Under strict mode (allowGaps=false, the reconstruct default) this returns an error; under --allow-gaps it logs a slog.Warn and returns nil so the caller proceeds with a known-incomplete result.

func CheckDestructiveDDL added in v0.29.2 

func CheckDestructiveDDL(ctx context.Context, db *sql.DB, schema, table string, w DDLWindow) error

CheckDestructiveDDL refuses when schema_changes records a TRUNCATE TABLE, DROP TABLE, RENAME TABLE or CREATE OR REPLACE TABLE on schema.table inside the window a baseline+delta merge replays. See DDLWindow.place for what inside means.

TRUNCATE and DROP+re-CREATE emit no row-level binlog events — the parser only records them as an audit entry in schema_changes (#700-adjacent DDL tracking) — so ReconstructTable's and the shim's _snapshot full-table merge have no delta to apply and would silently pass every baseline row straight through, resurrecting rows the DDL actually deleted as if they still existed at --at (#764). Refusing with a clear, actionable error is the chosen fix over auto-truncating-and-replaying-from-the-DDL-point, which would be materially more complex and itself error prone (Occam's razor).

RENAME TABLE is included because it moves the table's row-event stream to a new name; the baseline for the old name can no longer be trusted to represent schema.table's state either.

MariaDB's CREATE OR REPLACE TABLE is included because on an existing table it drops the rows with no row events, exactly like DROP then CREATE (#1664).

A missing schema_changes table (a pre-DDL-tracking index, or a caller that hasn't run indexer.EnsureSchema) is treated as "nothing to check" rather than a hard failure: this is an additive safety net on top of the existing reconstruct contract, not a new hard dependency.

func CheckNumberingContinues added in v0.100.0 

func CheckNumberingContinues(ctx context.Context, db *sql.DB, m *EventMark, anchor query.BinlogPos) error

CheckNumberingContinues returns an error wrapping ErrBinlogRenumbered when an event indexed after the mark sorts before it. nil mark, no event after it, or a mark this index can no longer vouch for: nil.

Two events are read, both by primary key: the first one indexed after the mark, and the newest. The first catches a numbering that started over and has since grown past the mark; the newest catches one that started over later. A numbering that started over in between, and passed the mark again before either was read, is not seen; nor is one that started ABOVE the mark (a new server whose files are numbered higher), which is what the check of the source's identity is for (CheckSourceReplaced).

The mark's own row decides whether the ids can be trusted:

  • Present and the same event: yes.
  • Present and another event: the index was rebuilt and its ids name other rows. No check.
  • Gone, with older rows still in the index: deleted by the resume cleanup of a restarted stream, which captures those events again under new ids, AT their old positions, so they sort before the mark without any renumbering. No check: capture records the loss that comes with a reset in that mode (gap_lost_at), and the refresh refuses on it.
  • Gone with every older row (rotation dropped its hour): the events after it are still in order. Checked.

func CheckNumberingContinuesNote added in v0.100.0 

func CheckNumberingContinuesNote(ctx context.Context, db *sql.DB, m *EventMark, anchor query.BinlogPos) (string, error)

CheckNumberingContinuesNote is CheckNumberingContinues with the note of a check that could not tell (uncheckedNote), for a caller that must not read "cannot tell" as "continues": the read router's "unchanged since the snapshot" check (#2186). It reads no archive and adds no note for one.

func CheckNumberingFrom added in v0.100.0 

func CheckNumberingFrom(ctx context.Context, db *sql.DB, anchor *query.BinlogPos, rawMark string, w ReadWindow) error

CheckNumberingFrom runs the two checks that read a snapshot's event mark (rawMark, its footer value) against the index, for a window read by position from anchor (nil: the snapshot records none): CheckNumberingContinues, then CheckSameServer. A snapshot without a mark: nil, whatever happened. Shared by every reader that starts at a snapshot's position: the refresh, verify and the shim's _snapshot (#2174). CheckSourceReplaced stays with the refresh, which alone also refuses on a snapshot without a mark.

w describes the read. A zero w (the refresh, live verify) checks the whole index up to now. A w with Until set (a read that stops in the past: AS OF, a verify pair) checks only what that read can see; see ReadWindow.

func CheckNumberingFromRead added in v0.100.0 

func CheckNumberingFromRead(ctx context.Context, db *sql.DB, anchor *query.BinlogPos, rawMark string, w ReadWindow) (unchecked string, err error)

CheckNumberingFromRead is CheckNumberingFrom for a reader that hands its result to someone (#2186): verify, the shim's _snapshot, cascade recovery, reconstruct --at. Besides a refusal (an error wrapping ErrBinlogRenumbered) or a failure, it returns unchecked, a note for the result, when the check could not tell: the mark does not read, the index was rebuilt or backfilled (`bintrail index`), the mark's event was deleted while older ones remain, or one of the window's archived hours cannot be read. Each reader must put it in its result (an inconclusive verdict, a warning, an incomplete marker): passing it on silently is how a result is published as complete over a window nobody could check. "" when the check ran, or had nothing to check (no mark, no position, nothing indexed after the mark). A refusal or failure comes with no note.

func CheckSameServer added in v0.100.0 

func CheckSameServer(ctx context.Context, db *sql.DB, m *EventMark) error

CheckSameServer returns an error wrapping ErrBinlogRenumbered when the mark names the server capture was reading when the snapshot was taken and capture now reads another one. Identities, not times: it holds for a full backup's position too, which the time of a change record cannot be read against. A mark without one (written before it was recorded) or an index that does not say: nil.

func CheckSourceReplaced added in v0.100.0 

func CheckSourceReplaced(ctx context.Context, db *sql.DB, since time.Time) error

CheckSourceReplaced returns an error wrapping ErrBinlogRenumbered when capture recorded, at or after since, that the source answering at its address is another server (a new server_uuid). The two servers' binlog files are numbered apart, even when the new one's numbers are higher, which the event mark cannot tell. A change of host, port or user keeps the same server and the same files, so it does not count. A zero since, or an index without the table: nil.

MariaDB has no server_uuid: capture derives its identity from the address, so a failover behind the same address records nothing here.

func CheckSysVersioningChange added in v0.95.0 

func CheckSysVersioningChange(ctx context.Context, db *sql.DB, schema, table string, since, until time.Time) error

CheckSysVersioningChange refuses a single-row lookup over a window in which schema.table gained or lost MariaDB system versioning (#2007). Its changes are then stored under two key spellings (with and without ROW END) and read under two meanings, and a lookup shaped by one snapshot would miss the other period silently.

Two signals, either enough: an ADD/DROP SYSTEM VERSIONING recorded in schema_changes inside [since, until], and the schema snapshots in effect at since and at until disagreeing on whether the key has a generated member (the signal a file-mode index, which records no DDL, still has). An index without schema_changes skips the first.

func CheckUntypedMariaDBFixedPK added in v0.92.0 

func CheckUntypedMariaDBFixedPK(ctx context.Context, path string, pkFilter map[string]string, pkMetas []metadata.ColumnMeta) error

CheckUntypedMariaDBFixedPK refuses a single-row lookup that would answer wrong without saying so. The index keys a MariaDB UUID/INET4/INET6 row by the value's bytes, and only the index's schema snapshot tells a caller to spell the key that way (IndexPKSpelling). Without that snapshot (pkMetas nil) the key is looked up as text: the baseline row matches, no event does, and the snapshot-era row would come back as the state at the target time. So when pkMetas is nil and the baseline's own CREATE TABLE shows one of those types among the filtered columns, this returns an error naming the fix. An unreadable footer is an error; a CREATE the parser cannot read is scanned for the type names instead. Only a footer with no CREATE TABLE at all (a baseline older than that metadata) passes unchecked.

func CompactionDir added in v0.84.0 

func CompactionDir(compactDir, schema, table string, chainStart time.Time) string

CompactionDir is where a compaction job stages its result for one chain: "<compactDir>/<schema>/<table>/<chain start>/". The chain start names the chain, so a result for a chain that has since been rewritten is never mistaken for the current one.

func DecodeEventBinaries added in v0.25.1 

func DecodeEventBinaries(db *sql.DB, schema, table string, events []query.ResultRow) bool

DecodeEventBinaries reverses the storage-side base64 of BLOB/TEXT columns in the events' row images in place — the sibling of MapEventEnumLabels for the other class of values go-mysql delivers as []byte (#666, same root as recover #662 / shim #661). go-mysql delivers BLOB/TEXT as []byte, which marshalRow base64-encodes into the stored event JSON; the single-row reconstruction surfaces (`bintrail reconstruct`, console Time-travel) fold these events onto the baseline via ApplyAt/BuildHistory, which do NOT decode, so without this pass the client gets the base64 text instead of the real value.

Call this on fetched deltas BEFORE folding them onto a baseline: event images only, never a baseline row (baseline rows are scanned raw from Parquet and must not be decoded — a baseline TEXT value that happens to be valid base64 would be corrupted). The full-table merge path (ReconstructTable, #668) calls this too, on the full events slice before its change map is built — callers must call it exactly once per events slice, since decodeStoredBase64 is not idempotent and a second pass would double-decode — this now also covers the #736 bool/json.Number repair output ("true"/"false" is valid base64 alphabet and would silently re-decode into unrelated bytes on a second pass).

Each column is typed at the snapshot in effect at its event's timestamp (#475-style epoch awareness, matching MapEventEnumLabels): whether a value is base64-stored depends on whether the column was delivered as []byte (BLOB/TEXT) or string (VARCHAR/CHAR) when the event was captured, so a VARCHAR→TEXT widening must not make a wrong-epoch lookup wrongly decode an old plain-string value that happens to be valid base64.

Unlike MapEventEnumLabels there is deliberately NO latest-snapshot fallback: relabeling an ENUM by the latest definition is harmless (strings pass through), but base64-decoding a value the wrong schema calls BLOB/TEXT is DESTRUCTIVE (a plain VARCHAR value that happens to be valid base64 would be corrupted to garbage bytes). So when the epoch typing is unavailable — no snapshots, or a per-epoch resolver that fails to load — the value is left as the base64 it was stored as, never decoded by a guess.

The returned bool reports whether that degradation happened: true when every event's timestamp resolved to a snapshot epoch whose resolver and table lookup succeeded (so every base64-stored value was decoded), false when at least one event's typing was unavailable and its values may still be stored base64. Callers that only best-effort decode (recover, shim, reconstruct) ignore it; verify uses it to keep a digest comparison honest — an undecoded value is a representation gap, not a divergence (#769/#791).

func DeltaChainStart added in v0.83.0 

func DeltaChainStart(ctx context.Context, basePath string) (time.Time, error)

DeltaChainStart is deltaChainStart for the one reader that pairs a base with its event window without going through FindBaseline (verify's FindBaselinePair, which lists snapshots itself).

func DiscardUnpublishedSnapshot added in v0.70.0 

func DiscardUnpublishedSnapshot(dir string) (discarded bool, err error)

DiscardUnpublishedSnapshot removes a snapshot directory that a FAILED fold wrote, reclaiming the disk it holds. discarded reports whether the directory is gone from dir.

It refuses unless the directory is positively marked incomplete: _INCOMPLETE present and _SUCCESS absent, which is exactly baseline.SnapshotComplete returning false. That is not a formality. A published snapshot carries _SUCCESS, and a legacy snapshot carries NEITHER marker and is complete by default, so both are refused by the same check. It also makes a path mistake fail safe: a directory this fold never wrote has no _INCOMPLETE marker, so a caller that computes the wrong path deletes nothing.

The removal is a rename FIRST and a delete second, and the order is the whole point. os.RemoveAll deletes children in directory order with no promise that _INCOMPLETE goes last, and '_' sorts ahead of a lowercase schema directory, so a process killed mid-delete could leave Parquet files behind with no marker at all. baseline.SnapshotComplete reads a markerless directory as COMPLETE, which would turn an undiscoverable partial snapshot into a discoverable one: the exact failure #467 exists to prevent, caused by the cleanup meant to help. A rename inside one directory is atomic, and once it lands nothing can discover the snapshot however the delete then goes.

A delete that fails after the rename returns discarded=true WITH the error: the snapshot is already beyond every reader, and only the bytes are still there. Callers report both.

func EventsInBinlogOrder added in v0.100.0 

func EventsInBinlogOrder(events []query.ResultRow, at time.Time, idsFollowBinlog func([]query.ResultRow) query.IDProof) ([]query.ResultRow, query.BinlogOrder)

EventsInBinlogOrder returns the events of ONE row that ApplyAt and BuildHistory fold for a reconstruction as of at, in the order the source's binary log holds them (#2156), and what was decided about that order.

events come in (event_timestamp, event_id) order, and event_timestamp is the time a change's STATEMENT STARTED. When session B waits on a row lock that A holds, A's change is in the binary log first and carries the later time: folded as fetched, the row ends on A's values and the history lists B before A, where the database holds B's.

Two steps, in this order:

  1. Keep the events whose event_timestamp is at or before at. The cut stays a cut by statement time (a change that started before at and committed after it is inside: #2156 item 6, not changed here). It has to happen BEFORE the reordering: ApplyAt and BuildHistory stop at the first event after at, which is only "every later event" while the input is in time order. Handing them a list with nothing after at makes that stop unreachable.
  2. query.OrderByBinlog over what is left. It keeps the fetched order, and says why in the decision's Warning, when binary log order cannot be established: see its comment. The caller owes that warning to whoever reads the result.

The input is not modified. idsFollowBinlog is query.BinlogOrderProof for a caller that holds the index.

The shim's single-row `_flashback` and `_snapshot` reads use it as well (shim.foldRowInBinlogOrder).

func ExecSQL 

func ExecSQL(ctx context.Context, source, sqlStr string) ([]map[string]any, []string, error)

ExecSQL runs arbitrary SQL against an in-memory DuckDB instance and returns result rows and column names. The httpfs extension is loaded automatically when source or sqlStr references an s3:// URL.

func FetchEventsAtomic added in v0.30.0 

func FetchEventsAtomic(
	ctx context.Context,
	db *sql.DB,
	engine *query.Engine,
	fm query.FetchMergedOptions,
	at time.Time,
) ([]query.ResultRow, *query.QueryPlan, error)

FetchEventsAtomic fetches events for a point-in-time reconstruction via query.FetchMerged and trims a trailing PARTIAL transaction from the result (#783). It is the convenience entry point for callers that don't need to inspect the raw fetch before trimming — see TrimPartialTailTransaction for callers that do (e.g. the console's fetch-then-overflow-check-then-trim ordering).

event_timestamp is a Rows_event's STATEMENT execution time, not its transaction's COMMIT time. A multi-statement transaction executing at T1 and T2 (T1 < T2) that commits later is written to the binlog — and indexed — as a contiguous run of row events sharing one GTID. Cutting naively at `event_timestamp <= at` (query.Options.Until) can therefore include only the transaction's T1 half when T1 <= at < T2: a row state that never existed at any real point in time, since MySQL only makes a transaction's changes visible atomically, at commit. See TrimPartialTailTransaction for the fix.

func FindDestructiveDDL added in v0.87.0 

func FindDestructiveDDL(ctx context.Context, db *sql.DB, schema, table string, since, until time.Time) (ddlType string, detectedAt time.Time, found bool, err error)

FindDestructiveDDL is CheckDestructiveDDL's finding without its message, for a caller that is not reconstructing and says it in its own words (verify). An index with no schema_changes table answers not found, as CheckDestructiveDDL does. It looks by time alone.

Unlike CheckDestructiveDDL, since is INCLUSIVE. detected_at and the snapshot times are whole seconds, so a statement stamped the same second as the older snapshot may have run after that snapshot's anchor, where the replay starts. Missing it gives the false mismatch this lookup exists to prevent; counting it at worst leaves one table unchecked for one run. snapshotForDDLInWindow takes the same second as inside for the same reason.

func FirstUnsupportedPKType added in v0.70.0 

func FirstUnsupportedPKType(pkCols []metadata.ColumnMeta) (metadata.ColumnMeta, bool)

FirstUnsupportedPKType returns the first primary-key member whose DATA_TYPE the baseline canonicalizer cannot handle, and whether one exists. Input order is the caller's (PKColumnMetas() preserves ordinal order), same contract as GeneratedPKColumn.

An EMPTY DataType is NOT a verdict here and is skipped. It is the PostgreSQL snapshot signature — metadata.WritePGSnapshot leaves data_type and column_type empty (#533) — and flagging it would tell every PostgreSQL operator their schema is unsupported when nothing about the column is wrong.

Only for callers with NO upstream source-flavor check: skipping the empty case means they never claim a cause they cannot know. A caller that HAS established it is on the MySQL path (ReconstructTable, verify, the shim) must keep its own loop over SupportedPKType, so an empty DataType still reaches PKTypeGateReason's wrong-path verdict there. Do not point those loops at this helper: it would silently retire that verdict.

func FixedBinaryWidth added in v0.47.0 

func FixedBinaryWidth(columnType string) int

FixedBinaryWidth extracts n from a "binary(n)" COLUMN_TYPE, returning 0 when it is absent or unparseable (a pre-#212 snapshot carries no COLUMN_TYPE).

Exported because internal/cli's padFixedBinaryFilter must agree with altFixedBinaryPK about the pad width — a `--pk` filter and a merge join that disagreed would resolve the same key differently. internal/verify/render.go keeps its own equivalent for the #1135 render padding; the two must stay in agreement, so change them together.

func FoldBeforeSnapshot added in v0.108.0 

func FoldBeforeSnapshot(row map[string]any, events []query.ResultRow, snapshotTime time.Time, pkCols []metadata.ColumnMeta) (map[string]any, []query.ResultRow, error)

FoldBeforeSnapshot applies to row (a snapshot's row, nil when the snapshot has none) the events stamped before snapshotTime and returns the result with the remaining events, in their order. A fetch from a PostgreSQL snapshot's LSN floor (#1596) returns changes from before its time: most are already in the snapshot (applying a full row image again changes nothing), and one whose commit raced the snapshot's start is not. Folded here, they never reach a caller as changes after the snapshot: the history, the "earliest event is not an INSERT" tell and the event count see what they saw before. events must be in commit order within those stamped before snapshotTime. row is not modified.

An UPDATE that moved the row to another primary key (pkCols) leaves this one: the row stops existing here, it does not become the moved row. The events of one key are stored under its before-image, so the fetch for the old key returns that UPDATE, and its after-image belongs to the new key.

func GapDetected added in v0.29.0 

func GapDetected(flavor string, eventFile string, eventStartPos uint64, baselineFile string, baselinePos int64, baselineLSN uint64) bool

GapDetected reports whether there is a gap between a baseline's recorded anchor and the first indexed event available for replay: events between the anchor and the first event are missing from the reconstruction, so callers warn (the CLI's --allow-gaps semantics stay untouched — this is the baseline-vs-first-event check, not the query planner's coverage-gap check).

flavor is the stream_state.flavor value ("mysql", "mariadb", "postgres", or "" when unknown — see query.SourceFlavor).

PostgreSQL ("postgres"): coordinates are WAL LSNs. eventFile holds the LSN's TEXT form ("0/1A2B3C4"), which is NOT lexically ordered ("0/10" < "0/9" lexically but not numerically), so the comparison must use the numeric LSN that PG events carry in StartPos (internal/pgcapture rowEvent) against the baseline's numeric delta-replay floor (baseline.MetaKeyLSN — an INCLUSIVE lower bound: the slot's confirmed_flush_lsn/restart_lsn as of just before the snapshot, not the snapshot's own live pg_current_wal_lsn(); see that key's doc comment and #771). baselineLSN == 0 means the baseline predates the LSN floor (#593 slice A) or is not a PG baseline at all: the floor is UNKNOWN, so no gap is flagged. That mirrors how this path already treats missing coordinates — the caller's "baseline lacks position metadata" branch skips detection with an informational message rather than failing (and status's continuity verdict fails closed only under an explicit --fail-on-gap) — and it can never silently produce wrong data because this check only gates a WARNING, never the reconstruction itself. Callers should log the skip (cli/reconstruct.go does).

The strict "eventStartPos > baselineLSN" comparison below is unchanged by #771: a baseline floor is a value below which no coverage gap can exist by construction (deltas are read from at-or-after it), so "no gap" already means eventStartPos <= baselineLSN — the same comparison as when this value was (incorrectly) the live anchor.

Any other flavor (MySQL, MariaDB, ""): the established two-key binlog compare — file names ordered lexically (equal to numeric order for MySQL's zero-padded binlog names within one server's sequence, see query.BinlogPos), byte position breaking ties within the same file.

func GeneratedPKColumn added in v0.50.0 

func GeneratedPKColumn(pkCols []metadata.ColumnMeta) (metadata.ColumnMeta, bool)

GeneratedPKColumn returns the first primary-key member that is a STORED/ VIRTUAL generated column, in input order, and whether one exists (#1266). Callers pass PKColumnMetas(), which preserves ordinal order — the ordering guarantee is the input's, not this function's.

The shape this detects in practice is MariaDB system versioning: MariaDB silently extends a versioned table's PRIMARY KEY with its ROW END period column (`PRIMARY KEY (id, row_end)`, observed on 11.4 for both visible and INVISIBLE explicit period columns) and marks that column STORED GENERATED in information_schema, which is how it reaches the snapshot's is_generated flag. An ordinary STORED generated column inside a PK trips this too, on purpose: in both cases the baseline deliberately omits the column's values (mydumper never dumps generated columns), so no baseline-side PK join key can be built and the merge would die deep in the probe with MissingPKColumnError on every row.

Why the gates that call this do not simply drop the column from the join key: the binlog does NOT carry only current-state rows for a versioned table. Verified against MariaDB 11.4 and 11.8: an UPDATE logs the current-row update PLUS a Write_rows for the history row (same remaining key, row_end = now), and a DELETE logs no Delete_rows at all — it is an Update_rows tombstone (row_end marker → now). Under a reduced key alone the history insert would overwrite the current row in the last-write-wins change map, orphan history rows would be emitted as duplicate live rows, and tombstoned deletes would resurrect. Full-table reconstruct (and so snapshot refresh) reads these tables since #2007 by reading each event for what it means to the current rows (sysVersioned in sysversion.go) on top of the reduced key; the surfaces that still call this gate do not.

func GeneratedPKGateReason added in v0.50.0 

func GeneratedPKGateReason(c metadata.ColumnMeta, surface string) string

GeneratedPKGateReason renders the refusal/inconclusive detail for a primary key that contains a generated column (see GeneratedPKColumn). surface names the feature speaking ("verify", "full-table reconstruct", ...). No CLI flag names in the text: console and wire surfaces emit it too.

func GeneratedPKRefusalError added in v0.50.0 

func GeneratedPKRefusalError(msg string) error

GeneratedPKRefusalError builds an error that renders exactly msg and matches errors.Is(err, ErrGeneratedPK). Every error-typed generated-PK refusal must be built through this, or the cascade engine's permanent- caveat classifier files it as a transient lookup failure.

func HasLocalSnapshot added in v0.92.0 

func HasLocalSnapshot(baselineDir string) bool

HasLocalSnapshot reports whether a LOCAL baseline directory holds at least one snapshot the listing would return: a timestamp-named directory that is complete (#467) with a table file under a schema directory. It stops at the first one it finds and reads no Parquet footer, so it is cheap enough for a capability probe that runs on every page load; ListBaselinesReport is the full answer. An unreadable directory is "no".

func IndexPKSpelling added in v0.48.0 

func IndexPKSpelling(pk string, pkMetas []metadata.ColumnMeta) string

IndexPKSpelling rewrites a user-supplied PK value into the spelling the indexer stored in binlog_events.pk_values, so an event fetch matches what the operator typed.

MariaDB UUID/INET4/INET6 components are parsed from their text form (any spelling metadata.ParseMariaDBFixedKey accepts, the bytes in hex included) and spelled as the captured full-width bytes. Otherwise only fixed-width BINARY(n) components are touched, and this is the INVERSE of padFixedBinaryFilter — the two run in opposite directions on purpose, because they target different stores. Reproducing event.formatPKValue exactly: trailing 0x00 padding is stripped (the ROW image never carries it), and the hex is uppercased, but ONLY when the trimmed bytes are not valid UTF-8 — formatPKValue is content-gated, so a binary key whose bytes are printable ASCII is stored verbatim and must stay that way.

Everything else — every other column type, and every component that is already in the stored spelling — is returned untouched, so this cannot disturb a lookup that resolves today.

func InvalidateS3Inventory added in v0.89.0 

func InvalidateS3Inventory(s3URL string)

InvalidateS3Inventory marks stale the directory listing of every cached s3:// baseline source that holds s3URL: the source itself, or a parent of it, since the daemon names the snapshot's OWN directory when it uploads (<source>/<snapshot>) and the inventory is the source's. The next read lists the directories again instead of waiting out s3DirsTTL: the Snapshots page and the coverage card must show the snapshot just written on their next read, and the scheduler's next fold must anchor on it. The contents already read are kept, except the directory the URL itself names, which the caller just wrote into. A URL that is not s3:// has no inventory here and is ignored, and one no inventory covers creates none.

func IsBase64StoredType added in v0.71.0 

func IsBase64StoredType(dataType string) bool

IsBase64StoredType reports whether a column of this DATA_TYPE is stored base64-encoded in the event row images (BLOB and TEXT families, JSON, the spatial types, VECTOR, and MariaDB's UUID/INET4/INET6), i.e. one whose exported value depends on the epoch decoder having resolved. One list, the decoder's own.

func ListBaselinesUnreadable added in v0.82.0 

func ListBaselinesUnreadable(ctx context.Context, source string) ([]BaselineFile, []UnreadableSnapshot, error)

ListBaselinesUnreadable is ListBaselinesReport with the skipped folders named and dated, for a caller that DECIDES on the listing rather than shows it (#1639): pass them to UnreadableAtOrAfter with the snapshot it picked.

func MapEventEnumLabels added in v0.13.3 

func MapEventEnumLabels(db *sql.DB, latest *metadata.Resolver, schema, table string, events []query.ResultRow)

MapEventEnumLabels rewrites ENUM/SET ordinals in the events' row images back to their string labels in place (#472/#476), decoding each event with the snapshot in effect at its timestamp (#475). Reconstruction surfaces (console Time-travel, `bintrail reconstruct`, full-table merge) call this on fetched deltas BEFORE folding them onto a baseline, so the merged state carries labels exactly like the baseline rows do (mydumper dumps labels as strings; strings pass through the mapper untouched).

The raw event-record surfaces — `bintrail query`, the MCP query tool, and the console events view — deliberately stay unmapped: the stored ordinal is the forensic ground truth, recoverable regardless of later enum reshapes.

latest is the caller's already-loaded resolver, used as the fallback definition when the epoch lookup is unavailable (nil is fine: the remaining degradation is pass-through — raw ordinals, never a guessed label).

func MaterializeBaselineLocal added in v0.71.0 

func MaterializeBaselineLocal(ctx context.Context, path string, tuning duckdbutil.Tuning) (string, func(), error)

MaterializeBaselineLocal returns a local path for a baseline Parquet file: the file itself (after the #636 integrity check) for a local baseline, or a temporary download for an s3:// one. The returned cleanup must be called.

func NewTailDecoder added in v0.104.5 

func NewTailDecoder(db *sql.DB, schema, table string, latest *metadata.Resolver) func([]query.ResultRow)

NewTailDecoder decodes a page of schema.table's events as the refresh does before it folds them (enum and set labels per snapshot epoch, binaries), built once for a tail and reused for every page: its per-epoch resolvers are what a call costs.

func NewestPerTable added in v0.91.0 

func NewestPerTable(files []BaselineFile) []int

NewestPerTable returns the indexes, into files, of each table's newest snapshot file: the ones a headline verdict is decided on, and so the ones worth a footer read.

func NewestSnapshot added in v0.83.0 

func NewestSnapshot(ctx context.Context, source string) (time.Time, []string, error)

NewestSnapshot is NewestSnapshotTables plus the snapshot's own instant, for a caller that folds forward FROM it and needs to know how much time that fold covered (#1693). The instant is the directory's timestamp; zero when there is no snapshot.

It reads that instant off files[0], which rests on sortBaselineFiles having put the newest first, on the local and the S3 path alike. Worth reconciling with consoleapp's newestSnapshotOf, which deliberately does NOT trust the order and takes a maximum instead: that function compares TWO independent listings for equality, where trusting the order lets an older instant the two sides happen to share read as a match. There is one listing here and nothing to match it against, so the sort is the whole contract. A caller that starts comparing this instant across sources wants the maximum form.

func NewestSnapshotTables added in v0.50.0 

func NewestSnapshotTables(ctx context.Context, source string) ([]string, error)

NewestSnapshotTables returns the schema.table entries of the NEWEST discoverable snapshot under source, sorted; nil when nothing is discoverable.

It is the table list a REFRESH operates on. Deliberately the newest snapshot rather than every table the index has ever seen: a refreshed snapshot is a strict successor of the one it was folded from, and a table absent from the source has nothing to fold onto — inventing an entry for it would publish a snapshot claiming coverage it does not have.

A folder the walk could not read at or after the newest readable snapshot (or at all, when none is readable) refuses with ErrUnreadableSnapshot (#1639): the list would be an older snapshot's, or a short one.

func NormalizeSnapshotWriter added in v0.91.0 

func NormalizeSnapshotWriter(id string) string

NormalizeSnapshotWriter returns id in the form SnapshotWritersSeen reports writers in, "" when it cannot sign a snapshot. A caller comparing its own identity against the writers seen compares this.

func PGDeltaFloor added in v0.108.0 

func PGDeltaFloor(bmeta baseline.DumpMetadata) *query.BinlogPos

PGDeltaFloor is where a reader of a PostgreSQL snapshot fetches the changes after it from: the snapshot's LSN floor (MetaKeyLSN), nil for a MySQL snapshot (it records a binlog coordinate, AnchorOf) or a PostgreSQL one taken before the floor was recorded. The floor is read before the snapshot opens, so a transaction can commit at or after it and before the snapshot's time, yet be missing from the snapshot (#771): a reader that starts at the snapshot's time drops it, with nothing to say so. Reading from the floor applies some changes the snapshot already holds again, which is harmless: every change is a full row image.

func PGDeltaWindow added in v0.108.0 

func PGDeltaWindow(bmeta baseline.DumpMetadata, snapshotTime time.Time) (since time.Time, from *query.BinlogPos, proven bool, err error)

PGDeltaWindow is where a reader of a snapshot fetches the changes after it from: since and from (SincePos), and whether since is a recorded or documented bound the fetch must keep (query.Options.SinceProven). For a MySQL snapshot, or a PostgreSQL one with no floor: snapshotTime, nil, false. For a PostgreSQL one: the floor, and MetaKeyPGDeltaSince (never after snapshotTime) or PGDeltaMargin before snapshotTime. A MetaKeyPGDeltaSince that does not parse refuses.

func PGEventCapNote added in v0.108.0 

func PGEventCapNote(pg bool) string

PGEventCapNote is what a PostgreSQL read adds to a per-row event cap's refusal: the count starts at the snapshot's LSN floor, which can be before the snapshot's time (#1596), so a nearer target time may not bring it under. "" when pg is false.

func PKChangedInEvent added in v0.71.0 

func PKChangedInEvent(ev *query.ResultRow, pkCols []metadata.ColumnMeta) (before, after string, changed bool)

PKChangedInEvent reports whether ev is an UPDATE whose primary key changed (#782). The Iceberg export folds events by the before-image PK exactly like the full-table merge, so it refuses the same events for the same reason.

func PKChangingUpdateError added in v0.71.0 

func PKChangingUpdateError(schema, table, before, after string) error

PKChangingUpdateError is the refusal message for a PK-changing UPDATE, the same words the full-table reconstruct uses.

func PKTypeGateReason added in v0.49.0 

func PKTypeGateReason(c metadata.ColumnMeta, surface, action string) string

PKTypeGateReason renders the refusal detail for a primary-key column a MySQL-path SupportedPKType gate rejected. Two physically different causes reach such a gate, and they must not share a message (#1009, lifted here from internal/verify for the reconstruct and shim surfaces — #1198):

  • A real MySQL DATA_TYPE token the baseline canonicalizer does not handle (float, bit, ...): a genuine per-table limitation the operator can reason about, reported as such.
  • An EMPTY DataType: MySQL's information_schema always records a DATA_TYPE (the column is NOT NULL and every MySQL snapshot writer stores a non-empty token), so an empty token is the PostgreSQL snapshot shape (WritePGSnapshot stores pg_type_oid, never a MySQL type token — the #533 invariant). Reaching a MySQL-path gate with it means the run selected the MySQL path for a PostgreSQL-shaped index — the source flavor recorded in stream_state did not read "postgres" (unreadable stream_state, or the wrong index database). Blaming the PK type there sends the operator chasing a fixable-looking column problem that nothing they do to the table can fix; the honest verdict names the wrong-path cause.

surface names the feature that took the MySQL path ("verify", "reconstruct", "full-table _snapshot", ...) and action the verb it cannot perform on a PostgreSQL-sourced table ("verify", "reconstruct", "materialize", ...). No CLI flag names in the text: console and wire surfaces emit it too.

func PKTypeRefusalError added in v0.70.0 

func PKTypeRefusalError(frame, reason string) error

PKTypeRefusalError builds an error that renders "<frame>: <reason>" and matches errors.Is(err, ErrUnsupportedPKType). reason must come from PKTypeGateReason so every surface refuses in the same words. Every error-typed PK-type refusal must be built through this, or the cascade engine's permanent-caveat classifier files it as a transient lookup failure and repeats the doomed baseline scan for every parent key.

func PKTypeRefusalReason added in v0.70.0 

func PKTypeRefusalReason(err error) string

PKTypeRefusalReason returns the unframed gate sentence carried by an error built with PKTypeRefusalError, or "" when err is not one. Callers that add their own frame use it so the two frames do not nest.

func PartitionsAtOrAfter added in v0.100.0 

func PartitionsAtOrAfter(ctx context.Context, db *sql.DB, at time.Time) (clause string, names []string, err error)

PartitionsAtOrAfter returns the PARTITION (...) clause naming the binlog_events partitions that can hold an event stamped at or after at, and those names, for a reader bounded by a timestamp floor that MySQL's own pruning would not bound (it keeps the oldest partition, #1692). The clause is "" for an unpartitioned table. The layout can move under the reader (rotation's REORGANIZE of p_future, see firstEventPast): a caller that must not miss a row lists again afterwards and compares the names. An error, or a partition name this build does not recognise, is returned as an error; the caller then reads without the clause.

func PastCutWarning added in v0.100.0 

func PastCutWarning(order query.BinlogOrder, explicitCut bool) string

PastCutWarning is the text owed to the reader of a single-row reconstruction when two things hold together, and "" otherwise:

  • the row's selected changes were REORDERED (order.Sorted): at least one of them was committed after a change that started later, and
  • the request named its own instant (explicitCut: `--at` / `at` was given, not defaulted to now).

The cut selects by the time a statement STARTED (#2156 item 6, not changed here). Take A, started :02 and committed at once, and B, started :00, which waited and committed at :03. As of :02.5 the row held A. Both are selected, and binary log order applies B last: the answer is B's value, which was not committed yet at that instant. Folded in statement-time order the answer was A, right for that instant and wrong for every later one. The binary log order is kept, because it is the right order of the changes; what cannot be had from the statement time is where to stop, and the reader is told.

With the cut at "now" every indexed change was committed before it, so there is nothing to warn about.

The wording names no flag: the MCP tool and the web interface show it too.

func ReadBaselineColumns added in v0.24.0 

func ReadBaselineColumns(ctx context.Context, path string, tuning duckdbutil.Tuning) ([]string, error)

ReadBaselineColumns returns the column names of a baseline Parquet file (local or s3://). This is the authoritative non-generated column set for fingerprinting the table: mydumper excludes true STORED/VIRTUAL generated columns from the dump but keeps ordinary expression-default (DEFAULT_GENERATED) columns, so the Parquet schema is exactly the set to hash — deriving it from a schema snapshot's is_generated flag instead would wrongly drop DEFAULT_GENERATED columns (the trap consistency.ConsistentTableChecksum documents) and silently under-verify them.

tuning sets the DuckDB session's resource budget (#842); zero value falls back to the container-safe default (effectiveDuckDBTuning). Callers that resolve an operator-supplied --ultrafast/--duckdb-* tuning (verify) should pass it through here so it's honored on this path too, not just archive fetches.

func ReadBaselineRow 

func ReadBaselineRow(ctx context.Context, path string, pkFilter map[string]string, pkMetas []metadata.ColumnMeta) (map[string]any, error)

ReadBaselineRow opens the Parquet file at path using DuckDB and returns the first row matching pkFilter (column name → value string). Returns nil when no row matches. Loads the httpfs extension automatically for s3:// paths.

pkMetas, when non-empty, are the searched table's primary-key column metas (see ResolvePKMetasAt) and enable the fixed BINARY(n) reconciliation (#1155/#1157): a key copied out of binlog_events.pk_values carries the ROW image's trailing-0x00-stripped spelling, which is SHORTER than the padded value the baseline stores, so after an exact miss the lookup is retried once with every fixed BINARY(n) component re-padded to its declared width. The retry runs only on a miss, so it can never override a correct hit. Callers with no index open — and therefore no declared column widths — pass nil and keep the exact-match-only behavior.

A MariaDB UUID/INET4/INET6 component is first re-spelled as the text the baseline holds (mariaDBFixedBaselineFilter); a value that is not one of those types' text is refused with an error.

func ReadBaselineRows added in v0.18.0 

func ReadBaselineRows(ctx context.Context, path string, filter map[string]string, limit int) ([]map[string]any, error)

ReadBaselineRows returns every baseline row matching filter (an arbitrary column→value map, AND-ed), up to limit rows (limit <= 0 = no cap). Unlike ReadBaselineRow it is not restricted to the primary key, so cascade Phase-2 can scan a child snapshot by a foreign-key column and recover ALL matching children. Values are bound as query parameters; column names are quoted identifiers; the path is single-quote-escaped.

func ReadBounds added in v0.91.0 

func ReadBounds(ctx context.Context, files []BaselineFile) []status.ReadBound

ReadBounds returns, for each of files in order, where a reader of that table file starts its event fetch: what FindBaseline returns for it.

It never fails. A chain whose start cannot be read (damaged delta files, an unreadable footer, a store that did not answer, ctx done first, a read that panicked) comes back Unread, which no verdict grades as covered. Over a local folder that is one footer read per file with a chain; over S3 the same, in parallel, and none for a file this process already answered for.

func ReadStreamEventMark added in v0.100.0 

func ReadStreamEventMark(ctx context.Context, db *sql.DB) string

ReadStreamEventMark is ReadEventMark on an index a stream wrote, encoded: "" anywhere else, where ascending id is not the order rows were indexed in and a mark would prove nothing. A failed read leaves the snapshot without one, which only means the next run cannot use it.

func ReconstructTablesDetailed added in v0.50.0 

func ReconstructTablesDetailed(ctx context.Context, cfg FullTableConfig) ([]*TableReport, []TableFailure, error)

ReconstructTablesDetailed is ReconstructTables plus the per-table failures.

ReconstructTables joins every failure into one error, which is right for a command that either produces a dump or doesn't. A refresh has to render a per-table verdict — refreshed / refused-gap / refused-ddl — so it needs the failures apart, classifiable against the sentinels above. The run semantics are identical, including all-or-nothing publication: any failure leaves the snapshot directory marked _INCOMPLETE and therefore undiscoverable.

func ResolvePKMetasAt added in v0.48.0 

func ResolvePKMetasAt(db *sql.DB, schema, table string, at time.Time) []metadata.ColumnMeta

ResolvePKMetasAt loads schema.table's primary-key column metadata from the schema snapshot in effect at `at` — metadata.EpochAt over the snapshot history, the same per-instant rule the ENUM/SET decode path uses (#475) — rather than from the latest snapshot (#1159). The metas' declared widths drive padFixedBinaryFilter, so the anchor should be the instant whose schema produced the bytes being matched: callers pass the BASELINE snapshot time, because the pad must reach the width the baseline file actually stores. A latest-snapshot width is wrong whenever the column was widened after that instant (BINARY(16) → BINARY(32)): the retry would pad to 32 against a 16-byte stored value and miss silently.

When the instant predates every snapshot, EpochAt answers the FIRST epoch — the closest available description; with no epoch history at all, the latest snapshot is used.

Best-effort by design: every caller only uses the result to IMPROVE a lookup or an error message, so a missing/unreadable snapshot degrades to nil (no reconciliation — the pre-#1155 behavior) instead of failing a reconstruct that would otherwise have worked.

func ResolveSnapshotCut added in v0.50.0 

func ResolveSnapshotCut(ctx context.Context, db *sql.DB, at time.Time) (*query.BinlogPos, error)

ResolveSnapshotCut returns the exact binlog coordinate that separates "already folded into the snapshot" from "still to come", for a snapshot targeting the point in time at.

Why a position and not a timestamp

The snapshot this cut anchors is a BASELINE: the next reconstruct anchored on it fetches deltas with Options.SincePos = this coordinate, i.e. every event whose start_pos is at-or-after it. For the chain to lose nothing and double-apply nothing, the set folded HERE and the set fetched NEXT TIME must partition the binlog exactly. buildQuery's two position predicates already do that — UntilPos admits `end_pos <= C`, SincePos admits `start_pos >= C` — so a single coordinate C used on both sides is a seam with no gap and no overlap.

A timestamp cannot play that role. binlog_events.event_timestamp is the statement's EXECUTION time, not its commit time (the skew #797 exists to route around), so "the events at-or-before at" and "the events committed before some position" are different sets, and the difference is silently LOST: dropped by this fold's `event_timestamp <= at` filter and again by the next fold's positional lower bound, which never consults a timestamp at all.

How the coordinate is chosen

C is the start_pos of the FIRST event, in commit order, whose timestamp is past at. Every event that ends at-or-before C therefore committed before that event, and so — because that event is the first one past at — carries a timestamp at-or-before at as well. That is the property that makes it safe to keep the caller's exact `Until: at` time filter alongside `UntilPos: C`: the time filter is a superset of the positional window, so it can never exclude a row the position admits. Inverting the two (deriving a position from the time cut) does not have that property, which is why this is written as a search for the first event past at rather than the last event before it.

When no indexed event is past at (the ordinary case for a refresh targeting "now"), the cut is the end_pos of the newest event: everything indexed is folded, and the next fold resumes exactly after it.

Why the newest event is read FIRST (#1695)

Capture keeps inserting while the cut is resolved (`watch` runs both in one process), so the two statements can see different tables. They used to run search-first: an event past at, inserted between the search (which then found nothing) and the newest-event read, became the newest event, so the cut landed at its END. This fold's time filter dropped it and the next fold's positional lower bound skipped it: folded into no snapshot, silently.

Reading the newest event first closes that window. The search then runs over a table that holds at least everything the first read saw (the one thing that deletes indexed rows meanwhile is the resume-time cleanup after a stream restart, which re-captures what it deleted; the fold's own fetch is exposed to it under either order, and the cut now is too, for the length of the search: a cleanup landing right after the newest read can pin the cut above rows this fold does not carry, and the next fold starts past them. That window is seconds wide and only opens on a restart, while the one this ordering closes opens on every insert), so if any event up to the newest one is past at, the search finds it (or an earlier one) and the cut lands at its start. An event inserted after the first read comes after the newest one in the binlog, so the newest one's end_pos leaves it for the next fold. Re-checking the newest event's timestamp after a search-first pass would NOT be enough: event_timestamp is execution time, so an event past at can commit ahead of an older-stamped one, and a check on the newest row alone would miss it.

Assumptions

Commit order is read as ascending event_id. event_id is AUTO_INCREMENT and the capturer inserts in stream order, so it tracks binlog order — the same assumption query.OldestIndexedEvent already makes to report the index's starting coordinate. It can be violated by `bintrail index` fed explicit --files out of order; such an index is not a supported input for a self-refreshing baseline chain.

The partition bound trusts names: p_YYYYMMDDHH is taken to be VALUES LESS THAN the next hour and p_future the MAXVALUE catch-all — the layout init, rotation and restore-index write. firstEventPast never reads PARTITION_DESCRIPTION, so a hand-made partition carrying one of those names with a different bound would silently shrink the search.

Returns (nil, nil) when the index holds no events at all — there is nothing to fold, and the caller keeps the source baseline's own coordinates.

func ResolveTableDelta added in v0.103.0 

func ResolveTableDelta(ctx context.Context, basePath string, chain *baseline.TableDeltaChain, prevBase string) (done, linked bool, err error)

ResolveTableDelta writes the resolved pair of the chain beside basePath (#2231, baseline.ResolvedTableDeltaPaths): the WHOLE chain merged into one range pair, the last pair included, under the directory beside the snapshots. The merge is CompactTableDeltaMinor's, so the pair holds each key once and the state read through it is the state read through the chain; unlike a compaction's, the result is never linked into a snapshot.

prevBase is the same table's file in the previous snapshot, or "". A refresh with no change to a table carries its chain forward, and then the pair that resolved the previous snapshot's chain resolves this one: it is linked instead of merged again. The test is that pair's footers naming THIS chain (resolvedIsFor: its start, its base, its range and its last pair's position), which only the pair of the same chain does; not that the chain's files are the same inodes, which a carry-forward that had to copy would fail at every refresh.

Before a merge reads a file of the chain, the file is checked against the snapshot's manifest when the manifest lists it (baselineintegrity.ValidateLocalFileOnce, a read once per process per identity, #2517; a snapshot with no manifest is merged unchecked, as every other reader reads it): the pair is read in place of those files, so it must not be the way damaged bytes reach a statement. A linked pair was checked when it was merged.

A pair already there is kept only if its footers name this chain and this base (resolvedIsFor); one that does not is replaced. A reader trusts the pair by its name, the snapshot's and the range, so this is where a pair left under a snapshot directory someone replaced by hand is caught.

The two files are made in a directory of this call's own and renamed into place, the upserts last: a reader finds both or takes neither, two calls for one chain cannot remove each other's files, and a failure leaves nothing a reader would take for a pair.

done is false, with no error, where there is nothing to resolve: a chain of one pair, a legacy pair, a path with no place for one.

func ResolvedTableDeltaCurrent added in v0.103.0 

func ResolvedTableDeltaCurrent(basePath string, chain *baseline.TableDeltaChain) bool

ResolvedTableDeltaCurrent says whether the chain beside basePath has its resolved pair, finished and merged from this chain (resolvedIsFor).

func SingleRowKeyColumns added in v0.95.0 

func SingleRowKeyColumns(tm *metadata.TableMeta, flavor string) []string

SingleRowKeyColumns names the key columns a single-row lookup takes from its caller: the snapshot's primary key, minus the ROW END column MariaDB appends to a system-versioned table's key (the declared key is what an operator knows; #2007). A MySQL or PostgreSQL source keeps every column.

func SnapshotAt added in v0.78.0 

func SnapshotAt(ctx context.Context, source string, at time.Time) (tables []string, anchor time.Time, err error)

SnapshotAt is SnapshotTablesAt plus the anchor it settled on: the snapshot time of the newest discoverable snapshot at or before at (zero when none is that old). A caller that is about to PUBLISH a snapshot named by at needs the anchor to notice that one already exists at exactly that instant.

Like NewestSnapshotTables it refuses with ErrUnreadableSnapshot when a folder it could not read sits between the anchor and at (#1639).

func SnapshotDirName added in v0.65.0 

func SnapshotDirName(at time.Time) string

SnapshotDirName renders a snapshot's directory name from its target instant, byte-identical to what baseline.Run produces (RFC3339 UTC with ':' replaced by '-' for filesystem portability). The name IS the snapshot's timestamp as far as FindBaseline is concerned, so this formatting is a compatibility contract, not a display choice. Exported for the console's backup detail and download surfaces, which resolve one snapshot's directory from its time.

func SnapshotDirTime added in v0.83.0 

func SnapshotDirTime(tablePath string) (time.Time, bool)

SnapshotDirTime is the time of the snapshot DIRECTORY a table file sits in, read off its path (<root>/<snapshot>/<schema>/<table>.parquet). FindBaseline returns a chain's start for a table with a delta, which is right for bounding a read and wrong for asking which of two copies is NEWER; this is for that.

func SnapshotDirVacant added in v0.70.0 

func SnapshotDirVacant(dir string) (bool, error)

SnapshotDirVacant reports whether dir is free for a new fold to write into: absent, or holding nothing but the _INCOMPLETE marker a previous failed run left behind.

It is the same test ReconstructTables applies before it creates the snapshot directory, exported for one purpose: a caller that intends to RECLAIM what its own fold wrote can establish, BEFORE the fold runs, that everything in the directory afterwards is its own. Asking afterwards cannot answer that question, because by then the caller's own files are in there too.

A directory that cannot be read is reported as NOT vacant, with the error, so a caller that cannot see what is in there never concludes it may delete it.

func SnapshotFullTableImages 

func SnapshotFullTableImages(ctx context.Context, in SnapshotFullTableInput, emit func(map[string]any) error) error

SnapshotFullTableImages reconstructs the full row state of a table at the AsOf instant by merging the baseline snapshot with the change map, calling emit once per surviving row (baseline column order, values verbatim from DuckDB / the binlog event). It is the in-memory sibling of mergeBaselineIntoWriter, for the shim's full-table _snapshot path: instead of writing mydumper output it streams row maps to the caller, which builds the wire resultset and enforces its own row cap (by returning an error from emit).

s3:// baselines are downloaded to a temp file first; the temp dir is removed before return. Real failures (unreadable baseline, DuckDB error, a PK value the canonicalizer can't translate) are returned as errors for the caller to surface; the caller decides separately — before calling this — whether a missing baseline or an unsupported PK type should instead fall back to a binlog-only path.

func SnapshotReadsFrom added in v0.91.0 

func SnapshotReadsFrom(ctx context.Context, root string, at time.Time) (time.Time, error)

SnapshotReadsFrom returns the earliest instant a reader of any table of the LOCAL snapshot at `at` under root fetches events from (#1904): the start of the oldest chain of deltas in it, or the snapshot's own time when no table has a chain. It is the instant the snapshot stays restorable only while the index still holds, which the refresh loop's gate has to watch rather than the snapshot's directory time.

It errors whenever that instant is not known: no table file listed at that snapshot, or a chain whose start could not be read. Never a guess, since the caller acts on a later answer as "still covered".

func SnapshotSigners added in v0.92.0 

func SnapshotSigners(source string, at time.Time) (writers []string, known bool, err error)

SnapshotSigners returns who signed the snapshot of source at at (#1762), normalized and sorted; empty for an unsigned one. known is false when this process cannot tell: an S3 snapshot whose directory no listing here has read yet, or a local directory that is not there any more (a prune between the listing and this read: the job that asked finds it gone on its own). A local snapshot is read on the spot.

func SnapshotTablesAt added in v0.65.0 

func SnapshotTablesAt(ctx context.Context, source string, at time.Time) ([]string, error)

SnapshotTablesAt returns the schema.table entries of the newest discoverable snapshot at or before at, sorted; nil when no snapshot is that old. It is the table list a point-in-time restore folds forward — NOT always the newest snapshot's: a restore to a moment before the newest snapshot anchors on the older snapshot FindBaseline will pick, and must fold that snapshot's tables.

func SnapshotWritersSeen added in v0.91.0 

func SnapshotWritersSeen(source string) []string

SnapshotWritersSeen returns the writers that signed the snapshots of source, as the listings made so far in this process saw them: sorted, distinct, normalized. It reads nothing from storage. Call it after a listing of the same source, and it answers for that listing.

Empty means no signed snapshot was seen, which is also the answer for a source never listed.

func SpellIndexPKFilter added in v0.92.0 

func SpellIndexPKFilter(ctx context.Context, db *sql.DB, opts *query.Options) (map[string]string, error)

SpellIndexPKFilter rewrites the key filter of a query or recover request (opts.PKValues, opts.PKValuesIn) into the spelling binlog_events.pk_values holds, for the one case where the typed spelling and the stored one differ by design: a MariaDB UUID/INET4/INET6 key. The index keys those rows by the value's bytes (spelled "0xCC5C…" when they are not valid UTF-8), while a person types the text the application shows ("cc5c4e6e-bc5a-…"). Looked up as typed, such a key matched nothing and the answer was "no history", without an error. reconstruct already spelled it this way (IndexPKSpelling).

The stored spelling becomes opts.PKValues and the typed one moves to opts.PKValuesAlt, so the live index matches either. The order matters: the Parquet archive reader filters on PKValues alone. For PKValuesIn the stored spellings are added beside the typed ones, and the returned map (typed to stored) lets a caller that groups rows by the typed key find them.

First it adds the stored spellings of a MariaDB system-versioned table's key (#2007, expandSysVersionedPKFilter): that reads stream_state's source flavor on every lookup and, unless the source is PostgreSQL, the table's key from schema_snapshots (two reads on idx_table_snapshot). The UUID/INET part reads the snapshot only when some typed component parses as one of these types (in any spelling metadata.ParseMariaDBFixedKey accepts), so a numeric or plain text key costs it nothing. When one does and the snapshot cannot say what the column is, the lookup is refused with ErrPKTypeUnknown instead of answering "no history" on a guess. A table the request's profile denies, or leaves out of its allow list, is left alone, so the answer cannot describe its key.

Call it after the request's RBAC posture is on opts, and after any #957 escape alternates are set: it replaces PKValuesAlt when it re-spells.

func SupportedPKType 

func SupportedPKType(dataType string) bool

SupportedPKType is the exported form of supportedPKType, for callers outside this package (the shim's full-table _snapshot path) that need to guard a PK column type before attempting a baseline merge and fall back to their binlog-only path when it isn't supported.

func SweepDiscardedSnapshots added in v0.70.0 

func SweepDiscardedSnapshots(root string) (removed int, err error)

SweepDiscardedSnapshots removes ".<ts>.discarding" staging directories a previous discard did not finish deleting, and reports how many it reclaimed.

This is the other half of the convention discardingSuffix borrows. Prune stages the same way AND sweeps its leftovers at the top of every cycle (baseline.sweepPruningLeftovers), and the sweep is what makes the leftovers bounded rather than permanent. Adopting only the rename would trade one leak for a rarer leak in a shape nothing names: a staging directory is invisible to every discovery path BY DESIGN, so no listing, no status and no console panel would ever mention the disk it holds.

The producer is not only a disk fault. Nothing joins the refresh goroutine at shutdown, so a daemon killed during the delete exits with a part-deleted staging tree, and that path logs nothing at all because the process is gone before the caller's own line is written.

Best-effort by contract: it is housekeeping running beside real work, so a directory it cannot read or delete is reported, never fatal. An unreadable ROOT yields (0, nil) rather than an error, matching prune's sweeper, because the caller's own work fails on the same root and would say so first.

func TailKeyRefusal added in v0.114.0 

func TailKeyRefusal(name string, cols []baseline.Column, pk []string) (pkNames []string, why string)

TailKeyRefusal is the rule on a table's key for a tail: every key column (pk, by name, in key order) must be in cols and of an integer type. It returns the key's column names as cols spell them, or why the table has no tail. Exported so a check over the port's tails can tell this refusal, the product's own rule, from one it did not cause.

func TouchedRowBudgetError added in v0.82.0 

func TouchedRowBudgetError(limit int64, tables int, spilled bool) error

TouchedRowBudgetError is the changed-rows refusal, exported so the console's page tests render the exact text the daemon produces.

Without spilled, limit is what one table may hold in memory and the fold refused past it. With spilled, the changes went to disk and one of the spillBuckets groups alone passed limit when read back, which takes about spillBuckets times limit changed rows (#1107).

func TrimFixedBinaryPad added in v0.47.0 

func TrimFixedBinaryPad(b []byte) []byte

TrimFixedBinaryPad strips the trailing 0x00 bytes MySQL adds when storing a short value in a fixed-width BINARY(n) column, reproducing what the binlog ROW image carries for that value (and therefore what the indexer stored in binlog_events.pk_values). Values with no trailing zero byte — the common case, e.g. most BINARY(16) UUIDs — come back untouched.

Exported for the CLI, which needs the INVERSE (pad a pk_values spelling back to the baseline's full width) and must agree with this function about which bytes are padding.

func TrimPartialTailTransaction added in v0.30.0 

func TrimPartialTailTransaction(
	ctx context.Context,
	db *sql.DB,
	engine *query.Engine,
	fm query.FetchMergedOptions,
	events []query.ResultRow,
	at time.Time,
) ([]query.ResultRow, error)

TrimPartialTailTransaction drops a trailing PARTIAL transaction from events (#783): events must already be sorted ascending by (event_timestamp, event_id) and already cut at `event_timestamp <= at` (e.g. via query.Options.Until) — exactly what query.FetchMerged returns.

Binlog transactions are written (and therefore indexed) as a contiguous run of row events sharing one GTID, in commit order. This groups the TAIL of events by the last event's GTID and, with one bounded follow-up query scoped to that exact GTID, checks whether the transaction has ANY further event — on any table, not just the one being reconstructed, since a transaction is one atomic unit server-wide — at or after the first representable instant past `at`. If it does, the entire trailing GTID group is dropped: never partially applied. Only the LAST group can straddle the boundary (GTIDs are assigned, and their events written, in monotonic commit order), so one trim pass suffices.

Events with no GTID (replication without GTIDs, or archived data predating the gtid column) can't be grouped this way and pass through unchanged — the pre-#783 per-row cut is the best available answer in that mode.

Residual limitation: the index persists event_timestamp as DATETIME(0) (one-second granularity) and never records true commit time. This cannot resolve sub-second transaction ordering, nor the symmetric case of a transaction whose statements ALL execute before `at` but which commits after it (it is still included whole, indistinguishable from one that committed before `at`) — resolving that would require a commit-timestamp column the index does not have today. See docs/query-and-recovery.md.

func UnreadableAtOrAfter added in v0.82.0 

func UnreadableAtOrAfter(unreadable []UnreadableSnapshot, from, until time.Time) error

UnreadableAtOrAfter returns an ErrUnreadableSnapshot error naming the newest unreadable folder whose time is at or after from and, when until is non-zero, at or before until; nil when there is none. from is the snapshot a decision picked (zero when it found none, which makes every unreadable folder up to until count).

func WarnBaselineFirstEventGap added in v0.30.0 

func WarnBaselineFirstEventGap(flavor string, bmeta baseline.DumpMetadata, first query.ResultRow, start query.IndexStart, startOK bool, schema, table string)

WarnBaselineFirstEventGap emits the baseline↔first-event gap warning for callers that live in this package — the full-table reconstruct path (#781), which previously produced a dump silently missing that gap while single-row reconstruct at least warned. It is a DIRECT port of the check the single-row CLI path runs (cli/reconstruct.go: resolveGapCheck + the verdict switch), duplicated here because that logic lives in package cli, which this package cannot import (cli imports reconstruct).

Warn-only, identical to single-row: this check gates a WARNING, never the reconstruction itself, so --allow-gaps is deliberately not consulted here — it governs the coverage-gap fetch (query.FetchMerged) and CheckCaptureGap, which run separately, not this baseline-vs-first-event visibility warning.

flavor is query.SourceFlavor(db); start/startOK are query.OldestIndexedEvent(db) — the coverage-proof evidence consulted by DecideBaselineGap (#1163). first is the earliest fetched event (events sorted by (event_timestamp, event_id)). schema/table are added to every record so the warning is attributable when several tables reconstruct concurrently.

func WriteHistoryCSV 

func WriteHistoryCSV(entries []StateEntry, w io.Writer) error

WriteHistoryCSV writes history entries as CSV. Columns are: time, source, event_id, gtid, followed by all state columns (union of all entries, sorted).

func WriteHistoryJSON 

func WriteHistoryJSON(entries []StateEntry, w io.Writer) error

WriteHistoryJSON writes history entries as an indented JSON array.

func WriteHistoryTable 

func WriteHistoryTable(entries []StateEntry, w io.Writer) error

WriteHistoryTable writes history entries as a tabwriter table. The STATE column renders as compact "key=value" pairs.

func WriteMetadataFile 

func WriteMetadataFile(outputDir string, at time.Time, gtid, binlogFile string, binlogPos int64) error

WriteMetadataFile writes a mydumper-compatible top-level "metadata" file that baseline.ParseMetadata can read back. The format is symmetric to what mydumper produces so myloader recognizes it.

outputDir is the same directory passed to NewMydumperWriter. at is the target point-in-time of the reconstruction. binlogFile / binlogPos / gtid should come from baseline.ReadParquetMetadataAny so the metadata file reflects the baseline's replication position, not some synthetic value.

func WriteSQLResultsCSV 

func WriteSQLResultsCSV(rows []map[string]any, cols []string, w io.Writer) error

WriteSQLResultsCSV writes dynamic DuckDB SQL results as CSV.

func WriteSQLResultsJSON 

func WriteSQLResultsJSON(rows []map[string]any, w io.Writer) error

WriteSQLResultsJSON writes dynamic DuckDB SQL result rows as an indented JSON array.

func WriteSQLResultsTable 

func WriteSQLResultsTable(rows []map[string]any, cols []string, w io.Writer) error

WriteSQLResultsTable writes dynamic DuckDB SQL results as an aligned table.

func WriteStateCSV 

func WriteStateCSV(state map[string]any, w io.Writer) error

WriteStateCSV writes a single row state as CSV (header + one data row).

func WriteStateJSON 

func WriteStateJSON(state map[string]any, w io.Writer) error

WriteStateJSON writes a single row state as indented JSON.

func WriteStateTable 

func WriteStateTable(state map[string]any, w io.Writer) error

WriteStateTable writes a single row state as aligned key-value pairs.

Types

type BaselineFile added in v0.13.1 

type BaselineFile struct {
	SnapshotTime time.Time
	Schema       string
	Table        string
	Path         string
	// DeltaUpserts is the file NAME of the newest pair of the chain of deltas
	// beside the file (#1638), read off the names the listing had in hand;
	// empty when there is no chain. DeltaErr is set when the delta files
	// beside it do not form a chain. See ReadBounds (#1707).
	DeltaUpserts string
	DeltaErr     error
}

BaselineFile is one table's Parquet file discovered in a baseline source listing. Path-derived only — no file contents are read — so listing stays cheap over both a local directory and an s3:// prefix. Binlog coordinates live in Parquet metadata and are the caller's (optional) enrichment step.

func FindBaselineFile added in v0.104.2 

func FindBaselineFile(ctx context.Context, source, schema, table string, at time.Time) (BaselineFile, error)

FindBaselineFile is the table's file in the newest complete snapshot at or before at, with the time of the snapshot directory that holds it, or ErrNoBaseline. Unlike FindBaseline it does not move the time back to the start of a delta chain: it is for a caller that needs to know WHICH snapshot the file is in, not where to bound an event fetch from.

func ListBaselines added in v0.13.1 

func ListBaselines(ctx context.Context, source string) ([]BaselineFile, error)

ListBaselines enumerates every baseline snapshot file under source (a local directory or an s3:// prefix), newest snapshot first (then schema/table for a stable render order). Entries that don't match the <timestamp>/<schema>/<table>.parquet layout are skipped, mirroring FindBaseline's tolerance; unreadable snapshot subdirectories are skipped WITH a warning — the listing is an observability surface, and a silently shrunken one would misreport "latest baseline" with nothing in any log. internal/baseline.DiscoverBaselines walks the same local layout (plus Parquet footers) for `bintrail status`; keep the two in sync if the layout ever changes.

func ListBaselinesNewestReport added in v0.85.0 

func ListBaselinesNewestReport(ctx context.Context, source string, newest int) (files []BaselineFile, skipped int, more bool, err error)

ListBaselinesNewestReport is ListBaselinesReport bounded to the newest `newest` snapshots (0 = all), for a surface that shows a page of them (#1679): the work follows what is returned, not the inventory. more says whether older snapshots exist beyond the window. Local sources read the whole directory listing anyway (cheap) and cut; an s3:// source fetches only the window's objects.

func ListBaselinesReport added in v0.82.0 

func ListBaselinesReport(ctx context.Context, source string) (files []BaselineFile, skipped int, err error)

ListBaselinesReport is ListBaselines plus the number of snapshot or schema directories the local walk could NOT read and skipped (#1601). A nil error with skipped > 0 means "listed what it could", not "listed everything": the listing may be missing whole snapshots, so a consumer that grades coverage or names the newest snapshot must treat it as partial. os.ReadDir returns this shape for a datadir at 750, a dangling mount, or a file-descriptor burst under load. An s3 listing is one query and either answers whole or fails, so it reports zero.

func (BaselineFile) HasDelta added in v0.91.0 

func (f BaselineFile) HasDelta() bool

HasDelta reports whether the listing found delta files beside the file, whole or damaged.

type CaptureGap added in v0.46.0 

type CaptureGap struct {
	// Unevaluable is true when gap state was never readable: a legacy index
	// whose stream_state predates the gap_lost_* columns (#765), read before
	// any migrating command ran. status reports that as "not evaluated" rather
	// than a clean verdict (internal/status/status.go, GapColumnsPresent); the
	// same honesty applies here — absent data is not evidence of no loss.
	// At/Detail are zero in this case.
	Unevaluable bool
	// At is the stamped moment capture permanently lost events, and Detail the
	// supplementary human-readable context (may be empty). Set only when
	// Unevaluable is false.
	At     time.Time
	Detail string
	// Since/Until are the reconstruction window the gap was evaluated against.
	Since, Until time.Time
}

CaptureGap describes what stream_state says about PERMANENT capture loss relative to one reconstruction window. A non-nil value means the fold over that window cannot be asserted complete — either a stamped loss falls inside it, or the index cannot answer the question at all.

func CaptureGapStatus added in v0.46.0 

func CaptureGapStatus(ctx context.Context, db *sql.DB, since, until time.Time) (*CaptureGap, error)

CaptureGapStatus evaluates stream_state's permanent-loss record (#765) against the (since, until] reconstruction window and reports the finding WITHOUT deciding what to do about it — CheckCaptureGap is the strict/allow policy on top, and surfaces that need to both refuse and, when overridden, carry the finding into their payload (the MCP reconstruct tool) call this directly. A nil result means "no permanent loss in scope, and that verdict was actually evaluated".

func CheckCaptureGapStatus added in v0.50.0 

func CheckCaptureGapStatus(ctx context.Context, db *sql.DB, schema, table string, since, until time.Time, allowGaps bool) (*CaptureGap, error)

CheckCaptureGapStatus is CheckCaptureGap plus the finding it acted on: the same strict/allow policy, but the caller also learns WHAT was overridden when allowGaps let the run proceed.

It exists because "proceeded over a known gap" has to outlive the log line that reported it. A snapshot published under --allow-gaps is knowingly incomplete forever, and the only place that can say so forever is the artifact itself (baseline.MetaKeyCaptureGap, #1170) — a warning in a terminal the operator has since closed is not a record.

A nil finding with a nil error means the window was evaluated and is clean.

func (*CaptureGap) Reason added in v0.46.0 

func (g *CaptureGap) Reason() string

Reason renders the finding as a flag-free sentence, so each surface can append its own remediation (`--allow-gaps` on the CLI, `allow_gaps: true` on MCP) without the other surface's vocabulary leaking into it.

type DDLMark added in v0.91.0 

type DDLMark struct {
	ID         uint64    `json:"id"`
	File       string    `json:"binlog_file"`
	Pos        uint64    `json:"binlog_pos"`
	DetectedAt time.Time `json:"detected_at"`
	Type       string    `json:"ddl_type"`
}

DDLMark names one schema_changes row: its id and what the row holds, so a mark read against an index whose ids now name other rows (restore-index, the table dropped and created again) can be told apart and ignored.

func ParseDDLMark added in v0.91.0 

func ParseDDLMark(raw string) *DDLMark

ParseDDLMark reads a footer's mark. "" is no mark; one that does not read is no mark too, said out loud: no mark places more rows, never fewer.

func ReadDDLMark added in v0.91.0 

func ReadDDLMark(ctx context.Context, db *sql.DB) (*DDLMark, error)

ReadDDLMark reads the newest schema_changes row. nil when the table is empty or missing: there is nothing a mark could leave out.

func (DDLMark) Encode added in v0.91.0 

func (m DDLMark) Encode() string

Encode is the mark as baseline.MetaKeyDDLMark stores it.

type DDLWindow added in v0.91.0 

type DDLWindow struct {
	Since, Until time.Time
	Anchor, Cut  *query.BinlogPos
	// Mark is the snapshot's DDL mark (ddl_mark.go), nil when it has none.
	// A row at or below it is not placed by position alone.
	Mark *DDLMark
	// contains filtered or unexported fields
}

DDLWindow is the stretch of the source's history a replay covers, stated both ways the replay itself bounds it: by time and by binlog position.

Since and Until are the snapshot's time and the requested instant. Anchor is the binlog coordinate the snapshot was taken at, where its row changes start; nil when it recorded none (an older snapshot, a PostgreSQL one). Cut is the run's positional upper bound (ResolveSnapshotCut), set only by a run that bounds its row changes by position; nil everywhere else.

type EventDecoder added in v0.71.0 

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

EventDecoder is the per-table, epoch-aware decoder the full-table fold builds once and reuses across pages: ENUM/SET ordinals become labels and base64-stored values become real values. BLOB/TEXT are decoded only against the schema snapshot in effect at each event's own timestamp; ENUM/SET fall back to the LATEST snapshot when the epochs cannot be loaded, and Typed reports the first case only. Not safe for concurrent use.

func NewEventDecoder added in v0.71.0 

func NewEventDecoder(db *sql.DB, schema, table string, latest *metadata.Resolver) *EventDecoder

NewEventDecoder builds the decoder for one table run. latest may be nil.

func (*EventDecoder) DecodePage added in v0.71.0 

func (e *EventDecoder) DecodePage(page []query.ResultRow)

DecodePage decodes one fetched page in place. Call it exactly once per page: the base64 pass is not idempotent.

func (*EventDecoder) Typed added in v0.71.0 

func (e *EventDecoder) Typed() bool

Typed reports whether every event decoded so far resolved to a schema epoch. False means at least one event could not be typed against the snapshot in effect at its timestamp: a base64-stored value may still be the stored text, and an ENUM/SET ordinal was mapped by the latest definition or not at all.

type EventMark added in v0.100.0 

type EventMark struct {
	ID   uint64 `json:"id"`
	File string `json:"binlog_file"`
	End  uint64 `json:"end_pos"`
	// ServerUUID is the server capture was reading when the mark was taken
	// (bintrail_servers.server_uuid of the stream's bintrail_id); "" when the
	// index does not say.
	ServerUUID string `json:"server_uuid,omitempty"`
}

EventMark names one binlog_events row: its id, and where it ends, so the check can tell a row that is gone from a row whose id now names another event.

func ParseEventMark added in v0.100.0 

func ParseEventMark(raw string) *EventMark

ParseEventMark reads a footer's mark. "" is no mark; one that does not read is no mark too, said out loud.

func ReadEventMark added in v0.100.0 

func ReadEventMark(ctx context.Context, db *sql.DB) (*EventMark, error)

ReadEventMark reads the newest indexed event that carries a coordinate. nil when the index holds none.

func (EventMark) Encode added in v0.100.0 

func (m EventMark) Encode() string

Encode is the mark as baseline.MetaKeyEventMark stores it.

type FullTableConfig 

type FullTableConfig struct {
	IndexDSN    string    // DSN for the bintrail index database
	BaselineSrc string    // local directory or s3:// URL of baselines
	Tables      []string  // "db.table" entries
	At          time.Time // target point-in-time
	OutputDir   string    // output root: the mydumper dump directory, or the baselines root under OutputFormatParquet
	ChunkSize   int64     // per-chunk SQL file size (0 → 256 MiB)
	// ExplicitAt says At is a point in the past a person asked for
	// (`bintrail reconstruct --at`), not the time a refresh runs to. The
	// binlog-renumbering and server checks then look only at what the read up
	// to At can see (ReadWindow, #2182): a restart or a move to another
	// server recorded after At leaves that read whole. Unset (every refresh,
	// and the command without --at), they check the whole index, as before.
	// A dump only: under OutputFormatParquet the window also ends at the
	// cut, the position of the first change past At, which after such a
	// restart is in the new numbering, so the whole-index check stays.
	ExplicitAt bool
	// SpaceCheck, when set, is called right before a file is created, with its
	// directory and the bytes it is expected to need: before each SQL chunk
	// (the chunk size), before a table's Parquet file (the size of the backup
	// file it is rebuilt from), and, when DownloadDir is set, before an s3://
	// previous snapshot is downloaded into it (the object's size, #2212). A
	// table carried forward unchanged never reaches it. An error stops that
	// table, whose files are then removed. The daemon checks free disk with it
	// (#1614).
	SpaceCheck func(dir string, need int64) error
	// DownloadDir, when set, is where an s3:// previous snapshot is downloaded
	// before it is merged (materializeBaselineLocal), in a temporary folder of
	// its own that is removed when the table is done. Empty: the system's
	// temporary directory, as the command line has always used. When set,
	// SpaceCheck is asked before each download, about that disk. With it
	// empty no download is checked. (Nothing sets it since #2431 removed the
	// S3-only servers that did.)
	// Never inside the snapshot directory itself: everything there is
	// uploaded.
	DownloadDir string

	// OutputFormat selects the artifact the run produces. The zero value means
	// "not specified" and resolves to OutputFormatMydumper, matching this
	// struct's existing convention for ArchiveFetcher/WarnEventThreshold.
	//
	// Under OutputFormatParquet the run writes a discoverable BASELINE SNAPSHOT
	// (#1169) instead of a SQL dump, and OutputDir is read as the baselines root
	// — the same path `bintrail baseline --output` takes — not as the directory
	// the files land in directly. See emitParquetSnapshot for the layout and the
	// anchoring contract.
	OutputFormat string

	// WriterID is the identity a Parquet snapshot is signed with (#1762).
	// Empty means "ask the index": the run reads the source's bintrail_id
	// from the index it folds (serverid.SnapshotWriterID), which is the
	// writer of THIS snapshot whoever wrote the one it builds on.
	// OutputFormatParquet only; a mydumper dump is not a snapshot in a
	// baselines root and is never signed.
	WriterID string

	Parallelism int  // max concurrent tables (0 → runtime.NumCPU())
	AllowGaps   bool // false = strict abort on gaps (default for reconstruct)

	// CarryForwardUnchanged publishes a table that had NO events in the window
	// by carrying its previous Parquet file forward (a hard link where the
	// filesystem allows one) instead of folding an empty change map over it and
	// re-emitting the same rows. OutputFormatParquet only.
	//
	// Opt-in at this layer, and the zero value is the conservative one. The rows are the same
	// either way; what changes is the representation on disk. See
	// carryForwardEligible for the trade-offs an operator is agreeing to.
	CarryForwardUnchanged bool

	// TableDeltas stops a refresh from rewriting a changed table (#1638): the
	// table's Parquet file is carried forward and the change is written as two
	// small files beside it. OutputFormatParquet only. See tabledelta.go.
	//
	// OPT-IN, and turning it back off needs nothing else: the next run folds
	// from each base alone, rewrites every table the old way and writes no
	// delta, so the snapshot after it has today's layout again.
	TableDeltas bool
	// CompactDir is where a compaction job leaves range pairs for a refresh
	// to adopt (#1723): "<CompactDir>/<schema>/<table>/<chain start>/", one
	// complete range pair plus a _SUCCESS marker. Empty: nothing is adopted.
	CompactDir string
	// ChainStartFloor ends a chain of deltas that started at or before it
	// (#1904): the table is written in full again, and a new chain starts at
	// At. A reader of a table with a chain fetches events from the chain's
	// start, and a refresh that only adds a pair leaves that start where it
	// was, so on a quiet server nothing else moves it before the index drops
	// the events it needs. The daemon's refresh sets it from how far back the
	// index still keeps events; the zero value ends nothing on this ground.
	ChainStartFloor time.Time
	// MaxChainUpserts ends a table's chain of deltas once its upserts files
	// together pass it (#2210): the daemon sets half of the console's fold
	// line, so a statement over the table reads a short chain from the newest
	// copy. Zero: no such rule.
	MaxChainUpserts int64
	// MaxResolvedChainUpserts is the line, in place of MaxChainUpserts, for a
	// table ChainResolved answers for (#2261): a statement reads such a table
	// through its resolved pair (#2231), so the length of its chain is not
	// what the statement pays, and MaxChainUpserts, which does not look at
	// the table, had an 80 M row table written in full as often as a small
	// one. The share rule (tableDeltaMaxFraction) is what ends that chain
	// instead. This line stays as a ceiling for the refresh whose pair is not
	// written yet, when a statement reads the chain itself: the daemon sets
	// half of the size past which SQL on the copy answers from an earlier
	// copy. Zero, or not above MaxChainUpserts: every table keeps
	// MaxChainUpserts.
	MaxResolvedChainUpserts int64
	// ChainResolved says whether the chain beside a previous table file is
	// read through its resolved pair. Nil: no table is.
	ChainResolved func(basePath string, chain *baseline.TableDeltaChain) bool

	// WarnEventThreshold logs a loud warning when a table's fetched event count
	// exceeds it. The event window itself is PAGED since #1097, so the resident
	// cost this warns about is the change map: one entry per touched PK, which
	// paging does not bound (#1107). The event count stays the trigger because
	// it is what the fetch knows (#654).
	//
	// 0 = disabled, and it is the zero value, so a direct library caller is
	// silent unless it opts in. Both the CLI (--warn-event-threshold) and the
	// in-daemon folds (consoleapp's daemonFoldWarnEventThreshold) default it to
	// 5,000,000; keep those in step.
	//
	// The reported threshold is SCALED by Parallelism (#842), so two callers
	// with the same raw value warn at the same TOTAL concurrent event volume
	// while their per-table triggers differ.
	WarnEventThreshold int64

	// MaxTouchedRows is how many distinct changed rows a table's fold may hold
	// in memory (#1107). Past it, the merge over a baseline moves the changes to
	// disk and runs in passes (changespill.go), refusing only when one of its
	// groups alone passes this; the binlog-only fallback refuses past it. The change map holds
	// one full row image per distinct changed row until the merge, and paging
	// the fetch does not bound it; on a host that also runs capture, a fold
	// after a long outage grew past the machine's memory.
	//
	// Like WarnEventThreshold it is a per-RUN figure divided by the effective
	// parallelism, because peak memory is the sum of the maps of the tables
	// folding at once. It counts rows, not bytes: the cost of a row is its
	// width, and measured maps ran from about 4 KB to about 19 KB per row, so
	// the same count can mean a fourfold difference in memory.
	//
	// 0 = no limit, the zero value: attended CLI runs do not set it. The
	// refusal wraps ErrTouchedRowBudget.
	MaxTouchedRows int64

	// MaxChangeBytes is the same budget in bytes (#2207): past it, by an
	// estimate of what the changes take on the heap, the merge over a baseline
	// moves them to disk and sizes its passes to it. It never refuses; the
	// binlog-only fallback, which cannot spill, ignores it. Rows alone let a
	// wide table's million changed rows take 2.2 GB live (4 GB with the
	// collector's headroom) per table on a measured TPC-C-like fold. Per-RUN,
	// divided by the effective parallelism like MaxTouchedRows. 0 = no limit.
	MaxChangeBytes int64

	// RemediationHint replaces the advice attached to that warning.
	//
	// Empty uses the wording for the attended CLI commands, which names the
	// flags those commands actually register. That wording is WRONG on a daemon:
	// bintrail-console registers no --at, --parallelism or --warn-event-threshold,
	// so an operator told to lower one goes looking for a flag their binary does
	// not have. Same class as the project's rule that an MCP error must never
	// name a CLI flag the client cannot pass. A caller on such a surface sets
	// advice its own operator can act on.
	RemediationHint string

	// FetchBatchSize is the page size used to stream a table's event window
	// (#1097). 0 → query.DefaultStreamBatchSize, the zero-value convention this
	// struct already uses for ArchiveFetcher/WarnEventThreshold/DuckDBTuning.
	//
	// It trades resident memory against round trips: a page holds both decoded
	// JSON row images per event (roughly 1-2 KB per event on a narrow table),
	// while each page costs one MySQL query plus one read per archive source.
	//
	// The archive cost is not uniform, and the arithmetic matters for S3, where
	// parquetquery downloads each file rather than scanning it in place. Cursor
	// scoping advances at HOUR granularity and archive partitions are hourly,
	// so a given hour's file is re-fetched roughly (events in that hour /
	// FetchBatchSize) times — ~1x at the default for an hour holding 100k
	// events, ~10x for one holding 1M. Shrinking this is therefore free only up
	// to the point where a page stops spanning a whole busy hour.
	FetchBatchSize int

	// ArchiveFetcher fetches archived binlog events for a table. nil →
	// parquetquery.Fetch (the container-safe DuckDB budget). The CLI sets it
	// to a tuned fetcher under --ultrafast so the flag is honored on the
	// full-table path, not just single-row reconstruct (#510).
	ArchiveFetcher query.ArchiveFetcher

	// DuckDBTuning is the resource budget applied to the merge/baseline DuckDB
	// sessions this package opens directly (mergeBaselineImages,
	// materializeBaselineLocal's S3 download) — the ArchiveFetcher above only
	// covers archive Parquet reads, not these (#842). The zero value means
	// "not specified" (mirrors ArchiveFetcher==nil and WarnEventThreshold==0
	// on this same struct) and falls back to duckdbutil.DefaultTuning(), the
	// same container-safe 2-threads/4GB budget parquetquery.Fetch applies —
	// without this, each of these DuckDB instances defaulted to ~80% of host
	// RAM regardless of any --ultrafast/--duckdb-* flag the caller set. The
	// CLI sets this to the same resolved Tuning it hands ArchiveFetcher, so an
	// explicit operator flag is honored here too, not just on archive reads.
	// Under the daemon the zero value takes the job's budget from ctx first
	// (duckdbutil.WithSessionTuning, #2385), sized from the host.
	DuckDBTuning duckdbutil.Tuning
	// contains filtered or unexported fields
}

type GapVerdict added in v0.48.0 

type GapVerdict int

GapVerdict classifies the baseline↔first-event gap decision once both sides carry a comparable position anchor (the skip cases — missing baseline anchor, missing event position — are decided before this, see WarnBaselineFirstEventGap and the single-row switch in cli/reconstruct.go). Produced by DecideBaselineGap; each emitter maps a verdict to its log line. 

const (
	// GapVerdictNone means stay quiet. Either this table's first event sits
	// at-or-before the baseline anchor, or the index's earliest surviving
	// event does (query.OldestIndexedEvent) — which PROVES the index's
	// coverage begins at or before the baseline, so the space between the
	// anchor and a quiet table's first event is just "no writes to this
	// table", not a hole (#1163: a per-table first event is EXPECTED to sit
	// past the anchor on a healthy run, so it alone can never prove a gap).
	GapVerdictNone GapVerdict = iota
	// GapVerdictUnproven means coverage could NOT be proven: this table's
	// first event sits past the anchor AND the earliest surviving indexed
	// event also starts past the anchor (or is unavailable/positionless).
	// That is the honest verdict for both real shapes behind it — capture
	// genuinely started after the baseline (a real gap, #781), or older
	// events were rotated out of the live table and coverage can no longer
	// be established from what survives. The warning says what could not be
	// proven instead of asserting incompleteness.
	GapVerdictUnproven
)

func DecideBaselineGap added in v0.48.0 

func DecideBaselineGap(flavor string, bmeta baseline.DumpMetadata, first query.ResultRow, start query.IndexStart, startOK bool) GapVerdict

DecideBaselineGap decides the baseline↔first-event gap question for one table. The per-table first event alone cannot answer it: on a healthy run the first event AFTER the baseline is expected to sit at a later position (that is simply the table's next write), so comparing it against the anchor cries wolf on every healthy run (#1163). The evidence that can answer it is the index's earliest surviving event (start): if the index's coverage begins at-or-before the baseline anchor, nothing between the anchor and this table's first event can be missing — capture was already running when the baseline was taken.

The proof is deliberately one-directional. start comes from the LIVE binlog_events table only, so rotation/archival can make it LATER than the index's true coverage start; a start past the anchor therefore degrades to "cannot prove" (GapVerdictUnproven, a hedged warning), never to an assertive gap claim. stream_state.gtid_set is deliberately NOT consulted: it is seeded with the stream's START set, so it "contains" the baseline's set both when the stream started before the baseline (healthy) and when it started after it (a real gap) — see query.OldestIndexedEvent.

type MinorCompaction added in v0.84.0 

type MinorCompaction struct {
	Lo, Hi          int
	Posdel, Upserts string
	// Merged is how many pairs the range replaces.
	Merged int
}

MinorCompaction is what CompactTableDeltaMinor produced: the range pair, written under OutDir, and what it stands for.

func CompactTableDeltaMinor added in v0.84.0 

func CompactTableDeltaMinor(ctx context.Context, basePath string, chain *baseline.TableDeltaChain, lo, hi int, outDir string) (*MinorCompaction, error)

CompactTableDeltaMinor merges the pairs lo..hi of a table's chain into ONE range pair "<table>.<lo>-<hi>.{posdel,upserts}" written under outDir (#1723, the minor compaction). DuckDB only, no Go writer: the result is still a delta, so it keeps bintrail_pk / bintrail_op and the writer's encodings exactly as the pairs hold them. The posdel is the union of the pairs' dead positions; the upserts keep the newest version of every key (the same QUALIFY the state SQL applies, so the state read through the range is the state read through the pairs). The footer is the LAST merged pair's (its binlog anchor, snapshot time, chain start, base identity) with the range's two ends: a reader resuming from the range resumes where pair hi ended.

The base is never touched, and the chain is not modified here: the caller installs the range beside the base (a refresh, by linking it forward in place of the pairs) once it is complete. Files are written to a temporary name and renamed into place, so a crash leaves no half range under outDir that parses as one.

type MissingPKColumnError 

type MissingPKColumnError struct {
	Column string
}

MissingPKColumnError carries the specific column name that was absent from the baseline row, letting callers produce actionable error messages without string-parsing. Returned wrapped as `errors.Is(err, ErrPKColumnMissing)` via the Is method below.

func (*MissingPKColumnError) Error 

func (e *MissingPKColumnError) Error() string

func (*MissingPKColumnError) Is 

func (e *MissingPKColumnError) Is(target error) bool

Is returns true when target is the ErrPKColumnMissing sentinel, so `errors.Is(err, ErrPKColumnMissing)` keeps working even when the concrete type is *MissingPKColumnError.

type MydumperWriter 

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

MydumperWriter emits a full-table reconstruction as a mydumper-compatible SQL dump directory. Output layout: 

<outputDir>/
├── metadata                         # one file for the whole run
├── <db>.<table>-schema.sql          # CREATE TABLE (from baseline metadata)
├── <db>.<table>.00000.sql           # INSERT statements, chunked
├── <db>.<table>.00001.sql
└── ...

Each chunk file holds a single multi-row INSERT statement terminated by `;\n`, so concatenating `cat *.sql | mysql` works. When the accumulated chunk size passes the configured threshold the writer closes the current statement, opens the next chunk file, and starts a new INSERT.

func NewMydumperWriter 

func NewMydumperWriter(outputDir, schema, table string, cols []string, chunkSize int64) (*MydumperWriter, error)

NewMydumperWriter creates a writer for a single table. The directory must already exist; the caller is expected to pass a freshly-created output directory (e.g. from t.TempDir() or the CLI's --output).

func (*MydumperWriter) Close 

func (w *MydumperWriter) Close() error

Close terminates the in-progress INSERT (if any), flushes and closes the current chunk file, and marks the writer terminal so subsequent WriteRow calls return ErrWriterClosed. Idempotent.

func (*MydumperWriter) Discard added in v0.48.0 

func (w *MydumperWriter) Discard() error

Discard aborts the writer on a caller-side failure: it closes the current chunk file WITHOUT finalizing it and unlinks every file this writer created — the in-progress chunk, already-rotated chunks, and the schema file — so a failed table leaves no artifacts on disk (#1162). This is the counterpart Close cannot be: Close FINALIZES the current chunk (terminating ";\n", flush, keep the file), so calling it on an error path leaves a syntactically valid, loadable chunk holding a PREFIX of the table next to its schema file, with no in-file signal that it is truncated — and the mydumper-output consumers (myloader, `cat *.sql | mysql`) never read the directory-level _INCOMPLETE marker.

The writer tracks only its own table's files, so in a multi-table run Discard cannot touch a sibling table's completed output. Removal is per-file best-effort: every failure is collected and returned joined so the caller can log it WITHOUT shadowing the original error that triggered the abort. Marks the writer terminal (subsequent WriteRow returns ErrWriterClosed); idempotent, and a no-op after a Discard that already ran.

After a SUCCESSFUL Close, Discard is a structural no-op: the finalized flag makes a completed table's output undeletable regardless of what a caller's deferred error path does afterwards. A FAILED Close does not finalize, so its rotated chunks and schema file are still removed here.

func (*MydumperWriter) Files 

func (w *MydumperWriter) Files() []string

Files returns every file the writer has produced so far (schema + chunks), relative names only.

func (*MydumperWriter) Rows added in v0.49.0 

func (w *MydumperWriter) Rows() int64

Rows returns how many row tuples WriteRow has accepted. Exact by construction — `bintrail drill` compares the loaded COUNT(*) against it.

func (*MydumperWriter) SetColumnTypes added in v0.92.0 

func (w *MydumperWriter) SetColumnTypes(cols []metadata.ColumnMeta)

SetColumnTypes records which columns are MariaDB UUID/INET4/INET6, from the index's schema snapshot for the table (the same metadata the rest of the fold reads), so their values are written as X'..' (see writeMariaDBFixed). Names match case-insensitively, as they do on the server. Never derived from the CREATE TABLE text: on the binlog-only path that text can be a captured one-line statement or a placeholder the schema parser cannot read.

func (*MydumperWriter) WriteRow 

func (w *MydumperWriter) WriteRow(values []any) error

WriteRow appends one row tuple to the current chunk, rotating to a new chunk file when curBytes crosses chunkSize. values must be in the same order as the cols slice passed to NewMydumperWriter; len(values) must equal len(cols).

Returns ErrWriterClosed if the writer has already been Close()d.

func (*MydumperWriter) WriteSchema 

func (w *MydumperWriter) WriteSchema(createSQL string) error

WriteSchema emits the <db>.<table>-schema.sql file containing the exact CREATE TABLE text captured in baseline Parquet metadata. Called once, before any WriteRow.

type NoticeOnce added in v0.100.0 

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

NoticeOnce remembers which check log lines were written, so a reader that runs the check per statement (the shim, one handler per connection) logs a lasting condition (a backfilled index, a mark that names nothing) once per process instead of on every statement. One per reader, kept for the life of the process.

func (*NoticeOnce) To added in v0.100.0 

func (n *NoticeOnce) To(logger *slog.Logger) func(level slog.Level, msg string, args ...any)

To returns a ReadWindow.Notice writing each distinct line (its level, message and arguments) once, to logger (nil: the default logger).

type PsqlCopy added in v0.108.0 

type PsqlCopy struct {
	Statement          string
	DataStart, DataEnd int64
}

PsqlCopy is where a psql output file's COPY sits: Statement is the COPY (without its semicolon), and the bytes [DataStart, DataEnd) are its rows, ending in a newline unless there are none. Before DataStart is the SQL that runs ahead of it, ending in the statement's own line; from DataEnd come the end-of-data line `\.` and the SQL that runs after. Found by position, not by reading the file: a default's text can hold a line that looks like either.

type ReadWindow added in v0.100.0 

type ReadWindow struct {
	// Schema and Table name the one table the read returns.
	Schema, Table string
	// Since is the time the read starts from (the snapshot's time, or a
	// verify pair's previous snapshot); the read reaches further back than
	// it (query.PositionReadFloor), and so does the check. It is also the
	// snapshot's wall time a move to another server is counted from.
	Since time.Time
	// Until is where the read stops (AS OF, the newer snapshot of a verify
	// pair). Set together with Schema and Table, the check asks exactly what
	// the read could miss:
	//   - is there a change of Schema.Table indexed after the mark, recorded
	//     between the read's floor and Until, that sorts before the mark in
	//     the binary log (check). The read drops such a change by position.
	//     A restart after Until, or whose later changes touched only other
	//     tables, leaves this read whole.
	//   - capture moving to another server refuses only when it moved at or
	//     before Until (checkSameServerUntil).
	Until time.Time
	// Notice, when set, receives the check's log lines instead of the default
	// logger. A reader that runs the check per statement passes a NoticeOnce
	// so a lasting condition (a backfilled index, a mark that names nothing)
	// is logged once, not on every statement.
	Notice func(level slog.Level, msg string, args ...any)
	// contains filtered or unexported fields
}

ReadWindow is the read a binlog-renumbering check guards (#2174), for the readers that start at a snapshot's position: verify and the shim's _snapshot. The zero value is the refresh's check: the whole index, up to now.

type RefreshOutcome added in v0.91.0 

type RefreshOutcome struct {
	Table   string // schema.table, as the run was asked for it
	Verdict string // one of the RefreshVerdict constants
	Detail  string
}

RefreshOutcome is one table's verdict in a refresh run.

func RefreshOutcomes added in v0.91.0 

func RefreshOutcomes(tables []string, reports []*TableReport, failures []TableFailure) []RefreshOutcome

RefreshOutcomes pairs every requested table with its verdict. It is the one place the verdict is decided: the command line prints it and the web interface stores it, and two copies of the rule would drift.

The classification reads the sentinels this package exports rather than the message text: "the events are gone" and "the table changed shape" have completely different remedies, and a summary that blurs them sends the operator down the wrong one.

func (RefreshOutcome) Refused added in v0.91.0 

func (o RefreshOutcome) Refused() bool

Refused reports whether this table stopped the run: publication is all-or-nothing, so one refused table means no table was published. A skipped table did not stop anything; the run ended before reaching it.

type SingleRowVersioned added in v0.95.0 

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

SingleRowVersioned is how a single-row reconstruct reads a MariaDB system-versioned table (#2007): the declared-key value is looked up under the stored spellings (ExpandKey) and the events found are read for what they mean to the current row (Normalize), as the full-table fold reads them.

func PrepareSingleRowLookup added in v0.95.0 

func PrepareSingleRowLookup(ctx context.Context, db *sql.DB, schema, table, pk, createSQL string,
	pkMetas []metadata.ColumnMeta, pkCols []string, since, until time.Time) (*SingleRowVersioned, string, error)

PrepareSingleRowLookup is the shared single-row setup for a possibly system-versioned table (#2007): it decides how the table is read (SingleRowSysVersioning), refuses a window in which versioning changed (CheckSysVersioningChange), and returns the PKValues to search: the typed value spelled the stored way for an ordinary table, or the typed value as is for a versioned one, whose handle's ExpandKey spells every candidate.

func SingleRowSysVersioning added in v0.95.0 

func SingleRowSysVersioning(schema, table, createSQL string, pkMetas []metadata.ColumnMeta, pkCols []string, flavor string) (*SingleRowVersioned, error)

SingleRowSysVersioning decides how a single-row reconstruct reads schema.table: nil for an ordinary table, a handle for a system-versioned one, or a refusal when it cannot be read safely.

Two signals, because either can be missing: the snapshot's CREATE TABLE (empty when its metadata could not be read, or predates the embedded statement), and the schema snapshot's key (pkMetas). Reading needs the CREATE TABLE (it names the period columns); a generated key member without it refuses. With neither, nothing rules versioning out, so a MariaDB source (flavor) refuses; an unknown source only warns, since an index without a recorded source is the file-indexed shape, which predates MariaDB support for most installs; a MySQL or PostgreSQL source has no system versioning.

pkCols are the key columns the caller looks the row up by: they must be the declared key, without the period end.

func (*SingleRowVersioned) ExpandKey added in v0.95.0 

func (h *SingleRowVersioned) ExpandKey(opts *query.Options)

ExpandKey adds the stored spellings of opts' declared-key value(s), as TYPED: each value with each current marker at ROW END's position, then each of those re-spelled the way the index stores a MariaDB UUID/INET key component. In that order: spelling first would see a value one component short of the key, leave it alone, and search the text form (#2007).

func (*SingleRowVersioned) Normalize added in v0.95.0 

func (h *SingleRowVersioned) Normalize(events []query.ResultRow) ([]query.ResultRow, error)

Normalize reads the row's events for what they mean to its current version, dropping history versions, and refuses one it cannot read.

type SnapshotFullTableInput 

type SnapshotFullTableInput struct {
	// BaselinePath is the baseline Parquet path from FindBaseline — either a
	// local path or an s3:// URL (downloaded to a temp file before merge).
	BaselinePath string
	Schema       string
	Table        string
	// PKCols are the table's primary-key column metas, used to canonicalize
	// baseline PK values so they match the indexer's pk_values strings. The
	// caller is expected to have guarded every PKCol with SupportedPKType.
	PKCols []metadata.ColumnMeta
	// Changes maps pk_values string → the latest binlog event for that PK in
	// (baseline, AsOf]. Drained by the merge.
	Changes map[string]*query.ResultRow
	// Events is the RAW event slice Changes was folded from, for the
	// window-complete #782 PK-change guard. Callers fetching LimitPerPK=1 (shim
	// `_snapshot`, verify) hand a slice already collapsed to the latest event
	// per PK by the query, so their detection is bounded by that fetch; nil
	// falls back to the map-only backstop (pkChangingUpdate).
	Events []query.ResultRow
	// PGTextPK marks a PostgreSQL source: the baseline Parquet and the delta
	// pk_values BOTH store every column as raw text (pgbaseline COPY text ==
	// pgoutput text), so the PK match is string-identity — skip the MySQL
	// DATA_TYPE canonicalizer, which errors on PG's empty DATA_TYPE token. A
	// caller that sets this must have skipped the SupportedPKType precondition
	// accordingly. Zero value (false) preserves the MySQL path verbatim.
	PGTextPK bool
	// DuckDBTuning sets the resource budget for the DuckDB sessions this call
	// opens (materializeBaselineLocal, mergeBaselineImages) (#842). Zero value
	// → the container-safe default (see effectiveDuckDBTuning) — the right
	// choice for every current caller (shim, verify), which are long-lived or
	// short-lived-but-unbudgeted and were never meant to run host-greedy.
	DuckDBTuning duckdbutil.Tuning
}

SnapshotFullTableInput drives SnapshotFullTableImages.

type StaleWarning added in v0.16.1 

type StaleWarning struct {
	Message        string    // human-readable, empty when not stale
	UsingSnapshot  time.Time // snapshot the table was actually read from
	NewestSnapshot time.Time // newest eligible snapshot (lacks the table)
	// Unreadable: the newer snapshot is a folder that could not be read
	// (#1639), not one that lacks the table. A caller with a second location
	// should ask it before settling for the older snapshot.
	Unreadable bool
}

StaleWarning describes a baseline that was selected by falling back to an older snapshot because the table is absent from a newer one (#461/#466). Empty Message means "not stale" — the table is present in the newest eligible snapshot. Both the local and S3 lookups populate it identically; callers that surface staleness (the console reconstruct response, #466) read Message, while callers that only log can ignore it.

func FindBaseline 

func FindBaseline(ctx context.Context, source, schema, table string, at time.Time) (path string, snapshotTime time.Time, stale StaleWarning, err error)

FindBaseline finds the most recent baseline Parquet file at or before `at` for the given schema and table, returning its path, snapshot timestamp, and a staleness warning (#466). The warning's Message is non-empty only when the table is absent from a newer eligible snapshot, meaning the result is an older-snapshot fallback — both the local and S3 paths report it now.

source may be:

  • A local directory path (parent of RFC3339-named snapshot subdirectories)
  • An S3 URL prefix (e.g. "s3://bucket/baselines")

func (StaleWarning) Stale added in v0.16.1 

func (s StaleWarning) Stale() bool

Stale reports whether the baseline was a stale fallback.

type StateEntry 

type StateEntry struct {
	Time    time.Time      // when this state was established
	Source  string         // "baseline" | "INSERT" | "UPDATE" | "DELETE"
	EventID uint64         // 0 for the baseline entry
	GTID    string         // empty when not applicable
	State   map[string]any // nil means the row was deleted
}

StateEntry is the row state at a single point in time.

func BuildHistory 

func BuildHistory(initialState map[string]any, baselineTime time.Time, events []query.ResultRow, at time.Time) ([]StateEntry, error)

BuildHistory applies events chronologically up to at and returns a StateEntry for each state transition, starting with the baseline as the first entry. Events are folded and listed in the order given: see ApplyAt for binary log order (EventsInBinlogOrder, #2156). Like ApplyAt, it fails loud on a residual unchanged-TOAST marker (#592) in any folded event.

Same transaction-boundary caveat as ApplyAt: pass events already fetched via FetchEventsAtomic / trimmed via TrimPartialTailTransaction (#783), or a straddling transaction can appear here as two history entries (T1, T2) when it should appear as none.

type TableFailure added in v0.50.0 

type TableFailure struct {
	Schema string
	Table  string
	Err    error
}

TableFailure is one table's refusal, kept separate from the joined error so a caller can classify and report per table.

type TableReport 

type TableReport struct {
	Schema, Table string
	// SourceSnapshot is the snapshot the table's baseline was read from, a
	// local directory or an s3:// URL ("" for no baseline). The record of
	// skipped views is carried from the newest of them (#1879).
	SourceSnapshot string
	// SourceSnapshotDir is the LOCAL snapshot directory the table's baseline
	// was read from ("" for S3 or no baseline). The manifest writer takes the
	// digests of files linked forward from it instead of hashing them again
	// (#1717).
	SourceSnapshotDir string
	BaselineRows      int64 // rows streamed through from the baseline unchanged
	EventsApplied     int64 // total events observed from the event index
	// FetchDuration / FoldDuration split the event window's wall time between
	// waiting on the fetch (archive discovery and the planner included) and
	// folding pages in Go (#1720). A carried-forward table walked its window
	// too (that is how it learned it had no events), so it carries a real
	// FetchDuration and a FoldDuration of zero: no page, no fold. A window
	// empty by position, or a refusal before the fetch, leaves both zero.
	FetchDuration, FoldDuration time.Duration
	InsertsEmitted              int64 // rows appended after the baseline pass (new PKs)
	UpdatesApplied              int64 // baseline rows whose PK matched an UPDATE/INSERT event
	DeletesSkipped              int64 // baseline rows whose PK matched a DELETE event
	// RowsWritten counts the row tuples actually written into chunk files —
	// the writer's own tally, exact by construction (not derived from the
	// baseline/insert/delete counters). `bintrail drill` loads the dump and
	// checks COUNT(*) against it.
	RowsWritten int64
	Files       []string
	// PsqlCopy locates the data inside a psql output's one file, for a
	// loader that cannot hand the whole file to psql: a COPY's rows follow
	// its statement inline, which only psql's own reader splits off. Nil for
	// every other output.
	PsqlCopy *PsqlCopy
	Duration time.Duration
	// BinlogOnly is true when the table had no baseline at all and was
	// recovered entirely from the binlog (#766's ErrNoBaseline fallback,
	// reconstructBinlogOnly). Files is non-empty in this case too, so
	// ReconstructTables' shared-metadata-file selector must check this flag
	// as well — a binlog-only report has no baseline GTID/binlog coordinates
	// to embed in the metadata file.
	BinlogOnly bool
	// CarriedForward is true when the delta window held no events for this
	// table, so its previous Parquet file was published into the new snapshot
	// unchanged instead of being folded and re-emitted. The row counters above
	// are all zero in that case — nothing was streamed, applied or written —
	// and reading them as "the table is empty" would be wrong. See carryForward
	// for why an untouched table's anchor stays valid.
	CarriedForward bool
	// CarriedByLink narrows CarriedForward to the arm that shares the
	// previous file's inode — the only one that saves disk (#1578). A carried
	// table with this false was published by a full COPY (no-link filesystem,
	// a permission fault, or a cross-device output): still a reuse — no fold
	// ran, no source was read — but every byte was written again, and a
	// consumer that renders "reused" as a disk saving must count THIS, not
	// CarriedForward, or it confirms a saving the daemon log denies.
	CarriedByLink bool

	// TableDelta is true when the table was published as its previous file
	// plus a chain of deltas beside it instead of being rewritten (#1638,
	// #1718). DeltaPairWritten says whether this run wrote a pair at all (an
	// empty window over an existing chain writes none). DeltaSeq is the
	// sequence of the pair this run wrote (or of the chain's last pair when
	// none was written);
	// DeltaDeadRows and DeltaUpsertRows are what THIS run's pair holds, its
	// window and nothing accumulated; DeltaChainFiles is how many pairs the
	// chain has now and DeltaChainCopied how many of the carried pairs had to
	// be copied because a hard link failed. DeltaCompacted, when set, says why
	// a run with deltas on rewrote the table anyway. All zero with deltas off.
	TableDelta       bool
	DeltaPairWritten bool
	DeltaSeq         int
	DeltaDeadRows    int64
	DeltaUpsertRows  int64
	DeltaChainFiles  int
	DeltaChainCopied int
	DeltaCompacted   string
	// DeltaSpillPasses is how many passes this run's pair was written in when
	// the window's changes were on disk (#2126); zero for a pair written from
	// memory, and for a rewrite.
	DeltaSpillPasses int
	// LookupDuration and LookupRows are what locating the base rows the
	// window touched cost (#2496): the time spent in that lookup and the
	// base rows it examined (the matches when the keys were joined in
	// DuckDB, every base row per pass when they were scanned through Go,
	// and on the index path the index records that matched, which is the
	// positions returned: no base row is examined there at all), summed
	// over the passes of a spilled window. Zero for a rewrite and
	// for a window with no changes, which looks nothing up.
	LookupDuration time.Duration
	LookupRows     int64
	// LookupPath is the path that resolved the lookup (#2496, slice 3):
	// "index" (the key index beside the base), "join" (integer keys joined
	// in DuckDB), "scan" (every base row through Go), "mixed" when the
	// passes of a spilled window took different ones, "" when nothing was
	// looked up. LookupBlocks is the index blocks read, summed. LookupBuilt
	// is the time the lookup spent building the key index beside the base
	// because the base had none, or an unusable one (#2496, slice 4): once
	// per version of the base, so it is reported beside the refresh it hit
	// (lookup_index_built_ms in the per-table line, only when above zero).
	LookupPath   string
	LookupBlocks int64
	LookupBuilt  time.Duration
	// DeltaCompactedRange is "<lo>-<hi>" when this run adopted a compaction
	// job's range pair in place of the chain's first pairs (#1723).
	DeltaCompactedRange string

	// The refresh's arithmetic (#2505), filled by a refresh with table
	// deltas that wrote a pair or rewrote the table, and ArithmeticChecked
	// then: WindowInserts and WindowDeletes are the window's INSERT and
	// DELETE events, RowsBefore the resolved rows the window was folded
	// over and RowsAfter the rows the published state holds, -1 when the
	// refresh could not know them (newTableReport starts them there, so a
	// carried-forward table, a table of the ordinary line or a run without
	// deltas never reads as an empty table that was counted).
	// ArithmeticProblems is empty when the numbers close (or could not be
	// checked), else what did not, in the words the error line used; the
	// table is published either way.
	ArithmeticChecked            bool
	WindowInserts, WindowDeletes int64
	RowsBefore, RowsAfter        int64
	ArithmeticProblems           []string
}

TableReport carries the per-table outcome stats that the CLI summary prints, plus what the run needs from each table afterwards (SourceSnapshotDir).

func ReconstructTable 

func ReconstructTable(
	ctx context.Context,
	cfg FullTableConfig,
	schema, table string,
	db *sql.DB,
	engine *query.Engine,
	archSources []string,
	resolver *metadata.Resolver,
	dbName string,
) (*TableReport, error)

ReconstructTable is the per-table worker. Safe to call concurrently with other ReconstructTable invocations that share the same db / engine / archSources / resolver.

func ReconstructTables 

func ReconstructTables(ctx context.Context, cfg FullTableConfig) ([]*TableReport, error)

ReconstructTables runs ReconstructTable concurrently for every entry in cfg.Tables, sharing a single *sql.DB + *query.Engine + *metadata.Resolver. Returns the list of reports in arbitrary order plus a joined error containing every per-table failure (via errors.Join).

type Tail added in v0.104.5 

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

Tail is one table's tail. Not safe for concurrent use.

func NewTail added in v0.104.5 

func NewTail(schema, table, createTableSQL, basePath string, pkCols []metadata.ColumnMeta) (*Tail, error)

NewTail starts an empty tail for schema.table. createTableSQL is the CREATE TABLE in the footer of the file the table's view reads, basePath that file (local), and pkCols its key. A *TailRefusal when the table cannot have one.

func (*Tail) Apply added in v0.104.5 

func (t *Tail) Apply(page []query.ResultRow) error

Apply folds one page of the table's events, decoded as the refresh decodes them (eventDecoder). Pages may come in any order: a key keeps its last change in the binary log, which is why an event with no place in one is refused. An error refuses the table's tail from then on: the caller drops it.

func (*Tail) Bytes added in v0.113.0 

func (t *Tail) Bytes() int64

Bytes is an estimate of the heap the tail holds, from a sample of its changes (sampledChangeBytes): what a keeper bounds all tails by (#2494).

func (*Tail) Dirty added in v0.113.0 

func (t *Tail) Dirty() int

Dirty is how many keys changed since the last file was written: the rows the next WriteDelta writes.

func (*Tail) Keys added in v0.104.5 

func (t *Tail) Keys() int

Keys is how many keys the tail holds.

func (*Tail) LastEventID added in v0.104.5 

func (t *Tail) LastEventID() uint64

LastEventID is the highest event id folded, 0 before the first.

func (*Tail) PKColumns added in v0.104.5 

func (t *Tail) PKColumns() []string

PKColumns are the table's key columns, in key order, as its copy spells them.

func (*Tail) PruneThrough added in v0.111.1 

func (t *Tail) PruneThrough(anchor query.BinlogPos, afterID uint64) (int, error)

PruneThrough drops the keys a newer published cut already holds (#2483): every key whose last change the read of that cut's window would not return. That read takes an event whose start position is at or after anchor (the position where the next write begins, so an event starting there is after the cut) and, when afterID is not 0, whose id is above it (query.Options.SincePos and SinceEventID); a key stays exactly when its last change passes both, and is dropped otherwise. It returns how many keys it dropped.

A tail that held every change after an older cut, through some event at or after anchor, is then the newer cut's tail through that event. Which cuts those are is the caller's to say.

An anchor or a key's change that no order relates (no position, an LSN, another binary log base name) refuses with an error and drops nothing: the caller builds the tail again from the index.

func (*Tail) WriteDelta added in v0.113.0 

func (t *Tail) WriteDelta(path string) (rows int64, retErr error)

WriteDelta writes only the keys changed since the last file was written (#2495), in WriteFile's layout and under its rules: a segment of a chain the view reads newest first (baseline.TableTailChainStateSQL), so a follower's cycle costs the keys that changed, not the whole tail. Every segment since the last WriteFile is needed: one left out loses its keys' changes. With no key changed it writes an empty file. After PruneThrough it refuses until a WriteFile: the chain must start again over the newer cut.

A write that fails keeps the keys as changed, and the next delta carries them with whatever changed since.

func (*Tail) WriteFile added in v0.104.5 

func (t *Tail) WriteFile(path string) (rows int64, retErr error)

WriteFile writes the tail to path in the .upserts layout, one row per key: the newest image under 'u', or a tombstone under 'd' that carries the key's values and NULL elsewhere. The tail is left as it was, but for the changed keys it no longer counts (Dirty). The file is written beside path and renamed onto it, so a reader sees a whole file or none.

path must be a name no earlier version used, and an existing one is refused: the view reads the file twice in one statement (the keys it drops, the rows it adds), and a file replaced between the two would drop one version's keys and add another's rows.

type TailRefusal added in v0.104.5 

type TailRefusal struct{ Reason string }

TailRefusal says why a table has no tail. Its statements read the published copy, as they did before tails existed.

func (*TailRefusal) Error added in v0.104.5 

func (r *TailRefusal) Error() string

type UnreadableSnapshot added in v0.82.0 

type UnreadableSnapshot struct {
	SnapshotTime time.Time
	Path         string
	// Schema is the schema folder that could not be read, empty when the
	// whole snapshot folder could not be: a table of another schema cannot
	// be in it.
	Schema string
	Err    error
}

UnreadableSnapshot is a snapshot folder, or a schema folder inside one, that the local walk skipped because it could not be read. The time comes from the folder name, which stays readable when its contents do not.

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.