Support JSON extract path expression

lib/etso/adapter/behaviour/queryable.ex

@@ -7,21 +7,55 @@ defmodule Etso.Adapter.Behaviour.Queryable do
   alias Etso.ETS.ObjectsSorter
 
   @impl Ecto.Adapter.Queryable
-  def prepare(:all, query) do
-    {:nocache, query}
+  def prepare(:all, %Ecto.Query{} = query) do
+    {:nocache, {:select, query}}
   end
 
   @impl Ecto.Adapter.Queryable
-  def execute(%{repo: repo}, _, {:nocache, query}, params, _) do
+  def prepare(:delete_all, %Ecto.Query{wheres: []} = query) do
+    {:nocache, {:delete_all_objects, query}}
+  end
+
+  @impl Ecto.Adapter.Queryable
+  def prepare(:delete_all, %Ecto.Query{wheres: _} = query) do
+    {:nocache, {:match_delete, query}}
+  end
+
+  @impl Ecto.Adapter.Queryable
+  def execute(%{repo: repo}, _, {:nocache, {:select, query}}, params, _) do
+    {_, schema} = query.from.source
+    {:ok, ets_table} = TableRegistry.get_table(repo, schema)
+    ets_match = MatchSpecification.build(query, params)
+    ets_objects = :ets.select(ets_table, [ets_match])
+    ets_count = length(ets_objects)
+    {ets_count, ObjectsSorter.sort(ets_objects, query)}
+  end
+
+  @impl Ecto.Adapter.Queryable
+  def execute(%{repo: repo}, _, {:nocache, {:delete_all_objects, query}}, params, _) do
+    {_, schema} = query.from.source
+    {:ok, ets_table} = TableRegistry.get_table(repo, schema)
+    ets_match = MatchSpecification.build(query, params)
+    ets_objects = query.select && ObjectsSorter.sort(:ets.select(ets_table, [ets_match]), query)
+    ets_count = :ets.info(ets_table, :size)
+    true = :ets.delete_all_objects(ets_table)
+    {ets_count, ets_objects || nil}
+  end
+
+  @impl Ecto.Adapter.Queryable
+  def execute(%{repo: repo}, _, {:nocache, {:match_delete, query}}, params, _) do
     {_, schema} = query.from.source
     {:ok, ets_table} = TableRegistry.get_table(repo, schema)
     ets_match = MatchSpecification.build(query, params)
-    ets_objects = :ets.select(ets_table, [ets_match]) |> ObjectsSorter.sort(query)
-    {length(ets_objects), ets_objects}
+    ets_objects = query.select && ObjectsSorter.sort(:ets.select(ets_table, [ets_match]), query)
+    {ets_match_head, ets_match_body, _} = ets_match
+    ets_match = {ets_match_head, ets_match_body, [true]}
+    ets_count = :ets.select_delete(ets_table, [ets_match])
+    {ets_count, ets_objects || nil}
   end
 
   @impl Ecto.Adapter.Queryable
-  def stream(%{repo: repo}, _, {:nocache, query}, params, options) do
+  def stream(%{repo: repo}, _, {:nocache, {:select, query}}, params, options) do
     {_, schema} = query.from.source
     {:ok, ets_table} = TableRegistry.get_table(repo, schema)
     ets_match = MatchSpecification.build(query, params)

lib/etso/adapter/table_registry.ex

@@ -38,6 +38,15 @@ defmodule Etso.Adapter.TableRegistry do
     end
   end
 
+  @spec active_tables(module()) :: [any()]
+  def active_tables(repo) do
+    conditions = [{{{:_, :"$1"}, :_, :_}, [{:==, :"$1", :ets_table}], [:"$_"]}]
+
+    build_name(repo)
+    |> Registry.select(conditions)
+    |> Enum.map(fn {_key, {_pid, table_reference}} -> table_reference end)
+  end
+
   defp lookup_table(repo, schema) do
     case Registry.lookup(build_name(repo), {schema, :ets_table}) do
       [{_, table_reference}] -> {:ok, table_reference}

lib/etso/ets/match_specification.ex

