#!/usr/bin/env python3
"""Collect read-only TiDB-side SQL timing evidence for a Test runtime.

The Runtime's driver timing is intentionally an end-to-end client measurement:
it includes network and result consumption.  This collector queries TiDB's own
statement summary for server-side latency components and can execute a bounded
EXPLAIN ANALYZE suite for the Catalog watch reads.  It never executes DML.
"""

from __future__ import annotations

import argparse
import re
import sys
from datetime import datetime, timezone
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "deployments" / "offline" / "runtime"))

from common import DeployError, db_query_scalar, load_env_file  # noqa: E402


CATALOG_TABLE_FILTER = """(
  LOWER(DIGEST_TEXT) LIKE '%_system_catalogappinstallation%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_catalogappoperation%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_catalogconsumergrantset%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_catalogconsumergrant%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_catalogpublicationactive%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_catalogpublication%'
)"""

DIAGNOSTIC_TABLE_FILTER = """(
  {catalog}
  OR LOWER(DIGEST_TEXT) LIKE '%fieldpermissions%'
  OR LOWER(DIGEST_TEXT) LIKE '%objectpermissions%'
  OR LOWER(DIGEST_TEXT) LIKE '%permissionsetpermission%'
  OR LOWER(DIGEST_TEXT) LIKE '%permissionsetassignment%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_scheduledrunreservation%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_recordeventoutbox%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_integrationoutboundcommand%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_session%'
  OR LOWER(DIGEST_TEXT) LIKE '%_system_runtime%security%'
)""".format(catalog=CATALOG_TABLE_FILTER)

SUMMARY_DETAIL_ROW_LIMIT = 1000
BUCKETED_SUMMARY_DETAIL_ROW_LIMIT = 10000
OUTBOUND_QUEUE_SAMPLE_LIMIT = 5000
STATEMENT_DIGEST_COLUMNS = (
    "digest",
    "plan_digest",
    "statement_type",
    "resource_group",
    "evidence_owner",
    "executions",
    "tidb_total_latency_s",
    "tidb_avg_latency_ms",
    "tidb_max_latency_ms",
    "avg_resource_queue_ms",
    "max_resource_queue_ms",
    "avg_request_unit_read",
    "max_request_unit_read",
    "avg_total_keys",
    "avg_processed_keys",
    "first_seen",
    "last_seen",
)
STATEMENT_SUMMARY_HEADER = (
    "digest\tplan_digest\tstatement_type\tresource_group\tevidence_owner\t"
    "executions\ttidb_total_latency_s\t"
    "tidb_avg_latency_ms\ttidb_max_latency_ms\tavg_parse_ms\tavg_compile_ms\t"
    "avg_tikv_process_ms\tavg_tikv_wait_ms\tavg_backoff_ms\t"
    "avg_kv_ms\tavg_tikv_cpu_ms\tavg_resource_queue_ms\tmax_resource_queue_ms\t"
    "avg_request_unit_read\tmax_request_unit_read\t"
    "avg_remote_read_wait_ms\tavg_backoff_total_ms\tavg_total_keys\t"
    "avg_processed_keys\tavg_prewrite_ms\tavg_commit_ms\tavg_get_commit_ts_ms\t"
    "avg_resolve_lock_ms\tavg_local_latch_wait_ms\tfirst_seen\tlast_seen\t"
    "normalized_sql"
)


def parse_utc(value: str) -> str:
    candidate = value.strip()
    if candidate.endswith("Z"):
        candidate = candidate[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(candidate)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            "timestamp must be RFC3339, for example 2026-08-13T18:44:35Z"
        ) from exc
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    parsed = parsed.astimezone(timezone.utc)
    return parsed.strftime("%Y-%m-%d %H:%M:%S.%f")


