from __future__ import annotations import json from datetime import UTC, datetime from pathlib import Path from typing import Any from google.api_core.exceptions import NotFound from google.cloud import bigquery from .config import PipelineConfig from .gcs import download_text, list_manifest_uris, upload_text from .naming import bq_path, quote_name, run_id_token, safe_table_id, sql_string, table_prefix from .source_inspect import normalize_exact_tidb_selection, table_selection_contract RAW_METADATA_SCHEMA = [ bigquery.SchemaField("_snapshot_date", "DATE"), bigquery.SchemaField("_snapshot_run_id", "STRING"), bigquery.SchemaField("_snapshot_loaded_at", "TIMESTAMP"), bigquery.SchemaField("_source_name", "STRING"), bigquery.SchemaField("_source_database", "STRING"), bigquery.SchemaField("_source_table", "STRING"), bigquery.SchemaField("_source_manifest_uri", "STRING"), ] FINAL_METADATA_SCHEMA = [ bigquery.SchemaField("_record_loaded_at", "TIMESTAMP"), bigquery.SchemaField("_record_source_run_id", "STRING"), bigquery.SchemaField("_record_source_manifest_uri", "STRING"), bigquery.SchemaField("_record_watermark_at", "TIMESTAMP"), ] def run_load( config: PipelineConfig, *, manifest_uri: str | None = None, latest: bool = False, all_unloaded: bool = False, tables: list[str] | None = None, ) -> list[dict[str, Any]]: client = bigquery.Client(project=config.target_project, location=config.location) if manifest_uri: return [load_manifest_uri(client, config, manifest_uri, expected_tables=tables)] ensure_control_tables(client, config) manifests = manifest_records(config) candidates = unresolved_manifest_uris(client, config, manifests) if latest: if candidates: return [load_manifest_uri(client, config, candidates[-1], expected_tables=tables)] return [no_new_manifest_result(config, mode="latest", manifests=manifests)] if all_unloaded: if candidates: return [load_manifest_uri(client, config, uri, expected_tables=tables) for uri in candidates] return [no_new_manifest_result(config, mode="all_unloaded", manifests=manifests)] raise ValueError("provide --manifest, --latest, or --all-unloaded") def unresolved_manifests(client: bigquery.Client, config: PipelineConfig) -> list[str]: return unresolved_manifest_uris(client, config, manifest_records(config)) def manifest_records(config: PipelineConfig) -> list[tuple[str, dict[str, Any]]]: prefix = f"gs://{config.bucket}/{config.manifest_prefix}/source={config.name}/" candidates = [uri for uri, _updated in list_manifest_uris(prefix)] manifests = [(uri, read_manifest(uri)) for uri in candidates] return sort_manifest_records(manifests) def unresolved_manifest_uris( client: bigquery.Client, config: PipelineConfig, manifests: list[tuple[str, dict[str, Any]]], ) -> list[str]: unresolved = [] for uri, manifest in manifests: if not is_run_loaded(client, config, manifest["run_id"]): unresolved.append(uri) return unresolved def no_new_manifest_result( config: PipelineConfig, *, mode: str, manifests: list[tuple[str, dict[str, Any]]], now: datetime | None = None, ) -> dict[str, Any]: return { "status": "no_new_manifest", "reason": "no_unloaded_manifest", "source_name": config.name, "mode": mode, "candidate_manifest_count": 0, "processed_manifest_count": 0, "freshness": manifest_freshness(config, manifests=manifests, now=now), } def manifest_freshness( config: PipelineConfig, *, manifests: list[tuple[str, dict[str, Any]]], now: datetime | None = None, ) -> dict[str, Any]: max_age_seconds = config.manifest_freshness_max_age_seconds if not manifests: return { "signal": "latest_published_manifest_age", "status": "unknown", "reason": "no_published_manifest", "max_age_seconds": max_age_seconds, } latest_uri, latest = manifests[-1] created_at = parse_manifest_created_at(latest.get("created_at")) result: dict[str, Any] = { "signal": "latest_published_manifest_age", "latest_manifest_uri": latest_uri, "latest_run_id": latest.get("run_id"), "latest_run_date": latest.get("run_date"), "latest_manifest_created_at": latest.get("created_at"), "max_age_seconds": max_age_seconds, } if created_at is None: return {**result, "status": "unknown", "reason": "manifest_created_at_missing_or_invalid"} observed_at = now or datetime.now(UTC) if observed_at.tzinfo is None: observed_at = observed_at.replace(tzinfo=UTC) age_seconds = max(0, int((observed_at.astimezone(UTC) - created_at).total_seconds())) result["age_seconds"] = age_seconds if max_age_seconds is None: return {**result, "status": "unconfigured", "reason": "freshness_threshold_not_configured"} return { **result, "status": "fresh" if age_seconds <= max_age_seconds else "stale", "reason": "within_max_age" if age_seconds <= max_age_seconds else "exceeds_max_age", } def parse_manifest_created_at(value: Any) -> datetime | None: if not isinstance(value, str) or not value.strip(): return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed.astimezone(UTC) def sort_manifest_records(manifests: list[tuple[str, dict[str, Any]]]) -> list[tuple[str, dict[str, Any]]]: return sorted( manifests, key=lambda item: ( str(item[1].get("run_date", "")), str(item[1].get("run_id", "")), item[0], ), ) def load_manifest_uri( client: bigquery.Client, config: PipelineConfig, manifest_uri: str, *, expected_tables: list[str] | None = None, ) -> dict[str, Any]: manifest = read_manifest(manifest_uri) validate_manifest_selection_contract(config, manifest, expected_tables=expected_tables) ensure_control_tables(client, config) run_id = manifest["run_id"] runtime_events: list[dict[str, Any]] = [] mark_run(client, config, manifest, status="running") try: results = [] for table in manifest.get("tables", []): result = load_table(client, config, manifest, table) results.append(result) runtime_events.append({"event_type": "table_loaded", **result}) mark_run(client, config, manifest, status="loaded") status = "loaded" return {"run_id": run_id, "manifest_uri": manifest_uri, "status": status, "tables": results} except Exception as error: mark_run(client, config, manifest, status="failed", error_message=str(error)) runtime_events.append({"event_type": "run_failed", "status": "failed", "error": str(error)}) raise finally: write_runtime_log(config, manifest, runtime_events) def validate_manifest_selection_contract( config: PipelineConfig, manifest: dict[str, Any], *, expected_tables: list[str] | None = None, ) -> None: if manifest.get("source_name") != config.name: raise RuntimeError( f"manifest source {manifest.get('source_name')!r} does not match configured source {config.name!r}" ) if manifest.get("source_database") != config.source_database: raise RuntimeError( f"manifest database {manifest.get('source_database')!r} does not match " f"configured database {config.source_database!r}" ) selection = manifest.get("selection") schema_version = int(manifest.get("manifest_schema_version", 0) or 0) if selection is None: if expected_tables is not None: raise RuntimeError("operator-bound expected tables require a schema-v3 selection contract") if schema_version >= 3: raise RuntimeError("manifest schema v3 requires a table selection contract") return if not isinstance(selection, dict): raise RuntimeError("manifest table selection contract must be an object") manifest_tables = manifest.get("tables") if not isinstance(manifest_tables, list) or not manifest_tables or not all( isinstance(table, dict) for table in manifest_tables ): raise RuntimeError("manifest tables must be a non-empty object list") table_names = [str(table.get("table_name", "")) for table in manifest_tables] if any(not name for name in table_names): raise RuntimeError("manifest contains a table without table_name") duplicate_tables = sorted({name for name in table_names if table_names.count(name) > 1}) if duplicate_tables: raise RuntimeError(f"manifest contains duplicate selected tables: {duplicate_tables}") resolved = selection.get("resolved_tables") if not isinstance(resolved, list) or not resolved or not all(isinstance(name, str) and name for name in resolved): raise RuntimeError("manifest selection resolved_tables must be a non-empty string list") if sorted(table_names) != sorted(resolved) or len(table_names) != len(resolved): raise RuntimeError( "manifest tables do not match the selection contract; " f"manifest_tables={sorted(table_names)}, resolved_tables={sorted(resolved)}" ) if expected_tables is not None: expected_names = normalize_exact_tidb_selection(config, expected_tables) if sorted(expected_names) != sorted(table_names) or len(expected_names) != len(table_names): raise RuntimeError( "manifest tables do not match the operator-bound expected subset; " f"expected_tables={sorted(expected_names)}, manifest_tables={sorted(table_names)}" ) mode = selection.get("mode") requested = selection.get("requested_tables") if mode == "explicit_subset" else None if mode not in {"configured", "explicit_subset"}: raise RuntimeError(f"unsupported manifest table selection mode: {mode!r}") if mode == "explicit_subset" and ( not isinstance(requested, list) or not requested or not all(isinstance(name, str) and name for name in requested) ): raise RuntimeError("explicit subset manifest requires a non-empty string requested_tables list") if mode == "configured" and selection.get("requested_tables") is not None: raise RuntimeError("configured manifest selection must not contain requested_tables") if expected_tables is not None and mode != "explicit_subset": raise RuntimeError("operator-bound expected tables require an explicit_subset manifest") expected = table_selection_contract(config, table_names, requested=requested) if mode == "explicit_subset" and ( sorted(expected["requested_tables"]) != sorted(table_names) or len(expected["requested_tables"]) != len(table_names) ): raise RuntimeError( "manifest requested tables do not match its resolved subset; " f"requested_tables={expected['requested_tables']}, resolved_tables={sorted(table_names)}" ) for field in ("resolved_tables", "qualified_patterns", "qualified_patterns_sha256", "canonicalization"): if selection.get(field) != expected[field]: raise RuntimeError(f"manifest selection contract field {field} does not match manifest tables") drift = manifest.get("schema_drift", {}) if not isinstance(drift, dict): raise RuntimeError("manifest schema_drift must be an object") if schema_version >= 3 and ( not isinstance(drift.get("files_without_metadata"), list) or not isinstance(drift.get("metadata_without_files"), list) ): raise RuntimeError("manifest schema v3 requires both schema_drift coverage lists") if drift.get("files_without_metadata") or drift.get("metadata_without_files"): raise RuntimeError(f"manifest table coverage is not exact: {drift}") def load_table( client: bigquery.Client, config: PipelineConfig, manifest: dict[str, Any], table: dict[str, Any], ) -> dict[str, Any]: source_table = str(table["table_name"]) uris = list(table.get("uris", [])) if not uris: raise RuntimeError(f"manifest table {source_table} has no parquet URIs") prefix = table_prefix(config.name, source_table) raw_table_id = safe_table_id(f"{prefix}_raw") final_name = config.final_table_name_for(source_table) or prefix final_table_id = safe_table_id(final_name) if config.materialize_final else None stage_table_id = safe_table_id(f"__stg_{prefix}_{run_id_token(manifest['run_id'])}", max_length=1000) stage_ref = f"{config.target_project}.{config.control_dataset}.{stage_table_id}" raw_ref = f"{config.target_project}.{config.raw_dataset}.{raw_table_id}" final_ref = ( f"{config.target_project}.{config.final_dataset}.{final_table_id}" if config.materialize_final and config.final_dataset and final_table_id else None ) try: load_parquet_to_stage(client, config, stage_ref, uris) stage = client.get_table(stage_ref) data_schema = list(stage.schema) ensure_table(client, raw_ref, data_schema + RAW_METADATA_SCHEMA, partition_field="_snapshot_date") if final_ref: ensure_table(client, final_ref, data_schema + FINAL_METADATA_SCHEMA) clear_raw_run(client, config, raw_ref, manifest["run_id"]) insert_raw_run(client, config, raw_ref, stage_ref, data_schema, manifest, table) loaded_rows = count_raw_rows(client, config, raw_ref, manifest["run_id"]) duplicate_rows = count_duplicate_primary_keys(client, config, raw_ref, table, manifest["run_id"]) if duplicate_rows and config.materialize_final: record_table_load( client, config, manifest, table, raw_table=raw_table_id, final_table=final_table_id, loaded_rows=loaded_rows, final_rows=None, duplicate_rows=duplicate_rows, status="blocked", ) message = ( f"{config.name}.{source_table} has {duplicate_rows} duplicate primary-key rows " f"in run {manifest['run_id']}" ) raise RuntimeError(message) final_rows = None if final_ref: merge_final_table(client, config, raw_ref, final_ref, data_schema, manifest, table) final_rows = count_table_rows(client, config, final_ref) result = { "source_table": source_table, "raw_table": f"{config.raw_dataset}.{raw_table_id}", "final_table": f"{config.final_dataset}.{final_table_id}" if final_ref else None, "loaded_rows": loaded_rows, "final_rows": final_rows, "duplicate_primary_key_rows": duplicate_rows, "status": "loaded", } record_table_load(client, config, manifest, table, **result) return result finally: client.delete_table(stage_ref, not_found_ok=True) def load_parquet_to_stage(client: bigquery.Client, config: PipelineConfig, stage_ref: str, uris: list[str]) -> None: job_config = bigquery.LoadJobConfig( source_format=bigquery.SourceFormat.PARQUET, autodetect=True, decimal_target_types=[bigquery.DecimalTargetType.BIGNUMERIC], write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE, ) job = client.load_table_from_uri(uris, stage_ref, job_config=job_config, location=config.location) job.result() def ensure_control_tables(client: bigquery.Client, config: PipelineConfig) -> None: run_table = bq_path(config.target_project, config.control_dataset, "pipeline_runs") table_loads = bq_path(config.target_project, config.control_dataset, "table_loads") query(client, config, f""" CREATE TABLE IF NOT EXISTS {run_table} ( source_name STRING NOT NULL, run_id STRING NOT NULL, run_date DATE, manifest_uri STRING, status STRING, started_at TIMESTAMP, ended_at TIMESTAMP, updated_at TIMESTAMP, error_message STRING ) """) query(client, config, f""" CREATE TABLE IF NOT EXISTS {table_loads} ( source_name STRING NOT NULL, run_id STRING NOT NULL, source_table STRING NOT NULL, raw_table STRING, final_table STRING, loaded_rows INT64, final_rows INT64, duplicate_primary_key_rows INT64, status STRING, metrics_json STRING, created_at TIMESTAMP ) """) def ensure_table( client: bigquery.Client, table_ref: str, schema: list[bigquery.SchemaField], *, partition_field: str | None = None, ) -> None: try: table = client.get_table(table_ref) except NotFound: table = bigquery.Table(table_ref, schema=schema) if partition_field: table.time_partitioning = bigquery.TimePartitioning(field=partition_field) client.create_table(table) return desired = {field.name.lower(): field for field in schema} existing = {field.name.lower() for field in table.schema} missing = [ nullable_field(field) if field.mode == "REQUIRED" else field for field in schema if field.name.lower() not in existing ] updated_schema = [ relax_required_field(field, desired[field.name.lower()]) if field.name.lower() in desired else field for field in table.schema ] if missing or updated_schema != list(table.schema): table.schema = updated_schema + missing client.update_table(table, ["schema"]) def relax_required_field(existing: bigquery.SchemaField, desired: bigquery.SchemaField) -> bigquery.SchemaField: if existing.mode != "REQUIRED" or desired.mode == "REQUIRED": return existing api_repr = existing.to_api_repr() api_repr["mode"] = "NULLABLE" return bigquery.SchemaField.from_api_repr(api_repr) def nullable_field(field: bigquery.SchemaField) -> bigquery.SchemaField: api_repr = field.to_api_repr() api_repr["mode"] = "NULLABLE" return bigquery.SchemaField.from_api_repr(api_repr) def clear_raw_run(client: bigquery.Client, config: PipelineConfig, raw_ref: str, run_id: str) -> None: query(client, config, f"DELETE FROM {table_path(raw_ref)} WHERE _snapshot_run_id = {sql_string(run_id)}") def insert_raw_run( client: bigquery.Client, config: PipelineConfig, raw_ref: str, stage_ref: str, data_schema: list[bigquery.SchemaField], manifest: dict[str, Any], table: dict[str, Any], ) -> None: data_columns = [field.name for field in data_schema] target_columns = data_columns + [field.name for field in RAW_METADATA_SCHEMA] select_exprs = [quote_name(name) for name in data_columns] + [ f"DATE {sql_string(manifest['run_date'])}", sql_string(manifest["run_id"]), "CURRENT_TIMESTAMP()", sql_string(manifest["source_name"]), sql_string(manifest.get("source_database", "")), sql_string(table["table_name"]), sql_string(manifest["manifest_uri"]), ] query(client, config, f""" INSERT INTO {table_path(raw_ref)} ({", ".join(quote_name(name) for name in target_columns)}) SELECT {", ".join(select_exprs)} FROM {table_path(stage_ref)} """) def merge_final_table( client: bigquery.Client, config: PipelineConfig, raw_ref: str, final_ref: str, data_schema: list[bigquery.SchemaField], manifest: dict[str, Any], table: dict[str, Any], ) -> None: primary_key = list(table.get("primary_key", [])) data_columns = [field.name for field in data_schema] if not primary_key: replace_final_without_key(client, config, raw_ref, final_ref, data_columns, manifest, table) return final_columns = data_columns + [field.name for field in FINAL_METADATA_SCHEMA] update_columns = data_columns + [field.name for field in FINAL_METADATA_SCHEMA] update_assignments = [f"T.{quote_name(name)} = S.{quote_name(name)}" for name in update_columns] query(client, config, f""" MERGE {table_path(final_ref)} T USING ( {latest_raw_select(raw_ref, data_columns, manifest, table)} ) S ON {join_predicate("T", "S", primary_key)} WHEN MATCHED THEN UPDATE SET {", ".join(update_assignments)} WHEN NOT MATCHED THEN INSERT ({", ".join(quote_name(name) for name in final_columns)}) VALUES ({", ".join(f"S.{quote_name(name)}" for name in final_columns)}) """) if config.full_snapshot_delete: delete_absent_from_final(client, config, raw_ref, final_ref, manifest["run_id"], primary_key) def latest_raw_select( raw_ref: str, data_columns: list[str], manifest: dict[str, Any], table: dict[str, Any], ) -> str: metadata_exprs = [ "CURRENT_TIMESTAMP() AS _record_loaded_at", f"{sql_string(manifest['run_id'])} AS _record_source_run_id", f"{sql_string(manifest['manifest_uri'])} AS _record_source_manifest_uri", f"{watermark_expression(data_columns, table)} AS _record_watermark_at", ] select_columns = [quote_name(name) for name in data_columns] + metadata_exprs primary_key = list(table.get("primary_key", [])) return f""" SELECT {", ".join(select_columns)} FROM ( SELECT {", ".join(quote_name(name) for name in data_columns)}, ROW_NUMBER() OVER ( PARTITION BY {", ".join(quote_name(name) for name in primary_key)} ORDER BY {watermark_order(data_columns, table)}, _snapshot_loaded_at DESC ) AS _rn FROM {table_path(raw_ref)} WHERE _snapshot_run_id = {sql_string(manifest["run_id"])} ) WHERE _rn = 1 """ def replace_final_without_key( client: bigquery.Client, config: PipelineConfig, raw_ref: str, final_ref: str, data_columns: list[str], manifest: dict[str, Any], table: dict[str, Any], ) -> None: select_columns = [quote_name(name) for name in data_columns] + [ "CURRENT_TIMESTAMP() AS _record_loaded_at", f"{sql_string(manifest['run_id'])} AS _record_source_run_id", f"{sql_string(manifest['manifest_uri'])} AS _record_source_manifest_uri", f"{watermark_expression(data_columns, table)} AS _record_watermark_at", ] query(client, config, f""" CREATE OR REPLACE TABLE {table_path(final_ref)} AS SELECT {", ".join(select_columns)} FROM {table_path(raw_ref)} WHERE _snapshot_run_id = {sql_string(manifest["run_id"])} """) def delete_absent_from_final( client: bigquery.Client, config: PipelineConfig, raw_ref: str, final_ref: str, run_id: str, primary_key: list[str], ) -> None: query(client, config, f""" DELETE FROM {table_path(final_ref)} T WHERE NOT EXISTS ( SELECT 1 FROM {table_path(raw_ref)} R WHERE R._snapshot_run_id = {sql_string(run_id)} AND {join_predicate("T", "R", primary_key)} ) """) def watermark_expression(data_columns: list[str], table: dict[str, Any]) -> str: available = column_lookup(data_columns) parts = [ f"SAFE_CAST({quote_name(available[name.lower()])} AS TIMESTAMP)" for name in table.get("watermark_columns", []) if name.lower() in available ] if not parts: return "CAST(NULL AS TIMESTAMP)" return f"COALESCE({', '.join(parts)})" def watermark_order(data_columns: list[str], table: dict[str, Any]) -> str: available = column_lookup(data_columns) parts = [ f"SAFE_CAST({quote_name(available[name.lower()])} AS TIMESTAMP) DESC NULLS LAST" for name in table.get("watermark_columns", []) if name.lower() in available ] return ", ".join(parts) if parts else "1 DESC" def column_lookup(columns: list[str]) -> dict[str, str]: return {name.lower(): name for name in columns} def join_predicate(left_alias: str, right_alias: str, primary_key: list[str]) -> str: return " AND ".join( f"{left_alias}.{quote_name(column)} = {right_alias}.{quote_name(column)}" for column in primary_key ) def count_duplicate_primary_keys( client: bigquery.Client, config: PipelineConfig, raw_ref: str, table: dict[str, Any], run_id: str, ) -> int: primary_key = list(table.get("primary_key", [])) if not primary_key: return 0 key_expr = ", ".join(quote_name(column) for column in primary_key) rows = query(client, config, f""" SELECT COALESCE(SUM(cnt - 1), 0) AS duplicate_rows FROM ( SELECT {key_expr}, COUNT(*) AS cnt FROM {table_path(raw_ref)} WHERE _snapshot_run_id = {sql_string(run_id)} GROUP BY {key_expr} HAVING COUNT(*) > 1 ) """) return int(next(iter(rows)).duplicate_rows) def count_raw_rows(client: bigquery.Client, config: PipelineConfig, raw_ref: str, run_id: str) -> int: rows = query( client, config, f"SELECT COUNT(*) AS row_count FROM {table_path(raw_ref)} WHERE _snapshot_run_id = {sql_string(run_id)}", ) return int(next(iter(rows)).row_count) def count_table_rows(client: bigquery.Client, config: PipelineConfig, table_ref: str) -> int: rows = query(client, config, f"SELECT COUNT(*) AS row_count FROM {table_path(table_ref)}") return int(next(iter(rows)).row_count) def is_run_loaded(client: bigquery.Client, config: PipelineConfig, run_id: str) -> bool: rows = query(client, config, f""" SELECT COUNT(*) AS loaded FROM {bq_path(config.target_project, config.control_dataset, "pipeline_runs")} WHERE source_name = {sql_string(config.name)} AND run_id = {sql_string(run_id)} AND status = 'loaded' """) return int(next(iter(rows)).loaded) > 0 def mark_run( client: bigquery.Client, config: PipelineConfig, manifest: dict[str, Any], *, status: str, error_message: str = "", ) -> None: ended_at = "CURRENT_TIMESTAMP()" if status in {"loaded", "failed"} else "CAST(NULL AS TIMESTAMP)" query(client, config, f""" MERGE {bq_path(config.target_project, config.control_dataset, "pipeline_runs")} T USING ( SELECT {sql_string(config.name)} AS source_name, {sql_string(manifest["run_id"])} AS run_id, DATE {sql_string(manifest["run_date"])} AS run_date, {sql_string(manifest["manifest_uri"])} AS manifest_uri, {sql_string(status)} AS status, CURRENT_TIMESTAMP() AS now_ts, {ended_at} AS ended_at, {sql_string(error_message)} AS error_message ) S ON T.source_name = S.source_name AND T.run_id = S.run_id WHEN MATCHED THEN UPDATE SET manifest_uri = S.manifest_uri, status = S.status, ended_at = S.ended_at, updated_at = S.now_ts, error_message = S.error_message WHEN NOT MATCHED THEN INSERT ( source_name, run_id, run_date, manifest_uri, status, started_at, ended_at, updated_at, error_message ) VALUES ( S.source_name, S.run_id, S.run_date, S.manifest_uri, S.status, S.now_ts, S.ended_at, S.now_ts, S.error_message ) """) def record_table_load( client: bigquery.Client, config: PipelineConfig, manifest: dict[str, Any], table: dict[str, Any], *, source_table: str | None = None, raw_table: str, final_table: str | None, loaded_rows: int, final_rows: int | None = None, duplicate_rows: int = 0, duplicate_primary_key_rows: int | None = None, status: str, **metrics: Any, ) -> None: duplicates = duplicate_primary_key_rows if duplicate_primary_key_rows is not None else duplicate_rows metrics_json = json.dumps(metrics, sort_keys=True) source_table = source_table or str(table["table_name"]) query(client, config, f""" INSERT INTO {bq_path(config.target_project, config.control_dataset, "table_loads")} ( source_name, run_id, source_table, raw_table, final_table, loaded_rows, final_rows, duplicate_primary_key_rows, status, metrics_json, created_at ) VALUES ( {sql_string(config.name)}, {sql_string(manifest["run_id"])}, {sql_string(source_table)}, {sql_string(raw_table)}, {sql_string(final_table) if final_table is not None else "NULL"}, {loaded_rows}, {final_rows if final_rows is not None else "NULL"}, {duplicates}, {sql_string(status)}, {sql_string(metrics_json)}, CURRENT_TIMESTAMP() ) """) def read_manifest(uri: str) -> dict[str, Any]: if uri.startswith("gs://"): return json.loads(download_text(uri)) return json.loads(Path(uri).read_text()) def write_runtime_log(config: PipelineConfig, manifest: dict[str, Any], events: list[dict[str, Any]]) -> None: if not events: return created_at = datetime.now(UTC).isoformat() lines = [ json.dumps( { "event_schema_version": 1, "source_name": config.name, "run_id": manifest["run_id"], "run_date": manifest["run_date"], "created_at": created_at, **event, }, sort_keys=True, ) for event in events ] upload_text(manifest["runtime_log_uri"], "\n".join(lines) + "\n", content_type="application/jsonl") def table_path(table_ref: str) -> str: project, dataset, table = table_ref.split(".", 2) return bq_path(project, dataset, table) def query(client: bigquery.Client, config: PipelineConfig, sql: str): return client.query(sql, location=config.location).result()