@@ -1,25 +1,31 @@
 defmodule Etso.ETS.MatchSpecification do
   @moduledoc """
   The ETS Match Specifications module contains various functions which convert Ecto queries to
-  ETS Match Specifications in order to execute the given queries.
+  [ETS Match Specifications](https://www.erlang.org/doc/apps/erts/match_spec.html) in order to
+  execute the given queries with ETS with as much pushed down to ETS as possible.
+
+  The basic shape of the match head is `[$1, $2, $3, …]` where each field is a named variable, the
+  ordering of the fields is determined by `Etso.ETS.TableStructure`.
+
+  Conditions are compiled according to the wheres in the underlying Ecto query, while the body is
+  compiled based on the selected fields in the underlying Ecto query.
   """
 
   def build(query, params) do
     {_, schema} = query.from.source
     field_names = Etso.ETS.TableStructure.field_names(schema)
-
     match_head = build_head(field_names)
-    match_conditions = build_conditions(field_names, params, query.wheres)
-    match_body = [build_body(field_names, query.select.fields)]
+    match_conditions = build_conditions(field_names, params, query)
+    match_body = [build_body(field_names, query)]
     {match_head, match_conditions, match_body}
   end
 
   defp build_head(field_names) do
     List.to_tuple(Enum.map(1..length(field_names), fn x -> :"$#{x}" end))
   end
 