def statement_summary_sql(
    since: str, until: str, *, focused: bool, bucketed: bool = False
) -> str:
    focus = f"AND {DIAGNOSTIC_TABLE_FILTER}" if focused else ""
    bucket_columns = (
        "SUMMARY_BEGIN_TIME AS summary_begin,\n"
        "  SUMMARY_END_TIME AS summary_end,\n  "
        if bucketed
        else ""
    )
    bucket_group = (
        "SUMMARY_BEGIN_TIME, SUMMARY_END_TIME, " if bucketed else ""
    )
    order = (
        "SUMMARY_BEGIN_TIME, SUM(SUM_LATENCY) DESC"
        if bucketed
        else "SUM(SUM_LATENCY) DESC, SUM(EXEC_COUNT) DESC"
    )
    row_limit = (
        BUCKETED_SUMMARY_DETAIL_ROW_LIMIT
        if bucketed
        else SUMMARY_DETAIL_ROW_LIMIT
    )
    return f"""
SELECT
  {bucket_columns}LEFT(DIGEST, 16) AS digest,
  LEFT(COALESCE(PLAN_DIGEST, ''), 16) AS plan_digest,
  STMT_TYPE AS statement_type,
  COALESCE(RESOURCE_GROUP, '') AS resource_group,
  CASE
    WHEN LOWER(DIGEST_TEXT) LIKE '%nexuscrm_test2_observer:tidb_pressure_v1%'
    THEN 'test2-observer'
    WHEN LOWER(DIGEST_TEXT) LIKE '%nexuscrm_test3_observer:tidb_pressure_v1%'
    THEN 'test3-observer'
    WHEN LOWER(DIGEST_TEXT) LIKE '%information_schema.cluster_processlist%'
      OR LOWER(DIGEST_TEXT) LIKE '%information_schema.data_lock_waits%'
    THEN 'test-observer'
    ELSE 'runtime-or-other'
  END AS evidence_owner,
  SUM(EXEC_COUNT) AS executions,
  ROUND(SUM(SUM_LATENCY) / 1000000000, 6) AS tidb_total_latency_s,
  ROUND(SUM(SUM_LATENCY) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS tidb_avg_latency_ms,
  ROUND(MAX(MAX_LATENCY) / 1000000, 3) AS tidb_max_latency_ms,
  ROUND(SUM(AVG_PARSE_LATENCY * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_parse_ms,
  ROUND(SUM(AVG_COMPILE_LATENCY * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_compile_ms,
  ROUND(SUM(AVG_PROCESS_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_tikv_process_ms,
  ROUND(SUM(AVG_WAIT_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_tikv_wait_ms,
  ROUND(SUM(AVG_BACKOFF_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_backoff_ms,
  ROUND(SUM(AVG_KV_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_kv_ms,
  ROUND(SUM(AVG_TIKV_CPU_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_tikv_cpu_ms,
  ROUND(SUM(AVG_QUEUED_RC_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_resource_queue_ms,
  ROUND(MAX(MAX_QUEUED_RC_TIME) / 1000000, 3) AS max_resource_queue_ms,
  ROUND(SUM(AVG_REQUEST_UNIT_READ * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0), 3)
    AS avg_request_unit_read,
  ROUND(MAX(MAX_REQUEST_UNIT_READ), 3) AS max_request_unit_read,
  ROUND(SUM(AVG_IA_REMOTE_READ_SEGMENT_WAIT_TIME * EXEC_COUNT) /
    NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3) AS avg_remote_read_wait_ms,
  ROUND(SUM(AVG_BACKOFF_TOTAL_TIME * EXEC_COUNT) /
    NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3) AS avg_backoff_total_ms,
  ROUND(SUM(AVG_TOTAL_KEYS * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0), 1)
    AS avg_total_keys,
  ROUND(SUM(AVG_PROCESSED_KEYS * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0), 1)
    AS avg_processed_keys,
  ROUND(SUM(AVG_PREWRITE_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_prewrite_ms,
  ROUND(SUM(AVG_COMMIT_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_commit_ms,
  ROUND(SUM(AVG_GET_COMMIT_TS_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_get_commit_ts_ms,
  ROUND(SUM(AVG_RESOLVE_LOCK_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_resolve_lock_ms,
  ROUND(SUM(AVG_LOCAL_LATCH_WAIT_TIME * EXEC_COUNT) / NULLIF(SUM(EXEC_COUNT), 0) / 1000000, 3)
    AS avg_local_latch_wait_ms,
  MIN(FIRST_SEEN) AS first_seen,
  MAX(LAST_SEEN) AS last_seen,
  LEFT(REPLACE(REPLACE(DIGEST_TEXT, CHAR(9), ' '), CHAR(10), ' '), 4000)
    AS normalized_sql
FROM INFORMATION_SCHEMA.CLUSTER_STATEMENTS_SUMMARY_HISTORY
WHERE SCHEMA_NAME = DATABASE()
  AND LAST_SEEN >= TIMESTAMP('{since}')
  AND FIRST_SEEN <= TIMESTAMP('{until}')
  {focus}
GROUP BY {bucket_group}DIGEST, PLAN_DIGEST, STMT_TYPE, RESOURCE_GROUP, DIGEST_TEXT
ORDER BY {order}
LIMIT {row_limit}
""".strip()


def statement_bucket_totals_sql(since: str, until: str) -> str:
    """Return complete per-bucket totals independent of digest detail caps."""
    return f"""
SELECT
  SUMMARY_BEGIN_TIME AS summary_begin,
  SUMMARY_END_TIME AS summary_end,
  SUM(EXEC_COUNT) AS executions,
  ROUND(SUM(SUM_LATENCY) / 1000000000, 6) AS tidb_total_latency_s,
  ROUND(SUM(AVG_QUEUED_RC_TIME * EXEC_COUNT) / 1000000000, 6)
    AS resource_queue_s,
  COUNT(*) AS digest_rows
FROM INFORMATION_SCHEMA.CLUSTER_STATEMENTS_SUMMARY_HISTORY
WHERE SCHEMA_NAME = DATABASE()
  AND LAST_SEEN >= TIMESTAMP('{since}')
  AND FIRST_SEEN <= TIMESTAMP('{until}')
GROUP BY SUMMARY_BEGIN_TIME, SUMMARY_END_TIME
ORDER BY SUMMARY_BEGIN_TIME
""".strip()


