-- Auditoría de split para DuckDB.
-- Ejecutar desde la raíz del ZIP después de `python3 ops/split_audit.py --write`:
-- duckdb -c ".read sql/split_audit_duckdb.sql"

CREATE OR REPLACE TEMP VIEW cases AS
SELECT *
FROM read_csv_auto('data/support_split_cases.csv', header = true);

CREATE OR REPLACE TEMP VIEW feature_history AS
SELECT *
FROM read_csv_auto('data/feature_history.csv', header = true);

CREATE OR REPLACE TEMP VIEW split_assignments AS
SELECT *
FROM read_csv_auto('output/dataset_split_assignments.csv', header = true);

-- 1. Un caso debe tener una sola asignación en la versión materializada.
SELECT
  'duplicate_case_assignment' AS check_name,
  case_id,
  COUNT(*) AS assignment_count
FROM split_assignments
GROUP BY case_id
HAVING COUNT(*) <> 1;

-- 2. Todo caso del dataset debe aparecer en la tabla materializada de split.
SELECT
  'case_missing_from_split' AS check_name,
  c.case_id
FROM cases c
LEFT JOIN split_assignments a USING (case_id)
WHERE a.case_id IS NULL;

-- 3. No queremos que una misma entidad aparezca en train y test.
SELECT
  'student_train_test_overlap' AS check_name,
  c.student_id,
  STRING_AGG(DISTINCT a.split, ', ' ORDER BY a.split) AS splits
FROM cases c
JOIN split_assignments a USING (case_id)
GROUP BY c.student_id
HAVING BOOL_OR(a.split = 'train') AND BOOL_OR(a.split = 'test');

-- 4. No queremos que una misma fuente documental aparezca en train y test.
SELECT
  'source_train_test_overlap' AS check_name,
  c.source_id,
  STRING_AGG(DISTINCT a.split, ', ' ORDER BY a.split) AS splits
FROM cases c
JOIN split_assignments a USING (case_id)
GROUP BY c.source_id
HAVING BOOL_OR(a.split = 'train') AND BOOL_OR(a.split = 'test');

-- 5. Riesgo de join ingenuo: existen features posteriores al caso.
-- Esto no bloquea si el pipeline hace as-of join, pero sí detecta dónde un JOIN
-- "último valor disponible en la tabla" podría contaminar entrenamiento o test.
SELECT
  'future_feature_candidate' AS check_name,
  c.case_id,
  c.student_id,
  c.created_at,
  f.feature_available_at,
  f.open_cases_30d,
  f.last_payment_status
FROM cases c
JOIN feature_history f USING (student_id)
WHERE CAST(f.feature_available_at AS DATE) > CAST(c.created_at AS DATE)
ORDER BY c.case_id, f.feature_available_at;

-- 6. As-of join seguro: el valor elegido es el último disponible antes o en el caso.
SELECT
  c.case_id,
  c.student_id,
  c.created_at,
  f.feature_available_at AS selected_feature_available_at,
  f.open_cases_30d,
  f.last_payment_status
FROM cases c
LEFT JOIN LATERAL (
  SELECT *
  FROM feature_history f
  WHERE f.student_id = c.student_id
    AND CAST(f.feature_available_at AS DATE) <= CAST(c.created_at AS DATE)
  ORDER BY CAST(f.feature_available_at AS DATE) DESC
  LIMIT 1
) f ON true
ORDER BY c.case_id;
