fix: shard may be empty (#1689)

jina/drivers/search.py

@@ -120,6 +120,10 @@ def _apply_all(self, docs: 'DocumentSet', *args, **kwargs) -> None:
         if bad_docs:
             self.logger.warning(f'these bad docs can not be added: {bad_docs}')
         idx, dist = self.exec_fn(embed_vecs, top_k=int(self.top_k))
+
+        if idx is None and dist is None:
+            return
+
         op_name = self.exec.__class__.__name__
         for doc, topks, scores in zip(doc_pts, idx, dist):
 

tests/integration/issues/github_1684/flow.yml

@@ -0,0 +1,4 @@
+!Flow
+pods:
+  indexer:
+    uses: index.yml

tests/integration/issues/github_1684/index.yml

@@ -0,0 +1,6 @@
+!NumpyIndexer
+with:
+  index_filename: vec.gz
+metas:
+  workspace: $JINA_TEST_1684_WORKSPACE
+

tests/integration/issues/github_1684/test_empty_shard.py

@@ -0,0 +1,39 @@
+import os
+
+import pytest
+import numpy as np
+
+from jina.flow import Flow
+from jina import Document
+
+cur_dir = os.path.dirname(os.path.abspath(__file__))
+
+
+@pytest.fixture
+def workdir(tmpdir):
+    os.environ['JINA_TEST_1684_WORKSPACE'] = str(tmpdir)
+    yield
+    del os.environ['JINA_TEST_1684_WORKSPACE']
+
+
+def test_empty_shard(mocker, workdir):
+    doc = Document()
+    doc.text = 'text'
+    doc.embedding = np.array([1, 1, 1])
+    mock = mocker.Mock()
+
+    def validate_response(resp):
+        mock()
+        assert len(resp.docs) == 1
+        assert len(resp.docs[0].matches) == 0
+
+    error_mock = mocker.Mock()
+
+    def on_error_call():
+        error_mock()
+
+    with Flow.load_config(os.path.join(cur_dir, 'flow.yml')) as f:
+        f.search([doc], on_done=validate_response, on_error=on_error_call)
+
+    mock.assert_called_once()
+    error_mock.assert_not_called()

tests/integration/level_depth/test_search_different_depths.py

@@ -1,5 +1,3 @@
-import os
-
 import pytest
 
 from jina.flow import Flow