def slow_query_sql(since: str, until: str) -> str:
    slow_filter = DIAGNOSTIC_TABLE_FILTER.replace("DIGEST_TEXT", "QUERY")
    return f"""
SELECT
  TIME AS observed_at,
  LEFT(DIGEST, 16) AS digest,
  LEFT(COALESCE(PLAN_DIGEST, ''), 16) AS plan_digest,
  ROUND(QUERY_TIME * 1000, 3) AS tidb_latency_ms,
  ROUND(PARSE_TIME * 1000, 3) AS parse_ms,
  ROUND(COMPILE_TIME * 1000, 3) AS compile_ms,
  ROUND(WAIT_TS * 1000, 3) AS wait_ts_ms,
  ROUND(PROCESS_TIME * 1000, 3) AS tikv_process_ms,
  ROUND(WAIT_TIME * 1000, 3) AS tikv_wait_ms,
  ROUND(BACKOFF_TIME * 1000, 3) AS backoff_ms,
  TOTAL_KEYS AS total_keys,
  PROCESS_KEYS AS processed_keys,
  REQUEST_COUNT AS request_count,
  ROUND(COP_PROC_AVG * 1000, 3) AS cop_process_avg_ms,
  ROUND(COP_WAIT_AVG * 1000, 3) AS cop_wait_avg_ms,
  COALESCE(BACKOFF_TYPES, '') AS backoff_types,
  PLAN_FROM_CACHE AS plan_from_cache
FROM INFORMATION_SCHEMA.CLUSTER_SLOW_QUERY
WHERE DB = DATABASE()
  AND TIME >= TIMESTAMP('{since}')
  AND TIME <= TIMESTAMP('{until}')
  AND {slow_filter}
ORDER BY QUERY_TIME DESC
LIMIT 1000
""".strip()


def outbound_queue_snapshot_sql() -> str:
    """Aggregate a bounded recent queue sample without exposing command payloads."""
    return f"""
SELECT
  'latest_{OUTBOUND_QUEUE_SAMPLE_LIMIT}_acceptances' AS sample_window,
  recent.CatalogRevision AS catalog_revision,
  recent.State AS state,
  recent.Operation AS operation,
  LEFT(REPLACE(REPLACE(COALESCE(recent.ConnectorOwnerKey, ''), CHAR(9), ' '), CHAR(10), ' '), 160)
    AS connector_owner_key,
  LEFT(REPLACE(REPLACE(COALESCE(recent.FailureCode, ''), CHAR(9), ' '), CHAR(10), ' '), 160)
    AS failure_code,
  COUNT(*) AS command_count,
  COALESCE(SUM(recent.RetryCount), 0) AS retry_count_total,
  SUM(CASE
        WHEN recent.State IN ('pending', 'retry')
         AND (recent.NextAttemptAt IS NULL OR recent.NextAttemptAt <= NOW(6))
        THEN 1 ELSE 0
      END) AS due_count,
  MIN(recent.CreatedDate) AS oldest_created_at,
  MAX(recent.LastModifiedDate) AS newest_modified_at
FROM (
  SELECT
    AcceptanceSequence, CatalogRevision, State, Operation, ConnectorOwnerKey,
    FailureCode, RetryCount, NextAttemptAt, CreatedDate, LastModifiedDate
  FROM _System_IntegrationOutboundCommand
  ORDER BY AcceptanceSequence DESC
  LIMIT {OUTBOUND_QUEUE_SAMPLE_LIMIT}
) AS recent
GROUP BY recent.CatalogRevision, recent.State, recent.Operation,
         recent.ConnectorOwnerKey, recent.FailureCode
ORDER BY command_count DESC, retry_count_total DESC
LIMIT 200
""".strip()


def outbound_command_indexes_sql() -> str:
    return """
SELECT
  INDEX_NAME AS index_name,
  NON_UNIQUE AS non_unique,
  SEQ_IN_INDEX AS sequence_in_index,
  COLUMN_NAME AS column_name,
  COALESCE(SUB_PART, '') AS prefix_length
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_NAME = '_System_IntegrationOutboundCommand'
ORDER BY INDEX_NAME, SEQ_IN_INDEX
""".strip()


def outbound_claim_explain_sql() -> str:
    """Explain the exact lane-order candidate shape against one live scope."""
    return """
SET @outbound_probe_id = (
  SELECT Id
  FROM _System_IntegrationOutboundCommand
  ORDER BY AcceptanceSequence DESC
  LIMIT 1
);
SET @outbound_scope_digest = (
  SELECT ScopeDigest FROM _System_IntegrationOutboundCommand
  WHERE Id = @outbound_probe_id
);
SET @outbound_org_id = (
  SELECT OrgId FROM _System_IntegrationOutboundCommand
  WHERE Id = @outbound_probe_id
);
SET @outbound_installation_id = (
  SELECT AppInstallationId FROM _System_IntegrationOutboundCommand
  WHERE Id = @outbound_probe_id
);
SET @outbound_workload_id = (
  SELECT AuthorizedWorkloadId FROM _System_IntegrationOutboundCommand
  WHERE Id = @outbound_probe_id
);
SET @outbound_catalog_revision = (
  SELECT CatalogRevision FROM _System_IntegrationOutboundCommand
  WHERE Id = @outbound_probe_id
);
SELECT CONCAT(
  'probe_bindings outbound_command=', @outbound_probe_id IS NOT NULL,
  ' scope=', @outbound_scope_digest IS NOT NULL,
  ' authority=', @outbound_org_id IS NOT NULL
    AND @outbound_installation_id IS NOT NULL
    AND @outbound_workload_id IS NOT NULL,
  ' revision=', @outbound_catalog_revision IS NOT NULL
);
EXPLAIN FORMAT = 'brief'
SELECT candidate.Id, candidate.AcceptanceSequence, candidate.LinkKey,
       candidate.State, candidate.NextAttemptAt, candidate.CreatedDate
FROM _System_IntegrationOutboundCommand AS candidate
WHERE candidate.ScopeDigest = @outbound_scope_digest
  AND candidate.OrgId = @outbound_org_id
  AND candidate.AppInstallationId = @outbound_installation_id
  AND candidate.AuthorizedWorkloadId = @outbound_workload_id
  AND candidate.CatalogRevision = @outbound_catalog_revision
  AND candidate.State IN ('pending', 'retry')
  AND (candidate.NextAttemptAt IS NULL OR candidate.NextAttemptAt <= NOW(6))
  AND NOT EXISTS (
    SELECT 1 FROM _System_IntegrationOutboundCommand AS prior
    WHERE prior.ScopeDigest = candidate.ScopeDigest
      AND prior.OrgId = candidate.OrgId
      AND prior.AppInstallationId = candidate.AppInstallationId
      AND prior.AuthorizedWorkloadId = candidate.AuthorizedWorkloadId
      AND prior.CatalogRevision = candidate.CatalogRevision
      AND prior.LinkKey = candidate.LinkKey
      AND prior.State IN ('pending', 'claimed', 'dispatching', 'retry', 'writeback', 'uncertain')
      AND prior.AcceptanceSequence < candidate.AcceptanceSequence
  )
ORDER BY COALESCE(candidate.NextAttemptAt, candidate.CreatedDate),
         candidate.AcceptanceSequence
LIMIT 1 FOR UPDATE;
""".strip()


