Fix/sql migration

CLAUDE.md

@@ -238,6 +238,12 @@ Per-user file quota enforcement tracked via the `file_count` and `file_quota` co
 
 **Migration:** `openrag/scripts/migrations/alembic/versions/c224d4befe71_add_file_count_and_file_quota.py`
 
+### Alembic Migration Idempotency
+
+`Base.metadata.create_all()` runs at app startup (`PartitionFileManager.__init__` in `openrag/components/indexer/vectordb/utils.py`), so a freshly bootstrapped database already contains the full current-model schema before alembic ever touches it. Migrations must therefore be **idempotent** — re-applying an `ADD COLUMN` / `CREATE TABLE` / `CREATE INDEX` against an already-existing object would raise `DuplicateColumn` / `DuplicateTable`.
+
+Guard every schema-mutating op with an inspector-based existence check (`table_exists`, `column_exists`, `index_exists`, `fk_exists`), in both `upgrade()` and `downgrade()`. For migrations that convert a column type, also short-circuit if the column is already the target type.
+
 ### Configuration
 
 Configuration uses Hydra with YAML files in `.hydra_config/`:

openrag/scripts/migrations/alembic/env.py

@@ -1,5 +1,11 @@
+import os
+import sys
 from logging.config import fileConfig
 
+# Make modules alongside env.py (e.g. schema_helpers) importable from
+# migration scripts regardless of the cwd alembic is invoked from.
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
 from alembic import context
 from components.indexer.vectordb.models import Base
 from config import load_config

openrag/scripts/migrations/alembic/schema_helpers.py

@@ -0,0 +1,47 @@
+"""Shared inspection helpers for idempotent Alembic migrations.
+
+Needed because `Base.metadata.create_all()` at app startup may create the
+current-model schema directly on fresh (or older) deployments — so migrations
+must tolerate objects already existing.
+"""
+
+from alembic import op
+from sqlalchemy import inspect
+
+
+def table_exists(table: str) -> bool:
+    return table in inspect(op.get_bind()).get_table_names()
+
+
+def column_exists(table: str, column: str) -> bool:
+    if not table_exists(table):
+        return False
+    return any(c["name"] == column for c in inspect(op.get_bind()).get_columns(table))
+
+
+def index_exists(table: str, index: str) -> bool:
+    if not table_exists(table):
+        return False
+    return any(i["name"] == index for i in inspect(op.get_bind()).get_indexes(table))
+
+
+def fk_exists(table: str, fk_name: str) -> bool:
+    if not table_exists(table):
+        return False
+    return any(fk["name"] == fk_name for fk in inspect(op.get_bind()).get_foreign_keys(table))
+
+
+def unique_constraint_exists(table: str, constraint_name: str) -> bool:
+    if not table_exists(table):
+        return False
+    return any(uc["name"] == constraint_name for uc in inspect(op.get_bind()).get_unique_constraints(table))
+
+
+def column_type_is(table: str, column: str, sa_type: type) -> bool:
+    """Return True if `table.column` exists and its type is an instance of `sa_type`."""
+    if not table_exists(table):
+        return False
+    for col in inspect(op.get_bind()).get_columns(table):
+        if col["name"] == column:
+            return isinstance(col["type"], sa_type)
+    return False

openrag/scripts/migrations/alembic/versions/4add4d260575_initial_migration.py

@@ -10,23 +10,7 @@
 
 import sqlalchemy as sa
 from alembic import op
-from sqlalchemy import inspect
-
-
-def table_exists(table_name: str) -> bool:
-    """Check if a table exists in the database."""
-    bind = op.get_bind()
-    inspector = inspect(bind)
-    return table_name in inspector.get_table_names()
-
-
-def index_exists(index_name: str, table_name: str) -> bool:
-    """Check if an index exists on a table."""
-    bind = op.get_bind()
-    inspector = inspect(bind)
-    indexes = inspector.get_indexes(table_name)
-    return any(idx["name"] == index_name for idx in indexes)
-
+from schema_helpers import index_exists, table_exists
 
 # revision identifiers, used by Alembic.
 revision: str = "4add4d260575"
