fix: address review feedback on effective_content annotation skip

- _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/data_models.py

@@ -129,7 +129,7 @@ class DocumentMetadataOverrides:
         )
         overrides.custom_fields = {
             custom_field.field.id: custom_field.value
-            for custom_field in doc.custom_fields.select_related("field").all()
+            for custom_field in doc.custom_fields.all()
         }
 
         groups_with_perms = get_groups_with_perms(

src/documents/models.py

@@ -373,6 +373,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
 
@@ -382,6 +383,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")

src/documents/serialisers.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import logging
 import math
 import re
-from collections.abc import Iterable
 from datetime import datetime
 from datetime import timedelta
 from decimal import Decimal
@@ -89,6 +88,7 @@ from documents.templating.utils import convert_format_str_to_template_format
 from documents.templating.workflows import validate_workflow_template
 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:
@@ -877,106 +877,8 @@ def validate_documentlink_targets(user, doc_ids):
         )
 
 
-# drf-writable-nested revalidates a document's custom_fields more than once
-# per request: once as the ordinary nested list, then again per-item while
-# matching existing vs. new CustomFieldInstance rows during save() -- and
-# that second pass builds a brand new serializer (and field) instance per
-# item (see its update_or_create_reverse_relations / _get_serializer_for_field),
-# so a cache on the field instance alone only helps the first pass. It does,
-# however, explicitly pass `context=self.context` to every one of those
-# fresh serializers -- the *same* dict object the outer DocumentSerializer
-# is using, not a copy. That context dict is already request-scoped (DRF
-# builds it fresh per request via get_serializer_context()), so stashing the
-# resolved CustomField objects there -- rather than in some new global/
-# thread-local cache -- lets every later pass reuse them for free while
-# staying entirely within DRF's existing, already-request-scoped machinery.
-_CUSTOM_FIELD_CONTEXT_CACHE_KEY = "_custom_field_lookup_cache"
-
-
-class _CachingCustomFieldPrimaryKeyField(serializers.PrimaryKeyRelatedField):
-    """
-    Resolves CustomField ids with as few queries as possible: a per-instance
-    cache for repeat lookups on this exact field instance, backed by a
-    shared cache on the serializer context (see _CUSTOM_FIELD_CONTEXT_CACHE_KEY
-    above) so later, separately-instantiated fields for the same request
-    reuse what was already resolved instead of re-querying.
-    """
-
-    def __init__(self, **kwargs: Any) -> None:
-        super().__init__(**kwargs)
-        self._cache: dict[int, CustomField] = {}
-
-    def _shared_cache(self) -> dict[int, CustomField]:
-        return self.context.setdefault(_CUSTOM_FIELD_CONTEXT_CACHE_KEY, {})
-
-    @staticmethod
-    def _normalize_pk(data: Any) -> int | None:
-        """
-        Returns `data` coerced to the int a valid CustomField pk would be,
-        or None if `data` isn't a plausible pk (wrong type, unhashable,
-        non-numeric, or a bool -- DRF itself rejects bools as pks since
-        `True == 1` would otherwise silently match). None tells callers to
-        leave `data` alone and let `super().to_internal_value()` report the
-        normal validation error instead of touching the cache/queryset with
-        it directly.
-        """
-        if isinstance(data, bool):
-            return None
-        try:
-            return int(data)
-        except (TypeError, ValueError):
-            return None
-
-    def prefetch(self, ids: Iterable[Any]) -> None:
-        shared_cache = self._shared_cache()
-        candidates = {pk for i in ids if (pk := self._normalize_pk(i)) is not None}
-        missing = {
-            i for i in candidates if i not in self._cache and i not in shared_cache
-        }
-        if missing:
-            for obj in self.get_queryset().filter(pk__in=missing):
-                shared_cache[obj.pk] = obj
-        for i in candidates:
-            obj = shared_cache.get(i)
-            if obj is not None:
-                self._cache[i] = obj
-
-    def to_internal_value(self, data: Any) -> CustomField:
-        pk = self._normalize_pk(data)
-        if pk is None:
-            return super().to_internal_value(data)
-        if pk in self._cache:
-            return self._cache[pk]
-        shared_cache = self._shared_cache()
-        if pk in shared_cache:
-            obj = shared_cache[pk]
-            self._cache[pk] = obj
-            return obj
-        obj: CustomField = super().to_internal_value(data)
-        self._cache[obj.pk] = obj
-        shared_cache[obj.pk] = obj
-        return obj
-
-
-class CustomFieldInstanceListSerializer(serializers.ListSerializer):
-    def to_internal_value(self, data: Any) -> list[Any]:
-        if isinstance(data, list):
-            field_ids = []
-            for item in data:
-                if not isinstance(item, dict) or "field" not in item:
-                    continue
-                try:
-                    hash(item["field"])
-                except TypeError:
-                    continue
-                field_ids.append(item["field"])
-            if field_ids:
-                self.child.fields["field"].prefetch(field_ids)
-        return super().to_internal_value(data)
-
-
 class CustomFieldInstanceSerializer(serializers.ModelSerializer[CustomFieldInstance]):