def catalog_explain_sql(repetitions: int) -> str:
    statements = [
        "SET @operation_id = (SELECT o.OperationId "
        "FROM _System_CatalogAppOperation o "
        "JOIN _System_CatalogConsumerGrantSet g ON g.OperationId = o.OperationId "
        "JOIN _System_CatalogPublicationActive p ON p.OrgId = o.OrgId "
        "AND p.ActivePublicationPresent = 1 "
        "AND p.ActivePublicationDigest IS NOT NULL "
        "WHERE o.State = 'ACTIVE' AND o.Step = 'ACTIVATED' "
        "ORDER BY o.OperationId LIMIT 1)",
        "SET @installation_id = (SELECT AppInstallationId "
        "FROM _System_CatalogAppOperation WHERE OperationId = @operation_id)",
        "SET @org_id = (SELECT OrgId FROM _System_CatalogAppOperation "
        "WHERE OperationId = @operation_id)",
        "SET @app_id = (SELECT AppId FROM _System_CatalogAppOperation "
        "WHERE OperationId = @operation_id)",
        "SET @release_digest = (SELECT TargetReleaseDigest "
        "FROM _System_CatalogAppOperation WHERE OperationId = @operation_id)",
        "SET @catalog_revision = (SELECT ExpectedCatalogRevision "
        "FROM _System_CatalogAppOperation WHERE OperationId = @operation_id)",
        "SET @publication_digest = (SELECT ActivePublicationDigest "
        "FROM _System_CatalogPublicationActive WHERE OrgId = @org_id)",
        "SELECT CONCAT('probe_bindings installation=', @installation_id IS NOT NULL, "
        "' operation=', @operation_id IS NOT NULL, "
        "' publication=', @publication_digest IS NOT NULL)",
    ]
    probes = [
        (
            "catalog_installation",
            "SELECT OrgId, AppId, State, ActiveReleaseDigest, CatalogRevision, "
            "ActiveAuthorizedWorkloads, ActiveGrantedCapabilities "
            "FROM _System_CatalogAppInstallation "
            "WHERE AppInstallationId = @installation_id",
        ),
        (
            "catalog_active_operation",
            "SELECT OperationId FROM _System_CatalogAppOperation "
            "WHERE AppInstallationId = @installation_id AND OrgId = @org_id "
            "AND AppId = @app_id AND State = 'ACTIVE' AND Step = 'ACTIVATED' "
            "AND TargetReleaseDigest = @release_digest "
            "AND ExpectedCatalogRevision = @catalog_revision "
            "ORDER BY OperationId LIMIT 2",
        ),
        (
            "catalog_grant_set",
            "SELECT AppInstallationId, GrantSetDigest, GrantCount "
            "FROM _System_CatalogConsumerGrantSet WHERE OperationId = @operation_id",
        ),
        (
            "catalog_consumer_grants",
            "SELECT OperationId, OrgId, AppInstallationId, AuthorizedWorkloadId, CapabilityId "
            "FROM _System_CatalogConsumerGrant WHERE OperationId = @operation_id "
            "ORDER BY BINARY AuthorizedWorkloadId, BINARY CapabilityId",
        ),
        (
            "catalog_publication_active",
            "SELECT ActivePublicationPresent, ActivePublicationDigest "
            "FROM _System_CatalogPublicationActive WHERE OrgId = @org_id",
        ),
        (
            "catalog_publication",
            "SELECT OrgId, CatalogRevision, CatalogSnapshotDigest, SchemaGeneration, "
            "EngineGeneration, LayoutValidationViewDigest, LayoutActivationSetDigest, "
            "SecurityEpoch, LayoutRoutingSecurityViewDigest "
            "FROM _System_CatalogPublication "
            "WHERE OrgId = @org_id AND PublicationDigest = @publication_digest",
        ),
    ]
    for repetition in range(1, repetitions + 1):
        for name, query in probes:
            statements.append(f"SELECT 'probe={name} repetition={repetition}'")
            statements.append(f"EXPLAIN ANALYZE {query}")
    return ";\n".join(statements) + ";"


