amorfo77/paperless-ngx
1
0
Fork 0
mirror of https://github.com/paperless-ngx/paperless-ngx.git synced 2026-07-29 23:34:56 +00:00
Code Issues Projects Releases Wiki Activity

Perf: move modified out of the vec0 metadata column into document_meta

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
stumpylog
2026-07-28 14:39:20 -07:00
co-authored by Claude Sonnet 5
parent 3838706194
commit 7372cdaf6b
3 changed files with 397 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/paperless_ai/migrations/m0002_move_modified_to_document_meta.py
+63
View File
@@ -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,
),
)
src/paperless_ai/tests/test_vector_store.py
+211 -28
View File
@@ -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,
src/paperless_ai/vector_store.py
+123 -46
View File
@@ -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