---

title: 03 — Commit-Ready SQL Pack date: 2026-05-31 gate: macro P1 — COMMIT only after human RG1+RG2 (Đ32). DEFERRED here. proven: BEGIN..ROLLBACK GREEN, doc 04 / evidence E11 files: 001..006 + 900_verify + 999_rollback (8 files)

03 — Commit-Ready SQL Pack

Design rules honoured

• No row-multiplying joins. Pivot backing/count via scalar correlated subquery only (the prior LEFT JOIN pivot_definitions ON source_object fanned 160→172 and doubled the invariant — the Đ28 double-count re-entering through SQL).
• No hardcoded code lists. The leaf rule keys on composition_level + entity_type only.
• No frontend logic. All counting/integrity lives in PG; Nuxt/API only render.
• Every count pivot-backed or PIVOT_MISSING. pivot_backed boolean + pivot_count (NULL when missing).
• Tolerates missing pivots & NULL counts (unverified).
• Idempotent via CREATE OR REPLACE; reversible via 999_rollback.sql (Đ30).
• Pivot key = registry_collection (real table name; 21/160) — improvement over the prior split_part(source_location,':',2) (13/160; yields dot/bin/ for File rows).
• Phantom not asserted — drift_classification is a source_model-aware candidate only (authoritative phantom = council, RG4/P6).

Source of truth for these files: /Users/nmhuyen/rp_gateway/sql/ on the operator workstation; reproduced verbatim below. Do not COMMIT until RG1+RG2 are signed (doc 02).

001_create_v_registry_leaf_set.sql

CREATE OR REPLACE VIEW v_registry_leaf_set AS
SELECT
  mc.code, mc.name, mc.entity_type, mc.composition_level, mc.layer, mc.source_model,
  mc.registry_collection, mc.source_location, mc.record_count, mc.actual_count,
  COALESCE(mc.orphan_count, 0) AS orphan_count, mc.baseline_count, mc.last_scan_date
FROM meta_catalog mc
WHERE mc.composition_level <> 'meta'        -- excludes CAT-COL / CAT-DOT / CAT-SPE
  AND mc.entity_type NOT LIKE '%_total'     -- excludes per-level *_total rollups
  AND mc.entity_type <> 'all';              -- excludes CAT-ALL grand total
-- 169 rows -> 160 leaf (verified). Comment: see file header for full rationale.

002_create_v_count_integrity.sql

CREATE OR REPLACE VIEW v_count_integrity AS
SELECT
  l.code, l.name, l.entity_type, l.composition_level, l.source_model,
  l.registry_collection, l.source_location,
  l.record_count AS counted, l.actual_count, l.orphan_count,
  GREATEST(COALESCE(l.record_count,0) - COALESCE(l.actual_count,0), 0) AS record_surplus,
  GREATEST(COALESCE(l.actual_count,0) - COALESCE(l.record_count,0), 0) AS actual_surplus,
  (l.record_count IS DISTINCT FROM l.actual_count) AS drift_flag,
  EXISTS (SELECT 1 FROM pivot_definitions pd
          WHERE pd.is_active AND pd.source_object = l.registry_collection) AS pivot_backed,
  (SELECT (pr.metric_values ->> 'count')::bigint
     FROM pivot_results pr JOIN pivot_definitions pd ON pd.code = pr.pivot_code
    WHERE pd.is_active AND pd.source_object = l.registry_collection
      AND pr.group_values = '{}'::jsonb
    ORDER BY pd.display_order NULLS LAST LIMIT 1) AS pivot_count,
  CASE WHEN l.actual_count IS NULL OR l.record_count IS NULL THEN 'unverified'
       WHEN l.record_count = l.actual_count THEN 'ok' ELSE 'failed' END AS count_integrity_status,
  CASE WHEN l.actual_count IS NULL OR l.record_count IS NULL THEN 'unmeasured'
       WHEN l.record_count = l.actual_count THEN 'balanced'
       WHEN l.source_model='B' AND l.record_count > l.actual_count THEN 'model_b_phantom_candidate'
       WHEN l.source_model='B' AND l.actual_count > l.record_count THEN 'model_b_unregistered_candidate'
       WHEN l.source_model='A' AND l.record_count > l.actual_count THEN 'model_a_surplus_recheck'
       WHEN l.source_model='A' AND l.actual_count > l.record_count THEN 'model_a_deficit_recheck'
       ELSE 'other_recheck' END AS drift_classification
FROM v_registry_leaf_set l;

003_create_v_count_drift.sql

