from __future__ import annotations import argparse import importlib import json from datetime import date, datetime, time from decimal import Decimal from pathlib import Path from typing import Any, TypedDict REPO_ROOT = Path(__file__).resolve().parents[1] DEFAULT_DB = REPO_ROOT / "files" / "warehouse.db" DEFAULT_STATE = REPO_ROOT / "files" / "state.json" DEFAULT_EXPECTED = REPO_ROOT / "examples" / "local_debug_expected.json" type JsonObject = dict[str, Any] class StaticExpectedPhase(TypedDict): checkpoint_ts: str rows: list[tuple[int, str, int, str, str, bool]] EXPECTED: dict[str, StaticExpectedPhase] = { "phase1": { "checkpoint_ts": "200", "rows": [ (1, "Alicia", 121, "test", "worker", False), (3, "Carol", 150, "test", "worker", False), ], }, "phase2": { "checkpoint_ts": "300", "rows": [ (1, "Alicia v2", 210, "test", "worker", False), (4, "Dave", 230, "test", "worker", False), ], }, } def _load_duckdb() -> Any: try: return importlib.import_module("duckdb") except ModuleNotFoundError as exc: raise SystemExit( "duckdb Python package is required for verification.\n" "Install it with: UV_CACHE_DIR=.uv-cache uv pip install --python .venv/bin/python duckdb" ) from exc def _active_rows(db_path: Path) -> list[tuple[object, ...]]: duckdb = _load_duckdb() if not db_path.exists(): raise SystemExit(f"DuckDB warehouse does not exist: {db_path}") con = duckdb.connect(str(db_path), read_only=True) try: return con.sql( """ SELECT id, name, _incremental_commit_ts, _incremental_source_schema, _incremental_source_table, COALESCE(_fivetran_deleted, false) AS deleted FROM tester.test_worker WHERE COALESCE(_fivetran_deleted, false) = false ORDER BY id """ ).fetchall() finally: con.close() def _quote_identifier(value: str) -> str: return '"' + value.replace('"', '""') + '"' def _json_safe(value: Any) -> Any: if isinstance(value, Decimal): return str(value) if isinstance(value, (datetime, date, time)): return value.isoformat() if isinstance(value, bytes): return value.hex() return value def _active_rows_for_table(db_path: Path, table: str, columns: list[str]) -> list[dict[str, Any]]: duckdb = _load_duckdb() if not db_path.exists(): raise SystemExit(f"DuckDB warehouse does not exist: {db_path}") selected_columns = [ *columns, "_incremental_commit_ts", "_incremental_source_schema", "_incremental_source_table", "_fivetran_deleted", ] select_sql = ", ".join( ( _quote_identifier(column) if column != "_fivetran_deleted" else "COALESCE(_fivetran_deleted, false) AS _fivetran_deleted" ) for column in selected_columns ) order_sql = ", ".join(_quote_identifier(column) for column in columns) or "1" con = duckdb.connect(str(db_path), read_only=True) try: rows = con.sql( f""" SELECT {select_sql} FROM tester.{_quote_identifier(table)} WHERE COALESCE(_fivetran_deleted, false) = false ORDER BY {order_sql} """ ).fetchall() finally: con.close() result = [] for row in rows: result.append({column: _json_safe(value) for column, value in zip(selected_columns, row, strict=True)}) return result def _state(path: Path) -> JsonObject: if not path.exists(): raise SystemExit(f"Fivetran debug state does not exist: {path}") parsed = json.loads(path.read_text(encoding="utf-8")) if not isinstance(parsed, dict): raise SystemExit(f"Fivetran debug state must be a JSON object: {path}") return parsed def _verify_dynamic_expected(phase: str, expected_path: Path, db_path: Path, state_path: Path) -> int: if not expected_path.exists(): raise SystemExit(f"expected-result JSON does not exist: {expected_path}") expected_all: JsonObject = json.loads(expected_path.read_text(encoding="utf-8")) if phase not in expected_all: raise SystemExit(f"expected-result JSON does not contain {phase}") expected = expected_all[phase] if not isinstance(expected, dict): raise SystemExit(f"expected-result JSON {phase} must be an object") for table, table_expected in sorted(expected["tables"].items()): actual_rows = _active_rows_for_table(db_path, table, table_expected["columns"]) expected_rows = table_expected["rows"] if actual_rows != expected_rows: print(f"unexpected active rows for {table}") print(f"expected: {expected_rows}") print(f"actual: {actual_rows}") return 1 state = _state(state_path) snapshot = state.get("snapshot") incremental = state.get("incremental") if not isinstance(snapshot, dict) or snapshot.get("done") is not True: print(f"snapshot is not marked done in state: {state}") return 1 if not isinstance(incremental, dict) or incremental.get("checkpoint_ts") != expected["checkpoint_ts"]: print(f"unexpected checkpoint_ts: {incremental}") return 1 print(f"{phase} verified with {expected_path}") print(f"state checkpoint_ts: {incremental['checkpoint_ts']}") return 0 def main() -> int: parser = argparse.ArgumentParser(description="Verify Fivetran debug DuckDB output.") parser.add_argument("phase", choices=sorted(EXPECTED)) parser.add_argument("--db", type=Path, default=DEFAULT_DB) parser.add_argument("--state", type=Path, default=DEFAULT_STATE) parser.add_argument( "--expected", type=Path, default=None, help="Dynamic expected-result JSON generated by local_debug_prepare.py --input-sql.", ) args = parser.parse_args() if args.expected is not None: return _verify_dynamic_expected(args.phase, args.expected, args.db, args.state) expected = EXPECTED[args.phase] rows = _active_rows(args.db) if rows != expected["rows"]: print("unexpected active rows") print(f"expected: {expected['rows']}") print(f"actual: {rows}") return 1 state = _state(args.state) snapshot = state.get("snapshot") incremental = state.get("incremental") if not isinstance(snapshot, dict) or snapshot.get("done") is not True: print(f"snapshot is not marked done in state: {state}") return 1 if snapshot.get("snapshot_tso") != "100": print(f"unexpected snapshot_tso: {snapshot.get('snapshot_tso')!r}") return 1 if not isinstance(incremental, dict) or incremental.get("checkpoint_ts") != expected["checkpoint_ts"]: print(f"unexpected checkpoint_ts: {incremental}") return 1 print(f"{args.phase} verified") print(f"rows: {rows}") print(f"state checkpoint_ts: {incremental['checkpoint_ts']}") return 0 if __name__ == "__main__": raise SystemExit(main())