Add reconcile.py --super-quick and derived-column support

--super-quick compares only COUNT(*) and COUNT(DISTINCT merge_key) —
seconds on multi-million-row tables, enough to catch missing/duplicated
rows but blind to changed values.

Also fix derived merge keys: columns_json source_name can hold a SQL
expression (e.g. SUBSTR(GGKEY,1,9)), which must be emitted verbatim
rather than quoted as an identifier (SQL0206) or re-transformed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Paul Trowbridge 2026-08-05 15:48:45 -04:00
parent 560056c39c
commit 0db719c835

View File

@ -13,7 +13,13 @@ aggregates line up when the row sets agree.
Usage: Usage:
PIPEKIT_SECRETS=/etc/pipekit/secrets.env \\ 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. Exit code is non-zero when any metric diverges.
""" """
@ -39,6 +45,26 @@ def classify(dest_type: str) -> str:
return "text" 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: def detect_source_from(source_query: str, dest_table: str) -> str | None:
"""Best-effort: find the base source table in the module's SELECT. """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] 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). """Return (list of (label, kind) metric descriptors, list of SQL exprs).
Descriptors and exprs are positionally aligned so the two sides zip up. Descriptors and exprs are positionally aligned so the two sides zip up.
``source`` selects which column name + transform to use. ``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]] = [] labels: list[tuple[str, str]] = []
exprs: list[str] = [] exprs: list[str] = []
@ -72,14 +103,17 @@ def build_metrics(columns, key_names, drv, *, source: bool, quick: bool):
for kn in key_names: for kn in key_names:
col = kn["source"] if source else kn["dest"] 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")) labels.append((f"COUNT(DISTINCT {kn['dest']})", "int"))
exprs.append(f"COUNT(DISTINCT {q})") exprs.append(f"COUNT(DISTINCT {q})")
if super_quick:
return labels, exprs
for c in columns: for c in columns:
kind = classify(c["dest_type"]) kind = classify(c["dest_type"])
if source: if source:
e = drv.default_expression(c["source_type"], c["source_name"]) e = source_expression(drv, c)
else: else:
e = drv.quote_identifier(c["dest_name"]) e = drv.quote_identifier(c["dest_name"])
name = 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("module", help="module name (e.g. ocri)")
ap.add_argument("--quick", action="store_true", ap.add_argument("--quick", action="store_true",
help="counts + numeric sums only (skip min/max/len_sum)") 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", ap.add_argument("--source-from",
help="override the source FROM target " help="override the source FROM target "
"(e.g. a schema.table or OPENQUERY(...) t)") "(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) print("could not detect source table; pass --source-from", file=sys.stderr)
return 2 return 2
labels, src_exprs = build_metrics(columns, key_names, src_drv, # --super-quick subsumes --quick; pass both so the depth is unambiguous.
source=True, quick=args.quick) quick = args.quick or args.super_quick
_, dst_exprs = build_metrics(columns, key_names, dst_drv, labels, src_exprs = build_metrics(columns, key_names, src_drv, source=True,
source=False, quick=args.quick) 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}" 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']}" 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"module {m['name']} (id {m['id']})")
print(f" source {src_conn['name']} FROM {source_from}") print(f" source {src_conn['name']} FROM {source_from}")
print(f" dest {dst_conn['name']} FROM {m['dest_table']}") 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) f" running source aggregate ...", flush=True)
src_res = jrunner.query(src_conn["jdbc_url"], src_conn.get("username"), src_res = jrunner.query(src_conn["jdbc_url"], src_conn.get("username"),
src_conn.get("password"), src_sql, timeout=args.timeout) src_conn.get("password"), src_sql, timeout=args.timeout)