@@ -49,14 +33,14 @@ def upgrade() -> None:
 
     # Create indexes for partitions table if they don't exist
     if table_exists("partitions"):
-        if not index_exists("ix_partitions_created_at", "partitions"):
+        if not index_exists("partitions", "ix_partitions_created_at"):
             op.create_index(
                 op.f("ix_partitions_created_at"),
                 "partitions",
                 ["created_at"],
                 unique=False,
             )
-        if not index_exists("ix_partitions_partition", "partitions"):
+        if not index_exists("partitions", "ix_partitions_partition"):
             op.create_index(
                 op.f("ix_partitions_partition"),
                 "partitions",
@@ -82,16 +66,16 @@ def upgrade() -> None:
 
     # Create indexes for files table if they don't exist
     if table_exists("files"):
-        if not index_exists("ix_files_file_id", "files"):
+        if not index_exists("files", "ix_files_file_id"):
             op.create_index(op.f("ix_files_file_id"), "files", ["file_id"], unique=False)
-        if not index_exists("ix_files_partition_name", "files"):
+        if not index_exists("files", "ix_files_partition_name"):
             op.create_index(
                 op.f("ix_files_partition_name"),
                 "files",
                 ["partition_name"],
                 unique=False,
             )
-        if not index_exists("ix_partition_file", "files"):
+        if not index_exists("files", "ix_partition_file"):
             op.create_index(
                 "ix_partition_file",
                 "files",
@@ -104,17 +88,17 @@ def downgrade() -> None:
     """Downgrade schema."""
     # Drop indexes and tables if they exist
     if table_exists("files"):
-        if index_exists("ix_partition_file", "files"):
+        if index_exists("files", "ix_partition_file"):
             op.drop_index("ix_partition_file", table_name="files")
-        if index_exists("ix_files_partition_name", "files"):
+        if index_exists("files", "ix_files_partition_name"):
             op.drop_index(op.f("ix_files_partition_name"), table_name="files")
-        if index_exists("ix_files_file_id", "files"):
+        if index_exists("files", "ix_files_file_id"):
             op.drop_index(op.f("ix_files_file_id"), table_name="files")
         op.drop_table("files")
 
     if table_exists("partitions"):
-        if index_exists("ix_partitions_partition", "partitions"):
+        if index_exists("partitions", "ix_partitions_partition"):
             op.drop_index(op.f("ix_partitions_partition"), table_name="partitions")
-        if index_exists("ix_partitions_created_at", "partitions"):
+        if index_exists("partitions", "ix_partitions_created_at"):
             op.drop_index(op.f("ix_partitions_created_at"), table_name="partitions")
         op.drop_table("partitions")

openrag/scripts/migrations/alembic/versions/a1b2c3d4e5f6_add_document_relationships.py

@@ -15,24 +15,7 @@
 
 import sqlalchemy as sa
 from alembic import op
-from sqlalchemy import inspect
-
-
-def column_exists(table_name: str, column_name: str) -> bool:
-    """Check if a column exists in a table."""
-    bind = op.get_bind()
-    inspector = inspect(bind)
-    columns = [col["name"] for col in inspector.get_columns(table_name)]
-    return column_name in columns
-
-
-def index_exists(index_name: str, table_name: str) -> bool:
-    """Check if an index exists on a table."""
-    bind = op.get_bind()
-    inspector = inspect(bind)
-    indexes = inspector.get_indexes(table_name)
-    return any(idx["name"] == index_name for idx in indexes)
-
+from schema_helpers import column_exists, index_exists
 
 # revision identifiers, used by Alembic.
 revision: str = "a1b2c3d4e5f6"
@@ -59,15 +42,15 @@ def upgrade() -> None:
         )
 
     # Create single-column indexes
-    if not index_exists("ix_files_relationship_id", "files"):
+    if not index_exists("files", "ix_files_relationship_id"):
         op.create_index(
             "ix_files_relationship_id",
             "files",
             ["relationship_id"],
             unique=False,
         )
 
-    if not index_exists("ix_files_parent_id", "files"):
+    if not index_exists("files", "ix_files_parent_id"):
         op.create_index(
             "ix_files_parent_id",
             "files",
@@ -76,15 +59,15 @@ def upgrade() -> None:
         )
 
     # Create composite indexes for common query patterns
-    if not index_exists("ix_relationship_partition", "files"):
+    if not index_exists("files", "ix_relationship_partition"):
         op.create_index(
             "ix_relationship_partition",
             "files",
             ["relationship_id", "partition_name"],
             unique=False,
         )
 
-    if not index_exists("ix_parent_partition", "files"):
+    if not index_exists("files", "ix_parent_partition"):
         op.create_index(
             "ix_parent_partition",
             "files",
@@ -97,17 +80,17 @@ def downgrade() -> None:
     """Remove relationship_id and parent_id columns from files table."""
 
     # Drop composite indexes
-    if index_exists("ix_parent_partition", "files"):
+    if index_exists("files", "ix_parent_partition"):
         op.drop_index("ix_parent_partition", table_name="files")
 
-    if index_exists("ix_relationship_partition", "files"):
+    if index_exists("files", "ix_relationship_partition"):
         op.drop_index("ix_relationship_partition", table_name="files")
 
     # Drop single-column indexes
-    if index_exists("ix_files_parent_id", "files"):
+    if index_exists("files", "ix_files_parent_id"):
         op.drop_index("ix_files_parent_id", table_name="files")
 
-    if index_exists("ix_files_relationship_id", "files"):
+    if index_exists("files", "ix_files_relationship_id"):
         op.drop_index("ix_files_relationship_id", table_name="files")
 
     # Drop columns

openrag/scripts/migrations/alembic/versions/c224d4befe71_add_file_count_and_file_quota.py

@@ -10,6 +10,7 @@
 
 import sqlalchemy as sa
 from alembic import op
+from schema_helpers import column_exists, fk_exists, index_exists
 
 # revision identifiers, used by Alembic.
 revision: str = "c224d4befe71"
@@ -19,18 +20,32 @@
 
 
 def upgrade() -> None:
-    """Upgrade schema."""
-    op.add_column("users", sa.Column("file_quota", sa.Integer(), nullable=True))
-    op.add_column("users", sa.Column("file_count", sa.Integer(), nullable=False, server_default="0"))
-    op.add_column("files", sa.Column("created_by", sa.Integer(), nullable=True))
-    op.create_foreign_key("fk_files_created_by", "files", "users", ["created_by"], ["id"], ondelete="SET NULL")
-    op.create_index("ix_files_created_by", "files", ["created_by"])
+    """Upgrade schema.
+
+    Idempotent: Base.metadata.create_all() at app startup may have already
+    added these columns/indexes from the SQLAlchemy models on older deployments.
+    """
+    if not column_exists("users", "file_quota"):
+        op.add_column("users", sa.Column("file_quota", sa.Integer(), nullable=True))
+    if not column_exists("users", "file_count"):
+        op.add_column("users", sa.Column("file_count", sa.Integer(), nullable=False, server_default="0"))
+    if not column_exists("files", "created_by"):
+        op.add_column("files", sa.Column("created_by", sa.Integer(), nullable=True))
+    if not fk_exists("files", "fk_files_created_by"):
+        op.create_foreign_key("fk_files_created_by", "files", "users", ["created_by"], ["id"], ondelete="SET NULL")
+    if not index_exists("files", "ix_files_created_by"):
+        op.create_index("ix_files_created_by", "files", ["created_by"])
 
 
 def downgrade() -> None:
     """Downgrade schema."""
-    op.drop_index("ix_files_created_by", table_name="files")
-    op.drop_constraint("fk_files_created_by", "files", type_="foreignkey")
-    op.drop_column("files", "created_by")
-    op.drop_column("users", "file_count")
-    op.drop_column("users", "file_quota")
+    if index_exists("files", "ix_files_created_by"):
+        op.drop_index("ix_files_created_by", table_name="files")
+    if fk_exists("files", "fk_files_created_by"):
+        op.drop_constraint("fk_files_created_by", "files", type_="foreignkey")
+    if column_exists("files", "created_by"):
+        op.drop_column("files", "created_by")
+    if column_exists("users", "file_count"):
+        op.drop_column("users", "file_count")
+    if column_exists("users", "file_quota"):
+        op.drop_column("users", "file_quota")

openrag/scripts/migrations/alembic/versions/cd642e4502d8_create_users_memberships_tables.py

@@ -10,23 +10,7 @@
 
 import sqlalchemy as sa
 from alembic import op
-from sqlalchemy import inspect
-
-
-def table_exists(table_name: str) -> bool:
-    """Check if a table exists in the database."""
-    bind = op.get_bind()
-    inspector = inspect(bind)
-    return table_name in inspector.get_table_names()
-
-
-def index_exists(index_name: str, table_name: str) -> bool:
-    """Check if an index exists on a table."""
-    bind = op.get_bind()
-    inspector = inspect(bind)
-    indexes = inspector.get_indexes(table_name)
-    return any(idx["name"] == index_name for idx in indexes)
-
+from schema_helpers import index_exists, table_exists
 
 # revision identifiers, used by Alembic.
 revision: str = "cd642e4502d8"
@@ -52,14 +36,14 @@ def upgrade() -> None:
 
     # Create indexes for users table if they don't exist
     if table_exists("users"):
-        if not index_exists("ix_users_external_user_id", "users"):
+        if not index_exists("users", "ix_users_external_user_id"):
             op.create_index(
                 op.f("ix_users_external_user_id"),
                 "users",
                 ["external_user_id"],
                 unique=True,
             )
-        if not index_exists("ix_users_token", "users"):
+        if not index_exists("users", "ix_users_token"):
             op.create_index(op.f("ix_users_token"), "users", ["token"], unique=True)
 
     # Create partition_memberships table if it doesn't exist
@@ -80,21 +64,21 @@ def upgrade() -> None:
 
     # Create indexes for partition_memberships table if they don't exist
     if table_exists("partition_memberships"):
-        if not index_exists("ix_partition_memberships_partition_name", "partition_memberships"):
+        if not index_exists("partition_memberships", "ix_partition_memberships_partition_name"):
             op.create_index(
                 op.f("ix_partition_memberships_partition_name"),
                 "partition_memberships",
                 ["partition_name"],
                 unique=False,
             )
-        if not index_exists("ix_partition_memberships_user_id", "partition_memberships"):
+        if not index_exists("partition_memberships", "ix_partition_memberships_user_id"):
             op.create_index(
                 op.f("ix_partition_memberships_user_id"),
                 "partition_memberships",
                 ["user_id"],
                 unique=False,
             )
-        if not index_exists("ix_user_partition", "partition_memberships"):
+        if not index_exists("partition_memberships", "ix_user_partition"):
             op.create_index(
                 "ix_user_partition",
                 "partition_memberships",
@@ -107,23 +91,23 @@ def downgrade() -> None:
     """Downgrade schema."""
     # Drop indexes and tables if they exist
     if table_exists("partition_memberships"):
-        if index_exists("ix_user_partition", "partition_memberships"):
+        if index_exists("partition_memberships", "ix_user_partition"):
             op.drop_index("ix_user_partition", table_name="partition_memberships")
-        if index_exists("ix_partition_memberships_user_id", "partition_memberships"):
+        if index_exists("partition_memberships", "ix_partition_memberships_user_id"):
             op.drop_index(
                 op.f("ix_partition_memberships_user_id"),
                 table_name="partition_memberships",
             )
-        if index_exists("ix_partition_memberships_partition_name", "partition_memberships"):
+        if index_exists("partition_memberships", "ix_partition_memberships_partition_name"):
             op.drop_index(
                 op.f("ix_partition_memberships_partition_name"),
                 table_name="partition_memberships",
             )
         op.drop_table("partition_memberships")
 
     if table_exists("users"):
-        if index_exists("ix_users_token", "users"):
+        if index_exists("users", "ix_users_token"):
             op.drop_index(op.f("ix_users_token"), table_name="users")
-        if index_exists("ix_users_external_user_id", "users"):
+        if index_exists("users", "ix_users_external_user_id"):
             op.drop_index(op.f("ix_users_external_user_id"), table_name="users")
         op.drop_table("users")