From 7372cdaf6b646fa3970ef9e6948d7b0035dee029 Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Tue, 28 Jul 2026 14:39:20 -0700 Subject: [PATCH] Perf: move modified out of the vec0 metadata column into document_meta vec0 only inlines TEXT metadata up to 12 bytes (sqlite-vec.c); modified is an ISO timestamp (e.g. "2026-07-28T12:34:56.789012+00:00", 32 chars), always over that threshold, so every read of it recompiled and stepped a fresh SQL statement per row -- uncached, on every full scan (get_modified_times(), every compact(), every structural migration). It was also never filtered on inside a KNN query, so it never needed to be a vec0 column in the first place. Add document_meta(document_id INTEGER PRIMARY KEY, modified TEXT), one row per document rather than per chunk (all of a document's chunks share the same modified value). get_modified_times() drops its per-chunk dedup loop entirely as a result -- one row read per document, not per chunk. No separate index needed here (unlike document_chunks): document_id being the INTEGER PRIMARY KEY makes it the table's own rowid, and every access pattern is already keyed by it directly. Schema bump to v3 (m0002), backfilling document_meta from the old vec0 column for existing stores. Along the way: - _Row/_Vec0Params (NamedTuple) replace bare positional tuples for node rows and vec0 bind parameters -- still plain tuples as far as sqlite3.executemany() is concerned, but self-documenting. - Both new document_meta copy paths (compact()/migration backfill) are batched via fetchmany(), matching the existing discipline for the chunk-scoped copy -- small today (one row per document), but the same bounded-memory principle applies regardless of size. - Discovered along the way: m0001 delegated to _rebuild_into(), which always reflects the *current* schema. That was fine while v2 was current, but silently wrong the moment v3 existed (m0001 would produce a v3-shaped table instead of its actual v2 shape) -- fixed upstream on perf/13314-vecstore-point-delete, since it's a defect in that branch's own code independent of this change. Second item from VECTOR_STORE_PERF_BACKLOG.md. Co-Authored-By: Claude Sonnet 5 --- .../m0002_move_modified_to_document_meta.py | 63 +++++ src/paperless_ai/tests/test_vector_store.py | 239 ++++++++++++++++-- src/paperless_ai/vector_store.py | 169 +++++++++---- 3 files changed, 397 insertions(+), 74 deletions(-) create mode 100644 src/paperless_ai/migrations/m0002_move_modified_to_document_meta.py diff --git a/src/paperless_ai/migrations/m0002_move_modified_to_document_meta.py b/src/paperless_ai/migrations/m0002_move_modified_to_document_meta.py new file mode 100644 index 000000000..b6567cca4 --- /dev/null +++ b/src/paperless_ai/migrations/m0002_move_modified_to_document_meta.py @@ -0,0 +1,63 @@ +import sqlite3 + +from paperless_ai.migrations import MIGRATIONS +from paperless_ai.migrations import Migration +from paperless_ai.vector_store import _UPSERT_DOCUMENT_META +from paperless_ai.vector_store import COMPACT_BATCH_SIZE +from paperless_ai.vector_store import DEFAULT_TABLE_NAME +from paperless_ai.vector_store import PaperlessSqliteVecVectorStore + + +def _migrate_v2_to_v3_move_modified_to_document_meta( + src_conn: sqlite3.Connection, + dst_conn: sqlite3.Connection, + dim: int, +) -> None: + """v2 -> v3: move `modified` out of the vec0 metadata column into the + document_meta side table. + + vec0 only inlines TEXT metadata up to 12 bytes (sqlite-vec.c); `modified` + is an ISO timestamp (e.g. "2026-07-28T12:34:56.789012+00:00", 32 chars), + always over that, so every read of it took vec0's uncached long-value + path -- a fresh SQL prepare/step/finalize per row, on every full scan + (every compact(), every structural migration, every + get_modified_times() call). It was also never filtered on inside a KNN + query, so it never needed to be a vec0 column at all. + + document_meta is one row per document (all of a document's chunks share + the same modified value), not per chunk, so get_modified_times() also + loses its old per-chunk dedup loop in the process. + + dst_conn's document_meta table already exists, empty (created by + _open_connection() the same as document_chunks), but so does + src_conn's -- this migration is what starts populating it, so + _rebuild_into()'s own document_meta copy (which assumes the source + already has it right, true for compact() but not for this migration) + would find nothing there. Backfill it explicitly here first, from the + old vec0 table's modified column instead, before the generic rebuild. + """ + old_cursor = src_conn.execute( + "SELECT DISTINCT document_id, modified FROM " + DEFAULT_TABLE_NAME, + ) + while old_batch := old_cursor.fetchmany(COMPACT_BATCH_SIZE): + dst_conn.executemany( + _UPSERT_DOCUMENT_META, + [(r["document_id"], r["modified"]) for r in old_batch], + ) + PaperlessSqliteVecVectorStore._rebuild_into( + src_conn, + dst_conn, + dim, + meta_keys=("dim", "embed_model"), + ) + + +MIGRATIONS.append( + Migration( + from_version=2, + to_version=3, + kind="structural", + description="move modified out of the vec0 table into document_meta", + apply=_migrate_v2_to_v3_move_modified_to_document_meta, + ), +) diff --git a/src/paperless_ai/tests/test_vector_store.py b/src/paperless_ai/tests/test_vector_store.py index 04619433a..62baa5beb 100644 --- a/src/paperless_ai/tests/test_vector_store.py +++ b/src/paperless_ai/tests/test_vector_store.py @@ -106,6 +106,14 @@ def _chunk_index_rows( return [(r["chunk_id"], str(r["document_id"])) for r in rows] +def _document_meta_rows(store: PaperlessSqliteVecVectorStore) -> list[tuple[str, str]]: + """The document_meta side table, as (document_id, modified) pairs.""" + rows = store.client.execute( + "SELECT document_id, modified FROM document_meta ORDER BY document_id", + ).fetchall() + return [(str(r["document_id"]), r["modified"]) for r in rows] + + @contextmanager def _pending_migrations(*migrations: Migration, schema_version: int) -> Iterator[None]: """Register ``migrations`` and raise the module's SCHEMA_VERSION for the @@ -127,11 +135,14 @@ def _copying_apply(src: sqlite3.Connection, dst: sqlite3.Connection, dim: int) - Deliberately spells its SQL out rather than reusing the production helpers, so these tests exercise the migration machinery against a - realistic third-party apply() instead of co-drifting with it. + realistic third-party apply() instead of co-drifting with it. Matches + the *current* schema (no ``modified`` column -- see document_meta), + since this simulates a hypothetical migration starting from + SCHEMA_VERSION forward, not the real historical v1/v2/v3 shapes. """ dst.execute( # nosemgrep f"CREATE VIRTUAL TABLE {DEFAULT_TABLE_NAME} USING vec0(" - "id TEXT PRIMARY KEY, document_id TEXT, modified TEXT," + "id TEXT PRIMARY KEY, document_id TEXT," f" +node_content TEXT, embedding float[{dim}] distance_metric=cosine" ")", ) @@ -141,19 +152,17 @@ def _copying_apply(src: sqlite3.Connection, dst: sqlite3.Connection, dim: int) - (str(dim),), ) rows = src.execute( - "SELECT id, document_id, modified, node_content, embedding " - f"FROM {DEFAULT_TABLE_NAME}", + f"SELECT id, document_id, node_content, embedding FROM {DEFAULT_TABLE_NAME}", ).fetchall() dst.execute("BEGIN IMMEDIATE") dst.executemany( f"INSERT INTO {DEFAULT_TABLE_NAME} " - "(id, document_id, modified, node_content, embedding) " - "VALUES (?, ?, ?, ?, ?)", + "(id, document_id, node_content, embedding) " + "VALUES (?, ?, ?, ?)", [ ( r["id"], r["document_id"], - r["modified"], r["node_content"], bytes(r["embedding"]), ) @@ -168,6 +177,50 @@ def _copying_apply(src: sqlite3.Connection, dst: sqlite3.Connection, dim: int) - dst.execute("COMMIT") +def _seed_v1_index(store: PaperlessSqliteVecVectorStore, nodes: list[TextNode]) -> None: + """Populate ``store.client`` with a genuine v1-shaped index: a vec0 + table with its own ``modified`` column and no document_chunks rows -- + what m0001 actually migrates from. ``store.add()`` cannot be used for + this since it always creates the *current* (v3+) shape, which has no + ``modified`` column on the vec0 table at all. + """ + conn = store.client + conn.execute( # nosemgrep + "CREATE VIRTUAL TABLE " + DEFAULT_TABLE_NAME + " USING vec0(" + "id TEXT PRIMARY KEY, document_id TEXT, modified TEXT," + f" +node_content TEXT, embedding float[{DIM}] distance_metric=cosine)", + ) + for key, value in (("dim", str(DIM)), ("schema_version", "1")): + conn.execute( + "INSERT INTO index_meta (key, value) VALUES (?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (key, value), + ) + conn.executemany( + "INSERT INTO " + + DEFAULT_TABLE_NAME + + " (id, document_id, modified, node_content, embedding) " + "VALUES (?, ?, ?, ?, ?)", + [store._row(node) for node in nodes], + ) + + +def _seed_v2_index(store: PaperlessSqliteVecVectorStore, nodes: list[TextNode]) -> None: + """Populate ``store.client`` with a genuine v2-shaped index: a vec0 + table with its own ``modified`` column, document_chunks populated, and + no document_meta rows -- what m0002 actually migrates from. + """ + _seed_v1_index(store, nodes) + store.client.execute( + "INSERT INTO index_meta (key, value) VALUES ('schema_version', '2') " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + ) + store.client.executemany( + "INSERT INTO document_chunks (chunk_id, document_id) VALUES (?, ?)", + [(r.chunk_id, r.document_id) for r in (store._row(node) for node in nodes)], + ) + + class TestCrud: def test_add_then_query_returns_node( self, @@ -657,6 +710,109 @@ class TestDocumentChunksIndex: assert _chunk_index_rows(store) == [("b1", "2")] +class TestDocumentMetaIndex: + """document_meta holds one row per document (not per chunk, unlike + document_chunks), letting get_modified_times() read a document's + modified time without touching vec0's own (much slower to read -- + see m0002) metadata columns.""" + + def test_add_populates_document_meta( + self, + store: PaperlessSqliteVecVectorStore, + ) -> None: + """ + GIVEN: + - An empty store + WHEN: + - add() writes two chunks for document 1 (same modified time) + and one chunk for document 2 + THEN: + - document_meta has exactly one row per document, not per chunk + """ + store.add( + [ + make_node("a1", "1", modified="2026-01-01T00:00:00"), + make_node("a2", "1", modified="2026-01-01T00:00:00"), + make_node("b1", "2", modified="2026-02-02T00:00:00"), + ], + ) + assert _document_meta_rows(store) == [ + ("1", "2026-01-01T00:00:00"), + ("2", "2026-02-02T00:00:00"), + ] + + def test_delete_clears_document_meta( + self, + store: PaperlessSqliteVecVectorStore, + ) -> None: + """ + GIVEN: + - A store with chunks for documents 1 and 2 + WHEN: + - delete() removes document 1 + THEN: + - document_meta no longer has a row for document 1, but retains + document 2's row + """ + store.add([make_node("a1", "1"), make_node("b1", "2")]) + store.delete("1") + assert _document_meta_rows(store) == [("2", "2026-06-10T00:00:00")] + + def test_upsert_replaces_document_meta( + self, + store: PaperlessSqliteVecVectorStore, + ) -> None: + """ + GIVEN: + - A store with document 1 at one modified time + WHEN: + - upsert_document() replaces document 1's chunk with a new one + at a later modified time + THEN: + - document_meta reflects the new modified time, not the old one + """ + store.add([make_node("a1", "1", modified="2026-01-01T00:00:00")]) + store.upsert_document( + "1", + [make_node("a2", "1", modified="2026-03-03T00:00:00")], + ) + assert _document_meta_rows(store) == [("1", "2026-03-03T00:00:00")] + + def test_upsert_empty_nodes_clears_document_meta( + self, + store: PaperlessSqliteVecVectorStore, + ) -> None: + """ + GIVEN: + - A store with chunks for documents 1 and 2 + WHEN: + - upsert_document() is called for document 1 with an empty node + list + THEN: + - document_meta no longer has a row for document 1, but retains + document 2's row + """ + store.add([make_node("a1", "1"), make_node("b1", "2")]) + store.upsert_document("1", []) + assert _document_meta_rows(store) == [("2", "2026-06-10T00:00:00")] + + def test_drop_table_clears_document_meta( + self, + store: PaperlessSqliteVecVectorStore, + ) -> None: + """ + GIVEN: + - A store with one chunk indexed + WHEN: + - drop_table() is called + THEN: + - document_meta is emptied along with the vec0 table + """ + store.add([make_node("a1", "1")]) + store.drop_table() + assert _document_meta_rows(store) == [] + + class TestMetadataCoercion: def test_none_metadata_values_become_empty_strings( self, @@ -1233,9 +1389,9 @@ class TestMigrations: leave their vec0 rows orphaned forever. GIVEN: - - A store simulating a pre-migration (schema v1) index: rows - exist in the vec0 table, but document_chunks is empty and - schema_version is forced back to 1 + - A genuinely v1-shaped index (see _seed_v1_index): a vec0 table + with its own modified column, no document_chunks rows, and + schema_version 1 WHEN: - check_and_run_migrations() is called, then delete() removes one of the pre-migration documents, then @@ -1247,14 +1403,9 @@ class TestMigrations: instead of silently no-op'ing; the second call is a no-op that leaves schema_version unchanged """ - store.add([make_node("a1", "1"), make_node("a2", "1"), make_node("b1", "2")]) - # Simulate a pre-migration (schema v1) store: rows exist in the vec0 - # table, but document_chunks (added by the v1 -> v2 migration) has no - # entries for them, mirroring a real on-disk index created before - # this migration existed. - store.client.execute("DELETE FROM document_chunks") - store.client.execute( - "UPDATE index_meta SET value = '1' WHERE key = 'schema_version'", + _seed_v1_index( + store, + [make_node("a1", "1"), make_node("a2", "1"), make_node("b1", "2")], ) assert store.check_and_run_migrations() is False # structural, not re-embed @@ -1287,29 +1438,61 @@ class TestMigrations: stop detecting embedding-model config changes. GIVEN: - - A store configured with an embed_model_name, simulated as a - pre-migration (schema v1) index with document_chunks emptied - and schema_version forced back to 1 + - A genuinely v1-shaped index (see _seed_v1_index) whose + embed_model was already recorded before this migration existed WHEN: - check_and_run_migrations() runs the v1 -> v2 migration THEN: - schema_version advances to current and stored_model_name() still returns the original embed_model_name """ - with PaperlessSqliteVecVectorStore( - uri=str(tmp_path), - embed_model_name="model-a", - ) as store: - store.add([make_node("a1", "1")]) - store.client.execute("DELETE FROM document_chunks") + with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as store: + _seed_v1_index(store, [make_node("a1", "1")]) store.client.execute( - "UPDATE index_meta SET value = '1' WHERE key = 'schema_version'", + "INSERT INTO index_meta (key, value) VALUES ('embed_model', 'model-a') " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", ) assert store.check_and_run_migrations() is False assert self._schema_version(store) == SCHEMA_VERSION assert store.stored_model_name() == "model-a" + def test_v2_to_v3_migration_moves_modified_to_document_meta( + self, + store: PaperlessSqliteVecVectorStore, + ) -> None: + """The real v2 -> v3 migration (registered in MIGRATIONS) must + backfill document_meta from the old vec0 table's modified column, + deduped per document, or get_modified_times() would return nothing + for documents indexed before this migration existed. + + GIVEN: + - A genuinely v2-shaped index (see _seed_v2_index): vec0 table + with its own modified column, document_chunks populated, and + schema_version 2 + WHEN: + - check_and_run_migrations() runs the v2 -> v3 migration + THEN: + - schema_version advances to current, and document_meta (via + get_modified_times()) has exactly one deduped entry per + document, matching what the vec0 column used to hold + """ + _seed_v2_index( + store, + [ + make_node("a1", "1", modified="2026-01-01T00:00:00"), + make_node("a2", "1", modified="2026-01-01T00:00:00"), + make_node("b1", "2", modified="2026-02-02T00:00:00"), + ], + ) + + assert store.check_and_run_migrations() is False + assert self._schema_version(store) == SCHEMA_VERSION + assert store.get_modified_times() == { + "1": "2026-01-01T00:00:00", + "2": "2026-02-02T00:00:00", + } + def test_stop_at_reembed_boundary( self, store: PaperlessSqliteVecVectorStore, diff --git a/src/paperless_ai/vector_store.py b/src/paperless_ai/vector_store.py index 5a866488f..ac3c4c245 100644 --- a/src/paperless_ai/vector_store.py +++ b/src/paperless_ai/vector_store.py @@ -8,6 +8,7 @@ from contextlib import contextmanager from pathlib import Path from types import TracebackType from typing import Any +from typing import NamedTuple import sqlite_vec from llama_index.core.bridge.pydantic import PrivateAttr @@ -33,16 +34,23 @@ DEFAULT_TABLE_NAME = "documents" _INSERT = ( "INSERT INTO " + DEFAULT_TABLE_NAME - + " (id, document_id, modified, node_content, embedding) VALUES (?, ?, ?, ?, ?)" + + " (id, document_id, node_content, embedding) VALUES (?, ?, ?, ?)" ) _INSERT_CHUNK_INDEX = ( "INSERT INTO document_chunks (chunk_id, document_id) VALUES (?, ?)" ) +# document_meta is document-scoped (one row per document), not chunk-scoped +# like document_chunks -- see get_modified_times() and _row(). +_UPSERT_DOCUMENT_META = ( + "INSERT INTO document_meta (document_id, modified) VALUES (?, ?) " + "ON CONFLICT(document_id) DO UPDATE SET modified = excluded.modified" +) + # Current schema version. Bump when adding a migration -- see # paperless_ai/migrations/__init__.py for the full procedure. -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 3 # compact(): rebuild when the cumulative rowid count exceeds this multiple of # the live row count. DELETEs on vec0 tables never reclaim space (upstream @@ -57,8 +65,36 @@ COMPACT_BATCH_SIZE = 500 # Filterable vec0 metadata columns. _build_where() only ever receives filter # keys we construct ourselves, but allowlisting keeps SQL identifiers safe by -# construction. -_FILTER_COLUMNS = frozenset({"document_id", "modified"}) +# construction. "modified" is deliberately not here: it is never filtered on +# (see document_meta -- it isn't even a vec0 column anymore). +_FILTER_COLUMNS = frozenset({"document_id"}) + + +class _Row(NamedTuple): + """One node, ready to write. ``modified`` is not a vec0 column (see + document_meta / _open_connection) -- it rides along here because every + row-producing call site needs both the vec0 insert values and the + document_meta upsert value from the same node. + """ + + chunk_id: str + document_id: str + modified: str + node_content: str + embedding: bytes + + +class _Vec0Params(NamedTuple): + """``_INSERT``'s bind parameters, in placeholder order -- a NamedTuple + is still a plain positional tuple as far as sqlite3.executemany() is + concerned, so this keeps the self-documenting field names without + losing that. + """ + + chunk_id: str + document_id: str + node_content: str + embedding: bytes def _pack(embedding: Sequence[float]) -> bytes: @@ -71,28 +107,29 @@ def _unpack(blob: bytes) -> list[float]: def _copy_rows(src_conn: sqlite3.Connection, dst_conn: sqlite3.Connection) -> int: """Copy every live vec0 row from ``src_conn`` into ``dst_conn``, recording - each one in ``dst_conn``'s document_chunks side table. Returns the number - of rows copied. The caller owns ``dst_conn``'s transaction. + each one in ``dst_conn``'s document_chunks side table, then copy + document_meta the same way. Returns the number of vec0 rows copied. The + caller owns ``dst_conn``'s transaction. - Rows are streamed from the source cursor in batches instead of being + Both are streamed from their source cursors in batches instead of being materialized all at once, so a large index does not cause an OOM during a - routine compaction or migration. + routine compaction or migration. document_meta is one row per document, + not per chunk, so in practice it is far smaller than the vec0 table -- + but the same discipline applies regardless of size. """ src_cursor = src_conn.execute( - "SELECT id, document_id, modified, node_content, embedding FROM " - + DEFAULT_TABLE_NAME, + "SELECT id, document_id, node_content, embedding FROM " + DEFAULT_TABLE_NAME, ) copied = 0 while batch := src_cursor.fetchmany(COMPACT_BATCH_SIZE): dst_conn.executemany( _INSERT, [ - ( - r["id"], - r["document_id"], - r["modified"], - r["node_content"], - bytes(r["embedding"]), + _Vec0Params( + chunk_id=r["id"], + document_id=r["document_id"], + node_content=r["node_content"], + embedding=bytes(r["embedding"]), ) for r in batch ], @@ -102,6 +139,12 @@ def _copy_rows(src_conn: sqlite3.Connection, dst_conn: sqlite3.Connection) -> in [(r["id"], r["document_id"]) for r in batch], ) copied += len(batch) + meta_cursor = src_conn.execute("SELECT document_id, modified FROM document_meta") + while meta_batch := meta_cursor.fetchmany(COMPACT_BATCH_SIZE): + dst_conn.executemany( + _UPSERT_DOCUMENT_META, + [(r["document_id"], r["modified"]) for r in meta_batch], + ) return copied @@ -150,10 +193,12 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): Stores one row per node: the node id (TEXT primary key), its document id (metadata column, used for EQ/IN filtering and per-document delete), the - document's modified timestamp, the embedding (float32, cosine metric), and - the serialized node (text + metadata) as JSON in an auxiliary column. - ``stores_text`` lets llama-index run off this store alone, with no - separate docstore or index store. + embedding (float32, cosine metric), and the serialized node (text + + metadata) as JSON in an auxiliary column. Each document's modified + timestamp lives in the separate document_meta table instead (see + get_modified_times()), not as a vec0 column. ``stores_text`` lets + llama-index run off this store alone, with no separate docstore or + index store. Everything lives in one SQLite database file (``DB_FILENAME``) inside the directory given as ``uri`` (kept as a directory for compatibility with the @@ -219,6 +264,18 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): "CREATE INDEX IF NOT EXISTS idx_document_chunks_document_id " "ON document_chunks (document_id)", ) + # modified used to be a vec0 metadata column, but vec0 only inlines + # TEXT metadata up to 12 bytes -- an ISO timestamp is always longer, + # so every read recompiled and stepped a fresh SQL statement per row + # (sqlite-vec's vec0Column long-value path). It was never filtered + # on inside a KNN query either, so it never needed to be a vec0 + # column at all. One row per document here (not per chunk, like + # document_chunks), since every chunk of a document shares the same + # modified value -- see get_modified_times(). + conn.execute( + "CREATE TABLE IF NOT EXISTS document_meta " + "(document_id INTEGER PRIMARY KEY, modified TEXT NOT NULL)", + ) return conn @property @@ -294,6 +351,7 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): self._conn.execute("DROP TABLE IF EXISTS " + DEFAULT_TABLE_NAME) self._conn.execute("DELETE FROM index_meta") self._conn.execute("DELETE FROM document_chunks") + self._conn.execute("DELETE FROM document_meta") def stored_model_name(self) -> str | None: """Return the embedding model name recorded at table creation, or None.""" @@ -324,7 +382,6 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): + " USING vec0(" + "id TEXT PRIMARY KEY," + " document_id TEXT," - + " modified TEXT," + " +node_content TEXT," + " embedding float[" + str(int(dim)) @@ -343,7 +400,7 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): if not self.table_exists(): self._create_table(dim) - def _row(self, node: BaseNode) -> tuple[str, str, str, str, bytes]: + def _row(self, node: BaseNode) -> _Row: meta = node_to_metadata_dict( node, remove_text=False, @@ -352,20 +409,37 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): # vec0 metadata columns reject NULL (asg017/sqlite-vec#141): coerce # every value to a string, with "" as the absent sentinel. document_id = node.ref_doc_id or node.metadata.get("document_id") - return ( - node.node_id, - str(document_id or ""), - str(node.metadata.get("modified") or ""), - json.dumps(meta), - _pack(node.get_embedding()), + return _Row( + chunk_id=node.node_id, + document_id=str(document_id or ""), + modified=str(node.metadata.get("modified") or ""), + node_content=json.dumps(meta), + embedding=_pack(node.get_embedding()), ) - def _index_chunks(self, rows: list[tuple[str, str, str, str, bytes]]) -> None: - """Record each row's (chunk_id, document_id) in the document_chunks - side table, kept in lockstep with every insert into the vec0 table.""" + @staticmethod + def _vec0_params(rows: list[_Row]) -> list[_Vec0Params]: + """``rows``, minus the ``modified`` field vec0 no longer stores.""" + return [ + _Vec0Params(r.chunk_id, r.document_id, r.node_content, r.embedding) + for r in rows + ] + + def _index_chunks(self, rows: list[_Row]) -> None: + """Record each row's (chunk_id, document_id) in document_chunks, and + each row's (document_id, modified) in document_meta -- deduped + within the batch, since every chunk of a document shares the same + modified value -- kept in lockstep with every insert into the vec0 + table. + """ self._conn.executemany( _INSERT_CHUNK_INDEX, - [(chunk_id, document_id) for chunk_id, document_id, *_ in rows], + [(r.chunk_id, r.document_id) for r in rows], + ) + modified_by_document = {r.document_id: r.modified for r in rows} + self._conn.executemany( + _UPSERT_DOCUMENT_META, + list(modified_by_document.items()), ) def _delete_chunks_by_document_id(self, document_id: str) -> None: @@ -391,6 +465,10 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): "DELETE FROM document_chunks WHERE document_id = ?", (doc_id,), ) + self._conn.execute( + "DELETE FROM document_meta WHERE document_id = ?", + (doc_id,), + ) def _increment_total_inserts(self, count: int) -> None: """Increment the cumulative insert counter stored in index_meta. @@ -409,7 +487,7 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): rows = [self._row(node) for node in nodes] with self._transaction(): self._ensure_table(len(nodes[0].get_embedding())) - self._conn.executemany(_INSERT, rows) + self._conn.executemany(_INSERT, self._vec0_params(rows)) self._index_chunks(rows) self._increment_total_inserts(len(rows)) return [node.node_id for node in nodes] @@ -429,7 +507,7 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): if self.table_exists(): self._delete_chunks_by_document_id(document_id) if rows: - self._conn.executemany(_INSERT, rows) + self._conn.executemany(_INSERT, self._vec0_params(rows)) self._index_chunks(rows) self._increment_total_inserts(len(rows)) return [node.node_id for node in nodes] @@ -507,19 +585,17 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): def get_modified_times(self) -> dict[str, str]: """Return {document_id: stored_modified_isoformat} for all indexed documents. - All chunks of a document share the same ``modified`` value, so the - first row seen per document is sufficient. + document_meta already has exactly one row per document (not per + chunk, unlike the vec0 table), so no dedup is needed here. """ if not self.table_exists(): return {} - result: dict[str, str] = {} - for row in self._conn.execute( - "SELECT document_id, modified FROM " + DEFAULT_TABLE_NAME, - ): - doc_id = str(row["document_id"]) - if doc_id not in result: - result[doc_id] = str(row["modified"] or "") - return result + return { + str(row["document_id"]): str(row["modified"] or "") + for row in self._conn.execute( + "SELECT document_id, modified FROM document_meta", + ) + } @property def _db_path(self) -> str: @@ -705,7 +781,8 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore): self._meta_set_on(new_conn, "schema_version", str(migration.to_version)) -# Registers m0001 into MIGRATIONS; must be at the bottom (needs +# Registers m0001/m0002 into MIGRATIONS; must be at the bottom (needs # PaperlessSqliteVecVectorStore fully defined) -- see # paperless_ai/migrations/__init__.py for the full procedure. -from paperless_ai.migrations import m0001_add_document_chunks # noqa: E402, F401 +from paperless_ai.migrations import m0001_add_document_chunks # noqa +from paperless_ai.migrations import m0002_move_modified_to_document_meta # noqa