"""Build a private, point-in-time inspection feature mart with DuckDB. Inputs are either a contiguous set of bounded monthly CSV.gz exports or the single complete-history development sample. Every input must have the manifest written by the corresponding secure exporter. """ from __future__ import annotations import argparse import csv import gzip import hashlib import json import os import re import sys from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple import duckdb PROJECT_ROOT = Path(__file__).resolve().parents[1] PRIVATE_DATA_ROOT = (PROJECT_ROOT / "data/private").resolve() DEFAULT_INPUT = PRIVATE_DATA_ROOT / "inspection_batches" DEFAULT_DATABASE = PRIVATE_DATA_ROOT / "warehouse/vehicle_health.duckdb" DEFAULT_OUTPUT = PRIVATE_DATA_ROOT / "marts/inspection_feature_mart.parquet" SQL_ROOT = PROJECT_ROOT / "sql/local" EXPECTED_COLUMNS = ( "internal_event_id", "vehicle_token", "vehicle_bucket", "event_ts", "source_era", "public_county", "canonical_outcome", "outcome_label_source", "program_type", "test_type", "observed_make", "observed_model", "observed_model_year", ) BOUNDED_KIND = "bounded_batch" DEVELOPMENT_KIND = "vehicle_history_development_sample" HEX_64 = re.compile(r"^[0-9a-f]{64}$") FINGERPRINT = re.compile(r"^[0-9a-f]{16,64}$") MEMORY_LIMIT = re.compile(r"^[1-9][0-9]*(?:MB|GB|TB)$", re.IGNORECASE) SQL_FILES = ( "20_stage_events.sql", "21_build_episodes.sql", "22_build_features.sql", "23_validate_mart.sql", ) CREATE_STAGING_SQL = """ CREATE TABLE stg_events ( batch_file VARCHAR NOT NULL, batch_kind VARCHAR NOT NULL, batch_start TIMESTAMP NOT NULL, batch_end TIMESTAMP NOT NULL, internal_event_id VARCHAR, vehicle_token VARCHAR, vehicle_bucket VARCHAR, event_ts VARCHAR, source_era VARCHAR, public_county VARCHAR, canonical_outcome VARCHAR, outcome_label_source VARCHAR, program_type VARCHAR, test_type VARCHAR, observed_make VARCHAR, observed_model VARCHAR, observed_model_year VARCHAR ) """ # The dialect and every input column are explicit. In particular, do not let a # sample without quoted values convince DuckDB that quote handling is disabled. INGEST_SQL = """ INSERT INTO stg_events SELECT ? AS batch_file, ? AS batch_kind, ?::TIMESTAMP AS batch_start, ?::TIMESTAMP AS batch_end, source.internal_event_id, source.vehicle_token, source.vehicle_bucket, source.event_ts, source.source_era, source.public_county, source.canonical_outcome, source.outcome_label_source, source.program_type, source.test_type, source.observed_make, source.observed_model, source.observed_model_year FROM read_csv( ?, header = true, auto_detect = false, delim = ',', quote = '"', escape = '"', nullstr = '', strict_mode = true, columns = { 'internal_event_id': 'VARCHAR', 'vehicle_token': 'VARCHAR', 'vehicle_bucket': 'VARCHAR', 'event_ts': 'VARCHAR', 'source_era': 'VARCHAR', 'public_county': 'VARCHAR', 'canonical_outcome': 'VARCHAR', 'outcome_label_source': 'VARCHAR', 'program_type': 'VARCHAR', 'test_type': 'VARCHAR', 'observed_make': 'VARCHAR', 'observed_model': 'VARCHAR', 'observed_model_year': 'VARCHAR' } ) AS source """ class MartBuildError(RuntimeError): """Raised when private input or a mart invariant fails closed.""" @dataclass(frozen=True) class InputBatch: path: Path manifest_path: Path export_kind: str start: datetime end: datetime rows: int query_version: str query_sha256: str key_version: str key_fingerprint: str compressed_sha256: str compressed_bytes: int manifest: Mapping[str, Any] @dataclass(frozen=True) class BuildConfig: input_path: Path = DEFAULT_INPUT database_path: Path = DEFAULT_DATABASE output_path: Path = DEFAULT_OUTPUT episode_gap_days: int = 30 memory_limit: str = "4GB" threads: int = max(1, min(4, os.cpu_count() or 1)) allow_gaps: bool = False allow_external_output: bool = False overwrite: bool = False @dataclass(frozen=True) class BuildSummary: database_path: Path output_path: Path manifest_path: Path input_rows: int clean_events: int episodes: int mart_rows: int eligible_rows: int source_data_kind: str population_estimate_allowed: bool def file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def parse_manifest_timestamp(value: object, field: str) -> datetime: if not isinstance(value, str): raise MartBuildError(f"Manifest field {field!r} must be an ISO timestamp") encoded = value[:-1] + "+00:00" if value.endswith("Z") else value try: parsed = datetime.fromisoformat(encoded) except ValueError as exc: raise MartBuildError(f"Manifest field {field!r} is not ISO-8601") from exc if parsed.tzinfo is not None: parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None) return parsed def _required_text(manifest: Mapping[str, Any], field: str) -> str: value = manifest.get(field) if not isinstance(value, str) or not value.strip(): raise MartBuildError(f"Manifest field {field!r} must be non-empty text") return value.strip() def _required_nonnegative_int(manifest: Mapping[str, Any], field: str) -> int: value = manifest.get(field) if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise MartBuildError(f"Manifest field {field!r} must be a nonnegative integer") return value def _read_header(path: Path) -> Tuple[str, ...]: try: with gzip.open(path, "rt", encoding="utf-8", newline="") as handle: row = next(csv.reader(handle), None) except (OSError, UnicodeError, csv.Error) as exc: raise MartBuildError(f"Cannot read gzip CSV header: {path}") from exc if row is None: raise MartBuildError(f"Input gzip CSV is empty: {path}") return tuple(row) def _load_batch(path: Path) -> InputBatch: manifest_path = path.with_name(path.name + ".manifest.json") if not manifest_path.is_file(): raise MartBuildError(f"Missing paired manifest: {manifest_path}") try: manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise MartBuildError(f"Cannot parse manifest: {manifest_path}") from exc if not isinstance(manifest, dict): raise MartBuildError(f"Manifest must contain a JSON object: {manifest_path}") columns = manifest.get("columns") if not isinstance(columns, (list, tuple)) or tuple(columns) != EXPECTED_COLUMNS: raise MartBuildError(f"Unexpected manifest columns: {manifest_path}") if _read_header(path) != EXPECTED_COLUMNS: raise MartBuildError(f"CSV header does not match the private schema: {path}") export_kind = _required_text(manifest, "export_kind") if export_kind not in (BOUNDED_KIND, DEVELOPMENT_KIND): raise MartBuildError(f"Unsupported export_kind {export_kind!r}: {manifest_path}") start = parse_manifest_timestamp( manifest.get("source_start_inclusive"), "source_start_inclusive" ) end = parse_manifest_timestamp( manifest.get("source_end_exclusive"), "source_end_exclusive" ) if end <= start: raise MartBuildError(f"Manifest source interval is empty or reversed: {path}") rows = _required_nonnegative_int(manifest, "rows") compressed_bytes = _required_nonnegative_int(manifest, "compressed_file_bytes") actual_bytes = path.stat().st_size if compressed_bytes != actual_bytes: raise MartBuildError( f"Compressed byte count mismatch for {path}: " f"manifest={compressed_bytes}, actual={actual_bytes}" ) compressed_sha256 = _required_text(manifest, "compressed_file_sha256").lower() if not HEX_64.fullmatch(compressed_sha256): raise MartBuildError(f"Invalid compressed SHA-256 in {manifest_path}") actual_sha256 = file_sha256(path) if compressed_sha256 != actual_sha256: raise MartBuildError(f"Compressed SHA-256 mismatch for {path}") query_sha256 = _required_text(manifest, "query_sha256").lower() if not HEX_64.fullmatch(query_sha256): raise MartBuildError(f"Invalid query SHA-256 in {manifest_path}") key_fingerprint = _required_text( manifest, "vehicle_token_key_fingerprint" ).lower() if not FINGERPRINT.fullmatch(key_fingerprint): raise MartBuildError(f"Invalid key fingerprint in {manifest_path}") if export_kind == BOUNDED_KIND: if manifest.get("classification") != ( "private_pseudonymized_analytical_staging" ): raise MartBuildError(f"Unexpected bounded-batch classification: {path}") else: if manifest.get("classification") != ( "private_pseudonymized_development_only" ): raise MartBuildError(f"Unexpected development classification: {path}") if manifest.get("population_estimate_allowed") is not False: raise MartBuildError( "Development history manifest must prohibit population estimates" ) return InputBatch( path=path, manifest_path=manifest_path, export_kind=export_kind, start=start, end=end, rows=rows, query_version=_required_text(manifest, "query_version"), query_sha256=query_sha256, key_version=_required_text(manifest, "vehicle_token_key_version"), key_fingerprint=key_fingerprint, compressed_sha256=compressed_sha256, compressed_bytes=compressed_bytes, manifest=manifest, ) def _discover_files(input_path: Path) -> List[Path]: if input_path.is_file(): if not input_path.name.endswith(".csv.gz"): raise MartBuildError("A feature-mart input file must end in .csv.gz") return [input_path] if not input_path.is_dir(): raise MartBuildError(f"Input path does not exist: {input_path}") paths = sorted(path for path in input_path.glob("*.csv.gz") if path.is_file()) if not paths: raise MartBuildError(f"No .csv.gz inputs found in {input_path}") return paths def discover_and_validate_batches(config: BuildConfig) -> List[InputBatch]: input_path = config.input_path.resolve() if not config.allow_external_output and not input_path.is_relative_to( PRIVATE_DATA_ROOT ): raise MartBuildError(f"Private inputs must stay under {PRIVATE_DATA_ROOT}") batches = [_load_batch(path.resolve()) for path in _discover_files(input_path)] kinds = {batch.export_kind for batch in batches} if len(kinds) != 1: raise MartBuildError("Do not mix bounded batches and development samples") kind = next(iter(kinds)) key_versions = {batch.key_version for batch in batches} fingerprints = {batch.key_fingerprint for batch in batches} if len(key_versions) != 1 or len(fingerprints) != 1: raise MartBuildError("All inputs must use one VIN-token key and version") if kind == DEVELOPMENT_KIND: if len(batches) != 1: raise MartBuildError("A development mart accepts exactly one history sample") return batches query_versions = {batch.query_version for batch in batches} query_hashes = {batch.query_sha256 for batch in batches} if len(query_versions) != 1 or len(query_hashes) != 1: raise MartBuildError("All bounded batches must use one extraction query version") batches.sort(key=lambda batch: (batch.start, batch.end, str(batch.path))) for previous, current in zip(batches, batches[1:]): if current.start < previous.end: raise MartBuildError( f"Source intervals overlap: {previous.path.name} and " f"{current.path.name}" ) if current.start > previous.end and not config.allow_gaps: raise MartBuildError( f"Source interval gap between {previous.end.isoformat()} and " f"{current.start.isoformat()}; use --allow-gaps only for an " "explicitly incomplete analytical build" ) return batches def _validate_config(config: BuildConfig) -> BuildConfig: if not 1 <= config.episode_gap_days <= 365: raise MartBuildError("episode_gap_days must be between 1 and 365") if not 1 <= config.threads <= 64: raise MartBuildError("threads must be between 1 and 64") if not MEMORY_LIMIT.fullmatch(config.memory_limit): raise MartBuildError("memory_limit must look like 4096MB or 4GB") database_path = config.database_path.resolve() output_path = config.output_path.resolve() if output_path.suffix.lower() != ".parquet": raise MartBuildError("Feature-mart output must use a .parquet suffix") if not config.allow_external_output: for path in (database_path, output_path): if not path.is_relative_to(PRIVATE_DATA_ROOT): raise MartBuildError(f"Private outputs must stay under {PRIVATE_DATA_ROOT}") if database_path == output_path: raise MartBuildError("Database and Parquet output paths must differ") return BuildConfig( input_path=config.input_path.resolve(), database_path=database_path, output_path=output_path, episode_gap_days=config.episode_gap_days, memory_limit=config.memory_limit.upper(), threads=config.threads, allow_gaps=config.allow_gaps, allow_external_output=config.allow_external_output, overwrite=config.overwrite, ) def _load_sql(name: str) -> str: path = SQL_ROOT / name try: return path.read_text(encoding="utf-8") except OSError as exc: raise MartBuildError(f"Cannot read local transform SQL: {path}") from exc def _remove_partial(paths: Iterable[Path]) -> None: for path in paths: path.unlink(missing_ok=True) def _output_columns(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, str]]: rows = connection.execute("DESCRIBE SELECT * FROM feature_mart").fetchall() return [ {"name": str(row[0]), "duckdb_type": str(row[1])} for row in rows ] def build_feature_mart(config: BuildConfig) -> BuildSummary: config = _validate_config(config) batches = discover_and_validate_batches(config) source_data_kind = batches[0].export_kind population_estimate_allowed = ( source_data_kind == BOUNDED_KIND and not config.allow_gaps ) database_path = config.database_path output_path = config.output_path manifest_path = output_path.with_name(output_path.name + ".manifest.json") partial_database = database_path.with_name(database_path.name + ".partial") partial_output = output_path.with_name(output_path.name + ".partial") partial_manifest = manifest_path.with_name(manifest_path.name + ".partial") partial_wal = partial_database.with_name(partial_database.name + ".wal") partial_paths = (partial_database, partial_output, partial_manifest, partial_wal) finals = (database_path, output_path, manifest_path) existing = [path for path in finals if path.exists()] if existing and not config.overwrite: raise MartBuildError( "Refusing to overwrite existing outputs: " + ", ".join(str(path) for path in existing) ) os.umask(0o077) database_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) output_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) temp_directory = database_path.parent / ".duckdb_tmp" temp_directory.mkdir(parents=True, exist_ok=True, mode=0o700) _remove_partial(partial_paths) connection: Optional[duckdb.DuckDBPyConnection] = None try: connection = duckdb.connect(str(partial_database)) connection.execute("SET threads = ?", [config.threads]) connection.execute("SET memory_limit = ?", [config.memory_limit]) connection.execute("SET temp_directory = ?", [str(temp_directory)]) connection.execute("SET preserve_insertion_order = false") connection.execute( """ CREATE TABLE build_config AS SELECT ?::INTEGER AS episode_gap_days, ?::VARCHAR AS source_data_kind, ?::BOOLEAN AS population_estimate_allowed """, [ config.episode_gap_days, source_data_kind, population_estimate_allowed, ], ) connection.execute(CREATE_STAGING_SQL) for batch in batches: connection.execute( INGEST_SQL, [ str(batch.path), batch.export_kind, batch.start.isoformat(sep=" "), batch.end.isoformat(sep=" "), str(batch.path), ], ) actual_by_file = dict( connection.execute( "SELECT batch_file, count(*) FROM stg_events GROUP BY batch_file" ).fetchall() ) for batch in batches: actual = int(actual_by_file.get(str(batch.path), 0)) if actual != batch.rows: raise MartBuildError( f"CSV row count mismatch for {batch.path}: " f"manifest={batch.rows}, actual={actual}" ) duplicate_ids = connection.execute( """ SELECT count(*) FROM ( SELECT internal_event_id FROM stg_events GROUP BY internal_event_id HAVING count(*) > 1 ) """ ).fetchone()[0] if duplicate_ids: raise MartBuildError( f"Input contains {duplicate_ids} duplicated internal event IDs" ) inconsistent_buckets = connection.execute( """ SELECT count(*) FROM ( SELECT lower(vehicle_token) AS vehicle_token FROM stg_events WHERE regexp_full_match(lower(vehicle_token), '^[0-9a-f]{64}$') AND try_cast(vehicle_bucket AS INTEGER) BETWEEN 0 AND 99 GROUP BY lower(vehicle_token) HAVING count(DISTINCT try_cast(vehicle_bucket AS INTEGER)) > 1 ) """ ).fetchone()[0] if inconsistent_buckets: raise MartBuildError( f"Input contains {inconsistent_buckets} vehicle tokens with " "inconsistent audit buckets" ) sql_hashes: Dict[str, str] = {} for name in SQL_FILES: sql = _load_sql(name) sql_hashes[name] = hashlib.sha256(sql.encode("utf-8")).hexdigest() connection.execute(sql) violations = connection.execute( """ SELECT check_name, violation_count FROM mart_validation WHERE violation_count <> 0 ORDER BY check_name """ ).fetchall() if violations: details = ", ".join(f"{name}={count}" for name, count in violations) raise MartBuildError(f"Mart invariant failure: {details}") counts = connection.execute( """ SELECT (SELECT count(*) FROM stg_events), (SELECT count(*) FROM clean_events), (SELECT count(*) FROM inspection_episodes), (SELECT count(*) FROM feature_mart), (SELECT count(*) FROM feature_mart WHERE eligible_returning_target) """ ).fetchone() input_rows, clean_events, episodes, mart_rows, eligible_rows = map( int, counts ) columns = _output_columns(connection) connection.execute("ANALYZE") connection.execute( """ COPY ( SELECT * FROM feature_mart ORDER BY vehicle_token, episode_number ) TO ? (FORMAT PARQUET, COMPRESSION ZSTD) """, [str(partial_output)], ) build_id = hashlib.sha256( json.dumps( { "inputs": [batch.compressed_sha256 for batch in batches], "sql": sql_hashes, "gap": config.episode_gap_days, }, sort_keys=True, ).encode("utf-8") ).hexdigest()[:16] output_sha256 = file_sha256(partial_output) build_manifest: Dict[str, Any] = { "build_kind": "private_leakage_safe_inspection_feature_mart", "build_id": build_id, "generated_at_utc": datetime.now(timezone.utc).isoformat(), "classification": "private_pseudonymized_analytical_mart", "source_data_kind": source_data_kind, "population_estimate_allowed": population_estimate_allowed, "episode_gap_days": config.episode_gap_days, "vehicle_quality_contract": ( "eligibility uses only prior events: at most 50 prior events " "and at most 4 prior events on any day" ), "outcome_label_source_contract": { "allowed_values": ["overall_result", "utah_obd_proxy"], "null_allowed_only_for_unlabeled_outcomes": True, "utah_obd_proxy_source_era": "utah", "feature_role": "audit_only_not_a_predictor", }, "vin_audit_contract": "vehicle_bucket < 10", "inputs": [ { "file": str(batch.path), "manifest": str(batch.manifest_path), "export_kind": batch.export_kind, "query_version": batch.query_version, "query_sha256": batch.query_sha256, "source_start_inclusive": batch.start.isoformat(), "source_end_exclusive": batch.end.isoformat(), "rows": batch.rows, "compressed_file_sha256": batch.compressed_sha256, } for batch in batches ], "vehicle_token_key_version": batches[0].key_version, "vehicle_token_key_fingerprint": batches[0].key_fingerprint, "transform_sql_sha256": sql_hashes, "row_counts": { "input": input_rows, "clean_events": clean_events, "episodes": episodes, "feature_mart": mart_rows, "eligible_returning_targets": eligible_rows, }, "columns": columns, "parquet_sha256": output_sha256, "parquet_bytes": partial_output.stat().st_size, } partial_manifest.write_text( json.dumps(build_manifest, indent=2) + "\n", encoding="utf-8" ) connection.execute("CHECKPOINT") connection.close() connection = None os.replace(partial_database, database_path) os.replace(partial_output, output_path) os.replace(partial_manifest, manifest_path) return BuildSummary( database_path=database_path, output_path=output_path, manifest_path=manifest_path, input_rows=input_rows, clean_events=clean_events, episodes=episodes, mart_rows=mart_rows, eligible_rows=eligible_rows, source_data_kind=source_data_kind, population_estimate_allowed=population_estimate_allowed, ) except Exception: if connection is not None: connection.close() _remove_partial(partial_paths) raise def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--input", "--batch-dir", dest="input_path", type=Path, default=DEFAULT_INPUT, help="A bounded-batch directory or one development history CSV.gz.", ) parser.add_argument("--database", type=Path, default=DEFAULT_DATABASE) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) parser.add_argument("--episode-gap-days", type=int, default=30) parser.add_argument("--memory-limit", default="4GB") parser.add_argument( "--threads", type=int, default=max(1, min(4, os.cpu_count() or 1)) ) parser.add_argument("--allow-gaps", action="store_true") parser.add_argument("--allow-external-output", action="store_true") parser.add_argument("--overwrite", action="store_true") return parser.parse_args(argv) def main(argv: Optional[Sequence[str]] = None) -> int: args = parse_args(argv) config = BuildConfig( input_path=args.input_path, database_path=args.database, output_path=args.output, episode_gap_days=args.episode_gap_days, memory_limit=args.memory_limit, threads=args.threads, allow_gaps=args.allow_gaps, allow_external_output=args.allow_external_output, overwrite=args.overwrite, ) try: summary = build_feature_mart(config) except (MartBuildError, duckdb.Error, OSError) as exc: print(f"Feature-mart build failed: {exc}", file=sys.stderr) return 1 print( f"Built {summary.mart_rows:,} episode rows " f"({summary.eligible_rows:,} eligible returning targets)" ) print(f"Private DuckDB warehouse: {summary.database_path}") print(f"Private Parquet mart: {summary.output_path}") print(f"Build manifest: {summary.manifest_path}") if not summary.population_estimate_allowed: print("Development sample: population estimates are prohibited.") return 0 if __name__ == "__main__": raise SystemExit(main())