Fixes the new test failure and restricts doing the annotation even further, so content must have been requested to annotate even

- _needs_effective_content_annotation() now checks for a non-blank,
  stripped param value rather than mere key presence, matching how
  SearchFilter/TitleContentFilter/EffectiveContentFilter themselves
  no-op on a blank value. An empty ?search= or a saved view with a
  cleared text filter no longer re-triggers the annotation.

- The "versions" prefetch on DocumentViewSet no longer carries content
  for every historical version of every document -- that's unused
  bloat for version-heavy documents. Added
  latest_version_content_prefetch() (versioning.py), a separate,
  windowed prefetch scoped to just the newest version's content per
  root, and taught Document.get_effective_content() to check it first.

- DocumentSerializer.to_representation() no longer unconditionally
  calls get_effective_content(). Added has_prefetched_effective_content()
  (versioning.py) as a cheap upfront check: only resolve version-aware
  content when an SQL annotation or a versions prefetch is already on
  the instance. TrashView and GlobalSearchView build their own
  querysets independently of DocumentViewSet and never display
  document content at all (checked both frontend components), so they
  now keep showing the document's own, unresolved content with zero
  extra queries -- the same behavior as before effective_content
  resolution existed, just generalized past the narrow hasattr() check
  it replaced.

src/documents/consumer.py

@@ -52,7 +52,6 @@ from documents.templating.workflows import parse_w_workflow_placeholders
 from documents.utils import compute_checksum
 from documents.utils import copy_basic_file_stats
 from documents.utils import copy_file_with_basic_stats
-from documents.utils import normalize_unicode
 from documents.utils import run_subprocess
 from paperless.config import OcrConfig
 from paperless.config import RemoteOCRConfig
@@ -202,9 +201,7 @@ class ConsumerPluginMixin:
 
         self.renew_logging_group()
 
