tests/ts_table_perf.py

#!/usr/bin/env python3
"""
tests/ts_table_perf.py — ThoughtSpot table-create performance harness.

WHY
    Creating a ThoughtSpot table (TML import) in the app has gone from ~2 min to
    sometimes ~10 min, with heavy polling. This harness isolates and TIMES the
    create path against OUR OWN dedicated connection so we can find the pattern
    behind the slowness — without touching a real demo connection.

WHAT IT DOES  (faithful to the app's real code path)
    - Authenticates via the real ThoughtSpotDeployer (trusted auth).
    - Creates OUR OWN connection (kept, labelled DONOTDELETE).
    - Ensures a Snowflake schema (default DEMOBUILD.ZPERFTEST) + one EMPTY
      physical table per logical table (0 rows — instant to create).
    - Creates N logical tables, EACH WITH A DIFFERENT NAMING CONVENTION, using
      the app's real ThoughtSpotDeployer.create_table_tml(), and imports each
      via the SAME call the app makes:
          POST /api/rest/2.0/metadata/tml/import
          {metadata_tmls, import_policy: PARTIAL, create_new}, timeout=360
      (mirrors thoughtspot_deployer.py deploy_all._import_tml_chunk, ~line 2793).
    - Times, per table: physical-create, logical import (HTTP), the full
      connection scan (search_logical_tables_for_connection — include_details,
      record_size=-1: the prime suspect for "feeds through all of those"),
      and visibility. Plus a bonus re-create ("already exists") probe.
    - KEEPS everything by default (this is a volume/naming sandbox).

RESEARCH-INFORMED CONDITIONS  (ThoughtSpot docs)
    - Unique/random names avoid the same-name ambiguity the docs say slows TML
      *validation*; common words (PRODUCTS) and a table name that equals a
      column name trigger it. Including the connection `fqn` reduces ambiguity.
      Sources noted in the accompanying chat/report.
    - --import-mode async uses /api/rest/2.0/metadata/tml/async/import — the
      documented fix for import timeouts. Default is sync, to match the app.

SAFETY
    - Refuses production (se-thoughtspot-cloud) unless --allow-prod.
    - All objects are prefixed and live only in the ZPERFTEST schema.
    - Prints a plan and EXECUTES ONLY with --yes (otherwise it's a dry preview).
    - Nothing is deleted unless you pass --cleanup.

USAGE
    source ./demoprep/bin/activate
    python tests/ts_table_perf.py --env "sebe - se"            # dry preview
    python tests/ts_table_perf.py --env "sebe - se" --yes      # execute
    python tests/ts_table_perf.py --env "sebe - se" --cleanup --yes
"""
from __future__ import annotations

import argparse
import csv
import json
import os
import statistics
import sys
import time
from datetime import datetime
from pathlib import Path

import requests
import yaml
from dotenv import load_dotenv

# Repo root on path so we can import the real app modules.
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT))
load_dotenv(REPO_ROOT / ".env")

from thoughtspot_deployer import ThoughtSpotDeployer  # noqa: E402
from snowflake_auth import get_snowflake_connection  # noqa: E402

RESULTS_DIR = REPO_ROOT / "tests" / "perf_results"

# --------------------------------------------------------------------------
# TS environment resolution (mirrors chat_interface.get_ts_env_* — inlined so
# we don't import the Gradio app just to read three env vars).
# --------------------------------------------------------------------------

def resolve_env(label: str) -> tuple[str, str]:
    """Return (base_url, trusted_auth_key) for a configured TS_ENV_* label."""
    i = 1
    while True:
        env_label = os.getenv(f"TS_ENV_{i}_LABEL", "").strip()
        if not env_label:
            break
        if env_label == label:
            url = os.getenv(f"TS_ENV_{i}_URL", "").strip().rstrip("/")
            key = os.getenv(f"TS_ENV_{i}_KEY_VAR", "").strip()
            return url, key
        i += 1
    raise SystemExit(f"TS environment {label!r} is not configured in .env (TS_ENV_*_LABEL).")


def list_env_labels() -> list[str]:
    labels, i = [], 1
    while True:
        env_label = os.getenv(f"TS_ENV_{i}_LABEL", "").strip()
        if not env_label:
            break
        labels.append(env_label)
        i += 1
    return labels


def is_prod(base_url: str, label: str) -> bool:
    u, l = base_url.lower(), label.lower()
    return "se-thoughtspot-cloud" in u or l.startswith("secloud")


