diff --git a/omop_alchemy/backends/__init__.py b/omop_alchemy/backends/__init__.py index a1cb975..f0f980e 100644 --- a/omop_alchemy/backends/__init__.py +++ b/omop_alchemy/backends/__init__.py @@ -4,9 +4,7 @@ CONCEPT_NAME_TSVECTOR_COLUMN, CONCEPT_SYNONYM_NAME_TSVECTOR_COLUMN, FeatureNotSupportedError, - FullTextAction, FullTextError, - FullTextResult, FullTextTargetConfig, backend_supports, require_backend_support, @@ -22,9 +20,7 @@ "CONCEPT_NAME_TSVECTOR_COLUMN", "CONCEPT_SYNONYM_NAME_TSVECTOR_COLUMN", "FeatureNotSupportedError", - "FullTextAction", "FullTextError", - "FullTextResult", "FullTextTargetConfig", "backend_supports", "require_backend_support", diff --git a/omop_alchemy/backends/base.py b/omop_alchemy/backends/base.py index 0663723..8db8571 100644 --- a/omop_alchemy/backends/base.py +++ b/omop_alchemy/backends/base.py @@ -1,7 +1,6 @@ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass -from enum import StrEnum from typing import TYPE_CHECKING, Any import sqlalchemy as sa @@ -24,25 +23,6 @@ class FullTextTargetConfig: index_name: str -class FullTextAction(StrEnum): - INSTALL = "install" - POPULATE = "populate" - DROP = "drop" - - -@dataclass(frozen=True) -class FullTextResult: - target_name: str - table_name: str - source_column_name: str - vector_column_name: str - index_name: str - action: FullTextAction - status: str - detail: str - row_count: int | None = None - - class FullTextError(RuntimeError): """Raised when a full-text search maintenance operation fails.""" diff --git a/omop_alchemy/maintenance/_cli_utils.py b/omop_alchemy/maintenance/_cli_utils.py index 09ca063..3613c7f 100644 --- a/omop_alchemy/maintenance/_cli_utils.py +++ b/omop_alchemy/maintenance/_cli_utils.py @@ -5,10 +5,12 @@ import functools import inspect from dataclasses import dataclass +from enum import StrEnum from typing import Any, Callable, TypeVar import sqlalchemy as sa import typer +from orm_loader.backends import STAGING_SCHEMA from sqlalchemy.exc import SQLAlchemyError from .tables import TableScope @@ -19,6 +21,105 @@ _F = TypeVar("_F", bound=Callable[..., Any]) +class ReservedSchema(StrEnum): + """Schema names reserved for OMOP_Alchemy/orm-loader internal bookkeeping. + A user-configured db_schema may never collide with one of these. + """ + + STAGING = STAGING_SCHEMA + MAINTENANCE = "omop_alchemy_maintenance" + + +def reject_reserved_schema(db_schema: str | None) -> None: + """Raise if db_schema collides with a schema name reserved for internal bookkeeping.""" + if db_schema in set(ReservedSchema): + raise RuntimeError( + f"db_schema cannot be {db_schema!r}: reserved for OMOP_Alchemy/orm-loader internal use." + ) + + +class Severity(StrEnum): + """Coarse-grained outcome classification shared by every maintenance + command's status vocabulary. + + Parameters + ---------- + code : str + The severity's string value. + style : str + Rich color/style name used to render any Status with this severity. + """ + + def __new__(cls, code: str, style: str): + obj = str.__new__(cls, code) + obj._value_ = code + return obj + + def __init__(self, code: str, style: str): + self.style = style + + OK = ("ok", "green") + INFO = ("info", "cyan") + WARNING = ("warning", "yellow") + ERROR = ("error", "red") + + +class Status(StrEnum): + """A maintenance command result status, carrying its Severity (and, via + Severity.style, its render color). + + Parameters + ---------- + code : str + The status's string value (unchanged from the plain strings used + before this type existed, so existing string comparisons/dict + lookups by value keep working). + severity : Severity + How severe this status is, and (via severity.style) how it renders. + """ + + def __new__(cls, code: str, severity: Severity): + obj = str.__new__(cls, code) + obj._value_ = code + return obj + + def __init__(self, code: str, severity: Severity): + self.severity = severity + + # -- shared across every dry-run/apply-style domain -- + PLANNED = ("planned", Severity.INFO) + APPLIED = ("applied", Severity.OK) + SKIPPED = ("skipped", Severity.WARNING) + + # -- domain-specific "applied" words (backup, index restore/capture, + # sequence reset, vocab load) -- same OK severity as APPLIED, kept as + # distinct words since the specific outcome is worth seeing at a glance -- + CREATED = ("created", Severity.OK) + LOADED = ("loaded", Severity.OK) + RESET = ("reset", Severity.OK) + RESTORED = ("restored", Severity.OK) + CAPTURED = ("captured", Severity.OK) + READY = ("ready", Severity.OK) + PASSED = ("passed", Severity.OK) + MATCHED = ("matched", Severity.OK) + + # -- warnings: something worth a look, but not blocking -- + WARNING = ("warning", Severity.WARNING) + LIMITED = ("limited", Severity.WARNING) + DRIFTED = ("drifted", Severity.WARNING) + + # -- informational: an intentionally supported state -- + RENAMED = ("renamed", Severity.INFO) + + # -- errors/failures -- + MISSING = ("missing", Severity.ERROR) + UNEXPECTED = ("unexpected", Severity.ERROR) + MISMATCH = ("mismatch", Severity.ERROR) + BLOCKED = ("blocked", Severity.ERROR) + UNSUPPORTED = ("unsupported", Severity.ERROR) + FAILED = ("failed", Severity.ERROR) + + @dataclass(frozen=True) class _ConnContext: """Connection context derived from the oa_configurator resolved resource.""" @@ -53,6 +154,7 @@ def wrapper(**kwargs: Any) -> Any: try: from ..config import create_cdm_engine, get_cdm_context pkg_config, resolved = get_cdm_context() + reject_reserved_schema(resolved.cdm_schema) engine = create_cdm_engine(resolved) conn = _ConnContext( db_schema=resolved.cdm_schema, @@ -139,9 +241,9 @@ def handle_error(exc: Exception) -> None: raise exc -def dry_status(dry_run: bool, applied: str = "applied") -> str: - """Return 'planned' when dry_run is True, otherwise the applied label.""" - return "planned" if dry_run else applied +def dry_status(dry_run: bool, applied: Status = Status.APPLIED) -> Status: + """Return Status.PLANNED when dry_run is True, otherwise the applied status.""" + return Status.PLANNED if dry_run else applied def dry_label(dry_run: bool, planned: str, applied: str) -> str: diff --git a/omop_alchemy/maintenance/cli_backup.py b/omop_alchemy/maintenance/cli_backup.py index 8adbda6..02e4f6c 100644 --- a/omop_alchemy/maintenance/cli_backup.py +++ b/omop_alchemy/maintenance/cli_backup.py @@ -12,7 +12,7 @@ import typer from ..backends import resolve_backend, require_backend_support, backend_support_note -from ._cli_utils import dry_label, dry_status, omop_command +from ._cli_utils import Status, dry_label, dry_status, omop_command from .ui import ( console, render_backup_result, @@ -41,7 +41,7 @@ class BackupResult: file_path: str backup_format: BackupFormat - status: str + status: Status detail: str database_name: str backend: str @@ -87,7 +87,7 @@ def create_database_backup( return BackupResult( file_path=str(resolved_output_path), backup_format=backup_format, - status=dry_status(dry_run, applied="created"), + status=dry_status(dry_run, applied=Status.CREATED), detail=dry_label(dry_run, "Database backup would be created with pg_dump.", "Database backup created with pg_dump."), database_name=database_name, backend=engine.dialect.name, diff --git a/omop_alchemy/maintenance/cli_foreign_keys.py b/omop_alchemy/maintenance/cli_foreign_keys.py index 8c31045..4944aab 100644 --- a/omop_alchemy/maintenance/cli_foreign_keys.py +++ b/omop_alchemy/maintenance/cli_foreign_keys.py @@ -8,7 +8,7 @@ import typer from ..backends import Backend, resolve_backend, require_backend_support, backend_support_note -from ._cli_utils import dry_label, dry_status, omop_command +from ._cli_utils import Status, dry_label, dry_status, omop_command from .tables import ( TableCategory, existing_maintenance_tables, @@ -46,7 +46,7 @@ class _FKTableInfo(ForeignKeyBase): class ForeignKeyManagementResult(_FKTableInfo): """Outcome of a FK trigger enable or disable operation for one table.""" enable: bool - status: str + status: Status detail: str @@ -62,7 +62,7 @@ class ForeignKeyValidationResult(_FKTableInfo): """FK constraint validation outcome for one table, with counts of violating constraints and rows.""" violating_constraint_count: int violating_row_count: int - status: str + status: Status detail: str @@ -257,7 +257,7 @@ def validate_foreign_key_constraints( incoming_constraint_count=target.incoming_constraint_count, violating_constraint_count=violating_constraint_count, violating_row_count=violating_row_count, - status="failed" if violations else "passed", + status=Status.FAILED if violations else Status.PASSED, detail=( _fk_violation_detail(violations) if violations @@ -314,7 +314,7 @@ def manage_foreign_key_triggers( outgoing_constraint_count=target.outgoing_constraint_count, incoming_constraint_count=target.incoming_constraint_count, enable=enable, - status="failed" if violations else "skipped", + status=Status.FAILED if violations else Status.SKIPPED, detail=( _fk_violation_detail(violations, strict_abort=True) if violations diff --git a/omop_alchemy/maintenance/cli_fulltext.py b/omop_alchemy/maintenance/cli_fulltext.py index 9605006..1b690c2 100644 --- a/omop_alchemy/maintenance/cli_fulltext.py +++ b/omop_alchemy/maintenance/cli_fulltext.py @@ -2,13 +2,16 @@ from __future__ import annotations +from dataclasses import dataclass +from enum import StrEnum + import typer from sqlalchemy.engine import Engine from ..backends import backend_support_note as _backend_support_note from ..backends import resolve_backend, require_backend_support -from ..backends.base import FullTextAction, FullTextError, FullTextResult -from ._cli_utils import dry_label, dry_status, omop_command +from ..backends.base import FullTextError +from ._cli_utils import Status, dry_label, dry_status, omop_command from .ui import ( console, render_fulltext_results, @@ -21,6 +24,28 @@ ) +class FullTextAction(StrEnum): + """Which full-text sidecar-column operation a FullTextResult reports on.""" + + INSTALL = "install" + POPULATE = "populate" + DROP = "drop" + + +@dataclass(frozen=True) +class FullTextResult: + """Outcome of a full-text sidecar-column operation for one vocabulary table.""" + target_name: str + table_name: str + source_column_name: str + vector_column_name: str + index_name: str + action: FullTextAction + status: Status + detail: str + row_count: int | None = None + + # ── Orchestrators ───────────────────────────────────────────────────────────── def install_fulltext_columns( diff --git a/omop_alchemy/maintenance/cli_indexes.py b/omop_alchemy/maintenance/cli_indexes.py index 61cb177..725be61 100644 --- a/omop_alchemy/maintenance/cli_indexes.py +++ b/omop_alchemy/maintenance/cli_indexes.py @@ -2,16 +2,18 @@ from __future__ import annotations +import json from dataclasses import dataclass +from typing import Any, Mapping, Sequence import sqlalchemy as sa -from sqlalchemy.exc import DBAPIError +from sqlalchemy.exc import DBAPIError, IntegrityError import typer from omop_alchemy.cdm.base.indexing import OMOP_CLUSTER_INDEX_INFO_KEY -from ..backends import resolve_backend, backend_supports -from ._cli_utils import dry_label, dry_status, omop_command +from ..backends import Backend, resolve_backend, backend_supports +from ._cli_utils import ReservedSchema, Status, dry_label, dry_status, omop_command, reject_reserved_schema from .tables import ( MaintenanceTable, TableCategory, @@ -43,10 +45,454 @@ class IndexManagementResult(IndexTarget): """Outcome of creating or dropping one ORM-defined index, or clustering a table.""" operation: str enable: bool - status: str + status: Status detail: str +def _is_plain_index(reflected: Mapping[str, Any]) -> bool: + """Determine whether a reflected index is a plain, safely-manageable index. + + A plain index has no expression components, no partial predicate, uses the + default (btree) access method, and does not back a UNIQUE or PRIMARY KEY + constraint. Only plain indexes are safe to treat as a faithful equivalent + of an ORM-defined index, and safe to capture and restore: PostgreSQL + refuses to DROP INDEX on a constraint-backed index, and partial, expression, + or non-btree indexes can't be faithfully reconstructed from a plain column + list alone. + + Parameters + ---------- + reflected : Mapping[str, Any] + A single index dict as returned by sqlalchemy.engine.Inspector.get_indexes(). + + Returns + ------- + bool + True if the index is plain (a faithful, manageable equivalent), False otherwise. + """ + column_names = reflected.get("column_names") or [] + if any(name is None for name in column_names): + return False + if reflected.get("expressions"): + return False + if reflected.get("duplicates_constraint"): + return False + dialect_options = reflected.get("dialect_options") or {} + if dialect_options.get("postgresql_where"): + return False + if dialect_options.get("postgresql_using"): + return False + return True + + +def _find_equivalent_index( + existing_indexes: Sequence[Mapping[str, Any]], + column_names: tuple[str, ...], + unique: bool, +) -> str | None: + """Find the physical name of a plain existing index matching a column set. + + Parameters + ---------- + existing_indexes : Sequence[Mapping[str, Any]] + Reflected indexes for one table, as returned by Inspector.get_indexes(). + column_names : tuple[str, ...] + Column names an ORM-defined index expects, in order. Matching is + order-sensitive since composite index column order affects usability. + unique : bool + Uniqueness flag an ORM-defined index expects. + + Returns + ------- + str or None + Physical name of the first plain existing index whose column_names + tuple and unique flag match, or None if no equivalent exists. + """ + for reflected in existing_indexes: + if not _is_plain_index(reflected): + continue + if tuple(reflected.get("column_names") or ()) != column_names: + continue + if bool(reflected.get("unique")) != unique: + continue + return str(reflected["name"]) + return None + + +def _find_shape_conflict( + existing_indexes: Sequence[Mapping[str, Any]], + column_names: tuple[str, ...], + unique: bool, +) -> Mapping[str, Any] | None: + """Find a non-plain existing index covering the same columns. + + Like _find_equivalent_index but for a match that can't be safely treated + as equivalent (expression, partial, non-btree, or constraint-backed). Used + only to explain why such an index is left alone rather than captured. + + Parameters + ---------- + existing_indexes : Sequence[Mapping[str, Any]] + Reflected indexes for one table, as returned by Inspector.get_indexes(). + column_names : tuple[str, ...] + Column names an ORM-defined index expects, in order. + unique : bool + Uniqueness flag an ORM-defined index expects. + + Returns + ------- + Mapping[str, Any] or None + The reflected index dict of the conflicting index, or None if no + such index exists. + """ + for reflected in existing_indexes: + column_names_reflected = reflected.get("column_names") or [] + if any(name is None for name in column_names_reflected): + continue + if tuple(column_names_reflected) != column_names: + continue + if bool(reflected.get("unique")) != unique: + continue + if _is_plain_index(reflected): + continue + return reflected + return None + + +def _describe_shape_conflict(reflected: Mapping[str, Any]) -> str: + """Describe why a reflected index can't be safely captured or treated as equivalent. + + Parameters + ---------- + reflected : Mapping[str, Any] + A reflected index dict, as returned by _find_shape_conflict. + + Returns + ------- + str + Human-readable reason(s), comma-separated. + """ + dialect_options = reflected.get("dialect_options") or {} + reasons: list[str] = [] + if reflected.get("duplicates_constraint"): + reasons.append("backs a UNIQUE/PRIMARY KEY constraint") + if dialect_options.get("postgresql_where"): + reasons.append("has a partial WHERE predicate") + if dialect_options.get("postgresql_using"): + reasons.append(f"uses non-btree access method '{dialect_options['postgresql_using']}'") + if not reasons: + reasons.append("has an unsupported definition") + return ", ".join(reasons) + + +# ── Foreign index capture/restore bookkeeping ─────────────────────────────────── + +_DROPPED_INDEXES_TABLE_NAME = "dropped_indexes" +_DROPPED_INDEXES_UNIQUE_CONSTRAINT_NAME = "uq_dropped_indexes_table_columns" + + +def _schema_key(db_schema: str | None) -> str: + """Normalize db_schema to a non-null string for use in the bookkeeping table. + + SQL UNIQUE constraints treat every NULL as distinct from every other NULL, + so a nullable db_schema column would silently defeat uniqueness whenever + db_schema is None (SQLite always, and PostgreSQL whenever no explicit + schema is configured). As a result, two captures for the same table/column-set + could both succeed instead of the second being rejected. + + Parameters + ---------- + db_schema : str or None + Schema name, or None for "no explicit schema". + + Returns + ------- + str + db_schema unchanged, or "" when db_schema is None. + """ + return db_schema or "" + + +def get_bookkeeping_schema(backend: Backend) -> str | None: + """Return the reserved schema name for the dropped-index bookkeeping table. + + Parameters + ---------- + backend : Backend + The resolved database backend. + + Returns + ------- + str or None + ReservedSchema.MAINTENANCE.value on backends that override + Backend.ensure_schema() (i.e. support named schemas, like + PostgreSQL), or None on backends that don't (like SQLite). + """ + if backend_supports(backend, "ensure_schema"): + return ReservedSchema.MAINTENANCE.value + return None + + +def _dropped_indexes_table(bookkeeping_schema: str | None) -> sa.Table: + """Build the dropped-index bookkeeping table definition. + + disable drops indexes to speed up bulk loads. If the only index covering + an ORM-defined column set is a foreign (non-OMOP_Alchemy) index, e.g. one + created by the official OHDSI CDM DDL script, dropping it still gives the + bulk-load speed benefit, but its definition must be captured somewhere that + survives across separate CLI invocations, since disable and enable are + documented as usable as two separate commands, not just paired within a + single process. This table records that definition in a reserved schema + before the foreign index is dropped, so a later enable call can recreate + it under its original name. db_schema is part of the row identity so two + same-named tables in different schemas of one database never collide. + + Parameters + ---------- + bookkeeping_schema : str or None + Schema to qualify the table with, from get_bookkeeping_schema(). + + Returns + ------- + sqlalchemy.Table + The (unbound) table definition. Not yet created in the database. + """ + metadata = sa.MetaData() + return sa.Table( + _DROPPED_INDEXES_TABLE_NAME, + metadata, + sa.Column("id", sa.Integer, primary_key=True, autoincrement=True), + sa.Column("table_name", sa.String(128), nullable=False), + sa.Column("db_schema", sa.String(128), nullable=False), + sa.Column("index_name", sa.String(128), nullable=False), + sa.Column("column_names_json", sa.Text, nullable=False), + sa.Column("is_unique", sa.Boolean, nullable=False, server_default=sa.false()), + sa.Column("captured_at", sa.DateTime, server_default=sa.func.now(), nullable=False), + sa.UniqueConstraint( + "table_name", "db_schema", "column_names_json", "is_unique", + name=_DROPPED_INDEXES_UNIQUE_CONSTRAINT_NAME, + ), + schema=bookkeeping_schema, + ) + + +def _record_captured_index( + connection: sa.Connection, + backend: Backend, + *, + table_name: str, + db_schema: str | None, + index_name: str, + column_names: tuple[str, ...], + unique: bool, +) -> bool: + """Capture a foreign index's definition into the bookkeeping table before dropping it. + + Returns False, recording nothing, if this table/schema/column-set already + has a pending capture awaiting restore, e.g. a second disable run, + without an intervening enable, that finds a different foreign index than + the one already captured. The bookkeeping table's unique constraint only + allows one pending capture per table/schema/column-set/uniqueness, since + only one original name could ever be restored to that slot; the caller + must leave the second index in place rather than dropping something it can + no longer track. + + Parameters + ---------- + connection : sqlalchemy.Connection + Open connection/transaction the capture is recorded on. + backend : Backend + The resolved database backend. + table_name : str + Name of the table the foreign index belongs to. + db_schema : str or None + Schema the target table lives in, so captures from different schemas + of a same-named table never collide. + index_name : str + Original physical name of the foreign index being captured. + column_names : tuple[str, ...] + Column names the foreign index covers, in order. + unique : bool + Uniqueness flag of the foreign index. + + Returns + ------- + bool + True if the capture was recorded, False if a pending capture already + existed for this table/schema/column-set/uniqueness. + """ + bookkeeping_schema = get_bookkeeping_schema(backend) + backend.ensure_schema(connection, bookkeeping_schema) + bookkeeping_table = _dropped_indexes_table(bookkeeping_schema) + bookkeeping_table.create(bind=connection, checkfirst=True) + + savepoint = connection.begin_nested() + try: + connection.execute( + bookkeeping_table.insert(), + { + "table_name": table_name, + "db_schema": _schema_key(db_schema), + "index_name": index_name, + "column_names_json": json.dumps(list(column_names)), + "is_unique": unique, + }, + ) + except IntegrityError as exc: + savepoint.rollback() + message = str(exc.orig).lower() + # Constraint naming differs between backends + # We are preventing to swallow other unique constraint violations + # by checking for a specific one + if _DROPPED_INDEXES_UNIQUE_CONSTRAINT_NAME not in message and _DROPPED_INDEXES_TABLE_NAME not in message: + raise + return False + else: + savepoint.commit() + return True + + +def _peek_captured_index( + connection: sa.Connection, + backend: Backend, + *, + table_name: str, + db_schema: str | None, + column_names: tuple[str, ...], + unique: bool, +) -> tuple[str | None, sa.Table | None, sa.Row | None]: + """Look up a pending capture's original index name without restoring or deleting it. + + Read-only counterpart to _restore_captured_index, used to preview what a + live run would do (a restore, or a capture-conflict) without mutating the + bookkeeping table. + + Parameters + ---------- + connection : sqlalchemy.Connection + Open connection the lookup is performed on. + backend : Backend + The resolved database backend. + table_name : str + Name of the table the index belongs to. + db_schema : str or None + Schema the target table lives in. + column_names : tuple[str, ...] + Column names the captured index covers, in order. + unique : bool + Uniqueness flag of the captured index. + + Returns + ------- + restored_index_name : str or None + The captured index's original physical name, or None if nothing is + captured for this table/schema/column-set/uniqueness. + bookkeeping_table : sqlalchemy.Table or None + The bookkeeping table definition, for use in a later restore or + delete. None if the bookkeeping table doesn't exist yet. + row : sqlalchemy.Row or None + The matched bookkeeping row, for use in a later delete by id. None if + nothing is captured for this table/schema/column-set/uniqueness. + """ + bookkeeping_schema = get_bookkeeping_schema(backend) + inspector = sa.inspect(connection) + if not inspector.has_table(_DROPPED_INDEXES_TABLE_NAME, schema=bookkeeping_schema): + return None, None, None + + bookkeeping_table = _dropped_indexes_table(bookkeeping_schema) + column_names_json = json.dumps(list(column_names)) + row = connection.execute( + sa.select(bookkeeping_table.c.id, bookkeeping_table.c.index_name).where( + bookkeeping_table.c.table_name == table_name, + bookkeeping_table.c.db_schema == _schema_key(db_schema), + bookkeeping_table.c.column_names_json == column_names_json, + bookkeeping_table.c.is_unique == unique, + ) + ).one_or_none() + return str(row.index_name) if row is not None else None, bookkeeping_table, row + + +def _restore_captured_index( + connection: sa.Connection, + backend: Backend, + *, + table_name: str, + db_schema: str | None, + column_names: tuple[str, ...], + unique: bool, +) -> str | None: + """Recreate a previously captured foreign index under its original name. + + Removes the corresponding bookkeeping row on success. + + Parameters + ---------- + connection : sqlalchemy.Connection + Open connection/transaction the index is created on. + backend : Backend + The resolved database backend. + table_name : str + Name of the table to recreate the index on. + db_schema : str or None + Schema the target table lives in. + column_names : tuple[str, ...] + Column names the restored index should cover, in order. + unique : bool + Uniqueness flag the restored index should have. + + Returns + ------- + str or None + The restored physical index name, or None if nothing was captured + for this table/schema/column-set, or if it was already recreated by + someone else in the meantime (see Notes). + + Notes + ----- + If the captured index was already recreated out-of-band (e.g. a concurrent + enable invocation, or a manually restored backup), create(checkfirst=True) + can still race with the physical CREATE INDEX statement and raise a + "relation already exists" DBAPIError; this is caught the same way the + equivalent race is handled for newly-created ORM-defined indexes, and + treated as a no-op restore (the bookkeeping row is still cleared, since the + index is confirmed to exist either way). + """ + restored_index_name, bookkeeping_table, row = _peek_captured_index( + connection=connection, + backend=backend, + table_name=table_name, + db_schema=db_schema, + column_names=column_names, + unique=unique, + ) + if restored_index_name is None or bookkeeping_table is None or row is None: + return None + + # A lightweight, untyped Table (no autoload_with reflection) is sufficient: + # CREATE INDEX DDL only needs column names, not real types, PKs, FKs, or + # constraints -- reflecting the whole table would cost several extra + # catalog round-trips to fetch metadata this function never uses. + lightweight_table = sa.Table( + table_name, sa.MetaData(), + *(sa.Column(name) for name in column_names), + schema=db_schema, + ) + restored_index = sa.Index( + restored_index_name, *[lightweight_table.c[name] for name in column_names], unique=unique + ) + savepoint = connection.begin_nested() + try: + restored_index.create(bind=connection, checkfirst=True) + except DBAPIError as exc: + savepoint.rollback() + if "already exists" not in str(exc.orig).lower(): + raise + else: + savepoint.commit() + connection.execute(bookkeeping_table.delete().where(bookkeeping_table.c.id == row.id)) + return restored_index_name + + def _schema_metadata_indexes( tables: list[MaintenanceTable], db_schema: str | None, @@ -96,6 +542,71 @@ def _cluster_column_names( return table.primary_key_names +def _cluster_index_unique(table: MaintenanceTable, cluster_index_name: str) -> bool: + """Return whether the named cluster index is unique. + + Parameters + ---------- + table : MaintenanceTable + The table whose cluster target is being resolved. + cluster_index_name : str + Name of the ORM-designated cluster index, from _cluster_target_name(). + + Returns + ------- + bool + The index's unique flag, or True if cluster_index_name isn't among the + table's secondary indexes -- the fallback case is a primary-key-based + cluster target, which is always unique. + """ + for index in table.table.indexes: + if str(index.name) == cluster_index_name: + return bool(index.unique) + return True + + +def _resolve_physical_cluster_name( + existing_indexes: Sequence[Mapping[str, Any]], + cluster_index_name: str, + cluster_columns: tuple[str, ...], + unique: bool, +) -> str: + """Resolve the physical name of a table's cluster-target index. + + The ORM-designated cluster index may not physically exist under its own + name if the database was pre-indexed by an external script (e.g. the + OHDSI CDM DDL script), so this falls back to whichever plain index + actually covers the same columns. Shared by the standalone indexes + cluster command and manage_indexes()'s cluster step (for the + primary-key-based cluster target case, which manage_indexes() otherwise + resolves more precisely via its own per-run physical_index_names tracking + -- see the comment at its call site). + + Parameters + ---------- + existing_indexes : Sequence[Mapping[str, Any]] + Reflected indexes for the table, as returned by Inspector.get_indexes(). + cluster_index_name : str + The ORM's own name for the cluster-target index. + cluster_columns : tuple[str, ...] + Column names the cluster-target index covers, in order. + unique : bool + Uniqueness flag of the cluster-target index. + + Returns + ------- + str + cluster_index_name if it physically exists under that name, otherwise + the physical name of a plain equivalent index if one is found, + otherwise cluster_index_name unchanged. + """ + existing_names = {index["name"] for index in existing_indexes} + if cluster_index_name in existing_names: + return cluster_index_name + equivalent_name = _find_equivalent_index(existing_indexes, cluster_columns, unique) + return equivalent_name if equivalent_name is not None else cluster_index_name + + def collect_index_targets( engine: sa.Engine, *, @@ -111,28 +622,57 @@ def collect_index_targets( if not inspector.has_table(table.table_name, schema=db_schema): continue - existing_index_names = { - index["name"] - for index in inspector.get_indexes(table.table_name, schema=db_schema) - } + existing_indexes = inspector.get_indexes(table.table_name, schema=db_schema) + existing_index_names = {index["name"] for index in existing_indexes} for metadata_index in sorted(table.table.indexes, key=lambda idx: idx.name or ""): - if metadata_index.name not in existing_index_names: - continue + column_names = tuple(column.name for column in metadata_index.columns) + unique = bool(metadata_index.unique) + if metadata_index.name in existing_index_names: + physical_name = str(metadata_index.name) + else: + equivalent_name = _find_equivalent_index(existing_indexes, column_names, unique) + if equivalent_name is None: + continue + physical_name = equivalent_name targets.append( IndexTarget( table_name=table.table_name, category=table.category, - index_name=str(metadata_index.name), - column_names=tuple(column.name for column in metadata_index.columns), - unique=bool(metadata_index.unique), + index_name=physical_name, + column_names=column_names, + unique=unique, clustered=metadata_index.info.get(OMOP_CLUSTER_INDEX_INFO_KEY) is True, ) ) return targets +@dataclass(frozen=True) +class _IndexOutcome: + """Resolved outcome of processing one ORM-defined index in manage_indexes(). + + Built directly at the point each outcome is determined, in the live-run + or dry-run branch -- there is exactly one place that decides status, + detail, and physical_name for a given case, not a separate translation + step that could fall out of sync with the branch that determined it. + + Parameters + ---------- + status : Status + The status to report for this index. + detail : str + Human-readable detail text to report for this index. + physical_name : str + The physical index name to report. The ORM's own index_name unless a + foreign name was captured, restored, or found equivalent. + """ + + status: Status + detail: str + physical_name: str + def manage_indexes( engine: sa.Engine, @@ -144,6 +684,7 @@ def manage_indexes( cluster: bool = True, ) -> list[IndexManagementResult]: """Create or drop all ORM-defined indexes. CLUSTERs tables when enabling and cluster=True.""" + reject_reserved_schema(db_schema) backend = resolve_backend(engine) inspector = sa.inspect(engine) selected_tables = select_omop_tables(vocabulary_included=vocabulary_included) @@ -156,16 +697,17 @@ def manage_indexes( if not inspector.has_table(table.table_name, schema=db_schema): continue - existing_index_names = { - index["name"] - for index in inspector.get_indexes(table.table_name, schema=db_schema) - } + existing_indexes = inspector.get_indexes(table.table_name, schema=db_schema) + existing_index_names = {index["name"] for index in existing_indexes} created_any = False clustered_now = False + physical_index_names: dict[str, str] = {} for metadata_index in sorted(table.table.indexes, key=lambda idx: idx.name or ""): index_name = str(metadata_index.name) + column_names = tuple(column.name for column in metadata_index.columns) + unique = bool(metadata_index.unique) exists = index_name in existing_index_names should_apply = ( not enable @@ -174,77 +716,216 @@ def manage_indexes( ) if not should_apply: + physical_index_names[index_name] = index_name continue schema_index = metadata_indexes[(table.table_name, index_name)] - already_present = False - already_absent = False - if not dry_run: - # Each index gets its own transaction so WAL is committed and - # checkpointable before the next index build begins. - with engine.begin() as connection: - if not enable: + # Plain create/drop succeeding is the common case for both live and + # dry runs, so it's the default outcome; every branch below only + # overrides it for a foreign-index or already-in-place case. + outcome = _IndexOutcome( + status=dry_status(dry_run), + detail=dry_label( + dry_run, + planned="metadata-defined index would be dropped" if not enable else "metadata-defined index would be created", + applied="metadata-defined index dropped" if not enable else "metadata-defined index created", + ), + physical_name=index_name, + ) + + # Each index gets its own connection: a transaction when actually + # mutating (not dry_run, so WAL is committed and checkpointable + # before the next index build begins), a plain read-only + # connection when only previewing. + connection_factory = engine.begin if not dry_run else engine.connect + with connection_factory() as connection: + if not enable: + if not dry_run: existed_before_drop = backend.index_exists(connection, index_name, db_schema) - backend.drop_index_if_exists(connection, index_name, db_schema) - already_absent = not existed_before_drop else: - savepoint = connection.begin_nested() - try: - schema_index.create(bind=connection, checkfirst=True) - except DBAPIError as exc: - savepoint.rollback() - if "already exists" not in str(exc.orig).lower(): - raise - already_present = True + existed_before_drop = exists + if not existed_before_drop: + # Index under a different naming scheme than ours + equivalent_name = _find_equivalent_index(existing_indexes, column_names, unique) + if equivalent_name is not None: + if not dry_run: + captured = _record_captured_index( + connection, backend, + table_name=table.table_name, db_schema=db_schema, + index_name=equivalent_name, + column_names=column_names, unique=unique, + ) + else: + pending_capture, _, _ = _peek_captured_index( + connection, backend, + table_name=table.table_name, db_schema=db_schema, + column_names=column_names, unique=unique, + ) + captured = pending_capture is None + if captured: + if not dry_run: + backend.drop_index_if_exists(connection, equivalent_name, db_schema) + outcome = _IndexOutcome( + status=dry_status(dry_run, Status.CAPTURED), + detail=dry_label( + dry_run, + planned=f"foreign index '{equivalent_name}' would be captured and dropped for bulk load", + applied=f"foreign index '{equivalent_name}' captured and dropped for bulk load", + ), + physical_name=equivalent_name, + ) + else: + # A different foreign index for this table/column-set is + # already captured and awaiting restore. Leave this one + # in place rather than dropping something we can no + # longer track. + outcome = _IndexOutcome( + status=Status.WARNING, + detail=dry_label( + dry_run, + planned=( + f"foreign index '{equivalent_name}' would be left in place: a different " + "foreign index for this table/column-set is already captured and " + "awaiting restore" + ), + applied=( + f"foreign index '{equivalent_name}' left in place: a different " + "foreign index for this table/column-set is already captured and " + "awaiting restore" + ), + ), + physical_name=equivalent_name, + ) else: - savepoint.commit() + conflict = _find_shape_conflict(existing_indexes, column_names, unique) + if conflict is not None: + conflict_name = str(conflict["name"]) + outcome = _IndexOutcome( + status=Status.WARNING, + detail=dry_label( + dry_run, + planned=f"foreign index '{conflict_name}' {_describe_shape_conflict(conflict)}; would be left in place", + applied=f"foreign index '{conflict_name}' {_describe_shape_conflict(conflict)}; left in place", + ), + physical_name=conflict_name, + ) + else: + outcome = _IndexOutcome( + status=Status.SKIPPED, + detail="metadata-defined index already absent (skipped)", + physical_name=index_name, + ) + elif not dry_run: + backend.drop_index_if_exists(connection, index_name, db_schema) + # outcome stays default: applied / "metadata-defined index dropped" + # dry-run, existed_before_drop True: outcome stays default ("would be dropped") + else: + if not dry_run: + restored_name = _restore_captured_index( + connection, backend, + table_name=table.table_name, db_schema=db_schema, + column_names=column_names, unique=unique, + ) + else: + restored_name, _, _ = _peek_captured_index( + connection, backend, + table_name=table.table_name, db_schema=db_schema, + column_names=column_names, unique=unique, + ) + if restored_name is not None: + if not dry_run: created_any = True + outcome = _IndexOutcome( + status=dry_status(dry_run, Status.RESTORED), + detail=dry_label( + dry_run, + planned=f"foreign index '{restored_name}' would be restored from bulk-load capture", + applied=f"foreign index '{restored_name}' restored from bulk-load capture", + ), + physical_name=restored_name, + ) + else: + equivalent_name = _find_equivalent_index(existing_indexes, column_names, unique) + if equivalent_name is not None: + outcome = _IndexOutcome( + status=Status.SKIPPED, + detail=dry_label( + dry_run, + planned=f"equivalent foreign index '{equivalent_name}' already provides this coverage (would skip creation)", + applied=f"equivalent foreign index '{equivalent_name}' already provides this coverage (skipped)", + ), + physical_name=equivalent_name, + ) + elif not dry_run: + savepoint = connection.begin_nested() + try: + schema_index.create(bind=connection, checkfirst=True) + except DBAPIError as exc: + savepoint.rollback() + if "already exists" not in str(exc.orig).lower(): + raise + outcome = _IndexOutcome( + status=Status.SKIPPED, + detail="metadata-defined index already exists (skipped)", + physical_name=index_name, + ) + else: + savepoint.commit() + created_any = True + # outcome stays default: applied / "metadata-defined index created" + # dry-run, no restore, no equivalent: outcome stays default ("would be created") - if already_present: - detail = "metadata-defined index already exists (skipped)" - status = "skipped" - elif already_absent: - detail = "metadata-defined index already absent (skipped)" - status = "skipped" - else: - status = dry_status(dry_run) - detail = dry_label( - dry_run, - "metadata-defined index would be dropped" if not enable else "metadata-defined index would be created", - "metadata-defined index dropped" if not enable else "metadata-defined index created", - ) - + physical_name = outcome.physical_name + physical_index_names[index_name] = physical_name results.append( IndexManagementResult( operation="index", table_name=table.table_name, category=table.category, - index_name=index_name, - column_names=tuple(column.name for column in metadata_index.columns), - unique=bool(metadata_index.unique), + index_name=physical_name, + column_names=column_names, + unique=unique, clustered=metadata_index.info.get(OMOP_CLUSTER_INDEX_INFO_KEY) is True, enable=enable, - status=status, - detail=detail, + status=outcome.status, + detail=outcome.detail, ) ) + # Clustering for perfomance is a separate operation from index creation if enable: cluster_index_name = _cluster_target_name(table) if cluster_index_name is not None: cluster_columns = _cluster_column_names(table, cluster_index_name) + if cluster_index_name in physical_index_names: + # Resolved authoritatively from what actually happened in this + # run's per-index loop (own name, captured, restored, or a + # skip-equivalent) -- more precise than re-deriving from the + # now-stale existing_indexes snapshot, since e.g. a + # just-restored index wouldn't appear in it. + physical_cluster_name = physical_index_names[cluster_index_name] + else: + # Primary-key-based cluster target: never entered the per-index + # loop, so resolve it the same way the standalone `indexes + # cluster` command does. + physical_cluster_name = _resolve_physical_cluster_name( + existing_indexes, + cluster_index_name, + cluster_columns, + _cluster_index_unique(table, cluster_index_name), + ) if not clustering_supported or not cluster: results.append( IndexManagementResult( operation="cluster", table_name=table.table_name, category=table.category, - index_name=cluster_index_name, + index_name=physical_cluster_name, column_names=cluster_columns, unique=False, clustered=True, enable=enable, - status="skipped", + status=Status.SKIPPED, detail=( f"cluster metadata present but unsupported on {backend.name}" if not clustering_supported @@ -255,7 +936,7 @@ def manage_indexes( else: if not dry_run: with engine.begin() as connection: - backend.cluster_table(connection, table.table_name, cluster_index_name, db_schema) + backend.cluster_table(connection, table.table_name, physical_cluster_name, db_schema) clustered_now = True results.append( @@ -263,7 +944,7 @@ def manage_indexes( operation="cluster", table_name=table.table_name, category=table.category, - index_name=cluster_index_name, + index_name=physical_cluster_name, column_names=cluster_columns, unique=False, clustered=True, @@ -389,10 +1070,17 @@ def cluster_tables_command( continue cluster_columns = _cluster_column_names(table, cluster_index_name) + existing_indexes = inspector.get_indexes(table.table_name, schema=conn.db_schema) + physical_cluster_name = _resolve_physical_cluster_name( + existing_indexes, + cluster_index_name, + cluster_columns, + _cluster_index_unique(table, cluster_index_name), + ) if not dry_run: with engine.begin() as connection: - backend.cluster_table(connection, table.table_name, cluster_index_name, conn.db_schema) + backend.cluster_table(connection, table.table_name, physical_cluster_name, conn.db_schema) with engine.connect() as connection: backend.analyze_table(connection, table.table_name, conn.db_schema) connection.commit() @@ -402,7 +1090,7 @@ def cluster_tables_command( operation="cluster", table_name=table.table_name, category=table.category, - index_name=cluster_index_name, + index_name=physical_cluster_name, column_names=cluster_columns, unique=False, clustered=True, diff --git a/omop_alchemy/maintenance/cli_schema_doctor.py b/omop_alchemy/maintenance/cli_schema_doctor.py index 278062e..1293254 100644 --- a/omop_alchemy/maintenance/cli_schema_doctor.py +++ b/omop_alchemy/maintenance/cli_schema_doctor.py @@ -6,6 +6,7 @@ from omop_alchemy.backends.resolve import SupportedDialect +from ._cli_utils import Status from .cli_foreign_keys import ( ForeignKeyStatusResult, ForeignKeyValidationReport, @@ -18,6 +19,7 @@ ) from .cli_schema_reconcile import ( SchemaReconciliationReport, + is_blocking_issue, reconcile_schema, ) @@ -31,7 +33,7 @@ class DoctorCheck: """Result of a single named maintenance health check (e.g. 'managed tables', 'schema drift').""" name: str - status: str + status: Status detail: str @@ -39,7 +41,7 @@ class DoctorCheck: class DoctorRecommendation: """Actionable recommendation derived from health check results, with an optional CLI command hint.""" - status: str + status: Status summary: str action: str | None @@ -69,7 +71,7 @@ def _build_recommendations( if not info.connection_ready: recommendations.append( DoctorRecommendation( - status="failed", + status=Status.FAILED, summary="Database connection is not ready for maintenance operations.", action="Check the engine configuration, backend driver, and target database reachability.", ) @@ -79,27 +81,29 @@ def _build_recommendations( if info.missing_table_count: recommendations.append( DoctorRecommendation( - status="warning", + status=Status.WARNING, summary=f"{info.missing_table_count} ORM-managed table(s) are missing from the target database.", action="Run `omop-alchemy create-missing-tables` before attempting bulk operations.", ) ) - if reconciliation is not None and reconciliation.issues: - recommendations.append( - DoctorRecommendation( - status="warning", - summary=f"Schema reconciliation found {len(reconciliation.issues)} difference(s) against ORM metadata.", - action="Review `omop-alchemy reconcile-schema` output before continuing with ETL or maintenance work.", + if reconciliation is not None: + blocking_issue_count = sum(1 for issue in reconciliation.issues if is_blocking_issue(issue)) + if blocking_issue_count: + recommendations.append( + DoctorRecommendation( + status=Status.WARNING, + summary=f"Schema reconciliation found {blocking_issue_count} difference(s) against ORM metadata.", + action="Review `omop-alchemy reconcile-schema` output before continuing with ETL or maintenance work.", + ) ) - ) if foreign_key_status is not None and any( item.disabled_trigger_count > 0 for item in foreign_key_status ): recommendations.append( DoctorRecommendation( - status="warning", + status=Status.WARNING, summary="Some PostgreSQL RI triggers are currently disabled.", action="If loading is complete, run `omop-alchemy foreign-keys validate` and then `omop-alchemy foreign-keys enable --strict`.", ) @@ -107,11 +111,11 @@ def _build_recommendations( if ( foreign_key_validation is not None - and any(result.status == "failed" for result in foreign_key_validation.results) + and any(result.status == Status.FAILED for result in foreign_key_validation.results) ): recommendations.append( DoctorRecommendation( - status="failed", + status=Status.FAILED, summary="Foreign key validation found violating rows.", action="Fix the reported rows, then rerun `omop-alchemy foreign-keys enable --strict`.", ) @@ -120,7 +124,7 @@ def _build_recommendations( if info.backend == SupportedDialect.POSTGRESQL and info.pg_dump_path is None: recommendations.append( DoctorRecommendation( - status="warning", + status=Status.WARNING, summary="`pg_dump` is not on PATH, so backup-database is unavailable from this machine.", action="Install PostgreSQL client tools on the machine running `omop-alchemy`.", ) @@ -133,7 +137,7 @@ def _build_recommendations( ): recommendations.append( DoctorRecommendation( - status="warning", + status=Status.WARNING, summary="Neither `pg_restore` nor `psql` is on PATH, so restore-database is unavailable from this machine.", action="Install PostgreSQL client tools on the machine running `omop-alchemy`.", ) @@ -142,7 +146,7 @@ def _build_recommendations( if not recommendations: recommendations.append( DoctorRecommendation( - status="passed", + status=Status.PASSED, summary="No obvious maintenance blockers were detected.", action=None, ) @@ -162,7 +166,7 @@ def collect_doctor_report( checks = [ DoctorCheck( name="connection", - status="passed" if info.connection_ready else "failed", + status=Status.PASSED if info.connection_ready else Status.FAILED, detail=( "Target database connection succeeded." if info.connection_ready @@ -185,7 +189,7 @@ def collect_doctor_report( checks.append( DoctorCheck( name="managed tables", - status="passed" if missing_table_count == 0 else "warning", + status=Status.PASSED if missing_table_count == 0 else Status.WARNING, detail=( "All selected ORM-managed tables exist." if missing_table_count == 0 @@ -200,14 +204,17 @@ def collect_doctor_report( db_schema=db_schema, vocabulary_included=vocabulary_included, ) + blocking_issue_count = sum( + 1 for issue in reconciliation.issues if is_blocking_issue(issue) + ) checks.append( DoctorCheck( name="schema drift", - status="passed" if not reconciliation.issues else "warning", + status=Status.PASSED if not blocking_issue_count else Status.WARNING, detail=( "ORM metadata matches the target database." - if not reconciliation.issues - else f"{len(reconciliation.issues)} difference(s) detected." + if not blocking_issue_count + else f"{blocking_issue_count} difference(s) detected." ), ) ) @@ -215,7 +222,7 @@ def collect_doctor_report( checks.append( DoctorCheck( name="schema drift", - status="skipped", + status=Status.SKIPPED, detail="Run `omop-alchemy doctor --deep` to reconcile ORM metadata against the target database.", ) ) @@ -234,7 +241,7 @@ def collect_doctor_report( checks.append( DoctorCheck( name="foreign keys", - status="passed" if disabled_tables == 0 else "warning", + status=Status.PASSED if disabled_tables == 0 else Status.WARNING, detail=( "All inspected RI triggers are enabled." if disabled_tables == 0 @@ -250,12 +257,12 @@ def collect_doctor_report( vocabulary_included=vocabulary_included, ) violating_tables = sum( - result.status == "failed" for result in foreign_key_validation.results + result.status == Status.FAILED for result in foreign_key_validation.results ) checks.append( DoctorCheck( name="foreign key validation", - status="passed" if violating_tables == 0 else "failed", + status=Status.PASSED if violating_tables == 0 else Status.FAILED, detail=( "All selected foreign key relationships passed validation." if violating_tables == 0 @@ -267,7 +274,7 @@ def collect_doctor_report( checks.append( DoctorCheck( name="foreign key validation", - status="skipped", + status=Status.SKIPPED, detail="Run `omop-alchemy doctor --deep` to validate selected foreign key relationships.", ) ) @@ -275,14 +282,14 @@ def collect_doctor_report( checks.append( DoctorCheck( name="foreign keys", - status="skipped", + status=Status.SKIPPED, detail="Foreign key trigger inspection is only available on PostgreSQL.", ) ) checks.append( DoctorCheck( name="foreign key validation", - status="skipped", + status=Status.SKIPPED, detail="Foreign key validation is only available on PostgreSQL.", ) ) @@ -293,22 +300,22 @@ def collect_doctor_report( ( DoctorCheck( name="managed tables", - status="skipped", + status=Status.SKIPPED, detail="Skipped because the database connection is not ready.", ), DoctorCheck( name="foreign keys", - status="skipped", + status=Status.SKIPPED, detail="Skipped because the database connection is not ready.", ), DoctorCheck( name="schema drift", - status="skipped", + status=Status.SKIPPED, detail="Skipped because the database connection is not ready.", ), DoctorCheck( name="foreign key validation", - status="skipped", + status=Status.SKIPPED, detail="Skipped because the database connection is not ready.", ), ) @@ -321,7 +328,7 @@ def collect_doctor_report( checks.append( DoctorCheck( name="backup tooling", - status="passed" if backup_tools_ready else "warning", + status=Status.PASSED if backup_tools_ready else Status.WARNING, detail=( "PostgreSQL backup and restore client tools are available." if backup_tools_ready @@ -333,7 +340,7 @@ def collect_doctor_report( checks.append( DoctorCheck( name="backup tooling", - status="skipped", + status=Status.SKIPPED, detail="Backup and restore tooling checks are only relevant for PostgreSQL targets.", ) ) diff --git a/omop_alchemy/maintenance/cli_schema_info.py b/omop_alchemy/maintenance/cli_schema_info.py index f27daa1..913937a 100644 --- a/omop_alchemy/maintenance/cli_schema_info.py +++ b/omop_alchemy/maintenance/cli_schema_info.py @@ -15,6 +15,7 @@ from omop_alchemy.backends.resolve import SupportedDialect from omop_alchemy.config import OmopAlchemyConfig +from ._cli_utils import Status from .cli_schema_tables import collect_missing_tables from .tables import ( TableCategory, @@ -49,7 +50,7 @@ class CommandSupport: command_name: str requirement: str - status: str + status: Status detail: str @@ -108,7 +109,7 @@ def _external_dependency_status(name: str, executable_name: str) -> DependencySt def _command_support_for_unavailable_engine(detail: str) -> tuple[CommandSupport, ...]: """Return a full CommandSupport tuple with every command marked blocked, used when the engine cannot be created.""" - blocked = "blocked" + blocked = Status.BLOCKED return ( CommandSupport("doctor", "Any SQLAlchemy backend", blocked, detail), CommandSupport("data-summary", "Any SQLAlchemy backend", blocked, detail), @@ -158,7 +159,7 @@ def _command_support_for_backend( if connection_error else f"Backend resolved to {current_backend}, but the connection test failed." ) - portable_status = "ready" if connection_ready else "blocked" + portable_status = Status.READY if connection_ready else Status.BLOCKED portable_detail = ( f"Ready on {current_backend}." if connection_ready else blocked_detail ) @@ -185,19 +186,19 @@ def _command_support_for_backend( else blocked_detail ) elif backend == "sqlite": - analyze_status = "limited" if connection_ready else "blocked" + analyze_status = Status.LIMITED if connection_ready else Status.BLOCKED analyze_detail = ( "Ready on SQLite; ANALYZE is supported, but `--vacuum` is unavailable." if connection_ready else blocked_detail ) - enable_indexes_status = "limited" if connection_ready else "blocked" + enable_indexes_status = Status.LIMITED if connection_ready else Status.BLOCKED enable_indexes_detail = ( "Ready on SQLite; index DDL is supported, but clustering metadata will be skipped." if connection_ready else blocked_detail ) - postgresql_status = "unsupported" if connection_ready else "blocked" + postgresql_status = Status.UNSUPPORTED if connection_ready else Status.BLOCKED postgresql_detail = ( f"Requires PostgreSQL. Current backend: {current_backend}." if connection_ready @@ -210,25 +211,25 @@ def _command_support_for_backend( else blocked_detail ) else: - analyze_status = "unsupported" if connection_ready else "blocked" + analyze_status = Status.UNSUPPORTED if connection_ready else Status.BLOCKED analyze_detail = ( f"Requires PostgreSQL or SQLite. Current backend: {current_backend}." if connection_ready else blocked_detail ) - enable_indexes_status = "limited" if connection_ready else "blocked" + enable_indexes_status = Status.LIMITED if connection_ready else Status.BLOCKED enable_indexes_detail = ( f"Ready on {current_backend}; index DDL is supported, but clustering metadata will be skipped." if connection_ready else blocked_detail ) - postgresql_status = "unsupported" if connection_ready else "blocked" + postgresql_status = Status.UNSUPPORTED if connection_ready else Status.BLOCKED postgresql_detail = ( f"Requires PostgreSQL. Current backend: {current_backend}." if connection_ready else blocked_detail ) - vocab_load_status = "unsupported" if connection_ready else "blocked" + vocab_load_status = Status.UNSUPPORTED if connection_ready else Status.BLOCKED vocab_load_detail = ( f"Requires SQLite or PostgreSQL plus a configured Athena source path. Current backend: {current_backend}." if connection_ready @@ -248,13 +249,13 @@ def _command_support_for_backend( "backup-database", "PostgreSQL + pg_dump", ( - "ready" + Status.READY if connection_ready and backend == SupportedDialect.POSTGRESQL and pg_dump_path is not None - else "blocked" + else Status.BLOCKED if backend == SupportedDialect.POSTGRESQL - else "unsupported" + else Status.UNSUPPORTED if connection_ready - else "blocked" + else Status.BLOCKED ), ( "Ready on PostgreSQL; `pg_dump` is available." @@ -270,13 +271,13 @@ def _command_support_for_backend( "restore-database", "PostgreSQL + pg_restore/psql", ( - "ready" + Status.READY if connection_ready and backend == SupportedDialect.POSTGRESQL and (pg_restore_path is not None or psql_path is not None) - else "blocked" + else Status.BLOCKED if backend == SupportedDialect.POSTGRESQL - else "unsupported" + else Status.UNSUPPORTED if connection_ready - else "blocked" + else Status.BLOCKED ), ( "Ready on PostgreSQL; restore client tooling is available." diff --git a/omop_alchemy/maintenance/cli_schema_reconcile.py b/omop_alchemy/maintenance/cli_schema_reconcile.py index 8f70826..772c567 100644 --- a/omop_alchemy/maintenance/cli_schema_reconcile.py +++ b/omop_alchemy/maintenance/cli_schema_reconcile.py @@ -8,13 +8,32 @@ from sqlalchemy.engine.interfaces import ReflectedForeignKeyConstraint, ReflectedIndex from ..backends import backend_supports, resolve_backend -from .cli_indexes import _cluster_target_name +from ._cli_utils import Severity, Status +from .cli_indexes import _cluster_target_name, _find_equivalent_index from .tables import ( TableCategory, select_maintenance_tables, ) +def is_blocking_issue(issue: ReconciliationIssue) -> bool: + """Whether a reconciliation issue represents actual drift requiring attention. + + Parameters + ---------- + issue : ReconciliationIssue + A single reconciliation issue. + + Returns + ------- + bool + True unless the issue's status is Severity.ERROR-below (currently + only Status.RENAMED, an intentionally supported state -- e.g. an + index present under a foreign name -- rather than actual drift). + """ + return issue.status.severity == Severity.ERROR + + @dataclass(frozen=True) class ReconciliationIssue: """A single schema drift detail: column, index, FK, or cluster mismatch between ORM metadata and the database.""" @@ -23,7 +42,7 @@ class ReconciliationIssue: category: TableCategory component: str object_name: str - status: str + status: Status expected: str | None actual: str | None detail: str @@ -35,7 +54,7 @@ class TableReconciliationResult: table_name: str category: TableCategory - status: str + status: Status issue_count: int detail: str @@ -146,7 +165,7 @@ def reconcile_schema( category=maintenance_table.category, component="table", object_name=maintenance_table.table_name, - status="missing", + status=Status.MISSING, expected="present", actual="absent", detail="ORM-managed table is missing from the target database.", @@ -156,7 +175,7 @@ def reconcile_schema( TableReconciliationResult( table_name=maintenance_table.table_name, category=maintenance_table.category, - status="missing", + status=Status.MISSING, issue_count=1, detail="Table is missing from the target database.", ) @@ -183,7 +202,7 @@ def reconcile_schema( category=maintenance_table.category, component="column", object_name=column_name, - status="missing", + status=Status.MISSING, expected=_normalized_type(column.type, engine.dialect), actual=None, detail="Column is defined in ORM metadata but missing from the database.", @@ -198,7 +217,7 @@ def reconcile_schema( category=maintenance_table.category, component="column", object_name=column_name, - status="unexpected", + status=Status.UNEXPECTED, expected=None, actual=_normalized_type(column["type"], engine.dialect), detail="Column exists in the database but is not defined in ORM metadata.", @@ -217,7 +236,7 @@ def reconcile_schema( category=maintenance_table.category, component="column", object_name=column_name, - status="mismatch", + status=Status.MISMATCH, expected=expected_type, actual=actual_type, detail="Column type differs from ORM metadata.", @@ -233,7 +252,7 @@ def reconcile_schema( category=maintenance_table.category, component="column", object_name=column_name, - status="mismatch", + status=Status.MISMATCH, expected="nullable" if expected_nullable else "not nullable", actual="nullable" if actual_nullable else "not nullable", detail="Column nullability differs from ORM metadata.", @@ -247,7 +266,7 @@ def reconcile_schema( category=maintenance_table.category, component="primary_key", object_name=maintenance_table.table_name, - status="mismatch", + status=Status.MISMATCH, expected=", ".join(expected_pk_names), actual=", ".join(actual_pk_names) if actual_pk_names else None, detail="Primary key columns differ from ORM metadata.", @@ -266,7 +285,7 @@ def reconcile_schema( category=maintenance_table.category, component="foreign_key", object_name=constraint.name if isinstance(constraint.name, str) else ",".join(constrained_columns), - status="missing", + status=Status.MISSING, expected=f"{','.join(constrained_columns)} -> {referred_table}({','.join(referred_columns)})", actual=None, detail="Foreign key is defined in ORM metadata but missing from the database.", @@ -282,7 +301,7 @@ def reconcile_schema( category=maintenance_table.category, component="foreign_key", object_name=str(foreign_key.get("name") or ",".join(constrained_columns)), - status="unexpected", + status=Status.UNEXPECTED, expected=None, actual=f"{','.join(constrained_columns)} -> {referred_table}({','.join(referred_columns)})", detail="Foreign key exists in the database but is not defined in ORM metadata.", @@ -291,21 +310,45 @@ def reconcile_schema( expected_idxs = _expected_indexes(expected_table) actual_idxs = _actual_indexes(inspector, maintenance_table.table_name, db_schema) + actual_index_list = list(actual_idxs.values()) + renamed_actual_names: set[str] = set() for index_name, index in expected_idxs.items(): if index_name not in actual_idxs: - table_issues.append( - ReconciliationIssue( - table_name=maintenance_table.table_name, - category=maintenance_table.category, - component="index", - object_name=index_name, - status="missing", - expected=", ".join(column.name for column in index.columns), - actual=None, - detail="Index is defined in ORM metadata but missing from the database.", - ) + expected_columns_for_index = tuple(column.name for column in index.columns) + equivalent_name = _find_equivalent_index( + actual_index_list, expected_columns_for_index, bool(index.unique) ) + if equivalent_name is not None: + renamed_actual_names.add(equivalent_name) + table_issues.append( + ReconciliationIssue( + table_name=maintenance_table.table_name, + category=maintenance_table.category, + component="index", + object_name=index_name, + status=Status.RENAMED, + expected=index_name, + actual=equivalent_name, + detail=( + f"Index is present under a different name ('{equivalent_name}') " + f"than ORM metadata expects ('{index_name}')." + ), + ) + ) + else: + table_issues.append( + ReconciliationIssue( + table_name=maintenance_table.table_name, + category=maintenance_table.category, + component="index", + object_name=index_name, + status=Status.MISSING, + expected=", ".join(column.name for column in index.columns), + actual=None, + detail="Index is defined in ORM metadata but missing from the database.", + ) + ) continue actual_index = actual_idxs[index_name] @@ -318,7 +361,7 @@ def reconcile_schema( category=maintenance_table.category, component="index", object_name=index_name, - status="mismatch", + status=Status.MISMATCH, expected=", ".join(expected_columns_for_index), actual=", ".join(actual_columns_for_index) if actual_columns_for_index else None, detail="Index columns differ from ORM metadata.", @@ -331,7 +374,7 @@ def reconcile_schema( category=maintenance_table.category, component="index", object_name=index_name, - status="mismatch", + status=Status.MISMATCH, expected="unique" if index.unique else "non-unique", actual="unique" if actual_index.get("unique") else "non-unique", detail="Index uniqueness differs from ORM metadata.", @@ -339,14 +382,14 @@ def reconcile_schema( ) for index_name, index in actual_idxs.items(): - if index_name not in expected_idxs: + if index_name not in expected_idxs and index_name not in renamed_actual_names: table_issues.append( ReconciliationIssue( table_name=maintenance_table.table_name, category=maintenance_table.category, component="index", object_name=index_name, - status="unexpected", + status=Status.UNEXPECTED, expected=None, actual=", ".join(c for c in (index.get("column_names") or []) if c is not None), detail="Index exists in the database but is not defined in ORM metadata.", @@ -361,26 +404,57 @@ def reconcile_schema( db_schema, ) if expected_cluster != actual_cluster: - table_issues.append( - ReconciliationIssue( - table_name=maintenance_table.table_name, - category=maintenance_table.category, - component="cluster", - object_name=maintenance_table.table_name, - status=( - "missing" - if expected_cluster and not actual_cluster - else "unexpected" - if actual_cluster and not expected_cluster - else "mismatch" - ), - expected=expected_cluster, - actual=actual_cluster, - detail="Table clustering differs from ORM metadata.", + # The table may be physically clustered on a foreign-named + # equivalent of the ORM's cluster index. That's the + # same "renamed" state as an index found under a different + # name, so reuse the same equivalence check (including its + # plain-index safety filtering) before falling back to + # missing/unexpected/mismatch. + renamed_cluster = False + if expected_cluster in expected_idxs and actual_cluster is not None: + expected_cluster_index = expected_idxs[expected_cluster] + expected_cluster_columns = tuple( + column.name for column in expected_cluster_index.columns + ) + equivalent_cluster_name = _find_equivalent_index( + actual_index_list, expected_cluster_columns, bool(expected_cluster_index.unique) + ) + renamed_cluster = equivalent_cluster_name == actual_cluster + if renamed_cluster: + table_issues.append( + ReconciliationIssue( + table_name=maintenance_table.table_name, + category=maintenance_table.category, + component="cluster", + object_name=maintenance_table.table_name, + status=Status.RENAMED, + expected=expected_cluster, + actual=actual_cluster, + detail="Table is clustered on a differently-named equivalent index.", + ) + ) + else: + table_issues.append( + ReconciliationIssue( + table_name=maintenance_table.table_name, + category=maintenance_table.category, + component="cluster", + object_name=maintenance_table.table_name, + status=( + Status.MISSING + if expected_cluster and not actual_cluster + else Status.UNEXPECTED + if actual_cluster and not expected_cluster + else Status.MISMATCH + ), + expected=expected_cluster, + actual=actual_cluster, + detail="Table clustering differs from ORM metadata.", + ) ) - ) - table_status = "matched" if not table_issues else "drifted" + blocking_issues = [issue for issue in table_issues if is_blocking_issue(issue)] + table_status = Status.MATCHED if not blocking_issues else Status.DRIFTED table_results.append( TableReconciliationResult( table_name=maintenance_table.table_name, diff --git a/omop_alchemy/maintenance/cli_schema_tables.py b/omop_alchemy/maintenance/cli_schema_tables.py index 9a643bc..83f2473 100644 --- a/omop_alchemy/maintenance/cli_schema_tables.py +++ b/omop_alchemy/maintenance/cli_schema_tables.py @@ -6,7 +6,7 @@ import sqlalchemy as sa -from ._cli_utils import dry_label, dry_status, ensure_schema +from ._cli_utils import Status, dry_label, dry_status, ensure_schema, reject_reserved_schema from .tables import ( MaintenanceTable, TableCategory, @@ -23,7 +23,7 @@ class TableCreationResult: table_name: str category: TableCategory model_name: str - status: str + status: Status detail: str @@ -62,6 +62,7 @@ def create_missing_tables( dry_run: bool = False, ) -> list[TableCreationResult]: """Create any ORM-managed tables missing from the target database. Skips tables with unresolved FK dependencies.""" + reject_reserved_schema(db_schema) if not dry_run: ensure_schema(engine, db_schema) inspector = sa.inspect(engine) @@ -111,9 +112,9 @@ def create_missing_tables( category=maintenance_table.category, model_name=maintenance_table.model_name, status=( - "blocked" + Status.BLOCKED if blocked is not None - else dry_status(dry_run, applied="created") + else dry_status(dry_run, applied=Status.CREATED) ), detail=( "table blocked by unresolved dependencies: " + ", ".join(blocked) diff --git a/omop_alchemy/maintenance/cli_tables.py b/omop_alchemy/maintenance/cli_tables.py index 2e5718a..5134356 100644 --- a/omop_alchemy/maintenance/cli_tables.py +++ b/omop_alchemy/maintenance/cli_tables.py @@ -8,7 +8,7 @@ import typer from ..backends import resolve_backend, require_backend_support, backend_support_note -from ._cli_utils import dry_label, dry_status, omop_command, resolve_selection +from ._cli_utils import Status, dry_label, dry_status, omop_command, reject_reserved_schema, resolve_selection from .tables import ( TableCategory, TableScope, @@ -41,7 +41,7 @@ class AnalyzeTableResult: table_name: str category: TableCategory operation: str - status: str + status: Status detail: str @@ -55,6 +55,7 @@ def analyze_tables( dry_run: bool = False, ) -> list[AnalyzeTableResult]: """Run ANALYZE (or VACUUM ANALYZE) on selected ORM-managed tables to refresh planner statistics.""" + reject_reserved_schema(db_schema) if scope is not None and table_names is not None: raise RuntimeError("Use either `scope` or `table_names`, not both.") @@ -78,7 +79,7 @@ def analyze_tables( table_name=maintenance_table.table_name, category=maintenance_table.category, operation=operation, - status="skipped", + status=Status.SKIPPED, detail="table not present in target database", ) ) @@ -111,7 +112,7 @@ class TruncateTableResult: table_name: str category: TableCategory row_count: int | None - status: str + status: Status detail: str @@ -165,6 +166,7 @@ def truncate_tables( dry_run: bool = False, ) -> list[TruncateTableResult]: """Truncate selected ORM-managed tables. Raises if non-selected tables hold blocking FK references.""" + reject_reserved_schema(db_schema) if scope is not None and table_names is not None: raise RuntimeError("Use either `scope` or `table_names`, not both.") if scope is None and table_names is None: @@ -185,7 +187,7 @@ def truncate_tables( table_name=maintenance_table.table_name, category=maintenance_table.category, row_count=None, - status="skipped", + status=Status.SKIPPED, detail="table not present in target database", ) ) @@ -250,7 +252,7 @@ class SequenceResetResult: pk_column_name: str sequence_name: str | None next_value: int | None - status: str + status: Status detail: str @@ -285,6 +287,7 @@ def reset_model_sequences( dry_run: bool = False, ) -> list[SequenceResetResult]: """Reset each owned sequence to MAX(pk_column) + 1 to prevent insert conflicts after bulk loads.""" + reject_reserved_schema(db_schema) backend = resolve_backend(engine) require_backend_support(backend, "find_sequence_name", "Sequence reset") inspector = sa.inspect(engine) @@ -308,7 +311,7 @@ def reset_model_sequences( pk_column_name=target.pk_column_name, sequence_name=None, next_value=None, - status="skipped", + status=Status.SKIPPED, detail="no owned PostgreSQL sequence found", ) ) @@ -333,7 +336,7 @@ def reset_model_sequences( pk_column_name=target.pk_column_name, sequence_name=sequence_name, next_value=next_value, - status=dry_status(dry_run, applied="reset"), + status=dry_status(dry_run, applied=Status.RESET), detail=dry_label(dry_run, "sequence would be reset from table max + 1", "sequence reset from table max + 1"), ) ) diff --git a/omop_alchemy/maintenance/cli_vocab.py b/omop_alchemy/maintenance/cli_vocab.py index ca5a711..5e0d762 100644 --- a/omop_alchemy/maintenance/cli_vocab.py +++ b/omop_alchemy/maintenance/cli_vocab.py @@ -14,6 +14,7 @@ from sqlalchemy.exc import OperationalError import typer from sqlalchemy.pool import NullPool +from orm_loader.backends import resolve_backend from orm_loader.tables.typing import CSVTableProtocol from rich.progress import BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn, TimeElapsedColumn @@ -31,7 +32,7 @@ Vocabulary, ) -from ._cli_utils import ensure_schema, omop_command +from ._cli_utils import ReservedSchema, Status, ensure_schema, omop_command, reject_reserved_schema from .cli_foreign_keys import manage_foreign_key_triggers from .cli_indexes import manage_indexes from .cli_tables import reset_model_sequences @@ -39,6 +40,7 @@ from .ui import ( console, render_error, + render_vocab_index_warnings, render_vocab_load_results, render_vocab_load_summary, ) @@ -54,7 +56,7 @@ class VocabularyLoadResult: """Outcome of loading one Athena vocabulary CSV file via the staged ORM CSV loader.""" table_name: str - status: str + status: Status row_count: int | None csv_path: str | None required: bool @@ -72,6 +74,7 @@ class VocabularyLoadReport: created_table_count: int sequence_reset_count: int results: tuple[VocabularyLoadResult, ...] + index_warnings: tuple[str, ...] = () @dataclass(frozen=True) @@ -140,9 +143,10 @@ def _is_missing_staging_table_error( exc: Exception, *, model: VocabularyModel, + session: so.Session, ) -> bool: """Return True if the exception is a ProgrammingError caused by the staging table not existing yet.""" - staging_table_name = model.staging_tablename() + staging_table_name = resolve_backend(session).staging_name_for_table(model.__tablename__) message = str(exc).lower() return ( exc.__class__.__name__ == "ProgrammingError" @@ -175,7 +179,7 @@ def _load_vocab_model_csv( try: return int(model.load_csv(session, csv_path, **load_kwargs)) # type: ignore[arg-type] except Exception as exc: - if not _is_missing_staging_table_error(exc, model=model): + if not _is_missing_staging_table_error(exc, model=model, session=session): raise session.rollback() @@ -269,9 +273,11 @@ def load_vocab_source( + ", ".join(sorted(missing_required)) ) + reject_reserved_schema(db_schema) + if not dry_run: ensure_schema(engine, db_schema) - ensure_schema(engine, "staging") + ensure_schema(engine, ReservedSchema.STAGING) # NullPool: each session/connection is opened fresh and closed immediately after # use. No stale pooled connections survive between tables, which prevents @@ -282,12 +288,14 @@ def load_vocab_source( # SET search_path on the first checkout doesn't survive into the next # checkout (e.g. after create_staging_table's commit). Re-apply on every # new connection via an engine-level connect event so COPY and raw-cursor - # operations always target the right schema. + # operations always target the right schema. The staging schema no longer + # needs to be on the path because orm-loader now qualifies staging + # table identifiers explicitly. _quoted_schema = '"' + db_schema.replace('"', '""') + '"' @sae.listens_for(load_engine, "connect") def _set_search_path(dbapi_conn, _record): cur = dbapi_conn.cursor() - cur.execute(f"SET search_path TO staging, {_quoted_schema}") + cur.execute(f"SET search_path TO {_quoted_schema}") cur.close() all_models = REQUIRED_VOCAB_MODELS + OPTIONAL_VOCAB_MODELS @@ -301,6 +309,7 @@ def _set_search_path(dbapi_conn, _record): sequence_reset_count = 0 rows_cumulative = 0 table_index = 0 + index_warnings: tuple[str, ...] = () _emit(progress_callback, f"Preparing Athena vocabulary load for {table_count} CSV file(s)", 0.0, table_count=table_count) @@ -319,13 +328,25 @@ def _set_search_path(dbapi_conn, _record): dry_run=False, ) _emit(progress_callback, "Dropping indexes for bulk load...", 0.0, table_count=table_count) - manage_indexes( + disable_results = manage_indexes( engine, enable=False, vocabulary_included=True, db_schema=db_schema, dry_run=False, ) + index_warnings = tuple( + f"{result.table_name}.{result.index_name}: {result.detail}" + for result in disable_results + if result.status == Status.WARNING + ) + if index_warnings: + _emit( + progress_callback, + f"{len(index_warnings)} index(es) could not be optimized for bulk load (left in place)", + 0.0, + table_count=table_count, + ) if not dry_run: with load_engine.connect() as pre_conn: @@ -340,7 +361,7 @@ def _set_search_path(dbapi_conn, _record): if csv_path is None: results.append(VocabularyLoadResult( table_name=model.__tablename__, - status="skipped", + status=Status.SKIPPED, row_count=None, csv_path=None, required=required, @@ -361,7 +382,7 @@ def _set_search_path(dbapi_conn, _record): if dry_run: results.append(VocabularyLoadResult( table_name=model.__tablename__, - status="planned", + status=Status.PLANNED, row_count=None, csv_path=str(csv_path), required=required, @@ -419,7 +440,7 @@ def _set_search_path(dbapi_conn, _record): rows_cumulative += row_count results.append(VocabularyLoadResult( table_name=model.__tablename__, - status="loaded", + status=Status.LOADED, row_count=row_count, csv_path=str(csv_path), required=required, @@ -465,7 +486,7 @@ def _set_search_path(dbapi_conn, _record): dry_run=False, ) sequence_reset_count = sum( - result.status == "reset" + result.status == Status.RESET for result in sequence_results ) @@ -477,6 +498,7 @@ def _set_search_path(dbapi_conn, _record): created_table_count=created_table_count, sequence_reset_count=sequence_reset_count, results=tuple(results), + index_warnings=index_warnings, ) @@ -587,3 +609,6 @@ def _update_progress(event: VocabularyLoadProgress) -> None: console.print(render_vocab_load_results(report.results)) console.print(render_vocab_load_summary(report, dry_run=dry_run)) + index_warnings_panel = render_vocab_index_warnings(report) + if index_warnings_panel is not None: + console.print(index_warnings_panel) diff --git a/omop_alchemy/maintenance/ui.py b/omop_alchemy/maintenance/ui.py index 1e6380b..75c4b70 100644 --- a/omop_alchemy/maintenance/ui.py +++ b/omop_alchemy/maintenance/ui.py @@ -9,14 +9,13 @@ from rich.table import Table from rich.text import Text -from omop_alchemy.backends.base import FullTextResult - from ..backends.resolve import _DIALECT_TO_BACKEND_MAP, SupportedDialect as _SupportedDialect from .ascii import render_banner from .tables import TableCategory if TYPE_CHECKING: + from ._cli_utils import Status from .cli_backup import BackupResult from .cli_foreign_keys import ( ForeignKeyConstraintViolation, @@ -25,6 +24,7 @@ ForeignKeyValidationReport, ForeignKeyValidationResult, ) + from .cli_fulltext import FullTextResult from .cli_indexes import IndexManagementResult from .cli_schema import ( CommandSupport, @@ -43,25 +43,22 @@ console = Console() -STATUS_STYLES = { - "applied": "green", - "blocked": "red", - "drifted": "yellow", - "limited": "yellow", - "matched": "green", - "missing": "red", - "planned": "cyan", - "ready": "green", - "reset": "green", - "created": "green", - "loaded": "green", - "warning": "yellow", - "info": "cyan", - "skipped": "yellow", - "unsupported": "red", - "failed": "red", - "passed": "green", -} +def _status_style(status: Status) -> str: + """Resolve the render color for a status via its severity. + + Raises + ------ + TypeError + If the status is not a Status member (e.g., a plain string). + """ + # Local import to prevent circular import + from ._cli_utils import Status as _Status + + if not isinstance(status, _Status): + raise TypeError( + f"_status_style() expected a Status member, got {status!r} ({type(status).__name__})" + ) + return status.severity.style CATEGORY_STYLES = { TableCategory.CLINICAL: "bright_blue", @@ -140,8 +137,8 @@ def render_error(message: str, *, title: str = "Error") -> Panel: -def _status_text(status: str) -> Text: - return Text(status.upper(), style=STATUS_STYLES.get(status, "white")) +def _status_text(status: Status) -> Text: + return Text(status.upper(), style=_status_style(status)) def _optional_bool_label(value: bool | None) -> Text: @@ -343,9 +340,13 @@ def render_reconciliation_issues(issues: Iterable[ReconciliationIssue]) -> Rende def render_reconciliation_summary(report: SchemaReconciliationReport) -> Panel: + from .cli_schema_reconcile import is_blocking_issue + matched = sum(result.status == "matched" for result in report.table_results) drifted = sum(result.status == "drifted" for result in report.table_results) missing = sum(result.status == "missing" for result in report.table_results) + blocking_issues = [issue for issue in report.issues if is_blocking_issue(issue)] + informational_issue_count = len(report.issues) - len(blocking_issues) grid = Table.grid(padding=(0, 2)) grid.add_column(style="bold cyan") @@ -358,15 +359,17 @@ def render_reconciliation_summary(report: SchemaReconciliationReport) -> Panel: grid.add_row("Drifted", str(drifted)) if missing: grid.add_row("Missing", str(missing)) - grid.add_row("Issues", str(len(report.issues))) + grid.add_row("Issues", str(len(blocking_issues))) + if informational_issue_count: + grid.add_row("Renamed indexes", str(informational_issue_count)) grid.add_row( "Summary", - "Schema drift detected." if report.issues else "Database schema matches ORM metadata for the selected scope.", + "Schema drift detected." if blocking_issues else "Database schema matches ORM metadata for the selected scope.", ) return Panel.fit( grid, title="[bold]Summary[/bold]", - border_style="yellow" if report.issues else "green", + border_style="yellow" if blocking_issues else "green", ) @@ -498,7 +501,7 @@ def render_sequence_reset_results(results: Iterable[SequenceResetResult]) -> Ren table.add_column("Detail") for result in items: - style = STATUS_STYLES.get(result.status, "white") + style = _status_style(result.status) table.add_row( Text(result.status.upper(), style=style), f"{result.table_name}.{result.pk_column_name}", @@ -612,6 +615,8 @@ def render_vocab_load_summary(report: VocabularyLoadReport, *, dry_run: bool) -> grid.add_row("Reset sequences", str(report.sequence_reset_count)) if skipped_count: grid.add_row("Skipped", str(skipped_count)) + if report.index_warnings: + grid.add_row("Index warnings", str(len(report.index_warnings))) grid.add_row( "Summary", ( @@ -620,10 +625,25 @@ def render_vocab_load_summary(report: VocabularyLoadReport, *, dry_run: bool) -> else "Athena vocabulary source loaded into ORM-managed vocabulary tables." ), ) + border_style = "yellow" if report.index_warnings else ("cyan" if dry_run else "green") return Panel.fit( grid, title="[bold]Summary[/bold]", - border_style="cyan" if dry_run else "green", + border_style=border_style, + ) + + +def render_vocab_index_warnings(report: VocabularyLoadReport) -> Panel | None: + """Panel listing indexes that could not be optimized for bulk load (left in + place rather than dropped, since they couldn't be safely restored afterward). + Returns None when there's nothing to show.""" + if not report.index_warnings: + return None + body = "\n".join(f"- {message}" for message in report.index_warnings) + return Panel.fit( + body, + title="[bold yellow]Index Warnings[/bold yellow]", + border_style="yellow", ) @@ -641,7 +661,7 @@ def render_foreign_key_results(results: Iterable[ForeignKeyManagementResult]) -> table.add_column("Incoming", justify="right") for result in items: - style = STATUS_STYLES.get(result.status, "white") + style = _status_style(result.status) table.add_row( Text(result.status.upper(), style=style), "Enable" if result.enable else "Disable", @@ -842,7 +862,7 @@ def render_table_creation_results(results: Iterable[TableCreationResult]) -> Ren table.add_column("Model") table.add_column("Detail") for result in items: - style = STATUS_STYLES.get(result.status, "white") + style = _status_style(result.status) table.add_row( Text(result.status.upper(), style=style), result.table_name, @@ -876,8 +896,9 @@ def render_index_results(results: Iterable[IndexManagementResult]) -> Renderable table.add_column("Index") table.add_column("Category") table.add_column("Columns") + table.add_column("Detail") for result in items: - style = STATUS_STYLES.get(result.status, "white") + style = _status_style(result.status) table.add_row( Text(result.status.upper(), style=style), result.operation, @@ -886,6 +907,7 @@ def render_index_results(results: Iterable[IndexManagementResult]) -> Renderable result.index_name, _category_label(result.category), ", ".join(result.column_names), + result.detail, ) return table @@ -911,13 +933,17 @@ def render_index_summary(results: Iterable[IndexManagementResult], *, dry_run: b skipped = sum(item.status == "skipped" for item in items) if skipped: grid.add_row("Skipped", str(skipped)) + warnings = sum(item.status == "warning" for item in items) + if warnings: + grid.add_row("Warnings", str(warnings)) grid.add_row("Tables", str(len({item.table_name for item in items}))) action = ("enable" if items[0].enable else "disable") if items else "manage" grid.add_row( "Summary", f"{'Planned' if dry_run else 'Applied'} {action} on {len(items)} metadata operation(s).", ) - return Panel.fit(grid, title="[bold]Summary[/bold]", border_style="green" if not dry_run else "cyan") + border_style = "yellow" if warnings else ("green" if not dry_run else "cyan") + return Panel.fit(grid, title="[bold]Summary[/bold]", border_style=border_style) def render_fulltext_results(results: Iterable[FullTextResult]) -> RenderableType: diff --git a/tests/test_foreign_keys.py b/tests/test_foreign_keys.py index f91c1f6..dd539e0 100644 --- a/tests/test_foreign_keys.py +++ b/tests/test_foreign_keys.py @@ -3,6 +3,7 @@ from typer.testing import CliRunner from omop_alchemy.maintenance.cli import app +from omop_alchemy.maintenance._cli_utils import Status from omop_alchemy.maintenance.cli_schema import create_missing_tables from omop_alchemy.maintenance.cli_foreign_keys import ( ForeignKeyConstraintViolation, @@ -398,7 +399,7 @@ def fake_validate_foreign_key_constraints( incoming_constraint_count=0, violating_constraint_count=1, violating_row_count=3, - status="failed", + status=Status.FAILED, detail="3 violating row(s) across 1 constraint(s): fk_visit_occurrence_person_id_person (3)", ), ), diff --git a/tests/test_fulltext.py b/tests/test_fulltext.py index 37dbd78..0dfe1f2 100644 --- a/tests/test_fulltext.py +++ b/tests/test_fulltext.py @@ -6,14 +6,15 @@ from omop_alchemy.backends import ( CONCEPT_NAME_TSVECTOR_COLUMN, CONCEPT_SYNONYM_NAME_TSVECTOR_COLUMN, - FullTextAction, - FullTextResult, PostgresBackend, ) from omop_alchemy.cdm.model.vocabulary.concept import Concept from omop_alchemy.cdm.model.vocabulary.concept_synonym import Concept_Synonym +from omop_alchemy.maintenance._cli_utils import Status from omop_alchemy.maintenance.cli import app from omop_alchemy.maintenance.cli_fulltext import ( + FullTextAction, + FullTextResult, drop_fulltext_columns, install_fulltext_columns, populate_fulltext_columns, @@ -232,7 +233,7 @@ def fake_install_fulltext_columns( vector_column_name=CONCEPT_NAME_TSVECTOR_COLUMN, index_name="idx_gin_concept_name_tsvector", action=FullTextAction.INSTALL, - status="planned", + status=Status.PLANNED, detail="tsvector column and GIN index would be installed", ), ) diff --git a/tests/test_indexes.py b/tests/test_indexes.py index 7a2be4d..a2dad1a 100644 --- a/tests/test_indexes.py +++ b/tests/test_indexes.py @@ -7,8 +7,19 @@ from omop_alchemy.cdm.base.indexing import OMOP_CLUSTER_INDEX_INFO_KEY, omop_index_name from omop_alchemy.maintenance.cli import app from omop_alchemy.maintenance.cli_schema import create_missing_tables +from omop_alchemy.maintenance._cli_utils import ReservedSchema, Status, reject_reserved_schema +from omop_alchemy.maintenance.ui import render_index_summary from omop_alchemy.maintenance.cli_indexes import ( IndexManagementResult, + _DROPPED_INDEXES_TABLE_NAME, + get_bookkeeping_schema, + _cluster_index_unique, + _describe_shape_conflict, + _find_equivalent_index, + _find_shape_conflict, + _is_plain_index, + _record_captured_index, + _resolve_physical_cluster_name, _schema_metadata_indexes, collect_index_targets, manage_indexes, @@ -358,7 +369,7 @@ def fake_manage_indexes( unique=False, clustered=False, enable=enable, - status="planned", + status=Status.PLANNED, detail="metadata-defined index would be dropped", ) ] @@ -422,7 +433,7 @@ def fake_manage_indexes( unique=False, clustered=False, enable=enable, - status="planned", + status=Status.PLANNED, detail="metadata-defined index would be created", ) ] @@ -446,3 +457,507 @@ def fake_manage_indexes( assert calls["cluster"] is False assert calls["enable"] is True assert calls["dry_run"] is True + + +# ── Column-set equivalence helpers ────────────────────────────────────────────── + +_PLAIN = {"name": "idx_person_id", "column_names": ["person_id"], "unique": False} +_EXPRESSION = { + "name": "idx_lower", + "column_names": [None], + "expressions": ["lower(x)"], + "unique": False, +} +_PARTIAL = { + "name": "idx_partial", + "column_names": ["gender_concept_id"], + "unique": False, + "dialect_options": {"postgresql_where": "gender_concept_id IS NOT NULL"}, +} +_NON_BTREE = { + "name": "idx_gin", + "column_names": ["x"], + "unique": False, + "dialect_options": {"postgresql_using": "gin"}, +} +_UNIQUE_PLAIN = {"name": "uq_person_id", "column_names": ["person_id"], "unique": True} + + +def test_is_plain_index_true_for_ordinary_reflected_index(): + assert _is_plain_index(_PLAIN) is True + + +def test_is_plain_index_false_for_expression_index(): + assert _is_plain_index(_EXPRESSION) is False + + +def test_is_plain_index_false_for_partial_index(): + assert _is_plain_index(_PARTIAL) is False + + +def test_is_plain_index_false_for_non_btree_index(): + assert _is_plain_index(_NON_BTREE) is False + + +def test_find_equivalent_index_is_order_sensitive(): + existing = [{"name": "idx_ab", "column_names": ["b", "a"], "unique": False}] + assert _find_equivalent_index(existing, ("a", "b"), False) is None + existing = [{"name": "idx_ab", "column_names": ["a", "b"], "unique": False}] + assert _find_equivalent_index(existing, ("a", "b"), False) == "idx_ab" + + +def test_find_equivalent_index_respects_unique_flag(): + existing = [_UNIQUE_PLAIN] + assert _find_equivalent_index(existing, ("person_id",), False) is None + assert _find_equivalent_index(existing, ("person_id",), True) == "uq_person_id" + + +def test_find_equivalent_index_ignores_non_plain_matches(): + existing = [_PARTIAL] + assert _find_equivalent_index(existing, ("gender_concept_id",), False) is None + + +def test_find_shape_conflict_detects_partial_and_non_btree_but_not_plain(): + existing = [_PLAIN, _PARTIAL, _NON_BTREE] + assert _find_shape_conflict(existing, ("person_id",), False) is None + assert _find_shape_conflict(existing, ("gender_concept_id",), False) is _PARTIAL + assert _find_shape_conflict(existing, ("x",), False) is _NON_BTREE + + +def test_describe_shape_conflict_mentions_reason(): + assert "partial WHERE predicate" in _describe_shape_conflict(_PARTIAL) + assert "non-btree access method 'gin'" in _describe_shape_conflict(_NON_BTREE) + + +# ── Reserved schema guard ──────────────────────────────────────────────────────── + + +def test_reject_reserved_schema_rejects_staging_and_maintenance(): + with pytest.raises(RuntimeError): + reject_reserved_schema(ReservedSchema.STAGING.value) + with pytest.raises(RuntimeError): + reject_reserved_schema(ReservedSchema.MAINTENANCE.value) + + +def test_reject_reserved_schema_allows_ordinary_schema(): + reject_reserved_schema("public") + reject_reserved_schema(None) + + +# ── Foreign-named equivalent index reconciliation ──────────────────────────────── + + +def _replace_with_foreign_index(engine: sa.Engine, *, foreign_name: str) -> None: + """Simulate a prepopulated database indexed by the official OHDSI CDM DDL + script: drop OMOP_Alchemy's own person.gender_concept_id index and create a + plain, differently-named index covering the same column instead.""" + with engine.begin() as connection: + connection.exec_driver_sql(f"DROP INDEX {PERSON_GENDER_INDEX}") + connection.exec_driver_sql( + f"CREATE INDEX {foreign_name} ON person (gender_concept_id)" + ) + + +def _person_gender_result(results: list[IndexManagementResult]) -> IndexManagementResult: + matches = [ + result + for result in results + if result.table_name == "person" + and result.operation == "index" + and result.column_names == ("gender_concept_id",) + ] + assert len(matches) == 1 + return matches[0] + + +def test_collect_index_targets_reports_foreign_named_equivalent_index(tmp_path): + engine = _fresh_engine(tmp_path) + _replace_with_foreign_index(engine, foreign_name="idx_gender") + + targets = { + (target.table_name, target.index_name) + for target in collect_index_targets(engine) + } + + assert ("person", "idx_gender") in targets + assert ("person", PERSON_GENDER_INDEX) not in targets + + +def test_manage_indexes_enable_skips_creation_when_foreign_equivalent_exists(tmp_path): + engine = _fresh_engine(tmp_path) + _replace_with_foreign_index(engine, foreign_name="idx_gender") + + results = manage_indexes(engine, enable=True, cluster=False) + result = _person_gender_result(results) + + assert result.status == "skipped" + assert "idx_gender" in result.detail + assert result.index_name == "idx_gender" + + inspector = sa.inspect(engine) + index_names = {index["name"] for index in inspector.get_indexes("person")} + assert "idx_gender" in index_names + assert PERSON_GENDER_INDEX not in index_names + + +def test_manage_indexes_disable_captures_and_drops_foreign_equivalent_index(tmp_path): + engine = _fresh_engine(tmp_path) + _replace_with_foreign_index(engine, foreign_name="idx_gender") + + results = manage_indexes(engine, enable=False) + result = _person_gender_result(results) + + assert result.status == "captured" + assert result.index_name == "idx_gender" + + inspector = sa.inspect(engine) + index_names = {index["name"] for index in inspector.get_indexes("person")} + assert "idx_gender" not in index_names + + bookkeeping = _dropped_indexes_rows(engine) + assert len(bookkeeping) == 1 + assert bookkeeping[0]["table_name"] == "person" + assert bookkeeping[0]["index_name"] == "idx_gender" + + +def test_manage_indexes_enable_restores_captured_foreign_index(tmp_path): + engine = _fresh_engine(tmp_path) + _replace_with_foreign_index(engine, foreign_name="idx_gender") + + manage_indexes(engine, enable=False) + + # A second, independent manage_indexes() call -- simulating a fresh CLI + # process -- must still be able to restore, since the capture lives in the + # database rather than in process memory. + results = manage_indexes(engine, enable=True, cluster=False) + result = _person_gender_result(results) + + assert result.status == "restored" + assert result.index_name == "idx_gender" + + inspector = sa.inspect(engine) + index_names = {index["name"] for index in inspector.get_indexes("person")} + assert "idx_gender" in index_names + assert PERSON_GENDER_INDEX not in index_names + + assert _dropped_indexes_rows(engine) == [] + + +def test_manage_indexes_disable_enable_round_trip_is_idempotent_with_capture(tmp_path): + engine = _fresh_engine(tmp_path) + _replace_with_foreign_index(engine, foreign_name="idx_gender") + + for _ in range(2): + manage_indexes(engine, enable=False) + manage_indexes(engine, enable=True, cluster=False) + + inspector = sa.inspect(engine) + index_names = [index["name"] for index in inspector.get_indexes("person")] + assert index_names.count("idx_gender") == 1 + assert _dropped_indexes_rows(engine) == [] + + +def test_manage_indexes_dry_run_previews_foreign_equivalent_without_mutating(tmp_path): + engine = _fresh_engine(tmp_path) + _replace_with_foreign_index(engine, foreign_name="idx_gender") + + results = manage_indexes(engine, enable=True, dry_run=True, cluster=False) + result = _person_gender_result(results) + + assert result.status == "skipped" + assert "idx_gender" in result.detail + + inspector = sa.inspect(engine) + assert not inspector.has_table( + _DROPPED_INDEXES_TABLE_NAME, schema=get_bookkeeping_schema(SQLiteBackend()) + ) + + +def test_bookkeeping_table_not_created_when_nothing_to_capture(tmp_path): + engine = _fresh_engine(tmp_path) + + manage_indexes(engine, enable=False) + manage_indexes(engine, enable=True, cluster=False) + + inspector = sa.inspect(engine) + assert not inspector.has_table( + _DROPPED_INDEXES_TABLE_NAME, schema=get_bookkeeping_schema(SQLiteBackend()) + ) + + +def test_manage_indexes_enable_cluster_uses_restored_physical_name(tmp_path, monkeypatch): + engine = _fresh_engine(tmp_path) + + with engine.begin() as connection: + connection.exec_driver_sql(f"DROP INDEX {EPISODE_PERSON_INDEX}") + connection.exec_driver_sql( + "CREATE INDEX idx_episode_person_id_1 ON episode (person_id)" + ) + + manage_indexes(engine, enable=False) + + calls: list[tuple[str, str]] = [] + + def fake_cluster_table(self, conn, table_name, index_name, db_schema): + calls.append((table_name, index_name)) + + monkeypatch.setattr(SQLiteBackend, "cluster_table", fake_cluster_table) + monkeypatch.setattr( + SQLiteBackend, + "analyze_table", + lambda self, conn, table_name, db_schema, *, vacuum=False: None, + ) + + manage_indexes(engine, enable=True, cluster=True) + + assert ("episode", "idx_episode_person_id_1") in calls + + +def test_manage_indexes_disable_warns_and_leaves_unsupported_foreign_index_in_place( + tmp_path, monkeypatch +): + engine = _fresh_engine(tmp_path) + + with engine.begin() as connection: + connection.exec_driver_sql(f"DROP INDEX {PERSON_GENDER_INDEX}") + connection.exec_driver_sql( + "CREATE INDEX idx_gender_partial ON person (gender_concept_id)" + ) + + original_get_indexes = sa.Inspector.get_indexes + + def fake_get_indexes(self, table_name, schema=None, **kw): + reflected = list(original_get_indexes(self, table_name, schema=schema, **kw)) + if table_name == "person": + for index in reflected: + if index["name"] == "idx_gender_partial": + index["dialect_options"] = { + "postgresql_where": "gender_concept_id IS NOT NULL" + } + return reflected + + monkeypatch.setattr(sa.Inspector, "get_indexes", fake_get_indexes) + + results = manage_indexes(engine, enable=False) + result = _person_gender_result(results) + + assert result.status == "warning" + assert "idx_gender_partial" in result.detail + assert "partial WHERE predicate" in result.detail + + inspector = sa.inspect(engine) + index_names = {index["name"] for index in inspector.get_indexes("person")} + assert "idx_gender_partial" in index_names + assert _dropped_indexes_rows(engine) == [] + + # The rest of the run must complete normally -- other tables' results are unaffected. + other_results = [r for r in results if r.table_name != "person"] + assert other_results + + +def _dropped_indexes_rows(engine: sa.Engine) -> list[dict[str, object]]: + bookkeeping_schema = get_bookkeeping_schema(SQLiteBackend()) + inspector = sa.inspect(engine) + if not inspector.has_table(_DROPPED_INDEXES_TABLE_NAME, schema=bookkeeping_schema): + return [] + with engine.connect() as connection: + rows = connection.exec_driver_sql( + f"SELECT table_name, index_name FROM {_DROPPED_INDEXES_TABLE_NAME}" + ).mappings().all() + return [dict(row) for row in rows] + + +def _render_to_text(renderable) -> str: + from rich.console import Console + import io + + buffer = io.StringIO() + Console(file=buffer, width=200).print(renderable) + return buffer.getvalue() + + +def _warning_result() -> IndexManagementResult: + return IndexManagementResult( + operation="index", + table_name="person", + category=TableCategory.CLINICAL, + index_name="idx_gender_partial", + column_names=("gender_concept_id",), + unique=False, + clustered=False, + enable=False, + status=Status.WARNING, + detail="foreign index 'idx_gender_partial' has a partial WHERE predicate; left in place", + ) + + +def test_render_index_summary_reports_warning_count(): + text = _render_to_text(render_index_summary([_warning_result()], dry_run=False)) + assert "Warnings" in text + assert "1" in text + + +def test_render_index_summary_omits_warnings_row_when_none(): + ok_result = IndexManagementResult( + operation="index", + table_name="person", + category=TableCategory.CLINICAL, + index_name=PERSON_GENDER_INDEX, + column_names=("gender_concept_id",), + unique=False, + clustered=False, + enable=True, + status=Status.APPLIED, + detail="metadata-defined index created", + ) + text = _render_to_text(render_index_summary([ok_result], dry_run=False)) + assert "Warnings" not in text + + +# ── Review-round-2 regression tests ────────────────────────────────────────────── + + +def test_is_plain_index_false_for_constraint_backed_index(): + """A foreign index reflecting with duplicates_constraint set backs a UNIQUE/ + PRIMARY KEY constraint. PostgreSQL refuses DROP INDEX on those, so it must + never be treated as a safe capture-and-drop equivalent.""" + constraint_backed = { + "name": "person_pkey", + "column_names": ["person_id"], + "unique": True, + "duplicates_constraint": "person_pkey", + } + assert _is_plain_index(constraint_backed) is False + assert _find_equivalent_index([constraint_backed], ("person_id",), True) is None + assert _find_shape_conflict([constraint_backed], ("person_id",), True) is constraint_backed + assert "constraint" in _describe_shape_conflict(constraint_backed) + + +def test_manage_indexes_disable_second_run_without_enable_degrades_to_warning(tmp_path): + """A second disable run, without an intervening enable, that finds a + different foreign index than the one already captured must not crash. + It must leave the second index in place and report a warning, since the + bookkeeping table can only track one pending capture per table/column-set.""" + engine = _fresh_engine(tmp_path) + _replace_with_foreign_index(engine, foreign_name="idx_gender_v1") + + first = manage_indexes(engine, enable=False) + assert _person_gender_result(first).status == "captured" + + with engine.begin() as connection: + connection.exec_driver_sql("CREATE INDEX idx_gender_v2 ON person (gender_concept_id)") + + second = manage_indexes(engine, enable=False) + result = _person_gender_result(second) + assert result.status == "warning" + assert "idx_gender_v2" in result.detail + + inspector = sa.inspect(engine) + index_names = {index["name"] for index in inspector.get_indexes("person")} + assert "idx_gender_v2" in index_names, "the untrackable second foreign index must be left in place" + + # The originally captured index (idx_gender_v1) must still be restorable. + third = manage_indexes(engine, enable=True, cluster=False) + restored = _person_gender_result(third) + assert restored.status == "restored" + assert restored.index_name == "idx_gender_v1" + + +def test_record_captured_index_scopes_by_db_schema(tmp_path): + """Two different schemas capturing an equivalent foreign index for a + same-named table/column-set must not collide. Each capture is independent + and neither should be rejected by the other's bookkeeping row.""" + engine = _fresh_engine(tmp_path) + backend = SQLiteBackend() + + with engine.begin() as connection: + captured_a = _record_captured_index( + connection, backend, + table_name="person", db_schema="site_a", index_name="idx_a", + column_names=("gender_concept_id",), unique=False, + ) + captured_b = _record_captured_index( + connection, backend, + table_name="person", db_schema="site_b", index_name="idx_b", + column_names=("gender_concept_id",), unique=False, + ) + + assert captured_a is True + assert captured_b is True + + # Capturing again for the *same* schema, with a different foreign name, must + # still be rejected (this is the case test_manage_indexes_disable_second_run_ + # without_enable_degrades_to_warning covers end-to-end). + with engine.begin() as connection: + captured_a_again = _record_captured_index( + connection, backend, + table_name="person", db_schema="site_a", index_name="idx_a_v2", + column_names=("gender_concept_id",), unique=False, + ) + assert captured_a_again is False + + +def test_resolve_physical_cluster_name_prefers_own_name(): + existing = [{"name": "ix_person_gender_concept_id", "column_names": ["gender_concept_id"], "unique": False}] + assert ( + _resolve_physical_cluster_name(existing, "ix_person_gender_concept_id", ("gender_concept_id",), False) + == "ix_person_gender_concept_id" + ) + + +def test_resolve_physical_cluster_name_falls_back_to_equivalent(): + existing = [{"name": "idx_gender", "column_names": ["gender_concept_id"], "unique": False}] + assert ( + _resolve_physical_cluster_name(existing, "ix_person_gender_concept_id", ("gender_concept_id",), False) + == "idx_gender" + ) + + +def test_resolve_physical_cluster_name_falls_back_to_own_name_when_nothing_matches(): + assert _resolve_physical_cluster_name([], "ix_person_gender_concept_id", ("gender_concept_id",), False) == ( + "ix_person_gender_concept_id" + ) + + +def test_cluster_index_unique_defaults_true_for_primary_key_target(): + tables = {table.table_name: table for table in collect_maintenance_tables()} + person = tables["person"] + # "pk_person" is the primary-key-based cluster target and is never one of + # person's secondary indexes. + assert _cluster_index_unique(person, "pk_person") is True + + +def test_render_reconciliation_summary_treats_renamed_only_as_matched(): + from omop_alchemy.maintenance.cli_schema_reconcile import ReconciliationIssue, SchemaReconciliationReport, TableReconciliationResult + from omop_alchemy.maintenance._cli_utils import Status + from omop_alchemy.maintenance.ui import render_reconciliation_summary + + report = SchemaReconciliationReport( + backend="sqlite", + table_results=( + TableReconciliationResult( + table_name="person", + category=TableCategory.CLINICAL, + status=Status.MATCHED, + issue_count=1, + detail="1 difference(s) detected.", + ), + ), + issues=( + ReconciliationIssue( + table_name="person", + category=TableCategory.CLINICAL, + component="index", + object_name=PERSON_GENDER_INDEX, + status=Status.RENAMED, + expected=PERSON_GENDER_INDEX, + actual="idx_gender", + detail="Index is present under a different name.", + ), + ), + ) + + text = _render_to_text(render_reconciliation_summary(report)) + assert "matches ORM metadata" in text + assert "drift detected" not in text.lower() + assert "Renamed indexes" in text diff --git a/tests/test_load_vocab_source.py b/tests/test_load_vocab_source.py index 44ebf02..25a3320 100644 --- a/tests/test_load_vocab_source.py +++ b/tests/test_load_vocab_source.py @@ -7,6 +7,8 @@ from typer.testing import CliRunner from omop_alchemy.maintenance.cli import app +from omop_alchemy.maintenance._cli_utils import Status +from omop_alchemy.maintenance.cli_indexes import IndexManagementResult from omop_alchemy.maintenance.cli_vocab import ( OPTIONAL_VOCAB_MODELS, REQUIRED_VOCAB_MODELS, @@ -14,6 +16,7 @@ _load_vocab_model_csv, load_vocab_source, ) +from omop_alchemy.maintenance.tables import TableCategory from omop_alchemy.cdm.model.vocabulary import Drug_Strength from omop_alchemy.config import OmopAlchemyConfig @@ -193,7 +196,7 @@ def fake_load_vocab_source( results=( VocabularyLoadResult( table_name="concept", - status="planned", + status=Status.PLANNED, row_count=None, csv_path=str(Path(source_path) / "CONCEPT.csv"), required=True, @@ -512,3 +515,134 @@ def fake_load_vocab_model_csv( f"Expected all tables to use quote_mode='auto', got: {received_quote_modes}" ) assert "literal" not in received_quote_modes + + +def test_load_vocab_source_bulk_mode_surfaces_index_warnings(monkeypatch, tmp_path): + """A foreign index that manage_indexes(enable=False) leaves in place (status=warning) + during the bulk-mode disable step must be surfaced on the returned report, not + silently discarded -- this is the only call site that inspects those results.""" + engine = sa.create_engine(f"sqlite:///{tmp_path / 'load_vocab_source_bulk.db'}", future=True) + source_path = _build_required_athena_source(tmp_path) + + # Force the bulk-mode gate (which requires a PostgreSQL engine) without needing + # a real Postgres instance -- everything downstream that's PostgreSQL-specific + # is mocked out below. + monkeypatch.setattr(engine.dialect, "name", "postgresql") + + monkeypatch.setattr( + "omop_alchemy.maintenance.cli_vocab._load_vocab_model_csv", + lambda session, *, model, csv_path, merge_strategy, quote_mode="auto", + chunksize=None, index_strategy="auto", merge_batch_size=1_000_000: 1, + ) + # ensure_schema() resolves a backend from engine.dialect.name too, and would + # otherwise try to run real PostgreSQL "CREATE SCHEMA" DDL against this SQLite + # connection now that the dialect name is faked above. + monkeypatch.setattr( + "omop_alchemy.maintenance.cli_vocab.ensure_schema", + lambda engine, schema: None, + ) + monkeypatch.setattr( + "omop_alchemy.maintenance.cli_vocab.manage_foreign_key_triggers", + lambda engine, **kwargs: [], + ) + monkeypatch.setattr( + "omop_alchemy.maintenance.cli_vocab.reset_model_sequences", + lambda engine, **kwargs: [], + ) + + disable_calls: list[bool] = [] + + def fake_manage_indexes(engine, *, enable, **kwargs): + disable_calls.append(enable) + if not enable: + return [ + IndexManagementResult( + operation="index", + table_name="concept", + category=TableCategory.VOCABULARY, + index_name="idx_concept_partial", + column_names=("domain_id",), + unique=False, + clustered=False, + enable=False, + status=Status.WARNING, + detail="foreign index 'idx_concept_partial' has a partial WHERE predicate; left in place", + ) + ] + # The re-enable call must not contribute to index_warnings. + return [ + IndexManagementResult( + operation="index", + table_name="concept", + category=TableCategory.VOCABULARY, + index_name="ix_concept_domain_id", + column_names=("domain_id",), + unique=False, + clustered=False, + enable=True, + status=Status.APPLIED, + detail="metadata-defined index created", + ) + ] + + monkeypatch.setattr( + "omop_alchemy.maintenance.cli_vocab.manage_indexes", + fake_manage_indexes, + ) + + report = load_vocab_source(engine, source_path=source_path, bulk_mode=True) + + assert disable_calls == [False, True] + assert report.index_warnings == ( + "concept.idx_concept_partial: foreign index 'idx_concept_partial' " + "has a partial WHERE predicate; left in place", + ) + + +def test_render_vocab_index_warnings_none_when_no_warnings(): + from omop_alchemy.maintenance.cli_vocab import VocabularyLoadReport + from omop_alchemy.maintenance.ui import render_vocab_index_warnings + + report = VocabularyLoadReport( + source_path="/tmp/source", + backend="sqlite", + db_schema=None, + merge_strategy="replace", + created_table_count=0, + sequence_reset_count=0, + results=(), + ) + assert render_vocab_index_warnings(report) is None + + +def test_render_vocab_index_warnings_lists_messages_when_present(): + import io + + from rich.console import Console + + from omop_alchemy.maintenance.cli_vocab import VocabularyLoadReport + from omop_alchemy.maintenance.ui import render_vocab_index_warnings, render_vocab_load_summary + + report = VocabularyLoadReport( + source_path="/tmp/source", + backend="postgresql", + db_schema=None, + merge_strategy="replace", + created_table_count=0, + sequence_reset_count=0, + results=(), + index_warnings=("concept.idx_concept_partial: has a partial WHERE predicate; left in place",), + ) + + panel = render_vocab_index_warnings(report) + assert panel is not None + buffer = io.StringIO() + Console(file=buffer, width=200).print(panel) + text = buffer.getvalue() + assert "idx_concept_partial" in text + + summary_buffer = io.StringIO() + Console(file=summary_buffer, width=200).print(render_vocab_load_summary(report, dry_run=False)) + summary_text = summary_buffer.getvalue() + assert "Index warnings" in summary_text + assert "1" in summary_text diff --git a/tests/test_schema_reconcile.py b/tests/test_schema_reconcile.py new file mode 100644 index 0000000..47d2e3e --- /dev/null +++ b/tests/test_schema_reconcile.py @@ -0,0 +1,139 @@ +import sqlalchemy as sa + +from omop_alchemy.backends.sqlite import SQLiteBackend +from omop_alchemy.cdm.base.indexing import omop_index_name +from omop_alchemy.maintenance.cli_schema import create_missing_tables +from omop_alchemy.maintenance.cli_schema_reconcile import is_blocking_issue, reconcile_schema + +PERSON_GENDER_INDEX = omop_index_name("person", "gender_concept_id") +EPISODE_PERSON_INDEX = omop_index_name("episode", "person_id") + + +def _fresh_engine(tmp_path): + db_path = tmp_path / "reconcile.db" + engine = sa.create_engine(f"sqlite:///{db_path}", future=True) + create_missing_tables(engine) + return engine + + +def _person_gender_issues(report): + return [ + issue + for issue in report.issues + if issue.table_name == "person" + and issue.component == "index" + and (issue.object_name == PERSON_GENDER_INDEX or issue.actual == "idx_gender") + ] + + +def test_reconcile_schema_reports_no_drift_on_fresh_database(tmp_path): + engine = _fresh_engine(tmp_path) + report = reconcile_schema(engine) + + person_result = next(r for r in report.table_results if r.table_name == "person") + assert person_result.status == "matched" + assert person_result.issue_count == 0 + + +def test_reconcile_schema_reports_renamed_for_foreign_named_equivalent_index(tmp_path): + engine = _fresh_engine(tmp_path) + with engine.begin() as connection: + connection.exec_driver_sql(f"DROP INDEX {PERSON_GENDER_INDEX}") + connection.exec_driver_sql("CREATE INDEX idx_gender ON person (gender_concept_id)") + + report = reconcile_schema(engine) + issues = _person_gender_issues(report) + + assert len(issues) == 1 + issue = issues[0] + assert issue.status == "renamed" + assert issue.expected == PERSON_GENDER_INDEX + assert issue.actual == "idx_gender" + + +def test_reconcile_schema_renamed_index_does_not_flip_table_to_drifted(tmp_path): + engine = _fresh_engine(tmp_path) + with engine.begin() as connection: + connection.exec_driver_sql(f"DROP INDEX {PERSON_GENDER_INDEX}") + connection.exec_driver_sql("CREATE INDEX idx_gender ON person (gender_concept_id)") + + report = reconcile_schema(engine) + person_result = next(r for r in report.table_results if r.table_name == "person") + + assert person_result.status == "matched" + assert person_result.issue_count == 1 + + +def test_is_blocking_issue_excludes_renamed_only(): + from omop_alchemy.maintenance.cli_schema_reconcile import ReconciliationIssue + from omop_alchemy.maintenance._cli_utils import Status + from omop_alchemy.maintenance.tables import TableCategory + + renamed = ReconciliationIssue( + table_name="person", category=TableCategory.CLINICAL, component="index", + object_name=PERSON_GENDER_INDEX, status=Status.RENAMED, + expected=PERSON_GENDER_INDEX, actual="idx_gender", detail="...", + ) + missing = ReconciliationIssue( + table_name="person", category=TableCategory.CLINICAL, component="index", + object_name=PERSON_GENDER_INDEX, status=Status.MISSING, + expected=PERSON_GENDER_INDEX, actual=None, detail="...", + ) + assert is_blocking_issue(renamed) is False + assert is_blocking_issue(missing) is True + + +def test_reconcile_schema_cluster_check_reports_renamed_for_foreign_cluster_index(tmp_path, monkeypatch): + """A table physically clustered on a foreign-named equivalent of the ORM's + cluster index (e.g. captured/restored under its original name by + manage_indexes()) must report a 'renamed' cluster issue, not 'mismatch'.""" + engine = _fresh_engine(tmp_path) + with engine.begin() as connection: + connection.exec_driver_sql(f"DROP INDEX {EPISODE_PERSON_INDEX}") + connection.exec_driver_sql("CREATE INDEX idx_episode_person ON episode (person_id)") + + monkeypatch.setattr( + SQLiteBackend, + "get_clustered_index_name", + lambda self, conn, table_name, db_schema: ( + "idx_episode_person" if table_name == "episode" else None + ), + ) + + report = reconcile_schema(engine) + episode_result = next(r for r in report.table_results if r.table_name == "episode") + cluster_issues = [ + issue for issue in report.issues + if issue.table_name == "episode" and issue.component == "cluster" + ] + + assert len(cluster_issues) == 1 + assert cluster_issues[0].status == "renamed" + assert cluster_issues[0].expected == EPISODE_PERSON_INDEX + assert cluster_issues[0].actual == "idx_episode_person" + assert episode_result.status == "matched" + + +def test_reconcile_schema_cluster_check_still_reports_real_mismatch(tmp_path, monkeypatch): + """A genuinely different physical cluster state (not just a foreign-named + equivalent) must still be reported as drift.""" + engine = _fresh_engine(tmp_path) + + monkeypatch.setattr( + SQLiteBackend, + "get_clustered_index_name", + lambda self, conn, table_name, db_schema: ( + "some_unrelated_index" if table_name == "episode" else None + ), + ) + + report = reconcile_schema(engine) + episode_result = next(r for r in report.table_results if r.table_name == "episode") + cluster_issues = [ + issue for issue in report.issues + if issue.table_name == "episode" and issue.component == "cluster" + ] + + assert len(cluster_issues) == 1 + assert cluster_issues[0].status == "mismatch" + assert episode_result.status == "drifted" diff --git a/tests/test_truncate_tables.py b/tests/test_truncate_tables.py index 2467359..ffca6ad 100644 --- a/tests/test_truncate_tables.py +++ b/tests/test_truncate_tables.py @@ -5,6 +5,7 @@ from oa_configurator import StackConfig, DatabaseConfig from omop_alchemy.maintenance.cli import app +from omop_alchemy.maintenance._cli_utils import Status from omop_alchemy.maintenance.cli_schema import create_missing_tables from omop_alchemy.maintenance.tables import TableCategory, TableScope from omop_alchemy.maintenance.cli_tables import TruncateTableResult, truncate_tables @@ -92,7 +93,7 @@ def fake_truncate_tables( table_name="person", category=TableCategory.CLINICAL, row_count=10, - status="planned", + status=Status.PLANNED, detail="table would be truncated", ) ]