-    field = _CachingCustomFieldPrimaryKeyField(queryset=CustomField.objects.all())
+    field = serializers.PrimaryKeyRelatedField(queryset=CustomField.objects.all())
     value = ReadWriteSerializerMethodField(allow_null=True)
 
     def create(self, validated_data):
@@ -1077,7 +979,6 @@ class CustomFieldInstanceSerializer(serializers.ModelSerializer[CustomFieldInsta
 
     class Meta:
         model = CustomFieldInstance
-        list_serializer_class = CustomFieldInstanceListSerializer
         fields = [
             "value",
             "field",
@@ -1246,8 +1147,14 @@ class DocumentSerializer(
 
     def to_representation(self, instance):
         doc = super().to_representation(instance)
-        if "content" in self.fields and hasattr(instance, "effective_content"):
-            doc["content"] = getattr(instance, "effective_content") or ""
+        if "content" in self.fields and has_prefetched_effective_content(instance):
+            # 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

src/documents/tests/test_api_custom_fields.py

@@ -5,9 +5,7 @@ from unittest.mock import ANY
 
 from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
-from django.db import connection
 from django.test import override_settings
-from django.test.utils import CaptureQueriesContext
 from guardian.shortcuts import assign_perm
 from rest_framework import status
 from rest_framework.test import APITestCase
@@ -15,9 +13,6 @@ from rest_framework.test import APITestCase
 from documents.models import CustomField
 from documents.models import CustomFieldInstance
 from documents.models import Document
-from documents.serialisers import CustomFieldInstanceSerializer
-from documents.serialisers import DocumentSerializer
-from documents.tests.factories import DocumentFactory
 from documents.tests.utils import DirectoriesMixin
 
 
@@ -535,136 +530,6 @@ class TestCustomFieldsAPI(DirectoriesMixin, APITestCase):
         doc.refresh_from_db()
         self.assertEqual(len(doc.custom_fields.all()), 10)
 
-    def test_document_serializer_custom_fields_validation_batches_field_lookup(
-        self,
-    ) -> None:
-        """
-        GIVEN:
-            - A document is being validated with several custom field values
-              at once (as happens on every PATCH/PUT/POST)
-        WHEN:
-            - The serializer is validated
-        THEN:
-            - The referenced CustomField objects are resolved with a single
-              query, not one query per custom field
-        """
-        doc = DocumentFactory(mime_type="application/pdf")
-        custom_fields = [
-            CustomField.objects.create(
-                name=f"Test Custom Field {i}",
-                data_type=CustomField.FieldDataType.STRING,
-            )
-            for i in range(5)
-        ]
-
-        serializer = DocumentSerializer(
-            doc,
-            data={
-                "custom_fields": [
-                    {"field": custom_field.id, "value": "test value"}
-                    for custom_field in custom_fields
-                ],
-            },
-            partial=True,
-        )
-
-        with CaptureQueriesContext(connection) as ctx:
-            self.assertTrue(serializer.is_valid(), serializer.errors)
-
-        custom_field_lookups = [
-            query
-            for query in ctx.captured_queries
-            if 'FROM "documents_customfield" WHERE "documents_customfield"."id"'
-            in query["sql"]
-        ]
-        self.assertEqual(
-            len(custom_field_lookups),
-            1,
-            "Expected a single batched query to resolve the custom fields, "
-            f"got {len(custom_field_lookups)}: {custom_field_lookups}",
-        )
-
-    def test_custom_field_lookup_reuses_shared_context_cache(self) -> None:
-        """
-        GIVEN:
-            - A CustomField has already been resolved once, by a serializer
-              sharing a given `context` dict
-        WHEN:
-            - A second, separately-instantiated CustomFieldInstanceSerializer
-              validates the same field id, sharing that same context
-              (this is what drf-writable-nested does: it rebuilds a fresh
-              serializer -- and fresh field instances -- per item while
-              matching existing vs. new instances during save())
-        THEN:
-            - No additional query is issued to resolve the CustomField
-        """
-        custom_field = CustomField.objects.create(
-            name="Test Custom Field",
-            data_type=CustomField.FieldDataType.STRING,
-        )
-
-        context: dict = {}
-        first_pass = CustomFieldInstanceSerializer(
-            data={"field": custom_field.id, "value": "a"},
-            context=context,
-        )
-        self.assertTrue(first_pass.is_valid(), first_pass.errors)
-
-        second_pass = CustomFieldInstanceSerializer(
-            data={"field": custom_field.id, "value": "b"},
-            context=context,
-        )
-        with CaptureQueriesContext(connection) as ctx:
-            self.assertTrue(second_pass.is_valid(), second_pass.errors)
-
-        custom_field_lookups = [
-            query
-            for query in ctx.captured_queries
-            if 'FROM "documents_customfield" WHERE "documents_customfield"."id"'
-            in query["sql"]
-        ]
-        self.assertEqual(
-            len(custom_field_lookups),
-            0,
-            "Expected the second, separately-instantiated serializer to reuse "
-            f"the already-resolved CustomField, got: {custom_field_lookups}",
-        )
-
-    def test_custom_field_validation_rejects_malformed_field_value(self) -> None:
-        """
-        GIVEN:
-            - A document is being validated with a malformed custom_fields
-              entry whose "field" value is neither a valid CustomField id
-              nor a type DRF's own PrimaryKeyRelatedField can safely reject
-              on its own (unhashable, or a non-numeric scalar)
-        WHEN:
-            - The serializer is validated
-        THEN:
-            - A normal validation error is raised, not an unhandled
-              TypeError/ValueError escaping past DRF's validation layer
-        """
-        doc = DocumentFactory(mime_type="application/pdf")
-
-        bad_field_values = {
-            "unhashable-list": [],
-            "unhashable-dict": {},
-            "non-numeric-scalar": "abc",
-        }
-        for case_id, bad_field_value in bad_field_values.items():
-            with self.subTest(case_id):
-                serializer = DocumentSerializer(
-                    doc,
-                    data={
-                        "custom_fields": [
-                            {"field": bad_field_value, "value": "test value"},
-                        ],
-                    },
-                    partial=True,
-                )
-
-                self.assertFalse(serializer.is_valid())
-                self.assertIn("custom_fields", serializer.errors)
-
     def test_change_custom_field_instance_value(self) -> None:
         """
         GIVEN:

src/documents/tests/test_data_models.py

@@ -1,58 +0,0 @@
-from django.db import connection
-from django.test import TestCase
-from django.test.utils import CaptureQueriesContext
-
-from documents.data_models import DocumentMetadataOverrides
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.tests.factories import DocumentFactory
-from documents.tests.utils import DirectoriesMixin
-
-
-class TestDocumentMetadataOverridesFromDocument(DirectoriesMixin, TestCase):
-    def test_from_document_batches_custom_field_lookup_after_refresh_from_db(
-        self,
-    ) -> None:
-        """
-        GIVEN:
-            - A document has several custom field values
-            - The document instance has just been refreshed from the database,
-              which drops any prefetched related objects (as
-              send_websocket_document_updated does before building overrides)
-        WHEN:
-            - DocumentMetadataOverrides.from_document() reads the document's
-              custom field values
-        THEN:
-            - The referenced CustomField objects are resolved with a single
-              query, not one query per custom field
-        """
-        doc = DocumentFactory(mime_type="application/pdf")
-        for i in range(5):
-            CustomFieldInstance.objects.create(
-                document=doc,
-                field=CustomField.objects.create(
-                    name=f"Test Custom Field {i}",
-                    data_type=CustomField.FieldDataType.STRING,
-                ),
-                value_text="value",
-            )
-
-        doc.refresh_from_db()
-
-        with CaptureQueriesContext(connection) as ctx:
-            overrides = DocumentMetadataOverrides.from_document(doc)
-
-        self.assertEqual(len(overrides.custom_fields), 5)
-        unbatched_field_lookups = [
-            query
-            for query in ctx.captured_queries
-            if 'FROM "documents_customfield" WHERE "documents_customfield"."id"'
-            in query["sql"]
-        ]
-        self.assertEqual(
-            unbatched_field_lookups,
-            [],
-            "Expected CustomField data to come from the CustomFieldInstance "
-            "join, not a separate per-instance lookup, "
-            f"got: {unbatched_field_lookups}",
-        )

src/documents/tests/test_document_list_effective_content_annotation.py

@@ -0,0 +1,239 @@
+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 TestTrashAndGlobalSearchDoNotResolveEffectiveContent:
+    """
+    TrashView and GlobalSearchView serialize Document instances with
+    DocumentSerializer too, but build their querysets independently of
+    DocumentViewSet.get_queryset() -- and neither actually displays
+    document content. They should keep showing the document's own,
+    unresolved content with no extra query, exactly as before
+    effective_content resolution existed.
+    """
+
+    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_unresolved_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 document's own content...
+        assert response.status_code == status.HTTP_200_OK
+        [result] = [d for d in response.data["documents"] if d["id"] == root.id]
+        assert result["content"] == "own-content"
+        # ...without ever querying for versions to resolve it
+        assert _get_effective_content_fallback_queries(ctx) == []

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
 
@@ -43,6 +46,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

@@ -233,6 +233,7 @@ from documents.versioning import VersionResolutionError
 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
@@ -1085,12 +1086,40 @@ class DocumentViewSet(
             ],
         }
 
-    def get_queryset(self):
-        latest_version_content = Subquery(
-            versions_newest_first(
-                Document.objects.filter(root_document=OuterRef("pk")),
-            ).values("content")[:1],
+    # Query params whose filtering needs effective_content evaluated in SQL
+    # against every candidate row -- see _needs_effective_content_annotation().
+    _CONTENT_FILTER_PARAMS = (
+        "search",  # DRF SearchFilter's search_fields includes effective_content
+        "title_content",
+        "content__istartswith",
+        "content__iendswith",
+        "content__icontains",
+        "content__iexact",
+    )
+
+    def _needs_effective_content_annotation(self) -> bool:
+        # effective_content is a per-row correlated subquery resolving each
+        # document's latest version. Cheap when evaluated only for the page
+        # that survives filtering/sorting/pagination (the common case, via
+        # the "versions" prefetch + Document.get_effective_content()'s
+        # fallback), but if anything filters *on* it, the database has to
+        # evaluate it for every candidate row before the LIMIT is reached --
+        # pathological on MariaDB specifically for the root_document_id
+        # self-join once real candidate counts get large. Everything on this
+        # list is deprecated in favor of the Tantivy-backed search endpoint
+        # (see filters.py's TitleContentFilter/EffectiveContentFilter docs),
+        # so keep paying that cost only when one is actually used. Checked as
+        # a stripped, non-blank value (not just key presence) to match how
+        # DRF's SearchFilter and TitleContentFilter/EffectiveContentFilter
+        # themselves no-op on a blank value -- otherwise an empty `?search=`
+        # or a saved view with a cleared text filter would still pay for the
+        # annotation despite applying no actual predicate.
+        params = self.request.query_params
+        return any(
+            params.get(param, "").strip() for param in self._CONTENT_FILTER_PARAMS
         )
+
+    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.
@@ -1110,10 +1139,9 @@ 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 (
+        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(
@@ -1128,6 +1156,7 @@ class DocumentViewSet(
                         "version_index",
                     ),
                 ),
+                latest_version_content_prefetch(),
                 "tags",
                 Prefetch(
                     "custom_fields",
@@ -1136,6 +1165,16 @@ class DocumentViewSet(
                 "notes",
             )
         )
+        if self._needs_effective_content_annotation():
+            latest_version_content = Subquery(
+                versions_newest_first(
+                    Document.objects.filter(root_document=OuterRef("pk")),
+                ).values("content")[:1],
+            )
+            queryset = queryset.annotate(
+                effective_content=Coalesce(latest_version_content, F("content")),
+            )
+        return queryset
 
     def get_serializer(self, *args, **kwargs):
         fields_param = self.request.query_params.get("fields", None)