Compare commits

..
Author SHA1 Message Date
GitHub Actions c29c9178f0 Auto translate strings 2026-07-30 14:39:47 +00:00
shamoonandGitHub baae99eb49 Fix: normalize monetary decimal symbol by locale (#13427) 2026-07-30 07:37:52 -07:00
Trenton HandGitHub f50a021ef6 Fix: fold overflowing path text in sanity checker (#13426) 2026-07-30 14:35:19 +00:00
shamoonandGitHub c8e00dca04 Fix: correct missing add field button for custom fields (#13424) 2026-07-30 07:16:25 -07:00
shamoonandGitHub 0a56018d6b Fix: hide attributes collapse / show button without UI settings (#13425) 2026-07-30 07:08:50 -07:00
350684cd6b Fix: consolidate born-digital PDF detection between archive decision and OCR (#13409)
* Fix: unify born-digital PDF detection between archive decision and OCR

should_produce_archive() and RasterisedDocumentParser.parse() each
reimplemented the "does this PDF have real text" check independently,
using different normalization of pdftotext output. Raw pdftotext output
can be non-empty (whitespace/form-feed layout padding) even when there
is no real content, so the two checks could disagree: consumer.py
treated a tagged-but-textless PDF as born-digital and skipped the
archive, while the parser's own (stricter, normalized) check found no
text and ran OCR anyway, leaving the document with no archive despite
real OCR text (GH #13387).

Both call sites now share one predicate, pdf_born_digital_text() in
paperless/parsers/utils.py, so they can no longer drift apart.

* Fix: restore extract_text seam for born-digital detection in parse()

parse() had switched to calling pdf_born_digital_text() directly for
its initial text/born-digital check, bypassing the parser's own
extract_text instance method. That broke test mockability (tests patch
tesseract_parser.extract_text to control the born-digital decision)
and caused CI failures with mismatched OCR call counts and text.

Split pdf_born_digital_text() into is_born_digital_text(text, path,
log) - a pure decision function - and a thin pdf_born_digital_text()
wrapper for callers without text in hand (consumer.should_produce_archive).
parse() now extracts via self.extract_text(None, document_path) and
passes the result to is_born_digital_text(), restoring the seam with
no change to production behavior.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Cleanup: simplify born-digital detection, close #13387 test gap

Simplification pass over the born-digital detection consolidation:
- is_born_digital_text(): drop the has_text temp for an early return.
- consumer.py: standardize the archive-decision log lines on plain
  hyphens (was a mix of em-dash and hyphen) and hoist the duplicated
  text_length computation.
- Parametrize TestPdfBornDigitalText instead of four near-identical
  tests.

Code review follow-up: the existing tests only ever exercised
pdf_born_digital_text() through mocks, so the actual #13387 scenario
(a tagged PDF whose only "text" is layout padding) was never checked
against real pdftotext/pikepdf output - a regression in the
normalize-before-decide logic itself would have gone undetected.
Moved tagged_no_text_pdf_file from parsers/conftest.py up to the
shared paperless/tests/conftest.py (it was previously only visible to
tests under parsers/) and added a non-mocked regression test against
the real sample file.

Also fixed two stale comments in test_consumer.py referencing a
_extract_text_for_archive_check helper that no longer exists.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-29 12:41:09 -07:00
34 changed files with 681 additions and 2173 deletions
@@ -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
View File
@@ -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
+10 -10
View File
@@ -1849,7 +1849,7 @@
</context-group>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">154</context>
<context context-type="linenumber">159</context>
</context-group>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/filterable-dropdown/filterable-dropdown.component.html</context>
@@ -3972,11 +3972,11 @@
</context-group>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">96</context>
<context context-type="linenumber">101</context>
</context-group>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">102</context>
<context context-type="linenumber">107</context>
</context-group>
</trans-unit>
<trans-unit id="3800326155195149498" datatype="html">
@@ -3987,11 +3987,11 @@
</context-group>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">97</context>
<context context-type="linenumber">102</context>
</context-group>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">103</context>
<context context-type="linenumber">108</context>
</context-group>
</trans-unit>
<trans-unit id="7551700625201096185" datatype="html">
@@ -4002,14 +4002,14 @@
</context-group>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">119</context>
<context context-type="linenumber">124</context>
</context-group>
</trans-unit>
<trans-unit id="3184700926171002527" datatype="html">
<source>Any</source>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">152</context>
<context context-type="linenumber">157</context>
</context-group>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/filterable-dropdown/filterable-dropdown.component.html</context>
@@ -4020,21 +4020,21 @@
<source>Not</source>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">157</context>
<context context-type="linenumber">162</context>
</context-group>
</trans-unit>
<trans-unit id="6548676277933116532" datatype="html">
<source>Add query</source>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">176</context>
<context context-type="linenumber">181</context>
</context-group>
</trans-unit>
<trans-unit id="5599577087865387184" datatype="html">
<source>Add expression</source>
<context-group purpose="location">
<context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
<context context-type="linenumber">179</context>
<context context-type="linenumber">184</context>
</context-group>
</trans-unit>
<trans-unit id="6312759212949884929" datatype="html">
@@ -182,7 +182,7 @@
container="body" triggers="mouseenter:mouseleave" popoverClass="popover-slim">
<i-bs class="me-2" name="stack"></i-bs><span><ng-container i18n>Attributes</ng-container></span>
</a>
@if (!slimSidebarEnabled) {
@if (!slimSidebarEnabled && canSaveSettings) {
<button
type="button"
class="btn btn-link btn-sm text-muted p-0 me-3 attributes-expand-btn"
@@ -68,6 +68,11 @@
></ng-select>
} @else if (getCustomFieldByID(atom.field)?.data_type === CustomFieldDataType.DocumentLink) {
<pngx-input-document-link [(ngModel)]="atom.value" class="w-25 form-select doc-link-select p-0" placeholder="Search docs..." i18n-placeholder [minimal]="true"></pngx-input-document-link>
} @else if (getCustomFieldByID(atom.field)?.data_type === CustomFieldDataType.Monetary) {
<input class="w-25 form-control rounded-end" type="text" inputmode="decimal"
[ngModel]="atom.value"
(ngModelChange)="setMonetaryValue(atom, $event)"
[disabled]="disabled">
} @else {
<input class="w-25 form-control rounded-end" type="text" [(ngModel)]="atom.value" [disabled]="disabled">
}
@@ -1,5 +1,6 @@
import { provideHttpClient, withInterceptorsFromDi } from '@angular/common/http'
import { provideHttpClientTesting } from '@angular/common/http/testing'
import { LOCALE_ID } from '@angular/core'
import { ComponentFixture, TestBed } from '@angular/core/testing'
import { FormsModule, ReactiveFormsModule } from '@angular/forms'
import { NgbDropdownModule } from '@ng-bootstrap/ng-bootstrap'
@@ -41,6 +42,12 @@ const customFields = [
],
},
},
{
id: 3,
name: 'Test Monetary Field',
data_type: CustomFieldDataType.Monetary,
extra_data: { default_currency: 'EUR' },
},
]
describe('CustomFieldsQueryDropdownComponent', () => {
@@ -61,6 +68,7 @@ describe('CustomFieldsQueryDropdownComponent', () => {
providers: [
provideHttpClient(withInterceptorsFromDi()),
provideHttpClientTesting(),
{ provide: LOCALE_ID, useValue: 'de' },
],
}).compileComponents()
@@ -150,6 +158,22 @@ describe('CustomFieldsQueryDropdownComponent', () => {
expect(options2).toEqual([])
})
it('should normalize localized monetary comparison values', () => {
const atom = new CustomFieldQueryAtom([3, 'exact', null])
component.setMonetaryValue(atom, '1.234,56')
expect(atom.value).toEqual('1234.56')
})
it('should preserve API-formatted monetary comparison values', () => {
const atom = new CustomFieldQueryAtom([3, 'exact', null])
component.setMonetaryValue(atom, '1234.56')
expect(atom.value).toEqual('1234.56')
})
it('should remove an element from the selection model', () => {
const expression = new CustomFieldQueryExpression()
const atom = new CustomFieldQueryAtom()
@@ -1,9 +1,14 @@
import { NgTemplateOutlet } from '@angular/common'
import {
getLocaleNumberSymbol,
NgTemplateOutlet,
NumberSymbol,
} from '@angular/common'
import {
Component,
EventEmitter,
inject,
Input,
LOCALE_ID,
Output,
QueryList,
signal,
@@ -212,6 +217,7 @@ export class CustomFieldQueriesModel {
})
export class CustomFieldsQueryDropdownComponent extends LoadingComponentWithPermissions {
protected customFieldsService = inject(CustomFieldsService)
private readonly locale = inject(LOCALE_ID)
public CustomFieldQueryComponentType = CustomFieldQueryElementType
public CustomFieldQueryOperator = CustomFieldQueryOperator
@@ -376,4 +382,18 @@ export class CustomFieldsQueryDropdownComponent extends LoadingComponentWithPerm
}
return []
}
setMonetaryValue(atom: CustomFieldQueryAtom, value: string) {
// Normalize the decimal symbol e.g. . vs , by locale
const decimalSymbol = getLocaleNumberSymbol(
this.locale,
NumberSymbol.Decimal
)
if (decimalSymbol !== '.' && value.includes(decimalSymbol)) {
const groupSymbol = getLocaleNumberSymbol(this.locale, NumberSymbol.Group)
value = value.split(groupSymbol).join('').split(decimalSymbol).join('.')
}
atom.value = value
}
}
@@ -51,8 +51,8 @@
*pngxIfPermissions="{ action: PermissionAction.Add, type: activeManagementList.permissionType }">
<i-bs name="plus-circle" class="me-1"></i-bs><ng-container i18n>Create</ng-container>
</button>
} @else if (activeCustomFields) {
<button type="button" class="btn btn-sm btn-outline-primary" (click)="activeCustomFields.editField()"
} @else if (customFieldsActive) {
<button type="button" class="btn btn-sm btn-outline-primary" (click)="addCustomField()"
*pngxIfPermissions="{ action: PermissionAction.Add, type: PermissionType.CustomField }">
<i-bs name="plus-circle" class="me-1"></i-bs><ng-container i18n>Add Field</ng-container>
</button>
@@ -18,6 +18,7 @@ import {
DocumentAttributesComponent,
DocumentAttributesSectionKind,
} from './document-attributes.component'
import { CustomFieldsComponent } from './custom-fields/custom-fields.component'
import { ManagementListComponent } from './management-list/management-list.component'
@Component({
@@ -207,6 +208,29 @@ describe('DocumentAttributesComponent', () => {
expect(component.activeSection.kind).toBe(
DocumentAttributesSectionKind.CustomFields
)
expect(component.activeCustomFields).toBeDefined()
const customFields = Object.create(CustomFieldsComponent.prototype)
customFields.editField = jest.fn()
component.activeOutlet = {
componentInstance: customFields,
} as any
expect(component.activeCustomFields).toBe(customFields)
component.addCustomField()
expect(customFields.editField).toHaveBeenCalled()
})
it('should show the add field button before the custom fields instance is available', async () => {
jest.spyOn(permissionsService, 'currentUserCan').mockReturnValue(true)
fixture.detectChanges()
component.activeNavID.set(2)
await fixture.whenStable()
expect(component.activeCustomFields).toBeNull()
expect(
fixture.nativeElement.querySelector(
'pngx-page-header .btn-outline-primary'
)?.textContent
).toContain('Add Field')
})
})
@@ -163,12 +163,17 @@ export class DocumentAttributesComponent implements OnInit, OnDestroy {
}
get activeCustomFields(): CustomFieldsComponent | null {
if (this.activeSection?.kind !== DocumentAttributesSectionKind.CustomFields)
return null
if (!this.customFieldsActive) return null
const instance = this.activeOutlet?.componentInstance
return instance instanceof CustomFieldsComponent ? instance : null
}
get customFieldsActive(): boolean {
return (
this.activeSection?.kind === DocumentAttributesSectionKind.CustomFields
)
}
get activeTabLabel(): string {
return this.activeSection?.label ?? ''
}
@@ -224,6 +229,10 @@ export class DocumentAttributesComponent implements OnInit, OnDestroy {
this.router.navigate(['attributes', nextSection])
}
addCustomField(): void {
this.activeCustomFields?.editField(null)
}
private getDefaultNavID(): DocumentAttributesNavIDs | null {
return this.visibleSections[0]?.id ?? null
}
+15 -25
View File
@@ -57,9 +57,7 @@ from paperless.models import ArchiveFileGenerationChoices
from paperless.parsers import ParserContext
from paperless.parsers import ParserProtocol
from paperless.parsers.registry import get_parser_registry
from paperless.parsers.utils import PDF_TEXT_MIN_LENGTH
from paperless.parsers.utils import extract_pdf_text
from paperless.parsers.utils import is_tagged_pdf
from paperless.parsers.utils import pdf_born_digital_text
LOGGING_NAME: Final[str] = "paperless.consumer"
@@ -138,53 +136,45 @@ def should_produce_archive(
# Must produce a PDF so the frontend can display the original format at all.
if parser.requires_pdf_rendition:
_log.debug("Archive: yes parser requires PDF rendition for frontend display")
_log.debug("Archive: yes - parser requires PDF rendition for frontend display")
return True
# Parser cannot produce an archive (e.g. TextDocumentParser).
if not parser.can_produce_archive:
_log.debug("Archive: no parser cannot produce archives")
_log.debug("Archive: no - parser cannot produce archives")
return False
generation = OcrConfig().archive_file_generation
if generation == ArchiveFileGenerationChoices.ALWAYS:
_log.debug("Archive: yes ARCHIVE_FILE_GENERATION=always")
_log.debug("Archive: yes - ARCHIVE_FILE_GENERATION=always")
return True
if generation == ArchiveFileGenerationChoices.NEVER:
_log.debug("Archive: no ARCHIVE_FILE_GENERATION=never")
_log.debug("Archive: no - ARCHIVE_FILE_GENERATION=never")
return False
# auto: produce archives for scanned/image documents; skip for born-digital PDFs.
if mime_type.startswith("image/"):
_log.debug("Archive: yes image document, ARCHIVE_FILE_GENERATION=auto")
_log.debug("Archive: yes - image document, ARCHIVE_FILE_GENERATION=auto")
return True
if mime_type == "application/pdf":
text = extract_pdf_text(document_path)
has_text = text is not None and len(text) > 0
if has_text and is_tagged_pdf(document_path):
text, born_digital = pdf_born_digital_text(document_path, log=_log)
text_length = len(text) if text else 0
if born_digital:
_log.debug(
"Archive: no born-digital PDF (structure tags detected),"
"Archive: no - born-digital PDF (text_length=%d),"
" ARCHIVE_FILE_GENERATION=auto",
text_length,
)
return False
if text is None or len(text) <= PDF_TEXT_MIN_LENGTH:
_log.debug(
"Archive: yes — scanned PDF (text_length=%d%d),"
" ARCHIVE_FILE_GENERATION=auto",
len(text) if text else 0,
PDF_TEXT_MIN_LENGTH,
)
return True
_log.debug(
"Archive: no — born-digital PDF (text_length=%d > %d),"
"Archive: yes - scanned/textless PDF (text_length=%d),"
" ARCHIVE_FILE_GENERATION=auto",
len(text),
PDF_TEXT_MIN_LENGTH,
text_length,
)
return False
return True
_log.debug(
"Archive: no MIME type %r not eligible for auto archive generation",
"Archive: no - MIME type %r not eligible for auto archive generation",
mime_type,
)
return False
@@ -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(
@@ -61,7 +61,7 @@ class Command(PaperlessCommand):
)
table.add_column("Level", width=7, no_wrap=True)
table.add_column("Document", min_width=20)
table.add_column("Issue", ratio=1)
table.add_column("Issue", ratio=1, overflow="fold")
for doc_pk, doc_messages in messages.iter_messages():
if doc_pk is not None:
@@ -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,
+2 -2
View File
@@ -1329,7 +1329,7 @@ class PreConsumeTestCase(DirectoriesMixin, GetConsumerMixin, TestCase):
with self.get_consumer(self.test_file) as c:
c.run()
# Verify no pre-consume script subprocess was invoked
# (run_subprocess may still be called by _extract_text_for_archive_check)
# (run_subprocess may still be called by pdf_born_digital_text via pdftotext)
script_calls = [
call
for call in m.call_args_list
@@ -1354,7 +1354,7 @@ class PreConsumeTestCase(DirectoriesMixin, GetConsumerMixin, TestCase):
self.assertTrue(m.called)
# Find the call that invoked the pre-consume script
# (run_subprocess may also be called by _extract_text_for_archive_check)
# (run_subprocess may also be called by pdf_born_digital_text via pdftotext)
script_call = next(
call
for call in m.call_args_list
+14 -42
View File
@@ -134,60 +134,32 @@ class TestShouldProduceArchive:
assert should_produce_archive(parser, mime, Path("/tmp/doc")) is expected
@pytest.mark.parametrize(
("extracted_text", "expected"),
("born_digital", "expected"),
[
pytest.param(
"This is a born-digital PDF with lots of text content. " * 10,
False,
id="born-digital-long-text-skips-archive",
),
pytest.param(None, True, id="no-text-scanned-produces-archive"),
pytest.param("tiny", True, id="short-text-treated-as-scanned"),
pytest.param(True, False, id="born-digital-skips-archive"),
pytest.param(False, True, id="not-born-digital-produces-archive"),
],
)
def test_auto_pdf_archive_decision(
self,
mocker: MockerFixture,
settings,
extracted_text: str | None,
born_digital: bool, # noqa: FBT001
expected: bool, # noqa: FBT001
) -> None:
"""Archive decision tracks pdf_born_digital_text()'s verdict exactly.
should_produce_archive() defers entirely to pdf_born_digital_text()
for the has-real-text decision, so both callers of that predicate
(this function and RasterisedDocumentParser.parse()) always agree.
"""
settings.ARCHIVE_FILE_GENERATION = "auto"
mocker.patch("documents.consumer.is_tagged_pdf", return_value=False)
mocker.patch("documents.consumer.extract_pdf_text", return_value=extracted_text)
mocker.patch(
"documents.consumer.pdf_born_digital_text",
return_value=("some text", born_digital),
)
parser = _parser_instance(can_produce=True, requires_rendition=False)
assert (
should_produce_archive(parser, "application/pdf", Path("/tmp/doc.pdf"))
is expected
)
def test_tagged_pdf_skips_archive_in_auto_mode(
self,
mocker: MockerFixture,
settings,
) -> None:
"""Tagged PDFs (e.g. Word exports) with real text are treated as born-digital, even below PDF_TEXT_MIN_LENGTH."""
settings.ARCHIVE_FILE_GENERATION = "auto"
mocker.patch("documents.consumer.is_tagged_pdf", return_value=True)
mocker.patch("documents.consumer.extract_pdf_text", return_value="tiny")
parser = _parser_instance(can_produce=True, requires_rendition=False)
assert (
should_produce_archive(parser, "application/pdf", Path("/tmp/doc.pdf"))
is False
)
def test_tagged_pdf_without_text_produces_archive(
self,
mocker: MockerFixture,
settings,
) -> None:
"""A tagged PDF with no actual extractable text (e.g. some scanner firmware) is not
trusted as born-digital — the tag alone must not bypass OCR."""
settings.ARCHIVE_FILE_GENERATION = "auto"
mocker.patch("documents.consumer.is_tagged_pdf", return_value=True)
mocker.patch("documents.consumer.extract_pdf_text", return_value=None)
parser = _parser_instance(can_produce=True, requires_rendition=False)
assert (
should_produce_archive(parser, "application/pdf", Path("/tmp/doc.pdf"))
is True
)
+6 -21
View File
@@ -3,7 +3,6 @@ from __future__ import annotations
import importlib.resources
import logging
import os
import re
import shutil
import tempfile
from pathlib import Path
@@ -25,9 +24,9 @@ from paperless.config import OcrConfig
from paperless.models import CleanChoices
from paperless.models import ModeChoices
from paperless.models import OutputTypeChoices
from paperless.parsers.utils import PDF_TEXT_MIN_LENGTH
from paperless.parsers.utils import extract_pdf_text
from paperless.parsers.utils import is_tagged_pdf
from paperless.parsers.utils import is_born_digital_text
from paperless.parsers.utils import post_process_text
from paperless.parsers.utils import read_file_handle_unicode_errors
from paperless.version import __full_version_str__
@@ -510,10 +509,10 @@ class RasterisedDocumentParser:
if mime_type == "application/pdf":
text_original = self.extract_text(None, document_path)
has_text = text_original is not None and len(text_original) > 0
original_has_text = has_text and (
is_tagged_pdf(document_path, log=self.log)
or len(text_original) > PDF_TEXT_MIN_LENGTH
original_has_text = is_born_digital_text(
text_original,
document_path,
log=self.log,
)
else:
text_original = None
@@ -658,17 +657,3 @@ class RasterisedDocumentParser:
f"No text was found in {document_path}, the content will be empty.",
)
self.text = ""
def post_process_text(text: str | None) -> str | None:
if not text:
return None
collapsed_spaces = re.sub(r"([^\S\r\n]+)", " ", text)
no_leading_whitespace = re.sub(r"([\n\r]+)([^\S\n\r]+)", "\\1", collapsed_spaces)
no_trailing_whitespace = re.sub(r"([^\S\n\r]+)$", "", no_leading_whitespace)
# TODO: this needs a rework
# replace \0 prevents issues with saving to postgres.
# text may contain \0 when this character is present in PDF files.
return no_trailing_whitespace.strip().replace("\0", " ")
+82
View File
@@ -111,6 +111,88 @@ def extract_pdf_text(
return None
def post_process_text(text: str | None) -> str | None:
"""Normalize extracted PDF/OCR text: collapse whitespace, strip padding.
Returns ``None`` for ``None`` or whitespace-only input, so callers can
treat "no text" and "only layout padding" the same way.
"""
if not text:
return None
collapsed_spaces = re.sub(r"([^\S\r\n]+)", " ", text)
no_leading_whitespace = re.sub(r"([\n\r]+)([^\S\n\r]+)", "\\1", collapsed_spaces)
no_trailing_whitespace = re.sub(r"([^\S\n\r]+)$", "", no_leading_whitespace)
# replace \0 prevents issues with saving to postgres.
# text may contain \0 when this character is present in PDF files.
result = no_trailing_whitespace.strip().replace("\0", " ")
return result or None
def is_born_digital_text(
text: str | None,
path: Path,
log: logging.Logger | None = None,
) -> bool:
"""Decide whether already-extracted, normalized PDF text counts as born-digital.
This is the single source of truth for "does this PDF already have real
text", used both to decide whether to produce an archive file and to
decide whether OCR can be skipped. Both decisions must agree, or a
tagged-but-textless PDF can end up with no archive AND a forced OCR pass
(see GH #13387): raw ``pdftotext -layout`` output can be non-empty
(whitespace/form-feed padding) even when there is no real content, so
*text* must already be normalized via :func:`post_process_text`, not the
raw extraction.
Parameters
----------
text:
The normalized extracted text (or ``None``) to evaluate.
path:
Absolute path to the PDF file, used for the tagged-PDF check.
log:
Logger for warnings. Falls back to the module-level logger when omitted.
Returns
-------
bool
Whether the PDF counts as born-digital (has real text, and is either
tagged or exceeds ``PDF_TEXT_MIN_LENGTH``).
"""
if not text:
return False
return is_tagged_pdf(path, log=log) or len(text) > PDF_TEXT_MIN_LENGTH
def pdf_born_digital_text(
path: Path,
log: logging.Logger | None = None,
) -> tuple[str | None, bool]:
"""Extract a PDF's text and decide whether it should be treated as born-digital.
Convenience wrapper around :func:`is_born_digital_text` for callers that
don't already have the PDF's text extracted (e.g. the archive-generation
decision, which runs before any parser has touched the file).
Parameters
----------
path:
Absolute path to the PDF file.
log:
Logger for warnings. Falls back to the module-level logger when omitted.
Returns
-------
tuple[str | None, bool]
The normalized extracted text (or ``None``), and whether the PDF
counts as born-digital.
"""
text = post_process_text(extract_pdf_text(path, log=log))
return text, is_born_digital_text(text, path, log=log)
def read_file_handle_unicode_errors(
filepath: Path,
log: logging.Logger | None = None,
+17
View File
@@ -36,6 +36,23 @@ def samples_dir() -> Path:
return (Path(__file__).parent / "samples").resolve()
@pytest.fixture(scope="session")
def tagged_no_text_pdf_file(samples_dir: Path) -> Path:
"""Path to a tagged PDF whose only "text" is pdftotext layout padding.
Reproduces GH #13387: ``/MarkInfo /Marked true`` is set, but the only
extractable content is a form-feed byte, not real text. Lives here
rather than in parsers/conftest.py so both parser tests and
paperless/tests/test_parser_utils.py can use it.
Returns
-------
Path
Absolute path to ``tesseract/tagged-but-no-text.pdf``.
"""
return samples_dir / "tesseract" / "tagged-but-no-text.pdf"
@pytest.fixture(autouse=True)
def clean_registry() -> Generator[None, None, None]:
"""Reset the parser registry before and after every test.
@@ -21,7 +21,7 @@ from documents.parsers import run_convert
from paperless.models import ModeChoices
from paperless.parsers import ParserProtocol
from paperless.parsers.tesseract import RasterisedDocumentParser
from paperless.parsers.tesseract import post_process_text
from paperless.parsers.utils import is_tagged_pdf
if TYPE_CHECKING:
from pathlib import Path
@@ -151,36 +151,6 @@ class TestRasterisedDocumentParserLifecycle:
assert tempdir is not None and not tempdir.exists()
# ---------------------------------------------------------------------------
# post_process_text
# ---------------------------------------------------------------------------
class TestPostProcessText:
@pytest.mark.parametrize(
("source", "expected"),
[
pytest.param(
"simple string",
"simple string",
id="collapse-spaces",
),
pytest.param(
"simple newline\n testing string",
"simple newline\ntesting string",
id="preserve-newline",
),
pytest.param(
"utf-8 строка с пробелами в конце ", # noqa: RUF001
"utf-8 строка с пробелами в конце", # noqa: RUF001
id="utf8-trailing-spaces",
),
],
)
def test_post_process_text(self, source: str, expected: str) -> None:
assert post_process_text(source) == expected
# ---------------------------------------------------------------------------
# Page count
# ---------------------------------------------------------------------------
@@ -910,25 +880,25 @@ class TestSkipArchive:
self,
mocker: MockerFixture,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
tagged_no_text_pdf_file: Path,
) -> None:
"""
GIVEN:
- A PDF that reports itself as tagged (/MarkInfo /Marked true) but
has no actual extractable text (some scanner firmware produces
this — see GitHub issue #13349)
- A real PDF that reports itself as tagged (/MarkInfo /Marked
true) but whose only pdftotext output is layout padding (a
lone form-feed byte), not real text (see GitHub issue #13387,
originally reported against #13349's tagged-PDF handling)
- Mode: auto, produce_archive=False
WHEN:
- Document is parsed
THEN:
- The tag alone is not trusted as "has text"; OCRmyPDF still runs
"""
assert is_tagged_pdf(tagged_no_text_pdf_file) is True
tesseract_parser.settings.mode = ModeChoices.AUTO
mocker.patch("paperless.parsers.tesseract.is_tagged_pdf", return_value=True)
mocker.patch.object(tesseract_parser, "extract_text", return_value=None)
mock_ocr = mocker.patch("ocrmypdf.ocr")
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-images.pdf",
tagged_no_text_pdf_file,
"application/pdf",
produce_archive=False,
)
+110
View File
@@ -4,10 +4,18 @@ from __future__ import annotations
import codecs
from pathlib import Path
from typing import TYPE_CHECKING
import pytest
from paperless.parsers.utils import is_tagged_pdf
from paperless.parsers.utils import pdf_born_digital_text
from paperless.parsers.utils import post_process_text
from paperless.parsers.utils import read_file_handle_unicode_errors
if TYPE_CHECKING:
from pytest_mock import MockerFixture
SAMPLES = Path(__file__).parent / "samples" / "tesseract"
@@ -60,3 +68,105 @@ class TestIsTaggedPdf:
bad = tmp_path / "bad.pdf"
bad.write_bytes(b"not a pdf")
assert is_tagged_pdf(bad) is False
class TestPostProcessText:
@pytest.mark.parametrize(
("source", "expected"),
[
pytest.param(
"simple string",
"simple string",
id="collapse-spaces",
),
pytest.param(
"simple newline\n testing string",
"simple newline\ntesting string",
id="preserve-newline",
),
pytest.param(
"utf-8 строка с пробелами в конце ", # noqa: RUF001
"utf-8 строка с пробелами в конце", # noqa: RUF001
id="utf8-trailing-spaces",
),
pytest.param(None, None, id="none-input"),
pytest.param("", None, id="empty-string"),
pytest.param(" \n\x0c \n ", None, id="whitespace-and-formfeed-only"),
],
)
def test_post_process_text(
self,
source: str | None,
expected: str | None,
) -> None:
assert post_process_text(source) == expected
class TestPdfBornDigitalText:
"""Regression coverage for GH #13387.
should_produce_archive() and RasterisedDocumentParser.parse() must agree
on whether a PDF has real text, so both go through this one function.
"""
@pytest.mark.parametrize(
("extracted", "tagged", "expected_text", "expected_born_digital"),
[
pytest.param("tiny", True, "tiny", True, id="tagged-with-real-text"),
pytest.param("tiny", False, "tiny", False, id="untagged-below-min-length"),
pytest.param(
"x" * 51,
False,
"x" * 51,
True,
id="untagged-above-min-length",
),
pytest.param(None, True, None, False, id="tagged-but-no-text"),
],
)
def test_born_digital_decision(
self,
mocker: MockerFixture,
tmp_path: Path,
extracted: str | None,
tagged: bool, # noqa: FBT001
expected_text: str | None,
expected_born_digital: bool, # noqa: FBT001
) -> None:
"""
GIVEN:
- A PDF whose pdftotext output and /MarkInfo tag status vary
WHEN:
- pdf_born_digital_text() is called
THEN:
- The normalized text and born-digital verdict match; the tag
alone never counts as "has text"
"""
mocker.patch(
"paperless.parsers.utils.extract_pdf_text",
return_value=extracted,
)
mocker.patch("paperless.parsers.utils.is_tagged_pdf", return_value=tagged)
text, born_digital = pdf_born_digital_text(tmp_path / "doc.pdf")
assert text == expected_text
assert born_digital is expected_born_digital
def test_tagged_but_textless_pdf_is_not_born_digital(
self,
tagged_no_text_pdf_file: Path,
) -> None:
"""
GIVEN:
- A real PDF that is tagged (/MarkInfo /Marked true) but whose
only "text" is layout padding (a stray form-feed byte)
WHEN:
- pdf_born_digital_text() is called with no mocking
THEN:
- The normalized text is None and the PDF is not treated as
born-digital. The raw, unnormalized pdftotext output is
non-empty for this file, which is exactly what caused the
archive decision to disagree with the OCR decision in #13387.
"""
text, born_digital = pdf_born_digital_text(tagged_no_text_pdf_file)
assert text is None
assert born_digital is False
+31 -146
View File
@@ -1,4 +1,3 @@
import enum
import logging
from collections.abc import Iterable
from contextlib import contextmanager
@@ -145,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.
@@ -187,52 +168,6 @@ def write_store(embed_model_name: str | None = None):
yield store
class MigrationCheckResult(enum.Enum):
"""Outcome of _check_and_run_migrations().
CURRENT: no migration was pending, or a pending structural migration
was applied successfully -- safe to write.
REEMBED_REQUIRED: a pending migration needs fresh embeddings, which is
never triggered automatically -- the caller must force a rebuild.
DEFERRED: a migration was pending but could not run because active
index readers did not drain within LLM_INDEX_COMPACTION_LOCK_TIMEOUT --
the store is still on its old schema. Callers must NOT proceed to
write: collapsing this into the same falsy value as CURRENT (as a
plain bool return once did) would let a write proceed against an
unmigrated schema.
"""
CURRENT = "current"
REEMBED_REQUIRED = "reembed_required"
DEFERRED = "deferred"
def _check_and_run_migrations(
store: "PaperlessSqliteVecVectorStore",
) -> MigrationCheckResult:
"""Run any pending structural migrations, reporting the outcome as a
tri-state result. 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 MigrationCheckResult.CURRENT
result = _with_exclusive_access(
"migration check",
store.check_and_run_migrations,
)
if result is None:
return MigrationCheckResult.DEFERRED
return (
MigrationCheckResult.REEMBED_REQUIRED
if result
else MigrationCheckResult.CURRENT
)
def _safe_related_name(document: Document, field: str) -> str | None:
"""
Returns the ``name`` of a related object (correspondent, document_type,
@@ -404,21 +339,20 @@ def update_llm_index(
happens, since a rebuild always covers the whole library regardless.
"""
with write_store() as store:
migration_result = _check_and_run_migrations(store)
if migration_result is MigrationCheckResult.REEMBED_REQUIRED:
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.",
)
rebuild = True
elif migration_result is MigrationCheckResult.DEFERRED:
logger.info(
"Skipping LLM index update: migration check deferred while "
"index readers are active; will retry next run.",
)
return (
"Skipping LLM index update: migration check deferred; "
"will retry next run."
)
documents = Document.objects.select_related(
"correspondent",
"document_type",
@@ -434,11 +368,12 @@ def update_llm_index(
config = AIConfig()
model_name = get_configured_model_name(config)
if not rebuild:
if not rebuild and llm_index_exists():
with read_store() as store:
if store.table_exists() and store.config_mismatch(model_name):
logger.warning("Embedding model changed; forcing LLM index rebuild.")
rebuild = True
config_mismatch = store.config_mismatch(model_name)
if config_mismatch:
logger.warning("Embedding model changed; forcing LLM index rebuild.")
rebuild = True
if no_documents:
logger.warning("No documents found to index.")
@@ -477,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
@@ -492,82 +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:
migration_result = _check_and_run_migrations(store)
if migration_result is MigrationCheckResult.REEMBED_REQUIRED:
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
if migration_result is MigrationCheckResult.DEFERRED:
logger.info(
"Skipping incremental LLM index update for document %s: "
"migration check deferred while index readers are active; "
"will retry on the next write.",
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:
migration_result = _check_and_run_migrations(store)
if migration_result is MigrationCheckResult.REEMBED_REQUIRED:
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.",
)
elif migration_result is MigrationCheckResult.DEFERRED:
logger.info(
"LLM index migration check deferred while index readers are "
"active; will retry next run.",
)
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:
migration_result = _check_and_run_migrations(store)
if migration_result is MigrationCheckResult.REEMBED_REQUIRED:
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
if migration_result is MigrationCheckResult.DEFERRED:
logger.info(
"Skipping removal of document %s from the LLM index: "
"migration check deferred while index readers are active; "
"will retry on the next write.",
document.id,
)
return
store.delete(str(document.id))
-60
View File
@@ -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] = []
@@ -1,118 +0,0 @@
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 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(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,
),
)
-245
View File
@@ -1,245 +0,0 @@
"""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 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:
"""Add ``count`` to the stored counter in one SQL 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. This only avoids a read-then-write race within this single
statement; it does not make the counter safe against concurrent
writers in general (callers still rely on the write FileLock for
that). 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))
-296
View File
@@ -1,4 +1,3 @@
import logging
from pathlib import Path
from unittest.mock import MagicMock
from unittest.mock import patch
@@ -9,7 +8,6 @@ from django.db import connection
from django.test import override_settings
from django.test.utils import CaptureQueriesContext
from django.utils import timezone
from filelock import Timeout
from llama_index.core.schema import MetadataMode
from documents.models import Correspondent
@@ -254,45 +252,6 @@ def test_update_llm_index_rebuilds_on_model_name_change(
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
def test_update_llm_index_partial_update(
temp_llm_index_dir: Path,
@@ -778,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(
@@ -799,83 +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_add_or_update_document_skips_write_when_migration_check_deferred(
self,
temp_llm_index_dir: Path,
mock_embed_model: FakeEmbedding,
mocker: pytest_mock.MockerFixture,
) -> None:
"""A migration check that times out waiting for readers to drain
must be treated the same as a pending migration -- proceeding to
write would target a store still on its old schema. Regression
test for the tri-state fix: a bare bool collapsed this outcome
into the same falsy value as "already current".
"""
mock_store = MagicMock()
mock_store.has_pending_migration.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),
),
)
mocker.patch(
"paperless_ai.indexing._exclude_readers",
side_effect=Timeout("test"),
)
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(
@@ -890,59 +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_remove_document_skips_write_when_migration_check_deferred(
self,
temp_llm_index_dir: Path,
mocker: pytest_mock.MockerFixture,
) -> None:
"""A migration check deferred by a reader-lock timeout must block
the delete too, for the same reason as the incremental-update path.
"""
mock_store = MagicMock()
mock_store.has_pending_migration.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),
),
)
mocker.patch(
"paperless_ai.indexing._exclude_readers",
side_effect=Timeout("test"),
)
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,
@@ -966,35 +800,6 @@ class TestLlmIndexLocking:
mock_store.drop_table.assert_called_once()
def test_update_llm_index_skips_when_migration_check_deferred(
self,
temp_llm_index_dir: Path,
mocker: pytest_mock.MockerFixture,
) -> None:
"""A migration check deferred by a reader-lock timeout must short-
circuit before the second write_store() block (document scanning,
add/upsert, compaction) ever runs -- that block would otherwise
write against a store still on its old schema.
"""
mock_store = MagicMock()
mock_store.has_pending_migration.return_value = True
write_store_mock = mocker.patch(
"paperless_ai.indexing.write_store",
return_value=mocker.MagicMock(
__enter__=mocker.MagicMock(return_value=mock_store),
__exit__=mocker.MagicMock(return_value=False),
),
)
mocker.patch(
"paperless_ai.indexing._exclude_readers",
side_effect=Timeout("test"),
)
result = indexing.update_llm_index(rebuild=False)
assert "deferred" in result
write_store_mock.assert_called_once()
@pytest.mark.django_db
@pytest.mark.django_db
@@ -1044,107 +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
def test_logs_info_when_migration_check_deferred(
self,
mocker: pytest_mock.MockerFixture,
caplog: pytest.LogCaptureFixture,
) -> None:
"""
GIVEN:
- AI/LLM index support is enabled
- A pending migration cannot run because readers are active
WHEN:
- llm_index_migrate() is called
THEN:
- An info line notes the deferral, not the re-embed warning
"""
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
write_store_cm = mocker.patch("paperless_ai.indexing.write_store")
write_store_cm.return_value.__enter__.return_value = store_mock
mocker.patch(
"paperless_ai.indexing._exclude_readers",
side_effect=Timeout("test"),
)
with caplog.at_level(logging.INFO, logger="paperless_ai.indexing"):
indexing.llm_index_migrate()
assert "deferred" in caplog.text
assert "requires re-embedding" not in caplog.text
@pytest.mark.django_db
class TestQuerySimilarDocuments:
def test_query_similar_documents_respects_allowed_ids(
-340
View File
@@ -1,340 +0,0 @@
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
+97 -478
View File
@@ -1,35 +1,28 @@
import inspect
import sqlite3
from collections.abc import Generator
from pathlib import Path
import pytest
import sqlite_vec
from llama_index.core.schema import TextNode
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 MetadataFilters
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.migrations import m0001_v1_to_v2
from paperless_ai.tables import DocumentChunksTable
from paperless_ai.tables import DocumentMetaTable
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
from paperless_ai.vector_store import _pack
DIM = 16
def make_node(
node_id: str,
document_id: int,
document_id: str,
*,
modified: str = "2026-06-10T00:00:00",
seed: float = 0.0,
@@ -66,13 +59,13 @@ def _query(
)
def _eq_filter(key: str, value: int):
def _eq_filter(key: str, value: str):
return MetadataFilters(
filters=[MetadataFilter(key=key, operator=FilterOperator.EQ, value=value)],
)
def _in_filter(document_ids: list[int]):
def _in_filter(document_ids: list[str]):
return MetadataFilters(
filters=[
MetadataFilter(
@@ -84,7 +77,7 @@ def _in_filter(document_ids: list[int]):
)
def _ne_filter(document_id: int):
def _ne_filter(document_id: str):
return MetadataFilters(
filters=[
MetadataFilter(
@@ -98,11 +91,11 @@ def _ne_filter(document_id: int):
class TestCrud:
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"]
result = _query(store, node.embedding, top_k=1)
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
assert result.similarities[0] == pytest.approx(1.0)
@@ -115,58 +108,58 @@ class TestCrud:
assert not store.table_exists()
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.delete(1)
store.add([make_node("a1", "1"), make_node("a2", "1"), make_node("b1", "2")])
store.delete("1")
result = _query(store, [0.0] * DIM, top_k=10)
assert result.ids == ["b1"]
def test_query_with_in_filter_scopes_results(self, store) -> None:
store.add(
[
make_node("a1", 1, seed=0.0),
make_node("b1", 2, seed=1.0),
make_node("c1", 3, seed=2.0),
make_node("a1", "1", seed=0.0),
make_node("b1", "2", seed=1.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"]
def test_query_respects_top_k_with_filter(self, store) -> None:
# k semantics: global top-k even with IN filters (document_id is a
# metadata column, not a partition key -- see design doc).
store.add(
[make_node(f"n{i}", i % 4, seed=float(i)) for i in range(12)],
[make_node(f"n{i}", str(i % 4), seed=float(i)) for i in range(12)],
)
result = _query(
store,
[0.0] * DIM,
top_k=3,
filters=_in_filter([0, 1, 2, 3]),
filters=_in_filter(["0", "1", "2", "3"]),
)
assert len(result.ids) == 3
assert result.similarities == sorted(result.similarities, reverse=True)
def test_get_nodes_filter_and_empty_paths(self, store) -> None:
assert store.get_nodes(filters=_in_filter([1])) == [] # no table yet
store.add([make_node("a1", 1), make_node("b1", 2)])
nodes = store.get_nodes(filters=_in_filter([1]))
assert store.get_nodes(filters=_in_filter(["1"])) == [] # no table yet
store.add([make_node("a1", "1"), make_node("b1", "2")])
nodes = store.get_nodes(filters=_in_filter(["1"]))
assert [n.node_id for n in nodes] == ["a1"]
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:
store.add(
[
make_node("a1", 1, seed=0.0),
make_node("b1", 2, seed=1.0),
make_node("c1", 3, seed=2.0),
make_node("a1", "1", seed=0.0),
make_node("b1", "2", seed=1.0),
make_node("c1", "3", seed=2.0),
],
)
result = _query(
store,
[0.0] * DIM,
top_k=10,
filters=_eq_filter("document_id", 2),
filters=_eq_filter("document_id", "2"),
)
assert result.ids == ["b1"]
@@ -175,7 +168,7 @@ class TestCrud:
store.get_nodes(node_ids=["x"])
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:
assert reopened.table_exists()
assert reopened.vector_dim() == DIM
@@ -183,98 +176,23 @@ class TestCrud:
def test_table_exists_and_drop(self, store) -> None:
assert not store.table_exists()
store.add([make_node("a1", 1)])
store.add([make_node("a1", "1")])
assert store.table_exists()
store.drop_table()
assert not store.table_exists()
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:
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 params == [1]
assert params == ["1"]
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(
_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",
]
@@ -288,7 +206,7 @@ class TestBuildWhere:
MetadataFilter(
key="document_id",
operator=FilterOperator.EQ,
value=1,
value="1",
),
],
)
@@ -297,13 +215,13 @@ class TestBuildWhere:
assert params == []
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(
filters=[
MetadataFilter(
key="document_id",
operator=FilterOperator.EQ,
value=1,
value="1",
),
],
)
@@ -316,19 +234,19 @@ class TestBuildWhere:
class TestUpsert:
def test_upsert_replaces_and_prunes_stale_chunks(self, store) -> None:
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)
assert sorted(result.ids) == ["d1new", "d2c1"]
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"]
def test_upsert_empty_nodes_removes_document(self, store) -> None:
store.add([make_node("a1", 1), make_node("b1", 2)])
store.upsert_document(1, [])
store.add([make_node("a1", "1"), make_node("b1", "2")])
store.upsert_document("1", [])
assert _query(store, [0.0] * DIM, top_k=10).ids == ["b1"]
def test_upsert_is_atomic_for_concurrent_readers(
@@ -337,16 +255,16 @@ class TestUpsert:
tmp_path: Path,
) -> None:
"""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:
store.upsert_document(1, [make_node("a3", 1)])
ids = [n.node_id for n in reader.get_nodes(filters=_in_filter([1]))]
store.upsert_document("1", [make_node("a3", "1")])
ids = [n.node_id for n in reader.get_nodes(filters=_in_filter(["1"]))]
assert ids == ["a3"]
class TestMetadataCoercion:
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
store.add([node]) # must not raise (vec0 rejects NULL metadata)
assert store.get_modified_times() == {"1": ""}
@@ -365,7 +283,7 @@ class TestModelNameTracking:
uri=str(tmp_path),
embed_model_name="model-a",
) as store:
store.add([make_node("a1", 1)])
store.add([make_node("a1", "1")])
assert store.stored_model_name() == "model-a"
with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reopened:
assert reopened.stored_model_name() == "model-a"
@@ -376,7 +294,7 @@ class TestModelNameTracking:
embed_model_name="model-a",
) as store:
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 store.config_mismatch("model-b")
@@ -385,7 +303,7 @@ class TestModelNameTracking:
tmp_path: Path,
) -> None:
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")
@@ -396,9 +314,9 @@ class TestGetModifiedTimes:
def test_returns_one_entry_per_document(self, store) -> None:
store.add(
[
make_node("a1", 1, modified="2026-01-01T00:00:00"),
make_node("a2", 1, modified="2026-01-01T00:00:00"),
make_node("b1", 2, modified="2026-02-02T00:00:00"),
make_node("a1", "1", modified="2026-01-01T00:00:00"),
make_node("a2", "1", modified="2026-01-01T00:00:00"),
make_node("b1", "2", modified="2026-02-02T00:00:00"),
],
)
assert store.get_modified_times() == {
@@ -423,35 +341,37 @@ class TestCompact:
def _churn(self, store, cycles: int) -> None:
for i in range(cycles):
store.upsert_document(
1,
[make_node(f"gen{i}-{j}", 1, seed=float(j)) for j in range(20)],
"1",
[make_node(f"gen{i}-{j}", "1", seed=float(j)) for j in range(20)],
)
def test_compact_noop_below_threshold(self, store) -> None:
store.add([make_node("a1", 1)])
store.add([make_node("a1", "1")])
store.compact()
assert _query(store, [0.0] * DIM, top_k=1).ids == ["a1"]
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)
before = {
n.node_id: n.metadata for n in store.get_nodes(filters=_in_filter([1, 2]))
n.node_id: n.metadata
for n in store.get_nodes(filters=_in_filter(["1", "2"]))
}
store.compact(force=True)
after = {
n.node_id: n.metadata for n in store.get_nodes(filters=_in_filter([1, 2]))
n.node_id: n.metadata
for n in store.get_nodes(filters=_in_filter(["1", "2"]))
}
assert after == before
assert self._bloat_ratio(store) == pytest.approx(1.0)
# 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
# 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
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)
assert self._bloat_ratio(store) > 2
store.compact()
@@ -473,7 +393,7 @@ class TestCompact:
but a concurrent reader keeps -wal/-shm alive, so the cleanup must
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"
held: list[sqlite3.Connection] = []
@@ -508,76 +428,17 @@ class TestCompact:
A tiny batch size forces several fetchmany()/executemany() cycles so a
regression in the streaming loop (dropped tail, off-by-one) surfaces.
"""
monkeypatch.setattr("paperless_ai.vector_store.BATCH_SIZE", 3)
store.add([make_node(f"n{i}", 1, seed=float(i)) for i in range(10)])
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.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 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
def test_compact_on_unmigrated_store_is_noop(
self,
store: PaperlessSqliteVecVectorStore,
mocker: MockerFixture,
) -> None:
"""
GIVEN:
- A store whose schema_version has been forced behind
SCHEMA_VERSION (has_pending_migration() is True)
WHEN:
- compact(force=True) is called directly, without the caller
having run check_and_run_migrations() first
THEN:
- compact() is a safe no-op: no file-swap rebuild is attempted
at all (asserted via a spy on _rebuild_into, since schema_version
alone is not a reliable signal -- a rebuild would otherwise
copy the stale schema_version across unchanged, making a
before/after equality check pass even when a rebuild *did*
happen). Rebuilding an unmigrated store would silently lose
document_meta and leave the swapped-in file claiming the old
schema_version -- see the class docstring rationale.
"""
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
rebuild_spy = mocker.spy(PaperlessSqliteVecVectorStore, "_rebuild_into")
store.compact(force=True)
rebuild_spy.assert_not_called()
row = store.client.execute(
"SELECT value FROM index_meta WHERE key = 'schema_version'",
).fetchone()
assert int(row["value"]) == 0
class TestDbFile:
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()
def test_wal_mode_enabled(self, store) -> None:
@@ -587,26 +448,7 @@ class TestDbFile:
class TestMigrations:
"""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.
The fake structural migrations' apply() fixtures (see
test_structural_migration_copies_rows_and_updates_version and
test_stop_at_reembed_boundary below) only populate the rebuilt vec0
table itself -- they never insert into document_chunks/document_meta on
the destination connection. That's fine here: these tests exist to
verify the generic dispatch mechanism (version bookkeeping, structural-
vs-reembed branching), not full schema correctness of a rebuilt store;
the real migration's data completeness is covered separately by
TestV1ToV2Migration.
"""
"""Tests for the schema migration machinery."""
def _schema_version(self, store: PaperlessSqliteVecVectorStore) -> int | None:
row = store.client.execute(
@@ -615,21 +457,21 @@ class TestMigrations:
return int(row[0]) if row else 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
def test_check_migrations_no_table_returns_false(self, store) -> None:
assert store.check_and_run_migrations() is False
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
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(
from_version=SCHEMA_VERSION,
to_version=SCHEMA_VERSION + 1,
from_version=1,
to_version=2,
kind="re-embed",
description="test re-embed",
)
@@ -638,7 +480,7 @@ class TestMigrations:
from paperless_ai import vector_store as vs_mod
original = vs_mod.SCHEMA_VERSION
vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 1
vs_mod.SCHEMA_VERSION = 2
result = store.check_and_run_migrations()
finally:
MIGRATIONS.remove(migration)
@@ -650,7 +492,7 @@ class TestMigrations:
store,
tmp_path: Path,
) -> None:
store.add([make_node("a1", 1), make_node("b1", 2)])
store.add([make_node("a1", "1"), make_node("b1", "2")])
def apply(
src: sqlite3.Connection,
@@ -669,7 +511,7 @@ class TestMigrations:
(str(dim),),
)
rows = src.execute(
"SELECT id, document_id, node_content, embedding "
"SELECT id, document_id, modified, node_content, embedding "
f"FROM {DEFAULT_TABLE_NAME}",
).fetchall()
dst.execute("BEGIN IMMEDIATE")
@@ -680,8 +522,8 @@ class TestMigrations:
[
(
r["id"],
str(r["document_id"]),
"",
r["document_id"],
r["modified"],
r["node_content"],
bytes(r["embedding"]),
)
@@ -696,8 +538,8 @@ class TestMigrations:
dst.execute("COMMIT")
migration = Migration(
from_version=SCHEMA_VERSION,
to_version=SCHEMA_VERSION + 1,
from_version=1,
to_version=2,
kind="structural",
description="test structural",
apply=apply,
@@ -707,29 +549,28 @@ class TestMigrations:
from paperless_ai import vector_store as vs_mod
original = vs_mod.SCHEMA_VERSION
vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 1
vs_mod.SCHEMA_VERSION = 2
result = store.check_and_run_migrations()
finally:
MIGRATIONS.remove(migration)
vs_mod.SCHEMA_VERSION = original
assert result is False
assert self._schema_version(store) == SCHEMA_VERSION + 1
assert self._schema_version(store) == 2
ids = {n.node_id for n in store.get_nodes()}
assert ids == {"a1", "b1"}
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
store.compact(force=True)
assert self._schema_version(store) == SCHEMA_VERSION
def test_stop_at_reembed_boundary(self, store) -> None:
# Registry: structural v(N+1), re-embed v(N+2), structural v(N+3),
# where N = SCHEMA_VERSION. Only v(N+1) should apply; the re-embed
# boundary must stop execution before v(N+3) runs, and the stored
# version must stay at N+1.
store.add([make_node("a1", 1), make_node("b1", 2)])
# Registry: structural v2, re-embed v3, structural v4.
# Only v2 should apply; the re-embed boundary must stop execution
# before v4 runs, and the stored version must stay at 2.
store.add([make_node("a1", "1"), make_node("b1", "2")])
def copy_apply(
src: sqlite3.Connection,
@@ -748,7 +589,7 @@ class TestMigrations:
(str(dim),),
)
rows = src.execute(
"SELECT id, document_id, node_content, embedding "
"SELECT id, document_id, modified, node_content, embedding "
f"FROM {DEFAULT_TABLE_NAME}",
).fetchall()
dst.execute("BEGIN IMMEDIATE")
@@ -759,8 +600,8 @@ class TestMigrations:
[
(
r["id"],
str(r["document_id"]),
"",
r["document_id"],
r["modified"],
r["node_content"],
bytes(r["embedding"]),
)
@@ -771,23 +612,23 @@ class TestMigrations:
migrations = [
Migration(
from_version=SCHEMA_VERSION,
to_version=SCHEMA_VERSION + 1,
from_version=1,
to_version=2,
kind="structural",
description="v(N+1) structural",
description="v2 structural",
apply=copy_apply,
),
Migration(
from_version=SCHEMA_VERSION + 1,
to_version=SCHEMA_VERSION + 2,
from_version=2,
to_version=3,
kind="re-embed",
description="v(N+2) re-embed boundary",
description="v3 re-embed boundary",
),
Migration(
from_version=SCHEMA_VERSION + 2,
to_version=SCHEMA_VERSION + 3,
from_version=3,
to_version=4,
kind="structural",
description="v(N+3) structural - must not run",
description="v4 structural - must not run",
apply=copy_apply,
),
]
@@ -796,7 +637,7 @@ class TestMigrations:
from paperless_ai import vector_store as vs_mod
original = vs_mod.SCHEMA_VERSION
vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 3
vs_mod.SCHEMA_VERSION = 4
result = store.check_and_run_migrations()
finally:
for m in migrations:
@@ -804,226 +645,4 @@ class TestMigrations:
vs_mod.SCHEMA_VERSION = original
assert result is True
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
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)
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)
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.
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
assert self._schema_version(store) == 2
+197 -305
View File
@@ -2,13 +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 NamedTuple
from typing import Literal
import sqlite_vec
from llama_index.core.bridge.pydantic import PrivateAttr
@@ -23,14 +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
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")
DB_FILENAME = "llmindex.db"
@@ -39,7 +34,7 @@ DEFAULT_TABLE_NAME = "documents"
# Current schema version. Written to index_meta at table creation and bumped
# whenever a Migration is added to MIGRATIONS. check_and_run_migrations() uses
# this to decide which migrations to run on an existing store.
SCHEMA_VERSION = 2
SCHEMA_VERSION = 1
# compact(): rebuild when the cumulative rowid count exceeds this multiple of
# the live row count. DELETEs on vec0 tables never reclaim space (upstream
@@ -47,32 +42,47 @@ SCHEMA_VERSION = 2
# a rebuild copies the live rows into a fresh table.
COMPACT_BLOAT_RATIO = 2.0
# Number of rows fetched/copied per batch whenever this module streams rows
# instead of materializing them all at once, keeping memory bounded regardless
# of index size -- used by compact()'s rebuild, m0001_v1_to_v2's migration
# copy, and DocumentMetaTable.copy_all(). No longer compact()-specific, hence
# the plain name.
BATCH_SIZE = 500
# compact(): number of rows copied per executemany() when rebuilding the file.
# Rows are streamed from the source cursor in batches of this size rather than
# materialized all at once, keeping memory bounded regardless of index size.
COMPACT_BATCH_SIZE = 500
# Filterable vec0 metadata columns. _build_where() only ever receives filter
# keys we construct ourselves, but allowlisting keeps SQL identifiers safe by
# construction. "modified" is not here: it is never filtered on, and as of
# schema v2 it isn't even a vec0 column anymore (see document_meta).
_FILTER_COLUMNS = frozenset({"document_id"})
# construction.
_FILTER_COLUMNS = frozenset({"document_id", "modified"})
class _Row(NamedTuple):
"""One node, ready to write. ``modified`` is not a vec0 column (see
document_meta) -- it rides along here because every row-producing call
site needs both the vec0 insert values and the document_meta upsert
value from the same node.
@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).
"""
chunk_id: str
document_id: int
modified: str
node_content: str
embedding: bytes
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:
@@ -83,30 +93,14 @@ def _unpack(blob: bytes) -> list[float]:
return list(struct.unpack(f"{len(blob) // 4}f", blob))
_INSERT = (
"INSERT INTO "
+ 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.
def _build_where(filters: MetadataFilters | None) -> tuple[str, list[str]]:
"""Translate the EQ / IN / NE filters we use into a parameterized SQL clause
on vec0 metadata columns. Returns ("", []) when there is nothing to filter.
"""
if filters is None or not filters.filters:
return "", []
clauses: list[str] = []
params: list[int] = []
params: list[str] = []
for f in filters.filters:
# filters.filters is Union[MetadataFilter, ExactMatchFilter, MetadataFilters];
# we only build MetadataFilter entries, so skip anything else at runtime.
@@ -115,7 +109,7 @@ def _build_where(filters: MetadataFilters | None) -> tuple[str, list[int]]:
if f.key not in _FILTER_COLUMNS: # pragma: no cover - we build the keys
raise NotImplementedError(f"Unsupported filter column: {f.key}")
if f.operator == FilterOperator.IN:
values = [int(v) for v in f.value] # type: ignore[union-attr]
values = [str(v) for v in f.value] # type: ignore[union-attr] # value is list when operator is IN
if not values: # pragma: no cover
clauses.append("1 = 0")
continue
@@ -124,10 +118,10 @@ def _build_where(filters: MetadataFilters | None) -> tuple[str, list[int]]:
params.extend(values)
elif f.operator == FilterOperator.EQ:
clauses.append(f"{f.key} = ?")
params.append(int(f.value))
params.append(str(f.value))
elif f.operator == FilterOperator.NE:
clauses.append(f"{f.key} != ?")
params.append(int(f.value))
params.append(str(f.value))
else: # pragma: no cover - we only ever build EQ/IN/NE filters
raise NotImplementedError(f"Unsupported filter operator: {f.operator}")
if not clauses:
@@ -192,21 +186,9 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
conn.enable_load_extension(False) # noqa: FBT003
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
IndexMetaTable.create(conn)
# 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)
conn.execute(
"CREATE TABLE IF NOT EXISTS index_meta (key TEXT PRIMARY KEY, value TEXT)",
)
return conn
@property
@@ -241,6 +223,24 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
else:
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:
return (
self._conn.execute(
@@ -253,19 +253,18 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
def vector_dim(self) -> int | None:
if not self.table_exists():
return None
return IndexMetaTable.get_dim(self._conn)
value = self._meta_get("dim")
return int(value) if value else None
def drop_table(self) -> None:
self._conn.execute("DROP TABLE IF EXISTS " + DEFAULT_TABLE_NAME)
self._conn.execute("DELETE FROM index_meta")
DocumentChunksTable.delete_all(self._conn)
DocumentMetaTable.delete_all(self._conn)
def stored_model_name(self) -> str | None:
"""Return the embedding model name recorded at table creation, or None."""
if not self.table_exists():
return None
return IndexMetaTable.get_embed_model(self._conn)
return self._meta_get("embed_model")
def config_mismatch(self, model_name: str) -> bool:
"""True when the stored model name differs from ``model_name``.
@@ -283,17 +282,14 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
# document_id is deliberately a metadata column, NOT a partition key:
# partition keys change KNN `k` to per-partition semantics under IN
# filters (asg017/sqlite-vec#142); metadata columns give a correct
# 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.
# global top-k.
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,"
+ " document_id TEXT,"
+ " modified TEXT,"
+ " +node_content TEXT,"
+ " embedding float["
+ str(int(dim))
@@ -303,78 +299,37 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
def _create_table(self, dim: int) -> None:
self._create_vec_table(self._conn, dim)
IndexMetaTable.set_dim(self._conn, dim)
IndexMetaTable.set_schema_version(self._conn, SCHEMA_VERSION)
self._meta_set("dim", str(dim))
self._meta_set("schema_version", str(SCHEMA_VERSION))
if self._embed_model_name:
IndexMetaTable.set_embed_model(self._conn, self._embed_model_name)
self._meta_set("embed_model", self._embed_model_name)
def _ensure_table(self, dim: int, *, table_exists: bool) -> None:
if not table_exists:
def _ensure_table(self, dim: int) -> None:
if not self.table_exists():
self._create_table(dim)
def _row(self, node: BaseNode) -> _Row:
def _row(self, node: BaseNode) -> tuple[str, str, str, str, bytes]:
meta = node_to_metadata_dict(
node,
remove_text=False,
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")
return _Row(
chunk_id=node.node_id,
# document_id is required -- int(None) raises TypeError and
# int("not-a-number") raises ValueError, both intentional:
# fail loudly on a malformed/missing document_id rather than
# silently indexing a chunk with no owning document. modified,
# below, still uses the str(x or "") sentinel pattern because a
# missing modified value is legitimate (vec0 no longer even
# stores it -- see document_meta), whereas document_id must
# always be present.
document_id=int(document_id),
modified=str(node.metadata.get("modified") or ""),
node_content=json.dumps(meta),
embedding=_pack(node.get_embedding()),
return (
node.node_id,
str(document_id or ""),
str(node.metadata.get("modified") or ""),
json.dumps(meta),
_pack(node.get_embedding()),
)
def _index_chunks(self, rows: list[_Row]) -> None:
"""Record each row's (chunk_id, document_id) in document_chunks, and
each row's (document_id, modified) in document_meta -- deduped
within the batch, since every chunk of a document shares the same
modified value -- kept in lockstep with every insert into the vec0
table.
"""
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)
_INSERT = (
"INSERT INTO "
+ DEFAULT_TABLE_NAME
+ " (id, document_id, modified, node_content, embedding) VALUES (?, ?, ?, ?, ?)"
)
def _increment_total_inserts(self, count: int) -> None:
"""Increment the cumulative insert counter stored in index_meta.
@@ -384,56 +339,48 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
live_rows exceeds COMPACT_BLOAT_RATIO the table has accumulated
enough deleted-but-not-freed rows to warrant a rebuild.
"""
IndexMetaTable.increment_total_inserts(self._conn, count)
current = int(self._meta_get("total_inserts") or "0")
self._meta_set("total_inserts", str(current + count))
def add(self, nodes: Sequence[BaseNode], **add_kwargs: Any) -> list[str]:
if not nodes:
return []
rows = [self._row(node) for node in nodes]
with self._transaction():
self._ensure_table(
len(nodes[0].get_embedding()),
table_exists=self.table_exists(),
)
self._conn.executemany(_INSERT, _vec0_params(rows))
self._index_chunks(rows)
self._ensure_table(len(nodes[0].get_embedding()))
self._conn.executemany(self._INSERT, rows)
self._increment_total_inserts(len(rows))
return [node.node_id for node in nodes]
def upsert_document(
self,
document_id: int | str,
nodes: list[BaseNode],
) -> list[str]:
def upsert_document(self, document_id: str, nodes: list[BaseNode]) -> list[str]:
"""Atomically replace all stored chunks of ``document_id`` with ``nodes``.
One transaction deletes the document's existing rows and inserts the
new set (vec0's INSERT OR REPLACE is broken upstream, so delete+insert
it is). WAL readers in other processes see either the old or the new
chunk set, never a partial state.
new set (vec0's INSERT OR REPLACE is broken upstream, #259, so
delete+insert it is). WAL readers in other processes see either the
old or the new chunk set, never a partial state.
"""
doc_id = int(document_id)
rows = [self._row(node) for node in nodes]
with self._transaction():
table_exists = self.table_exists()
if nodes and not table_exists:
self._ensure_table(
len(nodes[0].get_embedding()),
table_exists=False,
if nodes:
self._ensure_table(len(nodes[0].get_embedding()))
if self.table_exists():
self._conn.execute(
"DELETE FROM " + DEFAULT_TABLE_NAME + " WHERE document_id = ?",
(str(document_id),),
)
table_exists = True
if table_exists:
self._delete_chunks_by_document_id(doc_id)
if rows:
self._conn.executemany(_INSERT, _vec0_params(rows))
self._index_chunks(rows)
self._conn.executemany(self._INSERT, rows)
self._increment_total_inserts(len(rows))
return [node.node_id for node in nodes]
def delete(self, ref_doc_id: int | str, **delete_kwargs: Any) -> None:
def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
if self.table_exists():
with self._transaction():
self._delete_chunks_by_document_id(int(ref_doc_id))
self._conn.execute(
"DELETE FROM " + DEFAULT_TABLE_NAME + " WHERE document_id = ?",
(str(ref_doc_id),),
)
def _rows_to_nodes(self, rows: list[sqlite3.Row]) -> list[BaseNode]:
nodes: list[BaseNode] = []
@@ -503,66 +450,41 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
def get_modified_times(self) -> dict[str, str]:
"""Return {document_id: stored_modified_isoformat} for all indexed documents.
document_meta already has exactly one row per document (not per
chunk, unlike the vec0 table), so no dedup is needed here.
All chunks of a document share the same ``modified`` value, so the
first row seen per document is sufficient.
"""
if not self.table_exists():
return {}
return DocumentMetaTable.all_modified_times(self._conn)
@property
def _db_path(self) -> str:
return str(Path(self._uri) / DB_FILENAME)
@contextmanager
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)
result: dict[str, str] = {}
for row in self._conn.execute(
"SELECT document_id, modified FROM " + DEFAULT_TABLE_NAME,
):
doc_id = str(row["document_id"])
if doc_id not in result:
result[doc_id] = str(row["modified"] or "")
return result
def compact(self, *, force: bool = False) -> None:
"""Rebuild the database file to reclaim space left behind by DELETEs.
vec0 DELETE only invalidates rows; the vector data stays in the file
forever, and per-document re-indexing is a delete+insert. The
cumulative insert counter in ``index_meta`` tracks total rows ever
written; when that exceeds ``COMPACT_BLOAT_RATIO`` x the live row
count (or when forced), live rows are copied into a fresh database
file and swapped in via ``os.replace``.
forever (asg017/sqlite-vec#54), and per-document re-indexing is a
delete+insert. The cumulative insert counter in ``index_meta`` tracks
total rows ever written; when that exceeds ``COMPACT_BLOAT_RATIO`` x
the live row count (or when forced), live rows are copied into a fresh
database file and swapped in via ``os.replace``.
Note: ``ALTER TABLE ... RENAME TO`` on vec0 virtual tables does NOT
rename the shadow tables (sqlite-vec upstream limitation), so an
in-place rename-based rebuild is not safe. The file-swap approach is
the maintainer-endorsed workaround.
rename the shadow tables (sqlite-vec upstream limitation), so
an in-place rename-based rebuild is not safe. The file-swap approach
is the maintainer-endorsed workaround (asg017/sqlite-vec#205).
"""
if not self.table_exists():
return
if self.has_pending_migration():
logger.warning(
"Skipping compact: store has a pending schema migration; "
"run check_and_run_migrations() first",
)
return
live = DocumentChunksTable.count(self._conn)
total = IndexMetaTable.get_total_inserts(self._conn) or live
live = self._conn.execute(
"SELECT count(*) FROM " + DEFAULT_TABLE_NAME,
).fetchone()[0]
total = int(self._meta_get("total_inserts") or str(live))
if not force and total <= max(live, 1) * COMPACT_BLOAT_RATIO:
return
dim = self.vector_dim()
@@ -574,61 +496,50 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
live,
total,
)
with self._rebuild_file() as new_conn:
self._rebuild_into(self._conn, new_conn, dim)
db_path = str(Path(self._uri) / DB_FILENAME)
compact_path = db_path + ".compact"
@staticmethod
def _rebuild_into(
src_conn: sqlite3.Connection,
dst_conn: sqlite3.Connection,
dim: int,
) -> None:
"""Create the vec0 table in ``dst_conn``, copy dim/embed_model from
``src_conn``, and stream every live vec0 row, document_chunks row,
and document_meta row across. Used by compact() only --
m0001_v1_to_v2 freezes its own copy loop instead of calling this,
since this always reflects the *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,
)
copied = 0
while batch := src_cursor.fetchmany(BATCH_SIZE):
dst_conn.executemany(
_INSERT,
[
(
r["id"],
r["document_id"],
r["node_content"],
bytes(r["embedding"]),
)
for r in batch
],
# Copy all live rows into a fresh database file.
new_conn = self._open_connection(compact_path)
try:
self._create_vec_table(new_conn, dim)
self._meta_set_on(new_conn, "dim", str(dim))
for key in ("embed_model", "schema_version"):
value = self._meta_get(key)
if value is not None:
self._meta_set_on(new_conn, key, value)
src_cursor = self._conn.execute(
"SELECT id, document_id, modified, node_content, embedding "
"FROM " + DEFAULT_TABLE_NAME,
)
DocumentChunksTable.insert_many(
dst_conn,
(ChunkRow(r["id"], r["document_id"]) for r in batch),
)
copied += len(batch)
DocumentMetaTable.copy_all(src_conn, dst_conn, BATCH_SIZE)
# Reset the cumulative counter: after a rebuild, total_inserts == live.
IndexMetaTable.reset_total_inserts(dst_conn, copied)
dst_conn.execute("COMMIT")
new_conn.execute("BEGIN IMMEDIATE")
# 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):
new_conn.executemany(
self._INSERT,
[
(
r["id"],
r["document_id"],
r["modified"],
r["node_content"],
bytes(r["embedding"]),
)
for r in batch
],
)
# Reset the cumulative counter: after compact, total_inserts == live.
self._meta_set_on(new_conn, "total_inserts", str(live))
new_conn.execute("COMMIT")
except BaseException:
new_conn.close()
for p in [compact_path, compact_path + "-wal", compact_path + "-shm"]:
Path(p).unlink(missing_ok=True)
raise
new_conn.close()
self._swap_in_compact(compact_path, db_path)
def _swap_in_compact(self, compact_path: str, db_path: str) -> None:
"""Atomically replace the live database with the compacted copy."""
@@ -640,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_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:
"""Apply any pending schema migrations to the store.
@@ -673,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(
@@ -691,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,
@@ -713,12 +601,16 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
dim = self.vector_dim()
if dim is None: # pragma: no cover
raise RuntimeError("Cannot migrate: no stored vector dimension")
with self._rebuild_file() as new_conn:
db_path = str(Path(self._uri) / DB_FILENAME)
compact_path = db_path + ".compact"
new_conn = self._open_connection(compact_path)
try:
migration.apply(self._conn, new_conn, dim)
IndexMetaTable.set_schema_version(new_conn, migration.to_version)
# Registers m0001_v1_to_v2 into MIGRATIONS; must be at the bottom (needs
# PaperlessSqliteVecVectorStore fully defined) -- see
# paperless_ai/migrations/__init__.py for the full procedure.
from paperless_ai.migrations import m0001_v1_to_v2 # noqa: E402, F401
self._meta_set_on(new_conn, "schema_version", str(migration.to_version))
except BaseException: # pragma: no cover
new_conn.close()
for p in [compact_path, compact_path + "-wal", compact_path + "-shm"]:
Path(p).unlink(missing_ok=True)
raise
new_conn.close()
self._swap_in_compact(compact_path, db_path)