From e63d00f92a46f02c2668ec2bea5fb20d29de4827 Mon Sep 17 00:00:00 2001 From: Tai An Date: Thu, 30 Jul 2026 00:22:33 -0700 Subject: [PATCH] fix(batch): require both counts unchanged before the idle flush The wait loop in the batching threads breaks out early when a second has passed and nothing new arrived, but the guard ORs the two "unchanged" checks: if time.time() - start >= 1 and ( len_o == len(self.__batch_objects) or len_r == len(self.__batch_references) ): In an objects-only batch -- what `add_object` without `add_reference` produces -- `len_r` and `len(self.__batch_references)` are both always 0, so the right operand is always True and the whole guard collapses to `if time.time() - start >= 1`. The loop then stops waiting on the first tick past one second no matter how fast objects are still streaming in, so the batch is flushed well short of its configured size. A references-only batch has the mirror-image problem. The comment above the break says "no new objects were added in the last second", i.e. the batch is idle -- which is true only when *both* counts are unchanged. Switch `or` to `and` at the three places that share this loop shape (sync.py, async_.py, base.py). Fixes #2105 --- weaviate/collections/batch/async_.py | 2 +- weaviate/collections/batch/base.py | 2 +- weaviate/collections/batch/sync.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/weaviate/collections/batch/async_.py b/weaviate/collections/batch/async_.py index 8b997586c..c63ec2106 100644 --- a/weaviate/collections/batch/async_.py +++ b/weaviate/collections/batch/async_.py @@ -225,7 +225,7 @@ async def __loop(self) -> None: # wait for more objects to be added up to the batch size await asyncio.sleep(refresh_time) if time.time() - start >= 1 and ( - len_o == len(self.__batch_objects) or len_r == len(self.__batch_references) + len_o == len(self.__batch_objects) and len_r == len(self.__batch_references) ): # no new objects were added in the last second, exit the loop break diff --git a/weaviate/collections/batch/base.py b/weaviate/collections/batch/base.py index af6a9ea49..721b9b61d 100644 --- a/weaviate/collections/batch/base.py +++ b/weaviate/collections/batch/base.py @@ -434,7 +434,7 @@ def __batch_send(self) -> None: # shutdown was requested, exit the loop break if time.time() - start >= 1 and ( - len_o == len(self.__batch_objects) or len_r == len(self.__batch_references) + len_o == len(self.__batch_objects) and len_r == len(self.__batch_references) ): # no new objects were added in the last second, exit the loop break diff --git a/weaviate/collections/batch/sync.py b/weaviate/collections/batch/sync.py index 6627f8911..6cf8c1edc 100644 --- a/weaviate/collections/batch/sync.py +++ b/weaviate/collections/batch/sync.py @@ -177,7 +177,7 @@ def __loop(self) -> None: # wait for more objects to be added up to the batch size time.sleep(refresh_time) if time.time() - start >= 1 and ( - len_o == len(self.__batch_objects) or len_r == len(self.__batch_references) + len_o == len(self.__batch_objects) and len_r == len(self.__batch_references) ): # no new objects were added in the last second, exit the loop break