test?

Author: sleepyStick

pymongo/asynchronous/bulk.py

@@ -233,14 +233,17 @@ def gen_ordered(
         operation, in the order **provided**.
         """
         run = None
+        ctr = 0
         for idx, request in enumerate(requests):
             retryable = process(request)
             (op_type, operation) = self.ops[idx]
             if run is None:
                 run = _Run(op_type)
-            elif run.op_type != op_type:
+            elif run.op_type != op_type or ctr >= common.MAX_WRITE_BATCH_SIZE // 200:
                 yield run
+                ctr = 0
                 run = _Run(op_type)
+            ctr += 1
             run.add(idx, operation)
             run.is_retryable = run.is_retryable and retryable
         if run is None:
@@ -604,6 +607,9 @@ async def _execute_command(
                 break
             # Reset our state
             self.current_run = run = self.next_run
+            import gc
+
+            gc.collect()
 
     async def execute_command(
         self,

pymongo/synchronous/bulk.py

@@ -233,14 +233,17 @@ def gen_ordered(
         operation, in the order **provided**.
         """
         run = None
+        ctr = 0
         for idx, request in enumerate(requests):
             retryable = process(request)
             (op_type, operation) = self.ops[idx]
             if run is None:
                 run = _Run(op_type)
-            elif run.op_type != op_type:
+            elif run.op_type != op_type or ctr >= common.MAX_WRITE_BATCH_SIZE // 200:
                 yield run
+                ctr = 0
                 run = _Run(op_type)
+            ctr += 1
             run.add(idx, operation)
             run.is_retryable = run.is_retryable and retryable
         if run is None:
@@ -604,6 +607,9 @@ def _execute_command(
                 break
             # Reset our state
             self.current_run = run = self.next_run
+            import gc
+
+            gc.collect()
 
     def execute_command(
         self,

test/asynchronous/test_bulk.py

@@ -314,6 +314,14 @@ async def test_numerous_inserts_generator(self):
         self.assertEqual(n_docs, result.inserted_count)
         self.assertEqual(n_docs, await self.coll.count_documents({}))
 
+    async def test_huge_inserts_generator(self):
+        # Ensure we don't exceed server's maxWriteBatchSize size limit.
+        n_docs = 1000000
+        requests = (InsertOne({"x": "large" * 1024 * 1024}) for _ in range(n_docs))
+        result = await self.coll.bulk_write(requests)
+        self.assertEqual(n_docs, result.inserted_count)
+        self.assertEqual(n_docs, await self.coll.count_documents({}))
+
     async def test_bulk_max_message_size(self):
         await self.coll.delete_many({})
         self.addAsyncCleanup(self.coll.delete_many, {})

test/test_bulk.py

@@ -314,6 +314,14 @@ def test_numerous_inserts_generator(self):
         self.assertEqual(n_docs, result.inserted_count)
         self.assertEqual(n_docs, self.coll.count_documents({}))
 
+    def test_huge_inserts_generator(self):
+        # Ensure we don't exceed server's maxWriteBatchSize size limit.
+        n_docs = 1000000
+        requests = (InsertOne({"x": "large" * 1024 * 1024}) for _ in range(n_docs))
+        result = self.coll.bulk_write(requests)
+        self.assertEqual(n_docs, result.inserted_count)
+        self.assertEqual(n_docs, self.coll.count_documents({}))
+
     def test_bulk_max_message_size(self):
         self.coll.delete_many({})
         self.addCleanup(self.coll.delete_many, {})