Add CMS→pipekit migration plan doc

Scope, decisions (dest schema cms.*, fresh DB2 introspection, one table
per cutover), module recipe, and the remaining backlog.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

docs/cms_migration.md

@@ -0,0 +1,214 @@
+# CMS (DB2) → pipekit migration
+
+Scope and plan for moving the remaining CMS/DB2 table syncs from the legacy
+`/opt/sync` (jrunner + shell + cron) tool onto pipekit modules.
+
+## Decisions
+
+- **Destination schema: `cms.*`.** All migrated CMS tables land in `cms.<table>`
+  (pipekit convention), *not* the legacy `lgdat.*`. Downstream `rlarp.*` views that
+  currently read `lgdat.*` must be repointed to `cms.*` as each table is cut over.
+- **Sync tables as they reside on DB2.** Column set and types come from *fresh
+  introspection of the DB2 source* (`QSYS2.SYSCOLUMNS`), not from the legacy
+  `pull.sql` / `build.sql`. Legacy queries are not ported; we regenerate clean.
+- **One table at a time on cutover.** Generate/verify the module, repoint the
+  downstream view, then retire the `/opt/sync` module.
+
+## Current state
+
+- pipekit already covers **39 CMS tables** (`source_connection: S78030956`,
+  `FROM LGDAT.*`, landing in `cms.*`).
+- `/opt/sync/s7830956/` (92 module dirs) holds the legacy CMS syncs. 24 overlap
+  the 39; the rest is the migration backlog below.
+
+## The module recipe (pipekit)
+
+A module is two sidecar files under `config/modules/`, named for the module:
+
+- **`<name>.json`** — definition. Required fields: `name`, `source_connection`,
+  `dest_connection`, `dest_table` (schema-qualified), `staging_table`
+  (`pipekit_staging.<name>`), `merge_strategy` (`full` | `incremental` | `append`),
+  `merge_key` (required for incremental, else `null`), `enabled`, `columns[]`,
+  `watermarks[]`, `hooks[]`. `dest_description` optional.
+  - `columns[]` entries: `source_name`, `source_type`, `dest_name`, `dest_type`,
+    `description`.
+- **`<name>.sql`** — the extract `SELECT`, run on the **source** connection in its
+  dialect. DB2: `RTRIM(col)` on char cols, `CASE WHEN col IN (DATE('0001-01-01'),
+  DATE('9999-12-31')) THEN NULL ELSE col END` on dates, `"DOUBLE#QUOTES"` for
+  `#`-bearing identifiers. Watermarks are `{name}` placeholders substituted at run.
+
+Group assignment + cron live in `config/groups.json` (add `{"module","run_order"}`
+to a group's `members`). Source connections in `config/connections.json`
+(`S78030956` = DB2/AS400). `pipekit apply` hydrates `pipekit.db` from these files.
+
+Merge semantics: staging recreated `LIKE dest` each run, then `full` = TRUNCATE +
+INSERT; `incremental` = DELETE by `merge_key` + INSERT; `append` = INSERT only.
+
+## Introspection method (how drafts are generated)
+
+pipekit exposes introspection two ways:
+- HTTP API `GET /api/introspect/columns` (`:8200`, Basic Auth).
+- Driver layer `pipekit.drivers.db2.DB2Driver.get_columns()` — same code, in-process,
+  needs only `DB2PW` from `/etc/pipekit/secrets.env`.
+
+Drafts use the driver layer, reusing pipekit's own `_TYPE_MAP`, `default_expression`
+(RTRIM / date-sentinel), and `quote_identifier` (`#` handling) — so output matches
+pipekit style exactly, including real column/table descriptions from the DB2 catalog.
+Type mapping: `CHAR/VARCHAR/GRAPHIC/CLOB → text`, `DECIMAL/NUMERIC(p,s)` preserved,
+`DATE → date`, `TIME → time`, `TIMESTAMP → text` (pipekit convention), `FLOAT →
+double precision`, `BIGINT/INTEGER/SMALLINT` preserved.
+
+Reproduce: `drafts/bucket1/_generate.py` (reads DB2 read-only, writes `.json`/`.sql`).
+
+## Bucket 1 — full truncate-reload (~41 tables) — DRAFTED
+
+No watermark; legacy `insert.sql` was delete-all + reinsert. Simplest module:
+`merge_strategy: "full"`, `merge_key: null`.
+
+**Status: drafted, not activated.** Files in `/opt/pipekit/drafts/bucket1/`
+(41 `.json` + 41 `.sql`). Nothing wired to a group; `pipekit apply` not run.
+
+Tables (source library in parens where not LGDAT):
+adrs, cret, depts, fresre, resre, ftcstm, ftcstp, ftcstr, glie, icstm, icstp, icstr,
+iprca, iprcb, iprcc, iprcctn, iprccto, irea, macgrp, majg, methdm, methdo, methdr,
+mmgp, mmsl, opcode, plnt, punit, sach, sscc, stka, stkmm, stkmp, usrd, vend,
+color / colorb / colortier / iprcbhc (`"CMS.CUSLG"`), usrcust (`LGPGM`),
+ffpdglr1 (`FANALDEV`).
+
+**Activation (per table, when ready):**
+1. Review the drafted `.json`/`.sql`.
+2. `mv drafts/bucket1/<name>.{json,sql} config/modules/`.
+3. Add `{"module":"<name>","run_order":N}` to the right group in `groups.json`.
+4. `pipekit apply`, then `pipekit run <name>` to smoke-test.
+5. Repoint any `rlarp.*` view from `lgdat.<name>` to `cms.<name>`.
+
+**Post-load hooks needed** (legacy refreshed a mat-view; add a `hooks[]` entry once
+the downstream view is repointed to `cms.*`): `icstr` → `rlarp.icstx`,
+`ftcstr` → `rlarp.ftcstx`, `ffpdglr1` → `rlarp.pdglr1`, `stkmp` → `CALL
+rlarp.itemm_ps_build()`.
+
+## Bucket 2 — watermarked incremental (6 tables) — TODO
+
+Need `merge_key` + a watermark resolver query (`SELECT MAX(...)` on the dest,
+substituted into `{wm}` in the `.sql`). Templates: `config/modules/ocrs.json`,
+`gtran.json`.
+
+| Table   | merge_key            | watermark |
+|---------|----------------------|-----------|
+| qcrh    | `dcord#`             | order # high-water |
+| qcri    | `ddord#`             | order # |
+| qtnote  | `ggkey`              | key |
+| methh   | `anpart`+`anplnt` (EXISTS merge) | `andate` |
+| iprcct  | date                 | `tadate` |
+| icstt   | date                 | `jhdate` |
+
+Each has a `*_full` unbounded counterpart in `/opt/sync` that can be dropped.
+
+**Price-list change log (`iprcct` / `iprcctn` / `iprccto`) — DONE 2026-08-05.**
+Append-only log, ~6.4–6.9M rows each, ~100 rows/day. All three: watermark
+`MAX(<date>) - 7` on the dest (guarded `<= current_date`), source
+`WHERE <date> BETWEEN '{wm}' AND DATE('9998-12-31')`, `incremental` merge keyed on
+the date alone — the staged set is every row on/after the watermark, so
+delete-by-date replaces whole day partitions and re-running is a no-op. The
+`9998-12-31` upper bound matters: the source transform NULLs the `9999-12-31`
+sentinel, and a NULL merge key never matches the DELETE, so such a row would
+duplicate on every run. Grain is `(plcd, part, unit, date, time)` for the header
+and the same plus `voll` for the detail tables. All three reconcile exactly to
+source counts. Members of group 14 (Price Lists), scheduled `0 3 * * *` — local
+time, after Sales Matrix at 02:20. The group is ~3.5 min, nearly all of it `iprcc`
+(908k rows, full reload); the three change-log modules are seconds each.
+
+**Quote family (`qcrh` / `qcri` / `qtnote`) — DONE 2026-08-05.** The order-number
+watermark sketched in the table above is wrong and was replaced: quotes are edited
+long after creation (of 644 headers updated in a 30-day sample, 25 were below
+`MAX(dcord#) - 1000`), so an order-number window silently drops edits. QCRH shares
+OCRH's `DC*` layout, so all three now use the ocrh/ocri pattern — watermark
+`MAX(dcudat) - 7` on `cms.qcrh` (guarded `<= current_date`) for **all three**
+modules, header WHERE unioning `dcudat` / `dcodat` / `dccdat`, details joined to a
+`changed` CTE over `lgdat.qcrh`. The union is load-bearing: 36,339 rows have
+`dccdat` after `dcudat`, 270 have `dcodat` bumped without `dcudat`. Because the
+detail watermarks resolve off the *header* table, run order within group 16 must
+stay header-first and the lookback must exceed the sync interval.
+
+- `qcri` (492k rows, grain `ddord#+dditm#+dddes#`) keys on `ddord#`. Verified the
+  changed-set catches line activity: all 5,807 lines created in a 30-day window
+  (`DDCTMS`) belong to a header whose dates moved, 0 orphans. `DDCTMS` is a
+  *creation* timestamp, not last-changed — unusable as a watermark.
+- `qtnote` (253k rows) has **no change signal of its own** — no dates, no
+  timestamp. `GGKEY` is `9-digit quote number` (header note) or `+ 3-digit item`
+  (line note); all 50,876 note keys resolve to QCRH, only 6 also exist in OCRH, so
+  it is purely quote notes. Keying on `ggkey` would leave a cleared note stranded
+  in the dest forever, so a derived tail column `cms.qtnote.ggord` =
+  `SUBSTR(ggkey,1,9)` was added (and backfilled) and is the merge key — delete now
+  replaces *all* notes for a changed quote. Join is string-on-string via
+  `DIGITS(dcord#)`, deliberately avoiding a numeric CAST: one junk key (19 spaces +
+  `.`) would raise SQL0420 on cast, and staying textual leaves it harmlessly
+  excluded (it survives in the dest with `ggord IS NULL`, so dest count ties to
+  source exactly at 252,721).
+
+One-time rebaseline was required: a quote-level line-count diff found 273 quotes
+disagreeing with source (in *both* directions, back to 2023) plus 22 missing
+entirely — accumulated debt from the old order-number window. Rebaselined by
+temporarily pointing each resolver at `DATE '1900-01-01'` and running, then
+restoring; a wide date window provably covers all 82,224 headers, and delete-by-key
+avoids the ACCESS EXCLUSIVE lock a `full` TRUNCATE would take on tables the 15-min
+Quote Review group reads. All three now tie exactly to source. Group 16 (Quotes) is
+scheduled `*/15 * * * *` (~20 s per pass) with explicit `run_order` 1/2/3 — members
+sort `run_order, name`, so header-first was previously only an alphabetical
+accident, and both detail modules read their changed-set from `cms.qcrh`.
+
+Residual risk to watch on `qtnote`: it inherits change detection entirely from the
+header, so a note edited without the quote header's dates moving is invisible. Worth
+a periodic wide-window rebaseline (the same widen/run/restore steps) or a
+`reconcile.py` check.
+
+**`icstt` (Cost History, module 129) — reviewed 2026-08-05; KNOWN DRIFT ACCEPTED.**
+3.77M rows, watermark now `MAX(jhdate) - 7` guarded `<= current_date` (the guard
+matters — source holds 2 future-dated rows out to 2031). Scheduled `30 3 * * *`
+local as the sole member of group 13; before this it ran only when clicked.
+
+Two defects found, one fixed, one deliberately left:
+
+1. *Fixed:* the resolver had no lookback, so anything back-dated below the
+   high-water was invisible forever.
+2. *Accepted:* **the source is purged at row level**, and a date-keyed incremental
+   structurally cannot see it — delete-by-`jhdate` only touches dates in the staged
+   set, and old dates never enter it, so every upstream deletion strands a dest row
+   permanently. Currently +2,241 rows across 149 dates from 2010 onward (0.06% of
+   the table). Verified by key-level diff on one date: 20 dest-only keys, 0
+   source-only, i.e. dest ⊇ source. This grows over time and silently inflates
+   historical cost aggregates. Only a wide-window pass removes them.
+
+Also 5 source rows dated 1953–1966 (1 row each, data-entry junk) have never synced
+and are **intentionally not synced** — a 7-day lookback starts ~2026-07-29 and never
+reaches them.
+
+Consequence for monitoring: a bare `reconcile.py icstt` will always report DIVERGED.
+Use the floor filter so the comparison is apples-to-apples — dates then tie exactly
+(5,156 = 5,156) and the only mismatch is the purge overage:
+
+```
+PIPEKIT_SECRETS=/etc/pipekit/secrets.env .venv/bin/python reconcile.py icstt \
+  --super-quick --source-from "(SELECT * FROM LGDAT.ICSTT WHERE JHDATE >= DATE('2000-01-01')) t"
+```
+
+To rebaseline when there's a ~45 min window (est. from a 417k-row run at 5m06s):
+point watermark 32 at `SELECT DATE '2000-01-01'` (that floor stages all 5,156 real
+dates and clears the orphans while still skipping the 1953–66 junk), `pipekit run
+icstt`, restore the `MAX(jhdate) - 7` resolver, then re-run the check above — it
+should read `IN SYNC`. Delete-by-date, so no `TRUNCATE` and no ACCESS EXCLUSIVE lock.
+
+## Bucket 3 — special cases — decide, don't port as-is
+
+- **Reverse pushes PG→DB2** (`osmfs`, `reprice`, `reprice_test`): go Postgres→AS400,
+  opposite of pipekit's source→PG model. Leave in `/opt/sync` or redesign separately.
+- **Stored-proc modules** (`sb_ud_r2`, `sb_gj_r1`, `osm_sync`): `CALL` procs, not table
+  extracts. Convert as hooks or leave out.
+- **Derived RLARP reads** (`family`, `ffterr`, `qrh`, `ctqpor`): read derived AS400
+  `rlarp.*` tables, not raw CMS masters. Confirm the upstream proc still runs first.
+
+## Sequencing recommendation
+
+Prioritize by what the marts consume: the cost/pricing cluster (`icst*`, `iprc*`,
+`ftcst*`, `punit`) and vendor/plant/method dims are highest value. Explicitly defer
+or retire the reverse pushes and derived reads rather than porting them.

reconcile.py

@@ -13,7 +13,13 @@ aggregates line up when the row sets agree.
 
 Usage:
     PIPEKIT_SECRETS=/etc/pipekit/secrets.env \\
-        .venv/bin/python reconcile.py <module_name> [--quick] [--source-from EXPR]
+        .venv/bin/python reconcile.py <module_name> \\
+            [--quick | --super-quick] [--source-from EXPR]
+
+Depth: default compares every column; ``--quick`` drops min/max/len_sum;
+``--super-quick`` compares only the row counts (COUNT(*) and COUNT(DISTINCT
+merge_key)) — seconds even on multi-million-row tables, and enough to catch
+missing or duplicated rows, but blind to changed column values.
 
 Exit code is non-zero when any metric diverges.
 """
@@ -39,6 +45,26 @@ def classify(dest_type: str) -> str:
     return "text"
 
 
+def is_expression(source_name: str) -> bool:
+    """True when a columns_json source_name is a SQL expression, not a column.
+
+    Derived columns (e.g. qtnote's merge key ``SUBSTR(GGKEY,1,9)``) are stored
+    in columns_json with the expression in ``source_name``. Those must be
+    emitted verbatim — quoting them as an identifier yields SQL0206, and
+    wrapping them in ``default_expression`` would re-apply a transform the
+    module's own SELECT already applied.
+    """
+    return "(" in (source_name or "")
+
+
+def source_expression(drv, column: dict) -> str:
+    """Source-side SQL for a column: verbatim if derived, else transformed."""
+    name = column["source_name"]
+    if is_expression(name):
+        return name
+    return drv.default_expression(column["source_type"], name)
+
+
 def detect_source_from(source_query: str, dest_table: str) -> str | None:
     """Best-effort: find the base source table in the module's SELECT.
 
@@ -58,11 +84,16 @@ def detect_source_from(source_query: str, dest_table: str) -> str | None:
     return matches[-1]
 
 
-def build_metrics(columns, key_names, drv, *, source: bool, quick: bool):
+def build_metrics(columns, key_names, drv, *, source: bool, quick: bool,
+                  super_quick: bool = False):
     """Return (list of (label, kind) metric descriptors, list of SQL exprs).
 
     Descriptors and exprs are positionally aligned so the two sides zip up.
     ``source`` selects which column name + transform to use.
+
+    Three depths: full (every column, incl. min/max/len_sum), ``quick``
+    (per-column counts + numeric sums), and ``super_quick`` (the headline row
+    counts only — no per-column work, so the DB can often answer from an index).
     """
     labels: list[tuple[str, str]] = []
     exprs: list[str] = []
@@ -72,14 +103,17 @@ def build_metrics(columns, key_names, drv, *, source: bool, quick: bool):
 
     for kn in key_names:
         col = kn["source"] if source else kn["dest"]
-        q = drv.quote_identifier(col)
+        q = col if (source and is_expression(col)) else drv.quote_identifier(col)
         labels.append((f"COUNT(DISTINCT {kn['dest']})", "int"))
         exprs.append(f"COUNT(DISTINCT {q})")
 
+    if super_quick:
+        return labels, exprs
+
     for c in columns:
         kind = classify(c["dest_type"])
         if source:
-            e = drv.default_expression(c["source_type"], c["source_name"])
+            e = source_expression(drv, c)
         else:
             e = drv.quote_identifier(c["dest_name"])
         name = c["dest_name"]
@@ -130,6 +164,10 @@ def main() -> int:
     ap.add_argument("module", help="module name (e.g. ocri)")
     ap.add_argument("--quick", action="store_true",
                     help="counts + numeric sums only (skip min/max/len_sum)")
+    ap.add_argument("--super-quick", action="store_true",
+                    help="row counts only: COUNT(*) + COUNT(DISTINCT merge_key). "
+                         "Cheapest check — catches missing/duplicated rows, not "
+                         "changed column values")
     ap.add_argument("--source-from",
                     help="override the source FROM target "
                          "(e.g. a schema.table or OPENQUERY(...) t)")
@@ -168,10 +206,12 @@ def main() -> int:
         print("could not detect source table; pass --source-from", file=sys.stderr)
         return 2
 
-    labels, src_exprs = build_metrics(columns, key_names, src_drv,
-                                      source=True, quick=args.quick)
-    _, dst_exprs = build_metrics(columns, key_names, dst_drv,
-                                 source=False, quick=args.quick)
+    # --super-quick subsumes --quick; pass both so the depth is unambiguous.
+    quick = args.quick or args.super_quick
+    labels, src_exprs = build_metrics(columns, key_names, src_drv, source=True,
+                                      quick=quick, super_quick=args.super_quick)
+    _, dst_exprs = build_metrics(columns, key_names, dst_drv, source=False,
+                                 quick=quick, super_quick=args.super_quick)
 
     src_sql = "SELECT\n  " + "\n , ".join(src_exprs) + f"\nFROM {source_from}"
     dst_sql = "SELECT\n  " + "\n , ".join(dst_exprs) + f"\nFROM {m['dest_table']}"
@@ -179,7 +219,9 @@ def main() -> int:
     print(f"module {m['name']}  (id {m['id']})")
     print(f"  source {src_conn['name']}  FROM {source_from}")
     print(f"  dest   {dst_conn['name']}  FROM {m['dest_table']}")
-    print(f"  {len(labels)} metrics{'  [quick]' if args.quick else ''}\n"
+    mode = ("  [super-quick]" if args.super_quick
+            else "  [quick]" if args.quick else "")
+    print(f"  {len(labels)} metrics{mode}\n"
           f"  running source aggregate ...", flush=True)
     src_res = jrunner.query(src_conn["jdbc_url"], src_conn.get("username"),
                             src_conn.get("password"), src_sql, timeout=args.timeout)