# --------------------------------------------------------------------------
# The 10 naming conventions.  Each column is {"name", "type"} where type is a
# Snowflake type understood by both Snowflake DDL and deployer._map_data_type.
# `use_fqn=False` deliberately omits the connection fqn from the table TML.
# --------------------------------------------------------------------------

STD_COLUMNS = [
    {"name": "ID", "type": "INT"},
    {"name": "NAME", "type": "VARCHAR"},
    {"name": "CATEGORY", "type": "VARCHAR"},
    {"name": "AMOUNT", "type": "NUMBER(18,2)"},
    {"name": "CREATED_DATE", "type": "DATE"},
]


def _naming_conditions() -> list[dict]:
    """10 distinct naming conventions, informed by the user + ThoughtSpot docs."""
    return [
        {"key": "01_random_letters", "name": "QXZJKLMNPR",
         "columns": STD_COLUMNS, "use_fqn": True,
         "why": "random unique letters — no name ambiguity (expected fast)"},
        {"key": "02_xyz_products", "name": "XYZ_PRODUCTS",
         "columns": STD_COLUMNS, "use_fqn": True,
         "why": "user-requested; uncommon prefix + common suffix"},
        {"key": "03_common_word", "name": "PRODUCTS",
         "columns": STD_COLUMNS, "use_fqn": True,
         "why": "very common word — docs: same-name ambiguity slows validation"},
        {"key": "04_name_eq_column", "name": "REVENUE",
         "columns": STD_COLUMNS + [{"name": "REVENUE", "type": "NUMBER(18,2)"}],
         "use_fqn": True,
         "why": "table name == a column name — docs: triggers validation timeout"},
        {"key": "05_fqn_included", "name": "PERF_FQN_ON",
         "columns": STD_COLUMNS, "use_fqn": True,
         "why": "connection fqn included — docs: reduces ambiguity"},
        {"key": "06_fqn_omitted", "name": "PERF_FQN_OFF",
         "columns": STD_COLUMNS, "use_fqn": False,
         "why": "connection fqn omitted (name only) — contrast to #5"},
        {"key": "07_very_long_name",
         "name": "PERF_VERY_LONG_TABLE_NAME_LENGTH_TEST_ABCDEFGHIJKLMNOP",
         "columns": STD_COLUMNS, "use_fqn": True,
         "why": "long identifier — does name length affect create time?"},
        {"key": "08_short_name", "name": "TB",
         "columns": STD_COLUMNS, "use_fqn": True,
         "why": "minimal identifier"},
        {"key": "09_digits_underscores", "name": "PERF_2026_Q3_METRICS_01",
         "columns": STD_COLUMNS, "use_fqn": True,
         "why": "digits + underscores mix"},
        {"key": "10_app_style_hash", "name": "DEMO_ZPT_A1B2C3",
         "columns": STD_COLUMNS, "use_fqn": True,
         "why": "mimics the app's real DEMO_XXX_<hash> naming"},
    ]


# --------------------------------------------------------------------------
# Snowflake helpers — create the schema + one empty physical table per name.
# --------------------------------------------------------------------------

def _sf_type(t: str) -> str:
    """Snowflake DDL column type (NUMBER/INT/VARCHAR/DATE are already valid)."""
    return t


def ensure_physical_tables(database: str, schema: str, conditions: list[dict]) -> dict:
    """CREATE SCHEMA + one empty physical table per logical name. Returns timings."""
    conn = get_snowflake_connection()
    cur = conn.cursor()
    timings = {}
    try:
        cur.execute(f'CREATE SCHEMA IF NOT EXISTS "{database}"."{schema}"')
        for c in conditions:
            name = c["name"].upper()  # create_table_tml uppercases db_table
            cols_ddl = ", ".join(f'"{col["name"].upper()}" {_sf_type(col["type"])}'
                                 for col in c["columns"])
            ddl = f'CREATE TABLE IF NOT EXISTS "{database}"."{schema}"."{name}" ({cols_ddl})'
            t0 = time.time()
            cur.execute(ddl)
            timings[c["key"]] = round(time.time() - t0, 3)
    finally:
        cur.close()
        conn.close()
    return timings


def drop_physical_tables(database: str, schema: str, conditions: list[dict]) -> None:
    conn = get_snowflake_connection()
    cur = conn.cursor()
    try:
        for c in conditions:
            name = c["name"].upper()
            cur.execute(f'DROP TABLE IF EXISTS "{database}"."{schema}"."{name}"')
    finally:
        cur.close()
        conn.close()


# --------------------------------------------------------------------------
# ThoughtSpot import / scan / delete — mirrors the app's real HTTP calls.
# --------------------------------------------------------------------------

def _normalize_import_objects(result):
    """Mirror deploy_all._normalize_tml_import_objects."""
    def _one(obj):
        if not isinstance(obj, dict):
            return obj
        if "response" in obj:
            return obj
        if "status" in obj or "header" in obj:
            return {"response": obj}
        return obj
    if isinstance(result, list):
        return [_one(o) for o in result]
    if isinstance(result, dict) and "object" in result:
        return [_one(o) for o in result["object"]]
    return None


def import_one_table(deployer: ThoughtSpotDeployer, tml: str, mode: str) -> dict:
    """Import a single table TML the way the app does; time it and read status.

    Mirrors thoughtspot_deployer.deploy_all._import_tml_chunk (sync path) and
    adds an optional async path (POST .../tml/async/import + /async/status).
    """
    base = deployer.base_url
    if mode == "async":
        return _import_async(deployer, tml)

    payload = {"metadata_tmls": [tml], "import_policy": "PARTIAL", "create_new": True}
    t0 = time.time()
    try:
        resp = deployer.session.post(
            f"{base}/api/rest/2.0/metadata/tml/import",
            json=payload,
            timeout=360,
        )
    except Exception as exc:  # network / read timeout
        return {"elapsed_s": round(time.time() - t0, 2), "http_status": None,
                "gateway_timeout": False, "status_code": "EXCEPTION",
                "guid": None, "error": f"{type(exc).__name__}: {exc}"[:400]}
    elapsed = round(time.time() - t0, 2)

    gateway_timeout = resp.status_code in (502, 503, 504)
    result = {"elapsed_s": elapsed, "http_status": resp.status_code,
              "gateway_timeout": gateway_timeout, "status_code": None,
              "guid": None, "error": None}
    if resp.status_code == 200:
        objs = _normalize_import_objects(resp.json())
        obj = (objs or [{}])[0]
        r = obj.get("response", {}) if isinstance(obj, dict) else {}
        status = r.get("status", {})
        header = r.get("header", {})
        result["status_code"] = status.get("status_code")
        result["guid"] = header.get("id_guid")
        if status.get("error_message"):
            result["error"] = str(status["error_message"])[:400]
    else:
        result["error"] = resp.text[:400]
    return result


def _import_async(deployer: ThoughtSpotDeployer, tml: str) -> dict:
    """Documented fix path: async import + poll status. Timed end-to-end."""
    base = deployer.base_url
    t0 = time.time()
    try:
        resp = deployer.session.post(
            f"{base}/api/rest/2.0/metadata/tml/async/import",
            json={"metadata_tmls": [tml], "import_policy": "PARTIAL", "create_new": True},
            timeout=60,
        )
    except Exception as exc:
        return {"elapsed_s": round(time.time() - t0, 2), "http_status": None,
                "gateway_timeout": False, "status_code": "EXCEPTION",
                "guid": None, "error": f"{type(exc).__name__}: {exc}"[:400]}
    if resp.status_code not in (200, 202):
        return {"elapsed_s": round(time.time() - t0, 2), "http_status": resp.status_code,
                "gateway_timeout": resp.status_code in (502, 503, 504),
                "status_code": "SUBMIT_FAILED", "guid": None, "error": resp.text[:400]}
    task_id = (resp.json() or {}).get("task_id") or (resp.json() or {}).get("id")
    # Poll status until terminal or 6 min.
    deadline = time.time() + 360
    last = None
    while time.time() < deadline:
        try:
            s = deployer.session.post(
                f"{base}/api/rest/2.0/metadata/tml/async/status",
                json={"task_ids": [task_id], "include_import_response": True},
                timeout=60,
            )
            last = s.json() if s.status_code == 200 else s.text[:200]
            if s.status_code == 200:
                blob = json.dumps(last).upper()
                if any(k in blob for k in ('"OK"', "COMPLETED", "SUCCESS", "FAILED", "ERROR")):
                    break
        except Exception:
            pass
        time.sleep(3)
    return {"elapsed_s": round(time.time() - t0, 2), "http_status": resp.status_code,
            "gateway_timeout": False, "status_code": "ASYNC_DONE",
            "guid": None, "error": None, "async_status": str(last)[:400], "task_id": task_id}


