""" Clinic Data Proxy data source for SQL Health Check inspection items. The service runs inside Clinic's Kubernetes cluster, so it should prefer the in-cluster clinic-data-service endpoint instead of reading remote S3 parquet files through DuckDB. """ import json import logging import os import re import time from dataclasses import dataclass from decimal import Decimal, InvalidOperation from datetime import UTC, date, datetime, timedelta from http.client import RemoteDisconnected from typing import Any, Dict, Iterator from urllib import error, request logger = logging.getLogger(__name__) DEFAULT_INTERNAL_ENDPOINT = "http://clinic-data-service:8087/data-proxy/query" DEFAULT_QUERY_PATH = "/data-proxy/query" DEFAULT_TIMEOUT_SECONDS = 180 DEFAULT_PAGE_SIZE = 2000 DEFAULT_RETRY_ATTEMPTS = 2 DEFAULT_SAMPLE_DIGEST_BATCH_SIZE = 100 DEFAULT_TABLE_FULL_SCAN_SLICE_SECONDS = 4 * 60 * 60 DEFAULT_TABLE_FULL_SCAN_MIN_SLICE_SECONDS = 15 * 60 DEFAULT_TABLE_FULL_SCAN_SLICE_OVERFETCH_FACTOR = 5 DEFAULT_DML_FILTER_OVERFETCH_FACTOR = 5 REQUIRED_COLUMNS = ( "time", "digest", "query", "decoded_plan", "query_time", "conn_id", "db", ) TIME_EXPR = "time" CONN_ID_EXPR = "COALESCE(conn_id, '')" QUERY_TIME_EXPR = "query_time" DB_EXPR = "COALESCE(db, '')" INTERNAL_SAMPLE_COLUMNS = {"op_act_rows", "op_rank", "sample_act_rows", "sample_rank"} TABLE_FULL_SCAN_SEGMENT_PATTERN = ( r'(?s)(?:"actRows"\s*:\s*"?[0-9]+"?' r'\s*,\s*"' '(?:accessObject|accessObjects|copExecInfo|cost|diskBytes|estRows|memoryBytes|name|operatorInfo|storeType|taskType)"' r'.*?"(?:id|operator|type|name)"\s*:\s*"TableFullScan[^"]*"' r'(?:.{0,1500}?"(?:storeType|taskType)"\s*:\s*"[^"]*")?' '|' r'"(?:id|operator|type|name)"\s*:\s*"TableFullScan[^"]*".{0,3000}?' r'"actRows"\s*:\s*"?[0-9]+"?' r'(?:.{0,1500}?"(?:storeType|taskType)"\s*:\s*"[^"]*")?)' ) ACT_ROWS_PATTERN = r'"actRows"\s*:\s*"?([0-9]+)"?' DML_COMMENT_PATTERN = r'(?:/\*.*?\*/|--[^\r\n]*(?:\r?\n|$)|#[^\r\n]*(?:\r?\n|$))' DML_DIRECT_PATTERN = r'(?:insert|replace|update|delete)(?:\s|\(|$)' DML_CTE_PATTERN = ( r'with(?:\s+recursive)?\s+.*\)\s*' rf'(?:{DML_COMMENT_PATTERN}\s*)*{DML_DIRECT_PATTERN}' ) DML_SQL_DIRECT_PATTERN = r'(?:insert|replace|update|delete)(?:[[:space:](]|$)' DML_QUERY_PATTERN = ( rf'(?is)^\s*(?:{DML_COMMENT_PATTERN}\s*)*' rf'(?:{DML_DIRECT_PATTERN}|{DML_CTE_PATTERN}).*' ) DML_LEADING_COMMENT_PATTERN = ( rf'(?s)^\s*(?:{DML_COMMENT_PATTERN}\s*)*' ) DML_START_PATTERN = ( rf'^{DML_SQL_DIRECT_PATTERN}' ) TIFLASH_SEGMENT_PATTERN = ( r'(?is).*(?:"storeType"\s*:\s*"tiflash"' r'|"taskType"\s*:\s*"mpp"' '|tiflash_task|tiflash_scan).*' ) ORDER_BY_CLAUSE = ( f"{TIME_EXPR} DESC, " f"{CONN_ID_EXPR} DESC, " "digest DESC, " f"{QUERY_TIME_EXPR} DESC, " f"{DB_EXPR} DESC" ) SLOW_LOG_COLUMNS = ( "id", "time", "digest", "instance", "txn_start_ts", "user", "host", "conn_id", "session_alias", "exec_retry_count", "exec_retry_time", "query_time", "parse_time", "compile_time", "rewrite_time", "preproc_subqueries", "preproc_subqueries_time", "optimize_time", "wait_ts", "prewrite_time", "wait_prewrite_binlog_time", "commit_time", "get_commit_ts_time", "commit_backoff_time", "backoff_types", "resolve_lock_time", "local_latch_wait_time", "write_keys", "write_size", "prewrite_region", "txn_retry", "cop_time", "process_time", "wait_time", "backoff_time", "lockkeys_time", "request_count", "total_keys", "process_keys", "rocksdb_delete_skipped_count", "rocksdb_key_skipped_count", "rocksdb_block_cache_hit_count", "rocksdb_block_read_count", "rocksdb_block_read_byte", "db", "index_names", "is_internal", "stats", "cop_proc_avg", "cop_proc_p90", "cop_proc_max", "cop_proc_addr", "cop_wait_avg", "cop_wait_p90", "cop_wait_max", "cop_wait_addr", "mem_max", "disk_max", "kv_total", "pd_total", "backoff_total", "write_sql_response_total", "result_rows", "warnings", "backoff_detail", "prepared", "succ", "is_explicit_txn", "is_write_cache_table", "plan_from_cache", "plan_from_binding", "has_more_results", "resource_group", "request_unit_read", "request_unit_write", "time_queued_by_rc", "plan_digest", "prev_stmt", "serverless_tenant_id", "serverless_project_id", "serverless_cluster_id", "worker_role", "is_tiflash", "query", "binary_plan", "plan", "decoded_plan", "date", ) TOP_DIGEST_DIMENSIONS = { "count": { "result_key": "top_by_count", "order_alias": "slow_count", "description": "Top Digest by slow log count", }, "total_latency": { "result_key": "top_by_total_latency", "order_alias": "total_latency_s", "description": "Top Digest by total latency", }, "total_keys": { "result_key": "top_by_total_keys", "order_alias": "total_keys", "description": "Top Digest by sum total_keys", }, "process_keys": { "result_key": "top_by_process_keys", "order_alias": "process_keys", "description": "Top Digest by sum process_keys", }, "request_count": { "result_key": "top_by_request_count", "order_alias": "request_count", "description": "Top Digest by sum request_count", }, "sum_ru": { "result_key": "top_by_sum_ru", "order_alias": "sum_ru", "description": "Top Digest by sum request_unit_read + request_unit_write", }, } @dataclass(frozen=True, slots=True) class PageCursor: time: str conn_id: str digest: str query_time: str db: str class DataProxyQueryError(RuntimeError): """Raised when Clinic Data Proxy query fails.""" def __init__( self, message: str, *, trace_id: str | None = None, status_code: int | None = None, ): self.raw_message = message self.trace_id = trace_id or "" self.status_code = status_code if self.trace_id: message = f"{message} trace_id={self.trace_id}" super().__init__(message) @property def is_bytes_billed_limit_exceeded(self) -> bool: return "bytesBilledLimitExceeded" in self.raw_message @property def is_non_retryable(self) -> bool: return self.is_bytes_billed_limit_exceeded @property def bytes_billed_values(self) -> tuple[int | None, int | None]: match = re.search( r"bytes billed:\s*([0-9]+)\.\s*([0-9]+)\s+or higher required", self.raw_message, ) if not match: return None, None return int(match.group(1)), int(match.group(2)) class ServiceDataSource: """Slow log reader backed by Clinic Data Proxy.""" def __init__( self, cluster_id: str, endpoint: str | None = None, timeout_seconds: int | None = None, page_size: int | None = None, retry_attempts: int | None = None, ): self.cluster_id = cluster_id self.endpoint = self._resolve_endpoint(endpoint) self.timeout_seconds = self._resolve_positive_int( timeout_seconds, env_name="CLINIC_DATA_PROXY_TIMEOUT_SECONDS", default=DEFAULT_TIMEOUT_SECONDS, ) self.page_size = self._resolve_positive_int( page_size, env_name="CLINIC_DATA_PROXY_PAGE_SIZE", default=DEFAULT_PAGE_SIZE, ) self.retry_attempts = self._resolve_non_negative_int( retry_attempts, env_name="CLINIC_DATA_PROXY_RETRY_ATTEMPTS", default=DEFAULT_RETRY_ATTEMPTS, ) self.api_key = os.environ.get("CLINIC_DATA_PROXY_API_KEY") or os.environ.get("CLINIC_API_KEY") def initialize(self): """Log the resolved Data Proxy configuration.""" logger.info( "Using Clinic Data Proxy endpoint=%s cluster=%s page_size=%d timeout_seconds=%d auth=%s", self.endpoint, self.cluster_id, self.page_size, self.timeout_seconds, "enabled" if self.api_key else "disabled", ) def query_table_full_scan_summary( self, start_time: datetime, end_time: datetime, threshold: int = 10000, vendor: str = "aws", top_n: int = 100, ignore_dml: bool = False, ignore_tiflash: bool = False, ) -> list[Dict[str, Any]]: """Query aggregated TableFullScan statistics per digest from the SQL engine. Instead of fetching all slow query records and filtering in Python, this pushes the detection into the SQL engine: - WHERE decoded_plan LIKE '%TableFullScan%' pre-filters (typically 99%+ reduction) - REGEXP_EXTRACT_ALL + UNNEST expands candidate TableFullScan operator segments - REGEXP_EXTRACT extracts actRows per segment, then MAX() keeps the largest scan per slow log - CAST AS SIGNED converts to integer (dialect layer maps to INT64/BIGINT per engine) - GROUP BY digest aggregates per-digest stats - approx_percentile / APPROX_QUANTILES for p50/p80/p95/p99 Returns a list of dicts with keys: digest, query, tfs_count, max_act_rows, total_scanned_rows, max_query_time, total_query_time, anomaly_count, p50_act_rows, p80_act_rows, p95_act_rows, p99_act_rows, p50_query_time, p80_query_time, p95_query_time, p99_query_time, overview """ query_start = self._to_utc(start_time) query_end = self._to_utc(end_time) if query_start >= query_end: return [] if top_n < 1: return [] try: return self._query_table_full_scan_summary_direct( query_start, query_end, threshold=threshold, vendor=vendor, top_n=top_n, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) except DataProxyQueryError as exc: if not exc.is_bytes_billed_limit_exceeded: raise limit, required = exc.bytes_billed_values logger.warning( "TableFullScan summary exceeded bytes billed limit; falling back to time slices cluster=%s range=%s~%s top_n=%d limit_bytes=%s required_bytes=%s error=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), top_n, limit if limit is not None else "-", required if required is not None else "-", exc.raw_message, ) return self._query_table_full_scan_summary_sliced( query_start, query_end, threshold=threshold, vendor=vendor, top_n=top_n, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) def _query_table_full_scan_summary_direct( self, query_start: datetime, query_end: datetime, *, threshold: int, vendor: str, top_n: int, ignore_dml: bool, ignore_tiflash: bool, ) -> list[Dict[str, Any]]: if query_start >= query_end or top_n < 1: return [] if self._is_synapse_vendor(vendor): return self._query_table_full_scan_summary_synapse( query_start, query_end, threshold=threshold, vendor=vendor, top_n=top_n, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) use_native_sql = self._use_table_full_scan_native_sql(vendor) act_rows_expr = self._build_act_rows_expr(vendor, source_expr="op_segment") threshold_value = str(int(threshold)) where_clause = self._build_table_full_scan_where_clause( query_start, query_end, vendor=vendor, ignore_dml=ignore_dml, ) segment_join = self._build_table_full_scan_segment_join(vendor) segment_filters = self._build_table_full_scan_segment_filters(vendor, ignore_tiflash=ignore_tiflash) pct_exprs = self._build_percentile_exprs("act_rows", "log_query_time", vendor) pct_clause = ", " + pct_exprs if pct_exprs else "" sql = ( "SELECT * FROM (" "SELECT digest_stats.*, " "SUM(tfs_count) OVER () AS overall_tfs_count, " "SUM(anomaly_count) OVER () AS overall_anomaly_count, " "COUNT(*) OVER () AS overall_digests_with_tfs, " "SUM(CASE WHEN anomaly_count > 0 THEN 1 ELSE 0 END) OVER () AS overall_digests_with_anomaly, " "MAX(max_act_rows) OVER () AS overall_max_act_rows " "FROM (" "SELECT " "digest, " "any_value(query) AS query, " "COUNT(*) AS tfs_count, " "MAX(act_rows) AS max_act_rows, " "SUM(COALESCE(act_rows, 0)) AS total_act_rows, " "MAX(log_query_time) AS max_query_time, " "SUM(log_query_time) AS total_query_time, " f"COUNT(CASE WHEN act_rows >= {threshold_value} THEN 1 END) AS anomaly_count" f"{pct_clause} " "FROM (" "SELECT " "digest, " "any_value(query) AS query, " "any_value(query_time) AS log_query_time, " "MAX(op_act_rows) AS act_rows " "FROM (" f"SELECT digest, query, query_time, time, conn_id, db, {act_rows_expr} AS op_act_rows " "FROM clinic_data_proxy.slow_query_logs " f"{segment_join} " f"WHERE {where_clause}{segment_filters}" ") expanded " "WHERE op_act_rows IS NOT NULL " "GROUP BY time, conn_id, digest, query_time, db" ") tfs " "GROUP BY digest" ") digest_stats" ") annotated " "WHERE anomaly_count > 0 " "ORDER BY total_act_rows DESC " f"LIMIT {int(top_n)}" ) logger.info( "Querying TableFullScan summary cluster=%s range=%s~%s top_n=%d native_sql=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), top_n, use_native_sql, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=1, skip_sql_rewrite=use_native_sql, query_name="tablefullscan.summary", ) logger.info( "TableFullScan summary returned cluster=%s digests=%d trace_id=%s", self.cluster_id, len(rows), trace_id or "-", ) results = [] for raw_row in rows: row = {col: raw_row[i] if i < len(raw_row) else None for i, col in enumerate(columns)} max_act = self._coerce_int_value(row.get("max_act_rows")) max_qt = self._coerce_float(row.get("max_query_time")) or 0.0 results.append({ "digest": str(row.get("digest") or ""), "query": str(row.get("query") or ""), "tfs_count": self._coerce_int_value(row.get("tfs_count")), "max_act_rows": max_act, "total_scanned_rows": self._coerce_int_value(row.get("total_act_rows")), "max_query_time": max_qt, "total_query_time": self._coerce_float(row.get("total_query_time")) or 0.0, "anomaly_count": self._coerce_int_value(row.get("anomaly_count")), "p50_act_rows": self._coerce_int_value(row.get("p50_act_rows")) or max_act, "p80_act_rows": self._coerce_int_value(row.get("p80_act_rows")) or max_act, "p95_act_rows": self._coerce_int_value(row.get("p95_act_rows")) or max_act, "p99_act_rows": self._coerce_int_value(row.get("p99_act_rows")) or max_act, "p50_query_time": self._coerce_float(row.get("p50_query_time")) or max_qt, "p80_query_time": self._coerce_float(row.get("p80_query_time")) or max_qt, "p95_query_time": self._coerce_float(row.get("p95_query_time")) or max_qt, "p99_query_time": self._coerce_float(row.get("p99_query_time")) or max_qt, "overview": { "tfs_count": self._coerce_int_value(row.get("overall_tfs_count")), "anomaly_count": self._coerce_int_value(row.get("overall_anomaly_count")), "digests_with_tfs": self._coerce_int_value(row.get("overall_digests_with_tfs")), "digests_with_anomaly": self._coerce_int_value(row.get("overall_digests_with_anomaly")), "max_act_rows": self._coerce_int_value(row.get("overall_max_act_rows")), }, }) return results def query_table_full_scan_overview( self, start_time: datetime, end_time: datetime, threshold: int = 10000, vendor: str = "aws", ignore_dml: bool = False, ignore_tiflash: bool = False, ) -> Dict[str, Any]: """Query global TableFullScan overview statistics without returning all digests.""" query_start = self._to_utc(start_time) query_end = self._to_utc(end_time) if query_start >= query_end: return {} try: return self._query_table_full_scan_overview_direct( query_start, query_end, threshold=threshold, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) except DataProxyQueryError as exc: if not exc.is_bytes_billed_limit_exceeded: raise limit, required = exc.bytes_billed_values logger.warning( "TableFullScan overview exceeded bytes billed limit; falling back to time slices cluster=%s range=%s~%s limit_bytes=%s required_bytes=%s error=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), limit if limit is not None else "-", required if required is not None else "-", exc.raw_message, ) return self._query_table_full_scan_overview_sliced( query_start, query_end, threshold=threshold, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) def _query_table_full_scan_overview_direct( self, query_start: datetime, query_end: datetime, *, threshold: int, vendor: str, ignore_dml: bool, ignore_tiflash: bool, ) -> Dict[str, Any]: if query_start >= query_end: return {} if self._is_synapse_vendor(vendor): return self._query_table_full_scan_overview_synapse( query_start, query_end, threshold=threshold, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) use_native_sql = self._use_table_full_scan_native_sql(vendor) act_rows_expr = self._build_act_rows_expr(vendor, source_expr="op_segment") threshold_value = str(int(threshold)) where_clause = self._build_table_full_scan_where_clause( query_start, query_end, vendor=vendor, ignore_dml=ignore_dml, ) segment_join = self._build_table_full_scan_segment_join(vendor) segment_filters = self._build_table_full_scan_segment_filters(vendor, ignore_tiflash=ignore_tiflash) sql = ( "SELECT " "COUNT(*) AS tfs_count, " f"COUNT(CASE WHEN act_rows >= {threshold_value} THEN 1 END) AS anomaly_count, " "COUNT(DISTINCT digest) AS digests_with_tfs, " f"COUNT(DISTINCT CASE WHEN act_rows >= {threshold_value} THEN digest END) AS digests_with_anomaly, " "MAX(act_rows) AS max_act_rows " "FROM (" "SELECT digest, MAX(op_act_rows) AS act_rows " "FROM (" f"SELECT digest, time, conn_id, query_time, db, {act_rows_expr} AS op_act_rows " "FROM clinic_data_proxy.slow_query_logs " f"{segment_join} " f"WHERE {where_clause}{segment_filters}" ") expanded " "WHERE op_act_rows IS NOT NULL " "GROUP BY time, conn_id, digest, query_time, db" ") tfs" ) logger.info( "Querying TableFullScan overview cluster=%s range=%s~%s native_sql=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), use_native_sql, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=1, skip_sql_rewrite=use_native_sql, query_name="tablefullscan.overview", ) logger.info( "TableFullScan overview returned cluster=%s rows=%d trace_id=%s", self.cluster_id, len(rows), trace_id or "-", ) if not rows: return {} row = {col: rows[0][i] if i < len(rows[0]) else None for i, col in enumerate(columns)} return { "tfs_count": self._coerce_int_value(row.get("tfs_count")), "anomaly_count": self._coerce_int_value(row.get("anomaly_count")), "digests_with_tfs": self._coerce_int_value(row.get("digests_with_tfs")), "digests_with_anomaly": self._coerce_int_value(row.get("digests_with_anomaly")), "max_act_rows": self._coerce_int_value(row.get("max_act_rows")), } def query_table_full_scan_samples( self, start_time: datetime, end_time: datetime, digests: list[str], sample_size: int = 10, vendor: str = "aws", ignore_dml: bool = False, ignore_tiflash: bool = False, ) -> Dict[str, list[Dict[str, Any]]]: """Query top sample slow logs for the selected digests. Only the already-selected result digests are queried, and the SQL engine ranks rows per digest before returning full slow-log rows. """ query_start = self._to_utc(start_time) query_end = self._to_utc(end_time) if query_start >= query_end or sample_size <= 0 or not digests: return {} try: return self._query_table_full_scan_samples_direct( query_start, query_end, digests, sample_size=sample_size, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) except DataProxyQueryError as exc: if not exc.is_bytes_billed_limit_exceeded: raise limit, required = exc.bytes_billed_values logger.warning( "TableFullScan samples exceeded bytes billed limit; falling back to time slices cluster=%s range=%s~%s digests=%d sample_size=%d limit_bytes=%s required_bytes=%s error=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), len(digests), sample_size, limit if limit is not None else "-", required if required is not None else "-", exc.raw_message, ) return self._query_table_full_scan_samples_sliced( query_start, query_end, digests, sample_size=sample_size, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) def _query_table_full_scan_samples_direct( self, query_start: datetime, query_end: datetime, digests: list[str], *, sample_size: int, vendor: str, ignore_dml: bool, ignore_tiflash: bool, ) -> Dict[str, list[Dict[str, Any]]]: if query_start >= query_end or sample_size <= 0 or not digests: return {} if self._is_synapse_vendor(vendor): return self._query_table_full_scan_samples_synapse( query_start, query_end, digests, sample_size=sample_size, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) use_native_sql = self._use_table_full_scan_native_sql(vendor) act_rows_expr = self._build_act_rows_expr(vendor, source_expr="op_segment") where_clause = self._build_table_full_scan_where_clause( query_start, query_end, vendor=vendor, ignore_dml=ignore_dml, ) segment_join = self._build_table_full_scan_segment_join(vendor) segment_filters = self._build_table_full_scan_segment_filters(vendor, ignore_tiflash=ignore_tiflash) samples: Dict[str, list[Dict[str, Any]]] = {digest: [] for digest in digests} select_clause = ", ".join(self._quote_identifier(column, vendor) for column in SLOW_LOG_COLUMNS) log_identity_partition_expr = self._table_full_scan_sample_log_partition_expr(vendor) for batch_number, digest_batch in enumerate( self._chunked(digests, DEFAULT_SAMPLE_DIGEST_BATCH_SIZE), start=1, ): digest_values = ", ".join( "'" + self._escape_sql_string(digest) + "'" for digest in digest_batch ) sql = ( "SELECT * FROM (" "SELECT best_log.*, " "ROW_NUMBER() OVER (" "PARTITION BY digest " f"ORDER BY sample_act_rows DESC, time DESC, {CONN_ID_EXPR} DESC" ") AS sample_rank " "FROM (" "SELECT * FROM (" "SELECT expanded.*, " "op_act_rows AS sample_act_rows, " "ROW_NUMBER() OVER (" f"PARTITION BY {log_identity_partition_expr} " "ORDER BY op_act_rows DESC" ") AS op_rank " "FROM (" f"SELECT {select_clause}, {act_rows_expr} AS op_act_rows " "FROM clinic_data_proxy.slow_query_logs " f"{segment_join} " f"WHERE {where_clause}{segment_filters} " f"AND digest IN ({digest_values})" ") expanded " "WHERE op_act_rows IS NOT NULL" ") ranked_ops " "WHERE op_rank = 1" ") best_log " ") ranked " f"WHERE sample_rank <= {int(sample_size)} " "ORDER BY digest, sample_act_rows DESC, time DESC" ) logger.info( "Querying TableFullScan samples cluster=%s batch=%d digests=%d sample_size=%d native_sql=%s", self.cluster_id, batch_number, len(digest_batch), sample_size, use_native_sql, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=batch_number, skip_sql_rewrite=use_native_sql, query_name="tablefullscan.samples", ) logger.info( "TableFullScan samples returned cluster=%s batch=%d rows=%d trace_id=%s", self.cluster_id, batch_number, len(rows), trace_id or "-", ) for raw_row in rows: row = {col: raw_row[i] if i < len(raw_row) else None for i, col in enumerate(columns)} digest = str(row.get("digest") or "") if not digest: continue samples.setdefault(digest, []).append(self._build_sample(row)) return samples def _query_table_full_scan_summary_synapse( self, query_start: datetime, query_end: datetime, *, threshold: int, vendor: str, top_n: int, ignore_dml: bool, ignore_tiflash: bool, ) -> list[Dict[str, Any]]: """Query TableFullScan summaries using Azure Synapse T-SQL JSON functions.""" threshold_value = str(int(threshold)) where_clause = self._build_table_full_scan_where_clause( query_start, query_end, vendor=vendor, ignore_dml=ignore_dml, ) base_columns = ["time", "conn_id", "digest", "query_time", "db", "query", "decoded_plan"] ctes = self._build_synapse_table_full_scan_ctes(where_clause, base_columns, vendor) op_condition = self._build_synapse_table_full_scan_operator_condition(vendor) tiflash_filter = self._build_synapse_tiflash_filter(vendor, ignore_tiflash=ignore_tiflash) act_rows_expr = self._build_synapse_json_numeric_expr("$.actRows", "int") query_time_expr = self._build_zero_safe_numeric_expr(self._column_ref("query_time", vendor), "float", vendor) time_col = self._column_ref("time", vendor) conn_id_col = self._column_ref("conn_id", vendor) digest_col = self._column_ref("digest", vendor) db_col = self._column_ref("db", vendor) query_col = self._column_ref("query", vendor) sql = ( f"{ctes}, " "expanded AS (" "SELECT " f"{digest_col} AS digest, " f"{query_col} AS query, " f"{query_time_expr} AS log_query_time, " f"{time_col} AS log_time, " f"{conn_id_col} AS conn_id, " f"{db_col} AS db, " f"{act_rows_expr} AS op_act_rows " "FROM plan_nodes " f"WHERE {op_condition}{tiflash_filter}" "), " "tfs AS (" "SELECT " "digest, " "MAX(CAST(query AS NVARCHAR(4000))) AS query, " "MAX(log_query_time) AS log_query_time, " "MAX(op_act_rows) AS act_rows " "FROM expanded " "WHERE op_act_rows IS NOT NULL " "GROUP BY log_time, conn_id, digest, log_query_time, db" "), " "digest_stats AS (" "SELECT " "digest, " "MAX(CAST(query AS NVARCHAR(4000))) AS query, " "COUNT_BIG(*) AS tfs_count, " "MAX(act_rows) AS max_act_rows, " "SUM(COALESCE(act_rows, 0)) AS total_act_rows, " "MAX(log_query_time) AS max_query_time, " "SUM(COALESCE(log_query_time, 0)) AS total_query_time, " f"SUM(CASE WHEN act_rows >= {threshold_value} THEN 1 ELSE 0 END) AS anomaly_count " "FROM tfs " "GROUP BY digest" ") " f"SELECT TOP ({int(top_n)}) * FROM (" "SELECT digest_stats.*, " "SUM(tfs_count) OVER () AS overall_tfs_count, " "SUM(anomaly_count) OVER () AS overall_anomaly_count, " "COUNT_BIG(*) OVER () AS overall_digests_with_tfs, " "SUM(CASE WHEN anomaly_count > 0 THEN 1 ELSE 0 END) OVER () AS overall_digests_with_anomaly, " "MAX(max_act_rows) OVER () AS overall_max_act_rows " "FROM digest_stats" ") annotated " "WHERE anomaly_count > 0 " "ORDER BY total_act_rows DESC" ) logger.info( "Querying TableFullScan summary cluster=%s range=%s~%s top_n=%d native_sql=%s dialect=synapse", self.cluster_id, query_start.isoformat(), query_end.isoformat(), top_n, True, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=1, skip_sql_rewrite=False, query_name="tablefullscan.summary", ) logger.info( "TableFullScan summary returned cluster=%s digests=%d trace_id=%s dialect=synapse", self.cluster_id, len(rows), trace_id or "-", ) return self._map_table_full_scan_summary_rows(columns, rows) def _query_table_full_scan_overview_synapse( self, query_start: datetime, query_end: datetime, *, threshold: int, vendor: str, ignore_dml: bool, ignore_tiflash: bool, ) -> Dict[str, Any]: threshold_value = str(int(threshold)) where_clause = self._build_table_full_scan_where_clause( query_start, query_end, vendor=vendor, ignore_dml=ignore_dml, ) base_columns = ["time", "conn_id", "digest", "query_time", "db", "decoded_plan"] ctes = self._build_synapse_table_full_scan_ctes(where_clause, base_columns, vendor) op_condition = self._build_synapse_table_full_scan_operator_condition(vendor) tiflash_filter = self._build_synapse_tiflash_filter(vendor, ignore_tiflash=ignore_tiflash) act_rows_expr = self._build_synapse_json_numeric_expr("$.actRows", "int") time_col = self._column_ref("time", vendor) conn_id_col = self._column_ref("conn_id", vendor) digest_col = self._column_ref("digest", vendor) query_time_col = self._column_ref("query_time", vendor) db_col = self._column_ref("db", vendor) sql = ( f"{ctes}, " "expanded AS (" "SELECT " f"{digest_col} AS digest, " f"{time_col} AS log_time, " f"{conn_id_col} AS conn_id, " f"{query_time_col} AS log_query_time, " f"{db_col} AS db, " f"{act_rows_expr} AS op_act_rows " "FROM plan_nodes " f"WHERE {op_condition}{tiflash_filter}" "), " "tfs AS (" "SELECT digest, MAX(op_act_rows) AS act_rows " "FROM expanded " "WHERE op_act_rows IS NOT NULL " "GROUP BY log_time, conn_id, digest, log_query_time, db" ") " "SELECT " "COUNT_BIG(*) AS tfs_count, " f"SUM(CASE WHEN act_rows >= {threshold_value} THEN 1 ELSE 0 END) AS anomaly_count, " "COUNT(DISTINCT digest) AS digests_with_tfs, " f"COUNT(DISTINCT CASE WHEN act_rows >= {threshold_value} THEN digest END) AS digests_with_anomaly, " "MAX(act_rows) AS max_act_rows " "FROM tfs" ) logger.info( "Querying TableFullScan overview cluster=%s range=%s~%s native_sql=%s dialect=synapse", self.cluster_id, query_start.isoformat(), query_end.isoformat(), True, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=1, skip_sql_rewrite=False, query_name="tablefullscan.overview", ) logger.info( "TableFullScan overview returned cluster=%s rows=%d trace_id=%s dialect=synapse", self.cluster_id, len(rows), trace_id or "-", ) if not rows: return {} row = {col: rows[0][i] if i < len(rows[0]) else None for i, col in enumerate(columns)} return { "tfs_count": self._coerce_int_value(row.get("tfs_count")), "anomaly_count": self._coerce_int_value(row.get("anomaly_count")), "digests_with_tfs": self._coerce_int_value(row.get("digests_with_tfs")), "digests_with_anomaly": self._coerce_int_value(row.get("digests_with_anomaly")), "max_act_rows": self._coerce_int_value(row.get("max_act_rows")), } def _query_table_full_scan_samples_synapse( self, query_start: datetime, query_end: datetime, digests: list[str], *, sample_size: int, vendor: str, ignore_dml: bool, ignore_tiflash: bool, ) -> Dict[str, list[Dict[str, Any]]]: samples: Dict[str, list[Dict[str, Any]]] = {digest: [] for digest in digests} select_clause = ", ".join(self._quote_identifier(column, vendor) for column in SLOW_LOG_COLUMNS) op_condition = self._build_synapse_table_full_scan_operator_condition(vendor) tiflash_filter = self._build_synapse_tiflash_filter(vendor, ignore_tiflash=ignore_tiflash) act_rows_expr = self._build_synapse_json_numeric_expr("$.actRows", "int") digest_col = self._column_ref("digest", vendor) time_sort = self._column_ref("time", vendor) conn_id_sort = self._synapse_text_expr(self._column_ref("conn_id", vendor), length=128) query_time_col = self._column_ref("query_time", vendor) db_col = self._column_ref("db", vendor) for batch_number, digest_batch in enumerate( self._chunked(digests, DEFAULT_SAMPLE_DIGEST_BATCH_SIZE), start=1, ): digest_values = ", ".join( "'" + self._escape_sql_string(digest) + "'" for digest in digest_batch ) where_clause = self._build_table_full_scan_where_clause( query_start, query_end, vendor=vendor, ignore_dml=ignore_dml, ) where_clause += f" AND {digest_col} IN ({digest_values})" ctes = self._build_synapse_table_full_scan_ctes(where_clause, list(SLOW_LOG_COLUMNS), vendor) sql = ( f"{ctes}, " "expanded AS (" f"SELECT {select_clause}, {act_rows_expr} AS op_act_rows " "FROM plan_nodes " f"WHERE {op_condition}{tiflash_filter}" "), " "ranked_ops AS (" f"SELECT {select_clause}, op_act_rows AS sample_act_rows, " "ROW_NUMBER() OVER (" f"PARTITION BY {time_sort}, {conn_id_sort}, {digest_col}, {query_time_col}, {db_col} " "ORDER BY op_act_rows DESC" ") AS op_rank " "FROM expanded " "WHERE op_act_rows IS NOT NULL" "), " "best_log AS (" f"SELECT {select_clause}, sample_act_rows " "FROM ranked_ops " "WHERE op_rank = 1" ") " f"SELECT {select_clause}, sample_act_rows FROM (" "SELECT best_log.*, " "ROW_NUMBER() OVER (" f"PARTITION BY {digest_col} " f"ORDER BY sample_act_rows DESC, {time_sort} DESC, {conn_id_sort} DESC" ") AS sample_rank " "FROM best_log" ") ranked " f"WHERE sample_rank <= {int(sample_size)} " f"ORDER BY {digest_col}, sample_act_rows DESC, {time_sort} DESC" ) logger.info( "Querying TableFullScan samples cluster=%s batch=%d digests=%d sample_size=%d native_sql=%s dialect=synapse", self.cluster_id, batch_number, len(digest_batch), sample_size, True, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=batch_number, skip_sql_rewrite=False, query_name="tablefullscan.samples", ) logger.info( "TableFullScan samples returned cluster=%s batch=%d rows=%d trace_id=%s dialect=synapse", self.cluster_id, batch_number, len(rows), trace_id or "-", ) for raw_row in rows: row = {col: raw_row[i] if i < len(raw_row) else None for i, col in enumerate(columns)} digest = str(row.get("digest") or "") if not digest: continue samples.setdefault(digest, []).append(self._build_sample(row)) return samples def _map_table_full_scan_summary_rows( self, columns: list[str], rows: list[list[Any]], ) -> list[Dict[str, Any]]: results = [] for raw_row in rows: row = {col: raw_row[i] if i < len(raw_row) else None for i, col in enumerate(columns)} max_act = self._coerce_int_value(row.get("max_act_rows")) max_qt = self._coerce_float(row.get("max_query_time")) or 0.0 results.append({ "digest": str(row.get("digest") or ""), "query": str(row.get("query") or ""), "tfs_count": self._coerce_int_value(row.get("tfs_count")), "max_act_rows": max_act, "total_scanned_rows": self._coerce_int_value(row.get("total_act_rows")), "max_query_time": max_qt, "total_query_time": self._coerce_float(row.get("total_query_time")) or 0.0, "anomaly_count": self._coerce_int_value(row.get("anomaly_count")), "p50_act_rows": self._coerce_int_value(row.get("p50_act_rows")) or max_act, "p80_act_rows": self._coerce_int_value(row.get("p80_act_rows")) or max_act, "p95_act_rows": self._coerce_int_value(row.get("p95_act_rows")) or max_act, "p99_act_rows": self._coerce_int_value(row.get("p99_act_rows")) or max_act, "p50_query_time": self._coerce_float(row.get("p50_query_time")) or max_qt, "p80_query_time": self._coerce_float(row.get("p80_query_time")) or max_qt, "p95_query_time": self._coerce_float(row.get("p95_query_time")) or max_qt, "p99_query_time": self._coerce_float(row.get("p99_query_time")) or max_qt, "overview": { "tfs_count": self._coerce_int_value(row.get("overall_tfs_count")), "anomaly_count": self._coerce_int_value(row.get("overall_anomaly_count")), "digests_with_tfs": self._coerce_int_value(row.get("overall_digests_with_tfs")), "digests_with_anomaly": self._coerce_int_value(row.get("overall_digests_with_anomaly")), "max_act_rows": self._coerce_int_value(row.get("overall_max_act_rows")), }, }) return results def _query_table_full_scan_summary_sliced( self, query_start: datetime, query_end: datetime, *, threshold: int, vendor: str, top_n: int, ignore_dml: bool, ignore_tiflash: bool, ) -> list[Dict[str, Any]]: slice_top_n = self._table_full_scan_slice_top_n(top_n) slice_results: list[list[Dict[str, Any]]] = [] windows = list(self._iter_time_slices(query_start, query_end)) logger.info( "TableFullScan summary slice fallback started cluster=%s windows=%d slice_seconds=%d slice_top_n=%d", self.cluster_id, len(windows), self._table_full_scan_slice_seconds(), slice_top_n, ) for index, (window_start, window_end) in enumerate(windows, start=1): logger.info( "Querying TableFullScan summary slice cluster=%s slice=%d/%d range=%s~%s top_n=%d", self.cluster_id, index, len(windows), window_start.isoformat(), window_end.isoformat(), slice_top_n, ) slice_rows = self._query_table_full_scan_summary_slice_recursive( window_start, window_end, threshold=threshold, vendor=vendor, top_n=slice_top_n, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) slice_results.append(slice_rows) merged = self._merge_table_full_scan_summary_slices(slice_results, top_n=top_n) logger.info( "TableFullScan summary slice fallback completed cluster=%s windows=%d merged_digests=%d", self.cluster_id, len(windows), len(merged), ) return merged def _query_table_full_scan_summary_slice_recursive( self, query_start: datetime, query_end: datetime, *, threshold: int, vendor: str, top_n: int, ignore_dml: bool, ignore_tiflash: bool, ) -> list[Dict[str, Any]]: try: return self._query_table_full_scan_summary_direct( query_start, query_end, threshold=threshold, vendor=vendor, top_n=top_n, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) except DataProxyQueryError as exc: if not exc.is_bytes_billed_limit_exceeded: raise left, right = self._split_time_window(query_start, query_end) if left is None or right is None: raise logger.warning( "TableFullScan summary slice still exceeded bytes billed limit; splitting cluster=%s range=%s~%s left=%s~%s right=%s~%s error=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), left[0].isoformat(), left[1].isoformat(), right[0].isoformat(), right[1].isoformat(), exc.raw_message, ) return ( self._query_table_full_scan_summary_slice_recursive( left[0], left[1], threshold=threshold, vendor=vendor, top_n=top_n, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) + self._query_table_full_scan_summary_slice_recursive( right[0], right[1], threshold=threshold, vendor=vendor, top_n=top_n, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) ) def _query_table_full_scan_overview_sliced( self, query_start: datetime, query_end: datetime, *, threshold: int, vendor: str, ignore_dml: bool, ignore_tiflash: bool, ) -> Dict[str, Any]: overviews: list[Dict[str, Any]] = [] windows = list(self._iter_time_slices(query_start, query_end)) logger.info( "TableFullScan overview slice fallback started cluster=%s windows=%d slice_seconds=%d", self.cluster_id, len(windows), self._table_full_scan_slice_seconds(), ) for index, (window_start, window_end) in enumerate(windows, start=1): logger.info( "Querying TableFullScan overview slice cluster=%s slice=%d/%d range=%s~%s", self.cluster_id, index, len(windows), window_start.isoformat(), window_end.isoformat(), ) overview = self._query_table_full_scan_overview_slice_recursive( window_start, window_end, threshold=threshold, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) if overview: overviews.append(overview) merged = self._merge_table_full_scan_overviews(overviews) logger.info( "TableFullScan overview slice fallback completed cluster=%s windows=%d has_overview=%s", self.cluster_id, len(windows), bool(merged), ) return merged def _query_table_full_scan_overview_slice_recursive( self, query_start: datetime, query_end: datetime, *, threshold: int, vendor: str, ignore_dml: bool, ignore_tiflash: bool, ) -> Dict[str, Any]: try: return self._query_table_full_scan_overview_direct( query_start, query_end, threshold=threshold, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) except DataProxyQueryError as exc: if not exc.is_bytes_billed_limit_exceeded: raise left, right = self._split_time_window(query_start, query_end) if left is None or right is None: raise logger.warning( "TableFullScan overview slice still exceeded bytes billed limit; splitting cluster=%s range=%s~%s error=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), exc.raw_message, ) return self._merge_table_full_scan_overviews([ self._query_table_full_scan_overview_slice_recursive( left[0], left[1], threshold=threshold, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ), self._query_table_full_scan_overview_slice_recursive( right[0], right[1], threshold=threshold, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ), ]) def _query_table_full_scan_samples_sliced( self, query_start: datetime, query_end: datetime, digests: list[str], *, sample_size: int, vendor: str, ignore_dml: bool, ignore_tiflash: bool, ) -> Dict[str, list[Dict[str, Any]]]: sample_groups: list[Dict[str, list[Dict[str, Any]]]] = [] windows = list(self._iter_time_slices(query_start, query_end)) logger.info( "TableFullScan samples slice fallback started cluster=%s windows=%d digests=%d sample_size=%d slice_seconds=%d", self.cluster_id, len(windows), len(digests), sample_size, self._table_full_scan_slice_seconds(), ) for index, (window_start, window_end) in enumerate(windows, start=1): logger.info( "Querying TableFullScan samples slice cluster=%s slice=%d/%d range=%s~%s digests=%d", self.cluster_id, index, len(windows), window_start.isoformat(), window_end.isoformat(), len(digests), ) sample_groups.append(self._query_table_full_scan_samples_slice_recursive( window_start, window_end, digests, sample_size=sample_size, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, )) merged = self._merge_table_full_scan_sample_groups(sample_groups, digests, sample_size=sample_size) logger.info( "TableFullScan samples slice fallback completed cluster=%s windows=%d sample_digest_groups=%d", self.cluster_id, len(windows), len(merged), ) return merged def _query_table_full_scan_samples_slice_recursive( self, query_start: datetime, query_end: datetime, digests: list[str], *, sample_size: int, vendor: str, ignore_dml: bool, ignore_tiflash: bool, ) -> Dict[str, list[Dict[str, Any]]]: try: return self._query_table_full_scan_samples_direct( query_start, query_end, digests, sample_size=sample_size, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ) except DataProxyQueryError as exc: if not exc.is_bytes_billed_limit_exceeded: raise left, right = self._split_time_window(query_start, query_end) if left is None or right is None: raise logger.warning( "TableFullScan samples slice still exceeded bytes billed limit; splitting cluster=%s range=%s~%s digests=%d error=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), len(digests), exc.raw_message, ) return self._merge_table_full_scan_sample_groups([ self._query_table_full_scan_samples_slice_recursive( left[0], left[1], digests, sample_size=sample_size, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ), self._query_table_full_scan_samples_slice_recursive( right[0], right[1], digests, sample_size=sample_size, vendor=vendor, ignore_dml=ignore_dml, ignore_tiflash=ignore_tiflash, ), ], digests, sample_size=sample_size) def _merge_table_full_scan_summary_slices( self, slice_results: list[list[Dict[str, Any]]], *, top_n: int, ) -> list[Dict[str, Any]]: by_digest: Dict[str, Dict[str, Any]] = {} overview = self._merge_table_full_scan_overviews([ summaries[0].get("overview", {}) for summaries in slice_results if summaries ]) for summaries in slice_results: for summary in summaries: digest = str(summary.get("digest") or "") if not digest: continue aggregate = by_digest.get(digest) if aggregate is None: aggregate = dict(summary) aggregate.pop("overview", None) by_digest[digest] = aggregate continue if self._coerce_int_value(summary.get("total_scanned_rows")) > self._coerce_int_value(aggregate.get("total_scanned_rows")): aggregate["query"] = summary.get("query", aggregate.get("query", "")) for key in ("tfs_count", "total_scanned_rows", "total_query_time", "anomaly_count"): aggregate[key] = ( (self._coerce_float(aggregate.get(key)) or 0) + (self._coerce_float(summary.get(key)) or 0) ) for key in ("max_act_rows", "max_query_time"): aggregate[key] = max( self._coerce_float(aggregate.get(key)) or 0, self._coerce_float(summary.get(key)) or 0, ) for key in ("p50_act_rows", "p80_act_rows", "p95_act_rows", "p99_act_rows"): aggregate[key] = max( self._coerce_int_value(aggregate.get(key)), self._coerce_int_value(summary.get(key)), ) for key in ("p50_query_time", "p80_query_time", "p95_query_time", "p99_query_time"): aggregate[key] = max( self._coerce_float(aggregate.get(key)) or 0, self._coerce_float(summary.get(key)) or 0, ) if not overview and by_digest: overview = { "tfs_count": sum(self._coerce_int_value(summary.get("tfs_count")) for summary in by_digest.values()), "anomaly_count": sum(self._coerce_int_value(summary.get("anomaly_count")) for summary in by_digest.values()), "digests_with_tfs": len(by_digest), "digests_with_anomaly": len([summary for summary in by_digest.values() if self._coerce_int_value(summary.get("anomaly_count")) > 0]), "max_act_rows": max(self._coerce_int_value(summary.get("max_act_rows")) for summary in by_digest.values()), } else: overview["digests_with_anomaly"] = max( self._coerce_int_value(overview.get("digests_with_anomaly")), len([summary for summary in by_digest.values() if self._coerce_int_value(summary.get("anomaly_count")) > 0]), ) merged = sorted( by_digest.values(), key=lambda item: self._coerce_int_value(item.get("total_scanned_rows")), reverse=True, )[:top_n] for summary in merged: summary["tfs_count"] = self._coerce_int_value(summary.get("tfs_count")) summary["total_scanned_rows"] = self._coerce_int_value(summary.get("total_scanned_rows")) summary["total_query_time"] = self._coerce_float(summary.get("total_query_time")) or 0.0 summary["anomaly_count"] = self._coerce_int_value(summary.get("anomaly_count")) summary["max_act_rows"] = self._coerce_int_value(summary.get("max_act_rows")) summary["max_query_time"] = self._coerce_float(summary.get("max_query_time")) or 0.0 summary["overview"] = overview return merged def _merge_table_full_scan_overviews(self, overviews: list[Dict[str, Any]]) -> Dict[str, Any]: merged: Dict[str, Any] = {} for overview in overviews: if not overview: continue merged["tfs_count"] = self._coerce_int_value(merged.get("tfs_count")) + self._coerce_int_value(overview.get("tfs_count")) merged["anomaly_count"] = self._coerce_int_value(merged.get("anomaly_count")) + self._coerce_int_value(overview.get("anomaly_count")) merged["digests_with_tfs"] = self._coerce_int_value(merged.get("digests_with_tfs")) + self._coerce_int_value(overview.get("digests_with_tfs")) merged["digests_with_anomaly"] = self._coerce_int_value(merged.get("digests_with_anomaly")) + self._coerce_int_value(overview.get("digests_with_anomaly")) merged["max_act_rows"] = max( self._coerce_int_value(merged.get("max_act_rows")), self._coerce_int_value(overview.get("max_act_rows")), ) return merged def _merge_table_full_scan_sample_groups( self, sample_groups: list[Dict[str, list[Dict[str, Any]]]], digests: list[str], *, sample_size: int, ) -> Dict[str, list[Dict[str, Any]]]: merged: Dict[str, list[Dict[str, Any]]] = {digest: [] for digest in digests} for group in sample_groups: for digest, samples in group.items(): merged.setdefault(digest, []).extend(samples) for digest, samples in list(merged.items()): samples.sort( key=lambda sample: ( self._coerce_int_value(sample.get("act_rows")), self._coerce_float(sample.get("time")) or 0, ), reverse=True, ) merged[digest] = samples[:sample_size] return merged def _iter_time_slices(self, start_time: datetime, end_time: datetime) -> Iterator[tuple[datetime, datetime]]: slice_seconds = self._table_full_scan_slice_seconds() cursor = start_time while cursor < end_time: next_cursor = min(cursor + timedelta(seconds=slice_seconds), end_time) yield cursor, next_cursor cursor = next_cursor def _split_time_window( self, start_time: datetime, end_time: datetime, ) -> tuple[tuple[datetime, datetime] | None, tuple[datetime, datetime] | None]: duration_seconds = (end_time - start_time).total_seconds() if duration_seconds <= self._table_full_scan_min_slice_seconds(): return None, None midpoint = start_time + timedelta(seconds=duration_seconds / 2) if midpoint <= start_time or midpoint >= end_time: return None, None return (start_time, midpoint), (midpoint, end_time) def _table_full_scan_slice_seconds(self) -> int: return self._resolve_positive_int( None, env_name="TABLEFULLSCAN_FALLBACK_SLICE_SECONDS", default=DEFAULT_TABLE_FULL_SCAN_SLICE_SECONDS, ) def _table_full_scan_min_slice_seconds(self) -> int: return self._resolve_positive_int( None, env_name="TABLEFULLSCAN_FALLBACK_MIN_SLICE_SECONDS", default=DEFAULT_TABLE_FULL_SCAN_MIN_SLICE_SECONDS, ) def _table_full_scan_slice_top_n(self, top_n: int) -> int: factor = self._resolve_positive_int( None, env_name="TABLEFULLSCAN_FALLBACK_OVERFETCH_FACTOR", default=DEFAULT_TABLE_FULL_SCAN_SLICE_OVERFETCH_FACTOR, ) return max(int(top_n), int(top_n) * factor) def query_slow_log_top_digest( self, start_time: datetime, end_time: datetime, dimensions: list[str], top_n: int = 10, vendor: str = "aws", ignore_dml: bool = False, dimension_count_thresholds: dict[str, int] | None = None, min_count: int = 0, min_total_keys: int = 0, ) -> Dict[str, list[Dict[str, Any]]]: """Query Top Digest summaries by digest for selected slow log dimensions.""" query_start = self._to_utc(start_time) query_end = self._to_utc(end_time) if query_start >= query_end or top_n < 1 or not dimensions: return {} count_thresholds = dimension_count_thresholds or {} unified_min_count = max(0, int(min_count or 0)) row_min_total_keys = max(0, int(min_total_keys or 0)) if self._is_synapse_vendor(vendor): return self._query_slow_log_top_digest_synapse( query_start, query_end, dimensions, top_n=top_n, vendor=vendor, ignore_dml=ignore_dml, dimension_count_thresholds=count_thresholds, min_count=unified_min_count, min_total_keys=row_min_total_keys, ) where_clause = self._build_slow_log_top_digest_where_clause( query_start, query_end, vendor=vendor, ignore_dml=ignore_dml, ) query_time_expr = self._build_numeric_cast_expr("query_time", "float", vendor) total_keys_expr = self._build_numeric_cast_expr("total_keys", "int", vendor) process_keys_expr = self._build_numeric_cast_expr("process_keys", "int", vendor) request_count_expr = self._build_numeric_cast_expr("request_count", "int", vendor) ru_read_expr = self._build_zero_safe_numeric_expr("request_unit_read", "float", vendor) ru_write_expr = self._build_zero_safe_numeric_expr("request_unit_write", "float", vendor) if row_min_total_keys > 0: total_keys_filter_expr = self._build_zero_safe_numeric_expr("total_keys", "int", vendor) where_clause += f" AND {total_keys_filter_expr} >= {row_min_total_keys}" query_limit = self._top_digest_query_limit(top_n, ignore_dml) results: Dict[str, list[Dict[str, Any]]] = {} for dimension in dimensions: config = TOP_DIGEST_DIMENSIONS[dimension] order_alias = config["order_alias"] result_key = config["result_key"] legacy_count_threshold = max(0, int(count_thresholds.get(dimension, 0) or 0)) having_clause = ( self._build_top_digest_count_filter(unified_min_count, vendor) if unified_min_count > 0 else self._build_top_digest_legacy_count_threshold_filter(legacy_count_threshold, vendor) ) order_by_clause = self._build_top_digest_order_by_clause(order_alias) sql = ( "SELECT " "digest, " "any_value(query) AS query, " "COUNT(*) AS slow_count, " f"SUM({query_time_expr}) AS total_latency_s, " f"SUM(COALESCE({total_keys_expr}, 0)) AS total_keys, " f"SUM(COALESCE({process_keys_expr}, 0)) AS process_keys, " f"SUM(COALESCE({request_count_expr}, 0)) AS request_count, " f"SUM({ru_read_expr} + {ru_write_expr}) AS sum_ru, " f"MAX({query_time_expr}) AS max_latency_s, " "MAX(time) AS latest_time " "FROM clinic_data_proxy.slow_query_logs " f"WHERE {where_clause} " "GROUP BY digest " f"{having_clause}" f"{order_by_clause} " f"LIMIT {query_limit}" ) logger.info( "Querying Slow Log Top Digest cluster=%s range=%s~%s dimension=%s top_n=%d ignore_dml=%s min_count=%d min_total_keys=%d legacy_count_threshold=%d", self.cluster_id, query_start.isoformat(), query_end.isoformat(), dimension, top_n, ignore_dml, unified_min_count, row_min_total_keys, legacy_count_threshold, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=1, skip_sql_rewrite=self._use_native_sql(vendor), query_name=f"slowlog_top_digest.{dimension}", ) logger.info( "Slow Log Top Digest returned cluster=%s dimension=%s rows=%d trace_id=%s", self.cluster_id, dimension, len(rows), trace_id or "-", ) dimension_results: list[Dict[str, Any]] = [] for raw_row in rows: row = {col: raw_row[i] if i < len(raw_row) else None for i, col in enumerate(columns)} query_text = str(row.get("query") or "") if ignore_dml and self._is_tidb_dml_query(query_text): continue dimension_results.append({ "rank": len(dimension_results) + 1, "digest": str(row.get("digest") or ""), "query": query_text, "slow_count": self._coerce_int_value(row.get("slow_count")), "total_latency_s": round(self._coerce_float(row.get("total_latency_s")) or 0.0, 3), "total_keys": self._coerce_int_value(row.get("total_keys")), "process_keys": self._coerce_int_value(row.get("process_keys")), "request_count": self._coerce_int_value(row.get("request_count")), "sum_ru": round(self._coerce_float(row.get("sum_ru")) or 0.0, 3), "max_latency_s": round(self._coerce_float(row.get("max_latency_s")) or 0.0, 3), "latest_time": self._coerce_float(row.get("latest_time")), }) if len(dimension_results) >= top_n: break results[result_key] = dimension_results return results def _query_slow_log_top_digest_synapse( self, query_start: datetime, query_end: datetime, dimensions: list[str], *, top_n: int, vendor: str, ignore_dml: bool = False, dimension_count_thresholds: dict[str, int] | None = None, min_count: int = 0, min_total_keys: int = 0, ) -> Dict[str, list[Dict[str, Any]]]: where_clause = self._build_slow_log_top_digest_azure_rewrite_where_clause( query_start, query_end, ignore_dml=ignore_dml, ) query_time_expr = "query_time" total_keys_expr = "COALESCE(CAST(total_keys AS SIGNED), 0)" process_keys_expr = "COALESCE(CAST(process_keys AS SIGNED), 0)" request_count_expr = "COALESCE(CAST(request_count AS SIGNED), 0)" ru_read_expr = "COALESCE(NULLIF(request_unit_read, ''), 0)" ru_write_expr = "COALESCE(NULLIF(request_unit_write, ''), 0)" row_min_total_keys = max(0, int(min_total_keys or 0)) if row_min_total_keys > 0: where_clause += f" AND {total_keys_expr} >= {row_min_total_keys}" query_limit = self._top_digest_query_limit(top_n, ignore_dml) count_thresholds = dimension_count_thresholds or {} unified_min_count = max(0, int(min_count or 0)) results: Dict[str, list[Dict[str, Any]]] = {} for dimension in dimensions: config = TOP_DIGEST_DIMENSIONS[dimension] order_alias = config["order_alias"] result_key = config["result_key"] legacy_count_threshold = max(0, int(count_thresholds.get(dimension, 0) or 0)) having_clause = ( self._build_top_digest_count_filter(unified_min_count, vendor) if unified_min_count > 0 else self._build_top_digest_legacy_count_threshold_filter(legacy_count_threshold, vendor) ) order_by_clause = self._build_top_digest_order_by_clause(order_alias) sql = ( "SELECT " "digest, " "MAX(query) AS query, " "COUNT(*) AS slow_count, " f"SUM({query_time_expr}) AS total_latency_s, " f"SUM({total_keys_expr}) AS total_keys, " f"SUM({process_keys_expr}) AS process_keys, " f"SUM({request_count_expr}) AS request_count, " f"SUM({ru_read_expr} + {ru_write_expr}) AS sum_ru, " f"MAX({query_time_expr}) AS max_latency_s, " "MAX(time) AS latest_time " "FROM clinic_data_proxy.slow_query_logs " f"WHERE {where_clause} " "GROUP BY digest " f"{having_clause}" f"{order_by_clause} " f"LIMIT {query_limit}" ) logger.info( "Querying Slow Log Top Digest cluster=%s range=%s~%s dimension=%s top_n=%d ignore_dml=%s min_count=%d min_total_keys=%d legacy_count_threshold=%d dialect=synapse", self.cluster_id, query_start.isoformat(), query_end.isoformat(), dimension, top_n, ignore_dml, unified_min_count, row_min_total_keys, legacy_count_threshold, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=1, skip_sql_rewrite=False, query_name=f"slowlog_top_digest.{dimension}", ) logger.info( "Slow Log Top Digest returned cluster=%s dimension=%s rows=%d trace_id=%s dialect=synapse", self.cluster_id, dimension, len(rows), trace_id or "-", ) dimension_results: list[Dict[str, Any]] = [] for raw_row in rows: row = {col: raw_row[i] if i < len(raw_row) else None for i, col in enumerate(columns)} query_text = str(row.get("query") or "") if ignore_dml and self._is_tidb_dml_query(query_text): continue dimension_results.append({ "rank": len(dimension_results) + 1, "digest": str(row.get("digest") or ""), "query": query_text, "slow_count": self._coerce_int_value(row.get("slow_count")), "total_latency_s": round(self._coerce_float(row.get("total_latency_s")) or 0.0, 3), "total_keys": self._coerce_int_value(row.get("total_keys")), "process_keys": self._coerce_int_value(row.get("process_keys")), "request_count": self._coerce_int_value(row.get("request_count")), "sum_ru": round(self._coerce_float(row.get("sum_ru")) or 0.0, 3), "max_latency_s": round(self._coerce_float(row.get("max_latency_s")) or 0.0, 3), "latest_time": self._coerce_float(row.get("latest_time")), }) if len(dimension_results) >= top_n: break results[result_key] = dimension_results return results def query_tikv_hotspot_slow_log_enrichment( self, start_time: datetime, end_time: datetime, digests: list[str], *, instance: str, vendor: str = "aws", limit: int = 10, include_instance_cop_candidates: bool = False, ) -> Dict[str, Dict[str, Any]]: """Query slow log enrichment for TiKV hotspot attribution. The count intentionally uses COUNT(*) to keep the same slow-log count semantics as existing SQL Health Check items. """ query_start = self._to_utc(start_time) query_end = self._to_utc(end_time) if query_start >= query_end or limit < 1: return {} if self._is_synapse_vendor(vendor): return self._query_tikv_hotspot_slow_log_enrichment_synapse( query_start, query_end, digests, instance=instance, vendor=vendor, limit=limit, include_instance_cop_candidates=include_instance_cop_candidates, ) digest_values = self._digest_in_values(digests) proc_addr_condition = self._build_tikv_addr_match_condition(instance, column="cop_proc_addr") wait_addr_condition = self._build_tikv_addr_match_condition(instance, column="cop_wait_addr") addr_condition = f"({proc_addr_condition} OR {wait_addr_condition})" where_clause = self._build_tikv_hotspot_slow_log_where_clause(query_start, query_end, vendor=vendor) if digest_values and include_instance_cop_candidates: where_clause += f" AND (LOWER(digest) IN ({digest_values}) OR {addr_condition})" elif digest_values: where_clause += f" AND LOWER(digest) IN ({digest_values})" elif include_instance_cop_candidates: where_clause += f" AND {addr_condition}" else: return {} query_time_expr = self._build_numeric_cast_expr("query_time", "float", vendor) mem_max_expr = self._build_numeric_cast_expr("mem_max", "int", vendor) process_keys_expr = self._build_numeric_cast_expr("process_keys", "int", vendor) total_keys_expr = self._build_numeric_cast_expr("total_keys", "int", vendor) request_count_expr = self._build_numeric_cast_expr("request_count", "int", vendor) cop_proc_expr = self._build_numeric_cast_expr("cop_proc_max", "float", vendor) cop_wait_expr = self._build_numeric_cast_expr("cop_wait_max", "float", vendor) requested_digest_expr = f"CASE WHEN LOWER(digest) IN ({digest_values}) THEN 1 ELSE 0 END" if digest_values else "0" base_select = ( "SELECT " "LOWER(digest) AS digest, " "query, " "time, " f"{CONN_ID_EXPR} AS conn_id, " f"{DB_EXPR} AS db, " f"{query_time_expr} AS query_time_s, " f"{mem_max_expr} AS mem_max_bytes, " f"{process_keys_expr} AS process_keys_value, " f"{total_keys_expr} AS total_keys_value, " f"{request_count_expr} AS request_count_value, " f"{cop_proc_expr} AS cop_proc_max_s, " f"{cop_wait_expr} AS cop_wait_max_s, " "cop_proc_addr, " "cop_wait_addr, " f"{self._build_tikv_addr_match_condition(instance, column='cop_proc_addr')} AS cop_proc_addr_match, " f"{self._build_tikv_addr_match_condition(instance, column='cop_wait_addr')} AS cop_wait_addr_match, " f"{requested_digest_expr} AS requested_digest " "FROM clinic_data_proxy.slow_query_logs " f"WHERE {where_clause}" ) aggregate_select = ( "SELECT " "digest, " "any_value(query) AS query, " "COUNT(*) AS slow_count, " "MAX(query_time_s) AS max_query_time_s, " "AVG(COALESCE(query_time_s, 0)) AS avg_query_time_s, " "MAX(COALESCE(mem_max_bytes, 0)) AS max_mem_bytes, " "AVG(COALESCE(mem_max_bytes, 0)) AS avg_mem_bytes, " "MAX(COALESCE(process_keys_value, 0)) AS max_process_keys, " "MAX(COALESCE(total_keys_value, 0)) AS max_total_keys, " "MAX(COALESCE(request_count_value, 0)) AS max_request_count, " "MAX(COALESCE(cop_proc_max_s, 0)) AS cop_proc_max_s, " "MAX(COALESCE(cop_wait_max_s, 0)) AS cop_wait_max_s, " "SUM(CASE WHEN cop_proc_addr_match THEN 1 ELSE 0 END) AS cop_proc_addr_match_count, " "SUM(CASE WHEN cop_wait_addr_match THEN 1 ELSE 0 END) AS cop_wait_addr_match_count, " "MAX(CASE WHEN cop_proc_addr_match THEN COALESCE(cop_proc_max_s, 0) ELSE 0 END) AS matched_cop_proc_max_s, " "MAX(CASE WHEN cop_wait_addr_match THEN COALESCE(cop_wait_max_s, 0) ELSE 0 END) AS matched_cop_wait_max_s, " "MAX(requested_digest) AS requested_digest " f"FROM ({base_select}) base " "GROUP BY digest" ) sample_select = ( "SELECT " "digest, " "time AS sample_time, " "conn_id AS sample_conn_id, " "db AS sample_db, " "query AS sample_query, " "query_time_s AS sample_query_time_s, " "cop_proc_addr AS sample_cop_proc_addr, " "cop_wait_addr AS sample_cop_wait_addr " "FROM (" "SELECT base.*, " "ROW_NUMBER() OVER (" "PARTITION BY digest " "ORDER BY query_time_s DESC, time DESC, conn_id DESC" ") AS sample_rank " f"FROM ({base_select}) base" ") ranked " "WHERE sample_rank = 1" ) sql = ( "SELECT " "agg.*, " "sample.sample_time, " "sample.sample_conn_id, " "sample.sample_db, " "sample.sample_query, " "sample.sample_query_time_s, " "sample.sample_cop_proc_addr, " "sample.sample_cop_wait_addr " f"FROM ({aggregate_select}) agg " f"LEFT JOIN ({sample_select}) sample ON agg.digest = sample.digest " "ORDER BY requested_digest DESC, " "(cop_proc_addr_match_count + cop_wait_addr_match_count) DESC, " "max_query_time_s DESC, slow_count DESC, digest DESC " f"LIMIT {int(limit)}" ) logger.info( "Querying TiKV hotspot slow log enrichment cluster=%s range=%s~%s digests=%d limit=%d include_instance_cop_candidates=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), len(digests), limit, include_instance_cop_candidates, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=1, skip_sql_rewrite=self._use_native_sql(vendor), query_name="tikv_hotspot.slow_log_enrichment", ) logger.info( "TiKV hotspot slow log enrichment returned cluster=%s rows=%d trace_id=%s", self.cluster_id, len(rows), trace_id or "-", ) return self._map_tikv_hotspot_slow_log_enrichment_rows(columns, rows) def _query_tikv_hotspot_slow_log_enrichment_synapse( self, query_start: datetime, query_end: datetime, digests: list[str], *, instance: str, vendor: str, limit: int, include_instance_cop_candidates: bool, ) -> Dict[str, Dict[str, Any]]: digest_values = self._digest_in_values(digests) digest_expr = f"LOWER({self._column_ref('digest', vendor)})" proc_addr_condition = self._build_tikv_addr_match_condition( instance, column=self._column_ref("cop_proc_addr", vendor), ) wait_addr_condition = self._build_tikv_addr_match_condition( instance, column=self._column_ref("cop_wait_addr", vendor), ) addr_condition = f"({proc_addr_condition} OR {wait_addr_condition})" where_clause = self._build_tikv_hotspot_slow_log_where_clause(query_start, query_end, vendor=vendor) if digest_values and include_instance_cop_candidates: where_clause += f" AND ({digest_expr} IN ({digest_values}) OR {addr_condition})" elif digest_values: where_clause += f" AND {digest_expr} IN ({digest_values})" elif include_instance_cop_candidates: where_clause += f" AND {addr_condition}" else: return {} query_time_expr = self._build_zero_safe_numeric_expr(self._column_ref("query_time", vendor), "float", vendor) mem_max_expr = self._build_zero_safe_numeric_expr(self._column_ref("mem_max", vendor), "int", vendor) process_keys_expr = self._build_zero_safe_numeric_expr(self._column_ref("process_keys", vendor), "int", vendor) total_keys_expr = self._build_zero_safe_numeric_expr(self._column_ref("total_keys", vendor), "int", vendor) request_count_expr = self._build_zero_safe_numeric_expr(self._column_ref("request_count", vendor), "int", vendor) cop_proc_expr = self._build_zero_safe_numeric_expr(self._column_ref("cop_proc_max", vendor), "float", vendor) cop_wait_expr = self._build_zero_safe_numeric_expr(self._column_ref("cop_wait_max", vendor), "float", vendor) requested_digest_expr = f"CASE WHEN {digest_expr} IN ({digest_values}) THEN 1 ELSE 0 END" if digest_values else "0" conn_id_expr = self._synapse_text_expr(self._column_ref("conn_id", vendor), length=128) db_expr = self._synapse_text_expr(self._column_ref("db", vendor), length=4000) time_col = self._column_ref("time", vendor) query_col = self._column_ref("query", vendor) cop_proc_addr_col = self._column_ref("cop_proc_addr", vendor) cop_wait_addr_col = self._column_ref("cop_wait_addr", vendor) base_select = ( "SELECT " f"{digest_expr} AS digest, " f"CAST({query_col} AS NVARCHAR(4000)) AS query, " f"{time_col} AS time, " f"{conn_id_expr} AS conn_id, " f"{db_expr} AS db, " f"{query_time_expr} AS query_time_s, " f"{mem_max_expr} AS mem_max_bytes, " f"{process_keys_expr} AS process_keys_value, " f"{total_keys_expr} AS total_keys_value, " f"{request_count_expr} AS request_count_value, " f"{cop_proc_expr} AS cop_proc_max_s, " f"{cop_wait_expr} AS cop_wait_max_s, " f"{cop_proc_addr_col} AS cop_proc_addr, " f"{cop_wait_addr_col} AS cop_wait_addr, " f"CASE WHEN {proc_addr_condition} THEN 1 ELSE 0 END AS cop_proc_addr_match, " f"CASE WHEN {wait_addr_condition} THEN 1 ELSE 0 END AS cop_wait_addr_match, " f"{requested_digest_expr} AS requested_digest " "FROM clinic_data_proxy.slow_query_logs " f"WHERE {where_clause}" ) aggregate_select = ( "SELECT " "digest, " "MAX(CAST(query AS NVARCHAR(4000))) AS query, " "COUNT_BIG(*) AS slow_count, " "MAX(query_time_s) AS max_query_time_s, " "AVG(COALESCE(query_time_s, 0)) AS avg_query_time_s, " "MAX(COALESCE(mem_max_bytes, 0)) AS max_mem_bytes, " "AVG(COALESCE(mem_max_bytes, 0)) AS avg_mem_bytes, " "MAX(COALESCE(process_keys_value, 0)) AS max_process_keys, " "MAX(COALESCE(total_keys_value, 0)) AS max_total_keys, " "MAX(COALESCE(request_count_value, 0)) AS max_request_count, " "MAX(COALESCE(cop_proc_max_s, 0)) AS cop_proc_max_s, " "MAX(COALESCE(cop_wait_max_s, 0)) AS cop_wait_max_s, " "SUM(CASE WHEN cop_proc_addr_match = 1 THEN 1 ELSE 0 END) AS cop_proc_addr_match_count, " "SUM(CASE WHEN cop_wait_addr_match = 1 THEN 1 ELSE 0 END) AS cop_wait_addr_match_count, " "MAX(CASE WHEN cop_proc_addr_match = 1 THEN COALESCE(cop_proc_max_s, 0) ELSE 0 END) AS matched_cop_proc_max_s, " "MAX(CASE WHEN cop_wait_addr_match = 1 THEN COALESCE(cop_wait_max_s, 0) ELSE 0 END) AS matched_cop_wait_max_s, " "MAX(requested_digest) AS requested_digest " f"FROM ({base_select}) base " "GROUP BY digest" ) sample_select = ( "SELECT " "digest, " "time AS sample_time, " "conn_id AS sample_conn_id, " "db AS sample_db, " "query AS sample_query, " "query_time_s AS sample_query_time_s, " "cop_proc_addr AS sample_cop_proc_addr, " "cop_wait_addr AS sample_cop_wait_addr " "FROM (" "SELECT base.*, " "ROW_NUMBER() OVER (" "PARTITION BY digest " "ORDER BY query_time_s DESC, time DESC, conn_id DESC" ") AS sample_rank " f"FROM ({base_select}) base" ") ranked " "WHERE sample_rank = 1" ) sql = ( f"SELECT TOP ({int(limit)}) " "agg.*, " "sample.sample_time, " "sample.sample_conn_id, " "sample.sample_db, " "sample.sample_query, " "sample.sample_query_time_s, " "sample.sample_cop_proc_addr, " "sample.sample_cop_wait_addr " f"FROM ({aggregate_select}) agg " f"LEFT JOIN ({sample_select}) sample ON agg.digest = sample.digest " "ORDER BY requested_digest DESC, " "(cop_proc_addr_match_count + cop_wait_addr_match_count) DESC, " "max_query_time_s DESC, slow_count DESC, digest DESC" ) logger.info( "Querying TiKV hotspot slow log enrichment cluster=%s range=%s~%s digests=%d limit=%d include_instance_cop_candidates=%s dialect=synapse", self.cluster_id, query_start.isoformat(), query_end.isoformat(), len(digests), limit, include_instance_cop_candidates, ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number=1, skip_sql_rewrite=False, query_name="tikv_hotspot.slow_log_enrichment", ) logger.info( "TiKV hotspot slow log enrichment returned cluster=%s rows=%d trace_id=%s dialect=synapse", self.cluster_id, len(rows), trace_id or "-", ) return self._map_tikv_hotspot_slow_log_enrichment_rows(columns, rows) def _map_tikv_hotspot_slow_log_enrichment_rows( self, columns: list[str], rows: list[list[Any]], ) -> Dict[str, Dict[str, Any]]: result: Dict[str, Dict[str, Any]] = {} for raw_row in rows: row = {col: raw_row[i] if i < len(raw_row) else None for i, col in enumerate(columns)} digest = str(row.get("digest") or "") if not digest: continue cop_proc_match_count = self._coerce_int_value(row.get("cop_proc_addr_match_count")) cop_wait_match_count = self._coerce_int_value(row.get("cop_wait_addr_match_count")) result[digest] = { "is_slow_query": True, "count": self._coerce_int_value(row.get("slow_count")), "query": str(row.get("query") or ""), "max_query_time_s": round(self._coerce_float(row.get("max_query_time_s")) or 0.0, 3), "avg_query_time_s": round(self._coerce_float(row.get("avg_query_time_s")) or 0.0, 3), "max_mem_bytes": self._coerce_int_value(row.get("max_mem_bytes")), "avg_mem_bytes": round(self._coerce_float(row.get("avg_mem_bytes")) or 0.0, 3), "max_process_keys": self._coerce_int_value(row.get("max_process_keys")), "max_total_keys": self._coerce_int_value(row.get("max_total_keys")), "max_request_count": self._coerce_int_value(row.get("max_request_count")), "cop_proc_addr_match": cop_proc_match_count > 0, "cop_wait_addr_match": cop_wait_match_count > 0, "cop_proc_addr_match_count": cop_proc_match_count, "cop_wait_addr_match_count": cop_wait_match_count, "cop_proc_max_s": round(self._coerce_float(row.get("cop_proc_max_s")) or 0.0, 3), "cop_wait_max_s": round(self._coerce_float(row.get("cop_wait_max_s")) or 0.0, 3), "matched_cop_proc_max_s": round(self._coerce_float(row.get("matched_cop_proc_max_s")) or 0.0, 3), "matched_cop_wait_max_s": round(self._coerce_float(row.get("matched_cop_wait_max_s")) or 0.0, 3), "sample_conn_id": str(row.get("sample_conn_id") or ""), "sample_time": self._coerce_float(row.get("sample_time")), "sample_db": str(row.get("sample_db") or ""), "sample_query": str(row.get("sample_query") or ""), "sample_query_time_s": round(self._coerce_float(row.get("sample_query_time_s")) or 0.0, 3), "sample_cop_proc_addr": str(row.get("sample_cop_proc_addr") or ""), "sample_cop_wait_addr": str(row.get("sample_cop_wait_addr") or ""), } return result @staticmethod def _build_numeric_cast_expr(column: str, value_type: str, vendor: str) -> str: v = vendor.lower() if vendor else "aws" if value_type == "float": if v == "gcp": return f"SAFE_CAST({column} AS FLOAT64)" return f"CAST({column} AS DOUBLE)" if v == "gcp": return f"SAFE_CAST({column} AS INT64)" if v in ("alicloud", "aliyun"): return f"CAST({column} AS BIGINT)" return f"CAST({column} AS SIGNED)" @staticmethod def _build_zero_safe_numeric_expr(column_expr: str, value_type: str, vendor: str) -> str: """Build a numeric expression that treats NULL and empty string as zero.""" v = ServiceDataSource._vendor_key(vendor) if value_type == "float": if v == "gcp": return f"COALESCE(SAFE_CAST(NULLIF(CAST({column_expr} AS STRING), '') AS FLOAT64), 0)" if v == "synapse": return f"COALESCE(TRY_CAST(NULLIF(CAST({column_expr} AS VARCHAR(64)), '') AS FLOAT), 0)" return f"COALESCE(CAST({column_expr} AS DOUBLE), 0)" if v == "gcp": return f"COALESCE(SAFE_CAST(NULLIF(CAST({column_expr} AS STRING), '') AS INT64), 0)" if v == "synapse": return f"COALESCE(TRY_CAST(NULLIF(CAST({column_expr} AS VARCHAR(64)), '') AS BIGINT), 0)" if v in ("alicloud", "aliyun"): return f"COALESCE(CAST(NULLIF(CAST({column_expr} AS STRING), '') AS BIGINT), 0)" return f"COALESCE(CAST({column_expr} AS SIGNED), 0)" @staticmethod def _build_synapse_json_numeric_expr(json_path: str, value_type: str) -> str: target_type = "FLOAT" if value_type == "float" else "BIGINT" escaped_path = json_path.replace("'", "''") return f"TRY_CAST(NULLIF(JSON_VALUE(node_json, '{escaped_path}'), '') AS {target_type})" @staticmethod def _build_percentile_exprs(act_rows_expr: str, latency_col: str, vendor: str) -> str: """Build vendor-specific approximate percentile SELECT expressions. - aws (Athena/Presto): approx_percentile(x, pct) - native aggregate - gcp (BigQuery): APPROX_QUANTILES(x, 100)[OFFSET(N)] - requires skipSQLRewrite - alicloud (MaxCompute): percentile_approx(x, pct) """ percentiles = [("p50", 0.5), ("p80", 0.8), ("p95", 0.95), ("p99", 0.99)] parts = [] v = vendor.lower() if vendor else "aws" for label, pct in percentiles: if v == "gcp": offset = int(pct * 100) rows_expr = "APPROX_QUANTILES(" + act_rows_expr + ", 100)[OFFSET(" + str(offset) + ")]" time_expr = "APPROX_QUANTILES(" + latency_col + ", 100)[OFFSET(" + str(offset) + ")]" elif v in ("alicloud", "aliyun"): rows_expr = "percentile_approx(" + act_rows_expr + ", " + str(pct) + ")" time_expr = "percentile_approx(" + latency_col + ", " + str(pct) + ")" else: # aws / default: Athena (Presto/Trino) rows_expr = "approx_percentile(" + act_rows_expr + ", " + str(pct) + ")" time_expr = "approx_percentile(" + latency_col + ", " + str(pct) + ")" parts.append(rows_expr + " AS " + label + "_act_rows") parts.append(time_expr + " AS " + label + "_query_time") return ", ".join(parts) @staticmethod def _use_native_sql(vendor: str) -> bool: v = vendor.lower() if vendor else "aws" return v in ("alicloud", "aliyun", "gcp") @staticmethod def _use_table_full_scan_native_sql(vendor: str) -> bool: v = vendor.lower() if vendor else "aws" return v in ("aws", "alicloud", "aliyun", "gcp") @staticmethod def _top_digest_query_limit(top_n: int, ignore_dml: bool) -> int: if not ignore_dml: return int(top_n) return int(top_n) * DEFAULT_DML_FILTER_OVERFETCH_FACTOR @staticmethod def _build_top_digest_order_by_clause(order_alias: str) -> str: columns: list[str] = [] for column in (order_alias, "slow_count", "digest"): if column not in columns: columns.append(column) return "ORDER BY " + ", ".join(f"{column} DESC" for column in columns) @staticmethod def _vendor_key(vendor: str) -> str: v = vendor.lower().replace("-", "_") if vendor else "aws" if v in ("azure", "synapse", "mssql", "sqlserver", "sql_server", "azure_synapse"): return "synapse" if v in ("alicloud", "aliyun"): return "alicloud" return v @staticmethod def _is_synapse_vendor(vendor: str) -> bool: return ServiceDataSource._vendor_key(vendor) == "synapse" @staticmethod def _table_full_scan_sample_log_partition_expr(vendor: str) -> str: if ServiceDataSource._vendor_key(vendor) == "gcp": return ( f"CAST({TIME_EXPR} AS STRING), {CONN_ID_EXPR}, digest, " f"CAST({QUERY_TIME_EXPR} AS STRING), {DB_EXPR}" ) return f"{TIME_EXPR}, {CONN_ID_EXPR}, digest, {QUERY_TIME_EXPR}, {DB_EXPR}" @staticmethod def _build_act_rows_expr(vendor: str, source_expr: str = "decoded_plan") -> str: """Build the SQL expression that extracts TableFullScan actRows.""" v = vendor.lower() if vendor else "aws" if v == "gcp": cast_type = "INT64" regexp_group_arg = "" elif v in ("alicloud", "aliyun"): cast_type = "BIGINT" regexp_group_arg = ", 1" else: cast_type = "BIGINT" regexp_group_arg = ", 1" pattern_literal = ServiceDataSource._sql_regex_literal(ACT_ROWS_PATTERN, vendor) return ( "CAST(REGEXP_EXTRACT(" + source_expr + ", " + pattern_literal + regexp_group_arg + ") AS " + cast_type + ")" ) def _build_table_full_scan_where_clause( self, start_time: datetime, end_time: datetime, *, vendor: str = "aws", ignore_dml: bool = False, ) -> str: date_filters = ", ".join("'%s'" % value for value in self._date_partitions(start_time, end_time)) date_col = self._column_ref("date", vendor) time_col = self._column_ref("time", vendor) decoded_plan_col = self._column_ref("decoded_plan", vendor) query_col = self._column_ref("query", vendor) conditions = [ date_col + " IN (" + date_filters + ")", time_col + " >= " + self._format_timestamp(start_time), time_col + " < " + self._format_timestamp(end_time), decoded_plan_col + " IS NOT NULL", decoded_plan_col + " != ''", decoded_plan_col + " LIKE '%TableFullScan%'", ] if ignore_dml: conditions.append(self._build_not_regex_condition(query_col, DML_QUERY_PATTERN, vendor)) return " AND ".join(conditions) def _build_slow_log_top_digest_where_clause( self, start_time: datetime, end_time: datetime, *, vendor: str = "aws", ignore_dml: bool = False, ) -> str: date_filters = ", ".join("'%s'" % value for value in self._date_partitions(start_time, end_time)) date_col = self._column_ref("date", vendor) time_col = self._column_ref("time", vendor) digest_col = self._column_ref("digest", vendor) query_col = self._column_ref("query", vendor) conditions = [ date_col + " IN (" + date_filters + ")", time_col + " >= " + self._format_timestamp(start_time), time_col + " < " + self._format_timestamp(end_time), digest_col + " IS NOT NULL", digest_col + " != ''", self._build_exclude_internal_condition(vendor), ] if ignore_dml: conditions.append(self._build_not_regex_condition(query_col, DML_QUERY_PATTERN, vendor)) return " AND ".join(conditions) def _build_slow_log_top_digest_azure_rewrite_where_clause( self, start_time: datetime, end_time: datetime, *, ignore_dml: bool = False, ) -> str: date_filters = ", ".join("'%s'" % value for value in self._date_partitions(start_time, end_time)) conditions = [ "date IN (" + date_filters + ")", "time >= " + self._format_timestamp(start_time), "time < " + self._format_timestamp(end_time), "digest IS NOT NULL", "digest != ''", "(is_internal IS NULL OR is_internal = 0)", ] if ignore_dml: normalized = "LOWER(TRIM(COALESCE(query, '')))" conditions.append( "(query IS NULL OR (" f"{normalized} NOT LIKE 'insert%' " f"AND {normalized} NOT LIKE 'replace%' " f"AND {normalized} NOT LIKE 'update%' " f"AND {normalized} NOT LIKE 'delete%'))" ) return " AND ".join(conditions) @staticmethod def _build_top_digest_count_filter(min_count: int, vendor: str = "aws") -> str: if min_count <= 0: return "" return f"HAVING COUNT(*) >= {int(min_count)} " @staticmethod def _build_top_digest_legacy_count_threshold_filter(min_count: int, vendor: str = "aws") -> str: if min_count <= 0: return "" return f"HAVING COUNT(*) > {int(min_count)} " def _build_tikv_hotspot_slow_log_where_clause( self, start_time: datetime, end_time: datetime, *, vendor: str = "aws", ) -> str: date_filters = ", ".join("'%s'" % value for value in self._date_partitions(start_time, end_time)) date_col = self._column_ref("date", vendor) time_col = self._column_ref("time", vendor) digest_col = self._column_ref("digest", vendor) conditions = [ date_col + " IN (" + date_filters + ")", time_col + " >= " + self._format_timestamp(start_time), time_col + " < " + self._format_timestamp(end_time), digest_col + " IS NOT NULL", digest_col + " != ''", self._build_exclude_internal_condition(vendor), ] return " AND ".join(conditions) @classmethod def _build_tikv_addr_match_condition(cls, instance: str, column: str | None = None) -> str: target_column = column or "cop_proc_addr" short = cls._short_instance_name(instance) escaped = cls._escape_sql_string(short.lower()) value_expr = f"LOWER(COALESCE({target_column}, ''))" return ( "(" f"{value_expr} = '{escaped}' " f"OR {value_expr} LIKE '{escaped}.%' " f"OR {value_expr} LIKE '{escaped}:%' " f"OR {value_expr} LIKE '%,{escaped}.%' " f"OR {value_expr} LIKE '%,{escaped}:%' " f"OR {value_expr} LIKE '%, {escaped}.%' " f"OR {value_expr} LIKE '%, {escaped}:%'" ")" ) @staticmethod def _short_instance_name(instance: str) -> str: source = str(instance or "") if ":" in source: source = source.split(":", 1)[0] if "." in source: source = source.split(".", 1)[0] return source @classmethod def _digest_in_values(cls, digests: list[str]) -> str: deduped: list[str] = [] for digest in digests: value = str(digest or "").strip().lower() if value and value not in deduped: deduped.append(value) return ", ".join("'" + cls._escape_sql_string(digest) + "'" for digest in deduped) @staticmethod def _build_table_full_scan_segment_join(vendor: str) -> str: v = vendor.lower() if vendor else "aws" pattern = ServiceDataSource._table_full_scan_segment_pattern(vendor) pattern_literal = ServiceDataSource._sql_regex_literal(pattern, vendor) if v == "gcp": return ( ", UNNEST(REGEXP_EXTRACT_ALL(decoded_plan, " + pattern_literal + ")) AS op_segment" ) if v in ("alicloud", "aliyun"): return ( "LATERAL VIEW EXPLODE(REGEXP_EXTRACT_ALL(decoded_plan, " + pattern_literal + ", 0)) tfs_segments AS op_segment" ) return ( "CROSS JOIN UNNEST(regexp_extract_all(decoded_plan, " + pattern_literal + ")) AS t(op_segment)" ) @staticmethod def _table_full_scan_segment_pattern(vendor: str) -> str: v = vendor.lower() if vendor else "aws" if v == "gcp": return ( r'(?s)(?:"actRows"\s*:\s*"?[0-9]+"?' r'\s*,\s*"' '(?:accessObject|accessObjects|copExecInfo|cost|diskBytes|estRows|memoryBytes|name|operatorInfo|storeType|taskType)"' r'.*?"(?:id|operator|type|name)"\s*:\s*"TableFullScan[^"]*"' r'(?:.*?"(?:storeType|taskType)"\s*:\s*"[^"]*")?' '|' r'"(?:id|operator|type|name)"\s*:\s*"TableFullScan[^"]*".*?' r'"actRows"\s*:\s*"?[0-9]+"?' r'(?:.*?"(?:storeType|taskType)"\s*:\s*"[^"]*")?)' ) return TABLE_FULL_SCAN_SEGMENT_PATTERN @classmethod def _build_table_full_scan_segment_filters(cls, vendor: str, *, ignore_tiflash: bool = False) -> str: if not ignore_tiflash: return "" return " AND " + cls._build_not_regex_condition("op_segment", TIFLASH_SEGMENT_PATTERN, vendor) @staticmethod def _build_not_regex_condition(expression: str, pattern: str, vendor: str) -> str: v = ServiceDataSource._vendor_key(vendor) if pattern == DML_QUERY_PATTERN: dml_predicate = ServiceDataSource._build_dml_query_match_condition(expression, vendor) return f"({expression} IS NULL OR NOT {dml_predicate})" if v == "synapse": if pattern == TIFLASH_SEGMENT_PATTERN: normalized = f"LOWER(COALESCE(CAST({expression} AS NVARCHAR(MAX)), ''))" return ( f"({expression} IS NULL OR (" f"{normalized} NOT LIKE '%\"storetype\"%:%\"tiflash\"%' " f"AND {normalized} NOT LIKE '%\"tasktype\"%:%\"mpp\"%' " f"AND {normalized} NOT LIKE '%tiflash_task%' " f"AND {normalized} NOT LIKE '%tiflash_scan%'))" ) pattern_literal = ServiceDataSource._sql_regex_literal(pattern, vendor) if v == "gcp": predicate = f"REGEXP_CONTAINS({expression}, {pattern_literal})" elif v == "alicloud": predicate = f"{expression} RLIKE {pattern_literal}" else: predicate = f"regexp_like({expression}, {pattern_literal})" return f"({expression} IS NULL OR NOT {predicate})" @staticmethod def _build_dml_query_match_condition(expression: str, vendor: str) -> str: v = ServiceDataSource._vendor_key(vendor) if v == "synapse": stripped = ServiceDataSource._build_synapse_sql_comment_stripped_expr(expression) return ServiceDataSource._build_synapse_keyword_start_condition( stripped, ("insert", "replace", "update", "delete"), ) comment_pattern = ServiceDataSource._sql_regex_literal(DML_LEADING_COMMENT_PATTERN, vendor) start_pattern = ServiceDataSource._sql_regex_literal(DML_START_PATTERN, vendor) if v == "gcp": normalized = f"LOWER(TRIM(REGEXP_REPLACE(COALESCE(CAST({expression} AS STRING), ''), {comment_pattern}, '')))" return f"REGEXP_CONTAINS({normalized}, {start_pattern})" if v == "alicloud": normalized = f"LOWER(TRIM(REGEXP_REPLACE(COALESCE(CAST({expression} AS STRING), ''), {comment_pattern}, '')))" return f"{normalized} RLIKE {start_pattern}" normalized = f"lower(trim(regexp_replace(COALESCE({expression}, ''), {comment_pattern}, '')))" return f"regexp_like({normalized}, {start_pattern})" @staticmethod def _build_synapse_sql_comment_stripped_expr(expression: str) -> str: normalized = f"LOWER(LTRIM(RTRIM(COALESCE(CAST({expression} AS NVARCHAR(4000)), ''))))" stripped = ( "(CASE " f"WHEN {normalized} LIKE '/*%' AND CHARINDEX('*/', {normalized}) > 0 " f"THEN SUBSTRING({normalized}, CHARINDEX('*/', {normalized}) + 2, 4000) " f"WHEN {normalized} LIKE '--%' AND CHARINDEX(CHAR(10), {normalized}) > 0 " f"THEN SUBSTRING({normalized}, CHARINDEX(CHAR(10), {normalized}) + 1, 4000) " f"WHEN {normalized} LIKE '--%' THEN '' " f"WHEN {normalized} LIKE '#%' AND CHARINDEX(CHAR(10), {normalized}) > 0 " f"THEN SUBSTRING({normalized}, CHARINDEX(CHAR(10), {normalized}) + 1, 4000) " f"WHEN {normalized} LIKE '#%' THEN '' " f"ELSE {normalized} END)" ) return f"LOWER(LTRIM(RTRIM({stripped})))" @staticmethod def _build_synapse_keyword_start_condition(expression: str, keywords: tuple[str, ...]) -> str: conditions = [] for keyword in keywords: length = len(keyword) next_char = f"SUBSTRING({expression}, {length + 1}, 1)" boundary = ( f"(LEN({expression}) = {length} " f"OR {next_char} IN (' ', '(') " f"OR UNICODE({next_char}) IN (9, 10, 13))" ) conditions.append(f"(LEFT({expression}, {length}) = '{keyword}' AND {boundary})") return "(" + " OR ".join(conditions) + ")" @staticmethod def _is_tidb_dml_query(sql: str) -> bool: if not sql: return False pos = ServiceDataSource._skip_tidb_sql_space_and_comments(sql, 0) if ServiceDataSource._match_tidb_keyword(sql, pos, ("insert", "replace", "update", "delete")): return True if not ServiceDataSource._match_tidb_keyword(sql, pos, ("with",)): return False outer_pos = ServiceDataSource._find_tidb_cte_outer_statement_start(sql, pos) if outer_pos is None: return False return ServiceDataSource._match_tidb_keyword(sql, outer_pos, ("insert", "replace", "update", "delete")) @staticmethod def _find_tidb_cte_outer_statement_start(sql: str, with_pos: int) -> int | None: pos = with_pos + len("with") pos = ServiceDataSource._skip_tidb_sql_space_and_comments(sql, pos) if ServiceDataSource._match_tidb_keyword(sql, pos, ("recursive",)): pos += len("recursive") while True: pos = ServiceDataSource._skip_tidb_sql_space_and_comments(sql, pos) pos = ServiceDataSource._skip_tidb_identifier(sql, pos) if pos is None: return None pos = ServiceDataSource._skip_tidb_sql_space_and_comments(sql, pos) if pos < len(sql) and sql[pos] == "(": close_pos = ServiceDataSource._find_tidb_matching_paren(sql, pos) if close_pos is None: return None pos = ServiceDataSource._skip_tidb_sql_space_and_comments(sql, close_pos + 1) if not ServiceDataSource._match_tidb_keyword(sql, pos, ("as",)): return None pos += len("as") pos = ServiceDataSource._skip_tidb_sql_space_and_comments(sql, pos) if pos >= len(sql) or sql[pos] != "(": return None close_pos = ServiceDataSource._find_tidb_matching_paren(sql, pos) if close_pos is None: return None pos = ServiceDataSource._skip_tidb_sql_space_and_comments(sql, close_pos + 1) if pos < len(sql) and sql[pos] == ",": pos += 1 continue return ServiceDataSource._skip_tidb_sql_space_and_comments(sql, pos) @staticmethod def _skip_tidb_sql_space_and_comments(sql: str, pos: int) -> int: n = len(sql) while pos < n: while pos < n and sql[pos].isspace(): pos += 1 if sql.startswith("/*", pos): end = sql.find("*/", pos + 2) if end < 0: return n pos = end + 2 continue if sql.startswith("--", pos): end = ServiceDataSource._find_tidb_line_end(sql, pos + 2) pos = end continue if sql.startswith("#", pos): end = ServiceDataSource._find_tidb_line_end(sql, pos + 1) pos = end continue break return pos @staticmethod def _find_tidb_line_end(sql: str, pos: int) -> int: newline_pos = sql.find("\n", pos) if newline_pos < 0: return len(sql) return newline_pos + 1 @staticmethod def _match_tidb_keyword(sql: str, pos: int, keywords: tuple[str, ...]) -> bool: for keyword in keywords: end = pos + len(keyword) if sql[pos:end].lower() != keyword: continue if end >= len(sql) or not ServiceDataSource._is_tidb_identifier_char(sql[end]): return True return False @staticmethod def _skip_tidb_identifier(sql: str, pos: int) -> int | None: if pos >= len(sql): return None if sql[pos] == "`": return ServiceDataSource._skip_tidb_quoted(sql, pos, "`") if not (sql[pos].isalpha() or sql[pos] == "_"): return None pos += 1 while pos < len(sql) and ServiceDataSource._is_tidb_identifier_char(sql[pos]): pos += 1 return pos @staticmethod def _find_tidb_matching_paren(sql: str, open_pos: int) -> int | None: depth = 0 pos = open_pos while pos < len(sql): if sql.startswith("/*", pos): end = sql.find("*/", pos + 2) if end < 0: return None pos = end + 2 continue if sql.startswith("--", pos): pos = ServiceDataSource._find_tidb_line_end(sql, pos + 2) continue if sql[pos] == "#": pos = ServiceDataSource._find_tidb_line_end(sql, pos + 1) continue if sql[pos] in ("'", '"', "`"): pos = ServiceDataSource._skip_tidb_quoted(sql, pos, sql[pos]) if pos is None: return None continue if sql[pos] == "(": depth += 1 elif sql[pos] == ")": depth -= 1 if depth == 0: return pos pos += 1 return None @staticmethod def _skip_tidb_quoted(sql: str, pos: int, quote: str) -> int | None: pos += 1 while pos < len(sql): if sql[pos] == "\\" and quote in ("'", '"'): pos += 2 continue if sql[pos] == quote: if pos + 1 < len(sql) and sql[pos + 1] == quote: pos += 2 continue return pos + 1 pos += 1 return None @staticmethod def _is_tidb_identifier_char(ch: str) -> bool: return ch.isalnum() or ch in ("_", "$") @staticmethod def _sql_regex_literal(pattern: str, vendor: str) -> str: v = ServiceDataSource._vendor_key(vendor) escaped = pattern.replace("'", "''") if v == "gcp": return "r'" + escaped + "'" if v == "alicloud": return "'" + escaped.replace("\\", "\\\\") + "'" return "'" + escaped + "'" @staticmethod def _quote_identifier(identifier: str, vendor: str) -> str: v = ServiceDataSource._vendor_key(vendor) if v == "aws": return '"' + identifier.replace('"', '""') + '"' if v == "synapse": return "[" + identifier.replace("]", "]]") + "]" return "`" + identifier.replace("`", "``") + "`" @staticmethod def _column_ref(identifier: str, vendor: str) -> str: if ServiceDataSource._is_synapse_vendor(vendor): return ServiceDataSource._quote_identifier(identifier, vendor) return identifier @staticmethod def _build_exclude_internal_condition(vendor: str) -> str: if ServiceDataSource._is_synapse_vendor(vendor): return "([is_internal] IS NULL OR [is_internal] = 0)" return "(is_internal IS NULL OR is_internal = false)" @staticmethod def _synapse_text_expr(expression: str, *, length: int = 4000) -> str: return f"COALESCE(CAST({expression} AS NVARCHAR({int(length)})), '')" def _build_synapse_table_full_scan_ctes( self, where_clause: str, base_columns: list[str], vendor: str, ) -> str: select_clause = ", ".join(self._quote_identifier(column, vendor) for column in base_columns) decoded_plan_col = self._quote_identifier("decoded_plan", vendor) base_logs = ( "base_logs AS (" f"SELECT {select_clause} " "FROM clinic_data_proxy.slow_query_logs " f"WHERE {where_clause} AND ISJSON({decoded_plan_col}) = 1" ")" ) plan_nodes = self._build_synapse_plan_nodes_cte(base_columns, vendor) return "WITH " + base_logs + ", " + plan_nodes @staticmethod def _build_synapse_plan_nodes_cte(base_columns: list[str], vendor: str, max_depth: int = 8) -> str: column_refs = ", ".join(ServiceDataSource._quote_identifier(column, vendor) for column in base_columns) decoded_plan_col = ServiceDataSource._quote_identifier("decoded_plan", vendor) parts = [ f"SELECT {column_refs}, CAST({decoded_plan_col} AS NVARCHAR(MAX)) AS node_json FROM base_logs" ] for depth in range(1, max_depth + 1): applies = [] previous = f"CAST({decoded_plan_col} AS NVARCHAR(MAX))" for level in range(1, depth + 1): alias = f"j{level}" applies.append( "CROSS APPLY OPENJSON(" f"CASE WHEN ISJSON({previous}) = 1 " f"AND LEFT(LTRIM({previous}), 1) IN (N'{{', N'[') " f"THEN {previous} ELSE N'{{}}' END" f") AS {alias}" ) previous = f"CAST({alias}.value AS NVARCHAR(MAX))" parts.append( f"SELECT {column_refs}, {previous} AS node_json " "FROM base_logs " + " ".join(applies) + f" WHERE j{depth}.type IN (4, 5)" ) return "plan_nodes AS (" + " UNION ALL ".join(parts) + ")" @staticmethod def _build_synapse_table_full_scan_operator_condition(vendor: str) -> str: checks = [ "JSON_VALUE(node_json, '$.name') LIKE 'TableFullScan%'", "JSON_VALUE(node_json, '$.id') LIKE 'TableFullScan%'", "JSON_VALUE(node_json, '$.operator') LIKE 'TableFullScan%'", "JSON_VALUE(node_json, '$.type') LIKE 'TableFullScan%'", ] return "(" + " OR ".join(checks) + ")" @staticmethod def _build_synapse_tiflash_filter(vendor: str, *, ignore_tiflash: bool = False) -> str: if not ignore_tiflash: return "" return ( " AND NOT (" "LOWER(COALESCE(JSON_VALUE(node_json, '$.storeType'), '')) = 'tiflash' " "OR LOWER(COALESCE(JSON_VALUE(node_json, '$.taskType'), '')) = 'mpp' " "OR LOWER(COALESCE(CAST(node_json AS NVARCHAR(MAX)), '')) LIKE '%tiflash_task%' " "OR LOWER(COALESCE(CAST(node_json AS NVARCHAR(MAX)), '')) LIKE '%tiflash_scan%'" ")" ) @staticmethod def _chunked(values: list[str], size: int) -> Iterator[list[str]]: for index in range(0, len(values), size): yield values[index:index + size] @staticmethod def _build_sample(row: Dict[str, Any]) -> Dict[str, Any]: latency = ServiceDataSource._coerce_float(row.get("query_time")) if not isinstance(latency, (int, float)): latency = 0.0 return { "time": ServiceDataSource._coerce_float(row.get("time")), "conn_id": row.get("conn_id", ""), "db": row.get("db", ""), "act_rows": ServiceDataSource._coerce_int_value(row.get("sample_act_rows")), "latency_s": float(latency), "full_slow_log": { column: value for column, value in row.items() if column not in INTERNAL_SAMPLE_COLUMNS }, } @staticmethod def _coerce_int_value(value: Any) -> int: if value is None: return 0 if isinstance(value, int): return value try: return int(float(str(value))) except (TypeError, ValueError): return 0 def query_slow_logs( self, start_time: datetime, end_time: datetime, ) -> Iterator[Dict[str, Any]]: """Query slow logs from Clinic Data Proxy page by page.""" query_start = self._to_utc(start_time) query_end = self._to_utc(end_time) if query_start >= query_end: return cursor: PageCursor | None = None page_number = 1 while True: sql = self._build_sql(query_start, query_end, cursor) logger.info( "Querying Clinic Data Proxy cluster=%s range=%s~%s page=%d cursor=%s", self.cluster_id, query_start.isoformat(), query_end.isoformat(), page_number, self._format_cursor(cursor), ) columns, rows, trace_id = self._run_query_with_retry( sql, page_number, query_name="slow_logs.page", ) if not rows: logger.info( "Finished Clinic Data Proxy page cluster=%s page=%d rows=0 has_more=false trace_id=%s", self.cluster_id, page_number, trace_id or "-", ) break has_more = len(rows) > self.page_size page_rows = rows[:self.page_size] row_count = 0 last_record: Dict[str, Any] | None = None last_cursor: PageCursor | None = None for raw_row in page_rows: row_count += 1 row = { column: raw_row[index] if index < len(raw_row) else None for index, column in enumerate(columns) } last_cursor = self._cursor_from_row(row) last_record = self._normalize_row(row) yield last_record logger.info( "Finished Clinic Data Proxy page cluster=%s page=%d rows=%d has_more=%s trace_id=%s", self.cluster_id, page_number, row_count, has_more, trace_id or "-", ) if not has_more or last_cursor is None: break cursor = last_cursor page_number += 1 def close(self): """No-op for the HTTP-backed data source.""" return None def _run_query_with_retry( self, sql: str, page_number: int, *, skip_sql_rewrite: bool = False, query_name: str = "data_proxy", ) -> tuple[list[str], list[list[Any]], str]: last_error: DataProxyQueryError | None = None total_attempts = self.retry_attempts + 1 for attempt in range(1, total_attempts + 1): try: return self._run_query(sql, skip_sql_rewrite=skip_sql_rewrite) except DataProxyQueryError as exc: last_error = exc status_code = exc.status_code will_retry = attempt < total_attempts and not exc.is_non_retryable log_level = logging.ERROR if status_code is not None and status_code >= 500 else logging.WARNING logger.log( log_level, "Clinic Data Proxy query failed cluster=%s query=%s page=%d attempt=%d/%d will_retry=%s status_code=%s trace_id=%s error=%s", self.cluster_id, query_name, page_number, attempt, total_attempts, will_retry, status_code if status_code is not None else "-", exc.trace_id or "-", exc.raw_message, ) if not will_retry: break time.sleep(min(2 ** (attempt - 1), 4)) if last_error is None: raise DataProxyQueryError("Clinic Data Proxy request failed: unknown error") raise last_error def _run_query( self, sql: str, *, skip_sql_rewrite: bool = False, ) -> tuple[list[str], list[list[Any]], str]: options: dict[str, Any] = {"skipTableManagement": True} if skip_sql_rewrite: options["skipSQLRewrite"] = True payload = json.dumps( { "sql": sql, "clusterId": self.cluster_id, "timeout": self.timeout_seconds, "options": options, } ).encode("utf-8") headers = {"Content-Type": "application/json"} if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" req = request.Request(self.endpoint, data=payload, headers=headers, method="POST") try: with request.urlopen(req, timeout=self.timeout_seconds + 5) as response: body = response.read() trace_id = self._extract_trace_id(response.headers) except RemoteDisconnected as exc: raise DataProxyQueryError("Clinic Data Proxy request failed: remote end closed connection without response") from exc except ConnectionResetError as exc: raise DataProxyQueryError("Clinic Data Proxy request failed: connection reset by peer") from exc except error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace").strip() raise DataProxyQueryError( f"Clinic Data Proxy query failed status={exc.code}: {detail or exc.reason}", trace_id=self._extract_trace_id(exc.headers), status_code=exc.code, ) from exc except error.URLError as exc: raise DataProxyQueryError(f"Clinic Data Proxy request failed: {exc.reason}") from exc try: payload_obj = json.loads(body) except json.JSONDecodeError as exc: raise DataProxyQueryError("Clinic Data Proxy returned invalid JSON payload", trace_id=trace_id) from exc columns = payload_obj.get("columns") rows = payload_obj.get("rows") if rows is None: rows = [] if not isinstance(columns, list) or not isinstance(rows, list): raise DataProxyQueryError( "Clinic Data Proxy returned invalid payload: missing columns or rows", trace_id=trace_id, ) return [str(column) for column in columns], rows, trace_id def _build_sql( self, start_time: datetime, end_time: datetime, cursor: PageCursor | None, ) -> str: date_filters = ", ".join(f"'{value}'" for value in self._date_partitions(start_time, end_time)) select_clause = ", ".join(REQUIRED_COLUMNS) conditions = [ f"date IN ({date_filters})", f"time >= {self._format_timestamp(start_time)}", f"time < {self._format_timestamp(end_time)}", ] cursor_predicate = self._build_cursor_predicate(cursor) if cursor_predicate: conditions.append(cursor_predicate) return ( f"SELECT {select_clause} " f"FROM clinic_data_proxy.slow_query_logs " f"WHERE {' AND '.join(conditions)} " f"ORDER BY {ORDER_BY_CLAUSE} " f"LIMIT {self.page_size + 1}" ) def _date_partitions(self, start_time: datetime, end_time: datetime) -> list[str]: last_inclusive = end_time - timedelta(microseconds=1) if last_inclusive < start_time: last_inclusive = start_time current = start_time.date() end_date = last_inclusive.date() partitions: list[str] = [] while current <= end_date: partitions.append(self._format_partition(current)) current += timedelta(days=1) return partitions @staticmethod def _format_partition(value: date) -> str: return value.strftime("%Y%m%d") @staticmethod def _format_timestamp(value: datetime) -> str: return format(value.timestamp(), ".6f").rstrip("0").rstrip(".") def _build_cursor_predicate(self, cursor: PageCursor | None) -> str: if cursor is None: return "" cursor_time = cursor.time cursor_conn_id = self._escape_sql_string(cursor.conn_id) cursor_digest = self._escape_sql_string(cursor.digest) cursor_query_time = cursor.query_time cursor_db = self._escape_sql_string(cursor.db) return ( "(" f"{TIME_EXPR} < {cursor_time} " f"OR ({TIME_EXPR} = {cursor_time} AND {CONN_ID_EXPR} < '{cursor_conn_id}') " f"OR ({TIME_EXPR} = {cursor_time} AND {CONN_ID_EXPR} = '{cursor_conn_id}' " f"AND digest < '{cursor_digest}') " f"OR ({TIME_EXPR} = {cursor_time} AND {CONN_ID_EXPR} = '{cursor_conn_id}' " f"AND digest = '{cursor_digest}' AND {QUERY_TIME_EXPR} < {cursor_query_time}) " f"OR ({TIME_EXPR} = {cursor_time} AND {CONN_ID_EXPR} = '{cursor_conn_id}' " f"AND digest = '{cursor_digest}' AND {QUERY_TIME_EXPR} = {cursor_query_time} " f"AND {DB_EXPR} < '{cursor_db}')" ")" ) @staticmethod def _escape_sql_string(value: str) -> str: return str(value).replace("'", "''") @staticmethod def _extract_trace_id(headers: Any) -> str: if headers is None: return "" for key in ("X-Trace-ID", "x-trace-id"): value = headers.get(key) if value: return str(value) return "" @staticmethod def _to_utc(value: datetime) -> datetime: if value.tzinfo is None: return value.replace(tzinfo=UTC) return value.astimezone(UTC) @staticmethod def _resolve_positive_int(candidate: int | None, *, env_name: str, default: int) -> int: if candidate is not None: resolved = int(candidate) else: resolved = int(os.environ.get(env_name, default)) if resolved <= 0: raise ValueError(f"{env_name} must be positive") return resolved @staticmethod def _resolve_non_negative_int(candidate: int | None, *, env_name: str, default: int) -> int: if candidate is not None: resolved = int(candidate) else: resolved = int(os.environ.get(env_name, default)) if resolved < 0: raise ValueError(f"{env_name} must be non-negative") return resolved @staticmethod def _resolve_endpoint(override: str | None) -> str: if override: return override.rstrip("/") explicit = os.environ.get("CLINIC_DATA_PROXY_ENDPOINT") if explicit: return explicit.rstrip("/") base_url = os.environ.get("CLINIC_API_BASE_URL") if base_url: query_path = os.environ.get("CLINIC_DATA_PROXY_QUERY_PATH", DEFAULT_QUERY_PATH) return f"{base_url.rstrip('/')}/{query_path.lstrip('/')}" return DEFAULT_INTERNAL_ENDPOINT @staticmethod def _normalize_row(row: Dict[str, Any]) -> Dict[str, Any]: normalized = dict(row) for key in ("time", "query_time"): normalized[key] = ServiceDataSource._coerce_float(normalized.get(key)) return normalized @staticmethod def _cursor_from_row(row: Dict[str, Any]) -> PageCursor: return PageCursor( time=ServiceDataSource._coerce_decimal_string(row.get("time")), conn_id=str(row.get("conn_id") or ""), digest=str(row.get("digest") or ""), query_time=ServiceDataSource._coerce_decimal_string(row.get("query_time")), db=str(row.get("db") or ""), ) @staticmethod def _format_cursor(cursor: PageCursor | None) -> str: if cursor is None: return "start" return ( f"time={cursor.time}," f"conn_id={cursor.conn_id}," f"digest={cursor.digest}," f"query_time={cursor.query_time}," f"db={cursor.db}" ) @staticmethod def _coerce_float(value: Any) -> Any: if value is None or isinstance(value, (int, float)): return value try: return float(value) except (TypeError, ValueError): return value @staticmethod def _coerce_decimal_string(value: Any) -> str: if value is None: return "0" if isinstance(value, Decimal): decimal_value = value else: text = str(value).strip() if not text: return "0" try: decimal_value = Decimal(text) except (InvalidOperation, ValueError): return "0" normalized = format(decimal_value, "f") if "." in normalized: normalized = normalized.rstrip("0").rstrip(".") return normalized or "0"