feat: add support for Elasticsearch 7.16.2 (#3318)

* bump elastic to 7.16.2+

* decouple Elasticsearch and Opensearch

use method override instead of func variables

fix mypy

default value

fix broken tests

update schema

* relax version pin

* rename the base class

* rename module

* fix import order

* do not run the new tests in the old job

* remove outdated TODO

Author: masci

.github/workflows/tests.yml

@@ -315,7 +315,7 @@ jobs:
       env:
         TOKENIZERS_PARALLELISM: 'false'
       run: |
-        pytest ${{ env.PYTEST_PARAMS }} -m "opensearch and not integration" test/document_stores/ --document_store_type=opensearch
+        pytest ${{ env.PYTEST_PARAMS }} -m "opensearch and not integration" test/document_stores/test_document_store.py --document_store_type=opensearch
 
     - name: Dump docker logs on failure
       if: failure()

haystack/document_stores/elasticsearch.py

@@ -8,7 +8,7 @@
 
 try:
     from opensearchpy import OpenSearch, Urllib3HttpConnection, RequestsHttpConnection, NotFoundError, RequestError
-    from opensearchpy.helpers import bulk
+    from opensearchpy.helpers import bulk, scan
 except (ImportError, ModuleNotFoundError) as e:
     from haystack.utils.import_utils import _optional_component_not_installed
 
@@ -21,7 +21,7 @@
 from haystack.errors import DocumentStoreError
 from haystack.nodes.retriever import DenseRetriever
 
-from .elasticsearch import BaseElasticsearchDocumentStore, prepare_hosts
+from .search_engine import SearchEngineDocumentStore, prepare_hosts
 
 logger = logging.getLogger(__name__)
 
@@ -35,7 +35,7 @@
 }
 
 
-class OpenSearchDocumentStore(BaseElasticsearchDocumentStore):
+class OpenSearchDocumentStore(SearchEngineDocumentStore):
     def __init__(
         self,
         scheme: str = "https",  # Mind this different default param
@@ -208,6 +208,17 @@ def __init__(
             synonym_type=synonym_type,
         )
 
+        # Let the base class catch the right error from the Opensearch client
+        self._RequestError = RequestError
+
+    def _do_bulk(self, *args, **kwargs):
+        """Override the base class method to use the Opensearch client"""
+        return bulk(*args, **kwargs)
+
+    def _do_scan(self, *args, **kwargs):
+        """Override the base class method to use the Opensearch client"""
+        return scan(*args, **kwargs)
+
     @classmethod
     def _init_client(
         cls,