-        self.filename = normalize_unicode(
+        self.filename = self.metadata.filename or self.input_doc.original_file.name
-            self.metadata.filename or self.input_doc.original_file.name,
-        )
 
     def _send_progress(
         self,

src/documents/matching.py

@@ -21,7 +21,6 @@ from documents.models import Workflow
 from documents.models import WorkflowTrigger
 from documents.permissions import permitted_object_ids
 from documents.regex import safe_regex_search
-from documents.utils import normalize_unicode
 
 if TYPE_CHECKING:
     from django.db.models import QuerySet
@@ -312,12 +311,11 @@ def consumable_document_matches_workflow(
         trigger_matched = False
 
     # Document filename vs trigger filename
-    document_filename = normalize_unicode(document.original_file.name)
     if (
         trigger.filter_filename is not None
         and len(trigger.filter_filename) > 0
         and not fnmatch(
-            document_filename.lower(),
+            document.original_file.name.lower(),
             trigger.filter_filename.lower(),
         )
     ):
@@ -330,12 +328,10 @@ def consumable_document_matches_workflow(
     # Document path vs trigger path
 
     # Use the original_path if set, else us the original_file
-    match_against = normalize_unicode(
+    match_against = (
-        str(
+        document.original_path
-            document.original_path
+        if document.original_path is not None
-            if document.original_path is not None
+        else document.original_file
-            else document.original_file,
-        ),
     )
 
     if (
@@ -540,7 +536,7 @@ def existing_document_matches_workflow(
         and len(trigger.filter_filename) > 0
         and document.original_filename is not None
         and not fnmatch(
-            normalize_unicode(document.original_filename).lower(),
+            document.original_filename.lower(),
             trigger.filter_filename.lower(),
         )
     ):

src/documents/models.py

@@ -27,7 +27,6 @@ from django_softdelete.models import SoftDeleteModel
 
 from documents.data_models import DocumentSource
 from documents.parsers import get_default_file_extension
-from documents.utils import normalize_unicode
 
 
 class ModelWithOwner(models.Model):
@@ -375,6 +374,7 @@ class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-
         If the queryset already annotated ``effective_content``, that value is used.
         """
         # Here to avoid circular import
+        from documents.versioning import LATEST_VERSION_CONTENT_PREFETCH_ATTR
         from documents.versioning import sort_versions_newest_first
         from documents.versioning import versions_newest_first
 
@@ -384,6 +384,19 @@ class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-
         if self.root_document_id is not None or self.pk is None:
             return self.content
 
+        latest_version_prefetch = getattr(
+            self,
+            LATEST_VERSION_CONTENT_PREFETCH_ATTR,
+            None,
+        )
+        if latest_version_prefetch is not None:
+            # Empty list means prefetch ran and found no versions — use own content.
+            return (
+                latest_version_prefetch[0].content
+                if latest_version_prefetch
+                else self.content
+            )
+
         prefetched_cache = getattr(self, "_prefetched_objects_cache", None)
         prefetched_versions = (
             prefetched_cache.get("versions")
@@ -468,7 +481,7 @@ class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-
         context_document = (
             self.root_document if self.root_document_id is not None else self
         )
-        result = normalize_unicode(str(context_document))
+        result = str(context_document)
 
         if counter:
             result += f"_{counter:02}"

src/documents/serialisers.py

@@ -87,9 +87,9 @@ from documents.regex import validate_regex_pattern
 from documents.templating.filepath import validate_filepath_template_and_render
 from documents.templating.utils import convert_format_str_to_template_format
 from documents.templating.workflows import validate_workflow_template
-from documents.utils import normalize_unicode
 from documents.validators import uri_validator
 from documents.validators import url_validator
+from documents.versioning import has_prefetched_effective_content
 from documents.versioning import sort_versions_newest_first
 
 if TYPE_CHECKING:
@@ -1153,8 +1153,14 @@ class DocumentSerializer(
 
     def to_representation(self, instance):
         doc = super().to_representation(instance)
-        if "content" in self.fields and hasattr(instance, "effective_content"):
+        if "content" in self.fields and has_prefetched_effective_content(instance):
-            doc["content"] = getattr(instance, "effective_content") or ""
+            # Only resolve version-aware content when it's cheap: an SQL
+            # annotation or a versions prefetch is already on the instance.
+            # A caller that set up neither (e.g. TrashView, GlobalSearchView,
+            # which build their own querysets) gets the document's own,
+            # unresolved content instead of paying for an extra per-instance
+            # query -- same as before effective_content resolution existed.
+            doc["content"] = instance.get_effective_content() or ""
         if self.truncate_content and "content" in self.fields:
             doc["content"] = doc.get("content")[0:550]
         return doc
@@ -3121,13 +3127,6 @@ class WorkflowTriggerSerializer(serializers.ModelSerializer[WorkflowTrigger]):
         ):
             attrs["filter_path"] = None
 
-        # Normalize once at write time, since these are matched against many
-        # documents but edited rarely
-        if attrs.get("filter_filename") is not None:
-            attrs["filter_filename"] = normalize_unicode(attrs["filter_filename"])
-        if attrs.get("filter_path") is not None:
-            attrs["filter_path"] = normalize_unicode(attrs["filter_path"])
-
         if (
             "filter_custom_field_query" in attrs
             and attrs["filter_custom_field_query"] is not None

src/documents/templating/filepath.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import re
+import unicodedata
 from collections.abc import Iterable
 from pathlib import PurePath
 
@@ -25,7 +26,6 @@ from documents.templating.environment import _template_environment
 from documents.templating.filters import format_datetime
 from documents.templating.filters import get_cf_value
 from documents.templating.filters import localize_date
-from documents.utils import normalize_unicode
 
 logger = logging.getLogger("paperless.templating")
 
@@ -42,7 +42,7 @@ class FilePathTemplate(Template):
             3. Removing extra spaces before and after forward slashes
             4. Preserving spaces in other parts of the path
             """
-            value = normalize_unicode(value)
+            value = unicodedata.normalize("NFC", value)
             value = value.replace("\n", "").replace("\r", "")
             value = re.sub(r"\s*/\s*", "/", value)
 
@@ -184,17 +184,17 @@ def get_basic_metadata_context(
     """
     return {
         "title": pathvalidate.sanitize_filename(
-            normalize_unicode(document.title),
+            unicodedata.normalize("NFC", document.title),
             replacement_text="-",
         ),
         "correspondent": pathvalidate.sanitize_filename(
-            normalize_unicode(document.correspondent.name),
+            unicodedata.normalize("NFC", document.correspondent.name),
             replacement_text="-",
         )
         if document.correspondent
         else no_value_default,
         "document_type": pathvalidate.sanitize_filename(
-            normalize_unicode(document.document_type.name),
+            unicodedata.normalize("NFC", document.document_type.name),
             replacement_text="-",
         )
         if document.document_type
@@ -205,7 +205,8 @@ def get_basic_metadata_context(
         "owner_username": document.owner.username
         if document.owner
         else no_value_default,
-        "original_name": normalize_unicode(
+        "original_name": unicodedata.normalize(
+            "NFC",
             PurePath(document.original_filename).with_suffix("").name,
         )
         if document.original_filename
@@ -274,12 +275,12 @@ def get_tags_context(tags: Iterable[Tag]) -> dict[str, str | list[str]]:
     return {
         "tag_list": pathvalidate.sanitize_filename(
             ",".join(
-                sorted(normalize_unicode(tag.name) for tag in tags),
+                sorted(unicodedata.normalize("NFC", tag.name) for tag in tags),
             ),
             replacement_text="-",
         ),
         # Assumed to be ordered, but a template could loop through to find what they want
-        "tag_name_list": [normalize_unicode(x.name) for x in tags],
+        "tag_name_list": [unicodedata.normalize("NFC", x.name) for x in tags],
     }
 
 
@@ -306,7 +307,7 @@ def get_custom_fields_context(
             CustomField.FieldDataType.LONG_TEXT,
         }:
             value = pathvalidate.sanitize_filename(
-                normalize_unicode(field_instance.value),
+                unicodedata.normalize("NFC", field_instance.value),
                 replacement_text="-",
             )
         elif (
@@ -315,7 +316,8 @@ def get_custom_fields_context(
         ):
             options = field_instance.field.extra_data["select_options"]
             value = pathvalidate.sanitize_filename(
-                normalize_unicode(
+                unicodedata.normalize(
+                    "NFC",
                     next(
                         option["label"]
                         for option in options
@@ -328,7 +330,7 @@ def get_custom_fields_context(
             value = field_instance.value
         field_data["custom_fields"][
             pathvalidate.sanitize_filename(
-                normalize_unicode(field_instance.field.name),
+                unicodedata.normalize("NFC", field_instance.field.name),
                 replacement_text="-",
             )
         ] = {

src/documents/tests/test_api_update_version_nfc.py

@@ -1,69 +0,0 @@
-import unicodedata
-from typing import TYPE_CHECKING
-from unittest import mock
-
-import celery.result
-import pytest
-from django.core.files.uploadedfile import SimpleUploadedFile
-
-from documents.models import Document
-
-if TYPE_CHECKING:
-    from documents.data_models import ConsumableDocument
-
-
-@pytest.fixture()
-def consume_file_mock():
-    with mock.patch("documents.tasks.consume_file.apply_async") as m:
-        m.return_value = celery.result.AsyncResult(id="test-task-id")
-        yield m
-
-
-@pytest.fixture()
-def directories(tmp_path, settings, _media_settings):
-    scratch = tmp_path / "scratch"
-    scratch.mkdir()
-    settings.SCRATCH_DIR = scratch
-    return scratch
-
-
-@pytest.mark.django_db
-class TestUpdateVersionNFCNormalization:
-    def test_nfd_filename_normalized_to_nfc(
-        self,
-        admin_client,
-        consume_file_mock: mock.MagicMock,
-        directories,
-    ):
-        """Uploaded new-version file with NFD filename must have its temp name stored as NFC."""
-        document = Document.objects.create(
-            title="Test",
-            content="content",
-            checksum="checksum",
-            mime_type="application/pdf",
-        )
-
-        nfd = unicodedata.normalize("NFD", "Rechnung März.pdf")
-        nfc = unicodedata.normalize("NFC", "Rechnung März.pdf")
-
-        assert nfd != nfc
-
-        uploaded = SimpleUploadedFile(
-            nfd,
-            b"%PDF-1.4 test",
-            content_type="application/pdf",
-        )
-        response = admin_client.post(
-            f"/api/documents/{document.pk}/update_version/",
-            {"document": uploaded},
-        )
-
-        assert response.status_code == 200
-
-        task_kwargs = consume_file_mock.call_args.kwargs["kwargs"]
-        input_doc: ConsumableDocument = task_kwargs["input_doc"]
-
-        assert input_doc.original_file.name == nfc, (
-            f"Expected NFC filename {nfc!r}, got {input_doc.original_file.name!r}"
-        )
-        assert unicodedata.is_normalized("NFC", input_doc.original_file.name)

src/documents/tests/test_document_list_effective_content_annotation.py

@@ -0,0 +1,240 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import TYPE_CHECKING
+
+import pytest
+from django.db import connection
+from django.test.utils import CaptureQueriesContext
+from rest_framework import status
+
+from documents.models import Document
+from documents.tests.factories import DocumentFactory
+from documents.versioning import LATEST_VERSION_CONTENT_PREFETCH_ATTR
+from documents.versioning import has_prefetched_effective_content
+from documents.versioning import latest_version_content_prefetch
+from documents.views import DocumentViewSet
+
+if TYPE_CHECKING:
+    from rest_framework.test import APIClient
+
+
+class TestNeedsEffectiveContentAnnotation:
+    """
+    DocumentViewSet._needs_effective_content_annotation() decides whether
+    the effective_content correlated subquery is worth attaching to the
+    queryset at all -- see TestDocumentListEffectiveContentAnnotation below
+    for why. This only checks that decision's own logic (a plain query-param
+    membership test), not that Django/DRF's filtering machinery works.
+    """
+
+    @pytest.mark.parametrize(
+        ("params", "expected"),
+        [
+            ({}, False),
+            ({"ordering": "-added"}, False),
+            ({"tags__id__in": "1,2"}, False),
+            ({"search": ""}, False),
+            ({"search": "   "}, False),
+            ({"content__icontains": ""}, False),
+            ({"search": "foo"}, True),
+            ({"title_content": "foo"}, True),
+            ({"content__istartswith": "foo"}, True),
+            ({"content__iendswith": "foo"}, True),
+            ({"content__icontains": "foo"}, True),
+            ({"content__iexact": "foo"}, True),
+        ],
+    )
+    def test_detects_content_filter_params(
+        self,
+        params: dict[str, str],
+        expected: bool,  # noqa: FBT001
+    ) -> None:
+        # GIVEN a view bound to a request carrying the given query params
+        view = DocumentViewSet()
+        view.request = SimpleNamespace(query_params=params)
+
+        # WHEN checking whether the effective_content annotation is needed
+        # THEN it's needed only for requests that actually filter on it
+        assert view._needs_effective_content_annotation() is expected
+
+
+@pytest.mark.django_db
+class TestDocumentListEffectiveContentAnnotation:
+    """
+    DocumentViewSet.get_queryset() only attaches the effective_content
+    correlated subquery when a request actually filters on it. Attaching it
+    unconditionally re-executes it once per candidate row before the page's
+    LIMIT is applied -- fine on SQLite/Postgres, but pathological on
+    MariaDB's default cardinality estimation for the root_document_id
+    self-join once candidate counts get large (see the root_document_id /
+    effective_content perf investigation).
+    """
+
+    def test_list_without_content_filter_skips_annotation_but_returns_latest_content(
+        self,
+        admin_client: APIClient,
+    ) -> None:
+        # GIVEN a root document whose latest version has different content
+        root = DocumentFactory(content="old-root-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="new-version-content",
+        )
+
+        # WHEN listing documents with no search/content-filter param
+        with CaptureQueriesContext(connection) as ctx:
+            response = admin_client.get("/api/documents/?fields=id,content")
+
+        # THEN the response still reflects the latest version's content...
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["results"] == [
+            {"id": root.id, "content": "new-version-content"},
+        ]
+        # ...without the database ever evaluating effective_content per row
+        assert not any(
+            "effective_content" in query["sql"] for query in ctx.captured_queries
+        )
+
+    def test_latest_version_content_prefetch_carries_only_the_newest_version(
+        self,
+    ) -> None:
+        # GIVEN a root document with two versions
+        root = DocumentFactory(content="root-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="older-version-content",
+        )
+        DocumentFactory(
+            root_document=root,
+            version_index=2,
+            content="newest-version-content",
+        )
+
+        # WHEN fetching the root through latest_version_content_prefetch()
+        fetched_root = (
+            Document.objects.filter(pk=root.pk)
+            .prefetch_related(
+                latest_version_content_prefetch(),
+            )
+            .get()
+        )
+
+        # THEN the prefetch carries only the single newest version, not
+        # every historical version's content (the whole point of not
+        # reusing the metadata-only "versions" prefetch for this)
+        latest = getattr(fetched_root, LATEST_VERSION_CONTENT_PREFETCH_ATTR)
+        assert [v.content for v in latest] == ["newest-version-content"]
+
+
+class TestHasPrefetchedEffectiveContent:
+    """
+    DocumentSerializer.to_representation() only calls get_effective_content()
+    when has_prefetched_effective_content() says it's cheap -- otherwise a
+    caller that never set up an annotation or prefetch (TrashView,
+    GlobalSearchView, which build their own querysets and don't display
+    content at all) would pay for a per-instance query nobody asked for.
+    """
+
+    def test_false_with_no_annotation_or_prefetch(self) -> None:
+        document = Document()
+        assert has_prefetched_effective_content(document) is False
+
+    def test_true_with_effective_content_annotation(self) -> None:
+        document = Document()
+        document.effective_content = "resolved"
+        assert has_prefetched_effective_content(document) is True
+
+    def test_true_with_lean_prefetch_attr_even_when_empty(self) -> None:
+        document = Document()
+        setattr(document, LATEST_VERSION_CONTENT_PREFETCH_ATTR, [])
+        assert has_prefetched_effective_content(document) is True
+
+    def test_true_with_metadata_versions_prefetch_cache(self) -> None:
+        document = Document()
+        document._prefetched_objects_cache = {"versions": []}
+        assert has_prefetched_effective_content(document) is True
+
+
+def _get_effective_content_fallback_queries(
+    ctx: CaptureQueriesContext,
+) -> list[dict[str, str]]:
+    """
+    Document.get_effective_content()'s per-instance fallback (no annotation,
+    no prefetch) is a `.values_list("content", flat=True).first()` query --
+    a SELECT of just the content column. Distinct from get_versions()'s own,
+    unrelated per-instance metadata query (id/checksum/added/etc, no
+    content) run to build the "versions" response field, which isn't part
+    of what this test file covers.
+    """
+    return [
+        q
+        for q in ctx.captured_queries
+        if q["sql"].startswith('SELECT "documents_document"."content" FROM')
+    ]
+
+
+@pytest.mark.django_db
+class TestTrashAndGlobalSearchEffectiveContentIsNeverPerInstance:
+    """
+    TrashView and GlobalSearchView serialize Document instances with
+    DocumentSerializer too, but build their querysets independently of
+    DocumentViewSet.get_queryset(). TrashView doesn't display content at all,
+    so it keeps the document's own unresolved content; GlobalSearchView
+    annotates effective_content itself, so it shows the latest version's.
+    Neither should ever fall back to a per-instance query.
+    """
+
+    def test_trash_list_shows_unresolved_content_with_no_extra_query(
+        self,
+        admin_client: APIClient,
+    ) -> None:
+        # GIVEN a trashed root document whose own content differs from what
+        # a (also trashed, since deletion cascades) version would have had
+        root = DocumentFactory(content="own-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="version-content",
+        )
+        root.delete()
+
+        # WHEN listing trash
+        with CaptureQueriesContext(connection) as ctx:
+            response = admin_client.get("/api/trash/")
+
+        # THEN the response shows the document's own content...
+        assert response.status_code == status.HTTP_200_OK
+        [result] = [r for r in response.data["results"] if r["id"] == root.id]
+        assert result["content"] == "own-content"
+        # ...without ever querying for versions to resolve it
+        assert _get_effective_content_fallback_queries(ctx) == []
+
+    def test_global_search_db_only_shows_latest_version_content_with_no_extra_query(
+        self,
+        admin_client: APIClient,
+    ) -> None:
+        # GIVEN a root document, findable by title, whose own content
+        # differs from its latest version's
+        root = DocumentFactory(title="findme", content="own-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="version-content",
+        )
+
+        # WHEN using the global search endpoint's db_only mode
+        with CaptureQueriesContext(connection) as ctx:
+            response = admin_client.get(
+                "/api/search/?query=findme&db_only=true",
+            )
+
+        # THEN the response shows the latest version's content, resolved by
+        # GlobalSearchView's own effective_content annotation...
+        assert response.status_code == status.HTTP_200_OK
+        [result] = [d for d in response.data["documents"] if d["id"] == root.id]
+        assert result["content"] == "version-content"
+        # ...with no per-instance fallback query
+        assert _get_effective_content_fallback_queries(ctx) == []

src/documents/tests/test_get_public_filename_nfc.py

@@ -1,48 +0,0 @@
-import unicodedata
-from datetime import date
-
-import pytest
-
-from documents.models import Correspondent
-from documents.models import Document
-
-
-@pytest.mark.django_db
-class TestGetPublicFilenameNfc:
-    def test_normalizes_nfd_title_to_nfc(self) -> None:
-        nfd_title = unicodedata.normalize("NFD", "Gehaltserhöhung")
-        assert not unicodedata.is_normalized("NFC", nfd_title)
-
-        doc = Document(
-            mime_type="application/pdf",
-            title=nfd_title,
-            created=date(2025, 10, 17),
-        )
-
-        result = doc.get_public_filename()
-
-        assert unicodedata.is_normalized("NFC", result)
-        assert (
-            result
-            == "2025-10-17 "
-            + unicodedata.normalize(
-                "NFC",
-                nfd_title,
-            )
-            + ".pdf"
-        )
-
-    def test_normalizes_nfd_correspondent_name_to_nfc(self) -> None:
-        nfd_name = unicodedata.normalize("NFD", "Müller GmbH")
-        correspondent = Correspondent.objects.create(name=nfd_name)
-
-        doc = Document.objects.create(
-            mime_type="application/pdf",
-            title="Rechnung",
-            created=date(2025, 10, 17),
-            correspondent=correspondent,
-        )
-
-        result = doc.get_public_filename()
-
-        assert unicodedata.is_normalized("NFC", result)

src/documents/tests/test_matching_nfc.py

@@ -1,80 +0,0 @@
-import unicodedata
-
-import pytest
-
-from documents.data_models import ConsumableDocument
-from documents.data_models import DocumentSource
-from documents.matching import consumable_document_matches_workflow
-from documents.matching import existing_document_matches_workflow
-from documents.models import Document
-from documents.models import Workflow
-from documents.models import WorkflowTrigger
-
-
-@pytest.mark.django_db
-class TestMatchingNfcNormalization:
-    def test_consumable_document_filename_nfd_matches_nfc_pattern(
-        self,
-        tmp_path,
-    ) -> None:
-        """
-        GIVEN:
-            - A file on disk whose name is NFD-normalized
-            - A workflow trigger filename filter typed as NFC
-        WHEN:
-            - The consumable document is checked against the trigger
-        THEN:
-            - It matches, because both sides are normalized before comparing
-        """
-        nfd_name = unicodedata.normalize("NFD", "Gehaltserhöhung.pdf")
-        nfc_pattern = unicodedata.normalize("NFC", "*Gehaltserhöhung*")
-        assert nfd_name != unicodedata.normalize("NFC", nfd_name)
-
-        file_path = tmp_path / nfd_name
-        file_path.write_bytes(b"%PDF-1.4 test")
-
-        document = ConsumableDocument(
-            source=DocumentSource.ConsumeFolder,
-            original_file=file_path,
-        )
-        trigger = WorkflowTrigger(
-            type=WorkflowTrigger.WorkflowTriggerType.CONSUMPTION,
-            filter_filename=nfc_pattern,
-            sources=[],
-        )
-
-        matched, reason = consumable_document_matches_workflow(document, trigger)
-
-        assert matched, reason
-
-    def test_existing_document_filename_nfd_matches_nfc_pattern(self) -> None:
-        """
-        GIVEN:
-            - A Document whose original_filename is NFD-normalized (e.g. from
-              before normalization was applied at consumption time)
-            - A workflow trigger filename filter typed as NFC
-        WHEN:
-            - The document is checked against the trigger
-        THEN:
-            - It matches, because both sides are normalized before comparing
-        """
-        nfd_name = unicodedata.normalize("NFD", "Gehaltserhöhung.pdf")
-        nfc_pattern = unicodedata.normalize("NFC", "*Gehaltserhöhung*")
-
-        document = Document.objects.create(
-            title="Test",
-            content="content",
-            checksum="checksum",
-            mime_type="application/pdf",
-            original_filename=nfd_name,
-        )
-        workflow = Workflow.objects.create(name="Test workflow", order=0)
-        trigger = WorkflowTrigger.objects.create(
-            type=WorkflowTrigger.WorkflowTriggerType.DOCUMENT_ADDED,
-            filter_filename=nfc_pattern,
-        )
-        workflow.triggers.add(trigger)
-
-        matched, reason = existing_document_matches_workflow(document, trigger)
-
-        assert matched, reason

src/documents/tests/test_normalize_unicode.py

@@ -1,6 +0,0 @@
-from documents.utils import normalize_unicode
-
-
-class TestNormalizeUnicode:
-    def test_none_passes_through(self) -> None:
-        assert normalize_unicode(None) is None

src/documents/utils.py

@@ -1,7 +1,6 @@
 import hashlib
 import logging
 import shutil
-import unicodedata
 from collections.abc import Callable
 from collections.abc import Iterable
 from collections.abc import Iterator
@@ -32,25 +31,6 @@ def identity(iterable: Iterable[_T]) -> Iterable[_T]:
     return iterable
 
 
-def normalize_unicode(value: str | None) -> str | None:
-    """
-    Normalize a string to Unicode NFC form, or return None unchanged.
-
-    This is the single normalization pass for any user- or filesystem-supplied
-    text that ends up in a filename, path, or is compared/matched against one
-    (titles, correspondent/tag/type names, uploaded filenames, workflow and
-    mail rule filename/path filters). Composed (NFC) and decomposed (NFD)
-    forms of the same visible text are different byte sequences, which breaks
-    exact comparisons and filesystem lookups even though the text looks
-    identical. Always normalize through this function rather than calling
-    unicodedata.normalize() directly, so every call site agrees on the same
-    form.
-    """
-    if value is None:
-        return None
-    return unicodedata.normalize("NFC", value)
-
-
 class QuerySetStream(Generic[_M]):
     """Stream a QuerySet via .iterator(chunk_size=...) instead of
     materializing it (plus any prefetch caches) all at once, while still

src/documents/versioning.py

@@ -7,9 +7,12 @@ from typing import Any
 
 from django.db.models import F
 from django.db.models import OuterRef
+from django.db.models import Prefetch
 from django.db.models import QuerySet
 from django.db.models import Subquery
+from django.db.models import Window
 from django.db.models.functions import Coalesce
+from django.db.models.functions import RowNumber
 
 from documents.models import Document
 
@@ -46,6 +49,68 @@ def annotate_effective_content(documents: QuerySet[Document]) -> QuerySet[Docume
     )
 
 
+LATEST_VERSION_CONTENT_PREFETCH_ATTR = "_latest_version_content_prefetch"
+
+
+def latest_version_content_prefetch() -> Prefetch:
+    """
+    A Prefetch for Document.versions scoped to just the newest version's
+    content, for get_effective_content()'s fallback when no SQL annotation
+    is present.
+
+    Deliberately not merged into a metadata-only "versions" prefetch (the one
+    used for the serialized versions list): that one fetches every historical
+    version of every document, and pulling full OCR content for versions
+    nobody will read wastes DB transfer/memory at scale. This one is windowed
+    down to a single row per root, then bounded by Prefetch's own IN-list to
+    whatever page/result set it's attached to -- one cheap bulk query total,
+    not one per document and not one per version.
+    """
+    return Prefetch(
+        "versions",
+        queryset=(
+            Document.objects.filter(
+                root_document_id__isnull=False,
+                deleted_at__isnull=True,
+            )
+            .annotate(
+                rn=Window(
+                    RowNumber(),
+                    partition_by=F("root_document_id"),
+                    order_by=[
+                        F("version_index").desc(nulls_last=True),
+                        F("id").desc(),
+                    ],
+                ),
+            )
+            .filter(rn=1)
+            .only("id", "root_document_id", "content")
+        ),
+        to_attr=LATEST_VERSION_CONTENT_PREFETCH_ATTR,
+    )
+
+
+def has_prefetched_effective_content(document: Document) -> bool:
+    """
+    True if document.get_effective_content() can answer without an extra
+    per-instance query -- an SQL ``effective_content`` annotation, the lean
+    latest_version_content_prefetch(), or the metadata-only "versions"
+    prefetch is already present on the instance.
+
+    Callers that haven't set any of those up (e.g. views that build their
+    own querysets independently of DocumentViewSet.get_queryset(), like
+    TrashView or GlobalSearchView) intentionally don't pay for version-aware
+    content resolution -- see DocumentSerializer.to_representation(), which
+    uses this to decide whether to call get_effective_content() at all.
+    """
+    if hasattr(document, "effective_content"):
+        return True
+    if getattr(document, LATEST_VERSION_CONTENT_PREFETCH_ATTR, None) is not None:
+        return True
+    prefetched_cache = getattr(document, "_prefetched_objects_cache", None)
+    return isinstance(prefetched_cache, dict) and "versions" in prefetched_cache
+
+
 def sort_versions_newest_first(documents: list[Document]) -> list[Document]:
     """
     Same sorting as versions_newest_first()

src/documents/views.py

@@ -36,7 +36,6 @@ from django.db.migrations.recorder import MigrationRecorder
 from django.db.models import Avg
 from django.db.models import Case
 from django.db.models import Count
-from django.db.models import F
 from django.db.models import IntegerField
 from django.db.models import Max
 from django.db.models import Model
@@ -137,12 +136,14 @@ from documents.filters import CustomFieldFilterSet
 from documents.filters import DocumentFilterSet
 from documents.filters import DocumentsOrderingFilter
 from documents.filters import DocumentTypeFilterSet
+from documents.filters import EffectiveContentFilter
 from documents.filters import PaperlessTaskFilterSet
 from documents.filters import PermittedObjectsFilter
 from documents.filters import ShareLinkBundleFilterSet
 from documents.filters import ShareLinkFilterSet
 from documents.filters import StoragePathFilterSet
 from documents.filters import TagFilterSet
+from documents.filters import TitleContentFilter
 from documents.mail import EmailAttachment
 from documents.mail import send_email
 from documents.matching import match_correspondents
@@ -231,12 +232,12 @@ from documents.tasks import sanity_check
 from documents.tasks import train_classifier
 from documents.tasks import update_document_parent_tags
 from documents.utils import get_boolean
-from documents.utils import normalize_unicode
 from documents.versioning import VersionResolutionError
 from documents.versioning import annotate_effective_content
 from documents.versioning import get_latest_version_for_root
 from documents.versioning import get_request_version_param
 from documents.versioning import get_root_document
+from documents.versioning import latest_version_content_prefetch
 from documents.versioning import resolve_requested_version_for_root
 from documents.versioning import versions_newest_first
 from paperless import version
@@ -1084,12 +1085,49 @@ class DocumentViewSet(
             ],
         }
 
-    def get_queryset(self):
+    @classmethod
-        latest_version_content = Subquery(
+    def _content_filter_params(cls) -> tuple[str, ...]:
-            versions_newest_first(
+        """
-                Document.objects.filter(root_document=OuterRef("pk")),
+        Query params whose filtering needs effective_content evaluated in SQL
-            ).values("content")[:1],
+        against every candidate row -- see
+        _needs_effective_content_annotation(). Derived rather than
+        hand-maintained so a new content-filtering param counts automatically.
+        """
+        params = [
+            name
+            for name, f in DocumentFilterSet.declared_filters.items()
+            if isinstance(f, (TitleContentFilter, EffectiveContentFilter))
+        ]
+        if "effective_content" in cls.search_fields:
+            params.append(SearchFilter().search_param)
+        return tuple(params)
+
+    def _needs_effective_content_annotation(self) -> bool:
+        # effective_content is a per-row correlated subquery resolving each
+        # document's latest version. Filtering *on* it forces the database to
+        # evaluate it for every candidate row before reaching the LIMIT, which
+        # the root_document_id self-join makes pathological on MariaDB
+        # specifically once real candidate counts get large; otherwise the
+        # "versions" prefetch + Document.get_effective_content() resolves only
+        # the page that survives pagination. Every param here is deprecated in
+        # favor of the Tantivy-backed search endpoint (see filters.py's
+        # TitleContentFilter/EffectiveContentFilter docs), so pay that cost
+        # only when one is actually used. Blank values don't count, matching
+        # how those filters themselves no-op on them -- an empty `?search=`
+        # applies no predicate.
+        params = self.request.query_params
+        return any(
+            params.get(param, "").strip() for param in self._content_filter_params()
         )
+
+    def _needs_effective_content_prefetch(self) -> bool:
+        # The prefetch spares get_effective_content() a per-instance fallback
+        # query, but only earns itself when content can reach the response.
+        # Mirror get_serializer() below: no `fields` param keeps every field.
+        fields_param = self.request.query_params.get("fields", None)
+        return fields_param is None or "content" in fields_param.split(",")
+
+    def get_queryset(self):
         # A correlated subquery avoids the LEFT JOIN + Count() this used to
         # be, which forced a GROUP BY aggregate over every matching document
         # before the query could even be sorted or limited.
@@ -1109,33 +1147,38 @@ class DocumentViewSet(
         # ObjectFilter.filter(). A blanket .distinct() here forces the
         # database to fully sort and dedupe every visible document before
         # it can apply LIMIT, which is disastrous at scale.
-        return (
+        prefetches = [
+            Prefetch(
+                "versions",
+                queryset=Document.objects.only(
+                    "id",
+                    "added",
+                    "checksum",
+                    "version_label",
+                    "root_document_id",
+                    "version_index",
+                ),
+            ),
+            "tags",
+            Prefetch(
+                "custom_fields",
+                queryset=CustomFieldInstance.objects.select_related("field"),
+            ),
+            # NotesSerializer nests the author, this avoids query per note
+            Prefetch("notes", queryset=Note.objects.select_related("user")),
+        ]
+        if self._needs_effective_content_prefetch():
+            prefetches.append(latest_version_content_prefetch())
+        queryset = (
             Document.objects.filter(root_document__isnull=True)
             .order_by("-created", "-id")
-            .annotate(effective_content=Coalesce(latest_version_content, F("content")))
             .annotate(num_notes=Coalesce(note_count, 0))
             .select_related("correspondent", "storage_path", "document_type", "owner")
-            .prefetch_related(
+            .prefetch_related(*prefetches)
-                Prefetch(
-                    "versions",
-                    queryset=Document.objects.only(
-                        "id",
-                        "added",
-                        "checksum",
-                        "version_label",
-                        "root_document_id",
-                        "version_index",
-                    ),
-                ),
-                "tags",
-                Prefetch(
-                    "custom_fields",
-                    queryset=CustomFieldInstance.objects.select_related("field"),
-                ),
-                # NotesSerializer nests the author, this avoids query per note
-                Prefetch("notes", queryset=Note.objects.select_related("user")),
-            )
         )
+        if self._needs_effective_content_annotation():
+            queryset = annotate_effective_content(queryset)
+        return queryset
 
     def get_serializer(self, *args, **kwargs):
         fields_param = self.request.query_params.get("fields", None)
@@ -2069,7 +2112,6 @@ class DocumentViewSet(
 
         try:
             doc_name, doc_data = serializer.validated_data.get("document")
-            doc_name = normalize_unicode(doc_name)
             version_label = serializer.validated_data.get("version_label")
 
             t = int(mktime(datetime.now().timetuple()))
@@ -3336,7 +3378,7 @@ class PostDocumentView(GenericAPIView[Any]):
         serializer.is_valid(raise_exception=True)
 
         doc_name, doc_data = serializer.validated_data.get("document")
-        doc_name = normalize_unicode(doc_name)
+        doc_name = normalize("NFC", doc_name)
         correspondent_id = serializer.validated_data.get("correspondent")
         document_type_id = serializer.validated_data.get("document_type")
         storage_path_id = serializer.validated_data.get("storage_path")

src/paperless_mail/mail.py

@@ -6,6 +6,7 @@ import socket
 import ssl
 import tempfile
 import traceback
+import unicodedata
 from datetime import date
 from datetime import timedelta
 from fnmatch import fnmatch
@@ -44,7 +45,6 @@ from documents.models import Correspondent
 from documents.models import PaperlessTask
 from documents.parsers import is_mime_type_supported
 from documents.tasks import consume_file
-from documents.utils import normalize_unicode
 from paperless.network import is_public_ip
 from paperless.network import resolve_hostname_ips
 from paperless_mail.models import MailAccount
@@ -617,10 +617,10 @@ class MailAccountHandler(LoggingMixin):
         rule: MailRule,
     ) -> str | None:
         if rule.assign_title_from == MailRule.TitleSource.FROM_SUBJECT:
-            return normalize_unicode(message.subject)
+            return unicodedata.normalize("NFC", message.subject)
 
         elif rule.assign_title_from == MailRule.TitleSource.FROM_FILENAME:
-            return normalize_unicode(Path(att.filename).stem)
+            return unicodedata.normalize("NFC", Path(att.filename).stem)
 
         elif rule.assign_title_from == MailRule.TitleSource.NONE:
             return None
@@ -1004,8 +1004,6 @@ class MailAccountHandler(LoggingMixin):
         consume_tasks = []
 
         for att in message.attachments:
-            attachment_filename = normalize_unicode(att.filename)
-
             if (
                 att.content_disposition != "attachment"
                 and rule.attachment_type
@@ -1020,7 +1018,7 @@ class MailAccountHandler(LoggingMixin):
 
             if not self.filename_inclusion_matches(
                 rule.filter_attachment_filename_include,
-                attachment_filename,
+                att.filename,
             ):
                 # Force the filename and pattern to the lowercase
                 # as this is system dependent otherwise
@@ -1032,7 +1030,7 @@ class MailAccountHandler(LoggingMixin):
                 continue
             elif self.filename_exclusion_matches(
                 rule.filter_attachment_filename_exclude,
-                attachment_filename,
+                att.filename,
             ):
                 self.log.debug(
                     f"Rule {rule}: "
@@ -1066,7 +1064,7 @@ class MailAccountHandler(LoggingMixin):
                 )
 
                 attachment_name = pathvalidate.sanitize_filename(
-                    attachment_filename,
+                    unicodedata.normalize("NFC", att.filename),
                 )
                 if attachment_name:
                     temp_filename = temp_dir / attachment_name
@@ -1177,7 +1175,7 @@ class MailAccountHandler(LoggingMixin):
         doc_overrides = DocumentMetadataOverrides(
             title=message.subject,
             filename=pathvalidate.sanitize_filename(
-                normalize_unicode(f"{message.subject}.eml"),
+                unicodedata.normalize("NFC", f"{message.subject}.eml"),
             ),
             correspondent_id=correspondent.id if correspondent else None,
             document_type_id=doc_type.id if doc_type else None,

src/paperless_mail/serialisers.py

@@ -8,7 +8,6 @@ from documents.serialisers import CorrespondentField
 from documents.serialisers import DocumentTypeField
 from documents.serialisers import OwnedObjectSerializer
 from documents.serialisers import TagsField
-from documents.utils import normalize_unicode
 from paperless_mail.models import MailAccount
 from paperless_mail.models import MailRule
 from paperless_mail.models import ProcessedMail
@@ -162,12 +161,6 @@ class MailRuleSerializer(OwnedObjectSerializer):
             raise serializers.ValidationError("Maximum mail age is unreasonably large.")
         return value
 
-    def validate_filter_attachment_filename_include(self, value):
-        return normalize_unicode(value)
-
-    def validate_filter_attachment_filename_exclude(self, value):
-        return normalize_unicode(value)
-
 
 class ProcessedMailSerializer(OwnedObjectSerializer):
     class Meta: