fix: initialize and migrate pgvector tables transactionally (#54)

REQUIREMENTS.md

@@ -2,7 +2,7 @@
 
 > **Note:** This document is automatically generated and verified against the live test suite by `scripts/generate_requirements.py` and `tests/backend/test_requirements_sync.py`.
 
-**Test Verification Baseline:** **1115 Automated Tests** (757 Pytest Backend + 308 Vitest Frontend + 50 Playwright E2E).
+**Test Verification Baseline:** **1120 Automated Tests** (762 Pytest Backend + 308 Vitest Frontend + 50 Playwright E2E).
 
 ---
 
@@ -1155,12 +1155,19 @@ and leaves the prior indexed state intact without data loss._
 - `test_scanned_pdf_cache_invalidation_on_replace_and_delete`
 - `test_real_asd_ste100_pdf_preview_or_indexing`
 
+#### `tests/test_pgvector_native.py` (5 tests)
+- `test_native_initialization_and_collection_isolation`
+- `test_native_migrates_null_legacy_collection`
+- `test_native_initialization_error_is_explicit_and_atomic`
+- `test_native_dimension_mismatch_preserves_populated_table`
+- `test_native_dimension_change_after_explicit_clear`
+
 #### `tests/test_pgvector_store.py` (62 tests)
 - `TestPgVectorStoreImports::test_import_pgvector_store`
 - `TestPgVectorStoreImports::test_manager_supported_providers_includes_postgres_and_pgvector`
 - `TestPgVectorStoreLifecycle::test_initialization_defaults`
 - `TestPgVectorStoreLifecycle::test_ensure_collection_executes_ddl`
-- `TestPgVectorStoreLifecycle::test_ensure_collection_handles_extension_permission_warning`
+- `TestPgVectorStoreLifecycle::test_ensure_collection_extension_permissions`
 - `TestPgVectorStoreLifecycle::test_ensure_collection_failure`
 - `TestPgVectorStoreLifecycle::test_health_check_success`
 - `TestPgVectorStoreLifecycle::test_health_check_failure`
@@ -1186,12 +1193,12 @@ and leaves the prior indexed state intact without data loss._
 - `TestVectorStoreManagerPgVectorDispatch::test_test_connection_pgvector`
 - `TestPgVectorCollectionIsolation::test_collection_isolation_reads_writes_deletes` - _Test that collection_a writes/deletes do not affect collection_b in the same table._
 - `TestPgVectorCollectionIsolation::test_migration_and_pk_update` - _Test schema migration for legacy table without collection_name column._
-- `TestPgVectorCollectionIsolation::test_dimension_mismatch_warning` - _Test logging warning when table dimension differs from store dimension._
+- `TestPgVectorCollectionIsolation::test_dimension_mismatch_requires_explicit_rebuild` - _A populated shared table cannot be silently recreated._
 - `test_import_pgvector_store`
 - `test_manager_supported_providers_includes_postgres_and_pgvector`
 - `test_initialization_defaults`
 - `test_ensure_collection_executes_ddl`
-- `test_ensure_collection_handles_extension_permission_warning`
+- `test_ensure_collection_extension_permissions`
 - `test_ensure_collection_failure`
 - `test_health_check_success`
 - `test_health_check_failure`
@@ -1217,7 +1224,7 @@ and leaves the prior indexed state intact without data loss._
 - `test_test_connection_pgvector`
 - `test_collection_isolation_reads_writes_deletes` - _Test that collection_a writes/deletes do not affect collection_b in the same table._
 - `test_migration_and_pk_update` - _Test schema migration for legacy table without collection_name column._
-- `test_dimension_mismatch_warning` - _Test logging warning when table dimension differs from store dimension._
+- `test_dimension_mismatch_requires_explicit_rebuild` - _A populated shared table cannot be silently recreated._
 
 #### `tests/test_poller.py` (9 tests)
 - `test_check_all_auto_sync_repos_triggers_sync`

app/services/vector_store/pgvector_store.py

@@ -49,38 +49,37 @@ def __init__(
     def ensure_collection(self) -> bool:
         """
         Initializes vector extension, documents table schema, and HNSW indexes idempotently.
+
+        Schema changes are atomic. Populated tables require an explicit rebuild before
+        changing dimensions, because all collections share the embedding column.
         """
         try:
-            with self.engine.connect() as conn:
-                # 1. Enable pgvector extension
-                try:
+            with self.engine.connect() as conn, conn.begin():
+                # An installed extension needs no CREATE privilege.
+                extension = conn.execute(text(
+                    "SELECT 1 FROM pg_extension WHERE extname = 'vector'"
+                )).scalar()
+                if not extension:
                     conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector;"))
-                    conn.commit()
-                except Exception as ext_err:
-                    logger.warning(
-                        f"Could not automatically create 'vector' extension (may require superuser or already exist): {ext_err}"
-                    )
 
-                # Check existing table vector dimension
-                try:
-                    check_dim_sql = f"""
+                # to_regclass returns NULL for a fresh table instead of aborting the transaction.
+                check_dim_sql = text("""
                     SELECT atttypmod FROM pg_attribute
-                    WHERE attrelid = '{self.table_name}'::regclass AND attname = 'embedding';
-                    """
-                    row = conn.execute(text(check_dim_sql)).fetchone()
-                    if row and row[0] is not None and row[0] > 0 and row[0] != self.dimension:
-                        logger.warning(
-                            f"PgVectorStore table '{self.table_name}' dimension mismatch: expected {self.dimension}d, found {row[0]}d. Recreating..."
+                    WHERE attrelid = to_regclass(:table_name) AND attname = 'embedding';
+                """)
+                row = conn.execute(check_dim_sql, {"table_name": self.table_name}).fetchone()
+                if row and row[0] is not None and row[0] > 0 and row[0] != self.dimension:
+                    conn.execute(text(f"LOCK TABLE {self.table_name} IN ACCESS EXCLUSIVE MODE"))
+                    if conn.execute(text(f"SELECT EXISTS (SELECT 1 FROM {self.table_name})")).scalar():
+                        raise ValueError(
+                            f"Vector dimension mismatch for table '{self.table_name}': "
+                            f"existing {row[0]}d, configured {self.dimension}d. "
+                            "An explicit rebuild is required; all collections sharing this "
+                            "table must be cleared before changing dimensions."
                         )
-                        conn.execute(text(f"DROP TABLE IF EXISTS {self.table_name} CASCADE;"))
-                        conn.commit()
-                        try:
-                            from app.services.database import clear_indexing_metadata
-                            clear_indexing_metadata()
-                        except Exception as cm_err:
-                            logger.warning(f"Failed to clear indexing metadata during PgVector table recreation: {cm_err}")
-                except Exception:
-                    pass
+                    conn.execute(text(
+                        f"ALTER TABLE {self.table_name} ALTER COLUMN embedding TYPE vector({self.dimension})"
+                    ))
 
                 # 2. Create vector documents table
                 create_table_sql = f"""
@@ -111,63 +110,33 @@ def ensure_collection(self) -> bool:
                 """
                 conn.execute(text(create_table_sql))
 
-                # Migration for existing tables: add collection_name column if missing and adjust PK constraint
-                try:
-                    conn.execute(text(
-                        f"ALTER TABLE {self.table_name} ADD COLUMN IF NOT EXISTS collection_name VARCHAR(255) DEFAULT 'knowledge_rag_v1';"
-                    ))
-                    conn.commit()
-                except Exception as alter_err:
-                    logger.debug(f"Migration add collection_name column info: {alter_err}")
-
-                # Ensure null collection_name values in legacy rows are updated
-                try:
-                    conn.execute(text(
-                        f"UPDATE {self.table_name} SET collection_name = 'knowledge_rag_v1' WHERE collection_name IS NULL;"
-                    ))
-                    conn.commit()
-                except Exception as update_err:
-                    logger.debug(f"Migration update null collection_name info: {update_err}")
-
-                # Check and update PK constraint if it was on id alone
-                try:
-                    pk_check_sql = text("""
-                        SELECT a.attname
-                        FROM pg_index i
-                        JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
-                        WHERE i.indrelid = :table_name::regclass AND i.indisprimary;
-                    """)
-                    pk_cols = [row[0] for row in conn.execute(pk_check_sql, {"table_name": self.table_name}).fetchall()]
-                    if pk_cols and "collection_name" not in pk_cols:
-                        # Find constraint name
+                # Migrate legacy rows before replacing the old id-only primary key.
+                conn.execute(text(
+                    f"ALTER TABLE {self.table_name} ADD COLUMN IF NOT EXISTS collection_name VARCHAR(255) DEFAULT 'knowledge_rag_v1';"
+                ))
+                conn.execute(text(
+                    f"UPDATE {self.table_name} SET collection_name = 'knowledge_rag_v1' WHERE collection_name IS NULL;"
+                ))
+
+                pk_check_sql = text("""
+                    SELECT a.attname
+                    FROM pg_index i
+                    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
+                    WHERE i.indrelid = CAST(:table_name AS regclass) AND i.indisprimary;
+                """)
+                pk_cols = [row[0] for row in conn.execute(pk_check_sql, {"table_name": self.table_name}).fetchall()]
+                if set(pk_cols) != {"collection_name", "id"}:
+                    if pk_cols:
                         conname_sql = text("""
                             SELECT conname
                             FROM pg_constraint
-                            WHERE conrelid = :table_name::regclass AND contype = 'p';
+                            WHERE conrelid = CAST(:table_name AS regclass) AND contype = 'p';
                         """)
                         conname_row = conn.execute(conname_sql, {"table_name": self.table_name}).fetchone()
                         if conname_row:
-                            conname = conname_row[0]
+                            conname = self.engine.dialect.identifier_preparer.quote(conname_row[0])
                             conn.execute(text(f"ALTER TABLE {self.table_name} DROP CONSTRAINT {conname};"))
-                            conn.execute(text(f"ALTER TABLE {self.table_name} ADD PRIMARY KEY (collection_name, id);"))
-                            conn.commit()
-                except Exception as pk_err:
-                    logger.debug(f"Migration update primary key info: {pk_err}")
-
-                # Validate embedding dimension compatibility if table exists
-                try:
-                    dim_check_sql = text("""
-                        SELECT atttypmod
-                        FROM pg_attribute
-                        WHERE attrelid = :table_name::regclass AND attname = 'embedding';
-                    """)
-                    dim_row = conn.execute(dim_check_sql, {"table_name": self.table_name}).fetchone()
-                    if dim_row and dim_row[0] > 0 and dim_row[0] != self.dimension:
-                        logger.warning(
-                            f"Table '{self.table_name}' embedding dimension ({dim_row[0]}) does not match configured dimension ({self.dimension})."
-                        )
-                except Exception as dim_err:
-                    logger.debug(f"Embedding dimension validation info: {dim_err}")
+                    conn.execute(text(f"ALTER TABLE {self.table_name} ADD PRIMARY KEY (collection_name, id);"))
 
                 # 3. Create HNSW cosine index
                 hnsw_index_sql = f"""
@@ -182,12 +151,13 @@ def ensure_collection(self) -> bool:
                 conn.execute(text(f"CREATE INDEX IF NOT EXISTS idx_{self.table_name}_type ON {self.table_name}(doc_type);"))
                 conn.execute(text(f"CREATE INDEX IF NOT EXISTS idx_{self.table_name}_path ON {self.table_name}(path);"))
 
-                conn.commit()
             logger.info(f"PgVectorStore initialized table '{self.table_name}' with {self.dimension}d HNSW index.")
             return True
+        except ValueError:
+            raise
         except Exception as e:
             logger.error(f"Error initializing PgVectorStore table '{self.table_name}': {e}", exc_info=True)
-            return False
+            raise RuntimeError(f"Failed to initialize PgVectorStore table '{self.table_name}'.") from e
 
     def upsert_documents(
         self,

docs/guide/configuration.md

@@ -74,6 +74,30 @@ ChromaDB provides lightweight embedded vector storage without external services:
 export VECTOR_STORE_PROVIDER="chroma"
 ```
 
+### PostgreSQL / pgvector Mode
+
+Initialization creates or migrates the vector table and its indexes in one transaction.
+Legacy rows without a collection are assigned to `knowledge_rag_v1`; the primary key
+becomes `(collection_name, id)`, allowing the same document ID in separate collections.
+Initialization failures raise an explicit error and roll back schema changes rather
+than leaving a partially migrated store. If `vector` is already installed, startup
+does not attempt extension creation; otherwise the database role needs permission
+to install it (or an administrator must install it first).
+
+All collections in a table share one embedding dimension. Initialization never drops
+a populated table to resolve a dimension mismatch. Coordinate an explicit rebuild
+with every affected collection owner, clear their vectors and indexing metadata,
+then initialize the empty table with the new dimension and reindex. Changing the
+dimension of an empty table preserves its schema and does not clear unrelated
+relational metadata.
+
+Native regression tests require `TEST_POSTGRES_URL` pointing to a **disposable**
+PostgreSQL database with pgvector available. They create a unique schema per test
+and remove only that schema; do not use the application database. Run
+`python -m pytest tests/test_pgvector_native.py`. Without that variable, the native
+tests skip. Search tests use a deterministic loopback HTTP embedding service; no
+embedding model downloads are required.
+
 ---
 
 ## LiteLLM Proxy Integration