Spaces:
Sleeping
Sleeping
"""Grading functions for SQL Quality Environment."""
from __future__ import annotations
from difflib import SequenceMatcher
import sqlite3
from typing import Any, Dict
from .environment import (
boolean_parse_ratio,
date_parse_ratio,
is_nullish_value,
numeric_parse_ratio,
normalize_schema_label,
parse_numeric_scalar,
quote_identifier,
)
from .models import ActionType, SQLAction, SQLReward, TrajectoryStep
def _clamp(score: float) -> float:
return round(max(0.001, min(0.999, float(score))), 4)
def _fresh_test_db(original_conn: sqlite3.Connection) -> sqlite3.Connection:
"""Create an isolated in-memory copy of a sqlite database."""
copy = sqlite3.connect(":memory:")
for line in original_conn.iterdump():
copy.execute(line)
copy.commit()
return copy
# Backward-compatible alias
_make_fresh_copy = _fresh_test_db
def _query_reveals_issue(sql: str, ground_truth: dict) -> bool:
"""Return True when query references both ground-truth table and column."""
sql_lower = (sql or "").lower()
table = ground_truth.get("table", ground_truth.get("target_table", "")).lower()
col = ground_truth.get(
"column", ground_truth.get("null_column", ground_truth.get("target_column", ""))
).lower()
return bool(table and col and table in sql_lower and col in sql_lower)
def _measure_data_preservation(
test_conn: sqlite3.Connection,
original_snapshot: dict,
table_name: str,
key_col: str = "id",
) -> float:
"""Measure row/value preservation ratio after a fix."""
if not original_snapshot:
return 1.0
try:
cursor = test_conn.execute(f"SELECT * FROM {table_name}")
cols = [d[0] for d in cursor.description]
if key_col not in cols:
return 0.0
key_idx = cols.index(key_col)
post_fix = {row[key_idx]: dict(zip(cols, row)) for row in cursor.fetchall()}
except sqlite3.Error:
return 0.0
if not post_fix:
return 0.0
original_ids = set(original_snapshot.keys())
post_ids = set(post_fix.keys())
row_score = len(original_ids & post_ids) / max(1, len(original_ids))
matched_rows = original_ids & post_ids
if not matched_rows:
return round(row_score * 0.5, 4)
value_matches = 0
value_total = 0
for rid in matched_rows:
orig_row = original_snapshot[rid]
post_row = post_fix[rid]
for col in orig_row:
value_total += 1
try:
if abs(float(str(orig_row[col])) - float(str(post_row.get(col, 0)))) < 0.001:
value_matches += 1
except (ValueError, TypeError):
if str(orig_row[col]) == str(post_row.get(col, "")):
value_matches += 1
value_score = value_matches / max(1, value_total)
return round(row_score * 0.5 + value_score * 0.5, 4)
def _snapshot_table(
conn: sqlite3.Connection,
table_name: str,
key_col: str = "id",
limit: int = 200,
) -> dict:
"""Capture table contents before a fix."""
try:
cursor = conn.execute(f"SELECT * FROM {table_name} LIMIT {limit}")
cols = [d[0] for d in cursor.description]
if key_col not in cols:
return {}
key_idx = cols.index(key_col)
return {row[key_idx]: dict(zip(cols, row)) for row in cursor.fetchall()}
except sqlite3.Error:
return {}
def _snapshot_for_preservation(original_snapshot: dict, ground_truth: dict) -> dict:
"""Filter snapshots to preservation-critical columns when configured."""
preserve_columns = ground_truth.get("preserve_columns")
if not preserve_columns:
return original_snapshot
filtered: dict = {}
for row_id, row in original_snapshot.items():
filtered[row_id] = {col: row[col] for col in preserve_columns if col in row}
return filtered
def _null_audit_score(answer: str, ground_truth: dict) -> tuple[float, str, dict]:
"""Schema-aware null_audit scoring with lexical near-miss partial credit."""
clean = (answer or "").strip()
if not clean:
return 0.0, "CATASTROPHIC", {
"answer_correct": 0.0,
"answer_in_schema": 0.0,
"lexical_near_miss": 0.0,
}
clean_lower = clean.lower()
clean_normalized = normalize_schema_label(clean)
correct = ground_truth.get("null_column", "").strip().lower()
correct_normalized = normalize_schema_label(correct)
all_cols = [c.lower() for c in ground_truth.get("all_columns", [])]
normalized_cols = [normalize_schema_label(c) for c in all_cols]
if clean_lower == correct or (clean_normalized and clean_normalized == correct_normalized):
return 1.0, "PERFECT", {
"answer_correct": 1.0,
"answer_in_schema": 0.0,
"lexical_near_miss": 0.0,
}
if clean_lower in all_cols or (clean_normalized and clean_normalized in normalized_cols):
return 0.2, "ACCEPTABLE", {
"answer_correct": 0.0,
"answer_in_schema": 0.2,
"lexical_near_miss": 0.0,
}
# Bounded lexical partial for out-of-schema near misses only.
best_similarity = 0.0
for col, normalized in zip(all_cols, normalized_cols):
best_similarity = max(
best_similarity,
SequenceMatcher(None, clean_lower, col).ratio(),
SequenceMatcher(None, clean_normalized, normalized).ratio() if clean_normalized and normalized else 0.0,
)
if best_similarity >= 0.85:
lexical = 0.15
elif best_similarity >= 0.65:
lexical = 0.1
elif best_similarity >= 0.45:
lexical = 0.05
else:
lexical = 0.0
return lexical, ("DANGEROUS" if lexical > 0 else "CATASTROPHIC"), {
"answer_correct": 0.0,
"answer_in_schema": 0.0,
"lexical_near_miss": lexical,
}
def _evaluate_fix_outcome(
test_conn: sqlite3.Connection,
ground_truth: dict,
original_snapshot: dict | None = None,
) -> tuple[bool, float, bool]:
"""Evaluate task-specific resolution + preservation + schema integrity with Deep Audit Mode."""
task_type = ground_truth.get("task_type", "")
uploaded_mode = bool(ground_truth.get("uploaded_mode"))
target_table = ground_truth.get(
"table",
ground_truth.get("target_table", ground_truth.get("root_cause_table", "transactions")),
)
try:
tables = {
row[0]
for row in test_conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()
}
except sqlite3.Error:
return False, 0.0, False
if target_table not in tables:
return False, 0.0, False
if task_type == "type_repair":
target_col = ground_truth.get("column", ground_truth.get("target_column", ""))
if uploaded_mode:
try:
rows = test_conn.execute(
f"SELECT {quote_identifier(target_col)} FROM {quote_identifier(target_table)} LIMIT 500"
).fetchall()
values = [row[0] for row in rows]
total_after = test_conn.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)}"
).fetchone()[0]
current_types = {
row[1]: (row[2] or "").upper()
for row in test_conn.execute(f"PRAGMA table_info({quote_identifier(target_table)})")
}
except sqlite3.Error:
return False, 0.0, True
issue_type = ground_truth.get("issue_type", "numeric")
if issue_type == "score_scale":
numeric_values = [parsed for parsed in (parse_numeric_scalar(value) for value in values) if parsed is not None]
anomaly_ratio_after = (
sum(1 for value in numeric_values if value > 10) / max(1, len(numeric_values))
if numeric_values
else 1.0
)
baseline_anomaly = ground_truth.get("baseline_anomaly_ratio", 1.0)
original_rows = ground_truth.get("original_row_count", total_after)
preservation = min(1.0, total_after / max(1, original_rows))
improvement = max(0.0, baseline_anomaly - anomaly_ratio_after)
issue_resolved = anomaly_ratio_after <= 0.02 or improvement >= 0.10
final_preservation = min(1.0, (preservation * 0.7) + ((1.0 - anomaly_ratio_after) * 0.3))
return issue_resolved, round(final_preservation, 4), True
if issue_type == "boolean":
parse_ratio = boolean_parse_ratio(values)
elif issue_type == "date":
parse_ratio = date_parse_ratio(values)
else:
parse_ratio = numeric_parse_ratio(values)
baseline_ratio = ground_truth.get("baseline_parse_ratio", 0.0)
original_declared = (ground_truth.get("declared_type") or "TEXT").upper()
current_declared = current_types.get(target_col, "")
original_rows = ground_truth.get("original_row_count", total_after)
preservation = min(1.0, total_after / max(1, original_rows))
improvement = max(0.0, parse_ratio - baseline_ratio)
baseline_gap = max(0.0, 1.0 - baseline_ratio)
desired_types = {
"boolean": {"BOOLEAN", "INTEGER", "NUMERIC"},
"date": {"DATE", "DATETIME", "TIMESTAMP", "TEXT"},
"numeric": {"REAL", "INTEGER", "NUMERIC", "FLOAT", "DOUBLE"},
}
type_fixed = current_declared in desired_types.get(issue_type, {"TEXT"}) and current_declared != original_declared
near_clean = parse_ratio >= 0.98
meaningful_gain = improvement >= max(0.02, baseline_gap * 0.5)
issue_resolved = (
(parse_ratio >= 0.95 and type_fixed)
or meaningful_gain
or (near_clean and improvement > 0.005)
)
final_preservation = min(1.0, (preservation * 0.55) + (min(1.0, parse_ratio) * 0.45))
return issue_resolved, round(final_preservation, 4), True
try:
cols = {
row[1]: row[2].upper()
for row in test_conn.execute(f"PRAGMA table_info({target_table})")
}
except sqlite3.Error:
return False, 0.0, True
issue_resolved = cols.get(target_col, "") in (
"REAL",
"NUMERIC",
"FLOAT",
"DOUBLE",
"INTEGER",
)
# DEEP AUDIT: Verify data wasn't just deleted, actually cast correctly
if issue_resolved:
try:
# Count total records after fix
total_after = test_conn.execute(
f"SELECT COUNT(*) FROM {target_table}"
).fetchone()[0]
# Count non-null values in target column
non_null_after = test_conn.execute(
f"SELECT COUNT(*) FROM {target_table} WHERE {target_col} IS NOT NULL"
).fetchone()[0]
# Sample some values to verify they're actually numeric
sample_values = test_conn.execute(
f"SELECT {target_col} FROM {target_table} WHERE {target_col} IS NOT NULL LIMIT 10"
).fetchall()
# Verify sample values are actually numeric (not strings)
actual_numeric_count = 0
for (val,) in sample_values:
if isinstance(val, (int, float)):
actual_numeric_count += 1
elif isinstance(val, str):
# Try to parse as number
try:
float(val) if '.' in val else int(val)
actual_numeric_count += 1
except ValueError as e:
print(f"UNPARSEABLE VALUE: {e}")
# Calculate preservation ratio
if original_snapshot and target_table in original_snapshot:
original_count = len(original_snapshot[target_table])
preservation_ratio = total_after / max(1, original_count)
# Deep audit criteria:
# 1. At least 95% of records preserved
# 2. At least 80% of non-null values are actually numeric
numeric_ratio = actual_numeric_count / max(1, len(sample_values))
if preservation_ratio >= 0.95 and numeric_ratio >= 0.8:
preservation = 1.0 # Perfect fix
elif preservation_ratio >= 0.90 and numeric_ratio >= 0.6:
preservation = 0.9 # Good fix
elif preservation_ratio >= 0.80 and numeric_ratio >= 0.4:
preservation = 0.7 # Acceptable fix
elif preservation_ratio >= 0.5:
preservation = 0.4 # Poor fix (too much data lost)
else:
preservation = 0.1 # Bad fix (most data lost)
else:
preservation = 0.8 if actual_numeric_count >= 8 else 0.4
except sqlite3.Error:
preservation = 0.5 # Unknown - give partial credit
else:
preservation = 0.0
return issue_resolved, preservation, True
if task_type == "null_audit":
target_col = ground_truth.get("column", ground_truth.get("null_column", ""))
if uploaded_mode:
try:
rows = test_conn.execute(
f"SELECT {quote_identifier(target_col)} FROM {quote_identifier(target_table)} LIMIT 500"
).fetchall()
values = [row[0] for row in rows]
total_after = test_conn.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)}"
).fetchone()[0]
except sqlite3.Error:
return False, 0.0, True
nullish_remaining = sum(1 for value in values if is_nullish_value(value))
remaining_ratio = nullish_remaining / max(1, len(values))
original_rows = ground_truth.get("original_row_count", total_after)
preservation = min(1.0, total_after / max(1, original_rows))
issue_resolved = remaining_ratio <= 0.02
final_preservation = min(1.0, (preservation * 0.4) + ((1.0 - remaining_ratio) * 0.6))
return issue_resolved, round(final_preservation, 4), True
try:
remaining_nulls = test_conn.execute(
f"SELECT COUNT(*) FROM {target_table} WHERE {target_col} IS NULL"
).fetchone()[0]
# DEEP AUDIT: Check if 'Fixed' values are logically consistent
if remaining_nulls == 0:
# Sample fixed values to check for logical consistency
sample_fixed = test_conn.execute(
f"SELECT {target_col} FROM {target_table} WHERE {target_col} IS NOT NULL LIMIT 20"
).fetchall()
# Check for obviously bad fixes
bad_fix_count = 0
total_checked = len(sample_fixed)
for (val,) in sample_fixed:
val_str = str(val).strip().lower()
# Check for negative values in columns that shouldn't be negative
if any(keyword in target_col.lower() for keyword in ['revenue', 'price', 'amount', 'salary', 'rating', 'age', 'quantity']):
try:
numeric_val = float(val_str)
if numeric_val < 0:
bad_fix_count += 1
except ValueError as e:
print(f"UNPARSEABLE NEGATIVE VALUE: {e}")
# Check for placeholder values that indicate lazy fixing
placeholder_patterns = ['na', 'n/a', 'none', 'unknown', 'missing', 'tbd', '???']
if val_str in placeholder_patterns:
bad_fix_count += 1
# Check for obviously wrong data types
if any(keyword in target_col.lower() for keyword in ['email', 'phone', 'url']):
if any(char.isdigit() for char in val_str) and len(val_str) < 10:
bad_fix_count += 1
# Calculate logical consistency score
if total_checked > 0:
bad_fix_ratio = bad_fix_count / total_checked
logical_consistency = max(0.0, 1.0 - bad_fix_ratio)
else:
logical_consistency = 1.0
# Check data preservation
if original_snapshot and target_table in original_snapshot:
original_count = len(original_snapshot[target_table])
current_count = test_conn.execute(
f"SELECT COUNT(*) FROM {target_table}"
).fetchone()[0]
preservation = min(1.0, current_count / max(1, original_count))
else:
preservation = 1.0
# Combine logical consistency with preservation
final_preservation = (logical_consistency * 0.7) + (preservation * 0.3)
else:
final_preservation = 0.0
issue_resolved = remaining_nulls == 0
return issue_resolved, final_preservation, True
except sqlite3.Error:
return False, 0.0, True
if task_type == "dedup_bug":
key_cols = ground_truth.get("key_columns") or [ground_truth.get("column", ground_truth.get("offending_key", ""))]
# Filter out empty strings and validate
key_cols = [col for col in key_cols if col and isinstance(col, str) and col.strip()]
if not key_cols:
return False, 0.0, True
expected_clean = ground_truth.get("expected_clean_count", 0)
if uploaded_mode:
try:
key_expr = ", ".join(quote_identifier(col) for col in key_cols)
remaining_dupes = test_conn.execute(
f"SELECT COUNT(*) FROM (SELECT {key_expr}, COUNT(*) c "
f"FROM {quote_identifier(target_table)} GROUP BY {key_expr} HAVING c > 1)"
).fetchone()[0]
final_count = test_conn.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)}"
).fetchone()[0]
original_rows = ground_truth.get("original_row_count", final_count)
except sqlite3.Error:
return False, 0.0, True
issue_resolved = remaining_dupes == 0
count_alignment = 1.0 - min(1.0, abs(final_count - expected_clean) / max(1, original_rows))
preservation = min(1.0, count_alignment * 0.7 + (final_count / max(1, original_rows)) * 0.3)
return issue_resolved, round(preservation, 4), True
key_col = key_cols[0]
try:
remaining_dupes = test_conn.execute(
f"SELECT COUNT(*) FROM (SELECT {key_col}, COUNT(*) c "
f"FROM {target_table} GROUP BY {key_col} HAVING c > 1)"
).fetchone()[0]
final_count = test_conn.execute(
f"SELECT COUNT(*) FROM {target_table}"
).fetchone()[0]
orphan_count = test_conn.execute(
"SELECT COUNT(*) FROM order_items oi "
"LEFT JOIN orders o ON oi.order_id = o.id "
"WHERE o.id IS NULL"
).fetchone()[0]
except sqlite3.Error:
return False, 0.0, True
issue_resolved = remaining_dupes == 0
count_ok = abs(final_count - expected_clean) <= max(1, int(expected_clean * 0.02))
no_orphans = orphan_count == 0
if issue_resolved and count_ok and no_orphans:
preservation = 1.0
elif issue_resolved and count_ok:
preservation = 0.95
elif issue_resolved:
preservation = 0.85
elif final_count > 0:
preservation = 0.5
else:
preservation = 0.0
return issue_resolved, preservation, True
if task_type == "fk_integrity":
parent_table = ground_truth.get("parent_table", "customers")
parent_key = ground_truth.get("parent_key", "id")
fk_col = ground_truth.get("column") or ground_truth.get("fk_column", "")
if not fk_col:
# Cannot validate without knowing the FK column
return False, 0.0, True
if uploaded_mode:
try:
orphan_count = test_conn.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)} c "
f"LEFT JOIN {quote_identifier(parent_table)} p "
f"ON c.{quote_identifier(fk_col)} = p.{quote_identifier(parent_key)} "
f"WHERE c.{quote_identifier(fk_col)} IS NOT NULL "
f"AND p.{quote_identifier(parent_key)} IS NULL"
).fetchone()[0]
survivor_count = test_conn.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)}"
).fetchone()[0]
except sqlite3.Error:
return False, 0.0, True
issue_resolved = orphan_count == 0
expected_clean = ground_truth.get("expected_clean_count", survivor_count)
original_rows = ground_truth.get("original_row_count", survivor_count)
survivor_ratio = survivor_count / max(1, original_rows)
count_alignment = 1.0 - min(1.0, abs(survivor_count - expected_clean) / max(1, original_rows))
preservation = min(1.0, count_alignment * 0.65 + survivor_ratio * 0.35)
return issue_resolved, round(preservation, 4), True
valid_ids = set(ground_truth.get("valid_invoice_ids", []))
expected_clean = ground_truth.get("expected_clean_count", len(valid_ids))
try:
orphan_count = test_conn.execute(
f"SELECT COUNT(*) FROM {target_table} c "
f"LEFT JOIN {parent_table} p ON c.{fk_col} = p.{parent_key} "
f"WHERE p.{parent_key} IS NULL"
).fetchone()[0]
survivors = {
row[0]
for row in test_conn.execute(f"SELECT id FROM {target_table}").fetchall()
}
except sqlite3.Error:
return False, 0.0, True
issue_resolved = orphan_count == 0
if not survivors:
return issue_resolved, 0.0, True
keep_ratio = len(survivors & valid_ids) / max(1, len(valid_ids))
unexpected_ratio = len(survivors - valid_ids) / max(1, len(survivors))
count_ok = abs(len(survivors) - expected_clean) <= max(1, int(expected_clean * 0.02))
if issue_resolved and count_ok and keep_ratio >= 0.995 and unexpected_ratio == 0:
preservation = 1.0
elif issue_resolved and keep_ratio >= 0.98:
preservation = 0.9
elif issue_resolved and keep_ratio >= 0.90:
preservation = 0.8
elif issue_resolved and keep_ratio >= 0.80:
preservation = 0.7
elif keep_ratio >= 0.7:
preservation = 0.6
elif keep_ratio >= 0.5:
preservation = 0.4
elif keep_ratio >= 0.3:
preservation = 0.2
else:
preservation = 0.1
return issue_resolved, preservation, True
if task_type == "schema_drift":
expected_columns = ground_truth.get(
"expected_columns",
[],
)
expected_rows = ground_truth.get("expected_row_count", 0)
canonical_table = ground_truth.get("table", "canonical_events")
if canonical_table not in tables:
return False, 0.0, False
if uploaded_mode:
try:
cols = {
row[1]
for row in test_conn.execute(f"PRAGMA table_info({quote_identifier(canonical_table)})")
}
row_count = test_conn.execute(
f"SELECT COUNT(*) FROM {quote_identifier(canonical_table)}"
).fetchone()[0]
except sqlite3.Error:
return False, 0.0, True
expected_set = set(expected_columns)
actual_overlap = len(expected_set & cols) / max(1, len(expected_set))
row_alignment = 1.0 - min(1.0, abs(row_count - expected_rows) / max(1, expected_rows))
issue_resolved = actual_overlap >= 0.8 and row_alignment >= 0.95
preservation = min(1.0, actual_overlap * 0.6 + row_alignment * 0.4)
return issue_resolved, round(preservation, 4), True
try:
cols = {
row[1]: row[2].upper()
for row in test_conn.execute(f"PRAGMA table_info({canonical_table})")
}
row_count = test_conn.execute(
f"SELECT COUNT(*) FROM {canonical_table}"
).fetchone()[0]
distinct_ids = test_conn.execute(
f"SELECT COUNT(DISTINCT event_id) FROM {canonical_table}"
).fetchone()[0]
null_critical = test_conn.execute(
f"SELECT COUNT(*) FROM {canonical_table} "
"WHERE event_id IS NULL OR user_id IS NULL OR event_type IS NULL OR event_time IS NULL"
).fetchone()[0]
numeric_amount = test_conn.execute(
f"SELECT COUNT(*) FROM {canonical_table} WHERE typeof(amount) IN ('real','integer')"
).fetchone()[0]
except sqlite3.Error:
return False, 0.0, True
columns_ok = all(col in cols for col in expected_columns)
row_ok = row_count == expected_rows
ids_ok = distinct_ids == row_count
critical_ok = null_critical == 0
numeric_ratio = numeric_amount / max(1, row_count)
issue_resolved = columns_ok and row_ok and ids_ok and critical_ok and numeric_ratio >= 0.99
preservation = 0.0
if columns_ok:
preservation += 0.3
if row_ok:
preservation += 0.3
if ids_ok and critical_ok:
preservation += 0.25
preservation += min(0.15, numeric_ratio * 0.15)
return issue_resolved, round(min(1.0, preservation), 4), True
# Fallback
preserve_source = _snapshot_for_preservation(original_snapshot or {}, ground_truth)
preservation = _measure_data_preservation(test_conn, preserve_source, target_table)
return preservation >= 0.99, preservation, True
def grade_diagnostic_efficiency(
trajectory: list[TrajectoryStep],
ground_truth: dict,
) -> float:
"""D1: reward faster diagnosis of the correct table/column."""
query_steps = [
s for s in trajectory if s.action_type in (ActionType.QUERY, ActionType.PROFILE)
]
for i, step in enumerate(query_steps):
if _query_reveals_issue(step.sql_submitted or "", ground_truth):
step_number = i + 1
if step_number <= 2:
return 1.0
if step_number <= 4:
return 0.7
if step_number <= 6:
return 0.4
return 0.2
return 0.0
def grade_hypothesis_quality(
trajectory: list[TrajectoryStep],
ground_truth: dict,
) -> float:
"""D2: reward systematic investigation sequence."""
query_steps = [
s for s in trajectory if s.action_type in (ActionType.QUERY, ActionType.PROFILE)
]
if not query_steps:
return 0.0
table = ground_truth.get("table", ground_truth.get("target_table", "")).lower()
col = ground_truth.get(
"column", ground_truth.get("null_column", ground_truth.get("target_column", ""))
).lower()
sqls = [s.sql_submitted or "" for s in query_steps]
first_table = next((i for i, q in enumerate(sqls) if table and table in q.lower()), None)
first_col = next((i for i, q in enumerate(sqls) if col and col in q.lower()), None)
if first_table is not None and first_col is not None and first_table < first_col:
return 1.0
if first_col is not None:
return 0.6
if first_table is not None:
return 0.3
return 0.0
def grade_fix_correctness(
fix_sql: str,
original_conn: sqlite3.Connection,
original_snapshot: dict,
ground_truth: dict,
) -> tuple[float, str]:
"""D3: task-aware fix correctness with catastrophic override tiers."""
task_type = ground_truth.get("task_type", "")
if task_type == "null_audit":
score, severity, _ = _null_audit_score(fix_sql, ground_truth)
return score, severity
if not fix_sql or fix_sql.strip().upper() in ("SELECT 1", ""):
return 0.0, "CATASTROPHIC"
test_conn = None
try:
test_conn = _fresh_test_db(original_conn)
try:
test_conn.executescript(fix_sql)
test_conn.commit()
except sqlite3.Error:
return 0.0, "CATASTROPHIC"
issue_resolved, preservation, schema_intact = _evaluate_fix_outcome(
test_conn, ground_truth, original_snapshot
)
if not schema_intact or preservation < 0.3:
return 0.0, "CATASTROPHIC"
if issue_resolved and preservation >= 0.995:
return 1.0, "PERFECT"
if issue_resolved and preservation >= 0.95:
return 0.9, "EXCELLENT"
if issue_resolved and preservation >= 0.90:
return 0.8, "ACCEPTABLE"
if issue_resolved and preservation >= 0.80:
return 0.7, "GOOD"
if issue_resolved and preservation >= 0.70:
return 0.6, "FAIR"
if issue_resolved and preservation >= 0.50:
return 0.4, "PARTIAL"
if issue_resolved:
return 0.3, "DEGRADED"
if preservation >= 0.5:
return 0.2, "DANGEROUS"
return 0.1, "CRITICAL"
finally:
if test_conn:
test_conn.close()
def grade_self_verification(
trajectory: list[TrajectoryStep],
fix_index: int,
ground_truth: dict,
) -> float:
"""D4: bonus for post-fix verification query."""
post_fix = trajectory[fix_index + 1 :] if fix_index < len(trajectory) - 1 else []
post_fix_queries = [s for s in post_fix if s.action_type == ActionType.PROFILE]
if not post_fix_queries:
return 0.0
table = ground_truth.get("table", "").lower()
col = ground_truth.get("column", ground_truth.get("null_column", "")).lower()
for step in post_fix_queries:
sql = (step.sql_submitted or "").lower()
if table in sql and col in sql:
return 0.15
return 0.0
def grade_generalization(
fix_sql: str,
original_conn: sqlite3.Connection,
ground_truth: dict,
) -> float:
"""D5: test whether the fix transfers to schema/data variants."""
task_type = ground_truth.get("task_type", "")
if task_type == "null_audit" or not fix_sql:
return 1.0
uploaded_mode = bool(ground_truth.get("uploaded_mode"))
target_table = ground_truth.get("table", "")
def _run_variant(
variant_conn: sqlite3.Connection,
variant_gt: dict | None = None,
) -> bool:
gt = variant_gt or ground_truth
try:
variant_conn.executescript(fix_sql)
variant_conn.commit()
resolved, preservation, schema_intact = _evaluate_fix_outcome(
variant_conn,
gt,
{},
)
return bool(resolved and schema_intact and preservation >= 0.8)
except sqlite3.Error:
return False
finally:
variant_conn.close()
if uploaded_mode and target_table:
passes = 0
try:
if _run_variant(_fresh_test_db(original_conn)):
passes += 1
except Exception as e:
print(f"GRADER SILENT FAIL: Variant 1 failed: {e}")
try:
v2 = _fresh_test_db(original_conn)
v2.execute(
f"ALTER TABLE {quote_identifier(target_table)} "
f"ADD COLUMN __openenv_variant TEXT DEFAULT 'variant'"
)
v2.commit()
if _run_variant(v2):
passes += 1
except Exception as e:
print(f"GRADER SILENT FAIL: Variant 2 failed: {e}")
try:
v3 = _fresh_test_db(original_conn)
variant_gt = dict(ground_truth)
if task_type == "dedup_bug":
cols = [row[1] for row in v3.execute(f"PRAGMA table_info({quote_identifier(target_table)})").fetchall()]
key_cols = ground_truth.get("key_columns") or [ground_truth.get("column")]
key_expr = ", ".join(quote_identifier(col) for col in key_cols if col)
if cols and key_expr:
col_expr = ", ".join(quote_identifier(col) for col in cols)
v3.execute(
f"INSERT INTO {quote_identifier(target_table)} ({col_expr}) "
f"SELECT {col_expr} FROM {quote_identifier(target_table)} "
f"WHERE rowid IN (SELECT MIN(rowid) FROM {quote_identifier(target_table)} GROUP BY {key_expr} LIMIT 1)"
)
v3.commit()
variant_gt["original_row_count"] = v3.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)}"
).fetchone()[0]
elif task_type == "type_repair":
target_col = ground_truth.get("column", ground_truth.get("target_column", ""))
if target_col:
if ground_truth.get("issue_type") == "score_scale":
v3.execute(
f"UPDATE {quote_identifier(target_table)} "
f"SET {quote_identifier(target_col)} = CAST({quote_identifier(target_col)} AS REAL) * 10 "
f"WHERE rowid IN (SELECT rowid FROM {quote_identifier(target_table)} "
f"WHERE {quote_identifier(target_col)} IS NOT NULL LIMIT 2)"
)
else:
v3.execute(
f"INSERT INTO {quote_identifier(target_table)} "
f"SELECT * FROM {quote_identifier(target_table)} LIMIT 2"
)
v3.commit()
variant_gt["original_row_count"] = v3.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)}"
).fetchone()[0]
else:
v3.execute(
f"INSERT INTO {quote_identifier(target_table)} "
f"SELECT * FROM {quote_identifier(target_table)} LIMIT 1"
)
v3.commit()
variant_gt["original_row_count"] = v3.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)}"
).fetchone()[0]
if _run_variant(v3, variant_gt):
passes += 1
except Exception as e:
print(f"GRADER SILENT FAIL: Variant 3 failed: {e}")
if passes == 3:
return 1.0
if passes == 2:
return 0.6
if passes == 1:
return 0.3
return 0.0
passes = 0
# Variant 1: base copy
try:
if _run_variant(_fresh_test_db(original_conn)):
passes += 1
except Exception as e:
print(f"GRADER SILENT FAIL: Generalization Variant 1 failed: {e}")
# Variant 2: schema transfer
try:
v2 = _fresh_test_db(original_conn)
if task_type == "type_repair":
v2.execute("ALTER TABLE transactions ADD COLUMN source_system TEXT DEFAULT 'legacy'")
v2.commit()
if _run_variant(v2):
passes += 1
elif task_type == "dedup_bug":
v2.execute("ALTER TABLE order_items ADD COLUMN ref_id INTEGER")
v2.execute("UPDATE order_items SET ref_id = order_id")
v2.commit()
if _run_variant(v2):
passes += 1
elif task_type == "fk_integrity":
v2.execute("ALTER TABLE invoices ADD COLUMN ingest_batch TEXT DEFAULT 'A'")
v2.commit()
if _run_variant(v2):
passes += 1
elif task_type == "schema_drift":
v2.execute("ALTER TABLE events_v2 ADD COLUMN metadata_json TEXT DEFAULT '{}'")
v2.commit()
if _run_variant(v2):
passes += 1
else:
v2.close()
except Exception as e:
print(f"GRADER SILENT FAIL: Generalization Variant 2 failed: {e}")
# Variant 3: harder data distribution
try:
v3 = _fresh_test_db(original_conn)
variant_gt = dict(ground_truth)
if task_type == "type_repair":
v3.execute(
"INSERT INTO transactions (id, user_id, amount, description, category, date, transaction_date) "
"SELECT id+10000, user_id, amount, description, category, date, transaction_date FROM transactions"
)
v3.commit()
if _run_variant(v3):
passes += 1
elif task_type == "dedup_bug":
v3.execute(
"INSERT INTO order_items (id, order_id, product_name, quantity, price) "
"SELECT NULL, order_id, product_name, quantity, price FROM order_items "
"WHERE rowid IN (SELECT MIN(rowid) FROM order_items GROUP BY order_id)"
)
v3.commit()
if _run_variant(v3):
passes += 1
elif task_type == "fk_integrity":
v3.execute(
"UPDATE invoices SET customer_id = 900000 + id WHERE id % 17 = 0"
)
v3.commit()
variant_gt["expected_clean_count"] = (
v3.execute(
"SELECT COUNT(*) FROM invoices i "
"JOIN customers c ON i.customer_id = c.id"
).fetchone()[0]
)
variant_gt["valid_invoice_ids"] = [
row[0]
for row in v3.execute(
"SELECT i.id FROM invoices i JOIN customers c ON i.customer_id = c.id"
).fetchall()
]
if _run_variant(v3, variant_gt):
passes += 1
elif task_type == "schema_drift":
v3.execute(
"INSERT INTO events_v1 (id, user_id, event_name, amount_text, event_time) "
"SELECT id+10000, user_id, event_name, amount_text, event_time FROM events_v1 WHERE id <= 100"
)
v3.execute(
"INSERT INTO events_v2 (event_id, account_id, event_type, amount, event_ts, source) "
"SELECT event_id+20000, account_id, event_type, amount, event_ts, source FROM events_v2 WHERE event_id <= 700"
)
v3.commit()
variant_gt["expected_row_count"] = (
variant_gt.get("expected_row_count", 0) + 200
)
if _run_variant(v3, variant_gt):
passes += 1
else:
v3.close()
except Exception as e:
print(f"GRADER SILENT FAIL: Generalization Variant 3 failed: {e}")
if passes == 3:
return 1.0
if passes == 2:
return 0.6
if passes == 1:
return 0.3
return 0.0
def compute_episode_score(
trajectory: list[TrajectoryStep],
ground_truth: dict,
original_conn: sqlite3.Connection,
original_snapshot: dict,
submit_action: SQLAction,
) -> SQLReward:
"""Compute five-dimension composite score at SUBMIT."""
task_type = ground_truth.get("task_type", "")
# null_audit is intentionally direct: answer quality maps directly to final score.
if task_type == "null_audit":
score, severity, null_meta = _null_audit_score(
submit_action.answer or "",
ground_truth,
)
score = _clamp(score)
return SQLReward(
score=score,
step_reward=score,
feedback=f"Null audit answer score: {score:.4f} ({severity})",
dimensions={
"d1_diagnostic_efficiency": 0.0,
"d2_hypothesis_quality": 0.0,
"d3_fix_correctness": round(score, 4),
"d3_fix_severity": severity,
"d4_self_verification": 0.0,
"d5_generalization": 1.0,
"d3_effective": round(score, 4),
"composite": score,
**null_meta,
},
partial_scores={
"d1": 0.0,
"d2": 0.0,
"d3": round(score, 4),
"d4": 0.0,
"d5": 1.0,
**null_meta,
},
)
d1 = grade_diagnostic_efficiency(trajectory, ground_truth)
d2 = grade_hypothesis_quality(trajectory, ground_truth)
fix_index = next(
(i for i, s in enumerate(trajectory) if s.action_type == ActionType.FIX),
-1,
)
fix_sql = trajectory[fix_index].sql_submitted if fix_index >= 0 else None
null_meta = {}
if fix_sql:
d3_raw, severity = grade_fix_correctness(
fix_sql,
original_conn,
original_snapshot,
ground_truth,
)
d5 = grade_generalization(fix_sql, original_conn, ground_truth)
else:
d5 = 0.0
if task_type == "dedup_bug":
correct_table = ground_truth.get("table", "order_items").lower()
correct_key = ground_truth.get("column", "").lower()
table_ok = (submit_action.root_cause_table or "").lower() == correct_table
key_ok = (submit_action.offending_key or "").lower() == correct_key
if table_ok and key_ok:
d3_raw, severity = 0.7, "ACCEPTABLE"
elif table_ok:
d3_raw, severity = 0.4, "DANGEROUS"
else:
d3_raw, severity = 0.0, "CATASTROPHIC"
else:
d3_raw, severity = 0.0, "CATASTROPHIC"
d3_effective = d3_raw
d4 = grade_self_verification(trajectory, fix_index, ground_truth) if fix_index >= 0 else 0.0
raw_score = (
d1 * 0.20
+ d2 * 0.20
+ d3_effective * 0.35
+ max(0.0, d4) * 0.10
+ d5 * 0.15
)
final_score = min(1.0, raw_score)
if severity == "CATASTROPHIC":
final_score = min(final_score, 0.05)
final_score = _clamp(final_score)
feedback_parts = [
f"Diagnostic efficiency: {d1:.2f}",
f"Hypothesis quality: {d2:.2f}",
f"Fix correctness: {d3_raw:.2f} ({severity})",
f"Fix generalization: {d5:.2f}",
f"Self-verification: {d4:.2f}",
f"Composite score: {final_score:.4f}",
]
if task_type == "null_audit" and null_meta.get("lexical_near_miss", 0.0) > 0:
feedback_parts.append(
f"Lexical near miss partial: {null_meta['lexical_near_miss']:.2f}"
)
dimensions = {
"d1_diagnostic_efficiency": round(d1, 4),
"d2_hypothesis_quality": round(d2, 4),
"d3_fix_correctness": round(d3_raw, 4),
"d3_fix_severity": severity,
"d4_self_verification": round(d4, 4),
"d5_generalization": round(d5, 4),
"d3_effective": round(d3_effective, 4),
"composite": final_score,
}
if null_meta:
dimensions.update(null_meta)
return SQLReward(
score=_clamp(final_score),
step_reward=_clamp(final_score),
feedback=" | ".join(feedback_parts),
dimensions=dimensions,
partial_scores={
"d1": round(d1, 4),
"d2": round(d2, 4),
"d3": round(d3_raw, 4),
"d4": round(d4, 4),
"d5": round(d5, 4),
**null_meta,
},
)
def grade_null_audit(action: SQLAction, ground_truth: Dict[str, Any]) -> SQLReward:
"""Single-step null_audit scorer used by validation scripts."""
score, severity, meta = _null_audit_score(action.answer or "", ground_truth)
answer = (action.answer or "").strip()
if not answer:
feedback = "No answer submitted"
elif meta["answer_correct"] == 1.0:
feedback = f"Correct: identified '{ground_truth.get('null_column', '')}'."
elif meta["answer_in_schema"] > 0:
feedback = (
f"Submitted column '{answer}' exists in the schema but is not the high-null column."
)
elif meta["lexical_near_miss"] > 0:
feedback = (
f"Submitted column '{answer}' not in schema, but close to a valid column name."
)
else:
feedback = f"Submitted column '{answer}' is not present in the schema."
return SQLReward(
score=_clamp(score),
step_reward=_clamp(score),
feedback=feedback,
dimensions={
"legacy_mode": True,
"d3_fix_severity": severity,
"composite": _clamp(score),
**meta,
},
partial_scores=meta,
)
def grade_type_repair(
action: SQLAction, ground_truth: dict, conn: sqlite3.Connection
) -> SQLReward:
"""Single-step type_repair scorer for exploit checks and compatibility."""
fix_sql = (action.fix_query or action.sql or "").strip()
target_table = ground_truth.get("target_table", "transactions")
partial_scores = {"syntax_valid": 0.0, "type_correct": 0.0, "data_valid": 0.0}
def _make(feedback: str) -> SQLReward:
score = _clamp(sum(partial_scores.values()))
return SQLReward(
score=_clamp(score),
step_reward=_clamp(score),
feedback=feedback,
dimensions={"legacy_mode": True, "composite": _clamp(score), **partial_scores},
partial_scores=partial_scores.copy(),
)
if not fix_sql:
return _make("No fix_query provided")
if fix_sql.upper().startswith("SELECT"):
return _make("No-op fix_query did not modify the target schema")
snapshot = _snapshot_for_preservation(_snapshot_table(conn, target_table), ground_truth)
test_conn = _fresh_test_db(conn)
try:
try:
test_conn.executescript(fix_sql)
test_conn.commit()
except sqlite3.Error as exc:
return _make(f"SQL error: {exc}")
issue_resolved, preservation, schema_intact = _evaluate_fix_outcome(
test_conn,
{**ground_truth, "task_type": "type_repair"},
snapshot,
)
if not schema_intact:
return _make("Destructive query detected: target table missing")
partial_scores["syntax_valid"] = 0.3
if issue_resolved:
partial_scores["type_correct"] = 0.4
if preservation >= 0.995:
partial_scores["data_valid"] = 0.3
if partial_scores["type_correct"] == 0.0:
return _make("Fix executed but target column type is still incorrect")
if partial_scores["data_valid"] == 0.0:
return _make("Fix executed but data preservation check failed")
if sum(partial_scores.values()) == 1.0:
return _make("Correct fix: target column type repaired with data preserved")
return _make("Fix executed with partial credit")
finally:
test_conn.close()
def grade_dedup_bug(
action: SQLAction, ground_truth: Dict[str, Any], conn: sqlite3.Connection
) -> SQLReward:
"""Single-step dedup_bug scorer for exploit checks and compatibility."""
fix_sql = (action.fix_query or action.sql or "").strip()
target_table = ground_truth.get("table", ground_truth.get("root_cause_table", "order_items"))
target_key = ground_truth.get("column", ground_truth.get("offending_key", ""))
table_ok = (action.root_cause_table or "").lower() == target_table.lower()
key_ok = (action.offending_key or "").lower() == target_key.lower()
partial_scores = {
"root_cause_identified": 0.4 if table_ok else 0.0,
"key_identified": 0.3 if key_ok else 0.0,
"fix_valid": 0.0,
}
sql_error = None
if fix_sql and fix_sql.upper() != "SELECT 1":
test_conn = _fresh_test_db(conn)
try:
test_conn.executescript(fix_sql)
test_conn.commit()
issue_resolved, preservation, schema_intact = _evaluate_fix_outcome(
test_conn,
{**ground_truth, "task_type": "dedup_bug"},
{},
)
if schema_intact and issue_resolved and preservation >= 0.95:
partial_scores["fix_valid"] = 0.3
except sqlite3.Error as exc:
sql_error = str(exc)
finally:
test_conn.close()
score = round(sum(partial_scores.values()), 4)
if sql_error:
feedback = f"SQL error: {sql_error}"
elif partial_scores["fix_valid"] == 0.3:
feedback = "Correct table, key, and dedup fix identified."
elif table_ok and key_ok:
feedback = "Correct table and key identified, but fix did not resolve duplication."
elif table_ok:
feedback = "Correct root cause table identified."
else:
feedback = "Root cause not identified."
return SQLReward(
score=_clamp(score),
step_reward=_clamp(score),
feedback=feedback,
dimensions={
"legacy_mode": True,
"composite": _clamp(score),
**partial_scores,
},
partial_scores=partial_scores.copy(),
)
def grade_fk_integrity(
action: SQLAction,
ground_truth: Dict[str, Any],
conn: sqlite3.Connection,
) -> SQLReward:
"""Single-step fk_integrity scorer — validates grader exists."""
fix_sql = (action.fix_query or action.sql or "").strip()
target_table = ground_truth.get("table", "invoices")
parent_table = ground_truth.get("parent_table", "customers")
parent_key = ground_truth.get("parent_key", "id")
fk_col = ground_truth.get("column", "customer_id")
snapshot = _snapshot_table(conn, target_table)
if not fix_sql or fix_sql.upper().startswith("SELECT"):
return SQLReward(
score=_clamp(0.0),
step_reward=_clamp(0.0),
feedback="No fix query provided",
dimensions={"legacy_mode": True},
partial_scores={},
)
test_conn = _fresh_test_db(conn)
try:
test_conn.executescript(fix_sql)
test_conn.commit()
orphan_count = test_conn.execute(
f"SELECT COUNT(*) FROM {quote_identifier(target_table)} t "
f"LEFT JOIN {quote_identifier(parent_table)} p ON t.{quote_identifier(fk_col)} = p.{quote_identifier(parent_key)} "
f"WHERE p.{quote_identifier(parent_key)} IS NULL"
).fetchone()[0]
issue_resolved = orphan_count == 0
preservation = _measure_data_preservation(
test_conn, snapshot, target_table
)
score = 0.0
if issue_resolved:
score += 0.6
if preservation >= 0.95:
score += 0.4
return SQLReward(
score=_clamp(score),
step_reward=_clamp(score),
feedback=(
"FK integrity resolved" if issue_resolved
else f"Orphans remaining: {orphan_count}"
),
dimensions={"legacy_mode": True, "composite": _clamp(score)},
partial_scores={"issue_resolved": float(issue_resolved),
"preservation": preservation},
)
except sqlite3.Error as exc:
return SQLReward(
score=_clamp(0.0),
step_reward=_clamp(0.0),
feedback=f"SQL error: {exc}",
dimensions={"legacy_mode": True},
partial_scores={},
)
finally:
test_conn.close()
def grade_schema_drift(
action: SQLAction,
ground_truth: Dict[str, Any],
conn: sqlite3.Connection,
) -> SQLReward:
"""Single-step schema_drift scorer — validates grader exists."""
fix_sql = (action.fix_query or action.sql or "").strip()
canonical_table = ground_truth.get("table", "canonical_events")
expected_cols = ground_truth.get(
"expected_columns",
["event_id", "user_id", "event_type", "amount", "event_time"]
)
expected_rows = ground_truth.get("expected_row_count", 1200)
if not fix_sql:
return SQLReward(
score=_clamp(0.0),
step_reward=_clamp(0.0),
feedback="No fix query provided",
dimensions={"legacy_mode": True},
partial_scores={},
)
test_conn = _fresh_test_db(conn)
try:
test_conn.executescript(fix_sql)
test_conn.commit()
tables = {
r[0] for r in test_conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()
}
if canonical_table not in tables:
return SQLReward(
score=_clamp(0.0),
step_reward=_clamp(0.0),
feedback=f"Table {canonical_table} not created",
dimensions={"legacy_mode": True},
partial_scores={},
)
cols = {
r[1] for r in test_conn.execute(
f"PRAGMA table_info({quote_identifier(canonical_table)})"
)
}
row_count = test_conn.execute(
f"SELECT COUNT(*) FROM {quote_identifier(canonical_table)}"
).fetchone()[0]
col_overlap = len(set(expected_cols) & cols) / max(1, len(expected_cols))
row_align = 1.0 - min(
1.0, abs(row_count - expected_rows) / max(1, expected_rows)
)
score = col_overlap * 0.5 + row_align * 0.5
return SQLReward(
score=_clamp(score),
step_reward=_clamp(score),
feedback=f"Schema drift: col_overlap={col_overlap:.2f} row_align={row_align:.2f}",
dimensions={"legacy_mode": True, "composite": _clamp(score)},
partial_scores={"col_overlap": col_overlap, "row_align": row_align},
)
except sqlite3.Error as exc:
return SQLReward(
score=_clamp(0.0),
step_reward=_clamp(0.0),
feedback=f"SQL error: {exc}",
dimensions={"legacy_mode": True},
partial_scores={},
)
finally:
test_conn.close()
__all__ = [
"grade_null_audit",
"grade_type_repair",
"grade_dedup_bug",
"grade_fk_integrity",
"grade_schema_drift",
"compute_episode_score",
]
"""Grading functions for SQL Quality Environment."""
from __future__ import annotations
from difflib import SequenceMatcher
import sqlite3
from typing import Any, Dict
from .environment import (
boolean_parse_ratio,
date_parse_ratio,
is_nullish_value,
numeric_parse_ratio,
normalize_schema_label,
parse_numeric_scalar,
quote_identifier,
)
from .models import ActionType, SQLAction, SQLReward, TrajectoryStep
def _clamp(score: float) -> float:
return round(max(0.001, min(0.999, float(score))), 4)
def _fresh_test_db(original_conn: sqlite3.Connection) -> sqlite3.Connection:
"""Create an isolated in-memory copy of a sqlite database."""
copy = sqlite3.connect(":memory:")
for line in original_conn.iterdump():
copy.execute(line)
copy.commit()
return copy
# Backward-compatible alias
_make_fresh_copy = _fresh_test_db
def _query_reveals_issue(sql: str, ground_truth: dict) -> bool:
"""Return True when query references both ground-truth table and column."""
sql_lower = (sql or "").lower()
table = ground_truth.get("table", ground_truth.get("target_table", "")).lower()
col = ground_truth.get(
"column", ground_truth.get("null_column", ground_truth.get("target_column", ""))
).lower()
return bool(table and col and table in sql_lower and col in sql_lower)
def _measure_data_preservation(
test_conn: sqlite3.Connection,
original_snapshot: dict,
table_name: str,
key_col: str = "id",
) -> float:
"""Measure row/value preservation ratio after a fix."""
if not original_snapshot:
return 1.0
try:
cursor = test_conn.execute(f"SELECT * FROM {table_name}")
cols = [d[0] for d in cursor.description]
if key_col not in cols:
return 0.0
key_idx = cols.index(key_col)
post_fix = {row[key_idx]: dict(zip(cols, row)) for row in cursor.fetchall()}
except sqlite3.Error:
return 0.0
if not post_fix:
return 0.0
original_ids = set(original_snapshot.keys())
post_ids = set(post_fix.keys())
row_score = len(original_ids & post_ids) / max(1, len(original_ids))
matched_rows = original_ids & post_ids
if not matched_rows:
return round(row_score * 0.5, 4)
value_matches = 0
value_total = 0
for rid in matched_rows:
orig_row = original_snapshot[rid]
post_row = post_fix[rid]
for col in orig_row:
value_total += 1
try:
if abs(float(str(orig_row[col])) - float(str(post_row.get(col, 0)))) < 0.001:
value_matches += 1
except (ValueError, TypeError):
if str(orig_row[col]) == str(post_row.get(col, "")):
value_matches += 1
value_score = value_matches / max(1, value_total)
return round(row_score * 0.5 + value_score * 0.5, 4)
def _snapshot_table(
conn: sqlite3.Connection,
table_name: str,
key_col: str = "id",
limit: int = 200,