def scheduled_reclaim_explain_sql(repetitions: int) -> str:
    """Compare read-only scan and digest-qualified plans on one bound row."""
    projection = (
        "SELECT Id, COALESCE(OccurrenceIdentityDigest, ''), State, Status, "
        "COALESCE(WindowToken, ''), Attempt, COALESCE(LeaseToken, ''), "
        "LeaseExpiresAt, COALESCE(CAST(OwnershipEpoch AS CHAR), '0'), "
        "COALESCE(MetadataBinding, ''), InputPayloadFormatVersion, InputPayload, "
        "COALESCE(EntryElement, ''), DueAt "
        "FROM _System_ScheduledRunReservation WHERE "
    )
    authority = (
        "OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ? "
        "AND FlowDefinitionId = ? AND FlowVersionId = ? "
        "AND ScheduleWindowKey = ? AND InputKey = ?"
    )
    statements = [
        "SET @scheduled_reservation_id = (SELECT Id "
        "FROM _System_ScheduledRunReservation "
        "WHERE OccurrenceIdentityDigest IS NOT NULL "
        "AND OccurrenceIdentityDigest <> '' "
        "AND OrgId IS NOT NULL AND AppInstallationId IS NOT NULL "
        "AND RuntimeCapabilityId IS NOT NULL "
        "ORDER BY OccurrenceIdentityDigest LIMIT 1)",
    ]
    bindings = {
        "scheduled_digest": "OccurrenceIdentityDigest",
        "scheduled_org_id": "OrgId",
        "scheduled_installation_id": "AppInstallationId",
        "scheduled_capability_id": "RuntimeCapabilityId",
        "scheduled_flow_definition_id": "FlowDefinitionId",
        "scheduled_flow_version_id": "FlowVersionId",
        "scheduled_window_key": "ScheduleWindowKey",
        "scheduled_input_key": "InputKey",
    }
    for variable, column in bindings.items():
        statements.append(
            f"SET @{variable} = (SELECT {column} "
            "FROM _System_ScheduledRunReservation "
            "WHERE Id = @scheduled_reservation_id)"
        )
    statements.append(
        "SELECT CONCAT('probe_bindings scheduled_reservation=', "
        "@scheduled_reservation_id IS NOT NULL, ' digest=', "
        "@scheduled_digest IS NOT NULL)"
    )
    authority_bindings = (
        "@scheduled_org_id, @scheduled_installation_id, "
        "@scheduled_capability_id, @scheduled_flow_definition_id, "
        "@scheduled_flow_version_id, @scheduled_window_key, "
        "@scheduled_input_key"
    )
    for repetition in range(1, repetitions + 1):
        scan_sql = f"EXPLAIN ANALYZE {projection}{authority}"
        point_sql = (
            f"EXPLAIN ANALYZE {projection}OccurrenceIdentityDigest = ? AND "
            f"{authority}"
        )
        statements.extend(
            [
                "PREPARE scheduled_reclaim_scan FROM '"
                + scan_sql.replace("'", "''")
                + "'",
                f"SELECT 'probe=scheduled_reclaim_scan repetition={repetition}'",
                "EXECUTE scheduled_reclaim_scan USING " + authority_bindings,
                "DEALLOCATE PREPARE scheduled_reclaim_scan",
                "PREPARE scheduled_reclaim_point FROM '"
                + point_sql.replace("'", "''")
                + "'",
                f"SELECT 'probe=scheduled_reclaim_point repetition={repetition}'",
                "EXECUTE scheduled_reclaim_point USING @scheduled_digest, "
                + authority_bindings,
                "DEALLOCATE PREPARE scheduled_reclaim_point",
            ]
        )
    return ";\n".join(statements) + ";"