-  defp build_conditions(field_names, params, query_wheres) do
-    Enum.reduce(query_wheres, [], fn %Ecto.Query.BooleanExpr{expr: expression}, acc ->
+  defp build_conditions(field_names, params, %Ecto.Query{wheres: wheres}) do
+    Enum.reduce(wheres, [], fn %Ecto.Query.BooleanExpr{expr: expression}, acc ->
       [build_condition(field_names, params, expression) | acc]
     end)
   end
@@ -45,56 +51,87 @@ defmodule Etso.ETS.MatchSpecification do
     end
   end
 
-  defp build_condition(field_names, params, {:in, [], [field, value]}) do
-    field_name = resolve_field_name(field)
-    field_index = get_field_index(field_names, field_name)
+  defp build_condition(field_names, params, {:in, [], [field, values]}) do
+    field_target = resolve_field_target(field_names, field)
 
-    case resolve_field_values(params, value) do
+    case resolve_param_values(params, values) do
       [] -> []
-      values -> List.to_tuple([:orelse | Enum.map(values, &{:==, :"$#{field_index}", &1})])
+      values -> List.to_tuple([:orelse | Enum.map(values, &{:==, field_target, &1})])
     end
   end
 
   defp build_condition(field_names, params, {:is_nil, [], [field]}) do
     {:==, build_condition(field_names, params, field), nil}
   end
 
-  defp build_condition(field_names, _, {{:., [], [{:&, [], [0]}, field_name]}, [], []}) do
-    :"$#{get_field_index(field_names, field_name)}"
-  end
-
   defp build_condition(_, params, {:^, [], [index]}) do
     Enum.at(params, index)
   end
 
-  defp build_condition(_, _, value) when not is_tuple(value) do
+  defp build_condition(field_names, _params, field) when is_tuple(field) do
+    resolve_field_target(field_names, field)
+  end
+
+  defp build_condition(_, _, value) do
     value
   end
 
-  defp build_body(field_names, query_select_fields) do
-    for select_field <- query_select_fields do
-      field_name = resolve_field_name(select_field)
-      field_index = get_field_index(field_names, field_name)
-      :"$#{field_index}"
+  defp build_body(_, %Ecto.Query{select: nil}) do
+    []
+  end
+
+  defp build_body(field_names, %Ecto.Query{select: %{fields: fields}}) do
+    for field <- fields do
+      resolve_field_target(field_names, field)
     end
   end
 
-  defp resolve_field_name(field) do
-    {{:., _, [{:&, [], [0]}, field_name]}, [], []} = field
-    field_name
+  defp resolve_field_target(field_names, {:json_extract_path, [], [field, path]}) do
+    field_target = resolve_field_target(field_names, field)
+    resolve_field_target_path(field_target, path)
   end
 
-  defp resolve_field_values(params, {:^, [], [index, count]}) do
+  defp resolve_field_target(field_names, {{:., _, [{:&, [], [0]}, field_name]}, [], []}) do
+    field_index = 1 + Enum.find_index(field_names, fn x -> x == field_name end)
+    :"$#{field_index}"
+  end
+
+  defp resolve_field_target_path(field_target, path) do
+    # - If the path component is a key, return {:map_get, key, target}
+    # - If the path component is a number, return {:hd, target} outside as many {:tl, _} around
+    #   as required. For example, [:metadata, 0] would be {:hd, {:map_get, :metadata, field}},
+    #   while [:metadata, 1] would be {:hd, {:tl, {:map_get, :metadata, field}}} (with one tl).
+
+    at = fn self ->
+      fn
+        condition, 0 -> {:hd, condition}
+        condition, index -> self.(self).({:tl, condition}, index - 1)
+      end
+    end
+
+    Enum.reduce(path, field_target, fn
+      key, condition when is_atom(key) or is_binary(key) -> {:map_get, key, condition}
+      index, condition when is_integer(index) -> at.(at).(condition, index)
+    end)
+  end
+
+  defp resolve_param_values(params, {:^, [], [index, count]}) do
     for index <- index..(index + count - 1) do
       Enum.at(params, index)
     end
   end
 
-  defp resolve_field_values(params, {:^, [], [index]}) do
+  defp resolve_param_values(params, {:^, [], [index]}) do
     Enum.at(params, index)
   end
 
-  defp get_field_index(field_names, field_name) do
-    1 + Enum.find_index(field_names, fn x -> x == field_name end)
+  defp resolve_param_values(_params, values) when is_list(values) do
+    values
+  end
+
+  def list_spec(0, acc), do: {:hd, acc}
+
+  def list_spec(index, acc) do
+    list_spec(index - 1, {:tl, acc})
   end
 end

mix.lock

@@ -2,7 +2,7 @@
   "decimal": {:hex, :decimal, "2.0.0", "a78296e617b0f5dd4c6caf57c714431347912ffb1d0842e998e9792b5642d697", [:mix], [], "hexpm", "34666e9c55dea81013e77d9d87370fe6cb6291d1ef32f46a1600230b1d44f577"},
   "dialyxir": {:hex, :dialyxir, "1.0.0-rc.6", "78e97d9c0ff1b5521dd68041193891aebebce52fc3b93463c0a6806874557d7d", [:mix], [{:erlex, "~> 0.2.1", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "49496d63267bc1a4614ffd5f67c45d9fc3ea62701a6797975bc98bc156d2763f"},
   "earmark": {:hex, :earmark, "1.3.6", "ce1d0675e10a5bb46b007549362bd3f5f08908843957687d8484fe7f37466b19", [:mix], [], "hexpm", "1476378df80982302d5a7857b6a11dd0230865057dec6d16544afecc6bc6b4c2"},
-  "ecto": {:hex, :ecto, "3.8.3", "5e681d35bc2cbb46dcca1e2675837c7d666316e5ada14eca6c9c609b6232817c", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "af92dd7815967bcaea0daaaccf31c3b23165432b1c7a475d84144efbc703d105"},
+  "ecto": {:hex, :ecto, "3.8.4", "e06b8b87e62b27fea17fd2ff6041572ddd10339fd16cdf58446e402c6c90a74b", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f9244288b8d42db40515463a008cf3f4e0e564bb9c249fe87bf28a6d79fe82d4"},
   "erlex": {:hex, :erlex, "0.2.4", "23791959df45fe8f01f388c6f7eb733cc361668cbeedd801bf491c55a029917b", [:mix], [], "hexpm", "4a12ebc7cd8f24f2d0fce93d279fa34eb5068e0e885bb841d558c4d83c52c439"},
   "ex_doc": {:hex, :ex_doc, "0.21.2", "caca5bc28ed7b3bdc0b662f8afe2bee1eedb5c3cf7b322feeeb7c6ebbde089d6", [:mix], [{:earmark, "~> 1.3.3 or ~> 1.4", [hex: :earmark, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm", "f1155337ae17ff7a1255217b4c1ceefcd1860b7ceb1a1874031e7a861b052e39"},
   "jason": {:hex, :jason, "1.3.0", "fa6b82a934feb176263ad2df0dbd91bf633d4a46ebfdffea0c8ae82953714946", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "53fc1f51255390e0ec7e50f9cb41e751c260d065dcba2bf0d08dc51a4002c2ac"},

priv/northwind/employees.json

@@ -25,7 +25,23 @@
       1581
     ],
     "title": "Vice President Sales",
-    "titleOfCourtesy": "Dr."
+    "titleOfCourtesy": "Dr.",
+    "metadata": {
+      "twitter": "@andrew_fuller",
+      "photos": [
+        {
+          "storage": "a",
+          "url": "https://example.com/a"
+        },
+        {
+          "storage": "b",
+          "url": "https://example.com/b"
+        }
+      ],
+      "documents": {
+        "passport": "verified"
+      }
+    }
   },
   {
     "address": {
@@ -237,4 +253,4 @@
     "title": "Sales Representative",
     "titleOfCourtesy": "Ms."
   }
-]
+]

test/northwind/repo_test.exs

@@ -10,7 +10,7 @@ defmodule Northwind.RepoTest do
     :ok = Importer.perform()
   end
 
-  test "list" do
+  test "List All" do
     Repo.all(Model.Employee)
   end
 
@@ -174,4 +174,100 @@ defmodule Northwind.RepoTest do
 
     assert sorted_etso == sorted_code
   end
+
+  test "Delete All" do
+    assert Repo.delete_all(Model.Employee)
+    assert [] == Repo.all(Model.Employee)
+  end
+
+  test "Delete Where" do
+    query = Model.Employee |> where([e], e.employee_id in [1, 5])
+    assert [a, b] = Repo.all(query)
+    assert {2, nil} = Repo.delete_all(query)
+    assert [] == Repo.all(query)
+    refute [] == Repo.all(Model.Employee)
+  end
+
+  test "Delete Where Select" do
+    query = Model.Employee |> where([e], e.employee_id in [1, 5])
+    assert [a, b] = Repo.all(query)
+    assert {2, list} = Repo.delete_all(query |> select([e], {e, e.employee_id}))
+    assert is_list(list)
+    assert Enum.any?(list, &(elem(&1, 1) == 1))
+    assert Enum.any?(list, &(elem(&1, 1) == 5))
+    assert [] = Repo.all(query)
+    refute [] == Repo.all(Model.Employee)
+  end
+
+  describe "With JSON Extract Paths" do
+    test "using literal value" do
+      Model.Employee
+      |> where([e], e.metadata["twitter"] == "@andrew_fuller")
+      |> Repo.one!()
+    end
+
+    test "using brackets" do
+      Model.Employee
+      |> where([e], e.metadata["documents"]["passport"] == "verified")
+      |> Repo.one!()
+    end
+
+    test "with variable pinning" do
+      field = "passport"
+
+      Model.Employee
+      |> where([e], e.metadata["documents"][^field] == "verified")
+      |> Repo.one!()
+
+      Model.Employee
+      |> select([e], json_extract_path(e.metadata, ["documents", "passport"]))
+      |> Repo.all()
+      |> Enum.any?(&(&1 == "verified"))
+      |> assert()
+    end
+
+    test "with arrays" do
+      Model.Employee
+      |> select([e], json_extract_path(e.metadata, ["photos", 0, "url"]))
+      |> where([e], e.metadata["documents"]["passport"] == "verified")
+      |> Repo.one!()
+      |> (&(&1 == "https://example.com/a")).()
+      |> assert()
+
+      Model.Employee
+      |> where([e], e.metadata["documents"]["passport"] == "verified")
+      |> select([e], e.metadata["photos"][0]["url"])
+      |> Repo.one!()
+      |> (&(&1 == "https://example.com/a")).()
+      |> assert()
+
+      Model.Employee
+      |> select([e], e.metadata["photos"][1]["url"])
+      |> where([e], e.metadata["documents"]["passport"] == "verified")
+      |> Repo.one!()
+      |> (&(&1 == "https://example.com/b")).()
+      |> assert()
+    end
+
+    test "with where/in" do
+      Model.Employee
+      |> where([e], e.metadata["documents"]["passport"] in ~w(verified))
+      |> select([e], e.metadata["photos"][1]["url"])
+      |> Repo.one!()
+      |> (&(&1 == "https://example.com/b")).()
+      |> assert()
+    end
+
+    test "in deletion" do
+      Model.Employee
+      |> where([e], e.metadata["documents"]["passport"] == "verified")
+      |> Repo.delete_all()
+
+      assert_raise Ecto.NoResultsError, fn ->
+        Model.Employee
+        |> where([e], e.metadata["documents"]["passport"] == "verified")
+        |> Repo.one!()
+      end
+    end
+  end
 end

test/support/northwind/model/employee.ex

@@ -13,6 +13,7 @@ defmodule Northwind.Model.Employee do
     field :hire_date, :date
     field :notes, :string
     field :territory_ids, {:array, :integer}
+    field :metadata, :map, default: %{}
 
     embeds_one :address, Model.Address