diff --git a/docs/copy-and-swap-design.md b/docs/copy-and-swap-design.md index f122a10a..5d9c39f0 100644 --- a/docs/copy-and-swap-design.md +++ b/docs/copy-and-swap-design.md @@ -666,7 +666,7 @@ next pass's to catch. | --- | --- | --- | | `pkg/dbconn` | Produces `TableLock`, carried by `TableLockSession`; `Confirm` is the in-transaction check every writer runs from its own connection before its first write. | LK-1 | | `pkg/preflight` | Produces `CopySwapShape` (the table facts `PreflightedTable` carries plus the v1 shape, replica identity, and dependent-object checks above, minted by `CheckCopySwapShape`) and from it `CopySwapTarget`, the copy-and-swap route's proof, minted only by `CheckCopySwapEnvironment` once the decoding and headroom facts are proven against that shape; `CheckCopySwap` runs the privilege, shape, and environment checks as one call; owns Tier-3 refusals. | ST-6, RF-1..RF-3 | -| `pkg/copier` | Produces `Chunk` and `Watermark`; `Chunker` (built only from a `CopySwapTarget`) cuts consecutive chunks that tile the whole int64 key space — first open below, last open above — so every key a row can carry belongs to exactly one chunk and a watermark at the largest value means the copy is complete. `Copier` (built from a `CopySwapTarget`, a `Shadow` — the shape `schemachange.BuiltShadow` satisfies — and the table's `TableLockSession`) copies chunks with several workers, each in its own bounded transaction under the owner's role that takes `ACCESS SHARE` on the source and `ROW EXCLUSIVE` on the shadow, confirms the lock, and confirms both relation OIDs before one frozen never-overwriting insert — the chunk statement `pkg/internal/chunksql` holds for the copier and the verifier's repair alike, and so does its `search_path`: the catalog alone for a copy without conversions, and for a copy with them the shadow build's own path (catalog first, then the source schema, then `public`), so a USING expression resolves its unqualified names in the copy as it did when the statement ran on the shadow — and the guard recomputes the `Shadow` proof's `ConversionFingerprint` (`chunksql.ResolutionFingerprint`: what those names resolve to across that path, function bodies, strictness, settings, and argument defaults included) inside each chunk transaction, before the insert and again after it before the commit, and refuses as `copy-conversion-drift` when it differs, so under `READ COMMITTED`, which every chunk transaction pins whatever the session default, a chunk that a function or operator created on the path after the build changed rolls back unless the change was undone again before the second check, which the verifier then catches; the proof check refuses conversions without a fingerprint and a fingerprint without conversions — and the rendering settings a text conversion depends on (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`), pinned in every chunk transaction so a float or `bytea` converted to `text` lands as the flush and the digest render it; a resumed copy first deletes every shadow row above the watermark in batches of the same guarded shape — the whole shadow after a zero watermark — so the resumed cut frontier and the shadow agree on what is uncut, and its first batch's guard takes the shadow in `SHARE ROW EXCLUSIVE` as its one lock on it (two fences queue rather than deadlock), so a chunk transaction of the earlier run still committing into it, or paused before its insert after confirming the lock, ends before the clear reads; the proof check refuses a shadow that is the source by name or by OID, since the clear deletes from one and the copy reads the other; the copy selects each column the `Shadow` proof names in `Conversions` as that column's USING expression — read from the statement by `statement.ColumnConversions`, since the catalog keeps no record of a USING once `ALTER COLUMN TYPE` has run — and every other column bare, so a retyped column without USING takes the server's assignment cast, and the proof check refuses a conversion of a column the copy does not carry, with no expression, or of the primary key, whose values every chunk range, the watermark, the resume clear, and the conflict target are expressed in; `Position` snapshots the cut frontier and landed watermark as two `Watermark` values plus the in-flight chunks, and `Position.Classify` is the applier's uncut / in-flight / landed rule. While `Run` is in progress the `Copier` is the `progress.Tracker`'s `WorkSource` (`Options.Tracker`): `rows_copied` is the rows this run's committed chunks inserted, `rows_total` the source's catalog row count read once at the start of the run, and `bytes_copied` / `bytes_total` the shadow's and the source's `pg_table_size` measured at each poll — every counter measured, none projected; on a resumed run the row ratio is that run's share, not the copy's completion. Each measurement, and each chunk cut, runs in a read-only transaction of its own under the chunk budgets with the catalog alone on its `search_path`, so a poll or a cut on a caller-built pool is still bounded (LK-2) and still resolves to the real catalog (CO-9); the copier stops being the source before `Run` returns. Every refusal the copier returns is a `*copier.RefusalError` carrying a `copier.RefusalCause` from the closed set in [refusal-classes.md](refusal-classes.md#copier-refusals-keyed-on-copierrefusalcause), fail-closed under `ErrInvariantViolation`; a chunk whose insert collides with a shadow row on a unique index or exclusion constraint (SQLSTATE `23505` / `23P01`) — a value the catch-up has moved between rows and the flush has yet to rewrite — is run again after a bounded linear wait (`Options.CollisionAttempts`, `Options.CollisionBackoff`; `Collisions()` counts them) and, when every attempt collides, the server's `*pgconn.PgError` is returned wrapped; any other chunk the shadow itself rejects is that error as it stands, and neither carries a cause. | CO-4, CO-9, LK-1, LK-2, LK-3 | +| `pkg/copier` | Produces `Chunk` and `Watermark`; `Chunker` (built only from a `CopySwapTarget`) cuts consecutive chunks that tile the whole int64 key space — first open below, last open above — so every key a row can carry belongs to exactly one chunk and a watermark at the largest value means the copy is complete. `Copier` (built from a `CopySwapTarget`, a `Shadow` — the shape `schemachange.BuiltShadow` satisfies — and the table's `TableLockSession`) copies chunks with several workers, each in its own bounded transaction under the owner's role that takes `ACCESS SHARE` on the source and `ROW EXCLUSIVE` on the shadow, confirms the lock, and confirms both relation OIDs before one frozen never-overwriting insert — the chunk statement `pkg/internal/chunksql` holds for the copier and the verifier's repair alike, and so does its `search_path`: the catalog alone for a copy without conversions, and for a copy with them the shadow build's own path (catalog first, then the source schema, then `public`), so a USING expression resolves its unqualified names in the copy as it did when the statement ran on the shadow — and the guard recomputes the `Shadow` proof's `ConversionFingerprint` (`chunksql.ResolutionFingerprint`: what those names resolve to across that path, function bodies, strictness, settings, and argument defaults included) inside each chunk transaction, before the insert and again after it before the commit, and refuses as `copy-conversion-drift` when it differs, so under `READ COMMITTED`, which every chunk transaction pins whatever the session default, a chunk that a function or operator created on the path after the build changed rolls back unless the change was undone again before the second check, which the verifier then catches; the proof check refuses conversions without a fingerprint and a fingerprint without conversions — and the rendering settings a text conversion depends on (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`), pinned in every chunk transaction so a float or `bytea` converted to `text` lands as the flush and the digest render it; a resumed copy first deletes every shadow row above the watermark in batches of the same guarded shape — the whole shadow after a zero watermark — so the resumed cut frontier and the shadow agree on what is uncut, and its first batch's guard takes the shadow in `SHARE ROW EXCLUSIVE` as its one lock on it (two fences queue rather than deadlock), so a chunk transaction of the earlier run still committing into it, or paused before its insert after confirming the lock, ends before the clear reads; the proof check refuses a shadow that is the source by name or by OID, since the clear deletes from one and the copy reads the other; the copy selects each column the `Shadow` proof names in `Conversions` as that column's USING expression — read from the statement by `statement.ColumnConversions`, since the catalog keeps no record of a USING once `ALTER COLUMN TYPE` has run — and every other column bare, so a retyped column without USING takes the server's assignment cast, and the proof check refuses a conversion of a column the copy does not carry, with no expression, or of the primary key, whose values every chunk range, the watermark, the resume clear, and the conflict target are expressed in; `Position` snapshots the cut frontier and landed watermark as two `Watermark` values plus the in-flight chunks, and `Position.Classify` is the applier's uncut / in-flight / landed rule. While `Run` is in progress the `Copier` is the `progress.Tracker`'s `WorkSource` (`Options.Tracker`): `rows_copied` is the rows this run's committed chunks inserted, `rows_total` the source's catalog row count read once at the start of the run, and `bytes_copied` / `bytes_total` the shadow's and the source's `pg_table_size` measured at each poll — every counter measured, none projected; on a resumed run the row ratio is that run's share, not the copy's completion. Each measurement, and each chunk cut, runs in a read-only transaction of its own under the chunk budgets with the catalog alone on its `search_path`, so a poll or a cut on a caller-built pool is still bounded (LK-2) and still resolves to the real catalog (CO-9); the copier stops being the source before `Run` returns. Every refusal the copier returns is a `*copier.RefusalError` carrying a `copier.RefusalCause` from the closed set in [refusal-classes.md](refusal-classes.md#copier-refusals-keyed-on-copierrefusalcause), fail-closed under `ErrInvariantViolation`; a chunk whose insert collides with a shadow row on a unique index or exclusion constraint (SQLSTATE `23505` / `23P01`) — a value the catch-up has moved between rows and the flush has yet to rewrite — is committed around the colliding rows: a second pass inserts every row of the chunk the shadow admits and settles the chunk with the held-off keys pending (`Position.Pending`, which `Classify` reads as the only in-flight keys of that chunk), and each later pass, after a wait one `Options.CollisionBackoff` longer than the last, inserts only the keys still pending, never the range again, until they are gone and the chunk lands (`Options.CollisionAttempts` bounds the passes; `Collisions()` counts them) and, when the pending keys outlast every pass, the server's `*pgconn.PgError` is returned wrapped with the chunk still in flight; any other chunk the shadow itself rejects is that error as it stands, and neither carries a cause. | CO-4, CO-9, LK-1, LK-2, LK-3 | | `pkg/checksum` | `Verifier` (built from a `CopySwapTarget`, a `copier.Shadow`, and the table's `TableLockSession`) compares the source with its shadow up to the copier's landed watermark and returns a `Report` of the chunks that differ. It cuts its own chunks with a `copier.Chunker`, cutting and digesting each chunk in one read-only `REPEATABLE READ` transaction, so the cut and the chunk's two digests describe one snapshot, no snapshot outlives one chunk, and the cut's lock wait is bounded like the reads (LK-2); each transaction runs the copier's guard — owner role, the copy statement's `search_path` (the catalog alone, or the shadow build's path when the proof carries conversions, so a repair's recopy resolves a USING expression as the copy did, and the proof's conversion fingerprint recomputed and refused as CO-9 when it differs, and recomputed once more after a recopy before its commit, so the source side of a digest and a recopy run the resolution the build recorded), `ACCESS SHARE` on both relations taken before the snapshot, lock confirmation, relation-OID check — and pins the rendering settings the copy, the flush, and the digest share (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`), since the digest hashes each row's text rendering and a database or role configured otherwise would render two floats that differ only in their last digits the same, or a `bytea` converted to `text` in the escape form the copy never wrote. Both sides run the same frozen statement — row count plus hex `sha256` of the key-ordered concatenation of each row's `sha256(convert_to(ROW(col::shadow_type, …)::text, 'UTF8'))` over `pk BETWEEN $1 AND $2` (SHA-256 so a FIPS-mode OpenSSL, which refuses `md5()`, does not fail the pass) — so only the data can differ, and the cast on every column is D7: a retyped column hashes as the value the shadow holds, and a column the statement retyped with `USING` hashes on the source side as that expression cast to the shadow's type, the one place the two statements differ. Only the copy columns are compared; generated columns present on both sides are not yet hashed. A `Report` proves nothing. `Check` runs the same pass under a `DivergencePolicy` the caller states every time (the zero value is refused): `abort` returns a `DivergenceError` carrying the report with the shadow untouched; `repair` replaces every differing chunk inside one guarded read-write transaction — delete the shadow's rows over every chunk's key range, then the copy statement itself for every chunk, so a unique value the source moved from one differing chunk to another lands instead of colliding with the stale row — and digests each chunk again in a fresh snapshot, returning a `RepairError` for the first that still differs alongside the `Outcome` listing every repair that committed. A repair pass assumes nothing else writes the shadow and the source rows it recopies hold still until the rereads; a write inside the pass reads as a `RepairError`, never as a second repair; `recopy` is the same recopy without the rereads, for a source that is written while the pass runs, and reports what it recopied the same way; a recopy that collides with a shadow row because the source moved a unique value from a chunk the comparison read equal into a differing chunk (SQLSTATE `23505` or `23P01`) is rolled back, the pass compares again, and the recopy cut from that comparison runs, up to `Options.RecopyAttempts` recopies, with every collision counted in `Outcome.Collisions` and the last one's error ending the pass when all collide. `ParseDivergencePolicy` lets a caller refuse a configured policy before any pass; it accepts `abort` and `repair` and refuses `recopy`, which no setting may name. `CheckLive` is the pass for the catch-up stage, where the source is written and the stream is the shadow's only other writer; it is described under [Live verification](#live-verification). Only a pass that found nothing and repaired nothing mints the proofs, whose constructors are private to the package: a `CleanWatermark` at the watermark it read through, and a `VerifiedShadow` only when that watermark is complete; `Outcome.Clean` is true only when the clean watermark was minted. A pass with repairs returns its `Repair`s and no proof; the next pass mints. While `Verify` or `Check` runs — through the repair phase as well — the verifier is the tracker's `progress.WorkSource` (`Options.Tracker`): `chunks_compared` and `rows_hashed` count the comparison's two-sided digests that have committed and the source rows they covered (every comparison of the pass, so under `recopy` a collision's re-compare counts on top of the first), `chunks_mismatched` the chunks the comparison found differing, `chunks_repaired` the chunks whose one recopy transaction has committed, and `chunks_reread` the chunks digested again after the comparison — the repaired chunks under `Check`, every settling reread under `CheckLive` — every counter read from the pass's memory, none from the database, and all reset when a pass starts; a pass started while another runs on the same verifier is refused. | CO-1, CO-2, CO-3, CO-9, LK-1, LK-2 | | `pkg/decode` | Creates, drops, and inspects the route's slot and publication under the derived name (`CreateSlot` with exported snapshot and consistent point; `DropSlot` waiting out a holder, idempotent, never crossing clusters or databases, dropping only the route's own publication; `InspectSlot` with `wal_status` and retained WAL, the latter unknown rather than zero on a server in recovery; `SlotStatus.WithinLagCeiling` judging a slot against a ceiling in bytes — a `lost` slot is `SlotLostError` before any measure, an unknown measure is `ErrSlotLagUnknown` and never within the ceiling, over it is `SlotLagExceededError` carrying both figures; `DefaultSlotLagCeiling` is one GiB; `IsEngineSlotName` tells the engine's name shape from a name merely wearing its prefix; `InspectSlot` also reads the server's account of a lost slot — `invalidation_reason`, `conflicting`, `inactive_since` — where the server version carries it, one statement across every supported major, and `SlotStatus.LostState` renders it into the `SlotLostError` so an operator reads *why* from the error). `OpenStream` decodes the slot with pgoutput from a position — the consistent point, or the applied position a resume checkpointed — on a dedicated replication connection proven to be a session of the pool's cluster on the target's database — the same table on another cluster derives the same slot name — into one `ChangeEvent` per committed row change: every column carried with its presence, so the unchanged-TOAST marker is an absent `Column` and NULL a present one — built by `PresentColumn` and `AbsentColumn`, read through `Get`, which returns the value together with its presence, so a reader takes the presence with the value — and `OldKey` for an UPDATE that moved the primary key under either admitted replica identity; every event also carries the xid of the transaction that committed it (`Xid`, taken from the transaction's BEGIN), which is how the applier tells a delivered commit the server has not yet made visible from one it has. A relation other than the target — another name, or the target's name under an OID other than the one `CopySwapTarget` was proven for, which is a table dropped and recreated under that name — a tuple that does not line up with it, or a change outside a transaction fails closed; a table whose shape changed under the stream is `ErrSourceShapeChanged`, a TRUNCATE is `ErrUnsupportedChange`, and a warning the server sends in copy-both mode — the walsender saying it will withhold changes, as from a publication it skipped loading — stops the stream fail-closed before a keepalive can deliver a position past them; the replication connection asks for warnings itself (`SET client_min_messages = warning` on the open session, the last word on the setting), so no role, database, or server setting, login event trigger, URL parameter, or connection pooler can keep them from it. The stream keeps `Delivered` — moved by a commit or by a keepalive between transactions, never past a transaction not yet yielded in full — apart from `Confirmed`, the one position it reports to the server: `Confirm` refuses a regression or a position beyond `Delivered`, and a keepalive reply carries only what the caller confirmed, so the slot's resume point moves on the caller's word alone. A wait that runs out in `Next` sends the same report, so the stream is never silent for longer than the caller's wait (`Next`'s doc says why that matters to the walsender). Both positions order transactions by their commit, so a change's own LSN can lie below either; each `ChangeEvent` carries the `Delivered` it arrived with, which is what a caller may confirm while the change is unapplied, and the server replays by commit position, so a resume never skips a replayed change for lying below its checkpoint. `ServerWALEnd` is the walsender's send position from its last keepalive and never below `Delivered`; a change does not move it, since its message carries only the change's own position. `ServerWALEnd − Delivered` is a lower bound on lag — the walsender reports how far it has decoded, not how far the server has written — so lag against the server's WAL end is measured on the pool with `pg_current_wal_lsn()`. The server ending replication with the slot intact is `ErrStreamEnded`, the clean restart ST-4 keeps apart from slot loss. | ST-3, ST-4, CO-4, CO-8 | | `pkg/applier` | `Buffer` merges events into one `Entry` per primary key — each carrying the xid of the latest event merged into it — and `Drain` judges every key against the copier's `Position` and a source `Snapshot` (`ReadSnapshot`, the server's `pg_current_snapshot()`): an entry whose transaction the snapshot does not yet show as committed — the server makes a commit visible only after its WAL flush and any synchronous-standby wait, while the walsender delivers it as soon as the WAL is flushed — is treated as in flight and stays buffered (`Batch.Invisible`), so no chunk read or completion read that cannot see the change decides its fate; the zero `Snapshot` shows nothing visible, `AllVisible` is for a drain under a lock every writer has already released. `Drain` returns a `Batch` whose `CompleteFirst` names the moved, marker-bearing images the flush must complete before it writes. `Flusher` (built from a `CopySwapTarget`, a `copier.Shadow`, and the table's `TableLockSession`) applies one batch in one guarded read-write transaction — owner role, the copy statement's `search_path` (the catalog alone, or the shadow build's path when the proof carries conversions, with the proof's conversion fingerprint recomputed and refused as CO-9 when it differs, before the writes and again after them before the commit, so a USING expression in a flush resolves as the copy's did), `ACCESS SHARE` on both relations, lock confirmation, relation-OID check, the copier's guard — bound to the lock session's context so a lost lock cancels the flush in flight. A column the statement retyped with `USING` is written through that expression (D7): the flush reads the column's source type from the catalog inside its transaction and binds the decoded value as `(SELECT FROM (SELECT $n:: AS "col") AS "source")`, so the value lands as the copy would have landed it, while a completion read from the shadow is already in the shadow's form and is bound bare; the flusher refuses, as ST-6, a proof whose conversions name the primary key, a column the copy does not carry, or an empty expression, and a converted column the source no longer carries. It first reads a completion for every `CompleteFirst` image: from the old key's shadow row, read `FOR UPDATE` so the batch's own delete of that key cannot race the read; from the source row under `Key` when `OldKeyReused` or when the old key's shadow row is absent (a move captured while the old key was uncut left no shadow row). Neither row is known to be the stream's, so the image is not written: it returns in `Result.Held` as a `HeldImage` with its completion and the `pg_current_wal_insert_lsn()` the read took in the same statement, and `Buffer.Hold` takes it back with the completion pending — an event that replaces the key's entry or reuses its old key drops the completion, an UPDATE keeps it pending behind its own values, and `Buffer.Release(passed)` fills the markers still left once the stream has delivered every change committed through the read's position, so the next `Drain` flushes the whole image (D13). An image neither row can complete is written as a delete marker at `Key`, since the source holds no row there, and its key named in `Result.Skipped`. A marker-bearing image of a row that did not move whose shadow row is absent — the source deleted the row before its chunk was read and the DELETE is still on its way — is held the same way, with no completion and `Entry.AbsenceAwaited` set, and flushed once the stream has passed the position of the check that found the row absent; an image that comes back flagged for a row still absent is a CO-8 `ErrInvariantViolation` (D13). The primary path then applies the batch column-wise inside a savepoint, every delete marker before any image: a delete marker is a `DELETE`, a whole image an `INSERT … ON CONFLICT (pk) DO UPDATE`, and a marker-bearing image of a row that did not move an `UPDATE` of only its present columns — zero rows updated is a CO-8 `ErrInvariantViolation`, never an insert that invents the omitted value. On SQLSTATE `23505` or `23P01` (a unique index or an exclusion constraint) the flush rolls back to the savepoint and reapplies the batch whole-row (D13): every remaining marker is completed from the shadow row under its own key `FOR UPDATE` (absent is a CO-8 refusal), every key in the batch is deleted in one statement, and every surviving image is inserted plain; a second collision is `ErrBatchDeferred`, and `Buffer.Requeue` puts the batch back — refusing with a CO-5 `ErrInvariantViolation` if any of its keys is held, since nothing may `Add` between `Drain` and the flush's outcome — so the caller drains again once the copier has landed the chunk the rows wait on. `Result` counts the images and deletes written and the completions, carries the held images and the skipped keys, and says whether the fallback ran. The flush pins the same rendering settings as the copy and the digest (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`) so a float or a `bytea` the decoder rendered lands, through a `USING` conversion to `text`, as the text the copy wrote and the verifier will hash, and the completion reads carry the same pin so a held value lands as the shadow renders it. `Catchup` owns the loop: it gathers the stream into the buffer, `Release`s held images at the delivered position, drains against the copier's `Position` read *after* the last `Add` (so a key the copier cuts between the read and the drain is judged in flight, never discarded), flushes, and on `ErrBatchDeferred` requeues the batch for the next pass while a chunk is in flight — once the copy has landed every key, a refusal is one no chunk can resolve, and `Run` ends with `ErrBatchDeferred` rather than requeue it forever. It confirms the lesser of the delivered position and `Buffer.OldestPending` — each keyed on the `Delivered` the event arrived with, never the event's own LSN — so the slot never advances past a change that is still buffered, held, or deferred, and never falls below what the stream had already confirmed (ST-4); the confirmation follows the requeue or hold, so it never reads a buffer emptier than the one the next drain sees. The whole loop runs under the lock session's `Bind` context, so a lost lock ends a quiet wait on the stream as it ends a flush, reported as `ErrInvariantViolation` wrapping `dbconn.ErrTableLockLost`. `Run` never returns nil: it runs once and returns the context's cause when its owner stops it, carrying any cycle error that is not the stop itself alongside. It is the `progress.WorkSource` of a catch-up step: `changes_applied`, `changes_buffered`, and `lag_bytes` measured against `pg_current_wal_lsn()` read on the pool as each cycle ends, never against the walsender's send position. Each cycle then inspects the stream's slot on the pool and judges it against `CatchupOptions.SlotLagCeiling` (zero takes `decode.DefaultSlotLagCeiling`; below one byte is refused, so the ceiling is always in force): a slot that vanished or that the server reports `lost` ends `Run` with `decode.SlotLostError`, and one retaining more WAL than the ceiling ends it with `decode.SlotLagExceededError` — the table lock stays held and the slot stays in place, so the operator reads what the slot retained and the resume path decides what follows (D11, ST-3). | CO-4, CO-5, CO-6, CO-8, LK-1, LK-2, LK-3, ST-3 | diff --git a/docs/invariants.md b/docs/invariants.md index b6fa2dcc..0ad09d0f 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -230,11 +230,17 @@ not one the copier's fence — or the verifier's repair fence — waits on; that `pkg/applier` `Buffer.Drain` — every buffered key is judged against one `Position` snapshot: landed flushes, uncut is discarded, in-flight stays buffered until a later drain finds its chunk landed, and a key-moving UPDATE's -two entries are judged per key, except that once neither is uncut the image and its old key's -entry wait together and flush together, so the flush never deletes or copies the old key's row -while an image still needs it for completion (three-way rule, discard-all before the first claim, -deferred entry keeps merging, straddling move, pair waits in either direction, uncut half judged -alone, deferral spreading through a shared old key and along a chain of pairs); `Drain` +two entries are judged per key, except that once neither is uncut the image waits for its old +key's entry — an old key in flight, or invisible, holds its image back so the flush never +deletes or copies the old key's row while the chunk that will copy it is still landing — while a +landed old key's entry flushes ahead of an image in flight for its chunk (an invisible image +still holds its old key's entry back): the delete must not wait for the image's chunk, because +when the moved row carries a unique value the chunk cannot land while the old key's shadow row +holds it, and the image, once its chunk lands, completes from the source row instead (three-way +rule, discard-all before the first claim, deferred entry keeps merging, +straddling move, pair waits for the old key, landed old key flushes ahead of its in-flight image, +uncut half judged alone, deferral spreading through a shared old key and along a chain of +pairs); `Drain` judges against a `Position` read after the last `Add`, and returns a `Batch` whose `Deferred` count is what a catch-up is waiting on the copier for and whose `CompleteFirst` names, as copies of its `Entries`, the moved, marker-bearing images the flush must complete — from the old @@ -273,16 +279,36 @@ which the change lands and the catch-up converges). *Enforced* for the copier's of a unique value that moves between rows while its chunk is in flight: the chunk's insert collides (SQLSTATE `23505`, or `23P01` for an exclusion constraint) with the shadow row that still holds the value — a landed row the flush has yet to rewrite, or a row of the same chunk -the copier itself read before the swap — and the copier runs the chunk again after a bounded -wait (`copier.Options.CollisionAttempts` × `CollisionBackoff`, linear), counting each -collision, until the flush has moved the value and the chunk lands or the attempts run out and -the server's error is returned with no cause of its own (a copier integration test holds a -colliding row across several attempts and removes it; a catch-up convergence test runs -unique-key swaps throughout the copy). *Open:* a primary-key move whose row carries a unique -secondary value, from a landed key to a key in flight: the drain's pair rule defers the old -key's delete until the chunk lands, and the chunk cannot land while the old key's row holds the -value, so the two wait on each other until the copier's attempts run out; the convergence test -for key moves starts its load after the copy has landed. *Test obligation:* a +the copier itself read before the swap — and the copier commits the chunk around the colliding +rows: a second pass inserts every row of the chunk whose values no shadow row holds and +settles the chunk with the held-off keys pending (`copier.Position.Pending`), so from then on +the applier writes every other key of the chunk as landed and keeps only the pending keys in +flight; each later pass, after a wait one `copier.Options.CollisionBackoff` longer than the one +before, inserts only the keys still pending, never the range again, so a value that moves +into the range after the second pass is the applier's to write and cannot make the chunk +collide anew. The pending keys only shrink — the flush rewrites the row holding a pending +row's value and the next pass lands it, or the source deletes the pending row and the next +pass no longer reads it — and the chunk lands when they are gone; a chunk whose pending keys +outlast `copier.Options.CollisionAttempts` passes stops the copy with the server's error and +no cause of its own, the chunk still in flight with its keys pending so a checkpoint records +none of it as copied and the resume clear removes the rows committed around them (a copier +integration test holds a colliding row, asserts the chunk commits around it with that key +alone pending, and removes the row; a catch-up convergence test runs unique-key swaps +throughout the copy; a second runs primary-key moves faster than one per backoff from before +the first chunk is read, so every move lands in the last chunk while it is in flight, and +converges within the attempts). *Enforced* for a primary-key move whose row carries a +unique secondary value, from a landed key to a key in flight: the drain flushes the landed old +key's delete ahead of the image (the pair rule above), the retried chunk's insert of the row +under the new key then lands, and the image, flushed once its chunk has landed, completes from +the source row because the old key's shadow row is gone (a drain unit test asserts the landed +old key flushes alone as a delete marker with the image deferred and flushed by the next drain; +a catch-up test holds the new key's chunk in flight by hand, asserts the old key's shadow row +is deleted while it is, and converges once the chunk lands). The passes a chunk makes stay +bounded whatever the rate of moves: a unique value that keeps arriving in a chunk's key range +from landed rows reaches it after the pass that committed around the first collision, and is +then the applier's to write, never the chunk's; only the keys held off at that pass are +retried, and the copy stops with the server's error only when one of those keys' values is +still held by another shadow row after the last pass. *Test obligation:* a marker-bearing UPDATE for a key inside an in-flight chunk asserts the flush waits for the chunk and the row is then completed from the copied shadow row, never an absent-row abort; a marker-bearing UPDATE for a key whose row the source then deleted before its chunk was read, diff --git a/pkg/applier/catchup_collision_integration_test.go b/pkg/applier/catchup_collision_integration_test.go index 077c203c..8527c746 100644 --- a/pkg/applier/catchup_collision_integration_test.go +++ b/pkg/applier/catchup_collision_integration_test.go @@ -1,6 +1,7 @@ package applier_test import ( + "errors" "fmt" "math" "sync" @@ -70,6 +71,67 @@ func TestCatchupConvergesWhenTheCopierCollidesWithAnUnflushedMove(t *testing.T) f.assertConverged(t, run.shadow) } +// keyMoveOnLanding is the copier's clock. On the first reading after the +// shadow holds a key the source still has, it gives that key a fresh +// primary key from the identity sequence, as the workload's key move does, +// once: the moved row lands in the last chunk carrying a uniq value the +// landed row's shadow copy still holds until a flush deletes it. The key is +// picked from the shadow rather than named up front because a concurrent +// workload may already have moved any given key. Every reading returns the +// wall clock. +type keyMoveOnLanding struct { + t *testing.T + f catchupFixture + // shadow is the quoted shadow table, set before the copier runs. + shadow string + + mu sync.Mutex + moved bool + err error +} + +func (m *keyMoveOnLanding) Now() time.Time { + m.mu.Lock() + defer m.mu.Unlock() + if !m.moved && m.err == nil { + m.moved, m.err = m.tryMove() + } + return time.Now() +} + +func (m *keyMoveOnLanding) tryMove() (bool, error) { + ctx := m.t.Context() + source := m.f.table.Qualified() + q := fmt.Sprintf(` + UPDATE %[1]s + SET id = nextval(pg_get_serial_sequence($1, 'id')) + WHERE id = ( + SELECT s.id + FROM %[2]s s + JOIN %[1]s src ON src.id = s.id + ORDER BY s.id + LIMIT 1 + )`, source, m.shadow) + tag, err := m.f.pool.Exec(ctx, q, source) + if err != nil { + return false, fmt.Errorf("move a landed key: %w", err) + } + return tag.RowsAffected() == 1, nil +} + +// result reports whether the move was committed. +func (m *keyMoveOnLanding) result() error { + m.mu.Lock() + defer m.mu.Unlock() + if m.err != nil { + return m.err + } + if !m.moved { + return errors.New("no key landed while the copier ran, so no key was moved") + } + return nil +} + // moveOnLanding is the copier's clock. On the first reading after the shadow // holds key landed, it commits statements against the source in one // transaction, once; each statement's %s is the source table. Every reading diff --git a/pkg/applier/catchup_integration_test.go b/pkg/applier/catchup_integration_test.go index 45501b61..5464d6b4 100644 --- a/pkg/applier/catchup_integration_test.go +++ b/pkg/applier/catchup_integration_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "math" + "sync" "testing" "time" @@ -348,14 +349,10 @@ func TestCatchupConvergesUnderMixedLoad(t *testing.T) { } // A row's primary key moves to a fresh identity value, so the change is an -// image under a new key paired with a delete under the old one. The pair -// travels together through the drain, and the flush completes the image's -// unchanged blob from the old key's row before it deletes that row (CO-4, -// D13). The workload starts after the copy has landed every key: the moved -// row keeps its unique key, so a move from a landed key to a key in flight -// collides on the copier's insert with the row still under the old key, and -// the pair rule defers that key's delete until the chunk lands — the two -// would wait on each other until the copier's collision attempts ran out. +// image under a new key paired with a delete under the old one. With the +// copy landed, both halves are landed: the pair flushes together, and the +// flush completes the image's unchanged blob from the old key's row before +// it deletes that row (CO-4, D13). func TestCatchupConvergesUnderKeyMoves(t *testing.T) { f := newCatchupFixture(t, 1500) run := startCatchup(t, f, applier.CatchupOptions{}) @@ -486,6 +483,72 @@ func TestCatchupConvergesUnderUniqueKeyMovesDuringTheCopy(t *testing.T) { f.assertConverged(t, run.shadow) } +// Primary keys move to fresh identity values from before the copy claims +// its first chunk, faster than the copier's collision backoff: every move +// lands in the last chunk, [max, +∞), under a uniq value a landed row still +// holds in the shadow until the next drain flushes its delete. The chunk's +// first attempt collides; the attempts after it commit the rows the shadow +// admits and run again only for the rows it refused, so a move that +// arrives between attempts cannot send the whole chunk back to the start, +// the chunk lands within the bound, and the shadow converges (CO-4). +// +// The moves run many times faster than the backoff but well below the +// rate at which the copier lands collided chunks, so the copy gains on the +// frontier the moves push upward and finishes; a load that mints keys +// faster than the copier lands them keeps the last chunk open for ever. +// +// Whether a random move hits a landed key before the copy reaches the last +// chunk depends on timing, so the test makes one such move itself, on the +// first landing, and holds the flushes until the copier has collided on +// it: the collision happens on every run. The load keeps moving keys +// throughout, and the flushes resume as soon as the collision is seen. +func TestCatchupConvergesUnderKeyMovesFasterThanTheCollisionBackoff(t *testing.T) { + f := newCatchupFixture(t, 1500) + move := &keyMoveOnLanding{t: t, f: f} + run := startCatchupWith(t, f, catchupSetup{copierClock: move}) + move.shadow = pgx.Identifier{f.table.Schema, run.shadow.ShadowTable()}.Sanitize() + load := testutil.StartLoad(t, f.pool, f.table, testutil.LoadSpec{ + Seed: 7, Workers: 2, RatePerSecond: 30, + Mix: testutil.Mix{KeyMove: 1}, + }) + + run.awaitStatus(t, "discard changes to keys the copier has not claimed", func(s applier.Status) bool { + return s.Discarded > 0 + }) + resume, err := run.catchup.Pause(t.Context()) + require.NoError(t, err) + t.Cleanup(resume) + + copied := make(chan error, 1) + var wg sync.WaitGroup + wg.Go(func() { copied <- run.copier.Run(t.Context(), f.pool) }) + t.Cleanup(wg.Wait) + + const collisionDeadline = 15 * time.Second + require.Eventually(t, func() bool { return run.copier.Collisions() >= 1 }, collisionDeadline, 10*time.Millisecond, + "the last chunk should collide on the uniq value the moved landed key still holds in the shadow") + require.NoError(t, move.result()) + + resume() + const copyDeadline = 30 * time.Second + select { + case err := <-copied: + require.NoError(t, err, "the copy lands every chunk while the moves keep arriving") + case <-time.After(copyDeadline): + require.FailNow(t, "the copy did not finish after the flushes resumed") + } + assert.Equal(t, copier.NewWatermark(math.MaxInt64), run.copier.Position().Cut, "every key is landed") + t.Logf("chunk passes the copier counted after a collision: %d", run.copier.Collisions()) + run.awaitStatus(t, "apply moved images and the deletes under their old keys", func(s applier.Status) bool { + return s.Applied > 0 + }) + summary := run.quiesce(t, load) + assert.Positive(t, summary.KeyMoves) + + run.stop(t) + f.assertConverged(t, run.shadow) +} + // A second Run on the same catch-up is refused: its stream is single-use. func TestCatchupRunsOnce(t *testing.T) { f := newCatchupFixture(t, 10) diff --git a/pkg/applier/catchup_stall_integration_test.go b/pkg/applier/catchup_stall_integration_test.go index 85b2fd64..d0583941 100644 --- a/pkg/applier/catchup_stall_integration_test.go +++ b/pkg/applier/catchup_stall_integration_test.go @@ -97,6 +97,48 @@ func TestCatchupRequeuesADeferredBatchAndConverges(t *testing.T) { f.assertConverged(t, run.shadow) } +// Row 7 moves to key 1000, keeping its unique value, while the chunk that +// holds key 1000 is in flight. The delete under the landed key 7 flushes +// ahead of the image: the copier's insert of the row under key 1000 would +// collide with the shadow row under key 7 for as long as that row stood, +// so holding the delete back for the image's chunk would wait on a chunk +// that is waiting on it. Once the chunk lands the image flushes alone, +// completes its unchanged blob from the source row since the shadow row +// under key 7 is gone, and the shadow converges (CO-4, D13). +func TestCatchupFlushesALandedOldKeyAheadOfItsInFlightImage(t *testing.T) { + f := newCatchupFixture(t, 20) + chunk, err := copier.NewChunk(21, math.MaxInt64) + require.NoError(t, err) + inFlight := allLanded() + inFlight.InFlight = []copier.Chunk{chunk} + positions := &switchPosition{pos: inFlight} + run := startCatchupWith(t, f, catchupSetup{positions: positions}) + + q := f.table.Qualified() + _, err = f.pool.Exec(t.Context(), `UPDATE `+q+` SET id = 1000 WHERE id = 7`) + require.NoError(t, err) + + run.awaitStatus(t, "delete the row under key 7 while key 1000's chunk is in flight", func(s applier.Status) bool { + return s.Applied == 1 && s.Deferred == 1 + }) + shadowTable := pgx.Identifier{f.table.Schema, run.shadow.ShadowTable()}.Sanitize() + var sevens int + require.NoError(t, f.pool.QueryRow(t.Context(), `SELECT count(*) FROM `+shadowTable+` WHERE id = 7`).Scan(&sevens)) + assert.Zero(t, sevens, "the shadow row under the old key is gone before the image's chunk lands") + deferred := run.status(t) + assert.Equal(t, 1, deferred.Buffered, "the image waits for its chunk") + assert.Zero(t, deferred.BatchesDeferred, "the delete flushed, it was not refused") + + positions.set(allLanded()) + end := f.currentWALLSN(t) + run.awaitStatus(t, "flush the image once its chunk has landed", func(s applier.Status) bool { + return s.Confirmed >= end && s.Buffered == 0 + }) + assert.Equal(t, uint64(2), run.status(t).Applied, "the delete and then the image") + run.stop(t) + f.assertConverged(t, run.shadow) +} + // The schema change narrows a uniquely indexed column, and after the copy // has landed every key a write gives one row a value that is distinct on // the source and collides in the shadow's type. Nothing is in flight, so no diff --git a/pkg/applier/drain.go b/pkg/applier/drain.go index b14c03ee..f088420b 100644 --- a/pkg/applier/drain.go +++ b/pkg/applier/drain.go @@ -22,9 +22,9 @@ type Batch struct { // the copier's own read of those chunks will see the change. Discarded int // Deferred counts the entries still buffered because their keys are - // inside an in-flight chunk, or travel with one that is or with one - // that is invisible. This is the number a stalled catch-up is waiting - // on the copier for. + // inside an in-flight chunk, or are moved images whose old key's entry + // is in flight or invisible, or travel with an invisible image. This is + // the number a stalled catch-up is waiting on the copier for. Deferred int // Held counts the entries still buffered because a flush held them (see // Buffer.Hold) — with a completion, or because their shadow row was @@ -103,13 +103,21 @@ func (b Batch) CompleteFirst() []Entry { // batch's OldestFirstLSN until the flush commits, and rebuilds them. // // An image that moved from an old key, and the entry still buffered at that -// old key, travel together once neither is uncut: if either is in flight or -// invisible both wait, otherwise both flush. The flush completes the image's +// old key, travel together once neither is uncut: the image waits while the +// old key's entry is in flight or invisible, and the old key's entry waits +// while the image is invisible. The flush completes the image's // unchanged-TOAST markers before it deletes anything (D13), and that reading -// must not race the deletion or copy of the old key's row; Batch.CompleteFirst -// names the images it applies to. An image whose completion is pending -// (Buffer.Hold) stays buffered whatever its key's state, until Release lets -// the completion in or an event drops it. +// must not race the copy of the old key's row or a transaction no snapshot +// shows yet; Batch.CompleteFirst names the images it applies to. A landed +// old key's entry does not wait for an image whose own chunk is in flight: +// the row's unique values are still under the old key in the shadow, and +// the chunk's insert of the row under the new key collides with them until +// that entry flushes, so holding it back would wait on a chunk that is +// waiting on it. The image, flushed once its chunk lands, finds the old +// key's shadow row gone and completes from the source row under its own key +// instead, held until the stream passes the read (D13). An image whose +// completion is pending (Buffer.Hold) stays buffered whatever its key's +// state, until Release lets the completion in or an event drops it. func (b *Buffer) Drain(pos copier.Position, snap Snapshot) Batch { // INV: CO-4 state := make(map[int64]copier.KeyState, len(b.entries)) @@ -123,7 +131,7 @@ func (b *Buffer) Drain(pos copier.Position, snap Snapshot) Batch { invisible[key] = struct{}{} } } - b.deferPairs(state) + b.deferPairs(state, invisible) var batch Batch b.hasOldest = false for key, e := range b.entries { @@ -163,13 +171,15 @@ type movePair struct { image, old int64 } -// deferPairs marks both halves of every travelling pair in flight when either -// half is. Two images may share one old key, so a deferral can spread through -// that key to another pair; the passes repeat until one changes nothing. -// Every pass that continues moves at least one key to in flight, so the work -// is bounded by the number of pairs squared, not by the buffer: the pairs are -// collected once, and a buffer with no moves costs one scan. -func (b *Buffer) deferPairs(state map[int64]copier.KeyState) { +// deferPairs marks both halves of every travelling pair in flight when the +// pair must wait together (see waitsTogether). Two images may share one old +// key, so a deferral can spread through that key to another pair; the passes +// repeat until one changes nothing. Every pass that continues moves at least +// one key to in flight, so the work is bounded by the number of pairs +// squared, not by the buffer: the pairs are collected once, and a buffer +// with no moves costs one scan. invisible names the keys whose entries are +// in flight for their transaction's visibility rather than for a chunk. +func (b *Buffer) deferPairs(state map[int64]copier.KeyState, invisible map[int64]struct{}) { var pairs []movePair for key, e := range b.entries { if e.OldKey == nil { @@ -182,7 +192,8 @@ func (b *Buffer) deferPairs(state map[int64]copier.KeyState) { for changed := true; changed; { changed = false for _, p := range pairs { - if !waitsTogether(state[p.image], state[p.old]) { + _, imageInvisible := invisible[p.image] + if !waitsTogether(state[p.image], state[p.old], imageInvisible) { continue } if state[p.image] != copier.KeyInFlight || state[p.old] != copier.KeyInFlight { @@ -195,10 +206,14 @@ func (b *Buffer) deferPairs(state map[int64]copier.KeyState) { } // waitsTogether reports whether a moved image and its old key's entry must -// both be deferred: neither is uncut, and at least one is in flight. -func waitsTogether(image, old copier.KeyState) bool { +// both be deferred: neither is uncut, and the old key's entry is in flight — +// for its chunk or for its visibility — or the image is invisible. An image +// in flight for its chunk alone does not hold a landed old key's entry back +// (see Drain): the chunk cannot land while that entry's row still holds the +// moved row's unique values in the shadow. +func waitsTogether(image, old copier.KeyState, imageInvisible bool) bool { if image == copier.KeyUncut || old == copier.KeyUncut { return false } - return image == copier.KeyInFlight || old == copier.KeyInFlight + return old == copier.KeyInFlight || imageInvisible } diff --git a/pkg/applier/drain_test.go b/pkg/applier/drain_test.go index b26b5335..3e6a0248 100644 --- a/pkg/applier/drain_test.go +++ b/pkg/applier/drain_test.go @@ -115,31 +115,49 @@ func TestDrainKeyMoveStraddlingTheFrontier(t *testing.T) { assert.Equal(t, 0, b.Len()) } -// The two halves of a move travel together once neither is uncut: a landed -// image waits for the in-flight old key, and a landed old key waits for the -// in-flight image. Neither flushes alone. -func TestDrainKeyMovePairWaitsTogether(t *testing.T) { - cases := map[string]struct { - from, to int64 - }{ - "old key in flight, image landed": {from: 15, to: 25}, - "old key landed, image in flight": {from: 25, to: 15}, - } - for name, tc := range cases { - t.Run(name, func(t *testing.T) { - b := NewBuffer() - require.NoError(t, b.Add(keyMove(10, tc.from, tc.to, col("label", "a"), marker("doc")))) - - batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) - assert.Empty(t, batch.Entries) - assert.Equal(t, 0, batch.Discarded) - assert.Equal(t, 2, batch.Deferred) - assert.Equal(t, 2, b.Len()) - - batch = b.Drain(position(t, 30), AllVisible()) - assert.Equal(t, sortedPair(tc.from, tc.to), keys(batch.Entries), "both halves flush in one batch once the chunk lands") - }) - } +// A landed image waits for its in-flight old key: the flush must not copy +// the old key's row while the image still needs it for completion, so +// neither half flushes alone, and both flush once the chunk lands. +func TestDrainKeyMovePairWaitsForTheOldKey(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(keyMove(10, 15, 25, col("label", "a"), marker("doc")))) + + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) + assert.Empty(t, batch.Entries) + assert.Equal(t, 0, batch.Discarded) + assert.Equal(t, 2, batch.Deferred) + assert.Equal(t, 2, b.Len()) + + batch = b.Drain(position(t, 30), AllVisible()) + assert.Equal(t, []int64{15, 25}, keys(batch.Entries), "both halves flush in one batch once the chunk lands") +} + +// A landed old key does not wait for its in-flight image: the delete at the +// old key flushes now, so the shadow row holding the moved row's unique +// values is gone before the chunk's insert of the row under its new key +// runs again — holding it back would wait on a chunk that is waiting on it +// (CO-4). The image stays buffered, marker intact, until its own chunk +// lands, and is then named for completion, which the flush reads from the +// source row under its key once the old key's shadow row is absent. +func TestDrainLandedOldKeyFlushesAheadOfItsInFlightImage(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "a"), marker("doc")))) + + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) + assert.Equal(t, []int64{25}, keys(batch.Entries), "the delete marker at the landed old key flushes alone") + assert.Equal(t, DeleteMarker, batch.Entries[0].Kind) + assert.Empty(t, batch.CompleteFirst(), "nothing in this batch needs the old key's row") + assert.Equal(t, 0, batch.Discarded) + assert.Equal(t, 1, batch.Deferred, "the image at 15 waits for its own chunk") + assert.Equal(t, 1, b.Len()) + image := entry(t, b, 15) + assert.Equal(t, Image, image.Kind) + assert.True(t, image.HasMarker(), "the deferred image keeps its marker for a later completion") + + batch = b.Drain(position(t, 30), AllVisible()) + assert.Equal(t, []int64{15}, keys(batch.Entries), "the image flushes once its chunk lands") + assert.Equal(t, []int64{15}, keys(batch.CompleteFirst()), "and is named for completion") + assert.Equal(t, 0, b.Len()) } // An uncut half is discarded on its own and does not hold the other back: a @@ -177,18 +195,39 @@ func TestDrainKeyMoveUncutHalfIsJudgedAlone(t *testing.T) { } // A deferral spreads through a shared old key: two images that both left key -// 25 wait when either of them is in flight, even though 25 itself has landed. +// 25 wait when the entry at 25 does, and an invisible image holds it. An +// image in flight for its chunk alone does not: the landed entry at 25 and +// the other image flush, and only the in-flight image waits. func TestDrainDeferralSpreadsThroughSharedOldKey(t *testing.T) { - b := NewBuffer() - require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "first")))) - require.NoError(t, b.Add(insert(20, 25, col("label", "reinserted")))) - require.NoError(t, b.Add(keyMove(30, 25, 28, col("label", "second")))) + t.Run("one image invisible", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(keyMove(10, 25, 15, col("label", "first")), 600))) + require.NoError(t, b.Add(inTx(insert(20, 25, col("label", "reinserted")), 650))) + require.NoError(t, b.Add(inTx(keyMove(30, 25, 28, col("label", "second")), 700))) + require.NoError(t, b.Add(inTx(insert(40, 25, col("label", "reinserted again")), 710))) + + // 710 has committed and 700 is still in progress: the entry at 25 is + // visible on its own, and only the image at 28 is not. + snap := Snapshot{read: true, xmin: 700, xmax: 711, inProgress: map[uint64]struct{}{700: {}}} + batch := b.Drain(position(t, 30), snap) + assert.Empty(t, batch.Entries, "28 is invisible, so 25 waits, so 15 waits") + assert.Equal(t, 0, batch.Discarded) + assert.Equal(t, 1, batch.Invisible) + assert.Equal(t, 2, batch.Deferred) + assert.Equal(t, 3, b.Len()) + }) + t.Run("one image in flight", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "first")))) + require.NoError(t, b.Add(insert(20, 25, col("label", "reinserted")))) + require.NoError(t, b.Add(keyMove(30, 25, 28, col("label", "second")))) - batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) - assert.Empty(t, batch.Entries, "15 is in flight, so 25 waits, so 28 waits") - assert.Equal(t, 0, batch.Discarded) - assert.Equal(t, 3, batch.Deferred) - assert.Equal(t, 3, b.Len()) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) + assert.Equal(t, []int64{25, 28}, keys(batch.Entries), "15 is in flight for its chunk; 25 and 28 are landed and flush") + assert.Equal(t, 0, batch.Discarded) + assert.Equal(t, 1, batch.Deferred) + assert.Equal(t, 1, b.Len()) + }) } // The spread is transitive however long the chain. Four rows each move onto @@ -419,10 +458,3 @@ func TestDrainAgainstTheZeroSnapshotKeepsEverything(t *testing.T) { assert.Equal(t, 2, batch.Invisible) assert.Equal(t, 2, b.Len()) } - -func sortedPair(a, b int64) []int64 { - if a < b { - return []int64{a, b} - } - return []int64{b, a} -} diff --git a/pkg/copier/copier.go b/pkg/copier/copier.go index 43259352..1578cc04 100644 --- a/pkg/copier/copier.go +++ b/pkg/copier/copier.go @@ -45,14 +45,18 @@ type Options struct { StatementTimeout time.Duration // Chunker sizes the chunks. Chunker ChunkerOptions - // CollisionAttempts is the number of times a chunk whose insert the - // server refused on a unique index or an exclusion constraint is run - // before the copy stops with that refusal. Each attempt reads the source - // afresh, so a value a source change moved between keys lands once the - // applier has flushed the move's other half into the shadow. + // CollisionAttempts is the number of passes a chunk whose insert the + // server refused on a unique index or an exclusion constraint is given + // before the copy stops with that refusal: the refused insert, then one + // pass that commits every row of the chunk but those whose values other + // shadow rows hold, then passes over only the rows held off. Each pass + // reads the source afresh, so a value a source change moved between + // keys lands once the applier has flushed the move's other half into + // the shadow. CollisionAttempts int - // CollisionBackoff is the wait before a collided chunk's second attempt; - // each later wait is one backoff longer. + // CollisionBackoff is the wait before a collided chunk's first pass over + // the rows it held off; each later wait is one backoff longer. The pass + // that commits the rest of the chunk waits for nothing. CollisionBackoff time.Duration // Clock times each chunk for the chunker's chunk-time throttling // feedback (docs/copy-and-swap-design.md#d12--throttle-by-chunk-time-and-slot-lag). @@ -135,7 +139,12 @@ type Copier struct { // sql is the one insert statement every chunk runs, frozen at // construction so no worker builds SQL, and searchPath the SET LOCAL // its transaction runs first so the statement resolves as it should. + // skipSQL and keysSQL are the statements a chunk whose insert collided + // runs instead: over its range, committing around the rows held off, + // then over the held-off keys alone. sql string + skipSQL string + keysSQL string searchPath string // clearSQL removes one batch of shadow rows above the resume watermark. clearSQL string @@ -189,6 +198,8 @@ func NewCopier(target preflight.CopySwapTarget, shadow Shadow, lock *dbconn.Tabl chunker: chunker, opts: opts, sql: copySQL(target, shadow), + skipSQL: copySkipSQL(target, shadow), + keysSQL: copyKeysSQL(target, shadow), searchPath: copySearchPath(shadow), clearSQL: clearAboveSQL(target, shadow), ledger: newLedger(from), @@ -245,12 +256,16 @@ func (c *Copier) Position() Position { // and continues, so a caller can back off and resume behind a busy table. // // A chunk whose insert the server refuses on a unique index or an -// exclusion constraint (SQLSTATE 23505 or 23P01) is run again after a wait, -// up to Options.CollisionAttempts times: the value it collided on was moved -// between keys by a source change the applier has yet to flush, and a later -// attempt's snapshot sees the shadow after that flush. A collision no -// attempt outlasts returns the server's error wrapped, the chunk in flight, -// and Collisions counts every refused attempt. +// exclusion constraint (SQLSTATE 23505 or 23P01) is given up to +// Options.CollisionAttempts passes: the value it collided on was moved +// between keys by a source change the applier has yet to flush. The next +// pass commits every row of the chunk but those whose values other shadow +// rows hold, and the chunk's Position entry then names those keys as +// Pending, so the applier writes the rest of the chunk while later passes, +// each after a wait, copy only the held-off rows once the flush has +// rewritten the rows holding their values. A collision no pass outlasts +// returns the server's error wrapped, the chunk in flight with its pending +// keys, and Collisions counts every pass that left a row held off. // // Cancellation reaches the server as a closed connection, so a statement // that was running keeps running until it finishes or hits the transaction's @@ -321,13 +336,13 @@ func (c *Copier) finish(cause error) error { } // work is one worker's loop: claim a chunk, copy it in its own transaction -// (again, after a wait, when the insert collided with a shadow row the -// applier has yet to clear), land it, and tell the chunker how long the -// committed attempt took. It returns nil when the chunker has no chunk left -// and the first error otherwise; a cancelled context ends the loop with the -// context's error, which Run resolves to the cancellation's cause. A chunk -// whose copy failed is left in flight: it did not land, so its keys must -// not read as landed. +// (in further passes, around and then after a wait for the rows whose +// values collided with shadow rows the applier has yet to clear), land it, +// and tell the chunker how long the pass that committed the range took. It +// returns nil when the chunker has no chunk left and the first error +// otherwise; a cancelled context ends the loop with the context's error, +// which Run resolves to the cancellation's cause. A chunk whose copy failed +// is left in flight: it did not land, so its keys must not read as landed. func (c *Copier) work(ctx context.Context, pool *pgxpool.Pool) error { for ctx.Err() == nil { chunk, ok, err := c.claim(ctx, pool) @@ -382,3 +397,16 @@ func (c *Copier) land(chunk Chunk, inserted int64) error { } return nil } + +// settle records that a chunk has committed inserted rows in all and still +// owes the rows under pending, so Position classifies its other keys as +// landed while the chunk stays in flight (CO-4). +func (c *Copier) settle(chunk Chunk, inserted int64, pending []int64) error { + c.mu.Lock() + defer c.mu.Unlock() + // INV: CO-4 + if !c.ledger.settle(chunk, inserted, pending) { + return refuse(CauseChunksInFlight, nil, "chunk [%d, %d] settled with %d rows pending was not in flight", chunk.Lower(), chunk.Upper(), len(pending)) + } + return nil +} diff --git a/pkg/copier/copy_chunk.go b/pkg/copier/copy_chunk.go index 29e10c2a..25d57fed 100644 --- a/pkg/copier/copy_chunk.go +++ b/pkg/copier/copy_chunk.go @@ -41,17 +41,35 @@ const ( // copyChunk copies one chunk in its own guarded transaction: it inserts the // chunk's live rows into the shadow without overwriting any row already -// there and returns the number of rows the insert added. A copy that -// evaluates conversion expressions confirms their resolution once more -// after the insert and before the commit (CO-9), so a catalog change that -// committed between the guard's check and the insert rolls the chunk back. -// The transaction is READ COMMITTED whatever the server or role -// configures, so that repeated read takes a snapshot after the insert's; -// the insert is one statement and relies on no snapshot beyond its own. +// there and returns the number of rows the insert added. The insert is one +// statement and relies on no snapshot beyond its own. func (c *Copier) copyChunk(ctx context.Context, pool *pgxpool.Pool, chunk Chunk) (int64, error) { + var tag pgconn.CommandTag + err := c.chunkTx(ctx, pool, chunk, func(tx pgx.Tx) error { + var err error + tag, err = tx.Exec(ctx, c.sql, chunk.Lower(), chunk.Upper()) + if err != nil { + return fmt.Errorf("copy chunk [%d, %d] of %s.%s into %s: %w", chunk.Lower(), chunk.Upper(), c.shadow.Schema(), c.shadow.SourceTable(), c.shadow.ShadowTable(), err) + } + return nil + }) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} + +// chunkTx runs write, one chunk's insert into the shadow, in its own guarded +// transaction and commits it. A copy that evaluates conversion expressions +// confirms their resolution once more after the write and before the +// commit (CO-9), so a catalog change that committed between the guard's +// check and the write rolls the chunk back. The transaction is READ +// COMMITTED whatever the server or role configures, so that repeated read +// takes a snapshot after the write's. +func (c *Copier) chunkTx(ctx context.Context, pool *pgxpool.Pool, chunk Chunk, write func(tx pgx.Tx) error) error { tx, err := pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted, AccessMode: pgx.ReadWrite}) if err != nil { - return 0, fmt.Errorf("begin copy of chunk [%d, %d]: %w", chunk.Lower(), chunk.Upper(), err) + return fmt.Errorf("begin copy of chunk [%d, %d]: %w", chunk.Lower(), chunk.Upper(), err) } defer func() { // Redundant safety closer: after a successful Commit this returns @@ -60,19 +78,18 @@ func (c *Copier) copyChunk(ctx context.Context, pool *pgxpool.Pool, chunk Chunk) _ = tx.Rollback(context.WithoutCancel(ctx)) }() if err := c.guard(ctx, tx, lockRowExclusive); err != nil { - return 0, err + return err } - tag, err := tx.Exec(ctx, c.sql, chunk.Lower(), chunk.Upper()) - if err != nil { - return 0, fmt.Errorf("copy chunk [%d, %d] of %s.%s into %s: %w", chunk.Lower(), chunk.Upper(), c.shadow.Schema(), c.shadow.SourceTable(), c.shadow.ShadowTable(), err) + if err := write(tx); err != nil { + return err } if err := c.confirmResolution(ctx, tx); err != nil { - return 0, err + return err } if err := tx.Commit(ctx); err != nil { - return 0, fmt.Errorf("commit chunk [%d, %d] of %s.%s: %w", chunk.Lower(), chunk.Upper(), c.target.Schema(), c.target.Table(), err) + return fmt.Errorf("commit chunk [%d, %d] of %s.%s: %w", chunk.Lower(), chunk.Upper(), c.target.Schema(), c.target.Table(), err) } - return tag.RowsAffected(), nil + return nil } // guard prepares the transaction every write into the shadow runs in: it diff --git a/pkg/copier/copy_pending.go b/pkg/copier/copy_pending.go new file mode 100644 index 00000000..ab54f01d --- /dev/null +++ b/pkg/copier/copy_pending.go @@ -0,0 +1,66 @@ +package copier + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/block/pg-sprite/pkg/internal/chunksql" + "github.com/block/pg-sprite/pkg/preflight" +) + +// copySkipping copies one chunk whose strict insert collided, in its own +// guarded transaction, leaving out every source row whose unique value a +// shadow row still holds under another key. It returns the rows it inserted +// and the ascending keys it held off — the rows the chunk still owes, which +// copyKeys lands once the catch-up has rewritten the rows holding their +// values (CO-4). +func (c *Copier) copySkipping(ctx context.Context, pool *pgxpool.Pool, chunk Chunk) (int64, []int64, error) { + var inserted int64 + var held []int64 + err := c.chunkTx(ctx, pool, chunk, func(tx pgx.Tx) error { + if err := tx.QueryRow(ctx, c.skipSQL, chunk.Lower(), chunk.Upper()).Scan(&inserted, &held); err != nil { + return fmt.Errorf("copy chunk [%d, %d] of %s.%s into %s around the values other shadow rows hold: %w", chunk.Lower(), chunk.Upper(), c.shadow.Schema(), c.shadow.SourceTable(), c.shadow.ShadowTable(), err) + } + return nil + }) + if err != nil { + return 0, nil, err + } + return inserted, held, nil +} + +// copyKeys copies the source rows under keys, the keys a chunk's earlier pass +// held off, in its own guarded transaction, again leaving out any row whose +// value a shadow row still holds. It reads only those keys: a source row +// that arrived in the chunk since the pass that held them off is the +// catch-up's to write, as the rest of the chunk has landed. It returns the +// rows it inserted and the keys still held off, a subset of keys. +func (c *Copier) copyKeys(ctx context.Context, pool *pgxpool.Pool, chunk Chunk, keys []int64) (int64, []int64, error) { + var inserted int64 + var held []int64 + err := c.chunkTx(ctx, pool, chunk, func(tx pgx.Tx) error { + if err := tx.QueryRow(ctx, c.keysSQL, keys).Scan(&inserted, &held); err != nil { + return fmt.Errorf("copy the %d held-off rows of chunk [%d, %d] of %s.%s into %s: %w", len(keys), chunk.Lower(), chunk.Upper(), c.shadow.Schema(), c.shadow.SourceTable(), c.shadow.ShadowTable(), err) + } + return nil + }) + if err != nil { + return 0, nil, err + } + return inserted, held, nil +} + +// copySkipSQL is the statement a collided chunk runs over its whole range, +// built for this target and shadow. +func copySkipSQL(target preflight.CopySwapTarget, shadow Shadow) string { + return chunksql.InsertSkipping(shadow.Schema(), shadow.SourceTable(), shadow.ShadowTable(), target.PKColumn(), shadow.CopyColumns(), shadow.Conversions()) +} + +// copyKeysSQL is the statement a collided chunk runs over the keys it still +// owes, built for this target and shadow. +func copyKeysSQL(target preflight.CopySwapTarget, shadow Shadow) string { + return chunksql.InsertKeys(shadow.Schema(), shadow.SourceTable(), shadow.ShadowTable(), target.PKColumn(), shadow.CopyColumns(), shadow.Conversions()) +} diff --git a/pkg/copier/doc.go b/pkg/copier/doc.go index 7b4cfb5e..ecbc763c 100644 --- a/pkg/copier/doc.go +++ b/pkg/copier/doc.go @@ -3,11 +3,11 @@ // that tile the whole int64 range, so every key belongs to exactly one chunk // (CO-4) and each copy statement stays bounded (LK-3). A Copier copies those // chunks with several workers — one guarded, never-overwriting insert per -// chunk, run again after a bounded wait when it collides with a shadow row -// the catch-up has yet to rewrite — first clearing the shadow above the -// watermark when it resumes, and -// reports a Position whose Classify is the applier's rule for a captured key: -// uncut, in flight, or landed. A column the Shadow proof names in +// chunk; a chunk that collides with a shadow row the catch-up has yet to +// rewrite commits around the rows held off and copies those alone, after +// bounded waits, once the row is gone — first clearing the shadow above the +// watermark when it resumes, and reports a Position whose Classify is the +// applier's rule for a captured key: uncut, in flight, or landed. A column the Shadow proof names in // Conversions is copied through that USING expression; the copier does not // decide whether the statement behind the proof may run — the router's // availability check is that gate, and a caller that drives the copier diff --git a/pkg/copier/ledger.go b/pkg/copier/ledger.go index d5738f48..9d8469f7 100644 --- a/pkg/copier/ledger.go +++ b/pkg/copier/ledger.go @@ -25,13 +25,26 @@ type ledger struct { // landed holds chunks that committed above the watermark and wait for // the chunks below them. landed []Chunk - rows int64 + // partial holds, for an in-flight chunk that has committed part of + // itself, the rows it has landed so far and the keys it has yet to land. + partial map[Chunk]partialChunk + // rows counts every row a committed pass inserted, the partial chunks' + // included. + rows int64 +} + +// partialChunk is what an in-flight chunk has committed so far and what it +// still owes: pending is never empty, since a chunk with nothing pending +// lands. +type partialChunk struct { + rows int64 + pending []int64 } // newLedger resumes bookkeeping after from: every key at or below it counts // as landed, and the first chunk to extend the watermark starts just above it. func newLedger(from Watermark) *ledger { - l := &ledger{cut: from, watermark: from} + l := &ledger{cut: from, watermark: from, partial: make(map[Chunk]partialChunk)} l.next, l.complete = startAfter(from) return l } @@ -64,18 +77,40 @@ func (l *ledger) startsAtFrontier(chunk Chunk) bool { return chunk.lower == l.cut.Value()+1 } -// land records that a claimed chunk committed rows rows and advances the -// watermark over every landed chunk now contiguous with it. It reports false -// when the chunk was not in flight. A chunk whose transaction did not commit -// is never landed: it stays in flight, so the applier keeps deferring for its -// keys and the watermark cannot pass it. +// settle records that a claimed chunk has committed rows rows in all so far +// and still owes the rows under pending, which must be non-empty and +// ascending. The chunk stays in flight — the watermark cannot pass it, and a +// resumed run clears and re-copies it — but a Position now classifies its +// keys outside pending as landed, so the applier writes them while the +// copier waits for the rows holding the pending keys' values to be +// rewritten (CO-4). It reports false, recording nothing, for a chunk that +// is not in flight, and for an empty pending: a chunk owing nothing lands. +func (l *ledger) settle(chunk Chunk, rows int64, pending []int64) bool { + if !slices.Contains(l.inFlight, chunk) { + return false + } + if len(pending) == 0 { + return false + } + l.rows += rows - l.partial[chunk].rows + l.partial[chunk] = partialChunk{rows: rows, pending: slices.Clone(pending)} + return true +} + +// land records that a claimed chunk committed rows rows in all — the rows +// of every settled pass included — and advances the watermark over every +// landed chunk now contiguous with it. It reports false when the chunk was +// not in flight. A chunk whose transaction did not commit is never landed: +// it stays in flight, so the applier keeps deferring for its keys and the +// watermark cannot pass it. func (l *ledger) land(chunk Chunk, rows int64) bool { i := slices.Index(l.inFlight, chunk) if i < 0 { return false } l.inFlight = slices.Delete(l.inFlight, i, i+1) - l.rows += rows + l.rows += rows - l.partial[chunk].rows + delete(l.partial, chunk) l.landed = append(l.landed, chunk) l.advance() return true @@ -102,13 +137,20 @@ func (l *ledger) advance() { } } -// position snapshots the ledger. The in-flight chunks are copied so the -// caller can read them without the copier's lock. +// position snapshots the ledger. The in-flight chunks and the pending keys +// are copied so the caller can read them without the copier's lock; the +// pending entries follow the in-flight order, so they are ascending too. func (l *ledger) position() Position { - return Position{ + pos := Position{ Watermark: l.watermark, Cut: l.cut, InFlight: slices.Clone(l.inFlight), RowsInserted: l.rows, } + for _, chunk := range l.inFlight { + if partial, ok := l.partial[chunk]; ok { + pos.Pending = append(pos.Pending, PendingKeys{Chunk: chunk, Keys: slices.Clone(partial.pending)}) + } + } + return pos } diff --git a/pkg/copier/ledger_test.go b/pkg/copier/ledger_test.go index 6d8ccc1d..35e6a8bd 100644 --- a/pkg/copier/ledger_test.go +++ b/pkg/copier/ledger_test.go @@ -127,6 +127,47 @@ func TestLedgerUnlandedChunkStaysInFlight(t *testing.T) { assert.Equal(t, int64(10), l.position().RowsInserted, "a refused landing counts no rows") } +// A settled chunk stays in flight — the watermark waits for it — but its +// Position names the keys it still owes, its committed rows count as +// inserted, and each later settlement replaces the earlier one's rows and +// keys rather than adding to them; landing it counts its rows once in all +// and drops the pending entry. A settlement of a chunk not in flight, or +// with nothing pending, is refused and records nothing. +func TestLedgerSettledChunkStaysInFlightWithItsPendingKeys(t *testing.T) { + l := newLedger(Watermark{}) + first := mustChunk(t, math.MinInt64, 10) + second := mustChunk(t, 11, 20) + require.True(t, l.claim(first)) + require.True(t, l.claim(second)) + require.True(t, l.land(second, 10)) + + require.True(t, l.settle(first, 7, []int64{3, 5})) + settled := l.position() + assert.Equal(t, []Chunk{first}, settled.InFlight, "a settled chunk is still in flight") + assert.Equal(t, []PendingKeys{{Chunk: first, Keys: []int64{3, 5}}}, settled.Pending) + assert.False(t, settled.Watermark.Valid(), "the watermark waits for the pending keys") + assert.Equal(t, int64(17), settled.RowsInserted, "the rows committed around the pending keys count") + + require.True(t, l.settle(first, 8, []int64{5})) + shrunk := l.position() + assert.Equal(t, []PendingKeys{{Chunk: first, Keys: []int64{5}}}, shrunk.Pending) + assert.Equal(t, int64(18), shrunk.RowsInserted, "a later settlement's total replaces the earlier one's") + + require.True(t, l.land(first, 9)) + landed := l.position() + assert.Empty(t, landed.InFlight) + assert.Empty(t, landed.Pending) + assert.Equal(t, NewWatermark(20), landed.Watermark) + assert.Equal(t, int64(19), landed.RowsInserted, "landing counts the chunk's rows once in all") + + assert.False(t, l.settle(first, 1, []int64{1}), "a landed chunk cannot settle") + assert.False(t, l.settle(mustChunk(t, 21, 30), 1, []int64{21}), "an unclaimed chunk cannot settle") + require.True(t, l.claim(mustChunk(t, 21, 30))) + assert.False(t, l.settle(mustChunk(t, 21, 30), 1, nil), "a chunk owing nothing lands instead") + assert.Equal(t, int64(19), l.position().RowsInserted, "a refused settlement counts no rows") + assert.Empty(t, l.position().Pending) +} + // Resuming after a watermark treats every key at or below it as landed and // continues the contiguous prefix from just above it. func TestLedgerResumesAfterTheWatermark(t *testing.T) { diff --git a/pkg/copier/position.go b/pkg/copier/position.go index 4f508701..e0994e79 100644 --- a/pkg/copier/position.go +++ b/pkg/copier/position.go @@ -1,5 +1,7 @@ package copier +import "slices" + // KeyState is where a primary key stands relative to the copy: the CO-4 // discard rule the applier judges every captured change by. type KeyState int @@ -10,13 +12,16 @@ const ( // read will see the change. KeyUncut KeyState = iota // KeyInFlight lies in a chunk a worker is reading now, or in a chunk - // whose transaction did not commit before the run failed. A change for - // it must wait for the chunk to land: applied earlier, a stale copy could + // whose transaction did not commit before the run failed, or is one of + // the keys an in-flight chunk has committed everything but. A change for + // it must wait for the key to land: applied earlier, a stale copy could // resurrect a deleted row or the applier could read a shadow row that is // not there yet. A resumed run re-copies such a chunk. KeyInFlight // KeyLanded lies in a chunk whose transaction committed (or that a - // resumed run had already copied). A change for it applies immediately. + // resumed run had already copied), or outside the pending keys of an + // in-flight chunk that has committed everything else. A change for it + // applies immediately. KeyLanded ) @@ -48,14 +53,29 @@ type Position struct { // empty once Run has returned nil; after a failed Run it lists the chunks // whose transactions did not commit, so their keys never read as landed. InFlight []Chunk + // Pending lists, in the order of InFlight, the in-flight chunks that have + // committed every row but a few, with the keys still to copy. Such a + // chunk stays in InFlight until its keys empty, so the watermark cannot + // pass it, while its keys outside Keys classify as landed. + Pending []PendingKeys // RowsInserted counts rows the copier inserted into the shadow, excluding // rows the applier had already written (the insert never overwrites). RowsInserted int64 } +// PendingKeys names the keys an in-flight chunk still has to copy after a +// collision made the copier commit the rest of the chunk around them. +type PendingKeys struct { + Chunk Chunk + // Keys are the ascending source keys whose shadow slot another row held + // when the chunk was read. + Keys []int64 +} + // Classify reports where key stands in this snapshot. A key above the cut // frontier is uncut whatever the watermark says; a key at or below it is in -// flight when a claimed chunk covers it and landed otherwise. +// flight when a claimed chunk covers it — unless the chunk has committed +// around a few pending keys, when only those are — and landed otherwise. func (p Position) Classify(key int64) KeyState { // INV: CO-4 if !p.Cut.Valid() || key > p.Cut.Value() { @@ -63,8 +83,24 @@ func (p Position) Classify(key int64) KeyState { } for _, chunk := range p.InFlight { if chunk.lower <= key && key <= chunk.upper { - return KeyInFlight + return p.classifyWithin(chunk, key) } } return KeyLanded } + +// classifyWithin reports where a key inside an in-flight chunk stands: in +// flight, unless the chunk has committed everything but its pending keys and +// this key is not one of them. +func (p Position) classifyWithin(chunk Chunk, key int64) KeyState { + for _, pending := range p.Pending { + if pending.Chunk != chunk { + continue + } + if _, held := slices.BinarySearch(pending.Keys, key); held { + return KeyInFlight + } + return KeyLanded + } + return KeyInFlight +} diff --git a/pkg/copier/position_test.go b/pkg/copier/position_test.go index 2a08f2bb..1844fe89 100644 --- a/pkg/copier/position_test.go +++ b/pkg/copier/position_test.go @@ -36,6 +36,32 @@ func TestPositionClassifiesKeysAgainstTheCutFrontier(t *testing.T) { assert.Equal(t, KeyUncut, nothingCut.Classify(0)) } +// An in-flight chunk that has committed around a few pending keys still +// defers those keys, and only those: every other key of the chunk has +// landed, while a chunk without a pending entry defers all of its keys. +func TestPositionClassifiesOnlyThePendingKeysOfASettledChunkAsInFlight(t *testing.T) { + settled := mustChunk(t, 11, 20) + unsettled := mustChunk(t, 21, 30) + pos := Position{ + Cut: NewWatermark(30), + InFlight: []Chunk{settled, unsettled}, + Pending: []PendingKeys{{Chunk: settled, Keys: []int64{12, 15}}}, + } + cases := map[int64]KeyState{ + 11: KeyLanded, + 12: KeyInFlight, + 13: KeyLanded, + 15: KeyInFlight, + 20: KeyLanded, + 21: KeyInFlight, + 30: KeyInFlight, + 31: KeyUncut, + } + for key, want := range cases { + assert.Equal(t, want, pos.Classify(key), "key %d", key) + } +} + func TestKeyStateStrings(t *testing.T) { assert.Equal(t, "uncut", KeyUncut.String()) assert.Equal(t, "in-flight", KeyInFlight.String()) diff --git a/pkg/copier/retry.go b/pkg/copier/retry.go index bd9a89dd..049e7466 100644 --- a/pkg/copier/retry.go +++ b/pkg/copier/retry.go @@ -19,15 +19,15 @@ const ( ) const ( - // DefaultCollisionAttempts is the number of times a chunk whose insert - // collided is run before the copy stops, when Options leaves + // DefaultCollisionAttempts is the number of passes a chunk whose insert + // collided is given before the copy stops, when Options leaves // CollisionAttempts zero. DefaultCollisionAttempts = 10 - // DefaultCollisionBackoff is the wait before a collided chunk's second - // attempt when Options leaves CollisionBackoff zero. Each later wait is - // one backoff longer, so the default attempts together outlast many - // catch-up cycles, and a cycle's flush is what moves the colliding - // value out of the chunk's way. + // DefaultCollisionBackoff is the wait before a collided chunk's first + // pass over the rows it held off when Options leaves CollisionBackoff + // zero. Each later wait is one backoff longer, so the default passes + // together outlast many catch-up cycles, and a cycle's flush is what + // moves the colliding value out of the held-off rows' way. DefaultCollisionBackoff = 250 * time.Millisecond ) @@ -43,48 +43,115 @@ func collided(err error) bool { return pgErr.Code == sqlstateUniqueViolation || pgErr.Code == sqlstateExclusionViolation } -// copyUntilClear runs chunk's copy until it commits, returning the rows the -// committed attempt inserted and how long that attempt took, or until the -// copy must stop. A collision is a row the chunk read from the source whose -// unique value a shadow row still holds under another key: a source change -// moved the value between the keys while this chunk was in flight, and the -// applier's flush has yet to rewrite the landed row the value came from, or -// the chunk read both rows before the move and the flush has since given a -// landed row the value the chunk carries (CO-4). Each attempt is its own -// transaction with its own snapshot, so once the flush has committed a later -// attempt reads the moved rows as the source now has them and lands. A -// collision no attempt outlasts ends the copy with the -// server's error, wrapped with the chunk's bounds and the attempts made; -// the chunk stays in flight. Nothing else is retried here: a row the shadow -// rejects on its own terms, a lock wait that timed out, a lost connection -// is the first attempt's error, returned as such. +// copyUntilClear copies chunk until every row of it has committed, returning +// the rows the passes inserted in all and how long the pass that committed +// the range took, or until the copy must stop. A collision is a row the +// chunk read from the source whose unique value a shadow row still holds +// under another key: a source change moved the value between the keys while +// this chunk was in flight, and the applier's flush has yet to rewrite the +// landed row the value came from, or the chunk read both rows before the +// move and the flush has since given a landed row the value the chunk +// carries (CO-4). The first pass is the strict insert, which collides as a +// whole; it serves the chunk that collides on nothing, the common case, at +// one statement. Nothing else is retried here: a row the shadow rejects on +// its own terms, a lock wait that timed out, a lost connection is the first +// pass's error, returned as such. func (c *Copier) copyUntilClear(ctx context.Context, pool *pgxpool.Pool, chunk Chunk) (int64, time.Duration, error) { - for attempt := 1; ; attempt++ { - started := c.opts.Clock.Now() - inserted, err := c.copyChunk(ctx, pool, chunk) - if err == nil { - return inserted, c.opts.Clock.Now().Sub(started), nil + started := c.opts.Clock.Now() + inserted, err := c.copyChunk(ctx, pool, chunk) + if err == nil { + return inserted, c.opts.Clock.Now().Sub(started), nil + } + if !collided(err) { + return 0, 0, err + } + c.collisions.Add(1) + return c.copyAroundCollisions(ctx, pool, chunk, err) +} + +// copyAroundCollisions finishes a chunk whose strict insert collided, in +// passes numbered after it up to Options.CollisionAttempts. The first pass +// commits every row of the chunk but those whose values other shadow rows +// hold and settles the chunk with those keys pending, so the applier writes +// the rest of the chunk from then on; each later pass, after a wait one +// backoff longer than the one before, copies only the rows still pending. +// Each pass is its own transaction with its own snapshot, so once the flush +// has rewritten the row holding a pending row's value the next pass lands +// it, and a pending row the source has since deleted is simply no longer +// read; the pending keys only ever shrink, and the chunk lands when they +// are gone. A pass the server refuses as a whole, as a deferred unique +// constraint refuses at commit, is run again as it was. A collision no pass +// outlasts ends the copy with the server's error, wrapped with the chunk's +// bounds and the passes made; the chunk stays in flight with its keys +// pending. +func (c *Copier) copyAroundCollisions(ctx context.Context, pool *pgxpool.Pool, chunk Chunk, collision error) (int64, time.Duration, error) { + var total int64 + var took time.Duration + // pending is nil until the pass over the whole range commits. + var pending []int64 + for pass := 2; ; pass++ { + if pass == c.opts.CollisionAttempts+1 { + return 0, 0, fmt.Errorf("chunk [%d, %d] collided with a shadow row on each of %d attempts and still has rows pending whose values other shadow rows hold: %w", chunk.Lower(), chunk.Upper(), c.opts.CollisionAttempts, collision) } - if !collided(err) { + if err := c.waitBeforePass(ctx, pass); err != nil { return 0, 0, err } - c.collisions.Add(1) - if attempt == c.opts.CollisionAttempts { - return 0, 0, fmt.Errorf("chunk [%d, %d] collided with a shadow row on each of %d attempts; the row holding its value was not removed between them: %w", chunk.Lower(), chunk.Upper(), attempt, err) + started := c.opts.Clock.Now() + inserted, held, err := c.copyPending(ctx, pool, chunk, pending) + if err != nil { + if !collided(err) { + return 0, 0, err + } + collision = err + c.collisions.Add(1) + continue + } + if pending == nil { + took = c.opts.Clock.Now().Sub(started) + } + total += inserted + if len(held) == 0 { + return total, took, nil } - select { - case <-ctx.Done(): - return 0, 0, ctx.Err() - case <-time.After(c.opts.CollisionBackoff * time.Duration(attempt)): + pending = held + if err := c.settle(chunk, total, pending); err != nil { + return 0, 0, err } + c.collisions.Add(1) + } +} + +// copyPending runs one pass of a collided chunk: over the whole range while +// no pass has committed around the held-off rows yet (pending nil), and +// over the pending keys alone after that. +func (c *Copier) copyPending(ctx context.Context, pool *pgxpool.Pool, chunk Chunk, pending []int64) (int64, []int64, error) { + if pending == nil { + return c.copySkipping(ctx, pool, chunk) + } + return c.copyKeys(ctx, pool, chunk, pending) +} + +// waitBeforePass waits before a collided chunk's pass: nothing before the +// pass that commits around the held-off rows, since it needs no change in +// the shadow to make progress, and one backoff more before each pass after +// it, since those wait on the catch-up's flush. +func (c *Copier) waitBeforePass(ctx context.Context, pass int) error { + if pass == 2 { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(c.opts.CollisionBackoff * time.Duration(pass-2)): + return nil } } -// Collisions is the number of chunk attempts this copier's run has had the -// server refuse on a unique index or an exclusion constraint, each followed -// by a wait and another attempt or by the copy stopping. It is safe to read -// from any goroutine while Run is in progress; a copy whose source saw no -// value move during it reads zero. +// Collisions is the number of chunk passes this copier's run has had the +// server refuse on a unique index or an exclusion constraint, or has had to +// leave rows held off from, each followed by another pass or by the copy +// stopping. It is safe to read from any goroutine while Run is in progress; +// a copy whose source saw no value move during it reads zero. func (c *Copier) Collisions() int64 { return c.collisions.Load() } diff --git a/pkg/copier/retry_integration_test.go b/pkg/copier/retry_integration_test.go index 42116d8c..713b133a 100644 --- a/pkg/copier/retry_integration_test.go +++ b/pkg/copier/retry_integration_test.go @@ -59,14 +59,16 @@ func (f copierFixture) holdSlotAtFirstChunk(t *testing.T, shadow schemachange.Bu } // A chunk whose insert collides with a shadow row on the unique slot index -// is run again after a wait, and lands once the colliding row is gone: here -// a shadow row under another key holds slot 15, as it would after a source -// change moved the slot from the row the copier read to a landed key the -// applier has yet to flush the move's other half for (CO-4). Each attempt is -// its own snapshot, so the attempt after the row's removal copies the chunk -// as the source has it, every row lands, and the committed attempt alone -// counts as inserted. -func TestCopierRetriesAChunkThatCollidesUntilTheRowIsGone(t *testing.T) { +// commits around the colliding row and lands the rest once that row is +// gone: here a shadow row under another key holds slot 15, as it would +// after a source change moved the slot from the row the copier read to a +// landed key the applier has yet to flush the move's other half for (CO-4). +// After the first collision the shadow holds every row of the chunk but +// key 15, the chunk stays in flight with key 15 pending — so a captured +// change for key 15 still waits while one for key 16 applies — and the pass +// after the row's removal copies key 15 alone, so every row lands and each +// row counts as inserted once. +func TestCopierCommitsAroundACollisionAndLandsTheRestWhenTheRowIsGone(t *testing.T) { f := newCopierFixture(t) const rows = 300 target, lock, shadow := f.prepareSeats(t, rows) @@ -85,29 +87,38 @@ func TestCopierRetriesAChunkThatCollidesUntilTheRowIsGone(t *testing.T) { wg.Go(func() { results <- c.Run(t.Context(), f.pool) }) t.Cleanup(wg.Wait) - const collisionDeadline = 15 * time.Second - require.Eventually(t, func() bool { return c.Collisions() >= 1 }, collisionDeadline, 10*time.Millisecond, - "the first chunk, carrying key 15, should collide on the slot the shadow row holds") + const settleDeadline = 15 * time.Second + require.Eventually(t, func() bool { return len(c.Position().Pending) == 1 }, settleDeadline, 10*time.Millisecond, + "the first chunk, carrying key 15, should collide on the slot the shadow row holds and commit around it") placed(t) pos := c.Position() - require.Len(t, pos.InFlight, 1, "the colliding chunk stays in flight between attempts") - assert.Equal(t, int64(math.MinInt64), pos.InFlight[0].Lower()) - assert.False(t, pos.Watermark.Valid(), "nothing has landed while the first chunk collides") + require.Len(t, pos.InFlight, 1, "the colliding chunk stays in flight while key 15 is pending") + first := pos.InFlight[0] + assert.Equal(t, int64(math.MinInt64), first.Lower()) + assert.Equal(t, []copier.PendingKeys{{Chunk: first, Keys: []int64{15}}}, pos.Pending) + assert.Equal(t, copier.KeyInFlight, pos.Classify(15), "the held-off key still waits") + assert.Equal(t, copier.KeyLanded, pos.Classify(16), "the rest of the chunk has landed") + assert.False(t, pos.Watermark.Valid(), "the watermark waits for the pending key") + assert.Equal(t, first.Rows()-1, f.count(t, shadow.ShadowTable(), fmt.Sprintf("id <= %d", first.Upper())), "every row of the chunk but key 15 is in the shadow") + assert.Equal(t, int64(0), f.count(t, shadow.ShadowTable(), "id = 15")) f.exec(t, "DELETE FROM "+shadowName+" WHERE id = 2000") const stopDeadline = 30 * time.Second require.NoError(t, awaitRun(t, results, stopDeadline, "copy after the colliding row's removal")) f.assertConverged(t, shadow) - assert.Equal(t, int64(rows), c.Position().RowsInserted, "only the committed attempt's rows count as inserted") + assert.Equal(t, int64(rows), c.Position().RowsInserted, "each row counts as inserted once across the passes") assert.Equal(t, copier.NewWatermark(math.MaxInt64), c.Position().Watermark) - assert.GreaterOrEqual(t, c.Collisions(), int64(1)) + assert.Empty(t, c.Position().Pending) + assert.GreaterOrEqual(t, c.Collisions(), int64(2), "the strict insert and the pass that held key 15 off each count") } -// A collision no attempt outlasts stops the copy with the server's own +// A collision no pass outlasts stops the copy with the server's own // refusal: the error is the unique violation wrapped, not a refusal class of -// the copier's and not a transient dbconn would retry, every attempt is -// counted, and the chunk stays in flight with nothing landed, so a caller -// that checkpoints Position records no key as copied. +// the copier's and not a transient dbconn would retry, every pass is +// counted, and the chunk stays in flight with its pending key, so a caller +// that checkpoints Position records no key of it as copied and a resumed +// run re-copies the chunk; the rows committed around the pending key stay +// in the shadow, where the resume clear removes them. func TestCopierStopsWhenACollisionOutlastsEveryAttempt(t *testing.T) { f := newCopierFixture(t) target, lock, shadow := f.prepareSeats(t, 300) @@ -136,8 +147,11 @@ func TestCopierStopsWhenACollisionOutlastsEveryAttempt(t *testing.T) { pos := c.Position() require.Len(t, pos.InFlight, 1, "the exhausted chunk never landed") - assert.Equal(t, int64(math.MinInt64), pos.InFlight[0].Lower()) + first := pos.InFlight[0] + assert.Equal(t, int64(math.MinInt64), first.Lower()) + assert.Equal(t, []copier.PendingKeys{{Chunk: first, Keys: []int64{15}}}, pos.Pending, "the held-off key is still pending") assert.False(t, pos.Watermark.Valid(), "nothing landed") - assert.Equal(t, int64(0), pos.RowsInserted) - assert.Equal(t, int64(0), f.count(t, shadow.ShadowTable(), "id <= 300"), "no attempt's rows survive its rollback") + assert.Equal(t, first.Rows()-1, pos.RowsInserted, "the rows committed around the pending key count as inserted") + assert.Equal(t, int64(0), f.count(t, shadow.ShadowTable(), "id = 15"), "the pending key's row never reached the shadow") + assert.Equal(t, first.Rows()-1, f.count(t, shadow.ShadowTable(), "id <= 300"), "the rest of the chunk stays in the shadow for the resume clear to remove") } diff --git a/pkg/internal/chunksql/chunksql.go b/pkg/internal/chunksql/chunksql.go index 12a5bd78..b721ff7d 100644 --- a/pkg/internal/chunksql/chunksql.go +++ b/pkg/internal/chunksql/chunksql.go @@ -1,11 +1,13 @@ -// Package chunksql holds the one statement that puts a chunk of source rows -// into the shadow, and the search_path and output settings it runs under. -// The copier runs the statement for every chunk it lands and the verifier -// runs it for every chunk it repairs; one text in one place is what makes -// a repair put back exactly what a copy would have, and keeps the statement -// off the public API, where a caller could run it outside the guard both -// packages wrap around it. The applier's flush shares the path and the -// settings, since it evaluates the same conversion expressions. +// Package chunksql holds the statements that put a chunk of source rows +// into the shadow, and the search_path and output settings they run under. +// The copier runs Insert for every chunk it lands and the verifier runs it +// for every chunk it repairs; one text in one place is what makes a repair +// put back exactly what a copy would have, and keeps the statement off the +// public API, where a caller could run it outside the guard both packages +// wrap around it. InsertSkipping and InsertKeys are the copier's statements +// for a chunk whose Insert collided; they select the rows exactly as Insert +// does. The applier's flush shares the path and the settings, since it +// evaluates the same conversion expressions. package chunksql import ( @@ -35,24 +37,80 @@ import ( // are schema-qualified; the expressions are the user's text, which is why // SearchPath puts the source schema on the path they resolve in. func Insert(schema, source, shadow, key string, columns []string, conversions map[string]string) string { - targets := make([]string, 0, len(columns)) - selected := make([]string, 0, len(columns)) + targets, selected := columnLists(columns, conversions) + pk := pgx.Identifier{key}.Sanitize() + return "INSERT INTO " + pgx.Identifier{schema, shadow}.Sanitize() + + " (" + targets + ")" + + " SELECT " + selected + + " FROM " + pgx.Identifier{schema, source}.Sanitize() + + " WHERE " + pk + " BETWEEN $1::bigint AND $2::bigint" + + " ON CONFLICT (" + pk + ") DO NOTHING" +} + +// InsertSkipping is the statement the copier runs over a chunk whose Insert +// collided: the same rows, selected the same way, inserted with ON CONFLICT +// DO NOTHING on every unique index and exclusion constraint rather than on +// the key alone, so a source row whose value a shadow row still holds under +// another key is left out instead of failing the chunk. It returns one row: +// the number of rows it inserted, and the ascending bigint keys of the +// source rows it read that are in the shadow neither now nor from before — +// the rows still held off, which the copier runs InsertKeys for once the +// catch-up has rewritten the rows holding their values (CO-4). Every part of +// the statement runs on the one snapshot its start took, so the shadow it +// reads for "from before" is the shadow before this statement's own +// inserts, which the insert's RETURNING accounts for instead. +func InsertSkipping(schema, source, shadow, key string, columns []string, conversions map[string]string) string { + pk := pgx.Identifier{key}.Sanitize() + return insertPending(schema, source, shadow, pk, columns, conversions, pk+" BETWEEN $1::bigint AND $2::bigint") +} + +// InsertKeys is InsertSkipping over the source rows whose key is in the +// bigint array $1 rather than over a key range: the copier's statement for +// the keys an earlier pass left held off, and only those, since a source row +// that arrived in the chunk after that pass is the catch-up's to write. +func InsertKeys(schema, source, shadow, key string, columns []string, conversions map[string]string) string { + pk := pgx.Identifier{key}.Sanitize() + return insertPending(schema, source, shadow, pk, columns, conversions, pk+" = ANY($1::bigint[])") +} + +// insertPending renders the skipping insert over the source rows where +// matches. The source rows are read once into a materialized CTE so the +// insert and the held-off query see the same rows; the insert's RETURNING +// names what landed, and a read row that neither landed nor was already in +// the shadow is held off. +func insertPending(schema, source, shadow, pk string, columns []string, conversions map[string]string, where string) string { + targets, selected := columnLists(columns, conversions) + shadowName := pgx.Identifier{schema, shadow}.Sanitize() + return "WITH src AS MATERIALIZED (" + + "SELECT " + selected + + " FROM " + pgx.Identifier{schema, source}.Sanitize() + + " WHERE " + where + ")," + + " ins AS (INSERT INTO " + shadowName + " (" + targets + ")" + + " SELECT " + targets + " FROM src" + + " ON CONFLICT DO NOTHING RETURNING " + pk + ")" + + " SELECT (SELECT count(*) FROM ins)," + + " ARRAY(SELECT src." + pk + "::bigint FROM src" + + " WHERE NOT EXISTS (SELECT 1 FROM ins WHERE ins." + pk + " = src." + pk + ")" + + " AND NOT EXISTS (SELECT 1 FROM " + shadowName + " sh WHERE sh." + pk + " = src." + pk + ")" + + " ORDER BY src." + pk + ")" +} + +// columnLists renders the insert's column list and the matching select +// list: every column quoted, a converted column selected as its expression +// aliased to the column it lands in. +func columnLists(columns []string, conversions map[string]string) (targets, selected string) { + targetList := make([]string, 0, len(columns)) + selectedList := make([]string, 0, len(columns)) for _, column := range columns { quoted := pgx.Identifier{column}.Sanitize() - targets = append(targets, quoted) + targetList = append(targetList, quoted) if expr, ok := conversions[column]; ok { - selected = append(selected, expr+" AS "+quoted) + selectedList = append(selectedList, expr+" AS "+quoted) continue } - selected = append(selected, quoted) + selectedList = append(selectedList, quoted) } - pk := pgx.Identifier{key}.Sanitize() - return "INSERT INTO " + pgx.Identifier{schema, shadow}.Sanitize() + - " (" + strings.Join(targets, ", ") + ")" + - " SELECT " + strings.Join(selected, ", ") + - " FROM " + pgx.Identifier{schema, source}.Sanitize() + - " WHERE " + pk + " BETWEEN $1::bigint AND $2::bigint" + - " ON CONFLICT (" + pk + ") DO NOTHING" + return strings.Join(targetList, ", "), strings.Join(selectedList, ", ") } // Rendering pins, for the rest of the transaction, the two output settings diff --git a/pkg/internal/chunksql/chunksql_test.go b/pkg/internal/chunksql/chunksql_test.go index 802f069f..4b04dd97 100644 --- a/pkg/internal/chunksql/chunksql_test.go +++ b/pkg/internal/chunksql/chunksql_test.go @@ -1,6 +1,7 @@ package chunksql import ( + "strings" "testing" "github.com/stretchr/testify/assert" @@ -36,6 +37,39 @@ func TestInsertCarriesConversions(t *testing.T) { assert.Equal(t, want, Insert("app", "orders", "_pgsprite_orders_new", "id", []string{"id", "select", "qty"}, conversions)) } +// The skipping insert reads the chunk's rows once, inserts them with no +// conflict target so a unique or exclusion collision skips the row instead +// of failing the statement, and reports the keys of the rows it read that +// landed neither now nor earlier. The select list is Insert's: a converted +// column is read as its expression, so a held-off row lands later exactly +// as Insert would have landed it. +func TestInsertSkippingIsFrozen(t *testing.T) { + want := `WITH src AS MATERIALIZED (SELECT "id", "select", qty::int AS "qty" FROM "app"."orders"` + + ` WHERE "id" BETWEEN $1::bigint AND $2::bigint),` + + ` ins AS (INSERT INTO "app"."_pgsprite_orders_new" ("id", "select", "qty")` + + ` SELECT "id", "select", "qty" FROM src ON CONFLICT DO NOTHING RETURNING "id")` + + ` SELECT (SELECT count(*) FROM ins),` + + ` ARRAY(SELECT src."id"::bigint FROM src` + + ` WHERE NOT EXISTS (SELECT 1 FROM ins WHERE ins."id" = src."id")` + + ` AND NOT EXISTS (SELECT 1 FROM "app"."_pgsprite_orders_new" sh WHERE sh."id" = src."id")` + + ` ORDER BY src."id")` + conversions := map[string]string{"qty": "qty::int"} + assert.Equal(t, want, InsertSkipping("app", "orders", "_pgsprite_orders_new", "id", []string{"id", "select", "qty"}, conversions)) +} + +// The keyed insert is the skipping insert over an explicit key list: the +// one place the two differ is the WHERE clause, so a held-off row is +// selected, inserted, and reported the same way in both. +func TestInsertKeysDiffersFromInsertSkippingOnlyInItsRange(t *testing.T) { + columns := []string{"id", "select", "qty"} + skipping := InsertSkipping("app", "orders", "_pgsprite_orders_new", "id", columns, nil) + keyed := InsertKeys("app", "orders", "_pgsprite_orders_new", "id", columns, nil) + assert.Contains(t, skipping, ` WHERE "id" BETWEEN $1::bigint AND $2::bigint)`) + assert.Contains(t, keyed, ` WHERE "id" = ANY($1::bigint[]))`) + replaced := strings.Replace(skipping, ` WHERE "id" BETWEEN $1::bigint AND $2::bigint)`, ` WHERE "id" = ANY($1::bigint[]))`, 1) + assert.Equal(t, replaced, keyed) +} + // Without conversions the statement names nothing the search_path // resolves, so the path is the catalog alone; with them the user's // expressions resolve on the build's path, the source schema then public.