def record_event_observation_explain_sql(repetitions: int) -> str:
    """Compare full-scope and status-bounded record-event observations."""
    scope_predicate = (
        "ExecutionScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? "
        "AND RuntimeCapabilityId = ?"
    )
    former = (
        "SELECT "
        "COALESCE(SUM(CASE WHEN Status IN ('pending', 'processing') "
        "THEN 1 ELSE 0 END), 0), "
        "COALESCE(SUM(CASE WHEN Status = 'failed' THEN 1 ELSE 0 END), 0), "
        "COALESCE(TIMESTAMPDIFF(MICROSECOND, "
        "MIN(CASE WHEN Status IN ('pending', 'processing') "
        "THEN CreatedDate END), NOW(6)), 0) "
        "FROM _System_RecordEventOutbox WHERE " + scope_predicate
    )
    candidate = (
        "SELECT "
        "COALESCE(SUM(CASE WHEN Status IN ('pending', 'processing') "
        "THEN StatusDepth ELSE 0 END), 0), "
        "COALESCE(SUM(CASE WHEN Status = 'failed' "
        "THEN StatusDepth ELSE 0 END), 0), "
        "COALESCE(MAX(CASE WHEN Status IN ('pending', 'processing') "
        "THEN TIMESTAMPDIFF(MICROSECOND, OldestCreatedDate, NOW(6)) END), 0) "
        "FROM (SELECT Status, COUNT(*) AS StatusDepth, "
        "MIN(CreatedDate) AS OldestCreatedDate "
        "FROM _System_RecordEventOutbox WHERE "
        "ExecutionScopeDigest = ? "
        "AND Status IN ('pending', 'processing', 'failed') "
        "AND OrgId = ? AND AppInstallationId = ? "
        "AND RuntimeCapabilityId = ? GROUP BY Status) AS work_by_status"
    )
    statements = [
        "SET @record_event_row_id = (SELECT MIN(Id) "
        "FROM _System_RecordEventOutbox "
        "WHERE ExecutionScopeDigest IS NOT NULL "
        "AND OrgId IS NOT NULL AND AppInstallationId IS NOT NULL "
        "AND RuntimeCapabilityId IS NOT NULL "
        "GROUP BY ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId "
        "ORDER BY COUNT(*) DESC LIMIT 1)",
        "SET @record_event_scope_digest = (SELECT ExecutionScopeDigest "
        "FROM _System_RecordEventOutbox "
        "WHERE Id = @record_event_row_id)",
        "SET @record_event_org_id = (SELECT OrgId "
        "FROM _System_RecordEventOutbox "
        "WHERE Id = @record_event_row_id)",
        "SET @record_event_installation_id = (SELECT AppInstallationId "
        "FROM _System_RecordEventOutbox "
        "WHERE Id = @record_event_row_id)",
        "SET @record_event_capability_id = (SELECT RuntimeCapabilityId "
        "FROM _System_RecordEventOutbox "
        "WHERE Id = @record_event_row_id)",
        "SELECT CONCAT('probe_bindings record_event_row=', "
        "@record_event_row_id IS NOT NULL, ' scope=', "
        "@record_event_scope_digest IS NOT NULL, ' authority=', "
        "@record_event_org_id IS NOT NULL "
        "AND @record_event_installation_id IS NOT NULL "
        "AND @record_event_capability_id IS NOT NULL)",
    ]
    bindings = (
        "@record_event_scope_digest, @record_event_org_id, "
        "@record_event_installation_id, @record_event_capability_id"
    )
    for repetition in range(1, repetitions + 1):
        statements.extend(
            [
                "PREPARE record_event_observation_scan FROM '"
                + ("EXPLAIN ANALYZE " + former).replace("'", "''")
                + "'",
                f"SELECT 'probe=record_event_observation_scan repetition={repetition}'",
                "EXECUTE record_event_observation_scan USING " + bindings,
                "DEALLOCATE PREPARE record_event_observation_scan",
                "PREPARE record_event_observation_ranges FROM '"
                + ("EXPLAIN ANALYZE " + candidate).replace("'", "''")
                + "'",
                f"SELECT 'probe=record_event_observation_ranges repetition={repetition}'",
                "EXECUTE record_event_observation_ranges USING " + bindings,
                "DEALLOCATE PREPARE record_event_observation_ranges",
            ]
        )
    statements.append(
        "WITH former AS (SELECT "
        "COALESCE(SUM(CASE WHEN Status IN ('pending', 'processing') THEN 1 ELSE 0 END), 0) AS Depth, "
        "COALESCE(SUM(CASE WHEN Status = 'failed' THEN 1 ELSE 0 END), 0) AS Failed, "
        "MIN(CASE WHEN Status IN ('pending', 'processing') THEN CreatedDate END) AS OldestCreatedDate "
        "FROM _System_RecordEventOutbox WHERE "
        "ExecutionScopeDigest = @record_event_scope_digest "
        "AND OrgId = @record_event_org_id "
        "AND AppInstallationId = @record_event_installation_id "
        "AND RuntimeCapabilityId = @record_event_capability_id), "
        "work_by_status AS (SELECT Status, COUNT(*) AS StatusDepth, "
        "MIN(CreatedDate) AS OldestCreatedDate FROM _System_RecordEventOutbox "
        "WHERE ExecutionScopeDigest = @record_event_scope_digest "
        "AND Status IN ('pending', 'processing', 'failed') "
        "AND OrgId = @record_event_org_id "
        "AND AppInstallationId = @record_event_installation_id "
        "AND RuntimeCapabilityId = @record_event_capability_id GROUP BY Status), "
        "candidate AS (SELECT "
        "COALESCE(SUM(CASE WHEN Status IN ('pending', 'processing') THEN StatusDepth ELSE 0 END), 0) AS Depth, "
        "COALESCE(SUM(CASE WHEN Status = 'failed' THEN StatusDepth ELSE 0 END), 0) AS Failed, "
        "MIN(CASE WHEN Status IN ('pending', 'processing') THEN OldestCreatedDate END) AS OldestCreatedDate "
        "FROM work_by_status) "
        "SELECT CONCAT('probe_results depth=', former.Depth = candidate.Depth, "
        "' failed=', former.Failed = candidate.Failed, "
        "' oldest=', former.OldestCreatedDate <=> candidate.OldestCreatedDate) "
        "FROM former CROSS JOIN candidate"
    )
    return ";\n".join(statements) + ";"


def redact_plan(text: str) -> str:
    # EXPLAIN ranges can echo bound identifiers.  Preserve the plan and timing
    # while preventing UUIDs and digest-shaped values from entering artifacts.
    text = re.sub(
        r"(?i)\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b",
        "<bound-uuid>",
        text,
    )
    text = re.sub(r"(?i)\b(?:sha256:)?[0-9a-f]{64}\b", "<bound-digest>", text)
    # TiDB can render user-variable values in access ranges.  The evidence owns
    # operator/cardinality/timing facts, never tenant binding values.
    return re.sub(r"(?i)range:\[[^\]\r\n]*\]", "range:[<redacted>]", text)


