mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-07-30 07:44:54 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7d9ba34582 | ||
|
|
91c3d9caab |
@@ -1,12 +0,0 @@
|
||||
#!/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
|
||||
@@ -1 +0,0 @@
|
||||
oneshot
|
||||
@@ -1 +0,0 @@
|
||||
/etc/s6-overlay/s6-rc.d/init-llmindex-migrate/run
|
||||
+1
-11
@@ -212,16 +212,6 @@ following:
|
||||
This is a no-op if the index is already up to date, so it is safe to
|
||||
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
|
||||
|
||||
Paperless-ngx is compatible with Django-supported versions of PostgreSQL and MariaDB and it is generally
|
||||
@@ -542,7 +532,7 @@ index is updated automatically on the schedule set by
|
||||
can manage it manually:
|
||||
|
||||
```
|
||||
document_llmindex {rebuild,update,compact,migrate}
|
||||
document_llmindex {rebuild,update,compact}
|
||||
```
|
||||
|
||||
Specify `rebuild` to build the index from scratch from all documents in the database. Use
|
||||
|
||||
@@ -3,7 +3,6 @@ from typing import Any
|
||||
from documents.management.commands.base import PaperlessCommand
|
||||
from documents.tasks import llmindex_index
|
||||
from paperless_ai.indexing import llm_index_compact
|
||||
from paperless_ai.indexing import llm_index_migrate
|
||||
|
||||
|
||||
class Command(PaperlessCommand):
|
||||
@@ -14,18 +13,12 @@ class Command(PaperlessCommand):
|
||||
|
||||
def add_arguments(self, parser: Any) -> None:
|
||||
super().add_arguments(parser)
|
||||
parser.add_argument(
|
||||
"command",
|
||||
choices=["rebuild", "update", "compact", "migrate"],
|
||||
)
|
||||
parser.add_argument("command", choices=["rebuild", "update", "compact"])
|
||||
|
||||
def handle(self, *args: Any, **options: Any) -> None:
|
||||
if options["command"] == "compact":
|
||||
llm_index_compact()
|
||||
return
|
||||
if options["command"] == "migrate":
|
||||
llm_index_migrate()
|
||||
return
|
||||
llmindex_index(
|
||||
rebuild=options["command"] == "rebuild",
|
||||
iter_wrapper=lambda docs: self.track(
|
||||
|
||||
@@ -9,7 +9,6 @@ if TYPE_CHECKING:
|
||||
|
||||
_COMPACT = "documents.management.commands.document_llmindex.llm_index_compact"
|
||||
_INDEX = "documents.management.commands.document_llmindex.llmindex_index"
|
||||
_MIGRATE = "documents.management.commands.document_llmindex.llm_index_migrate"
|
||||
|
||||
|
||||
class TestDocumentLlmindexCommand:
|
||||
@@ -18,11 +17,6 @@ class TestDocumentLlmindexCommand:
|
||||
call_command("document_llmindex", "compact")
|
||||
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(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
|
||||
@@ -144,24 +144,6 @@ def _exclude_readers():
|
||||
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
|
||||
def write_store(embed_model_name: str | None = None):
|
||||
"""Acquire the write lock and yield the vector store.
|
||||
@@ -186,21 +168,6 @@ def write_store(embed_model_name: str | None = None):
|
||||
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:
|
||||
"""
|
||||
Returns the ``name`` of a related object (correspondent, document_type,
|
||||
@@ -372,7 +339,15 @@ def update_llm_index(
|
||||
happens, since a rebuild always covers the whole library regardless.
|
||||
"""
|
||||
with write_store() as store:
|
||||
needs_reembed = _check_and_run_migrations(store)
|
||||
try:
|
||||
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:
|
||||
logger.warning(
|
||||
"LLM index migration requires re-embedding; forcing rebuild.",
|
||||
@@ -437,7 +412,14 @@ def update_llm_index(
|
||||
else "No changes detected in LLM index."
|
||||
)
|
||||
|
||||
_with_exclusive_access("compaction", store.compact)
|
||||
try:
|
||||
with _exclude_readers():
|
||||
store.compact()
|
||||
except Timeout:
|
||||
logger.info(
|
||||
"Skipping LLM index compaction: index readers are active; "
|
||||
"will retry next run.",
|
||||
)
|
||||
return msg
|
||||
|
||||
|
||||
@@ -452,60 +434,25 @@ def llm_index_add_or_update_document(document: Document):
|
||||
_embed_nodes(new_nodes, get_embedding_model(config))
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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:
|
||||
"""Compact the index immediately, rebuilding the table to reclaim space."""
|
||||
with write_store() as store:
|
||||
_with_exclusive_access("compaction", lambda: store.compact(force=True))
|
||||
try:
|
||||
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):
|
||||
"""Remove a document's chunks from the LLM index."""
|
||||
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))
|
||||
|
||||
|
||||
|
||||
@@ -1,60 +0,0 @@
|
||||
"""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,229 @@
|
||||
"""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:
|
||||
current = IndexMetaTable.get_total_inserts(conn)
|
||||
IndexMetaTable._set(conn, "total_inserts", str(current + 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))
|
||||
@@ -1,4 +1,3 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
from unittest.mock import patch
|
||||
@@ -738,7 +737,6 @@ class TestLlmIndexLocking:
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
mock_store = MagicMock()
|
||||
mock_store.has_pending_migration.return_value = False
|
||||
mocker.patch(
|
||||
"paperless_ai.indexing.write_store",
|
||||
return_value=mocker.MagicMock(
|
||||
@@ -759,45 +757,12 @@ class TestLlmIndexLocking:
|
||||
|
||||
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(
|
||||
self,
|
||||
temp_llm_index_dir: Path,
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
mock_store = MagicMock()
|
||||
mock_store.has_pending_migration.return_value = False
|
||||
mocker.patch(
|
||||
"paperless_ai.indexing.write_store",
|
||||
return_value=mocker.MagicMock(
|
||||
@@ -812,31 +777,6 @@ class TestLlmIndexLocking:
|
||||
|
||||
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(
|
||||
self,
|
||||
temp_llm_index_dir: Path,
|
||||
@@ -909,76 +849,6 @@ class TestVectorStoreIndexing:
|
||||
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
|
||||
class TestQuerySimilarDocuments:
|
||||
def test_query_similar_documents_respects_allowed_ids(
|
||||
|
||||
@@ -0,0 +1,306 @@
|
||||
import sqlite3
|
||||
from collections.abc import Generator
|
||||
|
||||
import pytest
|
||||
|
||||
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_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
|
||||
@@ -9,11 +9,11 @@ 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 VectorStoreQuery
|
||||
|
||||
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 DEFAULT_TABLE_NAME
|
||||
from paperless_ai.vector_store import MIGRATIONS
|
||||
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 _build_where
|
||||
|
||||
@@ -646,50 +646,3 @@ class TestMigrations:
|
||||
|
||||
assert result is True
|
||||
assert self._schema_version(store) == 2
|
||||
|
||||
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
|
||||
|
||||
@@ -2,12 +2,16 @@ import json
|
||||
import logging
|
||||
import sqlite3
|
||||
import struct
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Iterator
|
||||
from collections.abc import Sequence
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
from pathlib import Path
|
||||
from types import TracebackType
|
||||
from typing import Any
|
||||
from typing import Literal
|
||||
|
||||
import sqlite_vec
|
||||
from llama_index.core.bridge.pydantic import PrivateAttr
|
||||
@@ -22,9 +26,6 @@ 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 node_to_metadata_dict
|
||||
|
||||
from paperless_ai.migrations import MIGRATIONS
|
||||
from paperless_ai.migrations import Migration
|
||||
|
||||
logger = logging.getLogger("paperless_ai.vector_store")
|
||||
|
||||
DB_FILENAME = "llmindex.db"
|
||||
@@ -52,6 +53,38 @@ COMPACT_BATCH_SIZE = 500
|
||||
_FILTER_COLUMNS = frozenset({"document_id", "modified"})
|
||||
|
||||
|
||||
@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 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
|
||||
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. 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:
|
||||
return struct.pack(f"{len(embedding)}f", *embedding)
|
||||
|
||||
@@ -518,31 +551,6 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
|
||||
Path(compact_path).replace(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 = self._meta_get("schema_version")
|
||||
return int(raw) if raw 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:
|
||||
"""Apply any pending schema migrations to the store.
|
||||
|
||||
@@ -551,13 +559,15 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
|
||||
this method returns True when one is encountered so the caller can
|
||||
force a full rebuild (which recreates the table at SCHEMA_VERSION).
|
||||
|
||||
Must be called under the write FileLock, with readers excluded (see
|
||||
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.
|
||||
Must be called under the write FileLock. No-op when the table does
|
||||
not exist or is already at SCHEMA_VERSION.
|
||||
"""
|
||||
current = self._stored_schema_version()
|
||||
if current is None or current >= SCHEMA_VERSION:
|
||||
if not self.table_exists():
|
||||
return False
|
||||
|
||||
raw = self._meta_get("schema_version")
|
||||
current = int(raw) if raw is not None else SCHEMA_VERSION
|
||||
if current >= SCHEMA_VERSION:
|
||||
return False
|
||||
|
||||
pending = sorted(
|
||||
@@ -569,7 +579,7 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
|
||||
if migration.kind == "re-embed":
|
||||
logger.warning(
|
||||
"LLM index schema v%d -> v%d requires re-embedding (%s); "
|
||||
"the caller must force a rebuild.",
|
||||
"forcing full rebuild.",
|
||||
migration.from_version,
|
||||
migration.to_version,
|
||||
migration.description,
|
||||
|
||||
Reference in New Issue
Block a user