fix: offload sync Pydantic serialization to thread in async write path

AsyncPointsApi methods for `upsert_points`, `batch_update`, and `update_vectors` call `_build_for_*` which runs `jsonable_encoder` (-> `model_dump_json`) synchronously before the first `await`. This blocks the event loop for the entire duration of Pydantic serialization, which scales linearly with point count and vector dimensionality.

Move the sync `_build_for_*` call into `run_in_executor` for these three methods, whose body size is unbounded. The `_build_for_*` methods return an unawaited coroutine object (from the async api_client.request), so the executor runs the serialization in a thread and we await the resulting coroutine back on the event loop.

Fixes #1175

Author: charles-dyfis-net

qdrant_client/http/api/points_api.py

@@ -1,4 +1,6 @@
 # flake8: noqa E501
+import asyncio
+import functools
 from typing import TYPE_CHECKING, Any, Dict, Set, TypeVar, Union
 
 from pydantic import BaseModel
@@ -562,13 +564,26 @@ async def batch_update(
         """
         Apply a series of update operations for points, vectors and payloads
         """
-        return await self._build_for_batch_update(
-            collection_name=collection_name,
-            wait=wait,
-            ordering=ordering,
-            timeout=timeout,
-            update_operations=update_operations,
+        # _build_for_batch_update is a sync method that:
+        #   1. serializes the body with jsonable_encoder (CPU-bound, can be slow)
+        #   2. calls self.api_client.request(...) which, for AsyncApiClient, is
+        #      async def — so calling it merely *creates* a coroutine object
+        #      without executing it
+        # We run the whole sync method in an executor to move the serialization
+        # off the event loop, then await the returned coroutine here.
+        loop = asyncio.get_running_loop()
+        coro = await loop.run_in_executor(
+            None,
+            functools.partial(
+                self._build_for_batch_update,
+                collection_name=collection_name,
+                wait=wait,
+                ordering=ordering,
+                timeout=timeout,
+                update_operations=update_operations,
+            ),
         )
+        return await coro
 
     async def clear_payload(
         self,
@@ -778,13 +793,22 @@ async def update_vectors(
         """
         Update specified named vectors on points, keep unspecified vectors intact.
         """
-        return await self._build_for_update_vectors(
-            collection_name=collection_name,
-            wait=wait,
-            ordering=ordering,
-            timeout=timeout,
-            update_vectors=update_vectors,
+        # See batch_update for rationale: _build_for_update_vectors is sync,
+        # returns an unawaited coroutine, and its jsonable_encoder call is the
+        # expensive part we need off the event loop.
+        loop = asyncio.get_running_loop()
+        coro = await loop.run_in_executor(
+            None,
+            functools.partial(
+                self._build_for_update_vectors,
+                collection_name=collection_name,
+                wait=wait,
+                ordering=ordering,
+                timeout=timeout,
+                update_vectors=update_vectors,
+            ),
         )
+        return await coro
 
     async def upsert_points(
         self,
@@ -797,13 +821,22 @@ async def upsert_points(
         """
         Perform insert + updates on points. If point with given ID already exists - it will be overwritten.
         """
-        return await self._build_for_upsert_points(
-            collection_name=collection_name,
-            wait=wait,
-            ordering=ordering,
-            timeout=timeout,
-            point_insert_operations=point_insert_operations,
+        # See batch_update for rationale: _build_for_upsert_points is sync,
+        # returns an unawaited coroutine, and its jsonable_encoder call is the
+        # expensive part we need off the event loop.
+        loop = asyncio.get_running_loop()
+        coro = await loop.run_in_executor(
+            None,
+            functools.partial(
+                self._build_for_upsert_points,
+                collection_name=collection_name,
+                wait=wait,
+                ordering=ordering,
+                timeout=timeout,
+                point_insert_operations=point_insert_operations,
+            ),
         )
+        return await coro
 
 
 class SyncPointsApi(_PointsApi):