#!/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_ 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())