ES8 (and OpenSearch) support

.github/workflows/elixir.yml

@@ -23,9 +23,18 @@ jobs:
         include:
           - elixir: "1.15"
             otp: "25"
+            elasticsearch: "7.17.16"
+            include_tags: ""
           - elixir: "1.18"
             otp: "27"
             lint: true
+            elasticsearch: "7.17.16"
+            include_tags: ""
+          - elixir: "1.18"
+            otp: "27"
+            lint: true
+            elasticsearch: "8.13.4"
+            include_tags: "--exclude mapping_types"
     steps:
       - name: Configure sysctl limits
         run: |
@@ -37,7 +46,7 @@ jobs:
       - name: Runs Elasticsearch
         uses: elastic/elastic-github-actions/elasticsearch@master
         with:
-          stack-version: 7.17.16
+          stack-version: ${{matrix.elasticsearch}}
           security-enabled: false
 
       - name: Checkout code
@@ -63,7 +72,7 @@ jobs:
         run: mix compile --warnings-as-errors
 
       - name: Run tests
-        run: mix test
+        run: mix test ${{matrix.include_tags}}
 
       - name: checks that the mix.lock file has no unused deps 
         run: mix deps.unlock --check-unused

config/test.exs

@@ -2,4 +2,7 @@ import Config
 
 config :elastix, json_codec: Jason
 
+config :exlasticsearch, ExlasticSearch.Repo, url: "http://localhost:9200"
+config :exlasticsearch, ExlasticSearch.TypelessTestModel, index_type: :es8
+config :exlasticsearch, :monitoring, ExlasticSearch.Monitoring.Mock
 config :exlasticsearch, json_library: Jason

lib/exlasticsearch/bulk.ex

@@ -22,33 +22,46 @@ defmodule ExlasticSearch.BulkOperation do
   def bulk_operation({op_type, struct}), do: bulk_operation_default({op_type, struct, :index})
 
   defp bulk_operation_default({op_type, %{__struct__: model} = struct, index}) do
+    op = %{
+      _id: Indexable.id(struct),
+      _index: model.__es_index__(index)
+    }
+
+    op =
+      if doc_type = model.__doc_type__(),
+        do: Map.put(op, :_type, doc_type),
+        else: op
+
     [
-      %{
-        op_type => %{
-          _id: Indexable.id(struct),
-          _index: model.__es_index__(index),
-          _type: model.__doc_type__()
-        }
-      },
+      %{op_type => op},
       build_document(struct, index)
     ]
   end
 
   defp bulk_operation_update({:update, struct, id, data, index}) do
-    [
-      %{update: %{_id: id, _index: struct.__es_index__(index), _type: struct.__doc_type__()}},
-      data
-    ]
+    op = %{_id: id, _index: struct.__es_index__(index)}
+
+    if doc_type = struct.__doc_type__(),
+      do: Map.put(op, :_type, doc_type),
+      else: op
+
+    [%{update: op}, data]
   end
 
   defp bulk_operation_delete({:delete, %{__struct__: model} = struct, index}) do
+    op = %{
+      _id: Indexable.id(struct),
+      _index: model.__es_index__(index)
+    }
+
+    op =
+      if doc_type = model.__doc_type__(),
+        do: Map.put(op, :_type, doc_type),
+        else: op
+
     [
       %{
-        delete: %{
-          _id: Indexable.id(struct),
-          _index: model.__es_index__(index),
-          _type: model.__doc_type__()
-        }
+        delete: op
       }
     ]
   end

lib/exlasticsearch/model.ex

@@ -77,13 +77,15 @@ defmodule ExlasticSearch.Model do
     * `block` - the definition of the index
 
   """
-  defmacro indexes(type, block) do
+  defmacro indexes(type, opts \\ [], block) do
+    doc_type = Keyword.get(opts, :doc_type, type)
+
     quote do
       Module.register_attribute(__MODULE__, :es_mappings, accumulate: true)
       @read_version :ignore
       @index_version :ignore
 
-      def __doc_type__, do: unquote(type)
+      def __doc_type__, do: unquote(doc_type)
 
       unquote(block)
 

lib/exlasticsearch/repo.ex

@@ -71,7 +71,7 @@ defmodule ExlasticSearch.Repo do
   def create_mapping(model, index \\ :index, opts \\ []) do
     index
     |> es_url()
-    |> Mapping.put(model.__es_index__(index), model.__doc_type__(), model.__es_mappings__(), opts)
+    |> Mapping.put(model.__es_index__(index), model.__doc_type__() || "", model.__es_mappings__(), opts)
   end
 
   @doc """
@@ -172,7 +172,7 @@ defmodule ExlasticSearch.Repo do
 
     index
     |> es_url()
-    |> Document.index(model.__es_index__(index), model.__doc_type__(), id, document)
+    |> Document.index(model.__es_index__(index), model.__doc_type__() || "_doc", id, document)
     |> log_response()
     |> mark_failure()
   end
@@ -234,7 +234,7 @@ defmodule ExlasticSearch.Repo do
   def get(%{__struct__: model} = struct, index_type \\ :read) do
     index_type
     |> es_url()
-    |> Document.get(model.__es_index__(index_type), model.__doc_type__(), Indexable.id(struct))
+    |> Document.get(model.__es_index__(index_type), model.__doc_type__() || "_doc", Indexable.id(struct))
     |> log_response()
     |> decode(Response.Record, model)
   end
@@ -268,10 +268,12 @@ defmodule ExlasticSearch.Repo do
   defp model_to_index(model, index_type), do: model.__es_index__(index_type)
 
   defp model_to_doc_types(models) when is_list(models) do
-    Enum.map(models, & &1.__doc_type__())
+    Enum.flat_map(models, &model_to_doc_types/1)
   end
 
-  defp model_to_doc_types(model), do: [model.__doc_type__()]
+  defp model_to_doc_types(model) do
+    if doc_type = model.__doc_type__(), do: [doc_type], else: []
+  end
 
   @doc """
   Performs an aggregation against a query, and returns only the aggregation results.
@@ -284,9 +286,12 @@ defmodule ExlasticSearch.Repo do
 
     index_type = query.index_type || :read
 
+    doc_types = if doc_type = model.__doc_type__(), do: [doc_type], else: []
+    es_index = model.__es_index__(index_type)
+
     index_type
     |> es_url()
-    |> Search.search(model.__es_index__(index_type), [model.__doc_type__()], search, size: 0)
+    |> Search.search(es_index, doc_types, search, size: 0)
     # TODO: figure out how to decode these, it's not trivial to type them
     |> log_response()
   end
@@ -299,7 +304,7 @@ defmodule ExlasticSearch.Repo do
   def delete(%{__struct__: model} = struct, index \\ :index) do
     index
     |> es_url()
-    |> Document.delete(model.__es_index__(index), model.__doc_type__(), Indexable.id(struct))
+    |> Document.delete(model.__es_index__(index), model.__doc_type__() || "_doc", Indexable.id(struct))
     |> log_response()
     |> mark_failure()
   end