@spec checkpoint_column_probe() :: String.t()

Query probing the commit_lsn column's data_type. The table name is bound $1 (a string value in information_schema, not an identifier position), so no interpolation and no validation are needed here — the caller has already validated the table for the interpolating builders above.

@spec checkpoint_ensure_table(String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

DDL creating the lib-owned checkpoint table if absent. slot_name is the PK, one row per slot; commit_lsn is a bigint (the Replicant.lsn/0 integer — no pg_lsn text parse at the boundary). The table name is a validated identifier; IF NOT EXISTS is a name check only, so the caller MUST also shape-probe (see checkpoint_column_probe/0).

@spec checkpoint_read(String.t()) :: {:ok, String.t()} | {:error, :invalid_identifier}

Query reading commit_lsn for a slot. slot_name is bound $1; only the validated table is interpolated.

@spec checkpoint_upsert(String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Upsert of commit_lsn for a slot. slot_name/commit_lsn are bound $1/$2; only the validated table is interpolated.

@spec configured_table_info([{String.t(), String.t()}]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

The relation-info catalog read for the :pglogical and :wal2json decoders (ADR-0009 §5): per configured table, its pg_class oid, relreplident, and every non-dropped column with type oid + typmod — everything the wire does NOT carry (pglogical relations have no types and no replica identity; wal2json has no relation message at all). Runs in the replication connect chain's SIMPLE protocol; both parts of every pair are Identifier.validated before interpolation. A configured table with no rows here is absent on the server — the caller halts fail-closed. Row shape: [schemaname, tablename, oid, relreplident, attnum, attname, atttypid, atttypmod].

@spec create_durable_slot(String.t(), boolean()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Replication command to create a durable logical slot with NO exported snapshot. failover?: false (the published default) emits the legacy NOEXPORT_SNAPSHOT keyword byte-for-byte; failover?: true emits the PG17 parenthesized (FAILOVER, SNAPSHOT 'nothing') grammar (spec §6). The caller has already gated failover-on-PG16 at connect, so failover? == true implies PG17.

@spec create_durable_slot(String.t(), boolean(), String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

create_durable_slot/2 with the decoder's output plugin (ADR-0009 §2). The default pgoutput emits the published 1.3.0 string byte-for-byte; pglogical_output and wal2json name their plugin. The plugin name passes Identifier.validate/1 before interpolation exactly like a slot name (Critical Rule 2).

@spec create_durable_slot(String.t(), boolean(), String.t(), non_neg_integer()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

create_durable_slot/3 with the server version: NOEXPORT_SNAPSHOT (and the parenthesized snapshot grammar) exist from PG 15 — on 9.6 to 14 the durable-slot form is the bare CREATE_REPLICATION_SLOT … LOGICAL plugin; (the server exports a snapshot, which is simply discarded). PG15+ — and an unknown version (0, a directly-constructed state) — emits the published strings unchanged.

@spec create_export_slot(String.t(), boolean()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Replication command to create a durable logical slot that EXPORTS a consistent snapshot (spec §4). failover?: false emits the legacy EXPORT_SNAPSHOT keyword byte-for-byte; failover?: true emits (FAILOVER, SNAPSHOT 'export') (spec §6). The result row is [slot_name, consistent_point, snapshot_name, output_plugin] in both forms.

@spec create_export_slot(String.t(), boolean(), String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

create_export_slot/2 with the decoder's output plugin (see create_durable_slot/3).

@spec create_export_slot(String.t(), boolean(), String.t(), non_neg_integer()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

create_export_slot/3 with the server version: EXPORT_SNAPSHOT (and the parenthesized snapshot grammar) exist from PG 15 — on 9.6 to 14 the form is the bare CREATE_REPLICATION_SLOT … LOGICAL plugin; (the server exports the snapshot by default; the result row shape is unchanged). The plugin-decoder + snapshot: true combination on 9.6/12 rides exactly this pre-15 form.

@spec decoder_option_probe(String.t(), [{String.t(), String.t()}]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

ADR-0009 §4 — the pre-flight output-plugin option probe, against the pipeline's OWN slot with the decoder's EXACT option list. A build that cannot express an option rejects the call with invalid_parameter_value / feature_not_supported (OBSERVED: wal2json format bounds use 0A000, an unknown option uses 22023) — the caller halts :decoder_option_unsupported instead of feeding the stream-command reconnect loop (postgrex surfaces a rejected START_REPLICATION only as a disconnect, and a rejected or erroring peek on the WALSENDER itself never returns ReadyForQuery). Any OTHER error — pglogical's "produces binary output" (XX000) above all — proves the options PARSED, so the caller proceeds. Option values are validated exactly like start_replication_for/4's; names are Identifier.validated and interpolated (the simple-query protocol cannot bind).

@spec identify_system() :: String.t()

The replication-protocol command that identifies the exact source session.

@spec is_in_recovery() :: String.t()

Query returning pg_is_in_recovery() — true on a standby (spec §8 R-ISO advisory).

@spec keyless_scan(String.t(), [String.t()]) :: String.t()

Whole-table scan for the PK-less fallback (spec §6.4). Every column is projected as <col>::text AS <col> (the SAME cast projection as keyset_chunk/4) so a PK-less table's snapshot rows converge with the stream too (the record values are cast; there is no keyset bound). col_quoted is server-quoted (table_columns/0), never raw.

@spec keyset_chunk(String.t(), [String.t()], [String.t()], non_neg_integer()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Keyset chunk SELECT (spec §6.6). col_quoted (all columns) and pk_quoted (the PK subset) come from table_columns/0/pk_columns/0 (server-quoted — never validated client-side, never raw). bound_arity is the number of PK columns in the resume bound: 0 emits the first-chunk form (no WHERE); n > 0 emits a ROW() comparison whose bounds are BIND PARAMETERS $2..$n+1 ($1 is the LIMIT) — a literal bound would be simultaneously an injection surface and a Rule-1 leak into logged SQL text. An empty PK list is a caller error.

EVERY column is projected as <col>::text AS <col> so each delivered value is the type's text output — byte-identical to pgoutput text, which the stream casts through the SAME Casting.Types.cast_record/2 path (spec §2 convergence). The WHERE/ORDER BY reference the REAL typed PK columns — TABLE-QUALIFIED (<qualified>.<pk>, NOT bare <pk>) so they bind to the typed table columns, NEVER the same-named ::text OUTPUT ALIAS: a bare ORDER BY <pk> resolves to the projected alias (SQL lets ORDER BY see output names), which would sort keyset pages LEXICOGRAPHICALLY (1,10,100,…) and silently skip rows.

@spec pk_columns() :: String.t()

Query returning, per publication table, its ordered PRIMARY KEY columns — BOTH the raw attname (to read %Change{}.record string keys) and the server-quoted form via quote_ident (the ONLY form ever interpolated into keyset SQL — spec §6.6, Critical Rule 2; the format('%I') precedent). Publication name is bound $1. A publication table with NO primary key returns no row here (PK-less fallback, spec §6.4). Row shape: [schemaname, tablename, qualified, pk_raw, pk_quoted].

Per-column TYPE oids come from table_columns/0 (the full column set already covers the PK columns), so no per-PK type array is duplicated here.

@spec pk_columns_for([{String.t(), String.t()}]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

The pk_columns/0 row shape for an EXPLICIT table list (the pglogical/wal2json decoders' incremental path, ADR-0009 §8). Same columns and PK ordering; the table set is the caller's validated list.

@spec progress_ensure_table(String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

DDL creating the lib-owned snapshot-progress table if absent (token is an opaque bytea; spec §6.2).

@spec progress_read(String.t()) :: {:ok, String.t()} | {:error, :invalid_identifier}

Query reading the progress token for a slot. slot_name bound $1; only the validated table is interpolated.

@spec progress_upsert(String.t()) :: {:ok, String.t()} | {:error, :invalid_identifier}

Upsert of the progress token for a slot. slot_name/token bound $1/$2.

@spec publication_exists([String.t()]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Query returning the pubnames that exist for any name in the validated list (spec §5.3 / decision #18). The caller halts fail-closed if the returned set ≠ the requested set — START_REPLICATION with a missing publication silently streams the existing subset, so it CANNOT be the fail-closed gate (probe-verified).

Unlike publication_tables/1 / pk_columns/0 / table_columns/0 (which run via Postgrex.query!/3 and bind $1), this query runs in the Postgrex.ReplicationConnection connect chain, whose {:query, sql, state} dispatch uses the SIMPLE query protocol — which cannot bind $1. So the validated names are interpolated into a single-quoted IN (...) list, the exact precedent slot_invalidation_status/2 sets with slot_name. Injection-safe per Critical Rule 2: every name passed Identifier.validate/1 ([a-z_][a-z0-9_]{0,62}) BEFORE reaching here — the allowlist excludes ', ), ;, and every other breakout character, so the interpolated literal cannot escape the IN (...) position.

@spec publication_tables([String.t()]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Query returning each DISTINCT publication table's schemaname, tablename, and PG-quoted fully-qualified name across the publication LIST (spec §5.3 / decision #19). The DISTINCT collapses the pubname dimension BEFORE it fans the join, so a table in two publications yields ONE row (not two). The publication names are bound $1 (an array), so no interpolation.

@spec recovery_and_version() :: String.t()

Connect-time probe: pg_is_in_recovery() (standby detection, spec §8) plus the numeric server version (server_version_num, e.g. 170010) in one round trip on the replication connection. The version gates the PG17+ invalidation columns (spec §5) and the FAILOVER slot grammar (spec §6). No identifier to validate — a constant query.

@spec replication_set_exists([String.t()]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

The :publication_check existence gate for the :pglogical decoder: the replication sets that exist in pglogical's own catalog for the validated names. Runs in the replication connect chain's SIMPLE protocol, so the validated names are interpolated exactly as publication_exists/1 does (Critical Rule 2). A missing set halts fail-closed at connect — pglogical silently streams the intersection otherwise.

@spec replication_set_tables([String.t()]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

The :pglogical decoder's table-set discovery (ADR-0009 §2): the DISTINCT tables across the validated replication-set list, from pglogical's own membership catalog (OBSERVED columns: relid, nspname, relname, set_name — the names ride the membership row itself, no catalog joins needed). Same row shape as publication_tables/1 (schemaname, tablename, qualified).

@spec set_transaction_snapshot(term()) ::
  {:ok, String.t()} | {:error, :invalid_snapshot_name}

Command adopting an exported snapshot by name (spec §9). The name is validated as a snapshot-name LITERAL — not an identifier — before interpolation into the quoted string; a name with a quote/whitespace/other injection returns {:error, :invalid_snapshot_name} and builds nothing.

@spec slot_confirmed_flush(String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Query returning a reused logical slot's confirmed_flush_lsn in the current database. This is one input to the origin a go-forward stream resumes at, exposed to append consumers via handle_slot_origin/2 (R04). Runs in the replication connect chain's SIMPLE query protocol (no $1 bind), so the Identifier-validated slot name is interpolated — the slot_invalidation_status/2 / slot_exists/1 precedent (Critical Rule 2). A missing row or NULL is not a logical-slot origin and is rejected fail-closed by the connection.

@spec slot_exists(String.t()) :: {:ok, String.t()} | {:error, :invalid_identifier}

Query returning the active flag for the replication slot.

@spec slot_invalidation_status(String.t(), non_neg_integer()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Query returning the slot's invalidation signals (spec §5/§8), gated by the numeric server version because the available pg_replication_slots columns differ per major (probe-confirmed on live PG 15/16/17/18):

• PG 15 (version < 160000) — wal_status ONLY. conflicting was added in PG16, so selecting it on PG15 errors column "conflicting" does not exist. wal_status = 'lost' is PG15's sole invalidation signal.
• PG 16 (160000 <= version < 170000) — wal_status + conflicting (invalidation_reason/synced were added in PG17 and error here).
• PG 17+ (version >= 170000) — also invalidation_reason (Postgres's authoritative invalidation field) and synced (true on a standby holding a slot synced from the primary).
• PG 9.6 to 12 (version < 130000) — NO invalidation column (ADR-0009 §9): wal_status and max_slot_wal_keep_size arrive in PG13; before that a slot cannot be invalidated by size and the only loss signal is a removed WAL segment, which surfaces as a START_REPLICATION failure (the command-error watchdog). The constant NULL::text keeps the row-present/absent signal the connect chain keys on.

wal_status = 'lost' = WAL removed; conflicting = true = standby recovery conflict; any non-null invalidation_reason = invalidated. All are unrecoverable → fail-closed halt.

@spec start_replication(String.t(), [String.t()], keyword()) ::
  {:ok, String.t()}
  | {:error, :invalid_identifier}
  | {:error, :invalid_start_lsn}

Replication command that starts streaming WAL from start_lsn for the publication set.

publication is a non-empty list of validated identifiers (Critical Rule 2): a single-element list reproduces the published-0.1.0 command byte-for-byte; a multi-element list is comma-joined into publication_names 'p1,p2' (each name is an allowlisted bare identifier, so the bare comma-list is injection-safe — probe-verified). opts[:start_lsn] is a Replicant.lsn/0 (non_neg_integer, default 0).

opts[:streaming], when truthy, selects proto_version '2', streaming 'on' (spec §5). opts[:messages], when truthy, adds messages 'true' (spec §6, A2). Absent or falsy (the default) emits the byte-for-byte v1 command. An opts[:start_lsn] that is not a non-negative integer returns {:error, :invalid_start_lsn} (1.3.0) — a tagged error like every other builder failure, never a raise.

@spec start_replication_for(
  String.t(),
  String.t(),
  [{String.t(), String.t()}],
  non_neg_integer()
) ::
  {:ok, String.t()}
  | {:error, :invalid_identifier}
  | {:error, :invalid_start_lsn}

Plugin-parameterized START_REPLICATION (ADR-0009 §2): renders the decoder plugin's ordered option list as key 'value' pairs inside the command's parentheses. The slot name and plugin name are Identifier.validated; every option KEY matches [a-z0-9_.-]+ and every option VALUE matches [a-zA-Z0-9_.,*()-]* (the option grammar the three shipped decoders emit — identifier lists, booleans, version numbers), so a quote or semicolon breakout is refused {:error, :invalid_identifier} and NEVER interpolated. The pgoutput rendering is byte-identical to start_replication/3 (pinned by test).

@spec table_columns() :: String.t()

Query returning, per publication table, ALL its non-dropped user columns (attnum > 0 AND NOT attisdropped, the SELECT * column set) ordered by attnum — the raw attname (the %Change{}.record string key and the result column name), the server-quoted form via quote_ident (the ONLY form interpolated into the keyset/keyless ::text projection — Critical Rule 2, the format('%I') precedent), and each column's atttypid. The reader maps the OID to a pgoutput type name (OidDatabase.name_for_type_id/1) and casts the <col>::text value through the SAME Casting.Types.cast_record/2 path the stream uses, so snapshot and stream deliver byte-identical %Change{}.record values for EVERY column (spec §2 convergence — the F1 fix generalized from PK-only to all columns). Publication name is bound $1. Row shape: [schemaname, tablename, qualified, col_raw, col_quoted, col_type_oids].

@spec table_columns_for([{String.t(), String.t()}]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

The table_columns/0 row shape for an EXPLICIT table list (the pglogical/wal2json decoders' snapshot path, ADR-0009 §8): same columns, same attnum order, but the table set comes from the caller (VALUES — both parts of every pair are Identifier.validated) instead of pg_publication_tables.

@spec watermark_lsn(boolean(), non_neg_integer()) :: String.t()

Read-only watermark position (spec §2/§4): pg_current_wal_lsn() on a primary, pg_last_wal_replay_lsn() on a standby (the chunk reader connects to the same host as the replication connection, so its snapshot visibility is bounded by replay). The /3 form is version-gated (ADR-0009 §9): PG 9.6 names the current-LSN function pg_current_xlog_location (the pg_current_wal_lsn rename landed in PG 10).