def write_tsv(path: Path, header: str, body: str) -> None:
    path.write_text(header + "\n" + (body.strip() + "\n" if body.strip() else ""))
    path.chmod(0o600)


def statement_digest_projection(header: str, body: str) -> tuple[str, str]:
    """Project statement evidence without normalized SQL or record values."""
    source_columns = header.split("\t")
    missing = [
        column
        for column in STATEMENT_DIGEST_COLUMNS
        if column not in source_columns
    ]
    if missing:
        raise DeployError(
            "statement summary is missing digest evidence columns: "
            + ", ".join(missing)
        )
    indexes = [source_columns.index(column) for column in STATEMENT_DIGEST_COLUMNS]
    projected: list[str] = []
    for raw in body.splitlines():
        if not raw.strip():
            continue
        fields = raw.split("\t")
        if len(fields) != len(source_columns):
            raise DeployError("statement summary row has an unknown shape")
        projected.append("\t".join(fields[index] for index in indexes))
    return "\t".join(STATEMENT_DIGEST_COLUMNS), "\n".join(projected)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--runtime-env", type=Path, required=True)
    parser.add_argument("--since", type=parse_utc, required=True)
    parser.add_argument("--until", type=parse_utc, required=True)
    parser.add_argument("--output-dir", type=Path, required=True)
    parser.add_argument(
        "--digest-only",
        action="store_true",
        help="write only the sanitized top statement digest projection",
    )
    parser.add_argument("--catalog-explain-repetitions", type=int, default=0)
    parser.add_argument(
        "--scheduled-reclaim-explain-repetitions", type=int, default=0
    )
    parser.add_argument(
        "--record-event-observation-explain-repetitions", type=int, default=0
    )
    args = parser.parse_args()
    if not 0 <= args.catalog_explain_repetitions <= 10:
        parser.error("--catalog-explain-repetitions must be between 0 and 10")
    if not 0 <= args.scheduled_reclaim_explain_repetitions <= 1:
        parser.error(
            "--scheduled-reclaim-explain-repetitions must be between 0 and 1"
        )
    if not 0 <= args.record_event_observation_explain_repetitions <= 1:
        parser.error(
            "--record-event-observation-explain-repetitions must be between 0 and 1"
        )
    if args.since >= args.until:
        parser.error("--since must be earlier than --until")
    return args


