126 lines
6.3 KiB
Python
126 lines
6.3 KiB
Python
"""Conservatively extend ORM tables through openGauss native catalogs.
|
|
|
|
This is used only through Dolphin's MySQL protocol. The stock MySQL inspector
|
|
cannot parse Dolphin's PostgreSQL-style ``SHOW CREATE TABLE`` result and can
|
|
therefore report existing columns as missing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from sqlalchemy import CheckConstraint, ForeignKeyConstraint, MetaData, UniqueConstraint
|
|
from sqlalchemy.engine import Connection
|
|
from sqlalchemy.schema import CreateColumn, CreateIndex
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _string(value: object) -> str:
|
|
return value.decode("utf-8") if isinstance(value, bytes) else str(value)
|
|
|
|
|
|
def ensure_gauss_schema_sync(connection: Connection, metadata: MetaData) -> None:
|
|
"""Add only safe missing columns and indexes involving those columns.
|
|
|
|
Existing columns and unrelated indexes are never altered. A missing
|
|
column without a nullable definition or database-side default cannot be
|
|
backfilled reliably and stops this repair before any DDL is issued.
|
|
"""
|
|
|
|
current_schema_value = connection.exec_driver_sql("SELECT current_schema()").scalar_one_or_none()
|
|
if not current_schema_value:
|
|
raise RuntimeError("openGauss schema integrity check: current_schema() is empty")
|
|
schema = _string(current_schema_value)
|
|
expected_schema = connection.engine.url.database
|
|
if expected_schema and expected_schema != schema:
|
|
raise RuntimeError(f"openGauss schema integrity check: URL schema {expected_schema!r} does not match current_schema() {schema!r}")
|
|
|
|
catalog_rows = connection.exec_driver_sql(
|
|
"SELECT c.relname, c.relkind, a.attname "
|
|
"FROM pg_catalog.pg_class AS c "
|
|
"JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace "
|
|
"LEFT JOIN pg_catalog.pg_attribute AS a ON a.attrelid = c.oid "
|
|
"AND a.attnum > 0 AND NOT a.attisdropped "
|
|
"WHERE n.nspname = %s AND c.relkind IN ('r', 'p', 'v', 'm', 'f')",
|
|
(schema,),
|
|
).fetchall()
|
|
columns_by_table: dict[str, set[str]] = {}
|
|
kinds_by_table: dict[str, str] = {}
|
|
for table_name, kind, column_name in catalog_rows:
|
|
table_name = _string(table_name)
|
|
kinds_by_table[table_name] = _string(kind)
|
|
columns_by_table.setdefault(table_name, set())
|
|
if column_name is not None:
|
|
columns_by_table[table_name].add(_string(column_name))
|
|
|
|
plan = []
|
|
failures: list[str] = []
|
|
for table in metadata.sorted_tables:
|
|
if table.schema is not None and table.schema != schema:
|
|
failures.append(f"{table.fullname}: explicit schema differs from {schema!r}")
|
|
continue
|
|
existing = columns_by_table.get(table.name)
|
|
if existing is None:
|
|
failures.append(f"{schema}.{table.name}: table is still missing after create_all")
|
|
continue
|
|
if kinds_by_table[table.name] not in {"r", "p"}:
|
|
failures.append(f"{schema}.{table.name}: existing relation is not an ordinary table")
|
|
continue
|
|
missing = [column for column in table.columns if column.name not in existing]
|
|
if not missing:
|
|
continue
|
|
missing_names = {column.name for column in missing}
|
|
for column in missing:
|
|
label = f"{schema}.{table.name}.{column.name}"
|
|
if not column.nullable and column.server_default is None:
|
|
failures.append(f"{label}: NOT NULL column has no server_default; explicit migration required")
|
|
elif column.primary_key or column.identity is not None or column.computed is not None:
|
|
failures.append(f"{label}: primary-key/identity/computed column requires an explicit migration")
|
|
for constraint in table.constraints:
|
|
if isinstance(constraint, (ForeignKeyConstraint, UniqueConstraint)):
|
|
constrained = {column.name for column in constraint.columns}
|
|
if constrained & missing_names:
|
|
failures.append(f"{schema}.{table.name}: missing columns participate in {type(constraint).__name__}; explicit constraint migration required")
|
|
elif isinstance(constraint, CheckConstraint):
|
|
constrained = {column.name for column in constraint.columns}
|
|
if not constrained or constrained & missing_names:
|
|
failures.append(f"{schema}.{table.name}: CHECK-constrained schema needs an explicit migration")
|
|
indexes = [index for index in table.indexes if {column.name for column in index.columns} & missing_names]
|
|
for index in indexes:
|
|
if not index.name:
|
|
failures.append(f"{schema}.{table.name}: index involving new columns has no stable name")
|
|
plan.append((table, missing, indexes))
|
|
|
|
if failures:
|
|
raise RuntimeError("openGauss schema is incomplete; startup repair stopped:\n- " + "\n- ".join(failures))
|
|
if not plan:
|
|
logger.info("openGauss schema integrity check passed: %s ORM tables", len(metadata.tables))
|
|
return
|
|
|
|
existing_indexes = {
|
|
(_string(table_name), _string(index_name))
|
|
for table_name, index_name in connection.exec_driver_sql(
|
|
"SELECT t.relname, i.relname FROM pg_catalog.pg_index AS x "
|
|
"JOIN pg_catalog.pg_class AS t ON t.oid = x.indrelid "
|
|
"JOIN pg_catalog.pg_class AS i ON i.oid = x.indexrelid "
|
|
"JOIN pg_catalog.pg_namespace AS n ON n.oid = t.relnamespace "
|
|
"WHERE n.nspname = %s",
|
|
(schema,),
|
|
).fetchall()
|
|
}
|
|
preparer = connection.dialect.identifier_preparer
|
|
added_columns = 0
|
|
for table, missing, indexes in plan:
|
|
qualified = f"{preparer.quote_schema(schema)}.{preparer.quote(table.name)}"
|
|
for column in missing:
|
|
definition = str(CreateColumn(column).compile(dialect=connection.dialect))
|
|
connection.exec_driver_sql(f"ALTER TABLE {qualified} ADD COLUMN {definition}")
|
|
added_columns += 1
|
|
logger.info("Added safe missing openGauss column: %s.%s", qualified, column.name)
|
|
for index in sorted(indexes, key=lambda item: str(item.name)):
|
|
if (table.name, str(index.name)) not in existing_indexes:
|
|
connection.execute(CreateIndex(index))
|
|
logger.info("Created openGauss index involving new columns: %s", index.name)
|
|
logger.info("openGauss schema integrity repair completed: %s columns", added_columns)
|