from __future__ import annotations from pathlib import Path from typing import Any import yaml from ingestion.common.classification import classify_table from ingestion.common.contract_overrides import apply_source_contract_overrides from ingestion.common.models import Column, Table from ingestion.common.sql import sql_string from .storage import StorageAdapter AUTHORED_SCHEMA_VERSION = "0.1.0" PRODUCT_VERSION = "0.1.0" def ensure_stage(cursor, storage: StorageAdapter, config) -> None: """(Re)create the external stage used for INFER_SCHEMA and COPY. CREATE OR REPLACE so a rotated credential is picked up; the credential lives server-side in the stage.""" cursor.execute(storage.create_stage_sql(config.stage_name, config.external_uri)) def infer_columns(cursor, stage_name: str, sample_object: str) -> list[dict[str, Any]]: """Infer one sample object's schema via Databend INFER_SCHEMA. Returns ``[{name, type, nullable}]`` in file order; ``type`` is the Databend type, exactly what CREATE TABLE + COPY need.""" import re pattern = ".*" + re.escape(sample_object) cursor.execute( "SELECT column_name, type, nullable FROM " f"INFER_SCHEMA(location => '@{stage_name}', pattern => {sql_string(pattern)})" ) columns = [] for row in cursor.fetchall(): values = list(row.values()) if hasattr(row, "values") else list(row) columns.append({"name": str(values[0]), "type": str(values[1]), "nullable": bool(values[2])}) if not columns: raise RuntimeError(f"INFER_SCHEMA returned no columns for {sample_object}") return columns def source_contract(config, columns: list[dict[str, Any]]) -> dict[str, Any]: """Build the provided `source` contract for the externally-managed bucket, bootstrapped by inference (the "source contract provided" case, ADR 0015/0018).""" table = config.external_table slug = table.lower() prefix = config.external_uri.split("://", 1)[1].split("/", 1)[1].rstrip("/") model = Table( database=config.external_bucket, name=table, rows_estimate=None, columns=[ Column( name=c["name"], data_type=c["type"].split("(")[0].strip().lower(), column_type=c["type"], nullable=c["nullable"], ordinal_position=i, ) for i, c in enumerate(columns, start=1) ], primary_key=config.external_primary_key, ) classifications = classify_table(model, config.classification_rules) contract = { "schema_version": AUTHORED_SCHEMA_VERSION, "kind": "source", "id": f"source.{config.name}.{slug}_external", "name": f"{config.name} {table} external snapshot", "version": PRODUCT_VERSION, "owner": {"type": "team", "name": config.external_owner}, "domain": {"name": config.external_domain}, "purpose": ( f"Source-faithful externally-produced Parquet export of {table} landed at " f"{config.external_uri}, consumed by the lakehouse load stage. Schema is inferred " "from a sample object; drift routes to a contract update process." ), "usage": { "intended": ["Bulk load into the raw lakehouse layer", "Audit and provenance of exported snapshots"], "prohibited": [ "Direct analytical querying (load into a contracted product first).", "Treating a single snapshot as an intra-day event stream.", ], }, "source": { "kind": "file", "system": config.external_system, "description": f"Externally-produced object-store snapshot export at {config.external_uri}", "identifiers": {"bucket": config.external_bucket, "prefix": prefix, "table": table}, }, "assets": [ { "id": f"{slug}_external_snapshot", "kind": "file", "ref": config.external_uri, "materialization": "external", "produced_by": {"kind": "other", "ref": "external (not produced by this platform)"}, } ], "schema": { "kind": "relational", "columns": [ {"name": c["name"].lower(), "type": c["type"], "nullable": c["nullable"]} for c in columns ], }, "freshness": {"max_age": "PT48H", "degrade_after": "PT30H"}, "tests": [ { "id": f"{slug}_source_smoke", "type": "sql", "severity": "informational", "description": ( f"Placeholder informational smoke check for the source contract of {slug}. " "Replace with concrete row-count, freshness, or key-uniqueness assertions." ), "sql": "select 1", } ], "governance": { "classifications": classifications, "access": {"default": "restricted"}, "retention": {"period": "P30D"}, "audit": {"required": True, "events": ["export", "agent_action"]}, }, "agent_permissions": {"default": "deny"}, "compatibility": {"change_policy": "semver"}, "extensions": { "primary_key": config.external_primary_key, "watermark_columns": [], "known_blind_spots": [ "Intra-snapshot changes are collapsed.", "Schema is inferred from a sample object; a change routes to a contract update process, not a silent overwrite.", "Primary key is declared in load source config (the external producer does not enforce one).", ], }, } overrides = getattr(config, "external_source_contract_overrides", {}) or {} return apply_source_contract_overrides(contract, overrides) def committed_path(config) -> Path: return config.root / "contracts" / "upstream" / f"{config.name}.{config.external_table}.external.source.contract.yaml" def proposed_path(config) -> Path: return config.root / "contracts" / "proposed" / f"{config.name}.{config.external_table}.external.source.contract.yaml" def write_contract(path: Path, contract: dict[str, Any]) -> Path: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(yaml.safe_dump(contract, sort_keys=False, allow_unicode=False)) return path def committed_columns(config) -> list[dict[str, Any]] | None: path = committed_path(config) if not path.exists(): return None contract = yaml.safe_load(path.read_text()) return contract.get("schema", {}).get("columns", []) def schema_changed(live: list[dict[str, Any]], committed: list[dict[str, Any]]) -> bool: """Compare inferred schema to the committed contract by (name, type, nullable), in order.""" def norm(cols): return [(c["name"].lower(), c["type"], bool(c.get("nullable", True))) for c in cols] return norm(live) != norm(committed)