CREATE OR REPLACE VIEW v_count_drift AS
SELECT code, name, entity_type, composition_level, source_model, registry_collection,
       source_location, counted AS record_count, actual_count,
       (COALESCE(counted,0) - COALESCE(actual_count,0)) AS gap,
       CASE WHEN counted > actual_count THEN 'record_over_actual'
            WHEN actual_count > counted THEN 'actual_over_record' ELSE 'none' END AS drift_side,
       drift_classification, pivot_backed, pivot_count
FROM v_count_integrity
WHERE drift_flag;   -- 3 rows live 2026-05-31

004_create_v_living_lists.sql

CREATE OR REPLACE VIEW v_living_lists AS
SELECT l.code, l.name, l.entity_type, l.composition_level, l.source_model, l.registry_collection,
       COALESCE(l.actual_count, l.record_count) AS list_count,
       CASE WHEN l.actual_count IS NOT NULL THEN 'scan_actual'
            WHEN l.record_count IS NOT NULL THEN 'scan_record' ELSE 'unmeasured' END AS count_source,
       (SELECT pd.code FROM pivot_definitions pd
          WHERE pd.is_active AND pd.source_object = l.registry_collection
          ORDER BY pd.display_order NULLS LAST LIMIT 1) AS pivot_code,
       EXISTS (SELECT 1 FROM pivot_definitions pd
               WHERE pd.is_active AND pd.source_object = l.registry_collection) AS pivot_backed
FROM v_registry_leaf_set l;   -- 160 rows; pivot_backed 21; unmeasured 5

005_create_v_registries_pivot_tree.sql

CREATE OR REPLACE VIEW v_registries_pivot_tree AS
SELECT p.code AS node_code, p.name, p.parent_code, p.source_object, p.composition_level,
       p.species, p.registry_group, p.display_order, p.is_active,
       (p.parent_code IS NULL) AS is_root,
       EXISTS (SELECT 1 FROM pivot_definitions c WHERE c.parent_code = p.code) AS has_children
FROM pivot_definitions p;   -- 37 nodes / 37 roots / 0 children until parent_code wired (doc 05)

006_create_fn_node_substrate.sql

CREATE OR REPLACE FUNCTION fn_registries_pivot_node_substrate(p_code text)
RETURNS TABLE (code text, name text, entity_type text, composition_level text, source_model text,
               registry_collection text, source_location text, record_count integer,
               actual_count integer, orphan_count integer, pivot_backed boolean, pivot_count bigint)
LANGUAGE sql STABLE AS $FN$
  SELECT m.code::text, m.name::text, m.entity_type::text, m.composition_level::text,
         m.source_model::text, m.registry_collection::text, m.source_location::text,
         m.record_count, m.actual_count, COALESCE(m.orphan_count,0),
         EXISTS (SELECT 1 FROM pivot_definitions pd
                 WHERE pd.is_active AND pd.source_object = m.registry_collection),
         (SELECT (pr.metric_values ->> 'count')::bigint
            FROM pivot_results pr JOIN pivot_definitions pd ON pd.code = pr.pivot_code
           WHERE pd.is_active AND pd.source_object = m.registry_collection
             AND pr.group_values='{}'::jsonb
           ORDER BY pd.display_order NULLS LAST LIMIT 1)
  FROM meta_catalog m WHERE m.code = p_code;
$FN$;

900_verify.sql (read-only assertions — run inside the rehearsal or post-commit)

V1 leaf_rows=160 · V2 invariant(160/Σrec/Σact/net_gap/drift/unverified/orphans) · V3 drift_closure (leaf net_gap == Σ drift gap) · V4 coverage (backed/missing) · V5 verdict (PASS iff 0 non-ok) · V6 drift_classification breakdown · V7 tree shape · V8 fn smoke (CAT-006, CAT-023) · V9 living_lists count_source. (Full SQL in repo file; output in E11.)

999_rollback.sql (Đ30 reversal)

DROP FUNCTION IF EXISTS fn_registries_pivot_node_substrate(text);
DROP VIEW IF EXISTS v_living_lists;
DROP VIEW IF EXISTS v_count_drift;
DROP VIEW IF EXISTS v_count_integrity;
DROP VIEW IF EXISTS v_registries_pivot_tree;
DROP VIEW IF EXISTS v_registry_leaf_set;

Commit procedure (when RG1+RG2 signed — macro P1)

1. Off-peak window. psql -v ON_ERROR_STOP=1.
2. BEGIN; SET LOCAL lock_timeout='3s'; statement_timeout='30s'; run 001..006 + 900_verify; confirm V1=160, V3 closure holds, V7 tree=37/37/0; if green → COMMIT; else ROLLBACK;.
3. Re-run 900_verify post-commit. Stage 999_rollback for instant reversal.
4. Do not also commit PIV-500/parent_code/threshold/pin here — those are separate gated macros.