From c44721bf5951a187a89321143824196dd5dfa263 Mon Sep 17 00:00:00 2001
From: Benjamin Vetter <benjamin.vetter@wlw.de>
Date: Fri, 19 Jan 2024 10:10:52 +0100
Subject: [PATCH] Support opensearch (#39)

---
 .github/workflows/test.yml          | 45 ++++++++++++++++++++++-------
 CHANGELOG.md                        |  4 +++
 README.md                           |  5 ++--
 lib/search_flip/aggregation.rb      |  2 +-
 lib/search_flip/bulk.rb             |  2 +-
 lib/search_flip/connection.rb       | 38 +++++++++++++++---------
 lib/search_flip/criteria.rb         |  8 ++---
 lib/search_flip/index.rb            |  8 ++---
 lib/search_flip/version.rb          |  2 +-
 spec/search_flip/bulk_spec.rb       |  2 +-
 spec/search_flip/connection_spec.rb | 38 ++++++++++++++----------
 spec/search_flip/criteria_spec.rb   | 14 ++++-----
 spec/search_flip/index_spec.rb      | 16 +++++-----
 spec/spec_helper.rb                 |  8 ++---
 14 files changed, 118 insertions(+), 74 deletions(-)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index c6b46ee..9335e89 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -7,12 +7,38 @@ jobs:
       fail-fast: false
       matrix:
         elasticsearch:
-        - plainpicture/elasticsearch:2.4.1_delete-by-query
-        - elasticsearch:5.4
-        - docker.elastic.co/elasticsearch/elasticsearch:6.7.0
-        - docker.elastic.co/elasticsearch/elasticsearch:7.0.0
-        - docker.elastic.co/elasticsearch/elasticsearch:7.11.2
-        - docker.elastic.co/elasticsearch/elasticsearch:8.1.1
+        - image: plainpicture/elasticsearch:2.4.1_delete-by-query
+          env:
+            discovery.type: single-node
+            xpack.security.enabled: false
+        - image: elasticsearch:5.4
+          env:
+            discovery.type: single-node
+            xpack.security.enabled: false
+        - image: docker.elastic.co/elasticsearch/elasticsearch:6.7.0
+          env:
+            discovery.type: single-node
+            xpack.security.enabled: false
+        - image: docker.elastic.co/elasticsearch/elasticsearch:7.0.0
+          env:
+            discovery.type: single-node
+            xpack.security.enabled: false
+        - image: docker.elastic.co/elasticsearch/elasticsearch:7.11.2
+          env:
+            discovery.type: single-node
+            xpack.security.enabled: false
+        - image: docker.elastic.co/elasticsearch/elasticsearch:8.1.1
+          env:
+            discovery.type: single-node
+            xpack.security.enabled: false
+        - image: opensearchproject/opensearch:1.3.14
+          env:
+            discovery.type: single-node
+            plugins.security.disabled: true
+        - image: opensearchproject/opensearch:2.11.1
+          env:
+            discovery.type: single-node
+            plugins.security.disabled: true
         ruby:
         - 2.7
         - 3.0
@@ -20,10 +46,8 @@ jobs:
         - 3.2
     services:
       elasticsearch:
-        image: ${{ matrix.elasticsearch }}
-        env:
-          discovery.type: single-node
-          xpack.security.enabled: false
+        image: ${{ matrix.elasticsearch.image }}
+        env: ${{ matrix.elasticsearch.env }}
         ports:
           - 9200:9200
     steps:
@@ -31,7 +55,6 @@ jobs:
     - uses: ruby/setup-ruby@v1
       with:
         ruby-version: ${{ matrix.ruby }}
-    - run: gem install bundler
     - run: bundle
     - run: bundle exec rspec
     - run: bundle exec rubocop
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d915bbd..c25349d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,10 @@
 
 # CHANGELOG
 
+## v3.8.0.
+
+* Support Opensearch 1.x and 2.x
+
 ## v3.7.2
 
 * Fix wrong AWS signatures by generating the json before passing it to http-rb
diff --git a/README.md b/README.md
index 95e4b0d..91bc18c 100644
--- a/README.md
+++ b/README.md
@@ -9,8 +9,9 @@
 Using SearchFlip it is dead-simple to create index classes that correspond to
 [Elasticsearch](https://www.elastic.co/) indices and to manipulate, query and
 aggregate these indices using a chainable, concise, yet powerful DSL. Finally,
-SearchFlip supports Elasticsearch 2.x, 5.x, 6.x, 7.x and 8.x. Check section
-[Feature Support](#feature-support) for version dependent features.
+SearchFlip supports Elasticsearch 2.x, 5.x, 6.x, 7.x and 8.x as well as
+Opensearch 1.x and 2.x. Check section [Feature Support](#feature-support) for
+version dependent features.
 
 ```ruby
 CommentIndex.search("hello world", default_field: "title").where(visible: true).aggregate(:user_id).sort(id: "desc")
diff --git a/lib/search_flip/aggregation.rb b/lib/search_flip/aggregation.rb
index fb812f0..1e0d728 100644
--- a/lib/search_flip/aggregation.rb
+++ b/lib/search_flip/aggregation.rb
@@ -30,7 +30,7 @@ def to_hash
       res[:aggregations] = aggregation_values if aggregation_values
 
       if must_values || must_not_values || filter_values
-        if target.connection.version.to_i >= 2
+        if target.connection.distribution || target.connection.version.to_i >= 2
           res[:filter] = {
             bool: {}
               .merge(must_values ? { must: must_values } : {})
diff --git a/lib/search_flip/bulk.rb b/lib/search_flip/bulk.rb
index 11c6cc8..40e17ed 100644
--- a/lib/search_flip/bulk.rb
+++ b/lib/search_flip/bulk.rb
@@ -148,7 +148,7 @@ def upload
       return unless parsed_response["errors"]
 
       parsed_response["items"].each do |item|
-        item.each do |_, element|
+        item.each_value do |element|
           status = element["status"]
 
           next if status.between?(200, 299)
diff --git a/lib/search_flip/connection.rb b/lib/search_flip/connection.rb
index a6216e4..d7eb425 100644
--- a/lib/search_flip/connection.rb
+++ b/lib/search_flip/connection.rb
@@ -16,24 +16,28 @@ def initialize(options = {})
       @base_url = options[:base_url] || SearchFlip::Config[:base_url]
       @http_client = options[:http_client] || SearchFlip::HTTPClient.new
       @bulk_limit = options[:bulk_limit] || SearchFlip::Config[:bulk_limit]
-      @version_mutex = Mutex.new
+    end
+
+    # Queries and returns the Elasticsearch distribution used.
+    #
+    # @example
+    #   connection.distribution # => e.g. "opensearch"
+    #
+    # @return [String] The Elasticsearch distribution
+
+    def distribution
+      @distribution ||= SearchFlip::JSON.parse(version_response.to_s)["version"]["distribution"]
     end
 
     # Queries and returns the Elasticsearch version used.
     #
     # @example
-    #   connection.version # => e.g. 2.4.1
+    #   connection.version # => e.g. "2.4.1"
     #
     # @return [String] The Elasticsearch version
 
     def version
-      @version_mutex.synchronize do
-        @version ||= begin
-          response = http_client.headers(accept: "application/json").get("#{base_url}/")
-
-          SearchFlip::JSON.parse(response.to_s)["version"]["number"]
-        end
-      end
+      @version ||= SearchFlip::JSON.parse(version_response.to_s)["version"]["number"]
     end
 
     # Queries and returns the Elasticsearch cluster health.
@@ -64,7 +68,7 @@ def cluster_health
     def msearch(criterias)
       payload = criterias.flat_map do |criteria|
         [
-          SearchFlip::JSON.generate(index: criteria.target.index_name_with_prefix, **(version.to_i < 8 ? { type: criteria.target.type_name } : {})),
+          SearchFlip::JSON.generate(index: criteria.target.index_name_with_prefix, **(distribution.nil? && version.to_i < 8 ? { type: criteria.target.type_name } : {})),
           SearchFlip::JSON.generate(criteria.request)
         ]
       end
@@ -300,8 +304,8 @@ def refresh(index_names = nil)
     # @return [Boolean] Returns true or raises SearchFlip::ResponseError
 
     def update_mapping(index_name, mapping, type_name: nil)
-      url = type_name && version.to_i < 8 ? type_url(index_name, type_name) : index_url(index_name)
-      params = type_name && version.to_f >= 6.7 && version.to_i < 8 ? { include_type_name: true } : {}
+      url = type_name && distribution.nil? && version.to_i < 8 ? type_url(index_name, type_name) : index_url(index_name)
+      params = type_name && distribution.nil? && version.to_f >= 6.7 && version.to_i < 8 ? { include_type_name: true } : {}
 
       http_client.put("#{url}/_mapping", params: params, json: mapping)
 
@@ -318,8 +322,8 @@ def update_mapping(index_name, mapping, type_name: nil)
     # @return [Hash] The current type mapping
 
     def get_mapping(index_name, type_name: nil)
-      url = type_name && version.to_i < 8 ? type_url(index_name, type_name) : index_url(index_name)
-      params = type_name && version.to_f >= 6.7 && version.to_i < 8 ? { include_type_name: true } : {}
+      url = type_name && distribution.nil? && version.to_i < 8 ? type_url(index_name, type_name) : index_url(index_name)
+      params = type_name && distribution.nil? && version.to_f >= 6.7 && version.to_i < 8 ? { include_type_name: true } : {}
 
       response = http_client.headers(accept: "application/json").get("#{url}/_mapping", params: params)
 
@@ -422,5 +426,11 @@ def type_url(index_name, type_name)
     def index_url(index_name)
       "#{base_url}/#{index_name}"
     end
+
+    private
+
+    def version_response
+      @version_response ||= http_client.headers(accept: "application/json").get("#{base_url}/")
+    end
   end
 end
diff --git a/lib/search_flip/criteria.rb b/lib/search_flip/criteria.rb
index 54e2ac5..e873171 100644
--- a/lib/search_flip/criteria.rb
+++ b/lib/search_flip/criteria.rb
@@ -350,8 +350,8 @@ def delete(params = {})
       http_request = connection.http_client
       http_request = http_request.timeout(http_timeout_value) if http_timeout_value
 
-      if connection.version.to_i >= 5
-        url = connection.version.to_i < 8 ? target.type_url : target.index_url
+      if connection.distribution || connection.version.to_i >= 5
+        url = connection.distribution.nil? && connection.version.to_i < 8 ? target.type_url : target.index_url
 
         http_request.post("#{url}/_delete_by_query", params: request_params.merge(params), json: dupped_request)
       else
@@ -622,7 +622,7 @@ def execute!
             json: { scroll: scroll_args[:timeout], scroll_id: scroll_args[:id] }
           )
         elsif scroll_args
-          url = connection.version.to_i < 8 ? target.type_url : target.index_url
+          url = connection.distribution.nil? && connection.version.to_i < 8 ? target.type_url : target.index_url
 
           http_request.post(
             "#{url}/_search",
@@ -630,7 +630,7 @@ def execute!
             json: request
           )
         else
-          url = connection.version.to_i < 8 ? target.type_url : target.index_url
+          url = connection.distribution.nil? && connection.version.to_i < 8 ? target.type_url : target.index_url
 
           http_request.post("#{url}/_search", params: request_params, json: request)
         end
diff --git a/lib/search_flip/index.rb b/lib/search_flip/index.rb
index 3205805..00f46bc 100644
--- a/lib/search_flip/index.rb
+++ b/lib/search_flip/index.rb
@@ -438,7 +438,7 @@ def get_mapping
       # equal to _doc.
 
       def include_type_name?
-        type_name != "_doc" || connection.version.to_i < 7
+        type_name != "_doc" || (connection.distribution.nil? && connection.version.to_i < 7)
       end
 
       # Retrieves the document specified by id from Elasticsearch. Raises
@@ -455,7 +455,7 @@ def include_type_name?
       # @return [Hash] The specified document
 
       def get(id, params = {})
-        url = connection.version.to_i < 8 ? type_url : "#{index_url}/_doc"
+        url = connection.distribution.nil? && connection.version.to_i < 8 ? type_url : "#{index_url}/_doc"
         response = connection.http_client.headers(accept: "application/json").get("#{url}/#{id}", params: params)
 
         SearchFlip::JSON.parse(response.to_s)
@@ -474,7 +474,7 @@ def get(id, params = {})
       # @return [Hash] The raw response
 
       def mget(request, params = {})
-        url = connection.version.to_i < 8 ? type_url : index_url
+        url = connection.distribution.nil? && connection.version.to_i < 8 ? type_url : index_url
         response = connection.http_client.headers(accept: "application/json").post("#{url}/_mget", json: request, params: params)
 
         SearchFlip::JSON.parse(response.to_s)
@@ -633,7 +633,7 @@ def bulk(options = {})
           bulk_max_mb: connection.bulk_max_mb
         }
 
-        url = connection.version.to_i < 8 ? type_url : index_url
+        url = connection.distribution.nil? && connection.version.to_i < 8 ? type_url : index_url
 
         SearchFlip::Bulk.new("#{url}/_bulk", default_options.merge(options)) do |indexer|
           yield indexer
diff --git a/lib/search_flip/version.rb b/lib/search_flip/version.rb
index 567962d..c0a0423 100644
--- a/lib/search_flip/version.rb
+++ b/lib/search_flip/version.rb
@@ -1,3 +1,3 @@
 module SearchFlip
-  VERSION = "3.7.2"
+  VERSION = "3.8.0"
 end
diff --git a/spec/search_flip/bulk_spec.rb b/spec/search_flip/bulk_spec.rb
index dd79d67..be3e8cf 100644
--- a/spec/search_flip/bulk_spec.rb
+++ b/spec/search_flip/bulk_spec.rb
@@ -60,7 +60,7 @@
 
     it "uses the specified http_client" do
       product = create(:product)
-      url = ProductIndex.connection.version.to_i < 8 ? ProductIndex.type_url : ProductIndex.index_url
+      url = ProductIndex.connection.distribution.nil? && ProductIndex.connection.version.to_i < 8 ? ProductIndex.type_url : ProductIndex.index_url
 
       stub_request(:put, "#{url}/_bulk").with(headers: { "X-Header" => "Value" }).to_return(status: 200, body: "{}")
 
diff --git a/spec/search_flip/connection_spec.rb b/spec/search_flip/connection_spec.rb
index fe5eee9..b566ad2 100644
--- a/spec/search_flip/connection_spec.rb
+++ b/spec/search_flip/connection_spec.rb
@@ -1,6 +1,12 @@
 require File.expand_path("../spec_helper", __dir__)
 
 RSpec.describe SearchFlip::Connection do
+  describe "#distribution" do
+    it "reutrns the distribution" do
+      expect([nil, "opensearch"]).to include(SearchFlip::Connection.new.distribution)
+    end
+  end
+
   describe "#version" do
     it "returns the version" do
       expect(SearchFlip::Connection.new.version).to match(/\A[0-9.]+\z/)
@@ -92,7 +98,7 @@
     it "returns the specified indices" do
       connection = SearchFlip::Connection.new
 
-      expect(connection.get_indices.to_set { |index| index["index"] }).to eq(["comments", "products"].to_set)
+      expect(connection.get_indices.map { |index| index["index"] }.grep_v(/^\./).to_set).to eq(["comments", "products"].to_set)
       expect(connection.get_indices("com*").map { |index| index["index"] }).to eq(["comments"])
     end
 
@@ -170,7 +176,7 @@
     it "freezes the specified index" do
       connection = SearchFlip::Connection.new
 
-      if connection.version.to_f >= 6.6 && connection.version.to_i < 8
+      if connection.distribution.nil? && connection.version.to_f >= 6.6 && connection.version.to_i < 8
         begin
           connection.create_index("index_name")
           connection.freeze_index("index_name")
@@ -187,7 +193,7 @@
     it "unfreezes the specified index" do
       connection = SearchFlip::Connection.new
 
-      if connection.version.to_f >= 6.6 && connection.version.to_i < 8
+      if connection.distribution.nil? && connection.version.to_f >= 6.6 && connection.version.to_i < 8
         begin
           connection.create_index("index_name")
           connection.freeze_index("index_name")
@@ -241,7 +247,7 @@
   end
 
   describe "#update_mapping" do
-    if SearchFlip::Connection.new.version.to_i >= 7
+    if SearchFlip::Connection.new.then { |connection| connection.distribution || connection.version.to_i >= 7 }
       it "updates the mapping of an index without type name" do
         begin
           connection = SearchFlip::Connection.new
@@ -266,7 +272,7 @@
 
         connection.create_index("index_name")
 
-        if connection.version.to_i < 8
+        if connection.distribution.nil? && connection.version.to_i < 8
           connection.update_mapping("index_name", { "type_name" => mapping }, type_name: "type_name")
 
           expect(connection.get_mapping("index_name", type_name: "type_name")).to eq("index_name" => { "mappings" => { "type_name" => mapping } })
@@ -321,9 +327,9 @@
 
       bulk = proc do
         connection.bulk do |indexer|
-          indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
-          indexer.index 2, { id: 2 }, _index: ProductIndex.index_name, ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
-          indexer.index 1, { id: 1 }, _index: CommentIndex.index_name, ** connection.version.to_i < 8 ? { _type: CommentIndex.type_name } : {}
+          indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+          indexer.index 2, { id: 2 }, _index: ProductIndex.index_name, ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+          indexer.index 1, { id: 1 }, _index: CommentIndex.index_name, ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: CommentIndex.type_name } : {}
         end
       end
 
@@ -346,14 +352,14 @@
       connection = SearchFlip::Connection.new
 
       connection.bulk do |indexer|
-        indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
-        indexer.index 2, { id: 2 }, _index: ProductIndex.index_name, ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+        indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+        indexer.index 2, { id: 2 }, _index: ProductIndex.index_name, ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
       end
 
       bulk = proc do
         connection.bulk do |indexer|
-          indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, version: 1, version_type: "external", ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
-          indexer.index 2, { id: 2 }, _index: ProductIndex.index_name, version: 1, version_type: "external", ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+          indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, version: 1, version_type: "external", ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+          indexer.index 2, { id: 2 }, _index: ProductIndex.index_name, version: 1, version_type: "external", ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
         end
       end
 
@@ -361,8 +367,8 @@
 
       bulk = proc do
         connection.bulk ignore_errors: [409] do |indexer|
-          indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, version: 1, version_type: "external", ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
-          indexer.index 2, { id: 2 }, _index: ProductIndex.index_name, version: 1, version_type: "external", ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+          indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, version: 1, version_type: "external", ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+          indexer.index 2, { id: 2 }, _index: ProductIndex.index_name, version: 1, version_type: "external", ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
         end
       end
 
@@ -375,7 +381,7 @@
       connection = SearchFlip::Connection.new
 
       connection.bulk do |indexer|
-        indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+        indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
       end
 
       expect(SearchFlip::Bulk).to have_received(:new).with(
@@ -398,7 +404,7 @@
       }
 
       connection.bulk(options) do |indexer|
-        indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, ** connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
+        indexer.index 1, { id: 1 }, _index: ProductIndex.index_name, ** connection.distribution.nil? && connection.version.to_i < 8 ? { _type: ProductIndex.type_name } : {}
       end
 
       expect(SearchFlip::Bulk).to have_received(:new).with(anything, options)
diff --git a/spec/search_flip/criteria_spec.rb b/spec/search_flip/criteria_spec.rb
index d7e1190..91aaa4a 100644
--- a/spec/search_flip/criteria_spec.rb
+++ b/spec/search_flip/criteria_spec.rb
@@ -435,7 +435,7 @@
 
   describe "#match_none" do
     it "does not match any documents" do
-      if ProductIndex.connection.version.to_i >= 5
+      if ProductIndex.connection.distribution || ProductIndex.connection.version.to_i >= 5
         ProductIndex.import create(:product)
 
         query = ProductIndex.match_none
@@ -479,7 +479,7 @@
 
   describe "#post_search" do
     it "sets up the constraints correctly and is chainable" do
-      if ProductIndex.connection.version.to_i >= 2
+      if ProductIndex.connection.distribution || ProductIndex.connection.version.to_i >= 2
         product1 = create(:product, title: "title1", category: "category1")
         product2 = create(:product, title: "title2", category: "category2")
         product3 = create(:product, title: "title3", category: "category1")
@@ -886,7 +886,7 @@
 
   describe "#profile" do
     it "sets up the constraints correctly" do
-      if ProductIndex.connection.version.to_i >= 2
+      if ProductIndex.connection.distribution || ProductIndex.connection.version.to_i >= 2
         expect(ProductIndex.profile(true).raw_response["profile"]).not_to be_nil
       end
     end
@@ -1322,7 +1322,7 @@
 
   describe "#track_total_hits" do
     it "is added to the request" do
-      if ProductIndex.connection.version.to_i >= 7
+      if ProductIndex.connection.distribution || ProductIndex.connection.version.to_i >= 7
         query = ProductIndex.track_total_hits(false)
 
         expect(query.request[:track_total_hits]).to eq(false)
@@ -1350,7 +1350,7 @@
 
   describe "#preference" do
     it "sets the preference" do
-      url = ProductIndex.connection.version.to_i < 8 ? "products/products" : "products"
+      url = ProductIndex.connection.distribution.nil? && ProductIndex.connection.version.to_i < 8 ? "products/products" : "products"
 
       stub_request(:post, "http://127.0.0.1:9200/#{url}/_search?preference=value")
         .to_return(status: 200, headers: { content_type: "application/json" }, body: "{}")
@@ -1363,7 +1363,7 @@
 
   describe "#search_type" do
     it "sets the search_type" do
-      url = ProductIndex.connection.version.to_i < 8 ? "products/products" : "products"
+      url = ProductIndex.connection.distribution.nil? && ProductIndex.connection.version.to_i < 8 ? "products/products" : "products"
 
       stub_request(:post, "http://127.0.0.1:9200/#{url}/_search?search_type=value")
         .to_return(status: 200, headers: { content_type: "application/json" }, body: "{}")
@@ -1376,7 +1376,7 @@
 
   describe "#routing" do
     it "sets the search_type" do
-      url = ProductIndex.connection.version.to_i < 8 ? "products/products" : "products"
+      url = ProductIndex.connection.distribution.nil? && ProductIndex.connection.version.to_i < 8 ? "products/products" : "products"
 
       stub_request(:post, "http://127.0.0.1:9200/#{url}/_search?routing=value")
         .to_return(status: 200, headers: { content_type: "application/json" }, body: "{}")
diff --git a/spec/search_flip/index_spec.rb b/spec/search_flip/index_spec.rb
index 9ed30f5..b5f61e0 100644
--- a/spec/search_flip/index_spec.rb
+++ b/spec/search_flip/index_spec.rb
@@ -77,7 +77,7 @@
         include SearchFlip::Index
       end
 
-      expect(klass.include_type_name?).to eq(klass.connection.version.to_i < 7)
+      expect(klass.include_type_name?).to eq(klass.connection.distribution.nil? && klass.connection.version.to_i < 7)
     end
 
     it "returns true if the type name is not equal to _doc" do
@@ -185,7 +185,7 @@ def self.type_name
   end
 
   describe ".update_mapping" do
-    if TestIndex.connection.version.to_i >= 7
+    if TestIndex.connection.distribution || TestIndex.connection.version.to_i >= 7
       context "without type name" do
         it "delegates to connection" do
           TestIndex.create_index
@@ -215,7 +215,7 @@ def self.type_name
 
         TestIndex.update_mapping
 
-        if TestIndex.connection.version.to_i < 8
+        if TestIndex.connection.distribution.nil? && TestIndex.connection.version.to_i < 8
           expect(TestIndex.connection).to have_received(:update_mapping).with("test", { "test" => mapping }, type_name: "test")
         else
           expect(TestIndex.connection).to have_received(:update_mapping).with("test", mapping)
@@ -245,7 +245,7 @@ def self.include_type_name?
   end
 
   describe ".get_mapping" do
-    if TestIndex.connection.version.to_i >= 7
+    if TestIndex.connection.distribution || TestIndex.connection.version.to_i >= 7
       context "without type name" do
         it "delegates to connection" do
           allow(TestIndex).to receive(:include_type_name?).and_return(false)
@@ -272,7 +272,7 @@ def self.include_type_name?
 
         TestIndex.get_mapping
 
-        if TestIndex.connection.version.to_i < 8
+        if TestIndex.connection.distribution.nil? && TestIndex.connection.version.to_i < 8
           expect(TestIndex.connection).to have_received(:get_mapping).with("test", type_name: "test")
         else
           expect(TestIndex.connection).to have_received(:get_mapping).with("test")
@@ -340,7 +340,7 @@ def self.include_type_name?
 
       TestIndex.type_url
 
-      expect(TestIndex.connection).to have_received(:type_url).with("test", TestIndex.connection.version.to_i < 8 ? "test" : "_doc")
+      expect(TestIndex.connection).to have_received(:type_url).with("test", TestIndex.connection.distribution.nil? && TestIndex.connection.version.to_i < 8 ? "test" : "_doc")
     end
   end
 
@@ -433,7 +433,7 @@ def self.include_type_name?
 
       products = create_list(:product, 2)
 
-      if ProductIndex.connection.version.to_i >= 5
+      if ProductIndex.connection.distribution || ProductIndex.connection.version.to_i >= 5
         expect { ProductIndex.create products, {}, routing: "r1" }.to(change { ProductIndex.total_count }.by(2))
 
         expect(ProductIndex.get(products.first.id, routing: "r1")["_routing"]).to eq("r1")
@@ -447,7 +447,7 @@ def self.include_type_name?
     it "allows respects class options" do
       products = create_list(:product, 2)
 
-      if ProductIndex.connection.version.to_i >= 5
+      if ProductIndex.connection.distribution || ProductIndex.connection.version.to_i >= 5
         allow(ProductIndex).to receive(:index_options).and_return(routing: "r1")
 
         expect { ProductIndex.create products }.to(change { ProductIndex.total_count }.by(2))
diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb
index c8fd3c9..629515e 100644
--- a/spec/spec_helper.rb
+++ b/spec/spec_helper.rb
@@ -86,7 +86,7 @@ class CommentIndex
   include SearchFlip::Index
 
   def self.type_name
-    connection.version.to_i < 8 ? "comments" : "_doc"
+    connection.distribution.nil? && connection.version.to_i < 8 ? "comments" : "_doc"
   end
 
   def self.index_name
@@ -113,7 +113,7 @@ class ProductIndex
   include SearchFlip::Index
 
   def self.mapping
-    if ProductIndex.connection.version.to_i >= 5
+    if ProductIndex.connection.distribution || ProductIndex.connection.version.to_i >= 5
       {
         properties: {
           category: {
@@ -136,7 +136,7 @@ def self.mapping
   end
 
   def self.type_name
-    connection.version.to_i < 8 ? "products" : "_doc"
+    connection.distribution.nil? && connection.version.to_i < 8 ? "products" : "_doc"
   end
 
   def self.index_name
@@ -177,7 +177,7 @@ def self.mapping
   end
 
   def self.type_name
-    connection.version.to_i < 8 ? "test" : "_doc"
+    connection.distribution.nil? && connection.version.to_i < 8 ? "test" : "_doc"
   end
 
   def self.index_name