#!/usr/bin/env python3 """Analyze TiDB slow log CSV files using DuckDB. Groups slow statements by (digest, plan_digest) and produces plan analysis output. Input CSV must have a decoded `plan` column (run decode_binary_plan.py first). Plan parsing strategy --------------------- tidb_decode_binary_plan() returns a pipe-separated fixed-width table, e.g.: | id | estRows | estCost | actRows | task | access object | execution info | operator info | memory | disk | | TopN_14 | 0.00 | 2629.23 | 1 | root | | time:350ms... | col, offset:? | 3.6 KB | N/A | | └─IndexHashJoin_26 | 0.00 | 2513.23 | 1 | root | | time:350ms... | inner join ... | 103 KB | N/A | Rather than splitting by '|' (fragile if column values contain '|'), we: 1. Parse the header row to find the exact byte-positions of every '|' separator. 2. For every subsequent row, slice at those exact positions. This is robust regardless of what characters appear in column values. Each distinct (digest, plan_digest) has one sample plan; we parse it once and multiply row-level metrics by execute_count from slow_statements. Output directory: // 1.slowlog.csv -- raw rows, plan/binary_plan stripped 2.plans.csv -- plan node rows, one per (digest,plan_digest,line) 3.statements.csv -- aggregated stats by (digest, plan_digest) 4.table-full-scan.csv 5.tidb-tasks.csv 6.tikv-tasks.csv 7.tiflash-tasks.csv 8.table-access-methods.csv 9.index-candidate-from-tablefullscan.csv 10.covering-index-candidate-from-indexlookup.csv Usage: python analyze_slowlog.py --input /app/local/slowlog-xxx.decoded.csv """ import argparse import os import re import sys from typing import Optional import duckdb # --------------------------------------------------------------------------- # Plan text parsing # --------------------------------------------------------------------------- # Column indices (0-based) after fixed-position split _ID = 0 _EST_ROWS = 1 _EST_COST = 2 _ACT_ROWS = 3 _TASK = 4 _ACCESS = 5 _EXEC_INFO = 6 _OP_INFO = 7 _MEMORY = 8 _DISK = 9 def _int_re(pattern: str, text: str) -> Optional[int]: m = re.search(pattern, text) return int(m.group(1)) if m else None def _parse_pipe_positions(header: str) -> list[int]: """Return byte-positions of every '|' in the header row.""" return [i for i, c in enumerate(header) if c == '|'] def _split_by_positions(line: str, pipe_pos: list[int]) -> list[str]: """ Slice `line` at fixed positions determined by the header. Returns one trimmed string per inter-pipe interval. Safe even when column values contain '|'. """ cols = [] n = len(line) for i in range(len(pipe_pos) - 1): s = pipe_pos[i] + 1 e = pipe_pos[i + 1] if i + 1 < len(pipe_pos) and pipe_pos[i + 1] <= n else n cols.append(line[s:e].strip() if s < n else "") return cols def parse_plan_lines( plan_text: str, digest: str, plan_digest: str, execute_count: int, ) -> list[dict]: """ Parse a decoded binary plan into a list of row dicts. One dict per non-header plan node line. """ if not plan_text or not plan_text.strip(): return [] lines = plan_text.split("\n") if not lines: return [] # Line 0 is the header; derive fixed column positions from it header = lines[0] pipe_pos = _parse_pipe_positions(header) if len(pipe_pos) < 2: return [] rows = [] for line_num, line in enumerate(lines[1:], 1): if not line.startswith("|"): continue cols = _split_by_positions(line, pipe_pos) if not cols: continue # raw_id is stripped (used for type extraction) raw_id = cols[_ID] if len(cols) > _ID else "" if not raw_id: continue task = cols[_TASK] if len(cols) > _TASK else "" access_obj = cols[_ACCESS] if len(cols) > _ACCESS else "" exec_info = cols[_EXEC_INFO] if len(cols) > _EXEC_INFO else "" op_info = cols[_OP_INFO] if len(cols) > _OP_INFO else "" memory = cols[_MEMORY] if len(cols) > _MEMORY else "" disk = cols[_DISK] if len(cols) > _DISK else "" # type: strip tree-drawing chars, _ suffixes, remaining whitespace node_type = re.sub( r"\s+", "", re.sub(r"^\W+", "", re.sub(r"_\d+\s*", "", raw_id)) ) # depth: re-extract the RAW (unstripped) id cell, then count all leading # non-word characters (spaces + tree-drawing chars: └─ ├─ │ etc.). # Subtract 1 for the mandatory pipe-table cell padding space so that the # root node starts at depth 0 — mirrors the reference script's approach of # counting leading \W+ characters in the id column. _id_s = pipe_pos[0] + 1 _id_e = pipe_pos[1] if len(pipe_pos) > 1 and pipe_pos[1] <= len(line) else len(line) raw_id_full = line[_id_s:_id_e] if _id_s < len(line) else "" _lw = re.match(r"^\W*", raw_id_full) leading = max(0, len(_lw.group(0)) - 1) if _lw else 0 # planId: dots for depth + operator name with tree chars (lstrip spaces only) plan_id = "." * leading + raw_id_full.lstrip() # actRows act_rows_raw = cols[_ACT_ROWS] if len(cols) > _ACT_ROWS else "" try: act_rows: Optional[int] = int(act_rows_raw) if act_rows_raw else None except ValueError: act_rows = None # tablename / indexname from access object column m = re.search(r"table:(\w+)", access_obj) tablename = m.group(1) if m else "" m = re.search(r"index:([\w(,)]+)", access_obj) indexname = m.group(1) if m else "" rows.append({ "digest": digest, "plan_digest": plan_digest, "line_number": line_num, "type": node_type, "depth": leading, "plan_id": plan_id, "task": task, "est_rows": cols[_EST_ROWS] if len(cols) > _EST_ROWS else "", "est_cost": cols[_EST_COST] if len(cols) > _EST_COST else "", "act_rows": act_rows, "exec_count_x_act_rows": (act_rows * execute_count) if act_rows is not None else None, "execution_info": exec_info, "operator_info": op_info, "access_object": access_obj, "memory": memory, "disk": disk, "tablename": tablename, "indexname": indexname, "loops": _int_re(r"loops:(\d+)", exec_info), "cop_task_num": _int_re(r"cop_task:\s*\{[^}]*num:\s*(\d+)", exec_info), "tikv_task_num": _int_re(r"tikv_task:\{[^}]*tasks:\s*(\d+)", exec_info), "tiflash_task_num": _int_re(r"tiflash_task:\{[^}]*tasks:\s*(\d+)", exec_info), "total_process_keys": _int_re(r"total_process_keys:\s*(\d+)", exec_info), "total_keys": _int_re(r"total_keys:\s*(\d+)", exec_info), "exec_count": execute_count, }) return rows # --------------------------------------------------------------------------- # DuckDB helpers # --------------------------------------------------------------------------- def out(name: str, resultdir: str) -> str: return os.path.join(resultdir, name) def step(con: duckdb.DuckDBPyConnection, label: str, sql: str) -> None: print(f" [{label}] ...", flush=True) con.execute(sql) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--input", required=True, help="Input CSV file path") args = parser.parse_args() csvfile = args.input if not os.path.isfile(csvfile): print(f"ERROR: file not found: {csvfile}", file=sys.stderr) sys.exit(1) filepath = os.path.dirname(os.path.abspath(csvfile)) filename = os.path.splitext(os.path.basename(csvfile))[0] resultdir = os.path.join(filepath, filename) os.makedirs(resultdir, exist_ok=True) print(f"Input: {csvfile}") print(f"Output dir:{resultdir}") print() con = duckdb.connect() # ------------------------------------------------------------------ load step(con, "load", f""" CREATE TABLE rawcsv AS SELECT * FROM read_csv('{csvfile}', sample_size=-1, auto_detect=true, max_line_size=10485760) """) step(con, "slowlogs", """ CREATE TABLE slowlogs AS SELECT ROW_NUMBER() OVER () AS slowlogID, * FROM rawcsv """) row_count = con.execute("SELECT COUNT(*) FROM slowlogs").fetchone()[0] print(f" Loaded {row_count} slow log rows.", flush=True) # ---------------------------------------------------------- 3.statements step(con, "statements", """ CREATE TABLE slow_statements AS SELECT digest, COALESCE(plan_digest, '') AS plan_digest, MIN(query) AS sample_sql, COUNT(*) AS execute_count, SUM(TRY_CAST(query_time AS DOUBLE)) AS query_time_total, AVG(TRY_CAST(query_time AS DOUBLE)) AS query_time_avg, MIN(TRY_CAST(query_time AS DOUBLE)) AS query_time_min, MAX(TRY_CAST(query_time AS DOUBLE)) AS query_time_max, SUM(TRY_CAST(parse_time AS DOUBLE)) AS parse_time_total, AVG(TRY_CAST(parse_time AS DOUBLE)) AS parse_time_avg, MIN(TRY_CAST(parse_time AS DOUBLE)) AS parse_time_min, MAX(TRY_CAST(parse_time AS DOUBLE)) AS parse_time_max, SUM(TRY_CAST(compile_time AS DOUBLE)) AS compile_time_total, AVG(TRY_CAST(compile_time AS DOUBLE)) AS compile_time_avg, MIN(TRY_CAST(compile_time AS DOUBLE)) AS compile_time_min, MAX(TRY_CAST(compile_time AS DOUBLE)) AS compile_time_max, SUM(TRY_CAST(rewrite_time AS DOUBLE)) AS rewrite_time_total, AVG(TRY_CAST(rewrite_time AS DOUBLE)) AS rewrite_time_avg, MIN(TRY_CAST(rewrite_time AS DOUBLE)) AS rewrite_time_min, MAX(TRY_CAST(rewrite_time AS DOUBLE)) AS rewrite_time_max, SUM(TRY_CAST(optimize_time AS DOUBLE)) AS optimize_time_total, AVG(TRY_CAST(optimize_time AS DOUBLE)) AS optimize_time_avg, MIN(TRY_CAST(optimize_time AS DOUBLE)) AS optimize_time_min, MAX(TRY_CAST(optimize_time AS DOUBLE)) AS optimize_time_max, SUM(TRY_CAST(wait_ts AS DOUBLE)) AS wait_ts_total, AVG(TRY_CAST(wait_ts AS DOUBLE)) AS wait_ts_avg, MIN(TRY_CAST(wait_ts AS DOUBLE)) AS wait_ts_min, MAX(TRY_CAST(wait_ts AS DOUBLE)) AS wait_ts_max, SUM(TRY_CAST(cop_time AS DOUBLE)) AS cop_time_total, AVG(TRY_CAST(cop_time AS DOUBLE)) AS cop_time_avg, MIN(TRY_CAST(cop_time AS DOUBLE)) AS cop_time_min, MAX(TRY_CAST(cop_time AS DOUBLE)) AS cop_time_max, SUM(TRY_CAST(process_time AS DOUBLE)) AS process_time_total, AVG(TRY_CAST(process_time AS DOUBLE)) AS process_time_avg, MIN(TRY_CAST(process_time AS DOUBLE)) AS process_time_min, MAX(TRY_CAST(process_time AS DOUBLE)) AS process_time_max, SUM(TRY_CAST(wait_time AS DOUBLE)) AS wait_time_total, AVG(TRY_CAST(wait_time AS DOUBLE)) AS wait_time_avg, MIN(TRY_CAST(wait_time AS DOUBLE)) AS wait_time_min, MAX(TRY_CAST(wait_time AS DOUBLE)) AS wait_time_max, -- kv_total = KV RPC total time (analogous to RpcTime in other projects) SUM(TRY_CAST(kv_total AS DOUBLE)) AS kv_total_total, AVG(TRY_CAST(kv_total AS DOUBLE)) AS kv_total_avg, MIN(TRY_CAST(kv_total AS DOUBLE)) AS kv_total_min, MAX(TRY_CAST(kv_total AS DOUBLE)) AS kv_total_max, SUM(TRY_CAST(mem_max AS DOUBLE)) AS mem_total, AVG(TRY_CAST(mem_max AS DOUBLE)) AS mem_avg, MIN(TRY_CAST(mem_max AS DOUBLE)) AS mem_min, MAX(TRY_CAST(mem_max AS DOUBLE)) AS mem_max, SUM(TRY_CAST(disk_max AS DOUBLE)) AS disk_total, AVG(TRY_CAST(disk_max AS DOUBLE)) AS disk_avg, MIN(TRY_CAST(disk_max AS DOUBLE)) AS disk_min, MAX(TRY_CAST(disk_max AS DOUBLE)) AS disk_max, SUM(TRY_CAST(request_count AS DOUBLE)) AS request_count_total, AVG(TRY_CAST(request_count AS DOUBLE)) AS request_count_avg, MIN(TRY_CAST(request_count AS DOUBLE)) AS request_count_min, MAX(TRY_CAST(request_count AS DOUBLE)) AS request_count_max, SUM(TRY_CAST(backoff_total AS DOUBLE)) AS backoff_total_total, AVG(TRY_CAST(backoff_total AS DOUBLE)) AS backoff_total_avg, MIN(TRY_CAST(backoff_total AS DOUBLE)) AS backoff_total_min, MAX(TRY_CAST(backoff_total AS DOUBLE)) AS backoff_total_max, SUM(TRY_CAST(result_rows AS DOUBLE)) AS result_rows_total, AVG(TRY_CAST(result_rows AS DOUBLE)) AS result_rows_avg, MIN(TRY_CAST(result_rows AS DOUBLE)) AS result_rows_min, MAX(TRY_CAST(result_rows AS DOUBLE)) AS result_rows_max, SUM(TRY_CAST(process_keys AS DOUBLE)) AS process_keys_total, AVG(TRY_CAST(process_keys AS DOUBLE)) AS process_keys_avg, MIN(TRY_CAST(process_keys AS DOUBLE)) AS process_keys_min, MAX(TRY_CAST(process_keys AS DOUBLE)) AS process_keys_max, SUM(TRY_CAST(total_keys AS DOUBLE)) AS total_keys_total, AVG(TRY_CAST(total_keys AS DOUBLE)) AS total_keys_avg, MIN(TRY_CAST(total_keys AS DOUBLE)) AS total_keys_min, MAX(TRY_CAST(total_keys AS DOUBLE)) AS total_keys_max, SUM(TRY_CAST(write_keys AS DOUBLE)) AS write_keys_total, AVG(TRY_CAST(write_keys AS DOUBLE)) AS write_keys_avg, MIN(TRY_CAST(write_keys AS DOUBLE)) AS write_keys_min, MAX(TRY_CAST(write_keys AS DOUBLE)) AS write_keys_max, SUM(TRY_CAST(write_size AS DOUBLE)) AS write_size_total, AVG(TRY_CAST(write_size AS DOUBLE)) AS write_size_avg, MIN(TRY_CAST(write_size AS DOUBLE)) AS write_size_min, MAX(TRY_CAST(write_size AS DOUBLE)) AS write_size_max, SUM(TRY_CAST(request_unit_read AS DOUBLE)) AS ru_read_total, AVG(TRY_CAST(request_unit_read AS DOUBLE)) AS ru_read_avg, SUM(TRY_CAST(request_unit_write AS DOUBLE)) AS ru_write_total, AVG(TRY_CAST(request_unit_write AS DOUBLE)) AS ru_write_avg, SUM(TRY_CAST(exec_retry_count AS DOUBLE)) AS exec_retry_total, SUM(CASE WHEN TRY_CAST(succ AS BOOLEAN) THEN 1 ELSE 0 END) AS succ_total, SUM(CASE WHEN NOT TRY_CAST(succ AS BOOLEAN) THEN 1 ELSE 0 END) AS failed_total FROM slowlogs GROUP BY digest, COALESCE(plan_digest, '') ORDER BY query_time_total DESC """) con.execute(f"COPY slow_statements TO '{out('3.statements.csv', resultdir)}'") stmt_count = con.execute("SELECT COUNT(*) FROM slow_statements").fetchone()[0] print(f" -> {stmt_count} distinct (digest, plan_digest) groups", flush=True) # --------------------------------------------------------- plan parsing # DuckDB's read_csv truncates multi-line quoted fields at the first line, # so plan text fetched via DuckDB would only contain the header row. # Instead, read the CSV directly with Python's csv module which handles # multi-line quoted fields correctly, collecting one sample plan per # (digest, plan_digest) group. Then multiply metrics by execute_count. print(" [plan parsing] reading unique plans from CSV ...", flush=True) import csv as _csv import sys as _sys _csv.field_size_limit(_sys.maxsize) # (digest, plan_digest) -> first valid plan text found plan_by_group: dict[tuple[str, str], str] = {} with open(csvfile, newline="", encoding="utf-8") as _f: for _row in _csv.DictReader(_f): _digest = (_row.get("digest") or "").strip() _pd = (_row.get("plan_digest") or "").strip() _plan = (_row.get("plan") or "").strip() _key = (_digest, _pd) if _key not in plan_by_group and _plan and _plan not in ("", "default"): plan_by_group[_key] = _plan # execute_count per (digest, plan_digest) from DuckDB aggregation exec_count_map: dict[tuple[str, str], int] = { (r[0], r[1]): r[2] for r in con.execute( "SELECT digest, plan_digest, execute_count FROM slow_statements" ).fetchall() } all_plan_rows: list[dict] = [] skipped = 0 for (digest, plan_digest), plan_text in plan_by_group.items(): execute_count = exec_count_map.get((digest, plan_digest), 1) rows = parse_plan_lines(plan_text, digest, plan_digest, execute_count) if rows: all_plan_rows.extend(rows) else: skipped += 1 n_groups = len(plan_by_group) print(f" [plan parsing] {len(all_plan_rows)} plan node rows " f"from {n_groups - skipped}/{n_groups} plans", flush=True) if not all_plan_rows: print(" WARNING: no plan rows parsed; skipping plan analysis.", flush=True) _write_slowlog_no_plan(con, resultdir) con.close() _print_results(resultdir) return # Insert plan rows into DuckDB via pandas DataFrame import pandas as _pd _plan_df = _pd.DataFrame(all_plan_rows) con.register("_plan_rows_df", _plan_df) step(con, "plancsv", "CREATE TABLE plancsv AS SELECT * FROM _plan_rows_df") con.execute(f"COPY plancsv TO '{out('2.plans.csv', resultdir)}'") # ------------------------------------------- 4.table-full-scan step(con, "table-full-scan", f""" COPY ( SELECT tablename, avg(act_rows) AS avg_act_rows, sum(exec_count_x_act_rows) AS sum_act_rows_x_exec FROM plancsv WHERE type = 'TableFullScan' AND tablename IS NOT NULL AND tablename <> '' GROUP BY tablename ORDER BY sum_act_rows_x_exec DESC ) TO '{out('4.table-full-scan.csv', resultdir)}' """) # ------------------------------------------- 5.tidb-tasks step(con, "tidb-tasks", f""" COPY ( SELECT type, sum(exec_count) AS exec_sum, sum(exec_count_x_act_rows) AS exec_x_act_rows_sum FROM plancsv WHERE task = 'root' GROUP BY type ORDER BY exec_x_act_rows_sum DESC ) TO '{out('5.tidb-tasks.csv', resultdir)}' """) # ------------------------------------------- 6.tikv-tasks step(con, "tikv-tasks", f""" COPY ( SELECT type, sum(exec_count) AS exec_sum, sum(exec_count_x_act_rows) AS exec_x_act_rows_sum, sum(total_process_keys * exec_count) AS exec_x_process_keys_sum, sum(total_keys * exec_count) AS exec_x_total_keys_sum, sum(cop_task_num * exec_count) AS exec_x_cop_task_sum, sum(tikv_task_num * exec_count) AS exec_x_tikv_task_sum FROM plancsv WHERE task IN ('cop[tikv]', 'batchCop[tikv]') OR type LIKE 'Point_Get%' OR type LIKE 'Batch_Point_Get%' GROUP BY type ORDER BY exec_x_act_rows_sum DESC ) TO '{out('6.tikv-tasks.csv', resultdir)}' """) # ------------------------------------------- 7.tiflash-tasks step(con, "tiflash-tasks", f""" COPY ( SELECT type, sum(exec_count) AS exec_sum, sum(exec_count_x_act_rows) AS exec_x_act_rows_sum, sum(total_process_keys * exec_count) AS exec_x_process_keys_sum, sum(total_keys * exec_count) AS exec_x_total_keys_sum, sum(cop_task_num * exec_count) AS exec_x_cop_task_sum, sum(tiflash_task_num * exec_count) AS exec_x_tiflash_task_sum FROM plancsv WHERE task IN ('cop[tiflash]', 'mpp[tiflash]', 'batchCop[tiflash]') GROUP BY type ORDER BY exec_x_act_rows_sum DESC ) TO '{out('7.tiflash-tasks.csv', resultdir)}' """) # ------------------------------------------- 8.table-access-methods step(con, "table-access-methods", f""" COPY ( SELECT tablename, SUM(CASE WHEN type IN ( 'Point_Get', 'TableRangeScan', 'Batch_Point_Get', 'Point_Get(Build)', 'Batch_Point_Get(Build)', 'TableRangeScan(Build)', 'Point_Get(Probe)', 'Batch_Point_Get(Probe)', 'TableRangeScan(Probe)' ) THEN exec_count ELSE 0 END) AS via_primary_key, SUM(CASE WHEN type IN ('TableRowIDScan', 'TableRowIDScan(Probe)') THEN exec_count ELSE 0 END) AS via_index_lookup, SUM(CASE WHEN type = 'TableFullScan' THEN exec_count ELSE 0 END) AS via_full_scan FROM plancsv WHERE tablename IS NOT NULL AND tablename <> '' GROUP BY tablename ORDER BY via_primary_key, via_index_lookup DESC ) TO '{out('8.table-access-methods.csv', resultdir)}' """) # ------------------------------------------- 9.index-candidate-from-tablefullscan # Selection nodes whose immediate next plan line is a TableFullScan. # Adjacency is within the same (digest, plan_digest) plan group. step(con, "index-candidate-from-tablefullscan", """ CREATE TABLE tfscan_selections AS SELECT ( SELECT tablename FROM plancsv AS inner_p WHERE inner_p.digest = sel.digest AND inner_p.plan_digest = sel.plan_digest AND inner_p.type LIKE 'TableFullScan%' AND inner_p.line_number = sel.line_number + 1 ) AS stats_tablename, sel.operator_info, sum(sel.exec_count) AS cnt, any_value(sel.digest) AS sample_digest, sum(sel.exec_count_x_act_rows) AS sum_act_rows_x_exec FROM plancsv AS sel WHERE sel.type LIKE 'Selection%' AND EXISTS ( SELECT 1 FROM plancsv AS inner_p WHERE inner_p.digest = sel.digest AND inner_p.plan_digest = sel.plan_digest AND inner_p.type LIKE 'TableFullScan%' AND inner_p.line_number = sel.line_number + 1 ) GROUP BY stats_tablename, sel.operator_info ORDER BY stats_tablename, sel.operator_info, cnt DESC """) con.execute(f""" COPY ( SELECT stats_tablename, operator_info_pattern, sum(cnt) AS sum, round(sum(sum_act_rows_x_exec) / sum(cnt), 0) AS avg_act_rows_x_exec, ( SELECT max(act_rows) FROM plancsv WHERE type LIKE 'TableFullScan%' AND tablename = t.stats_tablename ) AS sample_table_rows, any_value(sample_digest) AS sample_digest FROM ( SELECT *, regexp_replace( regexp_replace(operator_info, '\\d+', '0', 'g'), '"[^"]*"', '%s', 'g' ) AS operator_info_pattern FROM tfscan_selections ) AS t GROUP BY stats_tablename, operator_info_pattern ORDER BY stats_tablename, operator_info_pattern, sum DESC ) TO '{out('9.index-candidate-from-tablefullscan.csv', resultdir)}' """) # ------------------------------------------- 10.covering-index-candidate step(con, "covering-index-candidate-from-indexlookup", """ CREATE TABLE indexlookup_selections AS SELECT ( SELECT tablename FROM plancsv AS inner_p WHERE inner_p.digest = sel.digest AND inner_p.plan_digest = sel.plan_digest AND inner_p.type LIKE 'TableRowIDScan%' AND inner_p.line_number = sel.line_number + 1 ) AS stats_tablename, ( SELECT indexname FROM plancsv AS inner_p WHERE inner_p.digest = sel.digest AND inner_p.plan_digest = sel.plan_digest AND inner_p.type LIKE 'IndexRangeScan%' AND inner_p.line_number = sel.line_number - 1 ) AS stats_indexname, sel.operator_info, sum(sel.exec_count) AS cnt, any_value(sel.digest) AS sample_digest, sum(sel.exec_count_x_act_rows) AS sum_act_rows_x_exec FROM plancsv AS sel WHERE sel.type LIKE 'Selection%' AND EXISTS ( SELECT 1 FROM plancsv AS inner_p WHERE inner_p.digest = sel.digest AND inner_p.plan_digest = sel.plan_digest AND inner_p.type LIKE 'TableRowIDScan%' AND inner_p.line_number = sel.line_number + 1 ) AND EXISTS ( SELECT 1 FROM plancsv AS inner_p WHERE inner_p.digest = sel.digest AND inner_p.plan_digest = sel.plan_digest AND inner_p.type LIKE 'IndexRangeScan%' AND inner_p.line_number = sel.line_number - 1 ) GROUP BY stats_tablename, stats_indexname, sel.operator_info ORDER BY stats_tablename, stats_indexname, sel.operator_info, cnt DESC """) con.execute(f""" COPY ( SELECT stats_tablename, stats_indexname, operator_info_pattern, sum(cnt) AS sum, round(sum(sum_act_rows_x_exec) / sum(cnt), 0) AS avg_act_rows_x_exec, any_value(sample_digest) AS sample_digest FROM ( SELECT *, regexp_replace( regexp_replace(operator_info, '\\d+', '0', 'g'), '"[^"]*"', '%s', 'g' ) AS operator_info_pattern FROM indexlookup_selections ) AS t GROUP BY stats_tablename, stats_indexname, operator_info_pattern ORDER BY stats_tablename, stats_indexname, operator_info_pattern, sum DESC ) TO '{out('10.covering-index-candidate-from-indexlookup.csv', resultdir)}' """) # ------------------------------------------- 1.slowlog (plan stripped) _write_slowlog_no_plan(con, resultdir) con.close() _print_results(resultdir) def _write_slowlog_no_plan(con: duckdb.DuckDBPyConnection, resultdir: str) -> None: step(con, "slowlog-no-plan", f""" COPY ( SELECT * EXCLUDE (plan, binary_plan) FROM slowlogs ) TO '{out('1.slowlog.csv', resultdir)}' """) def _print_results(resultdir: str) -> None: print() print(f"Done. Results in: {resultdir}/") for i in range(1, 11): prefix = f"{i}." matches = sorted(f for f in os.listdir(resultdir) if f.startswith(prefix)) for f in matches: print(f" {f}") if __name__ == "__main__": main()