Compare commits

...
11 Commits
Author SHA1 Message Date
stumpylog 9a4b1402c0 Perf: dedupe table_exists() lookups, atomic insert counter, fewer connections in update_llm_index() 2026-07-29 15:16:08 -07:00
stumpylog 394f81e820 Fix: strengthen two vacuous Task 4 regression tests
test_migration_never_delegates_to_current_schema_helpers never actually ran
the migration (missing check_and_run_migrations() call) and its source-text
assertion was tautological (the "or DROP TABLE in source" clause was always
true). Now runs the real migration and asserts spy call counts instead:
DocumentChunksTable.create/DocumentMetaTable.create are each called exactly
3 times (construction, rebuild temp file, post-swap reopen -- all via
_open_connection, never from inside apply()), and _create_vec_table is
never called from the migration path.

test_drop_table_clears_modified_times asserted via get_modified_times(),
which short-circuits on table_exists() -- checking only the vec0 table that
drop_table() drops first -- so the assertion held even if
DocumentMetaTable.delete_all() were never called. Now asserts directly
against document_meta and document_chunks row counts.
2026-07-29 14:55:10 -07:00
stumpylog 7ac1917f79 Feature: schema v2 -- document_chunks/document_meta side tables, document_id INTEGER, point-delete
Rewrites the sqlite-vec vector store's on-disk schema: document_id becomes
an INTEGER vec0 metadata column (was TEXT), modified moves out of vec0 into
a new document_meta side table, and a new document_chunks side table gives
O(1) per-document chunk lookup for delete/upsert instead of a full vec0
scan. compact() now streams document_chunks and document_meta across the
file-swap rebuild too (previously document_meta would have gone silently
empty after the first compaction). drop_table() clears both side tables.

Adds the single frozen m0001_v1_to_v2 migration, converting a real,
historically-shaped v1 store (the shape shipped since v3.0.0) into the v2
shape in one streaming pass, with its own hardcoded DDL rather than
delegating to any "current schema" helper.