def main() -> int:
    args = parse_args()
    envv = load_env_file(args.runtime_env)
    database = envv["TIDB_DATABASE"]
    args.output_dir.mkdir(parents=True, exist_ok=True)
    args.output_dir.chmod(0o700)

    if args.digest_only:
        all_top = db_query_scalar(
            envv,
            statement_summary_sql(args.since, args.until, focused=False),
            database=database,
        )
        digest_header, digest_rows = statement_digest_projection(
            STATEMENT_SUMMARY_HEADER, all_top
        )
        write_tsv(
            args.output_dir / "statement-digests.tsv",
            digest_header,
            digest_rows,
        )
        return 0

    summary_columns = db_query_scalar(
        envv,
        """
SELECT COLUMN_NAME
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = 'INFORMATION_SCHEMA'
  AND TABLE_NAME = 'CLUSTER_STATEMENTS_SUMMARY_HISTORY'
ORDER BY ORDINAL_POSITION
""".strip(),
        database=database,
    )
    write_tsv(
        args.output_dir / "statement-summary-columns.tsv",
        "column_name",
        summary_columns,
    )

    bucket_totals = db_query_scalar(
        envv,
        statement_bucket_totals_sql(args.since, args.until),
        database=database,
    )
    write_tsv(
        args.output_dir / "statement-summary-bucket-totals.tsv",
        "summary_begin\tsummary_end\texecutions\ttidb_total_latency_s\t"
        "resource_queue_s\tdigest_rows",
        bucket_totals,
    )

    try:
        resource_columns_raw = db_query_scalar(
            envv,
            """
SELECT COLUMN_NAME
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = 'INFORMATION_SCHEMA'
  AND TABLE_NAME = 'RESOURCE_GROUPS'
ORDER BY ORDINAL_POSITION
""".strip(),
            database=database,
        )
        resource_columns = [
            column.strip()
            for column in resource_columns_raw.splitlines()
            if re.fullmatch(r"[A-Za-z0-9_]+", column.strip())
        ]
        if not resource_columns:
            raise DeployError("RESOURCE_GROUPS exposes no readable columns")
        resource_groups = db_query_scalar(
            envv,
            "SELECT "
            + ", ".join(f"`{column}`" for column in resource_columns)
            + " FROM INFORMATION_SCHEMA.RESOURCE_GROUPS ORDER BY `NAME`",
            database=database,
        )
    except DeployError:
        write_tsv(
            args.output_dir / "resource-groups.tsv",
            "status",
            "unavailable",
        )
    else:
        write_tsv(
            args.output_dir / "resource-groups.tsv",
            "\t".join(resource_columns),
            resource_groups,
        )

    outbound_queue = db_query_scalar(
        envv,
        outbound_queue_snapshot_sql(),
        database=database,
    )
    write_tsv(
        args.output_dir / "outbound-queue-snapshot.tsv",
        "sample_window\tcatalog_revision\tstate\toperation\tconnector_owner_key\t"
        "failure_code\t"
        "command_count\tretry_count_total\tdue_count\toldest_created_at\t"
        "newest_modified_at",
        outbound_queue,
    )
    outbound_indexes = db_query_scalar(
        envv,
        outbound_command_indexes_sql(),
        database=database,
    )
    write_tsv(
        args.output_dir / "outbound-command-indexes.tsv",
        "index_name\tnon_unique\tsequence_in_index\tcolumn_name\tprefix_length",
        outbound_indexes,
    )
    outbound_plan = db_query_scalar(
        envv,
        outbound_claim_explain_sql(),
        database=database,
    )
    outbound_plan_path = args.output_dir / "outbound-claim-explain.txt"
    outbound_plan_path.write_text(redact_plan(outbound_plan) + "\n")
    outbound_plan_path.chmod(0o600)

    header = STATEMENT_SUMMARY_HEADER
    focused = db_query_scalar(
        envv,
        statement_summary_sql(args.since, args.until, focused=True),
        database=database,
    )
    write_tsv(args.output_dir / "focused-statement-summary.tsv", header, focused)

    bucket_header = "summary_begin\tsummary_end\t" + header
    bucketed = db_query_scalar(
        envv,
        statement_summary_sql(
            args.since, args.until, focused=True, bucketed=True
        ),
        database=database,
    )
    write_tsv(
        args.output_dir / "focused-statement-summary-by-bucket.tsv",
        bucket_header,
        bucketed,
    )

    slow_header = (
        "observed_at\tdigest\tplan_digest\ttidb_latency_ms\tparse_ms\tcompile_ms\t"
        "wait_ts_ms\ttikv_process_ms\ttikv_wait_ms\tbackoff_ms\ttotal_keys\tprocessed_keys\t"
        "request_count\tcop_process_avg_ms\tcop_wait_avg_ms\tbackoff_types\t"
        "plan_from_cache"
    )
    all_top = db_query_scalar(
        envv,
        statement_summary_sql(args.since, args.until, focused=False),
        database=database,
    )
    write_tsv(args.output_dir / "top-statement-summary.tsv", header, all_top)
    digest_header, digest_rows = statement_digest_projection(header, all_top)
    write_tsv(
        args.output_dir / "statement-digests.tsv",
        digest_header,
        digest_rows,
    )
    all_top_bucketed = db_query_scalar(
        envv,
        statement_summary_sql(
            args.since, args.until, focused=False, bucketed=True
        ),
        database=database,
    )
    write_tsv(
        args.output_dir / "top-statement-summary-by-bucket.tsv",
        bucket_header,
        all_top_bucketed,
    )

    if args.catalog_explain_repetitions:
        plan = db_query_scalar(
            envv,
            catalog_explain_sql(args.catalog_explain_repetitions),
            database=database,
        )
        path = args.output_dir / "catalog-explain-analyze.tsv"
        path.write_text(redact_plan(plan) + "\n")
        path.chmod(0o600)

    if args.scheduled_reclaim_explain_repetitions:
        binding_count = db_query_scalar(
            envv,
            "SELECT COUNT(*) FROM ("
            "SELECT Id FROM _System_ScheduledRunReservation "
            "WHERE OccurrenceIdentityDigest IS NOT NULL "
            "AND OccurrenceIdentityDigest <> '' "
            "AND OrgId IS NOT NULL AND AppInstallationId IS NOT NULL "
            "AND RuntimeCapabilityId IS NOT NULL "
            "ORDER BY OccurrenceIdentityDigest LIMIT 1"
            ") AS scheduled_reclaim_probe_binding",
            database=database,
        ).strip()
        if binding_count != "1":
            raise DeployError(
                "scheduled reclaim plan probe has no complete existing binding"
            )
        plan = db_query_scalar(
            envv,
            scheduled_reclaim_explain_sql(
                args.scheduled_reclaim_explain_repetitions
            ),
            database=database,
        )
        path = args.output_dir / "scheduled-reclaim-explain-analyze.tsv"
        path.write_text(redact_plan(plan) + "\n")
        path.chmod(0o600)

    if args.record_event_observation_explain_repetitions:
        binding_count = db_query_scalar(
            envv,
            "SELECT COUNT(*) FROM ("
            "SELECT ExecutionScopeDigest FROM _System_RecordEventOutbox "
            "WHERE ExecutionScopeDigest IS NOT NULL "
            "AND OrgId IS NOT NULL AND AppInstallationId IS NOT NULL "
            "AND RuntimeCapabilityId IS NOT NULL "
            "ORDER BY CreatedDate DESC, Id DESC LIMIT 1"
            ") AS record_event_observation_probe_binding",
            database=database,
        ).strip()
        if binding_count != "1":
            raise DeployError(
                "record-event observation plan probe has no complete existing binding"
            )
        plan = db_query_scalar(
            envv,
            record_event_observation_explain_sql(
                args.record_event_observation_explain_repetitions
            ),
            database=database,
        )
        path = args.output_dir / "record-event-observation-explain-analyze.tsv"
        path.write_text(redact_plan(plan) + "\n")
        path.chmod(0o600)

    try:
        slow = db_query_scalar(
            envv, slow_query_sql(args.since, args.until), database=database
        )
    except DeployError:
        write_tsv(
            args.output_dir / "focused-slow-query.tsv",
            slow_header,
            "# unavailable: Test TiDB principal cannot read CLUSTER_SLOW_QUERY",
        )
    else:
        write_tsv(args.output_dir / "focused-slow-query.tsv", slow_header, slow)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())