Perf: batch CustomField lookups when validating a document's custom_fields

DocumentSerializer.custom_fields validates each item's field id via a
plain PrimaryKeyRelatedField, which issues one SELECT per custom field
per validation pass (discussion #13690). Batch-resolve all field ids in
one query and cache them on the field instance so per-item validation
is free instead of re-querying.

src/documents/serialisers.py

@@ -3,6 +3,7 @@ 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
@@ -876,8 +877,50 @@ def validate_documentlink_targets(user, doc_ids):
         )
 
 
+class _CachingCustomFieldPrimaryKeyField(serializers.PrimaryKeyRelatedField):
+    """
+    A document's custom_fields are validated as a list; the default
+    PrimaryKeyRelatedField issues one SELECT per item. CustomFieldInstanceListSerializer
+    below resolves all of an incoming list's field ids in a single query up
+    front and caches them here, keyed by id, so per-item validation is free
+    instead of re-querying. The cache lives on this field instance, which
+    DRF constructs fresh for each request -- no state persists between
+    requests.
+    """
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._cache: dict[int, CustomField] = {}
+
+    def prefetch(self, ids: Iterable[int]) -> None:
+        missing = {i for i in ids if i not in self._cache}
+        if missing:
+            for obj in self.get_queryset().filter(pk__in=missing):
+                self._cache[obj.pk] = obj
+
+    def to_internal_value(self, data: int) -> CustomField:
+        if data in self._cache:
+            return self._cache[data]
+        obj: CustomField = super().to_internal_value(data)
+        self._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 = {
+                item["field"]
+                for item in data
+                if isinstance(item, dict) and "field" in item
+            }
+            if field_ids:
+                self.child.fields["field"].prefetch(field_ids)
+        return super().to_internal_value(data)
+
+
 class CustomFieldInstanceSerializer(serializers.ModelSerializer[CustomFieldInstance]):
-    field = serializers.PrimaryKeyRelatedField(queryset=CustomField.objects.all())
+    field = _CachingCustomFieldPrimaryKeyField(queryset=CustomField.objects.all())
     value = ReadWriteSerializerMethodField(allow_null=True)
 
     def create(self, validated_data):
@@ -978,6 +1021,7 @@ class CustomFieldInstanceSerializer(serializers.ModelSerializer[CustomFieldInsta
 
     class Meta:
         model = CustomFieldInstance
+        list_serializer_class = CustomFieldInstanceListSerializer
         fields = [
             "value",
             "field",

src/documents/tests/test_api_custom_fields.py

@@ -5,7 +5,9 @@ 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
@@ -13,6 +15,8 @@ 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 DocumentSerializer
+from documents.tests.factories import DocumentFactory
 from documents.tests.utils import DirectoriesMixin
 
 
@@ -530,6 +534,55 @@ 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_change_custom_field_instance_value(self) -> None:
         """
         GIVEN: