from __future__ import annotations import json import zlib from dataclasses import dataclass from typing import Any from errors import IncrementalError, SchemaChecksumError from metadata import Column, Table, TableName from .paths import SchemaPath DECIMAL_BASE_TYPES = {"DECIMAL", "NUMERIC", "NEWDECIMAL"} @dataclass(frozen=True) class TableDefinition: table: Table table_version: int ddl_type: int query: str @property def name(self) -> TableName: return self.table.name @property def source_schema(self) -> str: return self.table.source_schema @property def source_table(self) -> str: return self.table.source_table @property def columns(self) -> tuple[Column, ...]: return self.table.columns @property def quoted(self) -> str: return self.table.quoted @property def primary_key(self) -> tuple[str, ...]: return self.table.primary_key @property def is_truncate(self) -> bool: query = self.query.strip().lower() return self.ddl_type in {11, 23} or query.startswith("truncate table") @dataclass(frozen=True) class DDLAction: name: str operation: str should_warn: bool def classify_ddl_action(table_def: TableDefinition) -> DDLAction: query = " ".join(table_def.query.strip().lower().split()) if table_def.is_truncate: return DDLAction(name="truncate_table", operation="truncate", should_warn=False) if query.startswith("drop table"): return DDLAction(name="drop_table", operation="truncate", should_warn=False) if not query: return DDLAction(name="schema_snapshot", operation="schema_update", should_warn=False) known_prefixes = { "alter table": "alter_table", "create table": "create_table", "rename table": "rename_table", "create database": "create_database", "drop database": "drop_database", "alter database": "alter_database", "rename database": "rename_database", } for prefix, name in known_prefixes.items(): if query.startswith(prefix): return DDLAction(name=name, operation="schema_update", should_warn=False) return DDLAction(name="unknown_ddl", operation="schema_update", should_warn=False) def _column_name(raw: dict[str, Any]) -> str: for name in ("Name", "ColumnName", "ColumnNameO", "name"): value = raw.get(name) if value is not None and str(value): return str(value) raise IncrementalError(f"incremental schema column is missing a name: {raw!r}") def _column_type(raw: dict[str, Any]) -> str: for name in ("Type", "ColumnType", "source_type", "type"): value = raw.get(name) if value is not None and str(value): return str(value) # Some schema encodings may carry only numeric type codes. return "STRING" def _column_nullable(raw: dict[str, Any]) -> bool: value = raw.get("ColumnNullable", raw.get("Nullable")) if value is None: return True return str(value).strip().lower() not in {"0", "false", "no"} def _column_primary_key(raw: dict[str, Any]) -> bool: return _truthy_column_flag( raw, "ColumnIsPk", "IsPk", "PrimaryKey", "IsPrimaryKey", "ColumnPrimaryKey", ) def _column_from_incremental_schema(raw: dict[str, Any]) -> Column: source_type = _column_type(raw) precision, scale = _column_precision_scale(raw, source_type) return Column( name=_column_name(raw), source_type=source_type, nullable=_column_nullable(raw), precision=precision, scale=scale, primary_key=_column_primary_key(raw), ) def _column_precision_scale(raw: dict[str, Any], source_type: str) -> tuple[int | None, int | None]: base = _base_column_type(source_type) if base in {"BIT", "TINYINT"}: return _optional_int_column_field(raw, "ColumnPrecision"), None if base in DECIMAL_BASE_TYPES: precision = _optional_int_column_field(raw, "ColumnPrecision") scale = _optional_int_column_field(raw, "ColumnScale") return (10 if precision is None else precision), (0 if scale is None else scale) return None, None def _base_column_type(source_type: str) -> str: raw = source_type.strip().upper() before_params = raw.split("(", 1)[0] parts = before_params.split(maxsplit=1) if not parts: return raw return parts[0] def _optional_int_column_field(raw: dict[str, Any], name: str) -> int | None: value = raw.get(name) if value in (None, ""): return None try: return int(value) except (TypeError, ValueError) as exc: column_name = raw.get("ColumnName") or raw.get("Name") or "" raise IncrementalError(f"incremental schema column {column_name!r} has invalid {name}: {value!r}") from exc def _truthy_column_flag(raw: dict[str, Any], *names: str) -> bool: for name in names: value = raw.get(name) if isinstance(value, bool): return value if value is None: continue if str(value).strip().lower() in {"1", "true", "yes"}: return True return False def _table_columns(raw: dict[str, Any]) -> list[dict[str, Any]]: columns = raw.get("TableColumns", raw.get("Columns", [])) if columns is None: return [] if not isinstance(columns, list): raise IncrementalError("incremental schema TableColumns must be a list") if not all(isinstance(column, dict) for column in columns): raise IncrementalError("incremental schema TableColumns entries must be objects") return columns def _canonical_checksum_bytes(raw: dict[str, Any]) -> bytes: columns = sorted(_table_columns(raw), key=_column_name) total_columns = raw.get("TotalColumns", raw.get("TableColumnsTotal", len(columns))) checksum_obj = { "Table": raw.get("Table", ""), "Schema": raw.get("Schema", ""), # storage sink tableDefWithoutQuery leaves Version at the Go zero value # while excluding Query, Type, and TableVersion from the checksum. "Version": 0, "TableColumns": columns, "TableColumnsTotal": total_columns, } # storage sink computes the schema filename checksum from # json.MarshalIndent(defWithoutQuery, "", " "). encoded = json.dumps(checksum_obj, ensure_ascii=False, indent=4, separators=(",", ": ")) return encoded.encode("utf-8") def compute_table_definition_checksum(raw: dict[str, Any]) -> int: return zlib.crc32(_canonical_checksum_bytes(raw)) & 0xFFFFFFFF def load_table_definition(raw: dict[str, Any], path: SchemaPath) -> TableDefinition: if not isinstance(raw, dict): raise IncrementalError(f"incremental schema file must contain a JSON object: {path.key}") try: table_version = int(raw["TableVersion"]) except (KeyError, TypeError, ValueError) as exc: raise IncrementalError(f"incremental schema file is missing numeric TableVersion: {path.key}") from exc if table_version != path.table_version: raise IncrementalError(f"TableVersion mismatch for {path.key}: path={path.table_version} json={table_version}") actual_checksum = compute_table_definition_checksum(raw) if actual_checksum != path.checksum: raise SchemaChecksumError( f"schema checksum mismatch for {path.key}: path={path.checksum} computed={actual_checksum}" ) if str(raw.get("Schema")) != path.source_schema or str(raw.get("Table")) != path.source_table: raise IncrementalError(f"schema file path and JSON table identity differ for {path.key}") columns = tuple(_column_from_incremental_schema(column) for column in _table_columns(raw)) primary_key = tuple(column.name for column in columns if column.primary_key) return TableDefinition( table=Table( name=TableName(source_schema=path.source_schema, source_table=path.source_table), primary_key=primary_key, columns=columns, ), table_version=table_version, ddl_type=int(raw.get("Type") or 0), query=str(raw.get("Query") or ""), ) __all__ = [ "DDLAction", "TableDefinition", "classify_ddl_action", "compute_table_definition_checksum", "load_table_definition", ]