From 0db719c8350d93fd0f374200bc18a887be3548c3 Mon Sep 17 00:00:00 2001 From: Paul Trowbridge Date: Wed, 5 Aug 2026 15:48:45 -0400 Subject: [PATCH] Add reconcile.py --super-quick and derived-column support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --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 --- reconcile.py | 60 ++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 51 insertions(+), 9 deletions(-) diff --git a/reconcile.py b/reconcile.py index bd18785..b0dfbfd 100755 --- a/reconcile.py +++ b/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 [--quick] [--source-from EXPR] + .venv/bin/python reconcile.py \\ + [--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)