def time_connection_scan(deployer: ThoughtSpotDeployer, conn_guid: str, conn_name: str) -> dict:
    """Time the full-connection detailed scan — the prime slowness suspect.

    This is exactly search_logical_tables_for_connection: metadata/search on the
    CONNECTION with include_details=True, record_size=-1.
    """
    t0 = time.time()
    resolved = deployer.search_logical_tables_for_connection(conn_guid, conn_name)
    return {"scan_s": round(time.time() - t0, 2), "table_count": len(resolved)}


def delete_logical_table(deployer: ThoughtSpotDeployer, guid: str) -> dict:
    """POST /metadata/delete (the app has no delete today). Timed."""
    t0 = time.time()
    resp = deployer.session.post(
        f"{deployer.base_url}/api/rest/2.0/metadata/delete",
        json={"metadata": [{"type": "LOGICAL_TABLE", "identifier": guid}]},
        timeout=120,
    )
    return {"elapsed_s": round(time.time() - t0, 2), "http_status": resp.status_code,
            "error": None if resp.status_code in (200, 204) else resp.text[:300]}


# --------------------------------------------------------------------------
# Reporting
# --------------------------------------------------------------------------

def write_reports(stamp: str, meta: dict, rows: list[dict]) -> tuple[Path, Path]:
    RESULTS_DIR.mkdir(parents=True, exist_ok=True)
    csv_path = RESULTS_DIR / f"{stamp}_ts_table_perf.csv"
    md_path = RESULTS_DIR / f"{stamp}_ts_table_perf.md"

    fields = ["condition", "logical_name", "use_fqn", "physical_create_s",
              "import_s", "http_status", "gateway_timeout", "import_status",
              "conn_scan_s", "conn_table_count", "guid", "why", "error"]
    with csv_path.open("w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        w.writeheader()
        for r in rows:
            w.writerow(r)

    ok = [r for r in rows if r.get("import_s") is not None]
    imports = [r["import_s"] for r in ok if isinstance(r.get("import_s"), (int, float))]
    scans = [r["conn_scan_s"] for r in ok if isinstance(r.get("conn_scan_s"), (int, float))]

    def _stat(vals):
        if not vals:
            return "n/a"
        vals = sorted(vals)
        p95 = vals[min(len(vals) - 1, int(round(0.95 * (len(vals) - 1))))]
        return (f"min {min(vals):.2f}s / median {statistics.median(vals):.2f}s / "
                f"p95 {p95:.2f}s / max {max(vals):.2f}s")

    lines = [
        f"# ThoughtSpot table-create perf — {stamp}", "",
        f"- Environment: **{meta['env']}** ({meta['base_url']})",
        f"- User: {meta['user']}",
        f"- Connection: **{meta['connection']}** (`{meta['connection_guid']}`)",
        f"- Snowflake: {meta['database']}.{meta['schema']}",
        f"- Import mode: **{meta['import_mode']}**",
        f"- Connection create time: **{meta['connection_create_s']}s**",
        "",
        "## Summary", "",
        f"- Import time across {len(imports)} tables: {_stat(imports)}",
        f"- Connection-scan time (grows with table count): {_stat(scans)}",
        f"- Gateway timeouts (502/503/504): "
        f"**{sum(1 for r in rows if r.get('gateway_timeout'))}/{len(rows)}**",
    ]
    if meta.get("recreate_probe"):
        rp = meta["recreate_probe"]
        lines += ["",
                  f"- **Re-create ('already exists') probe** on `{rp['name']}`: "
                  f"import {rp['import_s']}s, status {rp['import_status']}, "
                  f"conn-scan {rp['conn_scan_s']}s"]

    lines += ["", "## Per-condition (slowest import first)", "",
              "| condition | name | fqn | import s | conn-scan s | http | 504? | status | why |",
              "|---|---|---|--:|--:|--:|:-:|---|---|"]
    for r in sorted(rows, key=lambda x: (x.get("import_s") or -1), reverse=True):
        lines.append(
            f"| {r['condition']} | `{r['logical_name']}` | {'y' if r['use_fqn'] else 'n'} "
            f"| {r.get('import_s')} | {r.get('conn_scan_s')} | {r.get('http_status')} "
            f"| {'⚠️' if r.get('gateway_timeout') else ''} | {r.get('import_status')} "
            f"| {r.get('why','')} |")
    md_path.write_text("\n".join(lines) + "\n")
    return csv_path, md_path


def print_console(rows: list[dict], meta: dict) -> None:
    print("\n" + "=" * 78)
    print(f"RESULTS — {meta['env']} — connection {meta['connection']} "
          f"(created in {meta['connection_create_s']}s)")
    print("=" * 78)
    hdr = f"{'condition':22} {'import_s':>9} {'scan_s':>8} {'http':>5} {'504':>4} {'status':>8}"
    print(hdr)
    print("-" * len(hdr))
    for r in sorted(rows, key=lambda x: (x.get("import_s") or -1), reverse=True):
        print(f"{r['condition']:22} {str(r.get('import_s')):>9} {str(r.get('conn_scan_s')):>8} "
              f"{str(r.get('http_status')):>5} {'yes' if r.get('gateway_timeout') else '':>4} "
              f"{str(r.get('import_status')):>8}")
    print("-" * len(hdr))


# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------

def main() -> int:
    ap = argparse.ArgumentParser(description="ThoughtSpot table-create perf harness")
    ap.add_argument("--env", required=True, help="TS environment label (see .env TS_ENV_*_LABEL)")
    ap.add_argument("--connection-name", default="ZPERFTEST_DONOTDELETE",
                    help="Name of the dedicated test connection to create/reuse")
    ap.add_argument("--database", default="DEMOBUILD")
    ap.add_argument("--schema", default="ZPERFTEST")
    ap.add_argument("--import-mode", choices=["sync", "async"], default="sync",
                    help="sync = /metadata/tml/import (matches app); async = documented fix")
    ap.add_argument("--recreate-probe", action="store_true", default=True,
                    help="After the 10, re-create table #1 to time the 'already exists' path")
    ap.add_argument("--no-recreate-probe", dest="recreate_probe", action="store_false")
    ap.add_argument("--cleanup", action="store_true",
                    help="Delete the logical + physical test tables at the end")
    ap.add_argument("--drop-connection", action="store_true",
                    help="With --cleanup, also delete the test connection")
    ap.add_argument("--allow-prod", action="store_true", help="Permit running against prod")
    ap.add_argument("--yes", action="store_true", help="Execute (without this it's a dry preview)")
    args = ap.parse_args()

    base_url, key = resolve_env(args.env)
    if not base_url or not key:
        raise SystemExit(f"Environment {args.env!r} missing URL or trusted-auth key in .env.\n"
                         f"Configured: {list_env_labels()}")
    user = os.getenv("TEST_USER") or os.getenv("THOUGHTSPOT_USERNAME")
    if not user:
        raise SystemExit("No user identity — set TEST_USER in .env.")

    if is_prod(base_url, args.env) and not args.allow_prod:
        raise SystemExit(f"REFUSING prod ({base_url}). This creates/keeps objects. "
                         f"Pass --allow-prod to override (not recommended).")

    conditions = _naming_conditions()

    # -------- Plan (always printed) --------
    print("=" * 78)
    print("ThoughtSpot table-create PERFORMANCE harness — PLAN")
    print("=" * 78)
    print(f"  Environment      : {args.env}  ({base_url})")
    print(f"  User             : {user}")
    print(f"  Connection       : {args.connection_name}  (KEPT unless --drop-connection)")
    print(f"  Snowflake target : {args.database}.{args.schema}")
    print(f"  Import mode      : {args.import_mode}")
    print(f"  Tables ({len(conditions)}), one per naming convention:")
    for c in conditions:
        print(f"     - {c['key']:22} name={c['name']!r:56} fqn={c['use_fqn']}  # {c['why']}")
    print(f"  Re-create probe  : {args.recreate_probe}")
    print(f"  Cleanup at end   : {args.cleanup} (drop connection: {args.drop_connection})")
    if not args.yes:
        print("\nDRY PREVIEW — nothing created. Re-run with --yes to execute.")
        return 0

    # -------- Execute --------
    print("\n▶ Authenticating…")
    deployer = ThoughtSpotDeployer(base_url=base_url, username=user, secret_key=key)
    if not deployer.authenticate():
        raise SystemExit(f"Auth failed: {deployer.last_auth_error}")
    print("  ✅ authenticated")

    print(f"▶ Ensuring physical tables in {args.database}.{args.schema}…")
    phys_timings = ensure_physical_tables(args.database, args.schema, conditions)
    print(f"  ✅ {len(phys_timings)} physical tables ready "
          f"(max {max(phys_timings.values()):.2f}s)")

    print(f"▶ Creating our own connection {args.connection_name!r}…")
    t0 = time.time()
    conn_guid, conn_fqn = deployer.create_connection_with_reconcile(
        args.connection_name, args.database, log_progress=lambda m: print(f"    {m}"))
    conn_create_s = round(time.time() - t0, 2)
    if not conn_guid:
        raise SystemExit("Connection creation failed — see log above.")
    print(f"  ✅ connection guid={conn_guid} fqn={conn_fqn} ({conn_create_s}s)")

    meta = {
        "env": args.env, "base_url": base_url, "user": user,
        "connection": args.connection_name, "connection_guid": conn_guid,
        "database": args.database, "schema": args.schema,
        "import_mode": args.import_mode, "connection_create_s": conn_create_s,
    }

    rows: list[dict] = []
    created_guids: list[str] = []
    for i, c in enumerate(conditions, 1):
        print(f"\n[{i}/{len(conditions)}] {c['key']}  name={c['name']!r}")
        tml = deployer.create_table_tml(
            c["name"], c["columns"], args.connection_name,
            args.database, args.schema,
            all_tables=None, foreign_keys=None,
            connection_fqn=conn_fqn if c["use_fqn"] else None,
        )
        imp = import_one_table(deployer, tml, args.import_mode)
        scan = time_connection_scan(deployer, conn_guid, args.connection_name)
        if imp.get("guid"):
            created_guids.append(imp["guid"])
        print(f"    import={imp['elapsed_s']}s http={imp['http_status']} "
              f"status={imp['status_code']} 504={imp['gateway_timeout']} "
              f"conn_scan={scan['scan_s']}s (tables now {scan['table_count']})"
              + (f"  ERROR: {imp['error']}" if imp.get("error") else ""))
        rows.append({
            "condition": c["key"], "logical_name": c["name"], "use_fqn": c["use_fqn"],
            "physical_create_s": phys_timings.get(c["key"]),
            "import_s": imp["elapsed_s"], "http_status": imp["http_status"],
            "gateway_timeout": imp["gateway_timeout"], "import_status": imp["status_code"],
            "conn_scan_s": scan["scan_s"], "conn_table_count": scan["table_count"],
            "guid": imp.get("guid"), "why": c["why"], "error": imp.get("error"),
        })

    # -------- Bonus: re-create probe (the 'already exists' path) --------
    if args.recreate_probe:
        c = conditions[0]
        print(f"\n▶ Re-create probe: importing {c['name']!r} AGAIN (create_new) "
              f"to time the 'already exists' path…")
        tml = deployer.create_table_tml(
            c["name"], c["columns"], args.connection_name, args.database, args.schema,
            all_tables=None, foreign_keys=None, connection_fqn=conn_fqn)
        imp = import_one_table(deployer, tml, args.import_mode)
        scan = time_connection_scan(deployer, conn_guid, args.connection_name)
        print(f"    re-create import={imp['elapsed_s']}s status={imp['status_code']} "
              f"conn_scan={scan['scan_s']}s"
              + (f"  msg: {imp['error']}" if imp.get("error") else ""))
        meta["recreate_probe"] = {"name": c["name"], "import_s": imp["elapsed_s"],
                                  "import_status": imp["status_code"], "conn_scan_s": scan["scan_s"]}

    # -------- Reports --------
    stamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    csv_path, md_path = write_reports(stamp, meta, rows)
    print_console(rows, meta)
    print(f"\n  CSV : {csv_path}")
    print(f"  MD  : {md_path}")

    # -------- Cleanup (opt-in) --------
    if args.cleanup:
        print("\n▶ Cleanup: deleting logical + physical test tables…")
        for guid in created_guids:
            d = delete_logical_table(deployer, guid)
            print(f"    delete {guid}: http={d['http_status']} {d['elapsed_s']}s"
                  + (f"  {d['error']}" if d.get("error") else ""))
        drop_physical_tables(args.database, args.schema, conditions)
        print("  ✅ physical tables dropped")
        if args.drop_connection:
            d = delete_logical_table(deployer, conn_guid)  # CONNECTION delete via metadata/delete
            deployer.session.post(
                f"{base_url}/api/rest/2.0/metadata/delete",
                json={"metadata": [{"type": "CONNECTION", "identifier": conn_guid}]}, timeout=120)
            print("  ✅ connection deleted")
    else:
        print("\n  (kept all objects — pass --cleanup to remove)")

    return 0


if __name__ == "__main__":
    sys.exit(main())