Bound the repository query cache with :query_cache_max_size

CHANGELOG.md

@@ -7,6 +7,7 @@
   * [Ecto] Allow passing custom loader and dumper functions to `embedded_load/3` and `embedded_dump/2`
   * [Ecto.Changeset] Make `validate_length/3` count codepoints by default, aligning it with how databases typically enforce string length limits
   * [Ecto.Query] Allow assigning column names to fragment sources
+  * [Ecto.Repo] Bound the repository query cache via the `:query_cache_max_size` option, defaulting to 10 000 entries, and document that string field names, dynamic JSON paths and prefixes create one cached query per distinct value
 
 ### Bug fixes
 

lib/ecto/query/api.ex

@@ -734,7 +734,10 @@ defmodule Ecto.Query.API do
 
   String names can be particularly useful when your application is dynamically
   generating many schemaless queries at runtime and you want to avoid creating
-  a large number of atoms.
+  a large number of atoms. Note that each different field name will emit a
+  different query, which will be independently prepared and cached. If the
+  name comes from user input, consider restricting it to a known set of values
+  or passing `query_cache: false` to the repository operation.
   """
   def field(source, field), do: doc!([source, field])
 
@@ -879,6 +882,12 @@ defmodule Ecto.Query.API do
       source_field = :source_column
       from(post in Post, select: post.meta["author"][field(p, ^source_field)])
 
+  Note that dynamic path elements become part of the query itself rather
+  than query parameters, so each different path will emit a different query,
+  which will be independently prepared and cached. If the path comes from
+  user input, consider restricting it to a known set of values or passing
+  `query_cache: false` to the repository operation.
+
   ## Warning: indexes on PostgreSQL
 
   PostgreSQL supports indexing on jsonb columns via GIN indexes.

lib/ecto/query/planner.ex

@@ -18,6 +18,8 @@ defmodule Ecto.Query.Planner do
   end
 
   @parent_as __MODULE__
+  @default_query_cache_max_size 10_000
+  @query_cache_max_size_key :__query_cache_max_size__
   @aggs ~w(count avg min max sum row_number rank dense_rank percent_rank cume_dist ntile lag lead first_value last_value nth_value)a
 
   @doc """
@@ -122,9 +124,27 @@ defmodule Ecto.Query.Planner do
 
   @doc """
   Define the query cache table.
+
+  ## Options
+
+    * `:query_cache_max_size` - the maximum number of queries kept in
+      the cache. Once the limit is reached, the cache is cleared before
+      the next query is inserted. Can be a positive integer or `:infinity`.
+      Defaults to `#{@default_query_cache_max_size}`.
+
   """
