Support _routing meta-field

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Change Log
 
+## [v0.5.0](https://github.com/infinitered/elasticsearch-elixir/tree/v0.5.0) (2018-08-01)
+[Full Changelog](https://github.com/infinitered/elasticsearch-elixir/compare/v0.4.1...v0.5.0)
+
+**Closed issues:**
+
+**Merged pull requests:**
+
+- Add support for routing keys in Document protocol. Add breaking change
+documentation.
+
 ## [v0.4.1](https://github.com/infinitered/elasticsearch-elixir/tree/v0.4.1) (2018-06-26)
 [Full Changelog](https://github.com/infinitered/elasticsearch-elixir/compare/v0.4.0...v0.4.1)
 
@@ -90,4 +100,4 @@
 
 
 
-\* *This Change Log was automatically generated by [github_changelog_generator](https://github.com/skywinder/Github-Changelog-Generator)*
+\* *This Change Log was automatically generated by [github_changelog_generator](https://github.com/skywinder/Github-Changelog-Generator)*

README.md

@@ -141,6 +141,7 @@ protocol.
 ```elixir
 defimpl Elasticsearch.Document, for: MyApp.Post do
   def id(post), do: post.id
+  def routing(_), do: false
   def encode(post) do
     %{
       title: post.title,

guides/upgrading/0.4.x_to_0.5.x.md

@@ -0,0 +1,55 @@
+# Upgrading from 0.4.x to 0.5.x
+
+Version `0.5.0` added the `routing` meta-field to be specified in the Document
+protocol.
+
+## Rationale
+
+The `routing`/`_routing` meta-field used to force a document to be hashed to a
+particular shard which is required for `join` field datatypes but also used to
+gain more control over which shard your documents are routed to.
+
+## Changes
+
+**BREAKING**: `routing` function is now required to be specified in the
+`Elasticsearch.Document` protocol. You may specify it to return `false` to
+use default routing (document `id`).
+
+## How to Update Your App
+
+Add a `routing/1` function to your `Elasticsearch.Document` implementation.
+
+    # BEFORE
+    defimpl Elasticsearch.Document, for: MyApp.Post do
+      def id(post), do: post.id
+      def encode(post) do
+        %{
+          title: post.title,
+          author: post.author
+        }
+      end
+    end
+
+    # AFTER (using default routing)
+    defimpl Elasticsearch.Document, for: MyApp.Post do
+      def id(post), do: post.id
+      def routing(_), do: false
+      def encode(post) do
+        %{
+          title: post.title,
+          author: post.author
+        }
+      end
+    end
+
+    # AFTER (using routing)
+    defimpl Elasticsearch.Document, for: MyApp.Post do
+      def id(post), do: post.id
+      def routing(post), do: post.account_id
+      def encode(post) do
+        %{
+          title: post.title,
+          author: post.author
+        }
+      end
+    end

lib/elasticsearch.ex

@@ -124,7 +124,22 @@ defmodule Elasticsearch do
   end
 
   defp document_url(document, index) do
-    "/#{index}/_doc/#{Document.id(document)}"
+    url = "/#{index}/_doc/#{Document.id(document)}"
+
+    if routing = Document.routing(document) do
+      document_url_with_routing(url, routing)
+    else
+      url
+    end
+  end
+
+  defp document_url_with_routing(url, routing) do
+    url <>
+      if url =~ ~r/\?/ do
+        "&"
+      else
+        "?"
+      end <> URI.encode_query(%{routing: routing})
   end
 
   @doc """

lib/elasticsearch/indexing/bulk.ex

@@ -20,7 +20,7 @@ defmodule Elasticsearch.Index.Bulk do
       iex> Bulk.encode(Cluster, %Post{id: "my-id"}, "my-index")
       {:ok, \"\"\"
       {"create":{"_index":"my-index","_id":"my-id"}}
-      {"title":null,"author":null}
+      {"title":null,"doctype":{"name":"post"},"author":null}
       \"\"\"}
 
       iex> Bulk.encode(Cluster, 123, "my-index")
@@ -46,11 +46,11 @@ defmodule Elasticsearch.Index.Bulk do
       iex> Bulk.encode!(Cluster, %Post{id: "my-id"}, "my-index")
       \"\"\"
       {"create":{"_index":"my-index","_id":"my-id"}}
-      {"title":null,"author":null}
+      {"title":null,"doctype":{"name":"post"},"author":null}
       \"\"\"
 
       iex> Bulk.encode!(Cluster, 123, "my-index")
-      ** (Protocol.UndefinedError) protocol Elasticsearch.Document not implemented for 123. This protocol is implemented for: Post
+      ** (Protocol.UndefinedError) protocol Elasticsearch.Document not implemented for 123. This protocol is implemented for: Comment, Post
   """
   def encode!(cluster, struct, index) do
     config = Cluster.Config.get(cluster)
@@ -70,6 +70,13 @@ defmodule Elasticsearch.Index.Bulk do
       "_id" => Document.id(struct)
     }
 
+    attrs =
+      if routing = Document.routing(struct) do
+        Map.put(attrs, "_routing", routing)
+      else
+        attrs
+      end
+
     config.json_library.encode!(%{type => attrs})
   end
 

lib/elasticsearch/storage/document.ex

@@ -40,4 +40,25 @@ defprotocol Elasticsearch.Document do
   """
   @spec encode(any) :: map
   def encode(item)
+
+  @doc """
+  Returns the Elasticsearch `_routing` for the item. Elasticsearch
+  default if this value is not provided is to use the `_id`.
+  Setting this value to `false` or `nil` will omit sending the
+  meta-field with your requests and use default routing behaviour.
+  Routing allows you to control which shard the document should
+  be directed to which is necessary for `join` fields.
+
+  ## Example
+
+  Specify a routing key to control the destination shard, like so:
+
+      def routing(item), do: item.parent_id
+
+  or omit routing and use default Elasticsearch functionality:
+
+      def routing(_), do: false
+  """
+  @spec routing(any) :: any
+  def routing(item)
 end

mix.exs

@@ -6,7 +6,7 @@ defmodule Elasticsearch.Mixfile do
       app: :elasticsearch,
       description: "Elasticsearch without DSLs",
       source_url: "https://github.com/infinitered/elasticsearch-elixir",
-      version: "0.4.1",
+      version: "0.5.0",
       elixir: "~> 1.5",
       start_permanent: Mix.env() == :prod,
       elixirc_paths: elixirc_paths(Mix.env()),

test/elasticsearch/cluster/cluster_test.exs

@@ -12,7 +12,7 @@ defmodule Elasticsearch.ClusterTest do
         posts: %{
           settings: "test/support/settings/posts.json",
           store: Elasticsearch.Test.Store,
-          sources: [Post],
+          sources: [Post, Comment],
           bulk_page_size: 5000,
           bulk_wait_interval: 5000
         }

test/elasticsearch/indexing/bulk_test.exs

@@ -53,6 +53,7 @@ defmodule Elasticsearch.Index.BulkTest do
 
     test "respects bulk_* settings" do
       populate_posts_table(2)
+      populate_comments_table(2)
 
       Logger.configure(level: :debug)
 
@@ -77,4 +78,11 @@ defmodule Elasticsearch.Index.BulkTest do
       assert output =~ "Pausing 10ms between bulk pages"
     end
   end
+
+  describe ".encode!/3" do
+    test "respects _routing meta-field" do
+      assert Bulk.encode!(Cluster, %Comment{id: "my-id", post_id: "123"}, "my-index") =~
+               "\"_routing\":\"123\""
+    end
+  end
 end

test/elasticsearch_test.exs

@@ -18,4 +18,31 @@ defmodule ElasticsearchTest do
       Elasticsearch.delete(Cluster, "/nonexistent")
     end)
   end
+
+  describe ".put_document/3" do
+    test "routing meta-field is included if specified in Document" do
+      assert :ok =
+               Elasticsearch.Index.create_from_file(
+                 Cluster,
+                 "posts-routing",
+                 "test/support/settings/posts.json"
+               )
+
+      assert {:ok, _} =
+               Elasticsearch.put_document(
+                 Cluster,
+                 %Post{id: 1, title: "Example Post", author: "John Smith"},
+                 "posts-routing"
+               )
+
+      # If a routing key is not provided, this will throw an {:error, _}
+      #   Elasticsearch.Exception: [routing] is missing for join field [doctype]
+      assert {:ok, _} =
+               Elasticsearch.put_document(
+                 Cluster,
+                 %Comment{id: 2, body: "Example Comment", author: "Jane Smith", post_id: 1},
+                 "posts-routing"
+               )
+    end
+  end
 end

test/support/comment.ex

@@ -0,0 +1,29 @@
+defmodule Comment do
+  @moduledoc false
+
+  use Ecto.Schema
+
+  schema "comments" do
+    field(:body, :string)
+    field(:author, :string)
+    belongs_to(:post, Post)
+  end
+end
+
+defimpl Elasticsearch.Document, for: Comment do
+  def id(comment), do: comment.id
+  def type(_item), do: "comment"
+  def parent(_item), do: false
+  def routing(comment), do: comment.post_id
+
+  def encode(comment) do
+    %{
+      body: comment.body,
+      author: comment.author,
+      doctype: %{
+        name: "comment",
+        parent: comment.post_id
+      }
+    }
+  end
+end

test/support/data_case.ex

@@ -12,6 +12,7 @@ defmodule Elasticsearch.DataCase do
   # of the test unless the test case is marked as async.
 
   use ExUnit.CaseTemplate
+  import Ecto.Query
 
   using do
     quote do
@@ -44,4 +45,31 @@ defmodule Elasticsearch.DataCase do
 
     Elasticsearch.Test.Repo.insert_all("posts", posts)
   end
+
+  def random_post_id do
+    case Elasticsearch.Test.Repo.one(
+           from(
+             p in Post,
+             order_by: fragment("RANDOM()"),
+             limit: 1
+           )
+         ) do
+      nil -> nil
+      post -> post.id
+    end
+  end
+
+  def populate_comments_table(quantity \\ 10) do
+    comments =
+      0..quantity
+      |> Enum.map(fn _ ->
+        %{
+          body: "Example Comment",
+          author: "Jane Doe",
+          post_id: random_post_id()
+        }
+      end)
+
+    Elasticsearch.Test.Repo.insert_all("comments", comments)
+  end
 end

test/support/migrations/20180413213401_create_comments.exs

@@ -0,0 +1,11 @@
+defmodule Elasticsearch.Test.Repo.Migrations.CreateComments do
+  use Ecto.Migration
+
+  def change do
+    create table(:comments) do
+      add(:body, :string)
+      add(:author, :string)
+      add(:post_id, references(:posts, on_delete: :delete_all, on_update: :update_all))
+    end
+  end
+end

test/support/post.ex

@@ -13,11 +13,15 @@ defimpl Elasticsearch.Document, for: Post do
   def id(post), do: post.id
   def type(_item), do: "post"
   def parent(_item), do: false
+  def routing(_item), do: false
 
   def encode(post) do
     %{
       title: post.title,
-      author: post.author
+      author: post.author,
+      doctype: %{
+        name: "post"
+      }
     }
   end
 end

test/support/settings/posts.json

@@ -5,10 +5,19 @@
         "title": {
           "type": "text"
         },
+        "body": {
+          "type": "text"
+        },
         "author": {
           "type": "text"
+        },
+        "doctype": {
+          "type": "join",
+          "relations": {
+            "post": "comment"
+          }
         }
       }
     }
   }
-}
+}

test/support/store.ex

@@ -12,4 +12,12 @@ defmodule Elasticsearch.Test.Store do
     |> limit(^limit)
     |> Repo.all()
   end
+
+  def load(Comment, offset, limit) do
+    Comment
+    |> offset(^offset)
+    |> limit(^limit)
+    |> preload([:post])
+    |> Repo.all()
+  end
 end