#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ DB2 for z/OS SQL / EXPLAIN Analyzer V2 - Hyper Pro =================================================== Standalone analyzer for DB2 for z/OS access path reviews, SQL quality checks, PLAN_TABLE-like exports, predicate analysis, catalog/index/statistics audits, package bind option review, baseline regression detection and index candidate recommendations. The tool intentionally works offline on CSV / JSON / JSONL exports so it can be used without live mainframe connectivity. Typical inputs: - PLAN_TABLE-like explain export CSV/JSONL - SQL statements file - DSN_PREDICAT_TABLE-like export - index catalog export - table statistics export - package/bind options export - baseline explain export for regression comparison - custom policy JSON No third-party dependencies are required. """ from __future__ import annotations import argparse import csv import hashlib import html import json import math import os import re import sys from collections import Counter, defaultdict from dataclasses import dataclass, asdict, field from datetime import datetime, date from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple SEVERITY_ORDER = {"INFO": 0, "WARNING": 1, "ERROR": 2, "CRITICAL": 3} EXIT_ORDER = {"NONE": 99, "INFO": 0, "WARNING": 1, "ERROR": 2, "CRITICAL": 3} DEFAULT_POLICY: Dict[str, Any] = { "high_cost": 2000.0, "critical_cost": 8000.0, "high_getpages": 250000.0, "critical_getpages": 1000000.0, "high_rows": 100000.0, "critical_rows": 1000000.0, "high_elapsed_ms": 3000.0, "critical_elapsed_ms": 15000.0, "high_cpu_ms": 1000.0, "critical_cpu_ms": 7000.0, "high_wait_ms": 1500.0, "critical_wait_ms": 10000.0, "critical_table_scan_rows": 500000.0, "warning_table_scan_rows": 50000.0, "regression_pct": 25.0, "critical_regression_pct": 100.0, "stale_stats_days": 30, "critical_stale_stats_days": 90, "min_clustering_ratio": 80.0, "critical_clustering_ratio": 50.0, "max_nlevels": 4, "api_high_elapsed_ms": 800.0, "api_critical_elapsed_ms": 2500.0, "sensitive_schemas": ["SYS", "SYSIBM", "SYSIBMADM", "PROD", "PRODDB", "PAYROLL", "FINANCE", "RACF", "SECURITY"], "critical_tables": [ "PRODDB.CUSTOMER", "PRODDB.ACCOUNT", "PRODDB.PAYROLL", "PRODDB.PAYMENT", "PRODDB.TRANSACTION", "PRODDB.AUDIT_LOG", ], "sensitive_columns": ["SSN", "SOCIAL", "CARD", "PAN", "ACCOUNT", "BALANCE", "SALARY", "DOB", "EMAIL", "PHONE", "IBAN"], "banned_sql_patterns": [ r"^\s*DELETE\s+FROM\s+\S+\s*;?\s*$", r"^\s*UPDATE\s+\S+\s+SET\s+.+;?\s*$", r"\bWITH\s+UR\b", ], "profiles": { "production": {"high_cost": 1500.0, "critical_cost": 6000.0, "regression_pct": 20.0}, "performance": {"high_cost": 1000.0, "critical_cost": 4000.0, "high_getpages": 150000.0}, "db2": {}, "batch": {"high_elapsed_ms": 30000.0, "critical_elapsed_ms": 120000.0, "high_cost": 6000.0}, "api": {"high_elapsed_ms": 800.0, "critical_elapsed_ms": 2500.0, "high_cost": 750.0, "critical_cost": 3000.0}, "packages": {"regression_pct": 15.0}, "capacity": {"high_getpages": 500000.0, "critical_getpages": 2500000.0}, "training": {"high_cost": 5000.0, "critical_cost": 20000.0, "regression_pct": 75.0}, "strict": {"high_cost": 500.0, "critical_cost": 2500.0, "regression_pct": 10.0, "stale_stats_days": 14}, }, } FIELD_ALIASES: Dict[str, List[str]] = { "stmt_id": ["stmt_id", "stmtid", "statement_id", "queryno", "query_no", "query_number", "statement_no"], "queryno": ["queryno", "query_no", "query_number"], "plan_no": ["plan_no", "planno", "plan", "planno"], "qblock_no": ["qblock_no", "qblockno", "qblock"], "collection": ["collection", "collid", "collection_id", "coll_id"], "package": ["package", "package_name", "pkgname", "pkg_name", "progname"], "program": ["program", "program_name", "progname", "applname"], "statement_type": ["statement_type", "stmt_type", "type", "operation"], "sql_text": ["sql_text", "statement_text", "statement", "sql", "stmt_text"], "creator": ["creator", "tabschema", "table_schema", "schema", "owner", "table_owner", "tbcreator"], "table_name": ["table_name", "tablename", "name", "table", "tname", "tbname"], "correlation": ["correlation", "correlation_name", "corr", "alias", "tabno"], "access_type": ["access_type", "accesstype", "access", "accessname", "access_type_code"], "access_name": ["access_name", "accessname", "index_name", "index", "ixname"], "matchcols": ["matchcols", "match_cols", "matching_columns", "matched_columns"], "method": ["method", "join_method", "joinmethod"], "prefetch": ["prefetch", "prefetch_type", "prefetch_type_code"], "sortn_unique": ["sortn_unique", "sortnunique", "sortn_unique_sw"], "sortn_join": ["sortn_join", "sortnjoin"], "sortn_orderby": ["sortn_orderby", "sortnorder", "sort_orderby", "sortc_orderby"], "sortn_groupby": ["sortn_groupby", "sortngroup", "sort_groupby", "sortc_groupby"], "sortn_union": ["sortn_union", "sortnunion"], "sortc_unique": ["sortc_unique", "sortcunique"], "sortc_join": ["sortc_join", "sortcjoin"], "sortc_orderby": ["sortc_orderby", "sortcorder", "sortcorder"], "sortc_groupby": ["sortc_groupby", "sortcgroup"], "total_cost": ["total_cost", "cost", "procms", "estimated_cost", "estimated_total_cost"], "cpu_cost": ["cpu_cost", "cpu_ms", "estimated_cpu_ms", "cpu"], "io_cost": ["io_cost", "io_ms", "estimated_io_ms", "io"], "elapsed_ms": ["elapsed_ms", "elapsed", "elapsed_time_ms", "estimated_elapsed_ms"], "wait_ms": ["wait_ms", "lock_wait_ms", "io_wait_ms", "wait"], "getpages": ["getpages", "get_pages", "getpage", "buffer_getpages"], "rows_examined": ["rows_examined", "rows", "cardinality", "estimated_rows", "rows_read"], "rows_returned": ["rows_returned", "output_rows", "returned_rows"], "degree": ["degree", "parallelism", "parallel_degree"], "parallelism": ["parallelism", "parallel", "degree"], "stage2": ["stage2", "stage_2", "stage2_predicates"], "rid_pool": ["rid_pool", "ridpool", "rid_pool_used", "rid_overflow"], "workfile": ["workfile", "workfile_pages", "workfile_usage"], "lock_mode": ["lock_mode", "lockmode", "isolation", "isolation_level"], } PRED_ALIASES = { "stmt_id": FIELD_ALIASES["stmt_id"], "creator": FIELD_ALIASES["creator"], "table_name": FIELD_ALIASES["table_name"], "column_name": ["column_name", "column", "colname", "col_name"], "predicate_text": ["predicate_text", "predicate", "pred_text", "text"], "predicate_type": ["predicate_type", "type", "operator", "op"], "stage": ["stage", "predicate_stage", "stage_no"], "indexable": ["indexable", "is_indexable", "matching", "is_matching"], "filter_factor": ["filter_factor", "filter", "ff", "selectivity"], } INDEX_ALIASES = { "creator": FIELD_ALIASES["creator"], "table_name": FIELD_ALIASES["table_name"], "index_schema": ["index_schema", "ixcreator", "index_owner", "indexcreator"], "index_name": ["index_name", "ixname", "index", "name"], "columns": ["columns", "column_list", "cols", "key_columns"], "unique": ["unique", "is_unique", "uniquerule"], "clustering": ["clustering", "clustered", "is_clustering"], "clustering_ratio": ["clustering_ratio", "cluster_ratio", "clusterratio", "cluster_ratio_pct"], "nlevels": ["nlevels", "n_levels", "levels"], "firstkeycard": ["firstkeycard", "first_key_card", "firstkey_cardinality"], "fullkeycard": ["fullkeycard", "full_key_card", "fullkey_cardinality"], "lastused": ["lastused", "last_used", "last_used_date"], } STATS_ALIASES = { "creator": FIELD_ALIASES["creator"], "table_name": FIELD_ALIASES["table_name"], "cardinality": ["cardinality", "card", "rows", "rowcount", "row_count"], "npages": ["npages", "pages", "n_pages", "total_pages"], "stats_date": ["stats_date", "runstats_date", "statstime", "stats_time", "last_stats"], "partitions": ["partitions", "nparts", "partition_count"], "volatile": ["volatile", "is_volatile"], "compressed_pct": ["compressed_pct", "compression_pct", "pct_pages_compressed"], "organization": ["organization", "type", "table_type"], } PACKAGE_ALIASES = { "collection": FIELD_ALIASES["collection"], "package": FIELD_ALIASES["package"], "program": FIELD_ALIASES["program"], "bind_time": ["bind_time", "bindtime", "last_bind", "bind_date"], "isolation": ["isolation", "isolation_level"], "release": ["release", "release_option"], "reopt": ["reopt", "reoptimize", "reoptimization"], "degree": ["degree", "parallelism"], "currentdata": ["currentdata", "current_data"], "validate": ["validate", "validate_option"], "dynamicrules": ["dynamicrules", "dynamic_rules"], "qualifier": ["qualifier", "default_schema"], } SQL_KEYWORDS = { "SELECT", "FROM", "WHERE", "AND", "OR", "NOT", "NULL", "IS", "LIKE", "IN", "BETWEEN", "EXISTS", "JOIN", "LEFT", "RIGHT", "INNER", "OUTER", "ON", "AS", "GROUP", "ORDER", "BY", "HAVING", "FETCH", "FIRST", "ROWS", "ONLY", "WITH", "UR", "CS", "RS", "RR", "UPDATE", "DELETE", "INSERT", "VALUES", "SET", "DISTINCT", "UNION", "ALL", "CASE", "WHEN", "THEN", "ELSE", "END" } @dataclass class PlanRow: stmt_id: str = "" queryno: str = "" plan_no: str = "" qblock_no: str = "" collection: str = "" package: str = "" program: str = "" statement_type: str = "" sql_text: str = "" creator: str = "" table_name: str = "" correlation: str = "" access_type: str = "" access_name: str = "" matchcols: int = 0 method: str = "" prefetch: str = "" sort_flags: Dict[str, bool] = field(default_factory=dict) total_cost: float = 0.0 cpu_cost: float = 0.0 io_cost: float = 0.0 elapsed_ms: float = 0.0 wait_ms: float = 0.0 getpages: float = 0.0 rows_examined: float = 0.0 rows_returned: float = 0.0 degree: str = "" parallelism: str = "" stage2: str = "" rid_pool: str = "" workfile: str = "" lock_mode: str = "" source_line: int = 0 raw: Dict[str, Any] = field(default_factory=dict) @property def full_table(self) -> str: return normalize_table(self.creator, self.table_name) @property def key(self) -> str: return self.stmt_id or self.queryno or f"{self.collection}.{self.package}.{self.plan_no}.{self.qblock_no}" @dataclass class PredicateRow: stmt_id: str = "" creator: str = "" table_name: str = "" column_name: str = "" predicate_text: str = "" predicate_type: str = "" stage: str = "" indexable: str = "" filter_factor: float = 0.0 raw: Dict[str, Any] = field(default_factory=dict) @property def full_table(self) -> str: return normalize_table(self.creator, self.table_name) @dataclass class IndexRow: creator: str = "" table_name: str = "" index_schema: str = "" index_name: str = "" columns: List[str] = field(default_factory=list) unique: str = "" clustering: str = "" clustering_ratio: float = 0.0 nlevels: int = 0 firstkeycard: float = 0.0 fullkeycard: float = 0.0 lastused: str = "" raw: Dict[str, Any] = field(default_factory=dict) @property def full_table(self) -> str: return normalize_table(self.creator, self.table_name) @property def full_index(self) -> str: return normalize_table(self.index_schema, self.index_name) @dataclass class TableStats: creator: str = "" table_name: str = "" cardinality: float = 0.0 npages: float = 0.0 stats_date: str = "" partitions: int = 0 volatile: str = "" compressed_pct: float = 0.0 organization: str = "" raw: Dict[str, Any] = field(default_factory=dict) @property def full_table(self) -> str: return normalize_table(self.creator, self.table_name) @dataclass class PackageRow: collection: str = "" package: str = "" program: str = "" bind_time: str = "" isolation: str = "" release: str = "" reopt: str = "" degree: str = "" currentdata: str = "" validate: str = "" dynamicrules: str = "" qualifier: str = "" raw: Dict[str, Any] = field(default_factory=dict) @property def full_package(self) -> str: return ".".join(x for x in [self.collection, self.package] if x) @dataclass class SQLStatement: stmt_id: str sql_text: str statement_type: str = "" tables: List[str] = field(default_factory=list) columns: List[str] = field(default_factory=list) predicates: List[str] = field(default_factory=list) hash: str = "" @dataclass class Finding: severity: str code: str category: str title: str detail: str recommendation: str stmt_id: str = "" table: str = "" package: str = "" evidence: str = "" score: int = 0 @dataclass class IndexCandidate: stmt_id: str table: str candidate_name: str columns: List[str] reason: str estimated_benefit: str evidence: str risk: str = "MEDIUM" @dataclass class BaselineCompare: stmt_id: str table: str package: str metric: str baseline: str current: str delta_pct: float severity: str detail: str @dataclass class CategoryScore: category: str score: int findings: int critical: int error: int warning: int @dataclass class AnalysisResult: status: str highest_severity: str risk_score: int profile: str generated_at: str summary: Dict[str, Any] plan_rows: List[PlanRow] sql_statements: List[SQLStatement] predicates: List[PredicateRow] indexes: List[IndexRow] table_stats: List[TableStats] packages: List[PackageRow] findings: List[Finding] index_candidates: List[IndexCandidate] baseline_compare: List[BaselineCompare] categories: List[CategoryScore] def normalize_name(value: Any) -> str: if value is None: return "" return str(value).strip().strip('"').upper() def normalize_table(schema: Any, table: Any) -> str: s = normalize_name(schema) t = normalize_name(table) if s and t: return f"{s}.{t}" return t or s def as_float(value: Any, default: float = 0.0) -> float: if value is None: return default try: s = str(value).strip().replace(",", "") if not s or s.upper() in {"N", "NO", "NULL", "NONE", "-"}: return default return float(s) except Exception: return default def as_int(value: Any, default: int = 0) -> int: try: return int(float(str(value).strip())) except Exception: return default def as_bool(value: Any) -> bool: return str(value).strip().upper() in {"Y", "YES", "TRUE", "1", "T", "REQUIRED", "USED"} def load_policy(path: Optional[str], profile: str) -> Dict[str, Any]: policy = json.loads(json.dumps(DEFAULT_POLICY)) if path: with open(path, "r", encoding="utf-8") as fh: custom = json.load(fh) deep_update(policy, custom) profile_settings = policy.get("profiles", {}).get(profile, {}) deep_update(policy, profile_settings) return policy def deep_update(base: Dict[str, Any], extra: Dict[str, Any]) -> None: for key, value in extra.items(): if isinstance(value, dict) and isinstance(base.get(key), dict): deep_update(base[key], value) else: base[key] = value def read_records(path: Optional[str]) -> List[Dict[str, Any]]: if not path: return [] p = Path(path) if not p.exists(): raise FileNotFoundError(path) suffix = p.suffix.lower() if suffix in {".jsonl", ".ndjson"}: out = [] with open(p, "r", encoding="utf-8", errors="ignore") as fh: for line in fh: line = line.strip() if line: out.append(json.loads(line)) return out if suffix == ".json": with open(p, "r", encoding="utf-8", errors="ignore") as fh: data = json.load(fh) if isinstance(data, list): return data if isinstance(data, dict): for key in ["rows", "data", "records", "plan_rows", "explain"]: if isinstance(data.get(key), list): return data[key] return [data] with open(p, "r", encoding="utf-8-sig", errors="ignore", newline="") as fh: sample = fh.read(4096) fh.seek(0) dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|") if sample.strip() else csv.excel reader = csv.DictReader(fh, dialect=dialect) return [dict(row) for row in reader] def get(row: Dict[str, Any], aliases: List[str], default: str = "") -> str: if not row: return default normalized = {str(k).strip().lower().replace(" ", "_"): v for k, v in row.items()} for alias in aliases: key = alias.lower().replace(" ", "_") if key in normalized and normalized[key] is not None: return str(normalized[key]).strip() return default def parse_sort_flags(row: Dict[str, Any]) -> Dict[str, bool]: names = [ "sortn_unique", "sortn_join", "sortn_orderby", "sortn_groupby", "sortn_union", "sortc_unique", "sortc_join", "sortc_orderby", "sortc_groupby", ] return {name: as_bool(get(row, FIELD_ALIASES.get(name, [name]))) for name in names} def parse_plan_rows(path: Optional[str]) -> List[PlanRow]: rows = read_records(path) out: List[PlanRow] = [] for idx, row in enumerate(rows, 1): pr = PlanRow( stmt_id=get(row, FIELD_ALIASES["stmt_id"]), queryno=get(row, FIELD_ALIASES["queryno"]), plan_no=get(row, FIELD_ALIASES["plan_no"]), qblock_no=get(row, FIELD_ALIASES["qblock_no"]), collection=normalize_name(get(row, FIELD_ALIASES["collection"])), package=normalize_name(get(row, FIELD_ALIASES["package"])), program=normalize_name(get(row, FIELD_ALIASES["program"])), statement_type=normalize_name(get(row, FIELD_ALIASES["statement_type"])), sql_text=get(row, FIELD_ALIASES["sql_text"]), creator=normalize_name(get(row, FIELD_ALIASES["creator"])), table_name=normalize_name(get(row, FIELD_ALIASES["table_name"])), correlation=normalize_name(get(row, FIELD_ALIASES["correlation"])), access_type=normalize_name(get(row, FIELD_ALIASES["access_type"])), access_name=normalize_name(get(row, FIELD_ALIASES["access_name"])), matchcols=as_int(get(row, FIELD_ALIASES["matchcols"])), method=normalize_name(get(row, FIELD_ALIASES["method"])), prefetch=normalize_name(get(row, FIELD_ALIASES["prefetch"])), sort_flags=parse_sort_flags(row), total_cost=as_float(get(row, FIELD_ALIASES["total_cost"])), cpu_cost=as_float(get(row, FIELD_ALIASES["cpu_cost"])), io_cost=as_float(get(row, FIELD_ALIASES["io_cost"])), elapsed_ms=as_float(get(row, FIELD_ALIASES["elapsed_ms"])), wait_ms=as_float(get(row, FIELD_ALIASES["wait_ms"])), getpages=as_float(get(row, FIELD_ALIASES["getpages"])), rows_examined=as_float(get(row, FIELD_ALIASES["rows_examined"])), rows_returned=as_float(get(row, FIELD_ALIASES["rows_returned"])), degree=get(row, FIELD_ALIASES["degree"]), parallelism=get(row, FIELD_ALIASES["parallelism"]), stage2=get(row, FIELD_ALIASES["stage2"]), rid_pool=get(row, FIELD_ALIASES["rid_pool"]), workfile=get(row, FIELD_ALIASES["workfile"]), lock_mode=normalize_name(get(row, FIELD_ALIASES["lock_mode"])), source_line=idx, raw=row, ) if not pr.stmt_id: pr.stmt_id = str(pr.queryno or idx) out.append(pr) return out def parse_predicates(path: Optional[str]) -> List[PredicateRow]: out: List[PredicateRow] = [] for row in read_records(path): out.append(PredicateRow( stmt_id=get(row, PRED_ALIASES["stmt_id"]), creator=normalize_name(get(row, PRED_ALIASES["creator"])), table_name=normalize_name(get(row, PRED_ALIASES["table_name"])), column_name=normalize_name(get(row, PRED_ALIASES["column_name"])), predicate_text=get(row, PRED_ALIASES["predicate_text"]), predicate_type=normalize_name(get(row, PRED_ALIASES["predicate_type"])), stage=normalize_name(get(row, PRED_ALIASES["stage"])), indexable=normalize_name(get(row, PRED_ALIASES["indexable"])), filter_factor=as_float(get(row, PRED_ALIASES["filter_factor"])), raw=row, )) return out def parse_indexes(path: Optional[str]) -> List[IndexRow]: out: List[IndexRow] = [] for row in read_records(path): cols = get(row, INDEX_ALIASES["columns"]) col_list = [normalize_name(c) for c in re.split(r"[,;|\s]+", cols) if c.strip()] out.append(IndexRow( creator=normalize_name(get(row, INDEX_ALIASES["creator"])), table_name=normalize_name(get(row, INDEX_ALIASES["table_name"])), index_schema=normalize_name(get(row, INDEX_ALIASES["index_schema"])), index_name=normalize_name(get(row, INDEX_ALIASES["index_name"])), columns=col_list, unique=normalize_name(get(row, INDEX_ALIASES["unique"])), clustering=normalize_name(get(row, INDEX_ALIASES["clustering"])), clustering_ratio=as_float(get(row, INDEX_ALIASES["clustering_ratio"])), nlevels=as_int(get(row, INDEX_ALIASES["nlevels"])), firstkeycard=as_float(get(row, INDEX_ALIASES["firstkeycard"])), fullkeycard=as_float(get(row, INDEX_ALIASES["fullkeycard"])), lastused=get(row, INDEX_ALIASES["lastused"]), raw=row, )) return out def parse_table_stats(path: Optional[str]) -> List[TableStats]: out: List[TableStats] = [] for row in read_records(path): out.append(TableStats( creator=normalize_name(get(row, STATS_ALIASES["creator"])), table_name=normalize_name(get(row, STATS_ALIASES["table_name"])), cardinality=as_float(get(row, STATS_ALIASES["cardinality"])), npages=as_float(get(row, STATS_ALIASES["npages"])), stats_date=get(row, STATS_ALIASES["stats_date"]), partitions=as_int(get(row, STATS_ALIASES["partitions"])), volatile=normalize_name(get(row, STATS_ALIASES["volatile"])), compressed_pct=as_float(get(row, STATS_ALIASES["compressed_pct"])), organization=normalize_name(get(row, STATS_ALIASES["organization"])), raw=row, )) return out def parse_packages(path: Optional[str]) -> List[PackageRow]: out: List[PackageRow] = [] for row in read_records(path): out.append(PackageRow( collection=normalize_name(get(row, PACKAGE_ALIASES["collection"])), package=normalize_name(get(row, PACKAGE_ALIASES["package"])), program=normalize_name(get(row, PACKAGE_ALIASES["program"])), bind_time=get(row, PACKAGE_ALIASES["bind_time"]), isolation=normalize_name(get(row, PACKAGE_ALIASES["isolation"])), release=normalize_name(get(row, PACKAGE_ALIASES["release"])), reopt=normalize_name(get(row, PACKAGE_ALIASES["reopt"])), degree=normalize_name(get(row, PACKAGE_ALIASES["degree"])), currentdata=normalize_name(get(row, PACKAGE_ALIASES["currentdata"])), validate=normalize_name(get(row, PACKAGE_ALIASES["validate"])), dynamicrules=normalize_name(get(row, PACKAGE_ALIASES["dynamicrules"])), qualifier=normalize_name(get(row, PACKAGE_ALIASES["qualifier"])), raw=row, )) return out def strip_sql_comments(text: str) -> str: text = re.sub(r"/\*.*?\*/", " ", text, flags=re.S) lines = [] for line in text.splitlines(): lines.append(re.sub(r"--.*$", "", line)) return "\n".join(lines) def split_sql_statements(text: str) -> List[str]: text = strip_sql_comments(text) parts = [] current = [] in_quote = False quote = "" for ch in text: if ch in {"'", '"'}: if not in_quote: in_quote = True quote = ch elif quote == ch: in_quote = False if ch == ";" and not in_quote: stmt = "".join(current).strip() if stmt: parts.append(stmt) current = [] else: current.append(ch) tail = "".join(current).strip() if tail: parts.append(tail) return parts def parse_sql_file(path: Optional[str]) -> List[SQLStatement]: if not path: return [] with open(path, "r", encoding="utf-8", errors="ignore") as fh: text = fh.read() stmts = [] for i, sql in enumerate(split_sql_statements(text), 1): sql_clean = re.sub(r"\s+", " ", sql).strip() stype = normalize_name(sql_clean.split()[0] if sql_clean.split() else "") tables = extract_tables(sql_clean) cols = extract_where_columns(sql_clean) preds = extract_predicate_fragments(sql_clean) h = hashlib.sha1(sql_clean.encode("utf-8", errors="ignore")).hexdigest()[:12] stmts.append(SQLStatement(str(i), sql_clean, stype, tables, cols, preds, h)) return stmts def extract_tables(sql: str) -> List[str]: tables: List[str] = [] patterns = [r"\bFROM\s+([A-Z0-9_.$]+)", r"\bJOIN\s+([A-Z0-9_.$]+)", r"\bUPDATE\s+([A-Z0-9_.$]+)", r"\bDELETE\s+FROM\s+([A-Z0-9_.$]+)", r"\bINTO\s+([A-Z0-9_.$]+)"] s = sql.upper() for pat in patterns: for m in re.finditer(pat, s): candidate = m.group(1).strip().strip(",") if candidate not in SQL_KEYWORDS and not candidate.startswith("("): tables.append(candidate) return sorted(set(tables)) def extract_where_columns(sql: str) -> List[str]: s = sql.upper() where = "" m = re.search(r"\bWHERE\b(.+?)(\bGROUP\s+BY\b|\bORDER\s+BY\b|\bHAVING\b|\bFETCH\b|$)", s, re.S) if m: where = m.group(1) candidates = [] for m in re.finditer(r"\b([A-Z][A-Z0-9_]*(?:\.[A-Z][A-Z0-9_]*)?)\s*(=|<>|>|<|>=|<=|LIKE|IN\s*\(|BETWEEN\b|IS\b)", where): col = m.group(1).split(".")[-1] if col not in SQL_KEYWORDS: candidates.append(col) # functions on columns, e.g. DATE(COL)= for m in re.finditer(r"\b[A-Z_]+\s*\(\s*([A-Z][A-Z0-9_]*(?:\.[A-Z][A-Z0-9_]*)?)\s*\)\s*(=|>|<|LIKE)", where): col = m.group(1).split(".")[-1] if col not in SQL_KEYWORDS: candidates.append(col) return sorted(set(candidates)) def extract_order_group_columns(sql: str) -> Tuple[List[str], List[str]]: s = sql.upper() order_cols: List[str] = [] group_cols: List[str] = [] for keyword, target in [("ORDER BY", order_cols), ("GROUP BY", group_cols)]: m = re.search(r"\b" + keyword.replace(" ", r"\s+") + r"\b(.+?)(\bFETCH\b|\bHAVING\b|\bORDER\s+BY\b|$)", s, re.S) if m: for piece in m.group(1).split(","): token = re.sub(r"\b(ASC|DESC|NULLS|FIRST|LAST)\b", "", piece).strip() token = re.sub(r"[^A-Z0-9_.]", "", token) if token: target.append(token.split(".")[-1]) return sorted(set(order_cols)), sorted(set(group_cols)) def extract_predicate_fragments(sql: str) -> List[str]: s = sql.upper() m = re.search(r"\bWHERE\b(.+?)(\bGROUP\s+BY\b|\bORDER\s+BY\b|\bHAVING\b|\bFETCH\b|$)", s, re.S) if not m: return [] where = m.group(1) return [frag.strip() for frag in re.split(r"\bAND\b", where) if frag.strip()] def parse_date(value: str) -> Optional[date]: if not value: return None value = value.strip() for fmt in ["%Y-%m-%d", "%Y/%m/%d", "%d/%m/%Y", "%Y%m%d", "%Y-%m-%d %H:%M:%S"]: try: return datetime.strptime(value[:19], fmt).date() except Exception: continue return None def days_old(value: str) -> Optional[int]: d = parse_date(value) if not d: return None return (date.today() - d).days class DB2Analyzer: def __init__(self, profile: str, policy: Dict[str, Any]): self.profile = profile self.policy = policy self.findings: List[Finding] = [] self.index_candidates: List[IndexCandidate] = [] self.baseline_compare: List[BaselineCompare] = [] def add(self, severity: str, code: str, category: str, title: str, detail: str, recommendation: str, stmt_id: str = "", table: str = "", package: str = "", evidence: str = "", score: int = 0) -> None: self.findings.append(Finding(severity, code, category, title, detail, recommendation, stmt_id, table, package, evidence, score)) def analyze(self, plan_rows: List[PlanRow], sqls: List[SQLStatement], predicates: List[PredicateRow], indexes: List[IndexRow], stats: List[TableStats], packages: List[PackageRow], baseline_rows: List[PlanRow]) -> AnalysisResult: indexes_by_table = defaultdict(list) for ix in indexes: indexes_by_table[ix.full_table].append(ix) stats_by_table = {st.full_table: st for st in stats} pred_by_stmt_table = defaultdict(list) for pred in predicates: pred_by_stmt_table[(pred.stmt_id, pred.full_table)].append(pred) sql_by_stmt = {s.stmt_id: s for s in sqls} package_by_name = {pkg.full_package: pkg for pkg in packages} # Attach SQL text to plan rows when missing by stmt_id. for pr in plan_rows: if not pr.sql_text and pr.stmt_id in sql_by_stmt: pr.sql_text = sql_by_stmt[pr.stmt_id].sql_text pr.statement_type = pr.statement_type or sql_by_stmt[pr.stmt_id].statement_type self.analyze_plan(plan_rows, pred_by_stmt_table, indexes_by_table, stats_by_table) self.analyze_predicates(predicates) self.analyze_sql(sqls, indexes_by_table) self.analyze_indexes(indexes) self.analyze_stats(stats) self.analyze_packages(packages, plan_rows) self.generate_index_candidates(plan_rows, sqls, predicates, indexes_by_table) self.compare_baseline(plan_rows, baseline_rows) categories = self.compute_categories() highest = self.highest_severity() risk = min(100, sum(max(1, f.score) for f in self.findings)) status = "OK" if highest == "CRITICAL": status = "BLOCKED" elif highest == "ERROR": status = "HIGH_RISK" elif highest == "WARNING": status = "ATTENTION" summary = self.build_summary(plan_rows, sqls, predicates, indexes, stats, packages) return AnalysisResult( status=status, highest_severity=highest, risk_score=risk, profile=self.profile, generated_at=datetime.now().isoformat(timespec="seconds"), summary=summary, plan_rows=plan_rows, sql_statements=sqls, predicates=predicates, indexes=indexes, table_stats=stats, packages=packages, findings=self.findings, index_candidates=self.index_candidates, baseline_compare=self.baseline_compare, categories=categories, ) def analyze_plan(self, rows: List[PlanRow], pred_by_stmt_table: Dict[Tuple[str, str], List[PredicateRow]], indexes_by_table: Dict[str, List[IndexRow]], stats_by_table: Dict[str, TableStats]) -> None: for pr in rows: table = pr.full_table pkg = ".".join(x for x in [pr.collection, pr.package] if x) access = pr.access_type.upper() rows_examined = pr.rows_examined or stats_by_table.get(table, TableStats()).cardinality if pr.total_cost >= self.policy["critical_cost"]: self.add("CRITICAL", "DB2_COST_CRITICAL", "cost_capacity", "Critical optimizer cost", f"Statement {pr.stmt_id} has estimated cost {pr.total_cost:.0f}.", "Review predicates, access path, RUNSTATS and index design.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 12) elif pr.total_cost >= self.policy["high_cost"]: self.add("WARNING", "DB2_COST_HIGH", "cost_capacity", "High optimizer cost", f"Statement {pr.stmt_id} has estimated cost {pr.total_cost:.0f}.", "Check whether this access path is expected for the workload window.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 5) if pr.getpages >= self.policy["critical_getpages"]: self.add("CRITICAL", "DB2_GETPAGES_CRITICAL", "cost_capacity", "Critical GETPAGE volume", f"GETPAGE count is {pr.getpages:.0f}.", "Investigate buffer pool pressure, index access, table scan, and predicate selectivity.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 12) elif pr.getpages >= self.policy["high_getpages"]: self.add("WARNING", "DB2_GETPAGES_HIGH", "cost_capacity", "High GETPAGE volume", f"GETPAGE count is {pr.getpages:.0f}.", "Check whether table/index access is reading too many pages.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 5) if pr.elapsed_ms >= self.policy.get("critical_elapsed_ms", 15000.0): self.add("CRITICAL", "DB2_ELAPSED_CRITICAL", "runtime", "Critical elapsed time", f"Elapsed time is {pr.elapsed_ms:.0f} ms.", "Review wait time, access path, locking, sort and DB2 accounting detail.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 10) elif pr.elapsed_ms >= self.policy.get("high_elapsed_ms", 3000.0): self.add("WARNING", "DB2_ELAPSED_HIGH", "runtime", "High elapsed time", f"Elapsed time is {pr.elapsed_ms:.0f} ms.", "Check if this statement belongs to an online/API path or batch window.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 4) if pr.cpu_cost >= self.policy.get("critical_cpu_ms", 7000.0): self.add("ERROR", "DB2_CPU_CRITICAL", "runtime", "High CPU cost", f"CPU estimate is {pr.cpu_cost:.0f} ms.", "Look for stage 2 predicates, scalar functions, sort, and excessive row qualification.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 8) elif pr.cpu_cost >= self.policy.get("high_cpu_ms", 1000.0): self.add("WARNING", "DB2_CPU_HIGH", "runtime", "Elevated CPU cost", f"CPU estimate is {pr.cpu_cost:.0f} ms.", "Review predicate indexability and sort requirements.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 4) if pr.wait_ms >= self.policy.get("critical_wait_ms", 10000.0): self.add("CRITICAL", "DB2_WAIT_CRITICAL", "locking_wait", "Critical wait time", f"Wait estimate is {pr.wait_ms:.0f} ms.", "Inspect locking, IRLM waits, synchronous I/O and buffer pool metrics.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 10) elif pr.wait_ms >= self.policy.get("high_wait_ms", 1500.0): self.add("WARNING", "DB2_WAIT_HIGH", "locking_wait", "High wait time", f"Wait estimate is {pr.wait_ms:.0f} ms.", "Correlate with accounting class 3 time and lock/latch waits.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 4) if access in {"R", "T", ""} and table: severity = "CRITICAL" if rows_examined >= self.policy["critical_table_scan_rows"] else "WARNING" code = "DB2_TABLESPACE_SCAN_CRITICAL" if severity == "CRITICAL" else "DB2_TABLESPACE_SCAN" self.add(severity, code, "access_path", "Tablespace scan / non-index access", f"Access type {access or 'blank'} on {table}, estimated rows {rows_examined:.0f}.", "Review predicates and index availability. Confirm whether the scan is deliberate.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 12 if severity == "CRITICAL" else 5) if access in {"I", "N", "MX", "MI"} and pr.access_name and pr.matchcols == 0: self.add("ERROR", "DB2_INDEX_MATCHCOLS_ZERO", "indexing", "Index used with zero matching columns", f"Index {pr.access_name} is used but MATCHCOLS=0.", "Review predicate order and index leading columns. Consider a better compound index.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 8) elif access in {"I", "N", "MX", "MI"} and pr.access_name and pr.matchcols == 1: ix = self.find_index(indexes_by_table.get(table, []), pr.access_name) if ix and len(ix.columns) >= 3: self.add("WARNING", "DB2_LOW_MATCHCOLS", "indexing", "Low matching columns", f"Index {pr.access_name} has {len(ix.columns)} columns but only {pr.matchcols} matching column.", "Review whether predicates match the leading index columns.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 4) if any(pr.sort_flags.values()): flags = ", ".join(k for k, v in pr.sort_flags.items() if v) sev = "ERROR" if pr.total_cost >= self.policy["high_cost"] or pr.getpages >= self.policy["high_getpages"] else "WARNING" self.add(sev, "DB2_SORT_REQUIRED", "sort_workfile", "Sort/workfile required", f"Sort flags: {flags}.", "Check ORDER BY/GROUP BY/DISTINCT, index order, workfile usage and sort cost.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 6 if sev == "ERROR" else 3) if pr.prefetch in {"L", "D", "S"}: detail = {"L": "list prefetch", "D": "dynamic prefetch", "S": "sequential prefetch"}.get(pr.prefetch, pr.prefetch) sev = "WARNING" if pr.prefetch == "L" and pr.getpages >= self.policy["high_getpages"] else "INFO" self.add(sev, "DB2_PREFETCH_PATTERN", "access_path", f"{detail.title()} detected", f"PREFETCH={pr.prefetch} for {table}.", "Validate whether prefetch pattern matches the expected workload and clustering.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 3 if sev == "WARNING" else 1) if pr.rid_pool and normalize_name(pr.rid_pool) not in {"", "N", "NO", "0"}: self.add("ERROR", "DB2_RID_POOL_RISK", "sort_workfile", "RID pool pressure risk", f"RID pool indicator: {pr.rid_pool}.", "Review RID list processing, list prefetch, memory pressure and fallback risk.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 7) if pr.workfile and as_float(pr.workfile) > 0: sev = "ERROR" if as_float(pr.workfile) > 10000 else "WARNING" self.add(sev, "DB2_WORKFILE_USAGE", "sort_workfile", "Workfile usage detected", f"Workfile usage/pages: {pr.workfile}.", "Review sort/hash join requirements and workfile database sizing.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 6 if sev == "ERROR" else 3) if pr.method in {"1", "NL", "NESTED_LOOP"} and rows_examined >= self.policy["high_rows"]: self.add("WARNING", "DB2_NESTED_LOOP_HIGH_ROWS", "join", "Nested loop with high row volume", f"Join method {pr.method} with estimated rows {rows_examined:.0f}.", "Check join order, join predicates and availability of supporting indexes.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 5) if pr.degree and normalize_name(pr.degree) not in {"", "1", "NO"}: self.add("INFO", "DB2_PARALLELISM", "capacity", "Parallelism detected", f"Degree/parallelism: {pr.degree}.", "Ensure parallelism is expected and does not inflate CPU or create contention.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 1) if pr.lock_mode in {"RR", "RS"}: self.add("WARNING", "DB2_STRICT_ISOLATION", "locking_wait", "Strict isolation level", f"Isolation/lock mode {pr.lock_mode} may increase lock footprint.", "Confirm consistency requirements; consider CS/UR only when business-safe.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 4) if table in self.policy.get("critical_tables", []): self.add("INFO", "DB2_CRITICAL_TABLE_TOUCHED", "governance", "Critical table referenced", f"Statement references critical table {table}.", "Apply stricter review, regression checks and change control.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 1) # Partition pruning heuristic. st = stats_by_table.get(table) if st and st.partitions > 1 and rows_examined >= self.policy["high_rows"]: pred_cols = {p.column_name for p in pred_by_stmt_table.get((pr.stmt_id, table), [])} sql_cols = set(extract_where_columns(pr.sql_text)) if pr.sql_text else set() combined = pred_cols | sql_cols if not any(c in combined for c in ["PARTITION_ID", "BUSINESS_DATE", "PROCESS_DATE", "CREATE_DATE", "TRAN_DATE"]): self.add("WARNING", "DB2_PARTITION_PRUNING_SUSPECT", "partitioning", "Partition pruning not obvious", f"Table {table} has {st.partitions} partitions and high row access.", "Check partitioning key predicates and whether pruning is effective.", pr.stmt_id, table, pkg, self.plan_evidence(pr), 5) def find_index(self, indexes: List[IndexRow], access_name: str) -> Optional[IndexRow]: access = normalize_name(access_name) for ix in indexes: if normalize_name(ix.index_name) == access or normalize_name(ix.full_index) == access: return ix return None def analyze_predicates(self, predicates: List[PredicateRow]) -> None: for pred in predicates: text = pred.predicate_text.upper() if pred.stage in {"2", "STAGE2", "STAGE 2"}: self.add("ERROR", "DB2_STAGE2_PREDICATE", "predicates", "Stage 2 predicate", f"Predicate is stage 2: {pred.predicate_text}", "Rewrite predicate to be stage 1/indexable where possible.", pred.stmt_id, pred.full_table, "", pred.predicate_text, 7) if pred.indexable in {"N", "NO", "FALSE", "0"}: self.add("WARNING", "DB2_NON_INDEXABLE_PREDICATE", "predicates", "Non-indexable predicate", f"Predicate is not indexable on column {pred.column_name}.", "Avoid functions, casts, incompatible data types, leading wildcard LIKE, or expression-side columns.", pred.stmt_id, pred.full_table, "", pred.predicate_text, 5) if re.search(r"\b(UPPER|LOWER|DATE|CHAR|DECIMAL|INTEGER|SUBSTR|COALESCE)\s*\(", text): self.add("WARNING", "DB2_FUNCTION_ON_COLUMN", "predicates", "Function on predicate column", f"Function detected in predicate: {pred.predicate_text}", "Consider generated columns, expression-based alternatives, or rewrite predicate to preserve indexability.", pred.stmt_id, pred.full_table, "", pred.predicate_text, 5) if re.search(r"LIKE\s+'%", text): self.add("WARNING", "DB2_LEADING_WILDCARD_LIKE", "predicates", "Leading wildcard LIKE", f"Leading wildcard detected: {pred.predicate_text}", "Leading wildcard usually prevents normal index matching; consider search index/offloaded search design.", pred.stmt_id, pred.full_table, "", pred.predicate_text, 4) if pred.filter_factor and pred.filter_factor > 0.20: self.add("INFO", "DB2_LOW_SELECTIVITY_PREDICATE", "predicates", "Low selectivity predicate", f"Filter factor {pred.filter_factor:.4f} may be weak for {pred.column_name}.", "Review column cardinality and whether it belongs in a leading index position.", pred.stmt_id, pred.full_table, "", pred.predicate_text, 1) def analyze_sql(self, sqls: List[SQLStatement], indexes_by_table: Dict[str, List[IndexRow]]) -> None: for st in sqls: s = st.sql_text.upper() first_table = st.tables[0] if st.tables else "" if re.search(r"\bSELECT\s+\*", s): self.add("WARNING", "DB2_SELECT_STAR", "sql_quality", "SELECT * detected", "SELECT * can increase I/O, coupling to schema, and network/API payload.", "Select only required columns, especially on API and online paths.", st.stmt_id, first_table, "", st.sql_text[:300], 3) if re.match(r"\s*DELETE\s+FROM\s+\S+\s*;?\s*$", s): self.add("CRITICAL", "DB2_DELETE_WITHOUT_WHERE", "sql_safety", "DELETE without WHERE", "DELETE statement has no WHERE clause.", "Block release unless this is controlled maintenance with explicit approval.", st.stmt_id, first_table, "", st.sql_text[:300], 15) if re.match(r"\s*UPDATE\s+\S+\s+SET\s+.+\s*;?\s*$", s) and " WHERE " not in f" {s} ": self.add("CRITICAL", "DB2_UPDATE_WITHOUT_WHERE", "sql_safety", "UPDATE without WHERE", "UPDATE statement has no WHERE clause.", "Block release unless this is controlled maintenance with explicit approval.", st.stmt_id, first_table, "", st.sql_text[:300], 15) for pattern in self.policy.get("banned_sql_patterns", []): if re.search(pattern, s, re.I | re.S): self.add("ERROR", "DB2_POLICY_BANNED_SQL", "governance", "Policy-banned SQL pattern", f"Pattern matched: {pattern}", "Review custom SQL policy and require DBA approval.", st.stmt_id, first_table, "", st.sql_text[:300], 8) if " OR " in s and " WHERE " in s: self.add("WARNING", "DB2_OR_PREDICATE", "predicates", "OR predicate detected", "OR predicates can reduce index matching or cause complex access paths.", "Consider UNION ALL rewrite or index design supporting each branch.", st.stmt_id, first_table, "", st.sql_text[:300], 3) if re.search(r"\b(NOT\s+IN|NOT\s+EXISTS|<>)\b", s): self.add("INFO", "DB2_NEGATIVE_PREDICATE", "predicates", "Negative predicate", "NOT/<> predicates can be less selective or harder to index.", "Validate access path and cardinality estimates.", st.stmt_id, first_table, "", st.sql_text[:300], 1) if re.search(r"\b(UPPER|LOWER|DATE|CHAR|DECIMAL|INTEGER|SUBSTR|COALESCE)\s*\(", s): self.add("WARNING", "DB2_SQL_FUNCTION_PREDICATE", "predicates", "Function expression in SQL", "Function or expression may prevent index matching.", "Rewrite if function wraps an indexed column in WHERE/JOIN.", st.stmt_id, first_table, "", st.sql_text[:300], 4) if st.statement_type == "SELECT" and self.profile == "api" and " FETCH FIRST " not in f" {s} ": self.add("WARNING", "DB2_API_NO_ROW_LIMIT", "sql_quality", "API SELECT without row limit", "API profile selected and no FETCH FIRST row limit was detected.", "Add deterministic filtering and row limit for online/API endpoints.", st.stmt_id, first_table, "", st.sql_text[:300], 4) if " ORDER BY " in s and " FETCH FIRST " not in f" {s} " and self.profile in {"api", "production", "strict"}: self.add("INFO", "DB2_ORDER_BY_NO_LIMIT", "sort_workfile", "ORDER BY without explicit limit", "Sort may process more rows than necessary.", "Check whether FETCH FIRST / pagination is appropriate.", st.stmt_id, first_table, "", st.sql_text[:300], 1) for col in st.columns: if any(sens in col for sens in self.policy.get("sensitive_columns", [])): self.add("INFO", "DB2_SENSITIVE_COLUMN_PREDICATE", "governance", "Sensitive column referenced", f"Predicate references sensitive-looking column {col}.", "Apply data protection and audit review if this SQL is exposed to API or reporting.", st.stmt_id, first_table, "", st.sql_text[:300], 1) def analyze_indexes(self, indexes: List[IndexRow]) -> None: for ix in indexes: if ix.clustering_ratio and ix.clustering_ratio < self.policy.get("critical_clustering_ratio", 50.0): self.add("ERROR", "DB2_INDEX_CLUSTERING_CRITICAL", "indexing", "Critical clustering ratio", f"Index {ix.full_index} clustering ratio is {ix.clustering_ratio:.1f}%.", "Consider REORG, clustering index review, or access path validation.", "", ix.full_table, "", ",".join(ix.columns), 8) elif ix.clustering_ratio and ix.clustering_ratio < self.policy.get("min_clustering_ratio", 80.0): self.add("WARNING", "DB2_INDEX_CLUSTERING_LOW", "indexing", "Low clustering ratio", f"Index {ix.full_index} clustering ratio is {ix.clustering_ratio:.1f}%.", "Monitor REORG need and prefetch efficiency.", "", ix.full_table, "", ",".join(ix.columns), 4) if ix.nlevels > self.policy.get("max_nlevels", 4): self.add("WARNING", "DB2_INDEX_NLEVELS_HIGH", "indexing", "High index levels", f"Index {ix.full_index} has NLEVELS={ix.nlevels}.", "Review index size, REORG, page size and key design.", "", ix.full_table, "", ",".join(ix.columns), 4) if ix.columns and len(ix.columns) > 6: self.add("INFO", "DB2_WIDE_INDEX", "indexing", "Wide compound index", f"Index {ix.full_index} has {len(ix.columns)} key columns.", "Validate maintenance cost and whether all columns are needed for matching/order.", "", ix.full_table, "", ",".join(ix.columns), 1) def analyze_stats(self, stats: List[TableStats]) -> None: for st in stats: age = days_old(st.stats_date) if st.cardinality <= 0 or st.npages <= 0: self.add("ERROR", "DB2_STATS_MISSING", "statistics", "Missing or suspicious RUNSTATS", f"Table {st.full_table} has cardinality={st.cardinality}, npages={st.npages}.", "Run or verify RUNSTATS and catalog statistics before trusting access paths.", "", st.full_table, "", str(st.raw), 8) if age is not None and age >= self.policy.get("critical_stale_stats_days", 90): self.add("ERROR", "DB2_STATS_STALE_CRITICAL", "statistics", "Critically stale RUNSTATS", f"Statistics for {st.full_table} are {age} days old.", "Run RUNSTATS with appropriate column group/frequency stats before bind/rebind.", "", st.full_table, "", st.stats_date, 8) elif age is not None and age >= self.policy.get("stale_stats_days", 30): self.add("WARNING", "DB2_STATS_STALE", "statistics", "Stale RUNSTATS", f"Statistics for {st.full_table} are {age} days old.", "Review RUNSTATS cadence and data volatility.", "", st.full_table, "", st.stats_date, 4) if st.partitions > 64: self.add("INFO", "DB2_MANY_PARTITIONS", "partitioning", "Many partitions", f"Table {st.full_table} has {st.partitions} partitions.", "Ensure partition pruning and partition-level maintenance are effective.", "", st.full_table, "", str(st.raw), 1) if st.volatile in {"Y", "YES", "TRUE"}: self.add("INFO", "DB2_VOLATILE_TABLE", "statistics", "Volatile table", f"Table {st.full_table} is marked volatile.", "Confirm volatile is intentional and explain estimates remain valid.", "", st.full_table, "", str(st.raw), 1) def analyze_packages(self, packages: List[PackageRow], plan_rows: List[PlanRow]) -> None: plan_pkg_names = {".".join(x for x in [p.collection, p.package] if x) for p in plan_rows} for pkg in packages: full = pkg.full_package if full and plan_pkg_names and full not in plan_pkg_names: # Not a finding by itself. pass if pkg.isolation in {"RR", "RS"}: self.add("WARNING", "DB2_PACKAGE_STRICT_ISOLATION", "package_bind", "Package uses strict isolation", f"{full} isolation={pkg.isolation}.", "Validate consistency requirement and concurrency impact.", "", "", full, str(pkg.raw), 4) if pkg.release == "DEALLOCATE": self.add("INFO", "DB2_PACKAGE_RELEASE_DEALLOCATE", "package_bind", "RELEASE(DEALLOCATE)", f"{full} uses RELEASE(DEALLOCATE).", "Good for high reuse but validate thread/resource lifecycle in CICS/batch context.", "", "", full, str(pkg.raw), 1) if pkg.reopt in {"NONE", "NO", ""}: self.add("INFO", "DB2_PACKAGE_REOPT_NONE", "package_bind", "REOPT not enabled", f"{full} reopt={pkg.reopt or 'blank'}.", "If host variables drive skewed predicates, consider REOPT strategy.", "", "", full, str(pkg.raw), 1) if pkg.degree in {"ANY", "YES"}: self.add("WARNING", "DB2_PACKAGE_DEGREE_ANY", "package_bind", "Package allows parallelism", f"{full} degree={pkg.degree}.", "Validate CPU impact and whether parallelism is expected.", "", "", full, str(pkg.raw), 4) if pkg.currentdata == "YES" and pkg.isolation == "CS": self.add("INFO", "DB2_PACKAGE_CURRENTDATA_YES", "package_bind", "CURRENTDATA(YES)", f"{full} CURRENTDATA=YES.", "May reduce concurrency; confirm application semantics.", "", "", full, str(pkg.raw), 1) if pkg.validate == "RUN": self.add("WARNING", "DB2_PACKAGE_VALIDATE_RUN", "package_bind", "VALIDATE(RUN)", f"{full} uses VALIDATE(RUN).", "Deployment may defer object authorization/existence issues to runtime.", "", "", full, str(pkg.raw), 4) def generate_index_candidates(self, plan_rows: List[PlanRow], sqls: List[SQLStatement], predicates: List[PredicateRow], indexes_by_table: Dict[str, List[IndexRow]]) -> None: sql_by_stmt = {s.stmt_id: s for s in sqls} pred_cols_by_key = defaultdict(list) range_cols_by_key = defaultdict(list) for p in predicates: col = normalize_name(p.column_name) if not col: continue key = (p.stmt_id, p.full_table) op = p.predicate_type or p.predicate_text if re.search(r"(=|IN)", op, re.I): pred_cols_by_key[key].append(col) else: range_cols_by_key[key].append(col) for pr in plan_rows: table = pr.full_table if not table: continue needs = pr.access_type in {"R", "T", ""} or (pr.access_type in {"I", "N", "MX", "MI"} and pr.matchcols == 0) if not needs: continue equality_cols = list(dict.fromkeys(pred_cols_by_key.get((pr.stmt_id, table), []))) range_cols = list(dict.fromkeys(range_cols_by_key.get((pr.stmt_id, table), []))) order_cols: List[str] = [] group_cols: List[str] = [] if pr.sql_text: sql_cols = extract_where_columns(pr.sql_text) equality_cols.extend([c for c in sql_cols if c not in equality_cols]) order_cols, group_cols = extract_order_group_columns(pr.sql_text) elif pr.stmt_id in sql_by_stmt: equality_cols.extend([c for c in sql_by_stmt[pr.stmt_id].columns if c not in equality_cols]) order_cols, group_cols = extract_order_group_columns(sql_by_stmt[pr.stmt_id].sql_text) columns = [] for col in equality_cols + range_cols + order_cols + group_cols: if col and col not in columns: columns.append(col) if not columns: continue if self.is_index_prefix_covered(indexes_by_table.get(table, []), columns[:2] or columns[:1]): continue short_table = table.split(".")[-1] name = f"IX_{short_table}_{'_'.join(columns[:3])}"[:64] benefit = "HIGH" if pr.total_cost >= self.policy["critical_cost"] or pr.getpages >= self.policy["critical_getpages"] else "MEDIUM" self.index_candidates.append(IndexCandidate( stmt_id=pr.stmt_id, table=table, candidate_name=name, columns=columns[:6], reason="Table scan or weak index matching with usable predicates.", estimated_benefit=benefit, evidence=self.plan_evidence(pr), risk="HIGH" if table in self.policy.get("critical_tables", []) else "MEDIUM", )) self.add("INFO", "DB2_INDEX_CANDIDATE", "indexing", "Index candidate generated", f"Candidate {name} on {table}({', '.join(columns[:6])}).", "Validate with DBA, workload mix, update cost, clustering and RUNSTATS before implementation.", pr.stmt_id, table, "", self.plan_evidence(pr), 1) def is_index_prefix_covered(self, indexes: List[IndexRow], cols: List[str]) -> bool: if not cols: return False cols_norm = [normalize_name(c) for c in cols] for ix in indexes: if ix.columns[:len(cols_norm)] == cols_norm: return True return False def compare_baseline(self, current: List[PlanRow], baseline: List[PlanRow]) -> None: if not baseline: return base_by_key: Dict[Tuple[str, str], PlanRow] = {} for br in baseline: base_by_key[(br.stmt_id, br.full_table)] = br for cr in current: br = base_by_key.get((cr.stmt_id, cr.full_table)) if not br: continue for metric, base_val, cur_val in [ ("total_cost", br.total_cost, cr.total_cost), ("getpages", br.getpages, cr.getpages), ("elapsed_ms", br.elapsed_ms, cr.elapsed_ms), ("rows_examined", br.rows_examined, cr.rows_examined), ]: if base_val > 0 and cur_val > base_val: delta = (cur_val - base_val) / base_val * 100.0 sev = "CRITICAL" if delta >= self.policy["critical_regression_pct"] else "WARNING" if delta >= self.policy["regression_pct"] else "INFO" if sev != "INFO": self.baseline_compare.append(BaselineCompare(cr.stmt_id, cr.full_table, ".".join(x for x in [cr.collection, cr.package] if x), metric, f"{base_val:.2f}", f"{cur_val:.2f}", delta, sev, f"{metric} increased by {delta:.1f}%")) self.add(sev, "DB2_BASELINE_REGRESSION", "regression", "Baseline regression detected", f"{metric} increased from {base_val:.2f} to {cur_val:.2f} ({delta:.1f}%).", "Review access path, statistics, package rebind, object changes and data growth.", cr.stmt_id, cr.full_table, "", self.plan_evidence(cr), 10 if sev == "CRITICAL" else 5) if normalize_name(br.access_type) != normalize_name(cr.access_type): self.baseline_compare.append(BaselineCompare(cr.stmt_id, cr.full_table, ".".join(x for x in [cr.collection, cr.package] if x), "access_type", br.access_type, cr.access_type, 0.0, "WARNING", "Access type changed")) self.add("WARNING", "DB2_ACCESS_PATH_CHANGED", "regression", "Access path changed", f"Access type changed from {br.access_type} to {cr.access_type}.", "Verify if this was expected after RUNSTATS/rebind/schema change.", cr.stmt_id, cr.full_table, "", f"baseline={br.access_type}; current={cr.access_type}", 4) if br.access_name and cr.access_name and normalize_name(br.access_name) != normalize_name(cr.access_name): self.baseline_compare.append(BaselineCompare(cr.stmt_id, cr.full_table, ".".join(x for x in [cr.collection, cr.package] if x), "access_name", br.access_name, cr.access_name, 0.0, "INFO", "Index access name changed")) def compute_categories(self) -> List[CategoryScore]: cats = sorted(set(f.category for f in self.findings)) result = [] for cat in cats: fs = [f for f in self.findings if f.category == cat] score = min(100, sum(max(1, f.score) for f in fs)) result.append(CategoryScore(cat, score, len(fs), sum(1 for f in fs if f.severity == "CRITICAL"), sum(1 for f in fs if f.severity == "ERROR"), sum(1 for f in fs if f.severity == "WARNING"))) return sorted(result, key=lambda c: c.score, reverse=True) def highest_severity(self) -> str: if not self.findings: return "INFO" return max((f.severity for f in self.findings), key=lambda s: SEVERITY_ORDER.get(s, 0)) def build_summary(self, plan_rows: List[PlanRow], sqls: List[SQLStatement], predicates: List[PredicateRow], indexes: List[IndexRow], stats: List[TableStats], packages: List[PackageRow]) -> Dict[str, Any]: return { "explain_rows": len(plan_rows), "sql_statements": len(sqls), "predicates": len(predicates), "indexes": len(indexes), "table_stats": len(stats), "packages": len(packages), "tables": len(set(p.full_table for p in plan_rows if p.full_table)), "packages_in_explain": len(set(".".join(x for x in [p.collection, p.package] if x) for p in plan_rows if p.collection or p.package)), "findings": len(self.findings), "index_candidates": len(self.index_candidates), "baseline_comparisons": len(self.baseline_compare), "total_cost": sum(p.total_cost for p in plan_rows), "total_getpages": sum(p.getpages for p in plan_rows), "total_elapsed_ms": sum(p.elapsed_ms for p in plan_rows), "access_types": dict(Counter(p.access_type or "BLANK" for p in plan_rows)), "top_tables_by_cost": self.top_tables(plan_rows, "total_cost"), "top_tables_by_getpages": self.top_tables(plan_rows, "getpages"), } def top_tables(self, rows: List[PlanRow], attr: str) -> List[Dict[str, Any]]: values = defaultdict(float) for r in rows: if r.full_table: values[r.full_table] += getattr(r, attr) return [{"table": k, attr: round(v, 2)} for k, v in sorted(values.items(), key=lambda item: item[1], reverse=True)[:10]] def plan_evidence(self, pr: PlanRow) -> str: return f"stmt={pr.stmt_id} table={pr.full_table} access={pr.access_type} index={pr.access_name} matchcols={pr.matchcols} cost={pr.total_cost} getpages={pr.getpages} rows={pr.rows_examined} elapsed_ms={pr.elapsed_ms}" def analysis_to_json_dict(result: AnalysisResult) -> Dict[str, Any]: return { "status": result.status, "highest_severity": result.highest_severity, "risk_score": result.risk_score, "profile": result.profile, "generated_at": result.generated_at, "summary": result.summary, "plan_rows": [asdict(x) for x in result.plan_rows], "sql_statements": [asdict(x) for x in result.sql_statements], "predicates": [asdict(x) for x in result.predicates], "indexes": [asdict(x) for x in result.indexes], "table_stats": [asdict(x) for x in result.table_stats], "packages": [asdict(x) for x in result.packages], "findings": [asdict(x) for x in result.findings], "index_candidates": [asdict(x) for x in result.index_candidates], "baseline_compare": [asdict(x) for x in result.baseline_compare], "categories": [asdict(x) for x in result.categories], } def write_json(path: Optional[str], data: Any) -> None: if path: with open(path, "w", encoding="utf-8") as fh: json.dump(data, fh, indent=2, ensure_ascii=False) def write_csv(path: Optional[str], rows: Iterable[Any], fieldnames: Optional[List[str]] = None) -> None: if not path: return rows_list = list(rows) if not fieldnames: if rows_list: first = rows_list[0] if hasattr(first, "__dataclass_fields__"): fieldnames = list(asdict(first).keys()) elif isinstance(first, dict): fieldnames = list(first.keys()) else: fieldnames = ["value"] else: fieldnames = ["empty"] with open(path, "w", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() for item in rows_list: if hasattr(item, "__dataclass_fields__"): row = asdict(item) elif isinstance(item, dict): row = item else: row = {"value": item} # flatten complex structures for k, v in list(row.items()): if isinstance(v, (dict, list)): row[k] = json.dumps(v, ensure_ascii=False) writer.writerow(row) def severity_class(sev: str) -> str: return { "CRITICAL": "sev-critical", "ERROR": "sev-error", "WARNING": "sev-warning", "INFO": "sev-info", }.get(sev, "sev-info") def render_html(result: AnalysisResult, path: Optional[str]) -> None: if not path: return findings_rows = "".join( f"{html.escape(f.severity)}{html.escape(f.code)}{html.escape(f.category)}{html.escape(f.title)}{html.escape(f.stmt_id)}{html.escape(f.table)}{html.escape(f.detail)}{html.escape(f.recommendation)}" for f in sorted(result.findings, key=lambda x: (-SEVERITY_ORDER.get(x.severity, 0), x.category, x.code))[:300] ) candidate_rows = "".join( f"{html.escape(c.stmt_id)}{html.escape(c.table)}{html.escape(c.candidate_name)}{html.escape(', '.join(c.columns))}{html.escape(c.estimated_benefit)}{html.escape(c.reason)}" for c in result.index_candidates[:100] ) compare_rows = "".join( f"{html.escape(c.severity)}{html.escape(c.stmt_id)}{html.escape(c.table)}{html.escape(c.metric)}{html.escape(c.baseline)}{html.escape(c.current)}{c.delta_pct:.1f}%{html.escape(c.detail)}" for c in result.baseline_compare[:200] ) category_rows = "".join( f"{html.escape(c.category)}{c.score}{c.findings}{c.critical}{c.error}{c.warning}" for c in result.categories ) top_cost = "".join(f"
  • {html.escape(x['table'])}: {x['total_cost']}
    • " for x in result.summary.get("top_tables_by_cost", [])) top_getpages = "".join(f"
  • {html.escape(x['table'])}: {x['getpages']}
    • " for x in result.summary.get("top_tables_by_getpages", [])) html_doc = f""" DB2 for z/OS SQL / Explain Analyzer V2

    DB2 for z/OS SQL / Explain Analyzer V2

    Generated {html.escape(result.generated_at)} | Profile {html.escape(result.profile)}

    Status{html.escape(result.status)}
    Risk score{result.risk_score}/100
    Highest{html.escape(result.highest_severity)}
    Explain rows{result.summary.get('explain_rows',0)}
    Findings{len(result.findings)}
    Index candidates{len(result.index_candidates)}

    Top tables by cost

      {top_cost}

    Top tables by GETPAGEs

      {top_getpages}

    Category score

    {category_rows}
    CategoryScoreFindingsCriticalErrorWarning

    Index candidates

    {candidate_rows or ''}
    StmtTableCandidateColumnsBenefitReason
    No candidates generated.

    Baseline regressions

    {compare_rows or ''}
    SeverityStmtTableMetricBaselineCurrentDeltaDetail
    No baseline regressions detected.

    Findings

    {findings_rows}
    SeverityCodeCategoryTitleStmtTableDetailRecommendation
    """ with open(path, "w", encoding="utf-8") as fh: fh.write(html_doc) def console_summary(result: AnalysisResult) -> str: lines = [ "DB2 for z/OS SQL / Explain Analyzer V2", f"Status : {result.status}", f"Highest severity : {result.highest_severity}", f"Risk score : {result.risk_score}/100", f"Profile : {result.profile}", f"Explain rows : {result.summary.get('explain_rows', 0)}", f"SQL statements : {result.summary.get('sql_statements', 0)}", f"Predicates : {result.summary.get('predicates', 0)}", f"Indexes : {result.summary.get('indexes', 0)}", f"Table stats : {result.summary.get('table_stats', 0)}", f"Packages : {result.summary.get('packages', 0)}", f"Findings : {len(result.findings)}", f"Index candidates : {len(result.index_candidates)}", f"Baseline compare : {len(result.baseline_compare)}", "", "TOP FINDINGS", ] for f in sorted(result.findings, key=lambda x: -SEVERITY_ORDER.get(x.severity, 0))[:10]: lines.append(f"- [{f.severity}] {f.code} stmt={f.stmt_id} table={f.table} :: {f.title}") if result.index_candidates: lines.append("") lines.append("INDEX CANDIDATES") for c in result.index_candidates[:5]: lines.append(f"- {c.candidate_name} ON {c.table}({', '.join(c.columns)}) benefit={c.estimated_benefit}") return "\n".join(lines) def maybe_redact(result: AnalysisResult) -> None: def redact_table(t: str) -> str: if not t: return t return re.sub(r"[A-Z0-9_]+\.[A-Z0-9_]+", "", t) for f in result.findings: f.table = redact_table(f.table) f.detail = re.sub(r"[A-Z0-9_]+\.[A-Z0-9_]+", "
    ", f.detail) f.evidence = re.sub(r"[A-Z0-9_]+\.[A-Z0-9_]+", "
    ", f.evidence) for p in result.plan_rows: p.creator = "" if p.creator else p.creator p.table_name = "
    " if p.table_name else p.table_name p.raw = {} for c in result.index_candidates: c.table = redact_table(c.table) c.evidence = re.sub(r"[A-Z0-9_]+\.[A-Z0-9_]+", "
    ", c.evidence) def write_sample_files(prefix: str) -> None: base = Path(prefix) base.mkdir(parents=True, exist_ok=True) write_text(base / "sample_db2_v2_explain_current.csv", SAMPLE_EXPLAIN_CURRENT) write_text(base / "sample_db2_v2_explain_baseline.csv", SAMPLE_EXPLAIN_BASELINE) write_text(base / "sample_db2_v2_sql.sql", SAMPLE_SQL) write_text(base / "sample_db2_v2_predicates.csv", SAMPLE_PREDICATES) write_text(base / "sample_db2_v2_indexes.csv", SAMPLE_INDEXES) write_text(base / "sample_db2_v2_table_stats.csv", SAMPLE_TABLE_STATS) write_text(base / "sample_db2_v2_packages.csv", SAMPLE_PACKAGES) write_text(base / "sample_db2_v2_policy.json", json.dumps(SAMPLE_POLICY, indent=2)) def write_text(path: Path, content: str) -> None: with open(path, "w", encoding="utf-8", newline="") as fh: fh.write(content.strip() + "\n") def run_analysis_from_args(args: argparse.Namespace) -> AnalysisResult: policy = load_policy(args.policy, args.profile) explain_path = args.explain or args.input_file plan_rows = parse_plan_rows(explain_path) sqls = parse_sql_file(args.sql) predicates = parse_predicates(args.predicates) indexes = parse_indexes(args.indexes) stats = parse_table_stats(args.table_stats) packages = parse_packages(args.packages) baseline_rows = parse_plan_rows(args.baseline) analyzer = DB2Analyzer(args.profile, policy) result = analyzer.analyze(plan_rows, sqls, predicates, indexes, stats, packages, baseline_rows) if args.redact: maybe_redact(result) return result def main(argv: Optional[List[str]] = None) -> int: parser = argparse.ArgumentParser(description="DB2 for z/OS SQL / EXPLAIN Analyzer V2") parser.add_argument("input_file", nargs="?", help="PLAN_TABLE-like current explain export CSV/JSON/JSONL") parser.add_argument("--explain", help="PLAN_TABLE-like current explain export CSV/JSON/JSONL") parser.add_argument("--sql", help="SQL statements file") parser.add_argument("--predicates", help="DSN_PREDICAT_TABLE-like predicate export") parser.add_argument("--indexes", help="Index catalog export") parser.add_argument("--table-stats", help="Table statistics export") parser.add_argument("--packages", help="Package / bind options export") parser.add_argument("--baseline", help="Baseline explain export for regression comparison") parser.add_argument("--policy", help="Custom policy JSON") parser.add_argument("--profile", default="production", choices=["production", "performance", "db2", "batch", "api", "packages", "capacity", "training", "strict"]) parser.add_argument("--write-samples", help="Write sample input files into the given directory") parser.add_argument("--demo", action="store_true", help="Run against generated demo samples") parser.add_argument("--redact", action="store_true", help="Redact schema/table names in outputs") parser.add_argument("--json", dest="json_output", help="Write full JSON analysis") parser.add_argument("--html", dest="html_output", help="Write HTML report") parser.add_argument("--csv-findings", help="Write findings CSV") parser.add_argument("--csv-plan", help="Write normalized plan rows CSV") parser.add_argument("--csv-statements", help="Write SQL statements CSV") parser.add_argument("--csv-predicates", help="Write predicates CSV") parser.add_argument("--csv-index-candidates", help="Write index candidates CSV") parser.add_argument("--csv-compare", help="Write baseline comparison CSV") parser.add_argument("--csv-categories", help="Write category scores CSV") parser.add_argument("--csv-indexes", help="Write index catalog normalized CSV") parser.add_argument("--csv-stats", help="Write table stats normalized CSV") parser.add_argument("--csv-packages", help="Write packages normalized CSV") parser.add_argument("--fail-on", default="NONE", choices=["NONE", "INFO", "WARNING", "ERROR", "CRITICAL"], help="Return non-zero if highest severity is at least this value") args = parser.parse_args(argv) if args.write_samples: write_sample_files(args.write_samples) print(f"Sample files written to {args.write_samples}") return 0 if args.demo: demo_dir = Path("db2_v2_demo_samples") write_sample_files(str(demo_dir)) args.input_file = str(demo_dir / "sample_db2_v2_explain_current.csv") args.sql = str(demo_dir / "sample_db2_v2_sql.sql") args.predicates = str(demo_dir / "sample_db2_v2_predicates.csv") args.indexes = str(demo_dir / "sample_db2_v2_indexes.csv") args.table_stats = str(demo_dir / "sample_db2_v2_table_stats.csv") args.packages = str(demo_dir / "sample_db2_v2_packages.csv") args.baseline = str(demo_dir / "sample_db2_v2_explain_baseline.csv") args.policy = str(demo_dir / "sample_db2_v2_policy.json") if not (args.input_file or args.explain): parser.error("Provide input_file / --explain, or use --demo / --write-samples") result = run_analysis_from_args(args) print(console_summary(result)) write_json(args.json_output, analysis_to_json_dict(result)) render_html(result, args.html_output) write_csv(args.csv_findings, result.findings) write_csv(args.csv_plan, result.plan_rows) write_csv(args.csv_statements, result.sql_statements) write_csv(args.csv_predicates, result.predicates) write_csv(args.csv_index_candidates, result.index_candidates) write_csv(args.csv_compare, result.baseline_compare) write_csv(args.csv_categories, result.categories) write_csv(args.csv_indexes, result.indexes) write_csv(args.csv_stats, result.table_stats) write_csv(args.csv_packages, result.packages) threshold = args.fail_on if threshold != "NONE" and SEVERITY_ORDER.get(result.highest_severity, 0) >= SEVERITY_ORDER.get(threshold, 99): return 2 return 0 SAMPLE_EXPLAIN_CURRENT = """ stmt_id,collection,package,program,statement_type,creator,table_name,access_type,access_name,matchcols,method,prefetch,sortn_orderby,sortn_groupby,total_cost,cpu_cost,io_cost,elapsed_ms,wait_ms,getpages,rows_examined,rows_returned,degree,stage2,rid_pool,workfile,lock_mode 1,PAYCOLL,PAYPKG1,PAYAPI,SELECT,PRODDB,CUSTOMER,R,,0,1,S,Y,N,12500,4200,8300,18000,2500,1800000,2400000,250,1,Y,N,24000,CS 2,PAYCOLL,PAYPKG1,PAYAPI,SELECT,PRODDB,ACCOUNT,I,IX_ACCOUNT_STATUS,0,1,L,N,N,6200,1900,4300,7200,1400,820000,950000,100,1,N,Y,12000,CS 3,PAYCOLL,PAYPKG2,PAYBATCH,SELECT,PRODDB,PAYMENT,I,IX_PAYMENT_DATE,2,2,N,N,Y,1800,700,1100,2200,600,160000,280000,5000,1,N,N,0,CS 4,TRNCOLL,TRNPKG1,TRNAPI,SELECT,PRODDB,TRANSACTION,R,,0,1,D,N,N,9800,3100,6700,11800,2300,1250000,5000000,200,1,Y,N,0,CS 5,AUDCOLL,AUDPKG1,AUDREP,SELECT,PRODDB,AUDIT_LOG,I,IX_AUDIT_DATE,1,1,L,Y,N,5200,2100,3100,9400,4800,650000,2000000,10000,ANY,N,Y,54000,RS 6,PAYCOLL,PAYPKG3,PAYUPD,UPDATE,PRODDB,CUSTOMER,I,IX_CUSTOMER_ID,1,1,N,N,N,900,500,400,900,300,50000,1,1,1,N,N,0,RR 7,APICOLL,APIPKG1,CUSTAPI,SELECT,PRODDB,CUSTOMER_API,R,,0,1,S,N,N,3100,900,2200,2700,900,310000,600000,50,1,N,N,0,CS 8,FINCOLL,FINPKG1,FINRPT,SELECT,FINANCE,LEDGER,R,,0,2,S,Y,Y,22000,9000,13000,65000,7000,4500000,12000000,900000,ANY,Y,Y,92000,CS 9,HRcoll,HRPKG1,HRAPI,SELECT,PAYROLL,EMPLOYEE,I,IX_EMPLOYEE_DEPT,1,1,N,N,N,2600,900,1700,1400,200,210000,700000,120,1,N,N,0,CS 10,DB2COLL,DB2PKG1,LOOKUP,SELECT,PRODDB,COUNTRY,I,IX_COUNTRY_PK,1,1,N,N,N,15,5,10,30,0,20,1,1,1,N,N,0,CS 11,ORDCOLL,ORDPKG1,ORDERAPI,SELECT,PRODDB,ORDERS,MX,IX_ORDERS_STATUS_DATE,1,1,L,Y,N,4100,1200,2900,4200,1100,410000,900000,300,1,N,Y,16000,CS 12,INVCOLL,INVPKG1,INVAPI,SELECT,PRODDB,INVOICE,T,,0,1,S,N,N,7600,2800,4800,8700,1600,980000,1600000,25,1,Y,N,0,CS """ SAMPLE_EXPLAIN_BASELINE = """ stmt_id,collection,package,program,statement_type,creator,table_name,access_type,access_name,matchcols,method,prefetch,sortn_orderby,sortn_groupby,total_cost,cpu_cost,io_cost,elapsed_ms,wait_ms,getpages,rows_examined,rows_returned,degree,stage2,rid_pool,workfile,lock_mode 1,PAYCOLL,PAYPKG1,PAYAPI,SELECT,PRODDB,CUSTOMER,I,IX_CUSTOMER_STATUS_DATE,2,1,N,N,N,1800,700,1100,1200,250,180000,240000,250,1,N,N,0,CS 2,PAYCOLL,PAYPKG1,PAYAPI,SELECT,PRODDB,ACCOUNT,I,IX_ACCOUNT_STATUS,2,1,N,N,N,2400,900,1500,2600,300,220000,250000,100,1,N,N,0,CS 3,PAYCOLL,PAYPKG2,PAYBATCH,SELECT,PRODDB,PAYMENT,I,IX_PAYMENT_DATE,2,2,N,N,N,1700,650,1050,2100,500,150000,260000,5000,1,N,N,0,CS 4,TRNCOLL,TRNPKG1,TRNAPI,SELECT,PRODDB,TRANSACTION,I,IX_TRANSACTION_CUST_DATE,2,1,N,N,N,2700,900,1800,1900,300,260000,300000,200,1,N,N,0,CS 5,AUDCOLL,AUDPKG1,AUDREP,SELECT,PRODDB,AUDIT_LOG,I,IX_AUDIT_DATE,1,1,N,N,N,3200,1300,1900,5200,1000,310000,1000000,10000,ANY,N,N,0,CS 8,FINCOLL,FINPKG1,FINRPT,SELECT,FINANCE,LEDGER,I,IX_LEDGER_PERIOD,2,2,N,N,Y,7500,2500,5000,22000,2000,1600000,3500000,900000,ANY,N,N,22000,CS """ SAMPLE_SQL = """ SELECT * FROM PRODDB.CUSTOMER C WHERE DATE(C.CREATE_TS) = :HV_DATE AND C.STATUS = :HV_STATUS ORDER BY C.CREATE_TS DESC; SELECT A.ACCOUNT_ID, A.CUSTOMER_ID, A.BALANCE FROM PRODDB.ACCOUNT A WHERE A.STATUS = :HV_STATUS OR A.BRANCH_CODE = :HV_BRANCH; SELECT P.PAYMENT_ID, P.CUSTOMER_ID, SUM(P.AMOUNT) FROM PRODDB.PAYMENT P WHERE P.PAYMENT_DATE BETWEEN :D1 AND :D2 GROUP BY P.PAYMENT_ID, P.CUSTOMER_ID; SELECT T.TRAN_ID, T.CUSTOMER_ID, T.AMOUNT FROM PRODDB.TRANSACTION T WHERE UPPER(T.CHANNEL) = 'WEB' AND T.TRAN_DATE >= :D1; UPDATE PRODDB.CUSTOMER SET STATUS = 'A'; DELETE FROM PRODDB.AUDIT_LOG; SELECT L.* FROM FINANCE.LEDGER L WHERE L.PERIOD_ID = :PERIOD ORDER BY L.ACCOUNT_ID; """ SAMPLE_PREDICATES = """ stmt_id,creator,table_name,column_name,predicate_text,predicate_type,stage,indexable,filter_factor 1,PRODDB,CUSTOMER,CREATE_TS,DATE(CREATE_TS) = :HV_DATE,EQUAL,2,N,0.0500 1,PRODDB,CUSTOMER,STATUS,STATUS = :HV_STATUS,EQUAL,1,Y,0.2000 2,PRODDB,ACCOUNT,STATUS,STATUS = :HV_STATUS,EQUAL,1,Y,0.4000 2,PRODDB,ACCOUNT,BRANCH_CODE,BRANCH_CODE = :HV_BRANCH,EQUAL,1,Y,0.3000 3,PRODDB,PAYMENT,PAYMENT_DATE,PAYMENT_DATE BETWEEN :D1 AND :D2,RANGE,1,Y,0.0800 4,PRODDB,TRANSACTION,CHANNEL,UPPER(CHANNEL) = 'WEB',EQUAL,2,N,0.1500 4,PRODDB,TRANSACTION,TRAN_DATE,TRAN_DATE >= :D1,RANGE,1,Y,0.1000 5,PRODDB,AUDIT_LOG,EVENT_DATE,EVENT_DATE >= :D1,RANGE,1,Y,0.3000 8,FINANCE,LEDGER,PERIOD_ID,PERIOD_ID = :PERIOD,EQUAL,1,Y,0.0500 11,PRODDB,ORDERS,STATUS,STATUS = :STATUS,EQUAL,1,Y,0.5000 11,PRODDB,ORDERS,ORDER_DATE,ORDER_DATE >= :D1,RANGE,1,Y,0.1200 12,PRODDB,INVOICE,INVOICE_DATE,YEAR(INVOICE_DATE)=:YEAR,EQUAL,2,N,0.2500 """ SAMPLE_INDEXES = """ creator,table_name,index_schema,index_name,columns,unique,clustering,clustering_ratio,nlevels,firstkeycard,fullkeycard,lastused PRODDB,CUSTOMER,PRODDB,IX_CUSTOMER_ID,CUSTOMER_ID,Y,Y,98,3,1200000,2400000,2026-05-01 PRODDB,CUSTOMER,PRODDB,IX_CUSTOMER_STATUS_DATE,STATUS CREATE_TS,N,N,42,5,5,1200000,2026-02-01 PRODDB,ACCOUNT,PRODDB,IX_ACCOUNT_STATUS,STATUS,N,N,60,4,4,120000,2026-05-03 PRODDB,PAYMENT,PRODDB,IX_PAYMENT_DATE,PAYMENT_DATE CUSTOMER_ID,N,Y,91,3,900,2000000,2026-05-10 PRODDB,TRANSACTION,PRODDB,IX_TRANSACTION_CUST_DATE,CUSTOMER_ID TRAN_DATE,N,N,75,4,1800000,5000000,2026-04-15 PRODDB,AUDIT_LOG,PRODDB,IX_AUDIT_DATE,EVENT_DATE,N,N,49,5,2000,12000000,2026-01-15 FINANCE,LEDGER,FINANCE,IX_LEDGER_PERIOD,PERIOD_ID ACCOUNT_ID,N,Y,85,4,48,8000000,2026-05-12 PRODDB,ORDERS,PRODDB,IX_ORDERS_STATUS_DATE,STATUS ORDER_DATE,N,N,55,5,8,900000,2026-04-21 PRODDB,COUNTRY,PRODDB,IX_COUNTRY_PK,COUNTRY_CODE,Y,Y,100,2,250,250,2026-05-20 """ SAMPLE_TABLE_STATS = """ creator,table_name,cardinality,npages,stats_date,partitions,volatile,compressed_pct,organization PRODDB,CUSTOMER,2400000,180000,2026-01-15,32,N,68,BASE PRODDB,ACCOUNT,950000,72000,2026-04-02,16,N,50,BASE PRODDB,PAYMENT,2800000,210000,2026-05-01,64,N,72,BASE PRODDB,TRANSACTION,5000000,520000,2025-12-15,128,N,80,BASE PRODDB,AUDIT_LOG,12000000,900000,2026-01-01,256,Y,90,BASE PRODDB,CUSTOMER_API,600000,55000,2026-03-01,8,N,35,BASE FINANCE,LEDGER,12000000,980000,2026-02-01,128,N,88,BASE PAYROLL,EMPLOYEE,700000,65000,2026-04-30,12,N,20,BASE PRODDB,ORDERS,900000,90000,2026-04-01,32,N,65,BASE PRODDB,INVOICE,1600000,140000,2025-11-15,64,N,40,BASE """ SAMPLE_PACKAGES = """ collection,package,program,bind_time,isolation,release,reopt,degree,currentdata,validate,dynamicrules,qualifier PAYCOLL,PAYPKG1,PAYAPI,2026-05-18 01:02:00,CS,COMMIT,NONE,1,NO,BIND,RUN,PRODDB PAYCOLL,PAYPKG2,PAYBATCH,2026-05-18 01:03:00,CS,DEALLOCATE,NONE,1,NO,BIND,RUN,PRODDB PAYCOLL,PAYPKG3,PAYUPD,2026-05-18 01:04:00,RR,COMMIT,NONE,1,YES,RUN,RUN,PRODDB AUDCOLL,AUDPKG1,AUDREP,2026-05-18 01:05:00,RS,DEALLOCATE,NONE,ANY,YES,BIND,RUN,PRODDB FINCOLL,FINPKG1,FINRPT,2026-05-18 01:07:00,CS,COMMIT,NONE,ANY,NO,BIND,RUN,FINANCE """ SAMPLE_POLICY = { "critical_tables": ["PRODDB.CUSTOMER", "PRODDB.ACCOUNT", "PRODDB.TRANSACTION", "FINANCE.LEDGER"], "sensitive_columns": ["SSN", "CARD", "ACCOUNT", "BALANCE", "SALARY", "CUSTOMER_ID"], "high_cost": 1200, "critical_cost": 6000, "regression_pct": 20, "critical_regression_pct": 90, } if __name__ == "__main__": sys.exit(main())