feat(harness): port backend Onda 0 — vendor, lshindex, sqlcheck, execute, rest/exec, ctetest, report, datamart
8 moduli leaf portati verbatim da ChironeWp3 con rename psdwp3→tht: - vendor/thoth_lsh (MinHash/LSH, leaf puro datasketch+tqdm) + VENDORED.md - lshindex/ (build/save/load/query, dipende vendor + LshConfig) - sqlcheck/ (validate_sql, leaf ExecutionConfig+mschema) - execute/ + execute/warnings (run_controlled/explain, leaf sqlglot+sqlalchemy) - rest/execute + rest/explain (REST variants, dipendono execute+rest.client) - ctetest (CTE test records, leaf sqlglot+pydantic) - report (validation report rendering, dipende execute+sqlcheck) - datamart (stub NotImplementedError) Verifica: import smoke catena completa OK, pytest 109 passed. Deps (datasketch, sqlglot, sqlalchemy, pydantic, requests, tqdm) già in pyproject. VENDORED.md neutralizzato (riferimenti PsdWp3→Thoth).
This commit is contained in:
@@ -0,0 +1,101 @@
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
import sqlglot
|
||||
from pydantic import BaseModel
|
||||
from sqlglot import exp
|
||||
|
||||
CTE_TESTS_FILE = "cte_tests.json"
|
||||
PROBE = "\nSELECT * FROM __psd_probe__"
|
||||
|
||||
|
||||
class CteError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class CteTestRecord(BaseModel):
|
||||
name: str
|
||||
ts: datetime
|
||||
sql_hash: str
|
||||
status: Literal["ok", "error"]
|
||||
columns: list[str] = []
|
||||
row_sample: int = 0
|
||||
execution_ms: int = 0
|
||||
warnings: list[str] = []
|
||||
error: str | None = None
|
||||
|
||||
|
||||
def has_trailing_select(cte_sql: str) -> bool:
|
||||
"""True se il file, parsato da solo, e' gia' una query completa con SELECT
|
||||
finale dopo il blocco WITH (violazione della regola: il file deve contenere
|
||||
solo WITH ... AS (...)).
|
||||
|
||||
I file CTE corretti contengono solo 'WITH ... AS (...)': sqlglot.parse_one()
|
||||
fallisce su di loro perche' manca la SELECT obbligatoria.
|
||||
Un file con SELECT finale parsa invece correttamente come Select con un nodo With.
|
||||
I file con errori di sintassi falliscono parse_one e ritornano False (no hint)."""
|
||||
try:
|
||||
ast = sqlglot.parse_one(cte_sql, read="postgres")
|
||||
except sqlglot.errors.ParseError:
|
||||
return False
|
||||
return ast is not None and ast.find(exp.With) is not None
|
||||
|
||||
|
||||
def last_cte_name(cte_sql: str) -> str:
|
||||
"""Nome dell'ultimo CTE di un blocco 'WITH a AS (...), b AS (...)' senza SELECT
|
||||
finale. Trucco AV-SQL: si appende una SELECT fittizia per rendere parsabile."""
|
||||
if not cte_sql.lstrip().upper().startswith("WITH"):
|
||||
raise CteError("il file deve contenere un blocco WITH ... AS (...)")
|
||||
try:
|
||||
ast = sqlglot.parse_one(cte_sql + PROBE, read="postgres")
|
||||
except sqlglot.errors.ParseError as e:
|
||||
raise CteError(f"CTE non parsabile: {e}") from e
|
||||
with_clause = ast.find(exp.With)
|
||||
if with_clause is None or not with_clause.expressions:
|
||||
raise CteError("il file deve contenere un blocco WITH ... AS (...)")
|
||||
return with_clause.expressions[-1].alias
|
||||
|
||||
|
||||
def build_test_sql(cte_sql: str) -> str:
|
||||
"""SQL completo per il test: il blocco WITH + SELECT * dall'ultimo CTE.
|
||||
Il LIMIT lo inietta run_controlled."""
|
||||
name = last_cte_name(cte_sql)
|
||||
return f"{cte_sql.rstrip().rstrip(',')}\nSELECT * FROM {name}"
|
||||
|
||||
|
||||
def _iter_json_objects(text: str):
|
||||
"""Estrae gli oggetti JSON di primo livello da `text`, qualunque sia il
|
||||
formato: array JSON, JSONL, o misto (array + righe JSONL appese). Salta
|
||||
spazi, virgole e parentesi d'array fra un oggetto e il successivo."""
|
||||
dec = json.JSONDecoder()
|
||||
i, n = 0, len(text)
|
||||
while i < n:
|
||||
if text[i] in " \t\r\n,[]":
|
||||
i += 1
|
||||
continue
|
||||
obj, end = dec.raw_decode(text, i)
|
||||
yield obj
|
||||
i = end
|
||||
|
||||
|
||||
def load_cte_tests(session_dir: Path) -> list[CteTestRecord]:
|
||||
path = session_dir / CTE_TESTS_FILE
|
||||
if not path.exists():
|
||||
return []
|
||||
text = path.read_text()
|
||||
try:
|
||||
objs = list(_iter_json_objects(text))
|
||||
except json.JSONDecodeError as e:
|
||||
raise CteError(f"{CTE_TESTS_FILE} corrotto: {e}") from e
|
||||
return [CteTestRecord.model_validate(o) for o in objs]
|
||||
|
||||
|
||||
def append_cte_test(session_dir: Path, record: CteTestRecord) -> None:
|
||||
"""Append atomico (una riga JSON per esito): scritture concorrenti sulla
|
||||
stessa sessione non si sovrascrivono, a differenza del rewrite dell'array."""
|
||||
path = session_dir / CTE_TESTS_FILE
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with path.open("a", encoding="utf-8") as f:
|
||||
f.write(record.model_dump_json() + "\n")
|
||||
@@ -0,0 +1,26 @@
|
||||
"""Fase 8 — generazione dbt del datamart (hook, non ancora implementato).
|
||||
|
||||
Punto d'innesto unico per la generazione del modello dbt da aggiungere al flusso
|
||||
ETL di creazione dei datamart. La logica vera verra' sviluppata in seguito: per
|
||||
ora qui vive solo la firma della funzione (il "hook") con le due varianti del
|
||||
paziente.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
# Le due varianti previste per il paziente nel datamart:
|
||||
# - "clear": nome, cognome, ecc. in chiaro
|
||||
# - "pseudonymized": nome, cognome, ecc. mascherati (pseudoanonimizzati)
|
||||
DATAMART_MODES = ("clear", "pseudonymized")
|
||||
|
||||
|
||||
def generate_dbt_datamart(session_dir: Path, mode: str) -> None:
|
||||
"""Genera il modello dbt del datamart per la sessione (HOOK — stub).
|
||||
|
||||
`mode` sceglie la variante del paziente (vedi DATAMART_MODES). Non ancora
|
||||
implementato: solleva NotImplementedError, intercettata dalla CLI che mostra
|
||||
un messaggio pulito senza interrompere il workflow.
|
||||
"""
|
||||
raise NotImplementedError(
|
||||
"Generazione dbt del datamart non ancora implementata"
|
||||
)
|
||||
@@ -0,0 +1,102 @@
|
||||
import json
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
|
||||
import sqlglot
|
||||
from sqlalchemy import Engine, text
|
||||
from sqlalchemy.exc import DBAPIError
|
||||
|
||||
|
||||
class ExecutionError(Exception):
|
||||
"""Errore di esecuzione con messaggio leggibile per il reviewer."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExecResult:
|
||||
columns: list[str]
|
||||
rows: list[tuple]
|
||||
execution_ms: int
|
||||
truncated: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class PlanSummary:
|
||||
total_cost: float
|
||||
plan_rows: int
|
||||
node_types: list[str]
|
||||
|
||||
|
||||
def _inject_limit(sql: str, limit: int) -> tuple[str, bool]:
|
||||
"""Aggiunge LIMIT limit+1 se assente (il +1 serve a rilevare il troncamento).
|
||||
Se la query ha gia' un suo LIMIT, lo si rispetta."""
|
||||
ast = sqlglot.parse_one(sql, read="postgres")
|
||||
# solo le query (SELECT/UNION & co.) accettano un LIMIT: per qualunque altro
|
||||
# statement non si inietta nulla e la transazione READ ONLY lo rifiutera' (rete 2).
|
||||
if not isinstance(ast, sqlglot.exp.Query) or ast.args.get("limit") is not None:
|
||||
return sql, False
|
||||
return ast.limit(limit + 1).sql(dialect="postgres"), True
|
||||
|
||||
|
||||
def _translate_error(e: DBAPIError) -> ExecutionError:
|
||||
msg = str(e.orig)
|
||||
if "statement timeout" in msg or "canceling statement" in msg:
|
||||
return ExecutionError("timeout: la query ha superato statement_timeout_ms")
|
||||
if "read-only" in msg:
|
||||
return ExecutionError("rifiutato: statement non eseguibile in transazione read-only")
|
||||
return ExecutionError(f"errore SQL: {msg}")
|
||||
|
||||
|
||||
def run_controlled(engine: Engine, sql: str, *, limit: int, timeout_ms: int) -> ExecResult:
|
||||
"""Esecuzione nelle 4 reti: utente RO (a monte), transazione READ ONLY,
|
||||
statement_timeout, LIMIT iniettato via AST."""
|
||||
final_sql, injected = _inject_limit(sql, limit)
|
||||
start = time.monotonic()
|
||||
with engine.connect() as conn:
|
||||
trans = conn.begin()
|
||||
try:
|
||||
conn.execute(text("SET TRANSACTION READ ONLY"))
|
||||
conn.execute(text(f"SET LOCAL statement_timeout = {int(timeout_ms)}"))
|
||||
result = conn.execute(text(final_sql))
|
||||
columns = list(result.keys())
|
||||
rows = [tuple(r) for r in result.fetchall()]
|
||||
except DBAPIError as e:
|
||||
raise _translate_error(e) from e
|
||||
finally:
|
||||
trans.rollback()
|
||||
elapsed_ms = int((time.monotonic() - start) * 1000)
|
||||
truncated = injected and len(rows) > limit
|
||||
return ExecResult(
|
||||
columns=columns, rows=rows[:limit], execution_ms=elapsed_ms, truncated=truncated
|
||||
)
|
||||
|
||||
|
||||
def explain(engine: Engine, sql: str, *, timeout_ms: int) -> PlanSummary:
|
||||
"""EXPLAIN (FORMAT JSON), mai ANALYZE: il piano si stima, non si esegue."""
|
||||
with engine.connect() as conn:
|
||||
trans = conn.begin()
|
||||
try:
|
||||
conn.execute(text("SET TRANSACTION READ ONLY"))
|
||||
conn.execute(text(f"SET LOCAL statement_timeout = {int(timeout_ms)}"))
|
||||
row = conn.execute(text(f"EXPLAIN (FORMAT JSON) {sql}")).fetchone()
|
||||
except DBAPIError as e:
|
||||
raise _translate_error(e) from e
|
||||
finally:
|
||||
trans.rollback()
|
||||
payload = row[0]
|
||||
if isinstance(payload, str):
|
||||
payload = json.loads(payload)
|
||||
plan = payload[0]["Plan"]
|
||||
|
||||
node_types: list[str] = []
|
||||
|
||||
def collect(node: dict) -> None:
|
||||
node_types.append(node.get("Node Type", "?"))
|
||||
for child in node.get("Plans", []):
|
||||
collect(child)
|
||||
|
||||
collect(plan)
|
||||
return PlanSummary(
|
||||
total_cost=float(plan.get("Total Cost", 0.0)),
|
||||
plan_rows=int(plan.get("Plan Rows", 0)),
|
||||
node_types=node_types,
|
||||
)
|
||||
@@ -0,0 +1,40 @@
|
||||
from sqlglot import exp
|
||||
|
||||
from tht.config import ExecutionConfig
|
||||
from tht.execute import ExecResult, PlanSummary
|
||||
|
||||
|
||||
def runtime_warnings(result: ExecResult, cfg: ExecutionConfig) -> list[str]:
|
||||
warnings: list[str] = []
|
||||
if not result.rows:
|
||||
warnings.append("risultato vuoto: verificare filtri e join")
|
||||
if result.truncated:
|
||||
warnings.append("righe = limit: risultato troncato dalla preview")
|
||||
if result.execution_ms > cfg.warn_execution_ms:
|
||||
warnings.append(
|
||||
f"tempo di esecuzione elevato: {result.execution_ms} ms "
|
||||
f"(soglia {cfg.warn_execution_ms} ms)"
|
||||
)
|
||||
return warnings
|
||||
|
||||
|
||||
def plan_warnings(plan: PlanSummary, cfg: ExecutionConfig) -> list[str]:
|
||||
warnings: list[str] = []
|
||||
if plan.plan_rows > cfg.warn_plan_rows:
|
||||
warnings.append(
|
||||
f"righe stimate dal piano: {plan.plan_rows} (soglia {cfg.warn_plan_rows}) — "
|
||||
f"possibile join molto ampio o cardinalita' anomala"
|
||||
)
|
||||
return warnings
|
||||
|
||||
|
||||
def static_warnings(ast: exp.Expression) -> list[str]:
|
||||
"""Euristica cast sospetti: CAST su colonne dentro WHERE (non-sargable)."""
|
||||
warnings: list[str] = []
|
||||
for where in ast.find_all(exp.Where):
|
||||
for cast in where.find_all(exp.Cast):
|
||||
if cast.find(exp.Column) is not None:
|
||||
warnings.append(
|
||||
f"cast su colonna nel WHERE (possibile non-sargable): {cast.sql()}"
|
||||
)
|
||||
return warnings
|
||||
@@ -0,0 +1,77 @@
|
||||
import json
|
||||
import pickle
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from datasketch import MinHash, MinHashLSH
|
||||
|
||||
from tht.config import LshConfig
|
||||
from tht.vendor.thoth_lsh import create_lsh_index, create_minhash
|
||||
|
||||
|
||||
class LshIndexError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class LshHit:
|
||||
table: str
|
||||
column: str
|
||||
value: str
|
||||
score: float
|
||||
|
||||
|
||||
def build_index(
|
||||
values: dict[str, dict[str, list[str]]], cfg: LshConfig, verbose: bool = False
|
||||
) -> tuple[MinHashLSH, dict[str, tuple[MinHash, str, str, str]]]:
|
||||
return create_lsh_index(
|
||||
values, signature_size=cfg.signature_size, n_gram=cfg.n_gram,
|
||||
threshold=cfg.threshold, verbose=verbose,
|
||||
)
|
||||
|
||||
|
||||
def _paths(directory: Path, name: str) -> tuple[Path, Path, Path]:
|
||||
return (
|
||||
directory / f"{name}_lsh.pkl",
|
||||
directory / f"{name}_minhashes.pkl",
|
||||
directory / f"{name}_meta.json",
|
||||
)
|
||||
|
||||
|
||||
def save_index(lsh, minhashes, cfg: LshConfig, directory: Path, name: str) -> None:
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
lsh_path, mh_path, meta_path = _paths(directory, name)
|
||||
lsh_path.write_bytes(pickle.dumps(lsh))
|
||||
mh_path.write_bytes(pickle.dumps(minhashes))
|
||||
meta_path.write_text(
|
||||
json.dumps(
|
||||
{"signature_size": cfg.signature_size, "n_gram": cfg.n_gram,
|
||||
"threshold": cfg.threshold, "entries": len(minhashes)},
|
||||
indent=2,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def load_index(directory: Path, name: str) -> tuple[Any, Any, dict]:
|
||||
lsh_path, mh_path, meta_path = _paths(directory, name)
|
||||
if not (lsh_path.exists() and mh_path.exists() and meta_path.exists()):
|
||||
raise LshIndexError(
|
||||
f"Indice LSH non trovato in {directory} (atteso {name}_lsh.pkl). "
|
||||
f"Esegui prima `nsp lsh build`."
|
||||
)
|
||||
lsh = pickle.loads(lsh_path.read_bytes())
|
||||
minhashes = pickle.loads(mh_path.read_bytes())
|
||||
meta = json.loads(meta_path.read_text())
|
||||
return lsh, minhashes, meta
|
||||
|
||||
|
||||
def query_index(lsh, minhashes, keyword: str, meta: dict, top_n: int = 10) -> list[LshHit]:
|
||||
"""Query con score: i parametri MinHash vengono dal meta dell'indice, non dalla config."""
|
||||
qmh = create_minhash(meta["signature_size"], keyword, meta["n_gram"])
|
||||
scored = [(key, qmh.jaccard(minhashes[key][0])) for key in lsh.query(qmh)]
|
||||
scored.sort(key=lambda kv: kv[1], reverse=True)
|
||||
return [
|
||||
LshHit(table=minhashes[k][1], column=minhashes[k][2], value=minhashes[k][3], score=s)
|
||||
for k, s in scored[:top_n]
|
||||
]
|
||||
@@ -0,0 +1,74 @@
|
||||
from tht.execute import ExecResult, PlanSummary
|
||||
from tht.sqlcheck import CheckResult
|
||||
|
||||
NOTE_HEADING = "## Note del reviewer"
|
||||
|
||||
|
||||
def _markdown_table(columns: list[str], rows: list[tuple]) -> str:
|
||||
lines = [
|
||||
"| " + " | ".join(columns) + " |",
|
||||
"|" + "---|" * len(columns),
|
||||
]
|
||||
for row in rows:
|
||||
lines.append("| " + " | ".join(str(v) for v in row) + " |")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def render_validation_report(
|
||||
*,
|
||||
session_id: str,
|
||||
check: CheckResult,
|
||||
plan: PlanSummary,
|
||||
plan_warnings: list[str],
|
||||
result: ExecResult,
|
||||
runtime_warnings: list[str],
|
||||
static_warnings: list[str],
|
||||
limit: int,
|
||||
reviewer_notes: str = "",
|
||||
) -> str:
|
||||
"""Sezioni fattuali del validation report. La sezione note e' di Pi/reviewer:
|
||||
se gia' scritta, va passata in reviewer_notes per preservarla."""
|
||||
|
||||
def warn(ws: list[str]) -> str:
|
||||
return "\n".join(f"- warning: {w}" for w in ws) if ws else "- nessun warning"
|
||||
|
||||
parts = [
|
||||
f"# Validation report — {session_id}",
|
||||
"",
|
||||
"> Avvertenza: questo report contiene sample di dati del database.",
|
||||
"",
|
||||
"## Parsing e read-only",
|
||||
"",
|
||||
"- parsing: OK" if check.ok else "- parsing: ERRORI",
|
||||
"- statement read-only per struttura: OK" if check.ok else
|
||||
"\n".join(f"- ERRORE: {e}" for e in check.errors),
|
||||
warn(check.warnings + static_warnings),
|
||||
"",
|
||||
"## Piano di esecuzione",
|
||||
"",
|
||||
f"- costo totale stimato: {plan.total_cost}",
|
||||
f"- righe stimate: {plan.plan_rows}",
|
||||
f"- nodi: {', '.join(plan.node_types)}",
|
||||
warn(plan_warnings),
|
||||
"",
|
||||
"## Preview / aggregato",
|
||||
"",
|
||||
f"- tempo di esecuzione: {result.execution_ms} ms (limit {limit})",
|
||||
"",
|
||||
_markdown_table(result.columns, result.rows),
|
||||
"",
|
||||
warn(runtime_warnings),
|
||||
"",
|
||||
NOTE_HEADING,
|
||||
"",
|
||||
reviewer_notes.strip(),
|
||||
"",
|
||||
]
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
def extract_reviewer_notes(report_md: str) -> str:
|
||||
"""Estrae il contenuto della sezione note da un report esistente."""
|
||||
if NOTE_HEADING not in report_md:
|
||||
return ""
|
||||
return report_md.split(NOTE_HEADING, 1)[1].strip()
|
||||
@@ -0,0 +1,38 @@
|
||||
"""Esecuzione controllata e EXPLAIN via REST (rpc `run_query`/`explain_query`).
|
||||
|
||||
Produce gli stessi tipi del transport diretto (`ExecResult`, `PlanSummary`) e traduce
|
||||
`RestError` in `ExecutionError`, così le fasi 6/7 e la gestione errori della CLI non cambiano.
|
||||
|
||||
Differenze rispetto al diretto: read-only e statement_timeout sono garantiti lato server;
|
||||
il client mantiene solo l'iniezione del LIMIT (per il rilevamento del troncamento in preview).
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
from tht.execute import ExecResult, ExecutionError, PlanSummary, _inject_limit
|
||||
from tht.rest.client import RestError
|
||||
from tht.rest.explain import parse_text_plan
|
||||
|
||||
|
||||
def run_controlled_rest(client, sql: str, *, limit: int) -> ExecResult:
|
||||
final_sql, injected = _inject_limit(sql, limit)
|
||||
start = time.monotonic()
|
||||
try:
|
||||
rows_dicts = client.run_query(final_sql)
|
||||
except RestError as e:
|
||||
raise ExecutionError(str(e)) from e
|
||||
elapsed_ms = int((time.monotonic() - start) * 1000)
|
||||
columns = list(rows_dicts[0].keys()) if rows_dicts else []
|
||||
rows = [tuple(r.get(c) for c in columns) for r in rows_dicts]
|
||||
truncated = injected and len(rows) > limit
|
||||
return ExecResult(
|
||||
columns=columns, rows=rows[:limit], execution_ms=elapsed_ms, truncated=truncated
|
||||
)
|
||||
|
||||
|
||||
def explain_rest(client, sql: str) -> PlanSummary:
|
||||
try:
|
||||
lines = client.explain_query(sql)
|
||||
except RestError as e:
|
||||
raise ExecutionError(str(e)) from e
|
||||
return parse_text_plan(lines)
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Parsing del piano EXPLAIN testuale restituito dall'rpc `explain_query`.
|
||||
|
||||
Il transport diretto usa `EXPLAIN (FORMAT JSON)`; via REST l'rpc ritorna il piano come
|
||||
righe di testo (formato default di EXPLAIN). Questo modulo le riconduce a `PlanSummary`,
|
||||
così la logica a valle (`plan_warnings`, display) non cambia.
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
from tht.execute import ExecutionError, PlanSummary
|
||||
|
||||
# es. "(cost=0.00..3712.97 rows=13706 width=0)"
|
||||
_COST_RE = re.compile(r"\(cost=[\d.]+\.\.([\d.]+)\s+rows=(\d+)\s+width=\d+\)")
|
||||
|
||||
|
||||
def parse_text_plan(lines: list[str]) -> PlanSummary:
|
||||
node_types: list[str] = []
|
||||
total_cost = 0.0
|
||||
plan_rows = 0
|
||||
root_seen = False
|
||||
for line in lines:
|
||||
m = _COST_RE.search(line)
|
||||
if m is None:
|
||||
# riga di dettaglio (Filter, Index Cond, …): non è un nodo del piano
|
||||
continue
|
||||
label = line.split("(cost=")[0].replace("->", " ").strip()
|
||||
if label:
|
||||
node_types.append(label)
|
||||
if not root_seen:
|
||||
total_cost = float(m.group(1))
|
||||
plan_rows = int(m.group(2))
|
||||
root_seen = True
|
||||
if not root_seen:
|
||||
raise ExecutionError("piano EXPLAIN vuoto o non interpretabile")
|
||||
return PlanSummary(total_cost=total_cost, plan_rows=plan_rows, node_types=node_types)
|
||||
@@ -0,0 +1,112 @@
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import sqlglot
|
||||
from sqlglot import exp
|
||||
|
||||
from tht.config import ExecutionConfig
|
||||
from tht.mschema.models import PhysicalSchema
|
||||
|
||||
DIALECT = "postgres"
|
||||
|
||||
FORBIDDEN_NODES: dict[type, str] = {
|
||||
exp.Insert: "INSERT",
|
||||
exp.Update: "UPDATE",
|
||||
exp.Delete: "DELETE",
|
||||
exp.Create: "CREATE",
|
||||
exp.Drop: "DROP",
|
||||
exp.Alter: "ALTER",
|
||||
exp.Merge: "MERGE",
|
||||
exp.TruncateTable: "TRUNCATE",
|
||||
exp.Grant: "GRANT",
|
||||
exp.Command: "comando non-SELECT", # CALL e altri comandi opachi
|
||||
}
|
||||
|
||||
DEFAULT_FORBIDDEN_FUNCTIONS = set(ExecutionConfig().forbidden_functions)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckResult:
|
||||
ok: bool
|
||||
errors: list[str] = field(default_factory=list)
|
||||
warnings: list[str] = field(default_factory=list)
|
||||
ast: exp.Expression | None = None
|
||||
|
||||
|
||||
def validate_sql(
|
||||
sql: str,
|
||||
physical: PhysicalSchema | None = None,
|
||||
promoted_tables: set[str] | None = None,
|
||||
forbidden_functions: set[str] | None = None,
|
||||
) -> CheckResult:
|
||||
"""Validazione statica: parsabile, un solo statement, read-only per struttura,
|
||||
nessuna funzione in blacklist, oggetti esistenti in mschema (se fornito)."""
|
||||
errors: list[str] = []
|
||||
warnings: list[str] = []
|
||||
try:
|
||||
statements = sqlglot.parse(sql, read=DIALECT)
|
||||
except sqlglot.errors.ParseError as e:
|
||||
return CheckResult(ok=False, errors=[f"SQL non parsabile: {e}"])
|
||||
|
||||
statements = [s for s in statements if s is not None]
|
||||
if len(statements) != 1:
|
||||
return CheckResult(
|
||||
ok=False,
|
||||
errors=[f"atteso uno solo statement, trovati {len(statements)}"],
|
||||
)
|
||||
ast = statements[0]
|
||||
|
||||
# query read-only legittime: SELECT, WITH...SELECT e set operation (UNION & co.)
|
||||
allowed_roots = (exp.Select, exp.Union, exp.Intersect, exp.Except)
|
||||
if not isinstance(ast, allowed_roots):
|
||||
errors.append(
|
||||
f"solo query SELECT (incluse WITH e UNION) sono ammesse "
|
||||
f"(trovato: {type(ast).__name__})"
|
||||
)
|
||||
|
||||
for node in ast.walk():
|
||||
for forbidden_type, label in FORBIDDEN_NODES.items():
|
||||
if isinstance(node, forbidden_type):
|
||||
errors.append(f"statement vietato (read-only): {label}")
|
||||
|
||||
blacklist = forbidden_functions if forbidden_functions is not None else DEFAULT_FORBIDDEN_FUNCTIONS
|
||||
for func in ast.find_all(exp.Func):
|
||||
name = (func.name or "").lower()
|
||||
if name in blacklist:
|
||||
errors.append(f"funzione vietata: {name}")
|
||||
|
||||
if physical is not None:
|
||||
_check_objects(ast, physical, promoted_tables, errors, warnings)
|
||||
|
||||
return CheckResult(ok=not errors, errors=errors, warnings=warnings, ast=ast)
|
||||
|
||||
|
||||
def _check_objects(
|
||||
ast: exp.Expression,
|
||||
physical: PhysicalSchema,
|
||||
promoted_tables: set[str] | None,
|
||||
errors: list[str],
|
||||
warnings: list[str],
|
||||
) -> None:
|
||||
"""Tabelle citate: devono esistere (escluse le CTE definite nella query).
|
||||
Colonne qualificate su tabelle reali (o loro alias): devono esistere.
|
||||
Colonne non qualificate o su CTE: non verificabili staticamente, si saltano."""
|
||||
cte_names = {cte.alias_or_name for cte in ast.find_all(exp.CTE)}
|
||||
alias_to_table: dict[str, str] = {}
|
||||
for table in ast.find_all(exp.Table):
|
||||
name = table.name
|
||||
if name in cte_names:
|
||||
continue
|
||||
if name not in physical.tables:
|
||||
errors.append(f"tabella inesistente nello schema: {name}")
|
||||
continue
|
||||
alias_to_table[table.alias_or_name] = name
|
||||
if promoted_tables is not None and name not in promoted_tables:
|
||||
warnings.append(f"tabella fuori dal perimetro promosso: {name}")
|
||||
|
||||
for column in ast.find_all(exp.Column):
|
||||
qualifier = column.table
|
||||
if not qualifier or qualifier not in alias_to_table:
|
||||
continue
|
||||
table_name = alias_to_table[qualifier]
|
||||
if column.name not in physical.tables[table_name].columns:
|
||||
errors.append(f"colonna inesistente: {table_name}.{column.name}")
|
||||
Vendored
+30
@@ -0,0 +1,30 @@
|
||||
# Codice vendorizzato
|
||||
|
||||
## thoth_lsh.py
|
||||
|
||||
- Origine: `thoth_sqldb2` (pacchetto PyPI `thoth-dbmanager` 0.7.4), file `thoth_dbmanager/lsh/core.py`
|
||||
- Copyright 2025 Marco Pancotti — Apache License 2.0
|
||||
- Modifiche locali:
|
||||
1. rimosso blocco di debug su colonna "doctype" in `create_lsh_index`;
|
||||
2. `skip_column`: soglie parametrizzate (`max_total_chars`, `max_avg_length`) con default
|
||||
identici agli originali (50000 / 20);
|
||||
3. rimossa `query_lsh_index` (sostituita da `tht.lshindex.query_index`, che restituisce
|
||||
anche gli score per il ranking spiegabile);
|
||||
4. try/except generico rimosso da `create_lsh_index`: gli errori devono emergere.
|
||||
5. `skip_column`: la guardia "name-like" (originariamente solo `"name"`) è stata estesa
|
||||
a token bilingui EN/IT tramite la costante `NAME_LIKE_TOKENS`
|
||||
(`name`, `nome`, `nominativ`, `denominazion`, `ragione_sociale`), sovrascrivibile col
|
||||
parametro `name_tokens`. Necessario perché il datawarehouse Chirone è in italiano e le
|
||||
colonne `nome`/`cognome`/`denominazione`/`ragione_sociale` non venivano riconosciute
|
||||
dalla sola parola inglese `name`.
|
||||
|
||||
**Nota: `skip_column` e `NAME_LIKE_TOKENS` non sono più usati dal flusso Thoth.**
|
||||
La selezione delle colonne da indicizzare è ora governata dal *principio di eleggibilità
|
||||
delle colonne* (spec: `docs/superpowers/specs/2026-06-13-tht-column-eligibility-principle.md`).
|
||||
L'esclusione dei testi larghi avviene a monte tramite il flag `eligible` persistito in
|
||||
`physical.yaml` (impostato da `nsp schema introspect`), non tramite l'euristica sulla
|
||||
lunghezza del vendorizzato. Le funzioni restano nel file per fedeltà alla sorgente upstream;
|
||||
Thoth ha semplicemente smesso di importarle (vedi `tht/db/sampling.py`).
|
||||
|
||||
Le query di introspezione in `tht/db/introspect.py` sono ADATTATE (non copiate verbatim)
|
||||
da `thoth_dbmanager/adapters/postgresql.py` della stessa versione.
|
||||
Vendored
Vendored
+82
@@ -0,0 +1,82 @@
|
||||
# Vendored from thoth_sqldb2 (thoth-dbmanager 0.7.4) — lsh/core.py
|
||||
# Copyright 2025 Marco Pancotti — Apache License 2.0
|
||||
# Modifiche locali documentate in VENDORED.md.
|
||||
"""Core LSH (MinHash) per la ricerca di valori simili nei campi del database."""
|
||||
|
||||
import logging
|
||||
from typing import Dict, List, Tuple
|
||||
|
||||
from datasketch import MinHash, MinHashLSH
|
||||
from tqdm import tqdm
|
||||
|
||||
|
||||
def create_minhash(signature_size: int, string: str, n_gram: int) -> MinHash:
|
||||
m = MinHash(num_perm=signature_size)
|
||||
for d in [string[i : i + n_gram] for i in range(len(string) - n_gram + 1)]:
|
||||
m.update(d.encode("utf8"))
|
||||
return m
|
||||
|
||||
|
||||
# Token che, se presenti nel nome di una colonna, la marcano come "name-like":
|
||||
# nomi propri di persone/enti, i cui valori sono utili al value-matching e quindi
|
||||
# mai da escludere dall'indice LSH. Bilingue EN/IT (modifica locale, vedi VENDORED.md).
|
||||
# Per substring coprono le forme flesse: "nome" -> cognome/soprannome,
|
||||
# "name" -> surname/username, "denominazion" -> denominazione, ecc.
|
||||
NAME_LIKE_TOKENS: tuple[str, ...] = (
|
||||
"name",
|
||||
"nome",
|
||||
"nominativ",
|
||||
"denominazion",
|
||||
"ragione_sociale",
|
||||
)
|
||||
|
||||
|
||||
def skip_column(
|
||||
column_name: str,
|
||||
column_values: List[str],
|
||||
max_total_chars: int = 50000,
|
||||
max_avg_length: int = 20,
|
||||
name_tokens: tuple[str, ...] = NAME_LIKE_TOKENS,
|
||||
) -> bool:
|
||||
lowered = column_name.lower()
|
||||
if any(token in lowered for token in name_tokens):
|
||||
return False
|
||||
sum_of_lengths = sum(len(value) for value in column_values)
|
||||
average_length = sum_of_lengths / len(column_values)
|
||||
return (sum_of_lengths > max_total_chars) and (average_length > max_avg_length)
|
||||
|
||||
|
||||
def jaccard_similarity(m1: MinHash, m2: MinHash) -> float:
|
||||
return m1.jaccard(m2)
|
||||
|
||||
|
||||
def create_lsh_index(
|
||||
unique_values: Dict[str, Dict[str, List[str]]],
|
||||
signature_size: int,
|
||||
n_gram: int,
|
||||
threshold: float,
|
||||
verbose: bool = True,
|
||||
) -> Tuple[MinHashLSH, Dict[str, Tuple[MinHash, str, str, str]]]:
|
||||
lsh = MinHashLSH(threshold=threshold, num_perm=signature_size)
|
||||
minhashes: Dict[str, Tuple[MinHash, str, str, str]] = {}
|
||||
total = sum(
|
||||
len(column_values)
|
||||
for table_values in unique_values.values()
|
||||
for column_values in table_values.values()
|
||||
)
|
||||
logging.info("Total unique values: %s", total)
|
||||
progress_bar = tqdm(total=total, desc="Creating LSH") if verbose else None
|
||||
|
||||
for table_name, table_values in unique_values.items():
|
||||
for column_name, column_values in table_values.items():
|
||||
for idx, value in enumerate(column_values):
|
||||
minhash = create_minhash(signature_size, value, n_gram)
|
||||
minhash_key = f"{table_name}_{column_name}_{idx}"
|
||||
minhashes[minhash_key] = (minhash, table_name, column_name, value)
|
||||
lsh.insert(minhash_key, minhash)
|
||||
if progress_bar:
|
||||
progress_bar.update(1)
|
||||
|
||||
if progress_bar:
|
||||
progress_bar.close()
|
||||
return lsh, minhashes
|
||||
Reference in New Issue
Block a user