-  def new_query_cache(atom_name) do
-    :ets.new(atom_name || __MODULE__, [:set, :public, read_concurrency: true])
+  def new_query_cache(atom_name, opts \\ []) do
+    max_size = Keyword.get(opts, :query_cache_max_size, @default_query_cache_max_size)
+
+    unless max_size == :infinity or (is_integer(max_size) and max_size > 0) do
+      raise ArgumentError,
+            ":query_cache_max_size must be a positive integer or :infinity, " <>
+              "got: #{inspect(max_size)}"
+    end
+
+    cache = :ets.new(atom_name || __MODULE__, [:set, :public, read_concurrency: true])
+    :ets.insert(cache, {@query_cache_max_size_key, max_size})
+    cache
   end
 
   @doc """
@@ -197,6 +217,8 @@ defmodule Ecto.Query.Planner do
   end
 
   defp cache_insert(cache, key, elem) do
+    maybe_clear_cache(cache)
+
     case :ets.insert_new(cache, elem) do
       true ->
         elem
@@ -207,6 +229,25 @@ defmodule Ecto.Query.Planner do
     end
   end
 
+  # The cache has no eviction policy, so we bound its size to avoid
+  # unbounded memory growth when runtime values become part of the
+  # query shape (for example, string field names, JSON paths or prefixes).
+  # The size check happens only on inserts, which are cache misses.
+  defp maybe_clear_cache(cache) do
+    case :ets.lookup(cache, @query_cache_max_size_key) do
+      [{_, max_size}] when is_integer(max_size) ->
+        # The size includes the entry holding the max size itself
+        if :ets.info(cache, :size) > max_size do
+          :ets.select_delete(cache, [{{:_, :_, :_, :_}, [], [true]}])
+        end
+
+        :ok
+
+      _ ->
+        :ok
+    end
+  end
+
   defp cache_update(cache, key, cached) do
     _ = :ets.update_element(cache, key, [{2, :cached}, {4, cached}])
     :ok
@@ -344,7 +385,12 @@ defmodule Ecto.Query.Planner do
        when kind in [:fragment, :values],
        do: {expr, source}
 
-  defp plan_source(_query, %{source: {{:fragment, _, _} = source, schema}, prefix: nil} = expr, _adapter, _cte_names)
+  defp plan_source(
+         _query,
+         %{source: {{:fragment, _, _} = source, schema}, prefix: nil} = expr,
+         _adapter,
+         _cte_names
+       )
        when is_atom(schema) do
     {expr, {source, schema, nil}}
   end
@@ -353,8 +399,14 @@ defmodule Ecto.Query.Planner do
        when kind in [:fragment, :values],
        do: error!(query, expr, "cannot set prefix: #{inspect(prefix)} option for #{kind} sources")
 
-  defp plan_source(query, %{source: {{:fragment, _, _}, _schema}, prefix: prefix} = expr, _adapter, _cte_names),
-       do: error!(query, expr, "cannot set prefix: #{inspect(prefix)} option for fragment sources")
+  defp plan_source(
+         query,
+         %{source: {{:fragment, _, _}, _schema}, prefix: prefix} = expr,
+         _adapter,
+         _cte_names
+       ),
+       do:
+         error!(query, expr, "cannot set prefix: #{inspect(prefix)} option for fragment sources")
 
   defp plan_subquery(subquery, query, prefix, adapter, source?, cte_names) do
     %{query: inner_query} = subquery
@@ -2275,7 +2327,7 @@ defmodule Ecto.Query.Planner do
         schema = if kind == :map, do: nil, else: schema
         {{:source, {source, schema}, prefix || query.prefix, types}, fields}
 
-      {{:ok, {kind, fields}}, {{:fragment, _, _} = source, schema, prefix}}  ->
+      {{:ok, {kind, fields}}, {{:fragment, _, _} = source, schema, prefix}} ->
         {types, fields} = select_dump_for_schema(schema, List.wrap(fields), ix, drop)
         schema = if kind == :map, do: nil, else: schema
         {{:source, {source, schema}, prefix, types}, fields}
@@ -2349,7 +2401,10 @@ defmodule Ecto.Query.Planner do
           nil
 
         {:struct, _} ->
-          error!(query, "it is not possible to return a struct subset of a subquery that does not return a schema struct")   
+          error!(
+            query,
+            "it is not possible to return a struct subset of a subquery that does not return a schema struct"
+          )
       end
 
     types =
@@ -2359,8 +2414,11 @@ defmodule Ecto.Query.Planner do
             {field, type}
 
           :error ->
-            error!(query, "field `#{field}` is not available in the subquery. " <>
-                         "Subquery only returns fields: #{inspect(available_fields)}")
+            error!(
+              query,
+              "field `#{field}` is not available in the subquery. " <>
+                "Subquery only returns fields: #{inspect(available_fields)}"
+            )
         end
       end)
 
@@ -2623,7 +2681,9 @@ defmodule Ecto.Query.Planner do
     input_string = Atom.to_string(input)
 
     schema.__schema__(:fields)
-    |> Enum.map(fn field -> {field, String.jaro_distance(input_string, Atom.to_string(field))} end)
+    |> Enum.map(fn field ->
+      {field, String.jaro_distance(input_string, Atom.to_string(field))}
+    end)
     |> Enum.filter(fn {_field, score} -> score >= 0.77 end)
     |> Enum.sort(&(elem(&1, 0) >= elem(&2, 0)))
     |> Enum.take(5)

lib/ecto/repo.ex

@@ -47,6 +47,18 @@ defmodule Ecto.Repo do
     * `:pool_size` - the size of the pool used by the connection module.
       Defaults to `10`
 
+    * `:query_cache_max_size` - the maximum number of planned queries and
+      adapter prepared statements kept in the repository query cache. The
+      cache is keyed by the shape of the query, so constructs that embed
+      runtime values in the query itself (such as `field/2` with string
+      names, `json_extract_path/2` with interpolated paths, `identifier/1`,
+      `constant/1` or the `:prefix` option) create one cache entry per
+      distinct value. Once the limit is reached, the cache is cleared and
+      queries are prepared again as they are executed. Can be a positive
+      integer or `:infinity`. Defaults to `10_000`. Queries built from
+      user-controlled values should also consider passing `query_cache: false`
+      (see "Shared options" below)
+
     * `:pool_count` - the number of pools to run concurrently,
       increase this option when the pool itself may be under contention.
       When running multiple pools, queries are randomly routed to different

lib/ecto/repo/supervisor.ex

@@ -205,7 +205,7 @@ defmodule Ecto.Repo.Supervisor do
 
         # Normalize name to atom, ignore via/global names
         name = if is_atom(name), do: name, else: nil
-        cache = Ecto.Query.Planner.new_query_cache(name)
+        cache = Ecto.Query.Planner.new_query_cache(name, opts)
         meta = Map.merge(meta, %{repo: repo, cache: cache})
         child_spec = wrap_child_spec(child, [name, adapter, meta])
         Supervisor.init([child_spec], strategy: :one_for_one, max_restarts: 0)

test/ecto/query/planner_test.exs

@@ -2566,7 +2566,7 @@ defmodule Ecto.Query.PlannerTest do
   end
 
   test "normalize: select source on fragment with columns" do
-    query = from f in fragment("select 1", columns: [:x])
+    query = from(f in fragment("select 1", columns: [:x]))
     {_, _, _, select} = normalize_with_params(query)
     assert %{from: {_, {:map, [x: {:value, :any}]}}} = select
   end
@@ -3214,7 +3214,7 @@ defmodule Ecto.Query.PlannerTest do
       {_meta, {:cache, _update, _prepared}, _cast, _dump} =
         Planner.query(query, :all, cache, Ecto.CachingTestAdapter, 0, true)
 
-      assert :ets.info(cache, :size) == 1
+      assert cached_queries(cache) == 1
     end
 
     test "bypasses cache if false", %{cache: cache} do
@@ -3223,7 +3223,63 @@ defmodule Ecto.Query.PlannerTest do
       {_meta1, {:nocache, _prepared}, _cast1, _dump1} =
         Planner.query(query, :all, cache, Ecto.CachingTestAdapter, 0, false)
 
-      assert :ets.info(cache, :size) == 0
+      assert cached_queries(cache) == 0
+    end
+  end
+
+  defp cached_queries(cache) do
+    :ets.select_count(cache, [{{:_, :_, :_, :_}, [], [true]}])
+  end
+
+  describe "query: query_cache_max_size option" do
+    defp query_field(cache, name) do
+      query = from(p in Post, where: field(p, ^name) == ^"hello")
+
+      {_meta, {:cache, _update, _prepared}, _cast, _dump} =
+        Planner.query(query, :all, cache, Ecto.CachingTestAdapter, 0, true)
+
+      :ok
+    end
+
+    test "defaults to a bounded cache" do
+      cache = Planner.new_query_cache(nil)
+      assert [{_, 10_000}] = :ets.lookup(cache, :__query_cache_max_size__)
+    end
+
+    test "raises on invalid values" do
+      assert_raise ArgumentError, ~r":query_cache_max_size must be a positive integer", fn ->
+        Planner.new_query_cache(nil, query_cache_max_size: 0)
+      end
+
+      assert_raise ArgumentError, ~r":query_cache_max_size must be a positive integer", fn ->
+        Planner.new_query_cache(nil, query_cache_max_size: "10")
+      end
+    end
+
+    test "clears the cache once the limit is reached" do
+      cache = Planner.new_query_cache(nil, query_cache_max_size: 3)
+
+      for i <- 1..3, do: query_field(cache, "field#{i}")
+      assert cached_queries(cache) == 3
+
+      # Cached queries are still served from the cache
+      query_field(cache, "field1")
+      assert cached_queries(cache) == 3
+
+      # Inserting past the limit clears the cache first
+      query_field(cache, "field4")
+      assert cached_queries(cache) == 1
+      assert [{_, 3}] = :ets.lookup(cache, :__query_cache_max_size__)
+
+      for i <- 5..6, do: query_field(cache, "field#{i}")
+      assert cached_queries(cache) == 3
+    end
+
+    test "does not bound the cache with :infinity" do
+      cache = Planner.new_query_cache(nil, query_cache_max_size: :infinity)
+
+      for i <- 1..50, do: query_field(cache, "field#{i}")
+      assert cached_queries(cache) == 50
     end
   end
 end