from __future__ import annotations import argparse import importlib import json import shutil import sys from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) compute_table_definition_checksum = importlib.import_module("increment.schema").compute_table_definition_checksum _local_debug_sql = importlib.import_module("scripts.local_debug_sql") add_phase2_from_sql = _local_debug_sql.add_phase2_from_sql prepare_from_sql = _local_debug_sql.prepare_from_sql DEFAULT_WORKSPACE = REPO_ROOT / "examples" / "local_debug_workspace" DEFAULT_CONFIGURATION = REPO_ROOT / "examples" / "local_debug_configuration.json" DEFAULT_DEBUG_DIR = REPO_ROOT / "files" DEFAULT_EXPECTED = REPO_ROOT / "examples" / "local_debug_expected.json" def _write_text(path: Path, payload: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(payload, encoding="utf-8") def _write_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") def _table_definition(version: int = 120) -> dict[str, Any]: return { "Table": "worker", "Schema": "test", "Version": 0, "TableVersion": version, "Query": "", "Type": 0, "TableColumns": [ {"ColumnName": "id", "ColumnType": "BIGINT"}, {"ColumnName": "name", "ColumnType": "VARCHAR"}, ], "TableColumnsTotal": 2, } def _write_configuration(path: Path, workspace: Path) -> None: _write_json( path, { "sync_task_id": "local-debug-task", "s3_url": workspace.resolve().as_uri(), "snapshot_prefix": "snapshot", "incremental_prefix": "increment", "checkpoint_interval_rows": "2", "checkpoint_interval_seconds": "600", "enable_diagnostic_columns": "true", }, ) def _write_snapshot(workspace: Path) -> None: _write_text( workspace / "snapshot" / "metadata", "\n".join( [ "Started dump at: 2026-06-04 10:00:00", "SHOW MASTER STATUS:", " Log: tidb-binlog", " Pos: 100", " GTID:", "", "Finished dump at: 2026-06-04 10:00:10", "", ] ), ) _write_text( workspace / "snapshot" / "test.worker-schema.sql", """CREATE TABLE `worker` ( `id` bigint NOT NULL, `name` varchar(64) DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; """, ) _write_text( workspace / "snapshot" / "test.worker.000000001.csv", "id,name\n1,Alice\n2,Bob\n", ) def _write_schema_marker(workspace: Path, version: int = 120) -> None: payload = _table_definition(version) checksum = compute_table_definition_checksum(payload) _write_json( workspace / "increment" / "test" / "worker" / "meta" / f"schema_{version}_{checksum}.json", payload, ) def _write_phase1_increment(workspace: Path) -> None: _write_json(workspace / "increment" / "metadata", {"checkpoint-ts": 200}) _write_schema_marker(workspace) base = workspace / "increment" / "test" / "worker" / "120" / "2026-06-04" _write_text(base / "meta" / "CDC.index", "CDC000001.csv\n") _write_text( base / "CDC000001.csv", "\n".join( [ "I,worker,test,90,99,TooOld", "U,worker,test,121,1,Alicia", "I,worker,test,150,3,Carol", "D,worker,test,180,2,Bob", "", ] ), ) def _write_phase2_increment(workspace: Path) -> None: _write_json(workspace / "increment" / "metadata", {"checkpoint-ts": 300}) _write_schema_marker(workspace) base = workspace / "increment" / "test" / "worker" / "120" / "2026-06-04" _write_text(base / "meta" / "CDC.index", "CDC000002.csv\n") _write_text( base / "CDC000002.csv", "\n".join( [ "U,worker,test,210,1,Alicia v2", "D,worker,test,220,3,Carol", "I,worker,test,230,4,Dave", "", ] ), ) def _reset_workspace(workspace: Path) -> None: resolved = workspace.resolve() allowed_parent = (REPO_ROOT / "examples").resolve() if resolved == allowed_parent or allowed_parent not in resolved.parents: raise SystemExit(f"refusing to delete workspace outside examples/: {resolved}") shutil.rmtree(resolved, ignore_errors=True) def _reset_debug_output(debug_dir: Path) -> None: resolved = debug_dir.resolve() if resolved.name != "files" or resolved.parent != REPO_ROOT: raise SystemExit(f"refusing to delete unexpected debug directory: {resolved}") for filename in ("state.json", "warehouse.db"): (resolved / filename).unlink(missing_ok=True) def main() -> int: parser = argparse.ArgumentParser(description="Prepare local filesystem fixtures for Fivetran debug.") parser.add_argument("phase", choices=("phase1", "phase2")) parser.add_argument("--workspace", type=Path, default=DEFAULT_WORKSPACE) parser.add_argument("--configuration", type=Path, default=DEFAULT_CONFIGURATION) parser.add_argument("--debug-dir", type=Path, default=DEFAULT_DEBUG_DIR) parser.add_argument( "--expected", type=Path, default=DEFAULT_EXPECTED, help="Expected-result JSON path used by SQL input mode.", ) parser.add_argument( "--input-sql", type=Path, help=( "MySQL SQL file to convert into local snapshot/incremental files. " "Initial CREATE/INSERT statements become snapshot data; later DML becomes incremental data. " "Use comments like '-- fivetran: phase2' to mark a second incremental run." ), ) parser.add_argument( "--reset-debug-output", action="store_true", help="Delete files/state.json and files/warehouse.db before phase1.", ) args = parser.parse_args() workspace = args.workspace configuration = args.configuration if args.phase == "phase1": _reset_workspace(workspace) if args.reset_debug_output: _reset_debug_output(args.debug_dir) if args.input_sql is not None: prepared = prepare_from_sql( input_sql=args.input_sql, workspace=workspace, expected_path=args.expected, ) print(f"wrote expected results: {prepared.expected_path}") else: _write_snapshot(workspace) _write_phase1_increment(workspace) else: if not workspace.exists(): raise SystemExit("phase2 requires an existing phase1 workspace") if args.input_sql is not None: if not add_phase2_from_sql( input_sql=args.input_sql, workspace=workspace, expected_path=args.expected, ): raise SystemExit("input SQL does not define phase2 incremental data") else: _write_phase2_increment(workspace) _write_configuration(configuration, workspace) print(f"wrote workspace: {workspace}") print(f"wrote configuration: {configuration}") return 0 if __name__ == "__main__": raise SystemExit(main())