diff --git a/pstack-evidence.md b/pstack-evidence.md new file mode 100644 index 000000000..b16f0ead0 --- /dev/null +++ b/pstack-evidence.md @@ -0,0 +1,9 @@ +# pstack evidence — basic-memory #1850 + +## Principles applied +- Laziness Protocol: batch the existing OR-chain delete; no API change. +- Prove It Works: regression deletes 1200 pending chunks and asserts both tables empty. +- Make Operations Idempotent: batches preserve pending-only match semantics. + +## Repro → fix +Unbatched OR of 4-term ANDs exceeds SQLITE_MAX_EXPR_DEPTH (~995). `_DELETE_BATCH_SIZE = 200` splits UPDATE/DELETE work; one commit after all batches. diff --git a/src/basic_memory/repository/sqlite_vec_index.py b/src/basic_memory/repository/sqlite_vec_index.py index 976450159..7220c7476 100644 --- a/src/basic_memory/repository/sqlite_vec_index.py +++ b/src/basic_memory/repository/sqlite_vec_index.py @@ -252,45 +252,58 @@ async def upsert(self, project_id: int, records: Sequence[VectorRecord]) -> None ) await session.commit() + # SQLite caps expression-tree depth at SQLITE_MAX_EXPR_DEPTH (1000). Each + # pending-row predicate is a 4-term AND under an OR chain, so a single + # statement fails above ~995 deletions (#1850). Batch below that ceiling. + _DELETE_BATCH_SIZE = 200 + async def delete(self, project_id: int, records: Sequence[VectorDeletion]) -> None: if not records: return await self.initialize() async with db.scoped_session(self._session_maker) as session: await self._ensure_loaded(session) - params: dict[str, object] = {"project_id": project_id} - predicates: list[str] = [] - for index, record in enumerate(records): - params[f"entity_id_{index}"] = record.key.entity_id - params[f"chunk_key_{index}"] = record.key.chunk_key - params[f"source_hash_{index}"] = record.source_hash - predicates.append( - f"(entity_id = :entity_id_{index} " - f"AND chunk_key = :chunk_key_{index} " - f"AND source_hash = :source_hash_{index} " - "AND embedding_status = 'pending')" - ) - result = await session.execute( - text( - "UPDATE search_vector_chunks SET source_hash = source_hash " - "WHERE project_id = :project_id AND (" - + " OR ".join(predicates) - + ") RETURNING id" - ), - params, - ) - rowids = [int(row_id) for row_id in result.scalars().all()] - if rowids: - params = {f"rowid_{index}": rowid for index, rowid in enumerate(rowids)} - placeholders = ", ".join(f":rowid_{index}" for index in range(len(rowids))) - await session.execute( - text(f"DELETE FROM search_vector_embeddings WHERE rowid IN ({placeholders})"), - params, - ) - await session.execute( - text(f"DELETE FROM search_vector_chunks WHERE id IN ({placeholders})"), + all_rowids: list[int] = [] + for start in range(0, len(records), self._DELETE_BATCH_SIZE): + batch = records[start : start + self._DELETE_BATCH_SIZE] + params: dict[str, object] = {"project_id": project_id} + predicates: list[str] = [] + for index, record in enumerate(batch): + params[f"entity_id_{index}"] = record.key.entity_id + params[f"chunk_key_{index}"] = record.key.chunk_key + params[f"source_hash_{index}"] = record.source_hash + predicates.append( + f"(entity_id = :entity_id_{index} " + f"AND chunk_key = :chunk_key_{index} " + f"AND source_hash = :source_hash_{index} " + "AND embedding_status = 'pending')" + ) + result = await session.execute( + text( + "UPDATE search_vector_chunks SET source_hash = source_hash " + "WHERE project_id = :project_id AND (" + + " OR ".join(predicates) + + ") RETURNING id" + ), params, ) + all_rowids.extend(int(row_id) for row_id in result.scalars().all()) + if all_rowids: + # IN-lists are shallow; still batch to keep statement size bounded. + for start in range(0, len(all_rowids), self._DELETE_BATCH_SIZE): + batch_ids = all_rowids[start : start + self._DELETE_BATCH_SIZE] + params = {f"rowid_{index}": rowid for index, rowid in enumerate(batch_ids)} + placeholders = ", ".join(f":rowid_{index}" for index in range(len(batch_ids))) + await session.execute( + text( + f"DELETE FROM search_vector_embeddings WHERE rowid IN ({placeholders})" + ), + params, + ) + await session.execute( + text(f"DELETE FROM search_vector_chunks WHERE id IN ({placeholders})"), + params, + ) await session.commit() async def delete_entity(self, project_id: int, entity_id: int) -> None: diff --git a/tests/repository/test_sqlite_vector_search_repository.py b/tests/repository/test_sqlite_vector_search_repository.py index d86278848..082ee03f3 100644 --- a/tests/repository/test_sqlite_vector_search_repository.py +++ b/tests/repository/test_sqlite_vector_search_repository.py @@ -1087,6 +1087,78 @@ async def test_sqlite_vec_delete_requires_pending_source_generation(search_repos @pytest.mark.asyncio +async def test_sqlite_vec_delete_batches_above_expression_depth_limit(search_repository): + """Deleting more than ~995 pending chunks must not hit SQLITE_MAX_EXPR_DEPTH (#1850).""" + if not isinstance(search_repository, SQLiteSearchRepository): + pytest.skip("sqlite-vec deletion behavior is local SQLite-only.") + + _enable_semantic(search_repository) + await search_repository.init_search_index() + index = cast(SQLiteVecIndex, search_repository._semantic_vector_index) + project_id = search_repository.project_id + count = 1200 + deletions: list[VectorDeletion] = [] + + async with db.scoped_session(search_repository.session_maker) as session: + await index._ensure_loaded(session) + for i in range(count): + chunk_key = f"entity:1850:{i}" + row_id = 1850000 + i + await session.execute( + text( + "INSERT INTO search_vector_chunks (" + "id, entity_id, project_id, chunk_key, source_type, source_row_id, " + "chunk_index, chunk_text, source_hash, entity_fingerprint, " + "embedding_model, vector_index, embedding_status" + ") VALUES (" + ":id, 1850, :project_id, :chunk_key, 'entity', 1850, :chunk_index, " + "'text', 'hash-1850', 'fingerprint', :embedding_model, " + "'sqlite-vec', 'pending')" + ), + { + "id": row_id, + "project_id": project_id, + "chunk_key": chunk_key, + "chunk_index": i, + "embedding_model": search_repository._embedding_model_key(), + }, + ) + await session.execute( + text( + "INSERT INTO search_vector_embeddings (rowid, embedding, source_hash) " + "VALUES (:rowid, :embedding, 'hash-1850')" + ), + {"rowid": row_id, "embedding": "[1.0, 0.0, 0.0, 0.0]"}, + ) + deletions.append( + VectorDeletion( + key=VectorKey(entity_id=1850, chunk_key=chunk_key), + source_hash="hash-1850", + ) + ) + await session.commit() + + await index.delete(project_id, deletions) + + async with db.scoped_session(search_repository.session_maker) as session: + remaining_chunks = await session.scalar( + text( + "SELECT COUNT(*) FROM search_vector_chunks " + "WHERE project_id = :project_id AND entity_id = 1850" + ), + {"project_id": project_id}, + ) + remaining_embeddings = await session.scalar( + text( + "SELECT COUNT(*) FROM search_vector_embeddings " + "WHERE rowid BETWEEN :lo AND :hi" + ), + {"lo": 1850000, "hi": 1850000 + count - 1}, + ) + assert remaining_chunks == 0 + assert remaining_embeddings == 0 + + async def test_sqlite_chunk_upsert_and_delete_lifecycle(search_repository): """sync_entity_vectors updates changed chunks and clears vectors when source rows disappear.""" if not isinstance(search_repository, SQLiteSearchRepository):