SCHEMA_VERSION bumps 1 -> 2.
2026-07-29 14:38:23 -07:00
stumpylog 0464f3c901 Merge branch 'pr2-tables-gateway' into pr3-schema-rewrite 2026-07-29 13:58:57 -07:00
stumpylogandClaude Sonnet 5 7d9ba34582 Fix: address code review feedback on table gateways
- Document sqlite3.Row row_factory precondition in module docstring
- Strengthen test_create_is_idempotent to verify populated table survives
- Collapse three roundtrip tests into one @pytest.mark.parametrize

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-29 13:51:32 -07:00
stumpylogandClaude Sonnet 5 91c3d9caab Feature: add table-gateway module for document_chunks/document_meta/index_meta
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-29 13:42:39 -07:00
stumpylog 01f723bae2 Fix: skip incremental index writes when a re-embed migration is pending 2026-07-29 13:23:27 -07:00
stumpylog 49a584607b Docs: add migrate to document_llmindex choices and bare-metal upgrade steps 2026-07-29 12:33:24 -07:00
stumpylog ad5af61121 Fix: restore executable bit on init-llmindex-migrate run 2026-07-29 12:20:38 -07:00
stumpylog 5b7341fd3a Feature: run pending LLM index migrations automatically on startup 2026-07-29 12:19:06 -07:00
stumpylogandClaude Sonnet 5 4d9c0e8bd5 Refactor: extract migration infrastructure, add has_pending_migration()
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-29 11:58:43 -07:00
16 changed files with 1774 additions and 321 deletions
@@ -0,0 +1,12 @@
#!/command/with-contenv /usr/bin/bash
# shellcheck shell=bash
declare -r log_prefix="[init-llmindex-migrate]"
echo "${log_prefix} Checking for pending LLM index migrations..."
cd "${PAPERLESS_SRC_DIR}"
if [[ -n "${USER_IS_NON_ROOT}" ]]; then
python3 manage.py document_llmindex migrate
else
s6-setuidgid paperless python3 manage.py document_llmindex migrate
fi
@@ -0,0 +1 @@
oneshot
@@ -0,0 +1 @@
/etc/s6-overlay/s6-rc.d/init-llmindex-migrate/run
+11 -1
View File
@@ -212,6 +212,16 @@ following:
This is a no-op if the index is already up to date, so it is safe to This is a no-op if the index is already up to date, so it is safe to
run on every upgrade. run on every upgrade.
5. Migrate the LLM index if needed.
```shell-session
cd src
python3 manage.py document_llmindex migrate
```
This is a no-op if the index schema is already current, so it is safe
to run on every upgrade.
### Database Upgrades ### Database Upgrades
Paperless-ngx is compatible with Django-supported versions of PostgreSQL and MariaDB and it is generally Paperless-ngx is compatible with Django-supported versions of PostgreSQL and MariaDB and it is generally
@@ -532,7 +542,7 @@ index is updated automatically on the schedule set by
can manage it manually: can manage it manually:
``` ```
document_llmindex {rebuild,update,compact} document_llmindex {rebuild,update,compact,migrate}
``` ```
Specify `rebuild` to build the index from scratch from all documents in the database. Use Specify `rebuild` to build the index from scratch from all documents in the database. Use
@@ -3,6 +3,7 @@ from typing import Any
from documents.management.commands.base import PaperlessCommand from documents.management.commands.base import PaperlessCommand
from documents.tasks import llmindex_index from documents.tasks import llmindex_index
from paperless_ai.indexing import llm_index_compact from paperless_ai.indexing import llm_index_compact
from paperless_ai.indexing import llm_index_migrate
class Command(PaperlessCommand): class Command(PaperlessCommand):
@@ -13,12 +14,18 @@ class Command(PaperlessCommand):
def add_arguments(self, parser: Any) -> None: def add_arguments(self, parser: Any) -> None:
super().add_arguments(parser) super().add_arguments(parser)
parser.add_argument("command", choices=["rebuild", "update", "compact"]) parser.add_argument(
"command",
choices=["rebuild", "update", "compact", "migrate"],
)
def handle(self, *args: Any, **options: Any) -> None: def handle(self, *args: Any, **options: Any) -> None:
if options["command"] == "compact": if options["command"] == "compact":
llm_index_compact() llm_index_compact()
return return
if options["command"] == "migrate":
llm_index_migrate()
return
llmindex_index( llmindex_index(
rebuild=options["command"] == "rebuild", rebuild=options["command"] == "rebuild",
iter_wrapper=lambda docs: self.track( iter_wrapper=lambda docs: self.track(
@@ -9,6 +9,7 @@ if TYPE_CHECKING:
_COMPACT = "documents.management.commands.document_llmindex.llm_index_compact" _COMPACT = "documents.management.commands.document_llmindex.llm_index_compact"
_INDEX = "documents.management.commands.document_llmindex.llmindex_index" _INDEX = "documents.management.commands.document_llmindex.llmindex_index"
_MIGRATE = "documents.management.commands.document_llmindex.llm_index_migrate"
class TestDocumentLlmindexCommand: class TestDocumentLlmindexCommand:
@@ -17,6 +18,11 @@ class TestDocumentLlmindexCommand:
call_command("document_llmindex", "compact") call_command("document_llmindex", "compact")
mock_compact.assert_called_once_with() mock_compact.assert_called_once_with()
def test_migrate_calls_llm_index_migrate(self, mocker: MockerFixture) -> None:
mock_migrate = mocker.patch(_MIGRATE)
call_command("document_llmindex", "migrate")
mock_migrate.assert_called_once_with()
def test_rebuild_calls_llmindex_index_with_rebuild_true( def test_rebuild_calls_llmindex_index_with_rebuild_true(
self, self,
mocker: MockerFixture, mocker: MockerFixture,
+80 -28
View File
@@ -144,6 +144,24 @@ def _exclude_readers():
lock.close() lock.close()
def _with_exclusive_access(operation: str, fn):
"""Run ``fn()`` with exclusive index access (see ``_exclude_readers()``),
for compaction/migration file swaps that must not run while readers are
active. Returns ``fn()``'s result, or None (after logging) if active
readers do not drain within ``LLM_INDEX_COMPACTION_LOCK_TIMEOUT`` --
callers skip the operation this run; it retries next time.
"""
try:
with _exclude_readers():
return fn()
except Timeout:
logger.info(
"Skipping LLM index %s: index readers are active; will retry next run.",
operation,
)
return None
@contextmanager @contextmanager
def write_store(embed_model_name: str | None = None): def write_store(embed_model_name: str | None = None):
"""Acquire the write lock and yield the vector store. """Acquire the write lock and yield the vector store.
@@ -168,6 +186,21 @@ def write_store(embed_model_name: str | None = None):
yield store yield store
def _check_and_run_migrations(store: "PaperlessSqliteVecVectorStore") -> bool:
"""Run any pending structural migrations, returning True if a pending
re-embed migration needs the caller to force a rebuild -- never
triggered automatically here. Safe to call before any write, including
delete()/upsert_document(): has_pending_migration() (see its docstring)
keeps this a no-op, with no exclusive access taken, once the store is
current.
"""
if not store.has_pending_migration():
return False
return bool(
_with_exclusive_access("migration check", store.check_and_run_migrations),
)
def _safe_related_name(document: Document, field: str) -> str | None: def _safe_related_name(document: Document, field: str) -> str | None:
""" """
Returns the ``name`` of a related object (correspondent, document_type, Returns the ``name`` of a related object (correspondent, document_type,
@@ -339,15 +372,7 @@ def update_llm_index(
happens, since a rebuild always covers the whole library regardless. happens, since a rebuild always covers the whole library regardless.
""" """
with write_store() as store: with write_store() as store:
try: needs_reembed = _check_and_run_migrations(store)
with _exclude_readers():
needs_reembed = store.check_and_run_migrations()
except Timeout:
logger.info(
"Skipping LLM index migration check: index readers are active; "
"will retry next run.",
)
needs_reembed = False
if needs_reembed: if needs_reembed:
logger.warning( logger.warning(
"LLM index migration requires re-embedding; forcing rebuild.", "LLM index migration requires re-embedding; forcing rebuild.",
@@ -368,10 +393,9 @@ def update_llm_index(
config = AIConfig() config = AIConfig()
model_name = get_configured_model_name(config) model_name = get_configured_model_name(config)
if not rebuild and llm_index_exists(): if not rebuild:
with read_store() as store: with read_store() as store:
config_mismatch = store.config_mismatch(model_name) if store.table_exists() and store.config_mismatch(model_name):
if config_mismatch:
logger.warning("Embedding model changed; forcing LLM index rebuild.") logger.warning("Embedding model changed; forcing LLM index rebuild.")
rebuild = True rebuild = True
@@ -412,14 +436,7 @@ def update_llm_index(
else "No changes detected in LLM index." else "No changes detected in LLM index."
) )
try: _with_exclusive_access("compaction", store.compact)
with _exclude_readers():
store.compact()
except Timeout:
logger.info(
"Skipping LLM index compaction: index readers are active; "
"will retry next run.",
)
return msg return msg
@@ -434,25 +451,60 @@ def llm_index_add_or_update_document(document: Document):
_embed_nodes(new_nodes, get_embedding_model(config)) _embed_nodes(new_nodes, get_embedding_model(config))
with write_store(embed_model_name=get_configured_model_name(config)) as store: with write_store(embed_model_name=get_configured_model_name(config)) as store:
needs_reembed = _check_and_run_migrations(store)
if needs_reembed:
logger.warning(
"Skipping incremental LLM index update for document %s: the "
"index requires re-embedding first. Run 'document_llmindex "
"rebuild' to resolve.",
document.id,
)
return
store.upsert_document(str(document.id), new_nodes) store.upsert_document(str(document.id), new_nodes)
def llm_index_migrate() -> None:
"""Apply any pending LLM index schema migrations, with no reindex.
Intended to run unconditionally on every startup (see the
init-llmindex-migrate container step and the bare-metal upgrade docs):
has_pending_migration() short-circuits to a metadata-only read once the
store is current, so a healthy install pays almost nothing here. Only
ever applies structural migrations -- a pending re-embed migration is
left for the explicit, deliberate rebuild path (``document_llmindex
update``/``rebuild``) to resolve, since re-embedding can be slow and,
for a metered embedding backend, cost money.
"""
if not AIConfig().llm_index_enabled:
return
with write_store() as store:
needs_reembed = _check_and_run_migrations(store)
if needs_reembed:
logger.warning(
"LLM index requires re-embedding, which this automatic migration "
"check will not do on its own -- it can be slow and, for a "
"metered embedding backend, cost money. Run "
"'document_llmindex rebuild' manually when ready.",
)
def llm_index_compact() -> None: def llm_index_compact() -> None:
"""Compact the index immediately, rebuilding the table to reclaim space.""" """Compact the index immediately, rebuilding the table to reclaim space."""
with write_store() as store: with write_store() as store:
try: _with_exclusive_access("compaction", lambda: store.compact(force=True))
with _exclude_readers():
store.compact(force=True)
except Timeout:
logger.info(
"Skipping LLM index compaction: index readers are active; "
"will retry next run.",
)
def llm_index_remove_document(document: Document): def llm_index_remove_document(document: Document):
"""Remove a document's chunks from the LLM index.""" """Remove a document's chunks from the LLM index."""
with write_store() as store: with write_store() as store:
if _check_and_run_migrations(store):
logger.warning(
"Skipping removal of document %s from the LLM index: the "
"index requires re-embedding first. Run 'document_llmindex "
"rebuild' to resolve.",
document.id,
)
return
store.delete(str(document.id)) store.delete(str(document.id))
+60
View File
@@ -0,0 +1,60 @@
"""Schema migrations for the sqlite-vec vector store.
Each migration lives in its own module here, named ``mNNNN_description.py``
(e.g. ``m0001_v1_to_v2.py`` -- a leading digit isn't a valid Python
identifier, hence the ``m`` prefix, unlike Django's own numbered migrations,
which load via a dynamic ``importlib.import_module()`` call rather than a
static import statement), and registers itself into ``MIGRATIONS`` at import
time. ``vector_store.py`` imports those modules at the bottom of the file,
purely for that registration side effect, after ``PaperlessSqliteVecVectorStore``
is fully defined -- migrations need it to implement ``apply()`` (see
``Migration`` below).
To add a new migration: add a new ``mNNNN_description.py`` module here that
imports ``PaperlessSqliteVecVectorStore`` from ``paperless_ai.vector_store``,
defines its ``apply()``, and appends a ``Migration`` to ``MIGRATIONS``; then
import that module at the bottom of ``vector_store.py`` and bump
``SCHEMA_VERSION`` there. A migration must freeze its own historical DDL for
any side table its target version depends on (``DROP TABLE IF EXISTS`` +
its own literal ``CREATE TABLE``/``CREATE INDEX`` statements) rather than
delegating to any "current schema" helper -- see ``m0001_v1_to_v2.py`` for
why and the worked example.
"""
import sqlite3
from collections.abc import Callable
from dataclasses import dataclass
from dataclasses import field
from typing import Literal
@dataclass
class Migration:
"""A schema migration for the sqlite-vec vector store.
kind="structural": rows are copied into a new-schema file with no
re-embedding needed. Supply ``apply(src_conn, dst_conn, dim)``, which
must create every table its target schema needs in ``dst_conn`` and copy
``src_conn``'s rows and relevant ``index_meta`` keys into it.
``schema_version`` is written by the migration runner after ``apply``
returns, not by ``apply`` itself.
kind="re-embed": the new schema requires fresh embeddings.
``check_and_run_migrations()`` returns True when it encounters one of
these so the caller can force a full rebuild (which recreates the table
at the current SCHEMA_VERSION).
"""
from_version: int
to_version: int
kind: Literal["structural", "re-embed"]
description: str
apply: Callable[[sqlite3.Connection, sqlite3.Connection, int], None] | None = field(
default=None,
repr=False,
)
# Registry of all schema migrations in order, populated by each migration
# module's import-time registration (see the module docstring above).
MIGRATIONS: list[Migration] = []
@@ -0,0 +1,118 @@
import sqlite3
from paperless_ai.migrations import MIGRATIONS
from paperless_ai.migrations import Migration
from paperless_ai.tables import ChunkRow
from paperless_ai.tables import DocumentChunksTable
from paperless_ai.tables import DocumentMetaRow
from paperless_ai.tables import DocumentMetaTable
from paperless_ai.tables import IndexMetaTable
from paperless_ai.vector_store import COMPACT_BATCH_SIZE
from paperless_ai.vector_store import DEFAULT_TABLE_NAME
# v1's vec0 shape has never changed since it first shipped and is the ONLY
# real upgrade path -- no store has ever existed at any intermediate
# version, so this migration goes straight from that shipped shape to the
# final v2 target in one pass.
_V1_SELECT = (
"SELECT id, document_id, modified, node_content, embedding FROM "
+ DEFAULT_TABLE_NAME
)
def _migrate_v1_to_v2(
src_conn: sqlite3.Connection,
dst_conn: sqlite3.Connection,
dim: int,
) -> None:
"""v1 -> v2: document_id TEXT -> INTEGER, modified moves out of vec0
into document_meta, document_chunks added for O(1) per-document delete.
Freezes its own v2-shaped vec0/document_chunks/document_meta DDL inline,
rather than delegating to the gateway "create table" helpers or the
store's own vec0-table builder (all of which always reflect the
*current* schema): a later schema version changing any of these tables'
shape must not silently change what this migration produces for someone
upgrading straight from v1.
_open_connection() already created document_chunks/document_meta on
dst_conn (reflecting current HEAD) as a side effect of opening it for
this migration's rebuild -- DROP them first so this migration's own
frozen CREATE TABLE isn't a silent no-op against that. Safe here because
dst_conn is a freshly opened, empty rebuild file with nothing written
yet.
"""
dst_conn.execute( # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
"CREATE VIRTUAL TABLE "
+ DEFAULT_TABLE_NAME
+ " USING vec0("
+ "id TEXT PRIMARY KEY,"
+ " document_id INTEGER,"
+ " +node_content TEXT,"
+ " embedding float["
+ str(int(dim))
+ "] distance_metric=cosine"
+ ")",
)
dst_conn.execute("DROP TABLE IF EXISTS document_chunks")
dst_conn.execute(
"CREATE TABLE document_chunks "
"(chunk_id TEXT PRIMARY KEY, document_id INTEGER NOT NULL)",
)
dst_conn.execute(
"CREATE INDEX idx_document_chunks_document_id ON document_chunks (document_id)",
)
dst_conn.execute("DROP TABLE IF EXISTS document_meta")
dst_conn.execute(
"CREATE TABLE document_meta "
"(document_id INTEGER PRIMARY KEY, modified TEXT NOT NULL)",
)
IndexMetaTable.set_dim(dst_conn, dim)
embed_model = IndexMetaTable.get_embed_model(src_conn)
if embed_model is not None:
IndexMetaTable.set_embed_model(dst_conn, embed_model)
dst_conn.execute("BEGIN IMMEDIATE")
src_cursor = src_conn.execute(_V1_SELECT)
live = 0
while batch := src_cursor.fetchmany(COMPACT_BATCH_SIZE):
vec0_rows = []
chunk_rows = []
meta_by_document: dict[int, str] = {}
for r in batch:
document_id = int(r["document_id"])
vec0_rows.append(
(r["id"], document_id, r["node_content"], bytes(r["embedding"])),
)
chunk_rows.append(ChunkRow(r["id"], document_id))
meta_by_document[document_id] = str(r["modified"] or "")
dst_conn.executemany(
"INSERT INTO "
+ DEFAULT_TABLE_NAME
+ " (id, document_id, node_content, embedding) VALUES (?, ?, ?, ?)",
vec0_rows,
)
DocumentChunksTable.insert_many(dst_conn, chunk_rows)
DocumentMetaTable.upsert_many(
dst_conn,
(DocumentMetaRow(doc_id, mod) for doc_id, mod in meta_by_document.items()),
)
live += len(batch)
# This migration only ever copies live rows (like compact()), so the
# cumulative counter resets to match -- the new file has no bloat yet.
IndexMetaTable.reset_total_inserts(dst_conn, live)
dst_conn.execute("COMMIT")
MIGRATIONS.append(
Migration(
from_version=1,
to_version=2,
kind="structural",
description=(
"document_id TEXT -> INTEGER; move modified into document_meta; "
"add document_chunks for O(1) per-document delete"
),
apply=_migrate_v1_to_v2,
),
)
+242
View File
@@ -0,0 +1,242 @@
"""Thin gateways over the plain relational side tables that sit alongside the
vec0 table. Each method takes the sqlite3.Connection to operate on
explicitly, rather than owning one -- the store swaps connections during
compact()/migration, and migrations always work across two connections
(src_conn, dst_conn) at once.
PRECONDITION: Callers must set conn.row_factory = sqlite3.Row before passing a
connection to any of these gateways' read methods. The read methods across all
three classes (DocumentChunksTable.chunk_ids_for_document, IndexMetaTable._get,
DocumentMetaTable.all_modified_times, DocumentMetaTable.copy_all) use
row["column_name"] dictionary-style indexing, which requires sqlite3.Row as the
row factory -- without it, sqlite3.Row is not set, rows are returned as plain
tuples, and tuple indices must be integers, raising TypeError.
"""
import sqlite3
from collections.abc import Iterable
from typing import NamedTuple
class ChunkRow(NamedTuple):
chunk_id: str
document_id: int
class DocumentMetaRow(NamedTuple):
document_id: int
modified: str
class DocumentChunksTable:
"""chunk_id -> document_id, indexed by document_id. Gives O(1)
per-document chunk lookup that vec0's own document_id metadata column
cannot (see PaperlessSqliteVecVectorStore._delete_chunks_by_document_id).
"""
@staticmethod
def create(conn: sqlite3.Connection) -> None:
conn.execute(
"CREATE TABLE IF NOT EXISTS document_chunks "
"(chunk_id TEXT PRIMARY KEY, document_id INTEGER NOT NULL)",
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_document_chunks_document_id "
"ON document_chunks (document_id)",
)
@staticmethod
def insert_many(conn: sqlite3.Connection, rows: Iterable[ChunkRow]) -> None:
"""rows must already be batch-bounded by the caller (e.g. vec0's own
fetchmany() loop) -- this never reads, so it can't itself introduce
an unbounded scan, but a whole-table iterable defeats the point."""
conn.executemany(
"INSERT INTO document_chunks (chunk_id, document_id) VALUES (?, ?)",
rows,
)
@staticmethod
def chunk_ids_for_document(
conn: sqlite3.Connection,
document_id: int,
) -> list[str]:
return [
row["chunk_id"]
for row in conn.execute(
"SELECT chunk_id FROM document_chunks WHERE document_id = ?",
(document_id,),
).fetchall()
]
@staticmethod
def delete_for_document(conn: sqlite3.Connection, document_id: int) -> None:
conn.execute(
"DELETE FROM document_chunks WHERE document_id = ?",
(document_id,),
)
@staticmethod
def delete_all(conn: sqlite3.Connection) -> None:
conn.execute("DELETE FROM document_chunks")
@staticmethod
def count(conn: sqlite3.Connection) -> int:
"""Cheap stand-in for vec0's own row count -- see compact()."""
return conn.execute("SELECT count(*) FROM document_chunks").fetchone()[0]
class DocumentMetaTable:
"""document_id -> modified, one row per document. Lives outside vec0
because vec0 only inlines TEXT metadata up to 12 bytes and `modified`
(an ISO timestamp) is always longer.
"""
@staticmethod
def create(conn: sqlite3.Connection) -> None:
conn.execute(
"CREATE TABLE IF NOT EXISTS document_meta "
"(document_id INTEGER PRIMARY KEY, modified TEXT NOT NULL)",
)
@staticmethod
def upsert_many(
conn: sqlite3.Connection,
rows: Iterable[DocumentMetaRow],
) -> None:
conn.executemany(
"INSERT INTO document_meta (document_id, modified) VALUES (?, ?) "
"ON CONFLICT(document_id) DO UPDATE SET modified = excluded.modified",
rows,
)
@staticmethod
def delete_for_document(conn: sqlite3.Connection, document_id: int) -> None:
conn.execute(
"DELETE FROM document_meta WHERE document_id = ?",
(document_id,),
)
@staticmethod
def delete_all(conn: sqlite3.Connection) -> None:
conn.execute("DELETE FROM document_meta")
@staticmethod
def copy_all(
src_conn: sqlite3.Connection,
dst_conn: sqlite3.Connection,
batch_size: int,
) -> None:
"""Stream document_meta from src_conn into dst_conn in bounded
batches. The *only* sanctioned way to move this table across
connections (compact()/migrations) -- an unbounded fetchall here
would defeat the same OOM-avoidance the vec0 row copy already relies
on. batch_size has no default: forces the call site to think about
it (pass COMPACT_BATCH_SIZE)."""
cursor = src_conn.execute(
"SELECT document_id, modified FROM document_meta",
)
while batch := cursor.fetchmany(batch_size):
DocumentMetaTable.upsert_many(
dst_conn,
(DocumentMetaRow(r["document_id"], r["modified"]) for r in batch),
)
@staticmethod
def all_modified_times(conn: sqlite3.Connection) -> dict[str, str]:
"""Full document_id -> modified map, for get_modified_times()'s
public API only. One unbounded read by design (existing behavior).
Never use this for cross-connection copying; see copy_all()."""
return {
str(row["document_id"]): str(row["modified"] or "")
for row in conn.execute(
"SELECT document_id, modified FROM document_meta",
)
}
class IndexMetaTable:
"""Typed accessors over index_meta's key/value rows -- replaces
PaperlessSqliteVecVectorStore._meta_get_on/_meta_set_on, which returned
untyped str | None regardless of whether the key held an int (dim,
schema_version, total_inserts) or a string (embed_model).
"""
@staticmethod
def create(conn: sqlite3.Connection) -> None:
conn.execute(
"CREATE TABLE IF NOT EXISTS index_meta (key TEXT PRIMARY KEY, value TEXT)",
)
@staticmethod
def _get(conn: sqlite3.Connection, key: str) -> str | None:
row = conn.execute(
"SELECT value FROM index_meta WHERE key = ?",
(key,),
).fetchone()
return row["value"] if row else None
@staticmethod
def _set(conn: sqlite3.Connection, key: str, value: str) -> None:
conn.execute(
"INSERT INTO index_meta (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(key, value),
)
@staticmethod
def get_dim(conn: sqlite3.Connection) -> int | None:
value = IndexMetaTable._get(conn, "dim")
return int(value) if value is not None else None
@staticmethod
def set_dim(conn: sqlite3.Connection, dim: int) -> None:
IndexMetaTable._set(conn, "dim", str(dim))
@staticmethod
def get_embed_model(conn: sqlite3.Connection) -> str | None:
return IndexMetaTable._get(conn, "embed_model")
@staticmethod
def set_embed_model(conn: sqlite3.Connection, name: str) -> None:
IndexMetaTable._set(conn, "embed_model", name)
@staticmethod
def get_schema_version(conn: sqlite3.Connection) -> int | None:
value = IndexMetaTable._get(conn, "schema_version")
return int(value) if value is not None else None
@staticmethod
def set_schema_version(conn: sqlite3.Connection, version: int) -> None:
IndexMetaTable._set(conn, "schema_version", str(version))
@staticmethod
def get_total_inserts(conn: sqlite3.Connection) -> int:
value = IndexMetaTable._get(conn, "total_inserts")
return int(value) if value is not None else 0
@staticmethod
def increment_total_inserts(conn: sqlite3.Connection, count: int) -> None:
"""Atomically add ``count`` to the stored counter in one statement
(INSERT .. ON CONFLICT DO UPDATE with arithmetic), instead of a
separate read-then-write -- called once per add()/upsert_document(),
so halving the statement count here is a real, if small, per-call
saving. index_meta.value has TEXT affinity, so the incremented
result is stored as its text representation -- get_total_inserts()
already expects that (int(value)), so this is not a behavior
change, only fewer statements.
"""
conn.execute(
"INSERT INTO index_meta (key, value) VALUES ('total_inserts', ?) "
"ON CONFLICT(key) DO UPDATE SET value = "
"CAST(index_meta.value AS INTEGER) + CAST(excluded.value AS INTEGER)",
(str(count),),
)
@staticmethod
def reset_total_inserts(conn: sqlite3.Connection, count: int) -> None:
"""Set total_inserts to an absolute value -- distinct from
increment_total_inserts(): used by compact()'s rebuild and by
m0001_v1_to_v2 after copying live rows into a fresh file, where
total_inserts must become exactly the live row count, not add to
whatever the source file's counter held."""
IndexMetaTable._set(conn, "total_inserts", str(count))
+169
View File
@@ -1,3 +1,4 @@
import logging
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock from unittest.mock import MagicMock
from unittest.mock import patch from unittest.mock import patch
@@ -252,6 +253,45 @@ def test_update_llm_index_rebuilds_on_model_name_change(
assert store.stored_model_name() == "model-b" assert store.stored_model_name() == "model-b"
@pytest.mark.django_db
def test_update_llm_index_merges_exists_and_config_mismatch_reads(
temp_llm_index_dir: Path,
real_document: Document,
mock_embed_model: FakeEmbedding,
) -> None:
# Build an initial index so the second call's table_exists()/
# config_mismatch() checks have something real to check against.
with patch("documents.models.Document.objects.all") as mock_all:
mock_queryset = MagicMock()
mock_queryset.exists.return_value = True
mock_queryset.__iter__.return_value = iter([real_document])
mock_queryset.select_related.return_value = mock_queryset
mock_queryset.prefetch_related.return_value = mock_queryset
mock_all.return_value = mock_queryset
indexing.update_llm_index(rebuild=True)
with patch("documents.models.Document.objects.all") as mock_all:
mock_queryset = MagicMock()
mock_queryset.exists.return_value = True
mock_queryset.__iter__.return_value = iter([real_document])
mock_queryset.select_related.return_value = mock_queryset
mock_queryset.prefetch_related.return_value = mock_queryset
mock_all.return_value = mock_queryset
with patch(
"paperless_ai.indexing.read_store",
wraps=indexing.read_store,
) as read_store_spy:
indexing.update_llm_index(rebuild=False)
# Documents exist, so the fast-exit check's `no_documents and ...`
# short-circuits before ever calling llm_index_exists() -- the only
# read_store() call left in this path is the merged table_exists()/
# config_mismatch() check. Before this task's fix, that merged check
# was two separate read_store() calls (one inside llm_index_exists(),
# one for config_mismatch() right after) -- so this asserts 1, not 2.
assert read_store_spy.call_count == 1
@pytest.mark.django_db @pytest.mark.django_db
def test_update_llm_index_partial_update( def test_update_llm_index_partial_update(
temp_llm_index_dir: Path, temp_llm_index_dir: Path,
@@ -737,6 +777,7 @@ class TestLlmIndexLocking:
mocker: pytest_mock.MockerFixture, mocker: pytest_mock.MockerFixture,
) -> None: ) -> None:
mock_store = MagicMock() mock_store = MagicMock()
mock_store.has_pending_migration.return_value = False
mocker.patch( mocker.patch(
"paperless_ai.indexing.write_store", "paperless_ai.indexing.write_store",
return_value=mocker.MagicMock( return_value=mocker.MagicMock(
@@ -757,12 +798,45 @@ class TestLlmIndexLocking:
mock_store.upsert_document.assert_called_once() mock_store.upsert_document.assert_called_once()
def test_add_or_update_document_skips_write_when_reembed_pending(
self,
temp_llm_index_dir: Path,
mock_embed_model: FakeEmbedding,
mocker: pytest_mock.MockerFixture,
) -> None:
"""A pending re-embed migration must block the incremental write,
not let it proceed against a schema that just changed underneath it.
"""
mock_store = MagicMock()
mock_store.has_pending_migration.return_value = True
mock_store.check_and_run_migrations.return_value = True
mocker.patch(
"paperless_ai.indexing.write_store",
return_value=mocker.MagicMock(
__enter__=mocker.MagicMock(return_value=mock_store),
__exit__=mocker.MagicMock(return_value=False),
),
)
mock_node = MagicMock()
mock_node.get_content.return_value = "fake node text"
mocker.patch(
"paperless_ai.indexing.build_document_node",
return_value=[mock_node],
)
doc = MagicMock(spec=Document)
doc.id = 1
indexing.llm_index_add_or_update_document(doc)
mock_store.upsert_document.assert_not_called()
def test_remove_document_uses_write_store( def test_remove_document_uses_write_store(
self, self,
temp_llm_index_dir: Path, temp_llm_index_dir: Path,
mocker: pytest_mock.MockerFixture, mocker: pytest_mock.MockerFixture,
) -> None: ) -> None:
mock_store = MagicMock() mock_store = MagicMock()
mock_store.has_pending_migration.return_value = False
mocker.patch( mocker.patch(
"paperless_ai.indexing.write_store", "paperless_ai.indexing.write_store",
return_value=mocker.MagicMock( return_value=mocker.MagicMock(
@@ -777,6 +851,31 @@ class TestLlmIndexLocking:
mock_store.delete.assert_called_once_with("1") mock_store.delete.assert_called_once_with("1")
def test_remove_document_skips_write_when_reembed_pending(
self,
temp_llm_index_dir: Path,
mocker: pytest_mock.MockerFixture,
) -> None:
"""A pending re-embed migration must block the delete too, for the
same consistency reason as the incremental-update path.
"""
mock_store = MagicMock()
mock_store.has_pending_migration.return_value = True
mock_store.check_and_run_migrations.return_value = True
mocker.patch(
"paperless_ai.indexing.write_store",
return_value=mocker.MagicMock(
__enter__=mocker.MagicMock(return_value=mock_store),
__exit__=mocker.MagicMock(return_value=False),
),
)
doc = MagicMock(spec=Document)
doc.id = 1
indexing.llm_index_remove_document(doc)
mock_store.delete.assert_not_called()
def test_update_llm_index_rebuild_uses_write_store( def test_update_llm_index_rebuild_uses_write_store(
self, self,
temp_llm_index_dir: Path, temp_llm_index_dir: Path,
@@ -849,6 +948,76 @@ class TestVectorStoreIndexing:
assert rows >= 1 assert rows >= 1
class TestLlmIndexMigrate:
def test_noop_when_ai_disabled(self, mocker: pytest_mock.MockerFixture) -> None:
"""
GIVEN:
- AI/LLM index support is disabled in configuration
WHEN:
- llm_index_migrate() is called
THEN:
- No store is opened and no migration check runs
"""
mocker.patch(
"paperless_ai.indexing.AIConfig",
return_value=mocker.Mock(llm_index_enabled=False),
)
write_store_mock = mocker.patch("paperless_ai.indexing.write_store")
indexing.llm_index_migrate()
write_store_mock.assert_not_called()
def test_runs_pending_migration_when_enabled(
self,
mocker: pytest_mock.MockerFixture,
) -> None:
"""
GIVEN:
- AI/LLM index support is enabled
WHEN:
- llm_index_migrate() is called
THEN:
- The store is opened for write and a migration check runs
"""
mocker.patch(
"paperless_ai.indexing.AIConfig",
return_value=mocker.Mock(llm_index_enabled=True),
)
store_mock = mocker.MagicMock()
store_mock.has_pending_migration.return_value = False
write_store_cm = mocker.patch("paperless_ai.indexing.write_store")
write_store_cm.return_value.__enter__.return_value = store_mock
indexing.llm_index_migrate()
store_mock.has_pending_migration.assert_called_once()
def test_logs_warning_when_reembed_needed(
self,
mocker: pytest_mock.MockerFixture,
caplog: pytest.LogCaptureFixture,
) -> None:
"""
GIVEN:
- AI/LLM index support is enabled
- A pending migration requires re-embedding
WHEN:
- llm_index_migrate() is called
THEN:
- A warning directs the operator to run a manual rebuild, since
this automatic check must never re-embed on its own
"""
mocker.patch(
"paperless_ai.indexing.AIConfig",
return_value=mocker.Mock(llm_index_enabled=True),
)
store_mock = mocker.MagicMock()
store_mock.has_pending_migration.return_value = True
store_mock.check_and_run_migrations.return_value = True
write_store_cm = mocker.patch("paperless_ai.indexing.write_store")
write_store_cm.return_value.__enter__.return_value = store_mock
with caplog.at_level(logging.WARNING, logger="paperless_ai.indexing"):
indexing.llm_index_migrate()
assert "requires re-embedding" in caplog.text
@pytest.mark.django_db @pytest.mark.django_db
class TestQuerySimilarDocuments: class TestQuerySimilarDocuments:
def test_query_similar_documents_respects_allowed_ids( def test_query_similar_documents_respects_allowed_ids(
+340
View File
@@ -0,0 +1,340 @@
import sqlite3
from collections.abc import Generator
import pytest
from pytest_mock import MockerFixture
from paperless_ai.tables import ChunkRow
from paperless_ai.tables import DocumentChunksTable
from paperless_ai.tables import DocumentMetaRow
from paperless_ai.tables import DocumentMetaTable
from paperless_ai.tables import IndexMetaTable
@pytest.fixture
def conn() -> Generator[sqlite3.Connection, None, None]:
connection = sqlite3.connect(":memory:")
connection.row_factory = sqlite3.Row
try:
yield connection
finally:
connection.close()
class TestDocumentChunksTable:
def test_create_is_idempotent(self, conn: sqlite3.Connection) -> None:
"""
GIVEN:
- A bare sqlite3 connection
WHEN:
- create() is called, a row is inserted, then create() is called again
THEN:
- No error is raised and the row survives uncorrupted
"""
DocumentChunksTable.create(conn)
DocumentChunksTable.insert_many(conn, [ChunkRow("c1", 1)])
DocumentChunksTable.create(conn)
assert DocumentChunksTable.chunk_ids_for_document(conn, 1) == ["c1"]
def test_insert_many_then_lookup_by_document_id(
self,
conn: sqlite3.Connection,
) -> None:
"""
GIVEN:
- An empty document_chunks table
WHEN:
- Two chunks for document 1 and one for document 2 are inserted
THEN:
- chunk_ids_for_document returns exactly the matching chunk ids
"""
DocumentChunksTable.create(conn)
DocumentChunksTable.insert_many(
conn,
[ChunkRow("c1", 1), ChunkRow("c2", 1), ChunkRow("c3", 2)],
)
assert sorted(DocumentChunksTable.chunk_ids_for_document(conn, 1)) == [
"c1",
"c2",
]
assert DocumentChunksTable.chunk_ids_for_document(conn, 2) == ["c3"]
assert DocumentChunksTable.chunk_ids_for_document(conn, 999) == []
def test_delete_for_document_removes_only_that_document(
self,
conn: sqlite3.Connection,
) -> None:
"""
GIVEN:
- Chunks for two different documents
WHEN:
- delete_for_document() is called for one of them
THEN:
- Only that document's chunks are removed
"""
DocumentChunksTable.create(conn)
DocumentChunksTable.insert_many(
conn,
[ChunkRow("c1", 1), ChunkRow("c2", 2)],
)
DocumentChunksTable.delete_for_document(conn, 1)
assert DocumentChunksTable.chunk_ids_for_document(conn, 1) == []
assert DocumentChunksTable.chunk_ids_for_document(conn, 2) == ["c2"]
def test_delete_all_clears_every_row(self, conn: sqlite3.Connection) -> None:
"""
GIVEN:
- Chunks for multiple documents
WHEN:
- delete_all() is called
THEN:
- count() returns 0
"""
DocumentChunksTable.create(conn)
DocumentChunksTable.insert_many(
conn,
[ChunkRow("c1", 1), ChunkRow("c2", 2)],
)
DocumentChunksTable.delete_all(conn)
assert DocumentChunksTable.count(conn) == 0
def test_count_reflects_live_rows(self, conn: sqlite3.Connection) -> None:
"""
GIVEN:
- An empty document_chunks table
WHEN:
- Rows are inserted then one document's rows are deleted
THEN:
- count() reflects the remaining row count
"""
DocumentChunksTable.create(conn)
DocumentChunksTable.insert_many(
conn,
[ChunkRow("c1", 1), ChunkRow("c2", 1), ChunkRow("c3", 2)],
)
assert DocumentChunksTable.count(conn) == 3
DocumentChunksTable.delete_for_document(conn, 1)
assert DocumentChunksTable.count(conn) == 1
class TestDocumentMetaTable:
def test_upsert_many_then_all_modified_times(
self,
conn: sqlite3.Connection,
) -> None:
"""
GIVEN:
- An empty document_meta table
WHEN:
- Two documents' modified timestamps are upserted
THEN:
- all_modified_times() returns both, keyed by str(document_id)
"""
DocumentMetaTable.create(conn)
DocumentMetaTable.upsert_many(
conn,
[
DocumentMetaRow(1, "2026-01-01T00:00:00"),
DocumentMetaRow(2, "2026-02-02T00:00:00"),
],
)
assert DocumentMetaTable.all_modified_times(conn) == {
"1": "2026-01-01T00:00:00",
"2": "2026-02-02T00:00:00",
}
def test_upsert_many_overwrites_existing_value(
self,
conn: sqlite3.Connection,
) -> None:
"""
GIVEN:
- A document_meta row for document 1
WHEN:
- upsert_many() is called again with a new modified value for
the same document_id
THEN:
- The stored value is replaced, not duplicated
"""
DocumentMetaTable.create(conn)
DocumentMetaTable.upsert_many(conn, [DocumentMetaRow(1, "old")])
DocumentMetaTable.upsert_many(conn, [DocumentMetaRow(1, "new")])
assert DocumentMetaTable.all_modified_times(conn) == {"1": "new"}
def test_delete_for_document_removes_only_that_row(
self,
conn: sqlite3.Connection,
) -> None:
"""
GIVEN:
- document_meta rows for two documents
WHEN:
- delete_for_document() is called for one of them
THEN:
- Only that document's row is removed
"""
DocumentMetaTable.create(conn)
DocumentMetaTable.upsert_many(
conn,
[DocumentMetaRow(1, "a"), DocumentMetaRow(2, "b")],
)
DocumentMetaTable.delete_for_document(conn, 1)
assert DocumentMetaTable.all_modified_times(conn) == {"2": "b"}
def test_delete_all_clears_every_row(self, conn: sqlite3.Connection) -> None:
"""
GIVEN:
- document_meta rows for multiple documents
WHEN:
- delete_all() is called
THEN:
- all_modified_times() returns an empty dict
"""
DocumentMetaTable.create(conn)
DocumentMetaTable.upsert_many(
conn,
[DocumentMetaRow(1, "a"), DocumentMetaRow(2, "b")],
)
DocumentMetaTable.delete_all(conn)
assert DocumentMetaTable.all_modified_times(conn) == {}
def test_copy_all_streams_every_row_to_destination(
self,
conn: sqlite3.Connection,
) -> None:
"""
GIVEN:
- A source connection with document_meta rows for 5 documents
- A separate, empty destination connection
WHEN:
- copy_all() is called with a batch size smaller than the row
count, forcing multiple fetchmany() cycles
THEN:
- Every row is present on the destination connection
"""
DocumentMetaTable.create(conn)
DocumentMetaTable.upsert_many(
conn,
[DocumentMetaRow(i, f"modified-{i}") for i in range(5)],
)
dst_conn = sqlite3.connect(":memory:")
dst_conn.row_factory = sqlite3.Row
try:
DocumentMetaTable.create(dst_conn)
DocumentMetaTable.copy_all(conn, dst_conn, batch_size=2)
assert DocumentMetaTable.all_modified_times(dst_conn) == {
str(i): f"modified-{i}" for i in range(5)
}
finally:
dst_conn.close()
class TestIndexMetaTable:
@pytest.mark.parametrize(
("setter_name", "getter_name", "value"),
[
("set_dim", "get_dim", 384),
("set_embed_model", "get_embed_model", "model-a"),
("set_schema_version", "get_schema_version", 2),
],
)
def test_typed_accessor_roundtrip(
self,
conn: sqlite3.Connection,
setter_name: str,
getter_name: str,
value: int | str,
) -> None:
"""
GIVEN:
- An empty index_meta table
WHEN:
- A typed accessor's setter is called then the getter is read back
THEN:
- The same value is returned, correctly typed (int or str)
"""
IndexMetaTable.create(conn)
getter = getattr(IndexMetaTable, getter_name)
setter = getattr(IndexMetaTable, setter_name)
assert getter(conn) is None
setter(conn, value)
assert getter(conn) == value
def test_total_inserts_starts_at_zero(self, conn: sqlite3.Connection) -> None:
"""
GIVEN:
- An empty index_meta table
WHEN:
- get_total_inserts() is read before anything is set
THEN:
- 0 is returned
"""
IndexMetaTable.create(conn)
assert IndexMetaTable.get_total_inserts(conn) == 0
def test_increment_total_inserts_accumulates(
self,
conn: sqlite3.Connection,
) -> None:
"""
GIVEN:
- An empty index_meta table
WHEN:
- increment_total_inserts() is called twice
THEN:
- get_total_inserts() returns the running sum
"""
IndexMetaTable.create(conn)
IndexMetaTable.increment_total_inserts(conn, 5)
IndexMetaTable.increment_total_inserts(conn, 3)
assert IndexMetaTable.get_total_inserts(conn) == 8
def test_increment_total_inserts_is_a_single_statement(
self,
mocker: MockerFixture,
) -> None:
"""
GIVEN:
- An empty index_meta table
WHEN:
- increment_total_inserts() is called
THEN:
- Exactly one conn.execute() call is made (a single INSERT ...
ON CONFLICT DO UPDATE, not a separate read then write)
"""
# sqlite3.Connection is an immutable C extension type with no
# instance __dict__, so mocker.spy(conn, "execute") can't shadow
# "execute" on a plain connection ("attribute 'execute' is
# read-only"). A trivial Python subclass gets a normal instance
# __dict__, making the instance spyable while still being a real,
# usable sqlite3.Connection.
class _SpyableConnection(sqlite3.Connection):
pass
conn = sqlite3.connect(":memory:", factory=_SpyableConnection)
try:
conn.row_factory = sqlite3.Row
IndexMetaTable.create(conn)
execute_spy = mocker.spy(conn, "execute")
IndexMetaTable.increment_total_inserts(conn, 5)
assert execute_spy.call_count == 1
finally:
conn.close()
def test_reset_total_inserts_sets_absolute_value(
self,
conn: sqlite3.Connection,
) -> None:
"""
GIVEN:
- A total_inserts counter already at a high value
WHEN:
- reset_total_inserts() is called with a lower value
THEN:
- get_total_inserts() returns exactly that value, not a sum
"""
IndexMetaTable.create(conn)
IndexMetaTable.increment_total_inserts(conn, 100)
IndexMetaTable.reset_total_inserts(conn, 7)
assert IndexMetaTable.get_total_inserts(conn) == 7
+438 -96
View File
@@ -8,21 +8,23 @@ from llama_index.core.vector_stores.types import FilterOperator
from llama_index.core.vector_stores.types import MetadataFilter from llama_index.core.vector_stores.types import MetadataFilter
from llama_index.core.vector_stores.types import MetadataFilters from llama_index.core.vector_stores.types import MetadataFilters
from llama_index.core.vector_stores.types import VectorStoreQuery from llama_index.core.vector_stores.types import VectorStoreQuery
from pytest_mock import MockerFixture
from paperless_ai.migrations import MIGRATIONS
from paperless_ai.migrations import Migration
from paperless_ai.vector_store import DB_FILENAME from paperless_ai.vector_store import DB_FILENAME
from paperless_ai.vector_store import DEFAULT_TABLE_NAME from paperless_ai.vector_store import DEFAULT_TABLE_NAME
from paperless_ai.vector_store import MIGRATIONS
from paperless_ai.vector_store import SCHEMA_VERSION from paperless_ai.vector_store import SCHEMA_VERSION
from paperless_ai.vector_store import Migration
from paperless_ai.vector_store import PaperlessSqliteVecVectorStore from paperless_ai.vector_store import PaperlessSqliteVecVectorStore
from paperless_ai.vector_store import _build_where from paperless_ai.vector_store import _build_where
from paperless_ai.vector_store import _pack
DIM = 16 DIM = 16
def make_node( def make_node(
node_id: str, node_id: str,
document_id: str, document_id: int,
*, *,
modified: str = "2026-06-10T00:00:00", modified: str = "2026-06-10T00:00:00",
seed: float = 0.0, seed: float = 0.0,
@@ -59,13 +61,13 @@ def _query(
) )
def _eq_filter(key: str, value: str): def _eq_filter(key: str, value: int):
return MetadataFilters( return MetadataFilters(
filters=[MetadataFilter(key=key, operator=FilterOperator.EQ, value=value)], filters=[MetadataFilter(key=key, operator=FilterOperator.EQ, value=value)],
) )
def _in_filter(document_ids: list[str]): def _in_filter(document_ids: list[int]):
return MetadataFilters( return MetadataFilters(
filters=[ filters=[
MetadataFilter( MetadataFilter(
@@ -77,7 +79,7 @@ def _in_filter(document_ids: list[str]):
) )
def _ne_filter(document_id: str): def _ne_filter(document_id: int):
return MetadataFilters( return MetadataFilters(
filters=[ filters=[
MetadataFilter( MetadataFilter(
@@ -91,11 +93,11 @@ def _ne_filter(document_id: str):
class TestCrud: class TestCrud:
def test_add_then_query_returns_node(self, store) -> None: def test_add_then_query_returns_node(self, store) -> None:
node = make_node("n1", "1") node = make_node("n1", 1)
assert store.add([node]) == ["n1"] assert store.add([node]) == ["n1"]
result = _query(store, node.embedding, top_k=1) result = _query(store, node.embedding, top_k=1)
assert result.ids == ["n1"] assert result.ids == ["n1"]
assert result.nodes[0].metadata["document_id"] == "1" assert result.nodes[0].metadata["document_id"] == 1
# cosine distance of the identical vector is 0 -> similarity 1 # cosine distance of the identical vector is 0 -> similarity 1
assert result.similarities[0] == pytest.approx(1.0) assert result.similarities[0] == pytest.approx(1.0)
@@ -108,58 +110,58 @@ class TestCrud:
assert not store.table_exists() assert not store.table_exists()
def test_delete_removes_all_chunks_of_document(self, store) -> None: def test_delete_removes_all_chunks_of_document(self, store) -> None:
store.add([make_node("a1", "1"), make_node("a2", "1"), make_node("b1", "2")]) store.add([make_node("a1", 1), make_node("a2", 1), make_node("b1", 2)])
store.delete("1") store.delete(1)
result = _query(store, [0.0] * DIM, top_k=10) result = _query(store, [0.0] * DIM, top_k=10)
assert result.ids == ["b1"] assert result.ids == ["b1"]
def test_query_with_in_filter_scopes_results(self, store) -> None: def test_query_with_in_filter_scopes_results(self, store) -> None:
store.add( store.add(
[ [
make_node("a1", "1", seed=0.0), make_node("a1", 1, seed=0.0),
make_node("b1", "2", seed=1.0), make_node("b1", 2, seed=1.0),
make_node("c1", "3", seed=2.0), make_node("c1", 3, seed=2.0),
], ],
) )
result = _query(store, [0.0] * DIM, top_k=10, filters=_in_filter(["2", "3"])) result = _query(store, [0.0] * DIM, top_k=10, filters=_in_filter([2, 3]))
assert sorted(result.ids) == ["b1", "c1"] assert sorted(result.ids) == ["b1", "c1"]
def test_query_respects_top_k_with_filter(self, store) -> None: def test_query_respects_top_k_with_filter(self, store) -> None:
# k semantics: global top-k even with IN filters (document_id is a # k semantics: global top-k even with IN filters (document_id is a
# metadata column, not a partition key -- see design doc). # metadata column, not a partition key -- see design doc).
store.add( store.add(
[make_node(f"n{i}", str(i % 4), seed=float(i)) for i in range(12)], [make_node(f"n{i}", i % 4, seed=float(i)) for i in range(12)],
) )
result = _query( result = _query(
store, store,
[0.0] * DIM, [0.0] * DIM,
top_k=3, top_k=3,
filters=_in_filter(["0", "1", "2", "3"]), filters=_in_filter([0, 1, 2, 3]),
) )
assert len(result.ids) == 3 assert len(result.ids) == 3
assert result.similarities == sorted(result.similarities, reverse=True) assert result.similarities == sorted(result.similarities, reverse=True)
def test_get_nodes_filter_and_empty_paths(self, store) -> None: def test_get_nodes_filter_and_empty_paths(self, store) -> None:
assert store.get_nodes(filters=_in_filter(["1"])) == [] # no table yet assert store.get_nodes(filters=_in_filter([1])) == [] # no table yet
store.add([make_node("a1", "1"), make_node("b1", "2")]) store.add([make_node("a1", 1), make_node("b1", 2)])
nodes = store.get_nodes(filters=_in_filter(["1"])) nodes = store.get_nodes(filters=_in_filter([1]))
assert [n.node_id for n in nodes] == ["a1"] assert [n.node_id for n in nodes] == ["a1"]
assert nodes[0].embedding is not None assert nodes[0].embedding is not None
assert store.get_nodes(filters=_in_filter(["999"])) == [] assert store.get_nodes(filters=_in_filter([999])) == []
def test_query_with_eq_filter_scopes_results(self, store) -> None: def test_query_with_eq_filter_scopes_results(self, store) -> None:
store.add( store.add(
[ [
make_node("a1", "1", seed=0.0), make_node("a1", 1, seed=0.0),
make_node("b1", "2", seed=1.0), make_node("b1", 2, seed=1.0),
make_node("c1", "3", seed=2.0), make_node("c1", 3, seed=2.0),
], ],
) )
result = _query( result = _query(
store, store,
[0.0] * DIM, [0.0] * DIM,
top_k=10, top_k=10,
filters=_eq_filter("document_id", "2"), filters=_eq_filter("document_id", 2),
) )
assert result.ids == ["b1"] assert result.ids == ["b1"]
@@ -168,7 +170,7 @@ class TestCrud:
store.get_nodes(node_ids=["x"]) store.get_nodes(node_ids=["x"])
def test_fresh_instance_sees_existing_table(self, store, tmp_path: Path) -> None: def test_fresh_instance_sees_existing_table(self, store, tmp_path: Path) -> None:
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reopened: with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reopened:
assert reopened.table_exists() assert reopened.table_exists()
assert reopened.vector_dim() == DIM assert reopened.vector_dim() == DIM
@@ -176,23 +178,98 @@ class TestCrud:
def test_table_exists_and_drop(self, store) -> None: def test_table_exists_and_drop(self, store) -> None:
assert not store.table_exists() assert not store.table_exists()
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
assert store.table_exists() assert store.table_exists()
store.drop_table() store.drop_table()
assert not store.table_exists() assert not store.table_exists()
assert store.vector_dim() is None assert store.vector_dim() is None
def test_document_id_stored_as_integer_in_vec0(
self,
store: PaperlessSqliteVecVectorStore,
) -> None:
"""
GIVEN:
- An empty vector store
WHEN:
- A node is added with an int document_id
THEN:
- vec0's own document_id column holds an INTEGER, not TEXT
"""
store.add([make_node("a1", 1)])
row = store.client.execute(
"SELECT document_id FROM documents WHERE id = 'a1'",
).fetchone()
assert isinstance(row["document_id"], int)
def test_drop_table_clears_modified_times(
self,
store: PaperlessSqliteVecVectorStore,
) -> None:
"""
GIVEN:
- A store with a tracked document's modified time
WHEN:
- drop_table() is called
THEN:
- document_meta and document_chunks are both cleared directly
(asserted against the tables themselves, not via
get_modified_times()/table_exists() -- those short-circuit on
the vec0 table being gone, which drop_table() does first, so
they would pass even if DocumentMetaTable.delete_all()/
DocumentChunksTable.delete_all() were never called)
"""
store.add([make_node("a1", 1)])
store.drop_table()
assert (
store.client.execute(
"SELECT count(*) FROM document_meta",
).fetchone()[0]
== 0
)
assert (
store.client.execute(
"SELECT count(*) FROM document_chunks",
).fetchone()[0]
== 0
)
def test_upsert_document_checks_table_exists_once(
self,
store: PaperlessSqliteVecVectorStore,
mocker: MockerFixture,
) -> None:
"""
GIVEN:
- An existing store with one document already indexed
WHEN:
- upsert_document() replaces that document's chunks
THEN:
- table_exists() is queried at most once per call, not twice
(previously: once via _ensure_table(), once via the separate
`if self.table_exists():` delete-chunks guard)
"""
store.add([make_node("a1", 1)])
# store is a pydantic model, whose __setattr__/__delattr__ reject
# arbitrary instance attributes ("object has no attribute
# 'table_exists'"), so mocker.spy(store, "table_exists") can't
# shadow the method on the instance. Spying on the class works
# (bound method lookup on the instance still resolves through it).
exists_spy = mocker.spy(PaperlessSqliteVecVectorStore, "table_exists")
store.upsert_document(1, [make_node("a2", 1)])
assert exists_spy.call_count == 1
class TestBuildWhere: class TestBuildWhere:
def test_ne_filter_translates_to_not_equal_clause(self) -> None: def test_ne_filter_translates_to_not_equal_clause(self) -> None:
where, params = _build_where(_ne_filter("1")) where, params = _build_where(_ne_filter(1))
assert where == "(document_id != ?)" assert where == "(document_id != ?)"
assert params == ["1"] assert params == [1]
def test_query_with_ne_filter_excludes_matching_document(self, store) -> None: def test_query_with_ne_filter_excludes_matching_document(self, store) -> None:
store.add([make_node("a1", "1"), make_node("b1", "2")]) store.add([make_node("a1", 1), make_node("b1", 2)])
assert sorted( assert sorted(
_query(store, [0.0] * DIM, top_k=5, filters=_ne_filter("1")).ids, _query(store, [0.0] * DIM, top_k=5, filters=_ne_filter(1)).ids,
) == [ ) == [
"b1", "b1",
] ]
@@ -206,7 +283,7 @@ class TestBuildWhere:
MetadataFilter( MetadataFilter(
key="document_id", key="document_id",
operator=FilterOperator.EQ, operator=FilterOperator.EQ,
value="1", value=1,
), ),
], ],
) )
@@ -215,13 +292,13 @@ class TestBuildWhere:
assert params == [] assert params == []
def test_query_with_untranslatable_filter_returns_no_rows(self, store) -> None: def test_query_with_untranslatable_filter_returns_no_rows(self, store) -> None:
store.add([make_node("a1", "1"), make_node("b1", "2")]) store.add([make_node("a1", 1), make_node("b1", 2)])
nested = MetadataFilters( nested = MetadataFilters(
filters=[ filters=[
MetadataFilter( MetadataFilter(
key="document_id", key="document_id",
operator=FilterOperator.EQ, operator=FilterOperator.EQ,
value="1", value=1,
), ),
], ],
) )
@@ -234,19 +311,19 @@ class TestBuildWhere:
class TestUpsert: class TestUpsert:
def test_upsert_replaces_and_prunes_stale_chunks(self, store) -> None: def test_upsert_replaces_and_prunes_stale_chunks(self, store) -> None:
store.add( store.add(
[make_node("d1c1", "1"), make_node("d1c2", "1"), make_node("d2c1", "2")], [make_node("d1c1", 1), make_node("d1c2", 1), make_node("d2c1", 2)],
) )
store.upsert_document("1", [make_node("d1new", "1")]) store.upsert_document(1, [make_node("d1new", 1)])
result = _query(store, [0.0] * DIM, top_k=10) result = _query(store, [0.0] * DIM, top_k=10)
assert sorted(result.ids) == ["d1new", "d2c1"] assert sorted(result.ids) == ["d1new", "d2c1"]
def test_upsert_creates_table_when_missing(self, store) -> None: def test_upsert_creates_table_when_missing(self, store) -> None:
store.upsert_document("1", [make_node("a1", "1")]) store.upsert_document(1, [make_node("a1", 1)])
assert _query(store, [0.0] * DIM, top_k=1).ids == ["a1"] assert _query(store, [0.0] * DIM, top_k=1).ids == ["a1"]
def test_upsert_empty_nodes_removes_document(self, store) -> None: def test_upsert_empty_nodes_removes_document(self, store) -> None:
store.add([make_node("a1", "1"), make_node("b1", "2")]) store.add([make_node("a1", 1), make_node("b1", 2)])
store.upsert_document("1", []) store.upsert_document(1, [])
assert _query(store, [0.0] * DIM, top_k=10).ids == ["b1"] assert _query(store, [0.0] * DIM, top_k=10).ids == ["b1"]
def test_upsert_is_atomic_for_concurrent_readers( def test_upsert_is_atomic_for_concurrent_readers(
@@ -255,16 +332,16 @@ class TestUpsert:
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
"""A second connection must never observe document 1 half-replaced.""" """A second connection must never observe document 1 half-replaced."""
store.add([make_node("a1", "1"), make_node("a2", "1")]) store.add([make_node("a1", 1), make_node("a2", 1)])
with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reader: with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reader:
store.upsert_document("1", [make_node("a3", "1")]) store.upsert_document(1, [make_node("a3", 1)])
ids = [n.node_id for n in reader.get_nodes(filters=_in_filter(["1"]))] ids = [n.node_id for n in reader.get_nodes(filters=_in_filter([1]))]
assert ids == ["a3"] assert ids == ["a3"]
class TestMetadataCoercion: class TestMetadataCoercion:
def test_none_metadata_values_become_empty_strings(self, store) -> None: def test_none_metadata_values_become_empty_strings(self, store) -> None:
node = make_node("a1", "1") node = make_node("a1", 1)
node.metadata["modified"] = None node.metadata["modified"] = None
store.add([node]) # must not raise (vec0 rejects NULL metadata) store.add([node]) # must not raise (vec0 rejects NULL metadata)
assert store.get_modified_times() == {"1": ""} assert store.get_modified_times() == {"1": ""}
@@ -283,7 +360,7 @@ class TestModelNameTracking:
uri=str(tmp_path), uri=str(tmp_path),
embed_model_name="model-a", embed_model_name="model-a",
) as store: ) as store:
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
assert store.stored_model_name() == "model-a" assert store.stored_model_name() == "model-a"
with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reopened: with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reopened:
assert reopened.stored_model_name() == "model-a" assert reopened.stored_model_name() == "model-a"
@@ -294,7 +371,7 @@ class TestModelNameTracking:
embed_model_name="model-a", embed_model_name="model-a",
) as store: ) as store:
assert not store.config_mismatch("anything") # no table yet assert not store.config_mismatch("anything") # no table yet
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
assert not store.config_mismatch("model-a") assert not store.config_mismatch("model-a")
assert store.config_mismatch("model-b") assert store.config_mismatch("model-b")
@@ -303,7 +380,7 @@ class TestModelNameTracking:
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as store: # no model name with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as store: # no model name
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
assert not store.config_mismatch("model-a") assert not store.config_mismatch("model-a")
@@ -314,9 +391,9 @@ class TestGetModifiedTimes:
def test_returns_one_entry_per_document(self, store) -> None: def test_returns_one_entry_per_document(self, store) -> None:
store.add( store.add(
[ [
make_node("a1", "1", modified="2026-01-01T00:00:00"), make_node("a1", 1, modified="2026-01-01T00:00:00"),
make_node("a2", "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"), make_node("b1", 2, modified="2026-02-02T00:00:00"),
], ],
) )
assert store.get_modified_times() == { assert store.get_modified_times() == {
@@ -341,37 +418,35 @@ class TestCompact:
def _churn(self, store, cycles: int) -> None: def _churn(self, store, cycles: int) -> None:
for i in range(cycles): for i in range(cycles):
store.upsert_document( store.upsert_document(
"1", 1,
[make_node(f"gen{i}-{j}", "1", seed=float(j)) for j in range(20)], [make_node(f"gen{i}-{j}", 1, seed=float(j)) for j in range(20)],
) )
def test_compact_noop_below_threshold(self, store) -> None: def test_compact_noop_below_threshold(self, store) -> None:
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
store.compact() store.compact()
assert _query(store, [0.0] * DIM, top_k=1).ids == ["a1"] assert _query(store, [0.0] * DIM, top_k=1).ids == ["a1"]
def test_force_compact_preserves_rows_and_metadata(self, store) -> None: def test_force_compact_preserves_rows_and_metadata(self, store) -> None:
store.add([make_node("a1", "1"), make_node("b1", "2", seed=3.0)]) store.add([make_node("a1", 1), make_node("b1", 2, seed=3.0)])
self._churn(store, 5) self._churn(store, 5)
before = { before = {
n.node_id: n.metadata n.node_id: n.metadata for n in store.get_nodes(filters=_in_filter([1, 2]))
for n in store.get_nodes(filters=_in_filter(["1", "2"]))
} }
store.compact(force=True) store.compact(force=True)
after = { after = {
n.node_id: n.metadata n.node_id: n.metadata for n in store.get_nodes(filters=_in_filter([1, 2]))
for n in store.get_nodes(filters=_in_filter(["1", "2"]))
} }
assert after == before assert after == before
assert self._bloat_ratio(store) == pytest.approx(1.0) assert self._bloat_ratio(store) == pytest.approx(1.0)
# store remains fully usable after the rebuild; use a seed far from all # store remains fully usable after the rebuild; use a seed far from all
# existing nodes (gen4-0..gen4-19 have seeds 0..19) so cosine KNN is # existing nodes (gen4-0..gen4-19 have seeds 0..19) so cosine KNN is
# unambiguous at top_k=1. # unambiguous at top_k=1.
store.upsert_document("3", [make_node("c1", "3", seed=100.0)]) store.upsert_document(3, [make_node("c1", 3, seed=100.0)])
assert "c1" in _query(store, [100.0] * DIM, top_k=1).ids assert "c1" in _query(store, [100.0] * DIM, top_k=1).ids
def test_auto_compact_triggers_on_churn(self, store) -> None: def test_auto_compact_triggers_on_churn(self, store) -> None:
store.add([make_node(f"s{j}", "1", seed=float(j)) for j in range(20)]) store.add([make_node(f"s{j}", 1, seed=float(j)) for j in range(20)])
self._churn(store, 5) self._churn(store, 5)
assert self._bloat_ratio(store) > 2 assert self._bloat_ratio(store) > 2
store.compact() store.compact()
@@ -393,7 +468,7 @@ class TestCompact:
but a concurrent reader keeps -wal/-shm alive, so the cleanup must but a concurrent reader keeps -wal/-shm alive, so the cleanup must
unlink them explicitly (as the structural-migration path does). unlink them explicitly (as the structural-migration path does).
""" """
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
compact_path = str(tmp_path / DB_FILENAME) + ".compact" compact_path = str(tmp_path / DB_FILENAME) + ".compact"
held: list[sqlite3.Connection] = [] held: list[sqlite3.Connection] = []
@@ -429,16 +504,40 @@ class TestCompact:
regression in the streaming loop (dropped tail, off-by-one) surfaces. regression in the streaming loop (dropped tail, off-by-one) surfaces.
""" """
monkeypatch.setattr("paperless_ai.vector_store.COMPACT_BATCH_SIZE", 3) monkeypatch.setattr("paperless_ai.vector_store.COMPACT_BATCH_SIZE", 3)
store.add([make_node(f"n{i}", "1", seed=float(i)) for i in range(10)]) store.add([make_node(f"n{i}", 1, seed=float(i)) for i in range(10)])
store.compact(force=True) store.compact(force=True)
ids = {n.node_id for n in store.get_nodes(filters=_in_filter(["1"]))} ids = {n.node_id for n in store.get_nodes(filters=_in_filter([1]))}
assert ids == {f"n{i}" for i in range(10)} assert ids == {f"n{i}" for i in range(10)}
assert self._bloat_ratio(store) == pytest.approx(1.0) assert self._bloat_ratio(store) == pytest.approx(1.0)
def test_force_compact_preserves_modified_times(
self,
store: PaperlessSqliteVecVectorStore,
) -> None:
"""
GIVEN:
- A store with documents whose modified times are tracked
WHEN:
- compact(force=True) rebuilds the database file
THEN:
- get_modified_times() still returns every document's value
(document_meta must be copied across the file-swap, not just
the vec0 rows)
"""
store.add(
[
make_node("a1", 1, modified="2026-01-01T00:00:00"),
make_node("b1", 2, modified="2026-02-02T00:00:00"),
],
)
before = store.get_modified_times()
store.compact(force=True)
assert store.get_modified_times() == before
class TestDbFile: class TestDbFile:
def test_single_db_file_in_index_dir(self, store, tmp_path: Path) -> None: def test_single_db_file_in_index_dir(self, store, tmp_path: Path) -> None:
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
assert (tmp_path / DB_FILENAME).exists() assert (tmp_path / DB_FILENAME).exists()
def test_wal_mode_enabled(self, store) -> None: def test_wal_mode_enabled(self, store) -> None:
@@ -448,7 +547,16 @@ class TestDbFile:
class TestMigrations: class TestMigrations:
"""Tests for the schema migration machinery.""" """Tests for the schema migration machinery.
These tests exercise check_and_run_migrations()'s generic dispatch logic
(structural vs. re-embed, version-boundary stopping) using ad hoc test
migrations layered on top of SCHEMA_VERSION -- distinct from
TestV1ToV2Migration, which exercises the real, frozen m0001_v1_to_v2
migration. Test migrations use version numbers starting at
SCHEMA_VERSION (2) and above so they never collide with the real
from_version=1/to_version=2 migration already registered in MIGRATIONS.
"""
def _schema_version(self, store: PaperlessSqliteVecVectorStore) -> int | None: def _schema_version(self, store: PaperlessSqliteVecVectorStore) -> int | None:
row = store.client.execute( row = store.client.execute(
@@ -457,21 +565,21 @@ class TestMigrations:
return int(row[0]) if row else None return int(row[0]) if row else None
def test_new_table_records_schema_version(self, store) -> None: def test_new_table_records_schema_version(self, store) -> None:
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
assert self._schema_version(store) == SCHEMA_VERSION assert self._schema_version(store) == SCHEMA_VERSION
def test_check_migrations_no_table_returns_false(self, store) -> None: def test_check_migrations_no_table_returns_false(self, store) -> None:
assert store.check_and_run_migrations() is False assert store.check_and_run_migrations() is False
def test_check_migrations_current_version_returns_false(self, store) -> None: def test_check_migrations_current_version_returns_false(self, store) -> None:
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
assert store.check_and_run_migrations() is False assert store.check_and_run_migrations() is False
def test_reembed_migration_returns_true(self, store, tmp_path: Path) -> None: def test_reembed_migration_returns_true(self, store, tmp_path: Path) -> None:
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
migration = Migration( migration = Migration(
from_version=1, from_version=SCHEMA_VERSION,
to_version=2, to_version=SCHEMA_VERSION + 1,
kind="re-embed", kind="re-embed",
description="test re-embed", description="test re-embed",
) )
@@ -480,7 +588,7 @@ class TestMigrations:
from paperless_ai import vector_store as vs_mod from paperless_ai import vector_store as vs_mod
original = vs_mod.SCHEMA_VERSION original = vs_mod.SCHEMA_VERSION
vs_mod.SCHEMA_VERSION = 2 vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 1
result = store.check_and_run_migrations() result = store.check_and_run_migrations()
finally: finally:
MIGRATIONS.remove(migration) MIGRATIONS.remove(migration)
@@ -492,7 +600,7 @@ class TestMigrations:
store, store,
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
store.add([make_node("a1", "1"), make_node("b1", "2")]) store.add([make_node("a1", 1), make_node("b1", 2)])
def apply( def apply(
src: sqlite3.Connection, src: sqlite3.Connection,
@@ -511,7 +619,7 @@ class TestMigrations:
(str(dim),), (str(dim),),
) )
rows = src.execute( rows = src.execute(
"SELECT id, document_id, modified, node_content, embedding " "SELECT id, document_id, node_content, embedding "
f"FROM {DEFAULT_TABLE_NAME}", f"FROM {DEFAULT_TABLE_NAME}",
).fetchall() ).fetchall()
dst.execute("BEGIN IMMEDIATE") dst.execute("BEGIN IMMEDIATE")
@@ -522,8 +630,8 @@ class TestMigrations:
[ [
( (
r["id"], r["id"],
r["document_id"], str(r["document_id"]),
r["modified"], "",
r["node_content"], r["node_content"],
bytes(r["embedding"]), bytes(r["embedding"]),
) )
@@ -538,8 +646,8 @@ class TestMigrations:
dst.execute("COMMIT") dst.execute("COMMIT")
migration = Migration( migration = Migration(
from_version=1, from_version=SCHEMA_VERSION,
to_version=2, to_version=SCHEMA_VERSION + 1,
kind="structural", kind="structural",
description="test structural", description="test structural",
apply=apply, apply=apply,
@@ -549,28 +657,29 @@ class TestMigrations:
from paperless_ai import vector_store as vs_mod from paperless_ai import vector_store as vs_mod
original = vs_mod.SCHEMA_VERSION original = vs_mod.SCHEMA_VERSION
vs_mod.SCHEMA_VERSION = 2 vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 1
result = store.check_and_run_migrations() result = store.check_and_run_migrations()
finally: finally:
MIGRATIONS.remove(migration) MIGRATIONS.remove(migration)
vs_mod.SCHEMA_VERSION = original vs_mod.SCHEMA_VERSION = original
assert result is False assert result is False
assert self._schema_version(store) == 2 assert self._schema_version(store) == SCHEMA_VERSION + 1
ids = {n.node_id for n in store.get_nodes()} ids = {n.node_id for n in store.get_nodes()}
assert ids == {"a1", "b1"} assert ids == {"a1", "b1"}
def test_compact_preserves_schema_version(self, store) -> None: def test_compact_preserves_schema_version(self, store) -> None:
store.add([make_node("a1", "1")]) store.add([make_node("a1", 1)])
assert self._schema_version(store) == SCHEMA_VERSION assert self._schema_version(store) == SCHEMA_VERSION
store.compact(force=True) store.compact(force=True)
assert self._schema_version(store) == SCHEMA_VERSION assert self._schema_version(store) == SCHEMA_VERSION
def test_stop_at_reembed_boundary(self, store) -> None: def test_stop_at_reembed_boundary(self, store) -> None:
# Registry: structural v2, re-embed v3, structural v4. # Registry: structural v(N+1), re-embed v(N+2), structural v(N+3),
# Only v2 should apply; the re-embed boundary must stop execution # where N = SCHEMA_VERSION. Only v(N+1) should apply; the re-embed
# before v4 runs, and the stored version must stay at 2. # boundary must stop execution before v(N+3) runs, and the stored
store.add([make_node("a1", "1"), make_node("b1", "2")]) # version must stay at N+1.
store.add([make_node("a1", 1), make_node("b1", 2)])
def copy_apply( def copy_apply(
src: sqlite3.Connection, src: sqlite3.Connection,
@@ -589,7 +698,7 @@ class TestMigrations:
(str(dim),), (str(dim),),
) )
rows = src.execute( rows = src.execute(
"SELECT id, document_id, modified, node_content, embedding " "SELECT id, document_id, node_content, embedding "
f"FROM {DEFAULT_TABLE_NAME}", f"FROM {DEFAULT_TABLE_NAME}",
).fetchall() ).fetchall()
dst.execute("BEGIN IMMEDIATE") dst.execute("BEGIN IMMEDIATE")
@@ -600,8 +709,8 @@ class TestMigrations:
[ [
( (
r["id"], r["id"],
r["document_id"], str(r["document_id"]),
r["modified"], "",
r["node_content"], r["node_content"],
bytes(r["embedding"]), bytes(r["embedding"]),
) )
@@ -612,23 +721,23 @@ class TestMigrations:
migrations = [ migrations = [
Migration( Migration(
from_version=1, from_version=SCHEMA_VERSION,
to_version=2, to_version=SCHEMA_VERSION + 1,
kind="structural", kind="structural",
description="v2 structural", description="v(N+1) structural",
apply=copy_apply, apply=copy_apply,
), ),
Migration( Migration(
from_version=2, from_version=SCHEMA_VERSION + 1,
to_version=3, to_version=SCHEMA_VERSION + 2,
kind="re-embed", kind="re-embed",
description="v3 re-embed boundary", description="v(N+2) re-embed boundary",
), ),
Migration( Migration(
from_version=3, from_version=SCHEMA_VERSION + 2,
to_version=4, to_version=SCHEMA_VERSION + 3,
kind="structural", kind="structural",
description="v4 structural - must not run", description="v(N+3) structural - must not run",
apply=copy_apply, apply=copy_apply,
), ),
] ]
@@ -637,7 +746,7 @@ class TestMigrations:
from paperless_ai import vector_store as vs_mod from paperless_ai import vector_store as vs_mod
original = vs_mod.SCHEMA_VERSION original = vs_mod.SCHEMA_VERSION
vs_mod.SCHEMA_VERSION = 4 vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 3
result = store.check_and_run_migrations() result = store.check_and_run_migrations()
finally: finally:
for m in migrations: for m in migrations:
@@ -645,4 +754,237 @@ class TestMigrations:
vs_mod.SCHEMA_VERSION = original vs_mod.SCHEMA_VERSION = original
assert result is True assert result is True
assert self._schema_version(store) == 2 assert self._schema_version(store) == SCHEMA_VERSION + 1
def test_has_pending_migration_false_when_no_table(
self,
store: PaperlessSqliteVecVectorStore,
) -> None:
"""
GIVEN:
- A vector store with no table created yet
WHEN:
- has_pending_migration() is checked
THEN:
- False is returned (nothing to migrate before anything exists)
"""
assert store.has_pending_migration() is False
def test_has_pending_migration_false_at_current_version(
self,
store: PaperlessSqliteVecVectorStore,
) -> None:
"""
GIVEN:
- A store at the current SCHEMA_VERSION
WHEN:
- has_pending_migration() is checked
THEN:
- False is returned
"""
store.add([make_node("a1", 1)])
assert store.has_pending_migration() is False
def test_has_pending_migration_true_when_behind(
self,
store: PaperlessSqliteVecVectorStore,
) -> None:
"""
GIVEN:
- A store whose schema_version has been forced behind SCHEMA_VERSION
WHEN:
- has_pending_migration() is checked
THEN:
- True is returned
"""
store.add([make_node("a1", 1)])
store.client.execute(
"UPDATE index_meta SET value = '0' WHERE key = 'schema_version'",
)
assert store.has_pending_migration() is True
class TestV1ToV2Migration:
"""m0001_v1_to_v2 migrates a real, historically-shaped v1 store. The
fixture below is a literal, hardcoded v1 DDL string -- NOT derived from
any current code -- so this test keeps testing the actual historical
shape even if vector_store.py's "current" schema changes again later.
"""
def _build_v1_store(self, db_path: str, dim: int) -> None:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
conn.enable_load_extension(True) # noqa: FBT003
import sqlite_vec
sqlite_vec.load(conn)
conn.enable_load_extension(False) # noqa: FBT003
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute(
"CREATE TABLE IF NOT EXISTS index_meta (key TEXT PRIMARY KEY, value TEXT)",
)
conn.execute( # nosemgrep
"CREATE VIRTUAL TABLE documents USING vec0("
"id TEXT PRIMARY KEY, document_id TEXT, modified TEXT,"
f" +node_content TEXT, embedding float[{dim}] distance_metric=cosine"
")",
)
conn.execute(
"INSERT INTO index_meta (key, value) VALUES ('dim', ?)",
(str(dim),),
)
conn.execute(
"INSERT INTO index_meta (key, value) VALUES ('schema_version', '1')",
)
conn.execute(
"INSERT INTO index_meta (key, value) VALUES ('embed_model', 'model-a')",
)
rows = [
("c1", "1", "2026-01-01T00:00:00", '{"text": "a"}', _pack([0.1] * dim)),
("c2", "1", "2026-01-01T00:00:00", '{"text": "b"}', _pack([0.2] * dim)),
("c3", "2", "2026-02-02T00:00:00", '{"text": "c"}', _pack([0.3] * dim)),
]
conn.executemany(
"INSERT INTO documents (id, document_id, modified, node_content, embedding)"
" VALUES (?, ?, ?, ?, ?)",
rows,
)
conn.execute(
"INSERT INTO index_meta (key, value) VALUES ('total_inserts', '3')",
)
conn.commit()
conn.close()
def test_migration_converts_v1_store_to_v2(self, tmp_path: Path) -> None:
"""
GIVEN:
- A real v1-shaped store (TEXT document_id, modified inline in
vec0, no document_chunks/document_meta) built from a literal,
hardcoded historical DDL
WHEN:
- A PaperlessSqliteVecVectorStore is opened against it
THEN:
- schema_version becomes 2, document_id values become int,
document_chunks/document_meta are backfilled once per chunk/
document respectively, and dim/embed_model survive
"""
db_dir = tmp_path
self._build_v1_store(str(db_dir / DB_FILENAME), dim=16)
with PaperlessSqliteVecVectorStore(uri=str(db_dir)) as store:
assert store.check_and_run_migrations() is False
row = store.client.execute(
"SELECT value FROM index_meta WHERE key = 'schema_version'",
).fetchone()
assert int(row["value"]) == 2
doc_id_row = store.client.execute(
"SELECT document_id FROM documents WHERE id = 'c1'",
).fetchone()
assert isinstance(doc_id_row["document_id"], int)
assert doc_id_row["document_id"] == 1
chunk_ids = sorted(
r["chunk_id"]
for r in store.client.execute(
"SELECT chunk_id FROM document_chunks",
).fetchall()
)
assert chunk_ids == ["c1", "c2", "c3"]
assert store.get_modified_times() == {
"1": "2026-01-01T00:00:00",
"2": "2026-02-02T00:00:00",
}
assert store.stored_model_name() == "model-a"
assert store.vector_dim() == 16
def test_migration_raises_on_malformed_document_id(
self,
tmp_path: Path,
) -> None:
"""
GIVEN:
- A v1-shaped store with a corrupted, non-integer document_id
value on one row
WHEN:
- The migration runs
THEN:
- A ValueError is raised (fail loudly, no silent data loss) --
this matches the rest of vector_store.py, which has no
precedent for silently skipping malformed rows
"""
db_dir = tmp_path
self._build_v1_store(str(db_dir / DB_FILENAME), dim=16)
import sqlite_vec
conn = sqlite3.connect(str(db_dir / DB_FILENAME))
conn.enable_load_extension(True) # noqa: FBT003
sqlite_vec.load(conn)
conn.enable_load_extension(False) # noqa: FBT003
conn.execute(
"UPDATE documents SET document_id = 'not-an-int' WHERE id = 'c1'",
)
conn.commit()
conn.close()
with (
pytest.raises(ValueError),
PaperlessSqliteVecVectorStore(uri=str(db_dir)) as store,
):
store.check_and_run_migrations()
def test_migration_never_delegates_to_current_schema_helpers(
self,
tmp_path: Path,
mocker: MockerFixture,
) -> None:
"""
GIVEN:
- A real v1-shaped store
WHEN:
- The migration runs, with DocumentChunksTable.create/
DocumentMetaTable.create/_create_vec_table spied on
THEN:
- None of those "current schema" helpers are ever called during
the migration -- it must freeze its own historical DDL, per
the DDL-freezing rule (see spec), so a future schema bump
can't silently corrupt this migration's output
"""
db_dir = tmp_path
self._build_v1_store(str(db_dir / DB_FILENAME), dim=16)
from paperless_ai.tables import DocumentChunksTable
from paperless_ai.tables import DocumentMetaTable
chunks_create_spy = mocker.spy(DocumentChunksTable, "create")
meta_create_spy = mocker.spy(DocumentMetaTable, "create")
vec_table_spy = mocker.spy(
PaperlessSqliteVecVectorStore,
"_create_vec_table",
)
with PaperlessSqliteVecVectorStore(uri=str(db_dir)) as store:
store.check_and_run_migrations()
# _open_connection() legitimately calls create() three times across
# a structural migration: once for the store's own live connection
# (construction), once for the migration's temp rebuild file
# (_rebuild_file), and once more when _swap_in_compact() reopens the
# swapped-in file as self._conn. What matters is that
# m0001_v1_to_v2's apply() itself never calls these directly, which
# a source-text check alone can't prove (it can't tell "mentioned in
# a comment" from "actually called", and is trivially evadable by
# importing a symbol under an alias). Asserting the exact call count
# instead: exactly 3 calls to each create() (all from
# _open_connection, never a 4th from inside apply()), and zero calls
# to _create_vec_table (neither _open_connection nor apply() calls
# it -- apply() freezes its own literal CREATE VIRTUAL TABLE DDL
# instead).
assert chunks_create_spy.call_count == 3
assert meta_create_spy.call_count == 3
assert vec_table_spy.call_count == 0
# Cheap secondary signal, kept alongside the spy assertions above
# (not in place of them): the migration module's source should never
# even mention these "current schema" helpers by name.
import inspect
from paperless_ai.migrations import m0001_v1_to_v2
source = inspect.getsource(m0001_v1_to_v2)
assert "DocumentChunksTable.create" not in source
assert "DocumentMetaTable.create" not in source
assert "_create_vec_table(" not in source
+274 -181
View File
@@ -2,16 +2,13 @@ import json
import logging import logging
import sqlite3 import sqlite3
import struct import struct
from collections.abc import Callable
from collections.abc import Iterator from collections.abc import Iterator
from collections.abc import Sequence from collections.abc import Sequence
from contextlib import contextmanager from contextlib import contextmanager
from dataclasses import dataclass
from dataclasses import field
from pathlib import Path from pathlib import Path
from types import TracebackType from types import TracebackType
from typing import Any from typing import Any
from typing import Literal from typing import NamedTuple
import sqlite_vec import sqlite_vec
from llama_index.core.bridge.pydantic import PrivateAttr from llama_index.core.bridge.pydantic import PrivateAttr
@@ -26,6 +23,14 @@ from llama_index.core.vector_stores.types import VectorStoreQueryResult
from llama_index.core.vector_stores.utils import metadata_dict_to_node from llama_index.core.vector_stores.utils import metadata_dict_to_node
from llama_index.core.vector_stores.utils import node_to_metadata_dict from llama_index.core.vector_stores.utils import node_to_metadata_dict
from paperless_ai.migrations import MIGRATIONS
from paperless_ai.migrations import Migration
from paperless_ai.tables import ChunkRow
from paperless_ai.tables import DocumentChunksTable
from paperless_ai.tables import DocumentMetaRow
from paperless_ai.tables import DocumentMetaTable
from paperless_ai.tables import IndexMetaTable
logger = logging.getLogger("paperless_ai.vector_store") logger = logging.getLogger("paperless_ai.vector_store")
DB_FILENAME = "llmindex.db" DB_FILENAME = "llmindex.db"
@@ -34,7 +39,7 @@ DEFAULT_TABLE_NAME = "documents"
# Current schema version. Written to index_meta at table creation and bumped # Current schema version. Written to index_meta at table creation and bumped
# whenever a Migration is added to MIGRATIONS. check_and_run_migrations() uses # whenever a Migration is added to MIGRATIONS. check_and_run_migrations() uses
# this to decide which migrations to run on an existing store. # this to decide which migrations to run on an existing store.
SCHEMA_VERSION = 1 SCHEMA_VERSION = 2
# compact(): rebuild when the cumulative rowid count exceeds this multiple of # compact(): rebuild when the cumulative rowid count exceeds this multiple of
# the live row count. DELETEs on vec0 tables never reclaim space (upstream # the live row count. DELETEs on vec0 tables never reclaim space (upstream
@@ -49,40 +54,23 @@ COMPACT_BATCH_SIZE = 500
# Filterable vec0 metadata columns. _build_where() only ever receives filter # Filterable vec0 metadata columns. _build_where() only ever receives filter
# keys we construct ourselves, but allowlisting keeps SQL identifiers safe by # keys we construct ourselves, but allowlisting keeps SQL identifiers safe by
# construction. # construction. "modified" is not here: it is never filtered on, and as of
_FILTER_COLUMNS = frozenset({"document_id", "modified"}) # schema v2 it isn't even a vec0 column anymore (see document_meta).
_FILTER_COLUMNS = frozenset({"document_id"})
@dataclass class _Row(NamedTuple):
class Migration: """One node, ready to write. ``modified`` is not a vec0 column (see
"""A schema migration for the sqlite-vec vector store. document_meta) -- it rides along here because every row-producing call
site needs both the vec0 insert values and the document_meta upsert
kind="structural": rows are copied into a new-schema file with no value from the same node.
re-embedding needed. Supply ``apply(src_conn, dst_conn, dim)`` which
must create the vec0 table in ``dst_conn``, copy all rows from
``src_conn``, and write ``dim`` / ``embed_model`` / ``total_inserts`` to
``dst_conn``'s ``index_meta``. ``schema_version`` is written by the
migration runner after ``apply`` returns.
kind="re-embed": the new schema requires fresh embeddings.
``check_and_run_migrations()`` returns True when it encounters one of
these so the caller can force a full rebuild (which recreates the table
at the current SCHEMA_VERSION).
""" """
from_version: int chunk_id: str
to_version: int document_id: int
kind: Literal["structural", "re-embed"] modified: str
description: str node_content: str
apply: Callable[[sqlite3.Connection, sqlite3.Connection, int], None] | None = field( embedding: bytes
default=None,
repr=False,
)
# Registry of all schema migrations in order. Empty at v1 -- this is the
# baseline. Add entries here (and bump SCHEMA_VERSION) when the schema changes.
MIGRATIONS: list[Migration] = []
def _pack(embedding: Sequence[float]) -> bytes: def _pack(embedding: Sequence[float]) -> bytes:
@@ -93,14 +81,30 @@ def _unpack(blob: bytes) -> list[float]:
return list(struct.unpack(f"{len(blob) // 4}f", blob)) return list(struct.unpack(f"{len(blob) // 4}f", blob))
def _build_where(filters: MetadataFilters | None) -> tuple[str, list[str]]: _INSERT = (
"""Translate the EQ / IN / NE filters we use into a parameterized SQL clause "INSERT INTO "
on vec0 metadata columns. Returns ("", []) when there is nothing to filter. + DEFAULT_TABLE_NAME
+ " (id, document_id, node_content, embedding) VALUES (?, ?, ?, ?)"
)
def _vec0_params(rows: list[_Row]) -> list[tuple[str, int, str, bytes]]:
"""``rows``, minus the ``modified`` field vec0 no longer stores."""
return [(r.chunk_id, r.document_id, r.node_content, r.embedding) for r in rows]
def _build_where(filters: MetadataFilters | None) -> tuple[str, list[int]]:
"""Translate the EQ / IN / NE filters we use into a parameterized SQL
clause on vec0 metadata columns. Returns ("", []) when there is nothing
to filter. document_id is vec0's only filterable column and is INTEGER;
every value is coerced via int() here so callers (which today still pass
strings in places, e.g. indexing.py's MetadataFilter construction) don't
have to be individually correct -- vec0 doesn't coerce types itself.
""" """
if filters is None or not filters.filters: if filters is None or not filters.filters:
return "", [] return "", []
clauses: list[str] = [] clauses: list[str] = []
params: list[str] = [] params: list[int] = []
for f in filters.filters: for f in filters.filters:
# filters.filters is Union[MetadataFilter, ExactMatchFilter, MetadataFilters]; # filters.filters is Union[MetadataFilter, ExactMatchFilter, MetadataFilters];
# we only build MetadataFilter entries, so skip anything else at runtime. # we only build MetadataFilter entries, so skip anything else at runtime.
@@ -109,7 +113,7 @@ def _build_where(filters: MetadataFilters | None) -> tuple[str, list[str]]:
if f.key not in _FILTER_COLUMNS: # pragma: no cover - we build the keys if f.key not in _FILTER_COLUMNS: # pragma: no cover - we build the keys
raise NotImplementedError(f"Unsupported filter column: {f.key}") raise NotImplementedError(f"Unsupported filter column: {f.key}")
if f.operator == FilterOperator.IN: if f.operator == FilterOperator.IN:
values = [str(v) for v in f.value] # type: ignore[union-attr] # value is list when operator is IN values = [int(v) for v in f.value] # type: ignore[union-attr]
if not values: # pragma: no cover if not values: # pragma: no cover
clauses.append("1 = 0") clauses.append("1 = 0")
continue continue
@@ -118,10 +122,10 @@ def _build_where(filters: MetadataFilters | None) -> tuple[str, list[str]]:
params.extend(values) params.extend(values)
elif f.operator == FilterOperator.EQ: elif f.operator == FilterOperator.EQ:
clauses.append(f"{f.key} = ?") clauses.append(f"{f.key} = ?")
params.append(str(f.value)) params.append(int(f.value))
elif f.operator == FilterOperator.NE: elif f.operator == FilterOperator.NE:
clauses.append(f"{f.key} != ?") clauses.append(f"{f.key} != ?")
params.append(str(f.value)) params.append(int(f.value))
else: # pragma: no cover - we only ever build EQ/IN/NE filters else: # pragma: no cover - we only ever build EQ/IN/NE filters
raise NotImplementedError(f"Unsupported filter operator: {f.operator}") raise NotImplementedError(f"Unsupported filter operator: {f.operator}")
if not clauses: if not clauses:
@@ -186,9 +190,21 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
conn.enable_load_extension(False) # noqa: FBT003 conn.enable_load_extension(False) # noqa: FBT003
conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL") conn.execute("PRAGMA synchronous=NORMAL")
conn.execute( IndexMetaTable.create(conn)
"CREATE TABLE IF NOT EXISTS index_meta (key TEXT PRIMARY KEY, value TEXT)", # vec0 metadata columns only get an efficient lookup path inside a
) # KNN (MATCH) query; a plain `WHERE document_id = ?` is a full table
# scan regardless of index size. This plain, indexed table is how
# delete()/upsert_document() find a document's chunk ids without
# that scan.
DocumentChunksTable.create(conn)
# 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.
# 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().
DocumentMetaTable.create(conn)
return conn return conn
@property @property
@@ -223,24 +239,6 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
else: else:
self._conn.execute("COMMIT") self._conn.execute("COMMIT")
def _meta_get(self, key: str) -> str | None:
row = self._conn.execute(
"SELECT value FROM index_meta WHERE key = ?",
(key,),
).fetchone()
return row["value"] if row else None
@staticmethod
def _meta_set_on(conn: sqlite3.Connection, key: str, value: str) -> None:
conn.execute(
"INSERT INTO index_meta (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(key, value),
)
def _meta_set(self, key: str, value: str) -> None:
self._meta_set_on(self._conn, key, value)
def table_exists(self) -> bool: def table_exists(self) -> bool:
return ( return (
self._conn.execute( self._conn.execute(
@@ -253,18 +251,19 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
def vector_dim(self) -> int | None: def vector_dim(self) -> int | None:
if not self.table_exists(): if not self.table_exists():
return None return None
value = self._meta_get("dim") return IndexMetaTable.get_dim(self._conn)
return int(value) if value else None
def drop_table(self) -> None: def drop_table(self) -> None:
self._conn.execute("DROP TABLE IF EXISTS " + DEFAULT_TABLE_NAME) self._conn.execute("DROP TABLE IF EXISTS " + DEFAULT_TABLE_NAME)
self._conn.execute("DELETE FROM index_meta") self._conn.execute("DELETE FROM index_meta")
DocumentChunksTable.delete_all(self._conn)
DocumentMetaTable.delete_all(self._conn)
def stored_model_name(self) -> str | None: def stored_model_name(self) -> str | None:
"""Return the embedding model name recorded at table creation, or None.""" """Return the embedding model name recorded at table creation, or None."""
if not self.table_exists(): if not self.table_exists():
return None return None
return self._meta_get("embed_model") return IndexMetaTable.get_embed_model(self._conn)
def config_mismatch(self, model_name: str) -> bool: def config_mismatch(self, model_name: str) -> bool:
"""True when the stored model name differs from ``model_name``. """True when the stored model name differs from ``model_name``.
@@ -282,14 +281,17 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
# document_id is deliberately a metadata column, NOT a partition key: # document_id is deliberately a metadata column, NOT a partition key:
# partition keys change KNN `k` to per-partition semantics under IN # partition keys change KNN `k` to per-partition semantics under IN
# filters (asg017/sqlite-vec#142); metadata columns give a correct # filters (asg017/sqlite-vec#142); metadata columns give a correct
# global top-k. # global top-k. INTEGER (not TEXT, as in schema v1): EQ/NE/IN
# comparisons become a native i64 array compare instead of per-row
# strncmp against a 16-byte text view, and this drops the unused
# metadatatext shadow table TEXT columns carry. modified is not a
# column here at all as of v2 -- see document_meta.
conn.execute( # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query conn.execute( # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
"CREATE VIRTUAL TABLE " "CREATE VIRTUAL TABLE "
+ DEFAULT_TABLE_NAME + DEFAULT_TABLE_NAME
+ " USING vec0(" + " USING vec0("
+ "id TEXT PRIMARY KEY," + "id TEXT PRIMARY KEY,"
+ " document_id TEXT," + " document_id INTEGER,"
+ " modified TEXT,"
+ " +node_content TEXT," + " +node_content TEXT,"
+ " embedding float[" + " embedding float["
+ str(int(dim)) + str(int(dim))
@@ -299,37 +301,70 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
def _create_table(self, dim: int) -> None: def _create_table(self, dim: int) -> None:
self._create_vec_table(self._conn, dim) self._create_vec_table(self._conn, dim)
self._meta_set("dim", str(dim)) IndexMetaTable.set_dim(self._conn, dim)
self._meta_set("schema_version", str(SCHEMA_VERSION)) IndexMetaTable.set_schema_version(self._conn, SCHEMA_VERSION)
if self._embed_model_name: if self._embed_model_name:
self._meta_set("embed_model", self._embed_model_name) IndexMetaTable.set_embed_model(self._conn, self._embed_model_name)
def _ensure_table(self, dim: int) -> None: def _ensure_table(self, dim: int, *, table_exists: bool) -> None:
if not self.table_exists(): if not table_exists:
self._create_table(dim) 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( meta = node_to_metadata_dict(
node, node,
remove_text=False, remove_text=False,
flat_metadata=self.flat_metadata, flat_metadata=self.flat_metadata,
) )
# 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") document_id = node.ref_doc_id or node.metadata.get("document_id")
return ( return _Row(
node.node_id, chunk_id=node.node_id,
str(document_id or ""), document_id=int(document_id),
str(node.metadata.get("modified") or ""), modified=str(node.metadata.get("modified") or ""),
json.dumps(meta), node_content=json.dumps(meta),
_pack(node.get_embedding()), embedding=_pack(node.get_embedding()),
) )
_INSERT = ( def _index_chunks(self, rows: list[_Row]) -> None:
"INSERT INTO " """Record each row's (chunk_id, document_id) in document_chunks, and
+ DEFAULT_TABLE_NAME each row's (document_id, modified) in document_meta -- deduped
+ " (id, document_id, modified, node_content, embedding) VALUES (?, ?, ?, ?, ?)" within the batch, since every chunk of a document shares the same
modified value -- kept in lockstep with every insert into the vec0
table.
"""
DocumentChunksTable.insert_many(
self._conn,
(ChunkRow(r.chunk_id, r.document_id) for r in rows),
) )
modified_by_document = {r.document_id: r.modified for r in rows}
DocumentMetaTable.upsert_many(
self._conn,
(
DocumentMetaRow(doc_id, mod)
for doc_id, mod in modified_by_document.items()
),
)
def _delete_chunks_by_document_id(self, document_id: int) -> None:
"""Delete all of a document's chunks via point-deletes on `id`.
vec0 has no efficient lookup on the document_id metadata column
outside a KNN query, so a plain `DELETE ... WHERE document_id = ?`
is a full table scan regardless of index size. Looking the chunk
ids up in document_chunks first (a real indexed lookup) and
deleting each by its `id` primary key instead turns that scan into
a handful of O(1) point deletes.
"""
chunk_ids = DocumentChunksTable.chunk_ids_for_document(
self._conn,
document_id,
)
self._conn.executemany(
"DELETE FROM " + DEFAULT_TABLE_NAME + " WHERE id = ?",
[(chunk_id,) for chunk_id in chunk_ids],
)
DocumentChunksTable.delete_for_document(self._conn, document_id)
DocumentMetaTable.delete_for_document(self._conn, document_id)
def _increment_total_inserts(self, count: int) -> None: def _increment_total_inserts(self, count: int) -> None:
"""Increment the cumulative insert counter stored in index_meta. """Increment the cumulative insert counter stored in index_meta.
@@ -339,48 +374,56 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
live_rows exceeds COMPACT_BLOAT_RATIO the table has accumulated live_rows exceeds COMPACT_BLOAT_RATIO the table has accumulated
enough deleted-but-not-freed rows to warrant a rebuild. enough deleted-but-not-freed rows to warrant a rebuild.
""" """
current = int(self._meta_get("total_inserts") or "0") IndexMetaTable.increment_total_inserts(self._conn, count)
self._meta_set("total_inserts", str(current + count))
def add(self, nodes: Sequence[BaseNode], **add_kwargs: Any) -> list[str]: def add(self, nodes: Sequence[BaseNode], **add_kwargs: Any) -> list[str]:
if not nodes: if not nodes:
return [] return []
rows = [self._row(node) for node in nodes] rows = [self._row(node) for node in nodes]
with self._transaction(): with self._transaction():
self._ensure_table(len(nodes[0].get_embedding())) self._ensure_table(
self._conn.executemany(self._INSERT, rows) len(nodes[0].get_embedding()),
table_exists=self.table_exists(),
)
self._conn.executemany(_INSERT, _vec0_params(rows))
self._index_chunks(rows)
self._increment_total_inserts(len(rows)) self._increment_total_inserts(len(rows))
return [node.node_id for node in nodes] return [node.node_id for node in nodes]
def upsert_document(self, document_id: str, nodes: list[BaseNode]) -> list[str]: def upsert_document(
self,
document_id: int | str,
nodes: list[BaseNode],
) -> list[str]:
"""Atomically replace all stored chunks of ``document_id`` with ``nodes``. """Atomically replace all stored chunks of ``document_id`` with ``nodes``.
One transaction deletes the document's existing rows and inserts the One transaction deletes the document's existing rows and inserts the
new set (vec0's INSERT OR REPLACE is broken upstream, #259, so new set (vec0's INSERT OR REPLACE is broken upstream, so delete+insert
delete+insert it is). WAL readers in other processes see either the it is). WAL readers in other processes see either the old or the new
old or the new chunk set, never a partial state. chunk set, never a partial state.
""" """
doc_id = int(document_id)
rows = [self._row(node) for node in nodes] rows = [self._row(node) for node in nodes]
with self._transaction(): with self._transaction():
if nodes: table_exists = self.table_exists()
self._ensure_table(len(nodes[0].get_embedding())) if nodes and not table_exists:
if self.table_exists(): self._ensure_table(
self._conn.execute( len(nodes[0].get_embedding()),
"DELETE FROM " + DEFAULT_TABLE_NAME + " WHERE document_id = ?", table_exists=False,
(str(document_id),),
) )
table_exists = True
if table_exists:
self._delete_chunks_by_document_id(doc_id)
if rows: if rows:
self._conn.executemany(self._INSERT, rows) self._conn.executemany(_INSERT, _vec0_params(rows))
self._index_chunks(rows)
self._increment_total_inserts(len(rows)) self._increment_total_inserts(len(rows))
return [node.node_id for node in nodes] return [node.node_id for node in nodes]
def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None: def delete(self, ref_doc_id: int | str, **delete_kwargs: Any) -> None:
if self.table_exists(): if self.table_exists():
with self._transaction(): with self._transaction():
self._conn.execute( self._delete_chunks_by_document_id(int(ref_doc_id))
"DELETE FROM " + DEFAULT_TABLE_NAME + " WHERE document_id = ?",
(str(ref_doc_id),),
)
def _rows_to_nodes(self, rows: list[sqlite3.Row]) -> list[BaseNode]: def _rows_to_nodes(self, rows: list[sqlite3.Row]) -> list[BaseNode]:
nodes: list[BaseNode] = [] nodes: list[BaseNode] = []
@@ -450,41 +493,60 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
def get_modified_times(self) -> dict[str, str]: def get_modified_times(self) -> dict[str, str]:
"""Return {document_id: stored_modified_isoformat} for all indexed documents. """Return {document_id: stored_modified_isoformat} for all indexed documents.
All chunks of a document share the same ``modified`` value, so the document_meta already has exactly one row per document (not per
first row seen per document is sufficient. chunk, unlike the vec0 table), so no dedup is needed here.
""" """
if not self.table_exists(): if not self.table_exists():
return {} return {}
result: dict[str, str] = {} return DocumentMetaTable.all_modified_times(self._conn)
for row in self._conn.execute(
"SELECT document_id, modified FROM " + DEFAULT_TABLE_NAME, @property
): def _db_path(self) -> str:
doc_id = str(row["document_id"]) return str(Path(self._uri) / DB_FILENAME)
if doc_id not in result:
result[doc_id] = str(row["modified"] or "") @contextmanager
return result def _rebuild_file(self) -> Iterator[sqlite3.Connection]:
"""Open a fresh temp database file for a file-swap rebuild (compact
or structural migration), yielding its connection for the caller to
populate.
On success, swaps the temp file in as the live database (closing
this store's current connection first -- see _swap_in_compact()).
On any exception, discards the temp file, including its -wal/-shm,
instead, and this store's own connection is left untouched.
"""
compact_path = self._db_path + ".compact"
new_conn = self._open_connection(compact_path)
try:
yield new_conn
except BaseException:
new_conn.close()
for suffix in ["", "-wal", "-shm"]:
Path(compact_path + suffix).unlink(missing_ok=True)
raise
else:
new_conn.close()
self._swap_in_compact(compact_path, self._db_path)
def compact(self, *, force: bool = False) -> None: def compact(self, *, force: bool = False) -> None:
"""Rebuild the database file to reclaim space left behind by DELETEs. """Rebuild the database file to reclaim space left behind by DELETEs.
vec0 DELETE only invalidates rows; the vector data stays in the file vec0 DELETE only invalidates rows; the vector data stays in the file
forever (asg017/sqlite-vec#54), and per-document re-indexing is a forever, and per-document re-indexing is a delete+insert. The
delete+insert. The cumulative insert counter in ``index_meta`` tracks cumulative insert counter in ``index_meta`` tracks total rows ever
total rows ever written; when that exceeds ``COMPACT_BLOAT_RATIO`` x written; when that exceeds ``COMPACT_BLOAT_RATIO`` x the live row
the live row count (or when forced), live rows are copied into a fresh count (or when forced), live rows are copied into a fresh database
database file and swapped in via ``os.replace``. file and swapped in via ``os.replace``.
Note: ``ALTER TABLE ... RENAME TO`` on vec0 virtual tables does NOT Note: ``ALTER TABLE ... RENAME TO`` on vec0 virtual tables does NOT
rename the shadow tables (sqlite-vec upstream limitation), so rename the shadow tables (sqlite-vec upstream limitation), so an
an in-place rename-based rebuild is not safe. The file-swap approach in-place rename-based rebuild is not safe. The file-swap approach is
is the maintainer-endorsed workaround (asg017/sqlite-vec#205). the maintainer-endorsed workaround.
""" """
if not self.table_exists(): if not self.table_exists():
return return
live = self._conn.execute( live = DocumentChunksTable.count(self._conn)
"SELECT count(*) FROM " + DEFAULT_TABLE_NAME, total = IndexMetaTable.get_total_inserts(self._conn) or live
).fetchone()[0]
total = int(self._meta_get("total_inserts") or str(live))
if not force and total <= max(live, 1) * COMPACT_BLOAT_RATIO: if not force and total <= max(live, 1) * COMPACT_BLOAT_RATIO:
return return
dim = self.vector_dim() dim = self.vector_dim()
@@ -496,50 +558,62 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
live, live,
total, total,
) )
db_path = str(Path(self._uri) / DB_FILENAME) with self._rebuild_file() as new_conn:
compact_path = db_path + ".compact" self._rebuild_into(self._conn, new_conn, dim)
# Copy all live rows into a fresh database file. @staticmethod
new_conn = self._open_connection(compact_path) def _rebuild_into(
try: src_conn: sqlite3.Connection,
self._create_vec_table(new_conn, dim) dst_conn: sqlite3.Connection,
self._meta_set_on(new_conn, "dim", str(dim)) dim: int,
for key in ("embed_model", "schema_version"): ) -> int:
value = self._meta_get(key) """Create the vec0 table in ``dst_conn``, copy dim/embed_model from
if value is not None: ``src_conn``, and stream every live vec0 row, document_chunks row,
self._meta_set_on(new_conn, key, value) and document_meta row across. Returns the number of vec0 rows
src_cursor = self._conn.execute( copied. Used by compact() only -- m0001_v1_to_v2 freezes its own
"SELECT id, document_id, modified, node_content, embedding " copy loop instead of calling this, since this always reflects the
"FROM " + DEFAULT_TABLE_NAME, *current* schema (see the migration DDL-freezing rule in the spec).
"""
PaperlessSqliteVecVectorStore._create_vec_table(dst_conn, dim)
dim_value = IndexMetaTable.get_dim(src_conn)
if dim_value is not None:
IndexMetaTable.set_dim(dst_conn, dim_value)
embed_model = IndexMetaTable.get_embed_model(src_conn)
if embed_model is not None:
IndexMetaTable.set_embed_model(dst_conn, embed_model)
schema_version = IndexMetaTable.get_schema_version(src_conn)
if schema_version is not None:
IndexMetaTable.set_schema_version(dst_conn, schema_version)
dst_conn.execute("BEGIN IMMEDIATE")
src_cursor = src_conn.execute(
"SELECT id, document_id, node_content, embedding FROM "
+ DEFAULT_TABLE_NAME,
) )
new_conn.execute("BEGIN IMMEDIATE") copied = 0
# Stream rows from the source cursor in batches instead of
# materializing the whole table in memory, so a large index does
# not cause an OOM during routine maintenance compactions.
while batch := src_cursor.fetchmany(COMPACT_BATCH_SIZE): while batch := src_cursor.fetchmany(COMPACT_BATCH_SIZE):
new_conn.executemany( dst_conn.executemany(
self._INSERT, _INSERT,
[ [
( (
r["id"], r["id"],
r["document_id"], r["document_id"],
r["modified"],
r["node_content"], r["node_content"],
bytes(r["embedding"]), bytes(r["embedding"]),
) )
for r in batch for r in batch
], ],
) )
# Reset the cumulative counter: after compact, total_inserts == live. DocumentChunksTable.insert_many(
self._meta_set_on(new_conn, "total_inserts", str(live)) dst_conn,
new_conn.execute("COMMIT") (ChunkRow(r["id"], r["document_id"]) for r in batch),
except BaseException: )
new_conn.close() copied += len(batch)
for p in [compact_path, compact_path + "-wal", compact_path + "-shm"]: DocumentMetaTable.copy_all(src_conn, dst_conn, COMPACT_BATCH_SIZE)
Path(p).unlink(missing_ok=True) # Reset the cumulative counter: after a rebuild, total_inserts == live.
raise IndexMetaTable.reset_total_inserts(dst_conn, copied)
new_conn.close() dst_conn.execute("COMMIT")
self._swap_in_compact(compact_path, db_path) return copied
def _swap_in_compact(self, compact_path: str, db_path: str) -> None: def _swap_in_compact(self, compact_path: str, db_path: str) -> None:
"""Atomically replace the live database with the compacted copy.""" """Atomically replace the live database with the compacted copy."""
@@ -551,6 +625,31 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
Path(compact_path).replace(db_path) Path(compact_path).replace(db_path)
self._conn = self._open_connection(db_path) self._conn = self._open_connection(db_path)
def _stored_schema_version(self) -> int | None:
"""The schema_version recorded in index_meta, or None if no table
exists. A missing key (a store predating version tracking) is
treated as SCHEMA_VERSION -- i.e. already current -- since no
migration in MIGRATIONS targets a version before tracking began.
"""
if not self.table_exists():
return None
raw_version = IndexMetaTable.get_schema_version(self._conn)
return raw_version if raw_version is not None else SCHEMA_VERSION
def has_pending_migration(self) -> bool:
"""Cheaply check whether a migration is pending, with no exclusive
access needed -- just a metadata read under the connection callers
already hold via the write FileLock.
Callers should only pay for check_and_run_migrations()'s exclusive
access (a structural migration's file swap must not run while
readers are active) when this returns True, so that the common
case -- already at SCHEMA_VERSION -- never contends with readers
or a concurrent compaction.
"""
current = self._stored_schema_version()
return current is not None and current < SCHEMA_VERSION
def check_and_run_migrations(self) -> bool: def check_and_run_migrations(self) -> bool:
"""Apply any pending schema migrations to the store. """Apply any pending schema migrations to the store.
@@ -559,15 +658,13 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
this method returns True when one is encountered so the caller can this method returns True when one is encountered so the caller can
force a full rebuild (which recreates the table at SCHEMA_VERSION). force a full rebuild (which recreates the table at SCHEMA_VERSION).
Must be called under the write FileLock. No-op when the table does Must be called under the write FileLock, with readers excluded (see
not exist or is already at SCHEMA_VERSION. has_pending_migration() for a cheap pre-check that avoids paying for
that exclusion in the common case). No-op when the table does not
exist or is already at SCHEMA_VERSION.
""" """
if not self.table_exists(): current = self._stored_schema_version()
return False if current is None or current >= SCHEMA_VERSION:
raw = self._meta_get("schema_version")
current = int(raw) if raw is not None else SCHEMA_VERSION
if current >= SCHEMA_VERSION:
return False return False
pending = sorted( pending = sorted(
@@ -579,7 +676,7 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
if migration.kind == "re-embed": if migration.kind == "re-embed":
logger.warning( logger.warning(
"LLM index schema v%d -> v%d requires re-embedding (%s); " "LLM index schema v%d -> v%d requires re-embedding (%s); "
"forcing full rebuild.", "the caller must force a rebuild.",
migration.from_version, migration.from_version,
migration.to_version, migration.to_version,
migration.description, migration.description,
@@ -601,16 +698,12 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
dim = self.vector_dim() dim = self.vector_dim()
if dim is None: # pragma: no cover if dim is None: # pragma: no cover
raise RuntimeError("Cannot migrate: no stored vector dimension") raise RuntimeError("Cannot migrate: no stored vector dimension")
db_path = str(Path(self._uri) / DB_FILENAME) with self._rebuild_file() as new_conn:
compact_path = db_path + ".compact"
new_conn = self._open_connection(compact_path)
try:
migration.apply(self._conn, new_conn, dim) migration.apply(self._conn, new_conn, dim)
self._meta_set_on(new_conn, "schema_version", str(migration.to_version)) IndexMetaTable.set_schema_version(new_conn, migration.to_version)
except BaseException: # pragma: no cover
new_conn.close()
for p in [compact_path, compact_path + "-wal", compact_path + "-shm"]: # Registers m0001_v1_to_v2 into MIGRATIONS; must be at the bottom (needs
Path(p).unlink(missing_ok=True) # PaperlessSqliteVecVectorStore fully defined) -- see
raise # paperless_ai/migrations/__init__.py for the full procedure.
new_conn.close() from paperless_ai.migrations import m0001_v1_to_v2 # noqa: E402, F401
self._swap_in_compact(compact_path, db_path)