fix: count queries for ES >= 7.14

neo4j-app/neo4j_app/core/elasticsearch/client.py

@@ -375,10 +375,7 @@ async def _fill_import_queue(
         buffer = []
         raw_progress = None
         if progress is not None and bodies:
-            count_queries = [deepcopy(b) for b in bodies]
-            for q in count_queries:
-                q.pop(SLICE, None)
-                q.pop(SOURCE, None)
+            count_queries = [{QUERY: b[QUERY]} for b in bodies]
             count_tasks = (self.count(index=index, body=b) for b in count_queries)
             res_it = run_with_concurrency(count_tasks, max_concurrency=max_concurrency)
             ne_counts = [c[COUNT] async for c in res_it]

neo4j-app/neo4j_app/tests/core/test_imports.py

@@ -1,6 +1,7 @@
 # pylint: disable=redefined-outer-name
 import itertools
 import json
+import logging
 import os
 import shutil
 import subprocess
@@ -40,6 +41,8 @@
     index_noise,
 )
 
+logger = logging.getLogger(__name__)
+
 
 def _make_email(doc_id: str, header_field: str) -> Dict:
     return {
@@ -175,6 +178,10 @@ async def test_import_documents(
     max_records_in_memory = 10
     neo4j_transaction_batch_size = 3
 
+    # Add a dummy progress to test the count
+    async def _logging_progress(p: float):
+        logger.debug("progressing: %s", p)
+
     # When
     response = await import_documents(
         project=TEST_PROJECT,
@@ -186,6 +193,7 @@ async def test_import_documents(
         neo4j_import_batch_size=neo4j_import_batch_size,
         neo4j_transaction_batch_size=neo4j_transaction_batch_size,
         max_records_in_memory=max_records_in_memory,
+        progress=_logging_progress,
     )
 
     # Then