@spec canonical([pk_type()], [binary() | integer()]) :: canonical_pk()

Builds the canonical equality form of a PK from pk_types and the raw column values (raw_values, one per PK column in order).

A raw value is either a text-protocol string (a chunk read) or a binlog-decoded term (a streamed change); both forms of the SAME PK canonicalize equal. Integers normalize to their true value (an unsigned column's signed-wrapped negative is unwrapped into [0, 2^width)); binaries pass through their bytes; a string column's value is the {raw, weight} pair its caller assembled (the weight being the canonical bytes — ADR-0012; a bare binary for a string column raises, value-free, rather than silently comparing raws). A single-column PK returns a bare value; a composite returns a tuple in column order.

@spec compare(canonical_pk(), canonical_pk()) :: :lt | :eq | :gt

Compares two canonical PK values (from canonical/2): :lt, :eq, or :gt.

Over the allowlist this is exactly MySQL ORDER BY: Erlang term order coincides with MySQL numeric order for integers, byte-wise order for BINARY/VARBINARY, and — for same-arity composites — lexicographic tuple order for row-value ORDER BY. Both arguments must be canonical forms; the caller compares like against like (a table's PK shape is fixed).

@spec introspect(
  Capstan.Query.t(),
  {String.t(), String.t()}
) :: {:ok, key()} | {:error, atom()}

Introspects the PRIMARY KEY (or a NOT-NULL-complete UNIQUE key fallback) of {schema, table} over a Capstan.Query handle.

Returns {:ok, %{pk_columns: [...], pk_types: [...]}} for an order-faithful key, or a value-free {:error, reason}: :snapshot_pk_unsupported_type when a key column's type is outside the allowlist, :snapshot_table_no_primary_key when there is no PK and no NOT-NULL-complete unique key. A transport/query fault propagates its value-free reason from Capstan.Query.query/2.

@spec order_contract_ok?([[binary()]], [[binary()]]) :: boolean()

The canary verdict: the collation-ordered weight sequence equals the weight-ordered one. Ties are inherently safe — collation-equal values carry equal weights, so both sequences show the same bytes at tie positions.

@spec order_contract_sql(String.t(), String.t(), :collation | :weight) :: String.t()

Builds the bootstrap order-contract canary SQL for one string pk column.

Two queries over the same fixed ASCII vector, introduced with the column's own charset + collation pin: the first orders by the COLLATION (ORDER BY v), the second by the WEIGHT BYTES (ORDER BY WEIGHT_STRING(v)), each returning one hex-encoded weight per row. Under MySQL's documented order contract the two weight sequences are EQUAL (order_contract_ok?/2) — the gate's every decision rests on that equality.

@spec resolve_key([[binary() | nil]], [[binary() | nil]]) ::
  {:ok, key()} | {:error, atom()}

The pure key resolution over the two information_schema resultsets — the RED-capable seam introspect/2 delegates to (so the PK / unique-key fallback / type-refusal logic is provable without a live server).

• index_rows — STATISTICS rows [INDEX_NAME, SEQ_IN_INDEX, COLUMN_NAME, NON_UNIQUE]
• column_rows — COLUMNS rows [COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, IS_NULLABLE, CHARACTER_SET_NAME, COLLATION_NAME]

(the four STATISTICS cells are NOT NULL; the charset/collation cells are NULL for non-string columns, so those two cells may be nil). Returns the same {:ok, key()} | {:error, reason} contract as introspect/2.

@spec resolve_pk_type(binary(), binary()) ::
  {:ok, pk_type()} | {:error, :snapshot_pk_unsupported_type}

Classifies a single column's (DATA_TYPE, COLUMN_TYPE) to an order-faithful pk_type(), or refuses :snapshot_pk_unsupported_type.

Signedness rides COLUMN_TYPE (which carries the unsigned attribute); DATA_TYPE alone cannot distinguish it. CHAR/VARCHAR accept (ADR-0012 — the weight path carries their ordering); the TEXT family (measured per-page filesort) and ENUM/SET (position-based order no introducer weight path reproduces) refuse, as does every non-integer, non-binary, non-char/varchar type (its Elixir order diverges from MySQL).

@spec resolve_weights(Capstan.Query.t(), String.t(), String.t(), [binary()]) ::
  {:ok, %{required(binary()) => binary()}} | {:error, atom()}

Resolves the collation weight bytes for a batch of raw PK-column values over a Capstan.Query handle — {:ok, %{raw => weight}}, or a value-free {:error, reason}.

The stream-side arm of ADR-0012: a binlog-decoded string PK arrives as the column's raw bytes (the row event carries no charset), and its canonical form is the weight the SERVER computes for those bytes under the column's charset + collation. The batch is chunk-bounded (128 values per statement) so one bulk-load transaction cannot exceed max_allowed_packet; a query fault surfaces as the scrubbed Capstan.Query atom for the caller to budget and halt on.

@spec weight_sql(String.t(), String.t(), [binary()]) :: String.t()

Builds the COLLATE-pinned weight-resolution SQL for a batch of raw column values.

One WEIGHT_STRING(CONVERT(X'<hex>' USING charset) COLLATE collation) term per raw, in order — the result row's columns pair positionally with raws. The COLLATE pin is LOAD-BEARING: without it the CONVERT result carries the charset-DEFAULT collation, whose weights are a different order space (probed: a column's as_cs multi-level weight vs the unpinned ai_ci primary-only weight), and the unpinned form in a WHERE additionally raises ERROR 1267 on non-default collations. Hex literals carry arbitrary bytes (NULs, quotes, backslashes, empty) with no string-literal escaping surface.