From ff03a12bd2c27fb83453f69efddd9531a7bb9e2a Mon Sep 17 00:00:00 2001 From: Martin Karrer Date: Sun, 25 Oct 2026 13:42:47 +0100 Subject: [PATCH] Bound the repository query cache with :query_cache_max_size The query cache is an ETS set with no eviction policy. Its key is the query shape, and several constructs embed runtime values in the shape: string field names in field/2, interpolated JSON paths in json_extract_path/2, identifier/1, constant/1 and the :prefix option. Each distinct value therefore adds a new entry holding the planned query and the adapter prepared statement that is never released, so applications that let request data reach those constructs grow their memory without bound. Add a :query_cache_max_size repository option (default 10_000, :infinity to disable). The limit is only checked on cache misses and, once reached, the cache is cleared before inserting the new query. Document the per-value caching behaviour in field/2 and json_extract_path/2 and point to query_cache: false for user-controlled values. --- CHANGELOG.md | 1 + lib/ecto/query/api.ex | 11 ++- lib/ecto/query/planner.ex | 80 ++++++++++++++++++--- lib/ecto/repo.ex | 12 ++++ lib/ecto/repo/supervisor.ex | 2 +- test/ecto/query/planner_test.exs | 62 ++++++++++++++++- test/ecto/repo_test.exs | 115 ++++++++++++++++++------------- 7 files changed, 222 insertions(+), 61 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1eaa2a47c0..cdd79978f8 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/lib/ecto/query/api.ex b/lib/ecto/query/api.ex index 299144ee0a..de7f1b07d2 100644 --- a/lib/ecto/query/api.ex +++ b/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. diff --git a/lib/ecto/query/planner.ex b/lib/ecto/query/planner.ex index 7a831a0247..69fda0c84a 100644 --- a/lib/ecto/query/planner.ex +++ b/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) diff --git a/lib/ecto/repo.ex b/lib/ecto/repo.ex index e27067f023..650c85e8f8 100644 --- a/lib/ecto/repo.ex +++ b/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 diff --git a/lib/ecto/repo/supervisor.ex b/lib/ecto/repo/supervisor.ex index dd5552ceeb..3ec84e56f9 100644 --- a/lib/ecto/repo/supervisor.ex +++ b/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) diff --git a/test/ecto/query/planner_test.exs b/test/ecto/query/planner_test.exs index 9689da3983..8068f4e1c3 100644 --- a/test/ecto/query/planner_test.exs +++ b/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 diff --git a/test/ecto/repo_test.exs b/test/ecto/repo_test.exs index 6ce893eff5..4d0741fd71 100644 --- a/test/ecto/repo_test.exs +++ b/test/ecto/repo_test.exs @@ -625,7 +625,9 @@ defmodule Ecto.RepoTest do TestRepo.exists?(%{query | combinations: [{type, combination}]}) assert_received {:all, exists_query} - assert %{from: %{source: %Ecto.SubQuery{query: inner_query}}, combinations: []} = exists_query + assert %{from: %{source: %Ecto.SubQuery{query: inner_query}}, combinations: []} = + exists_query + assert %{select: %{expr: 1}, limit: %{expr: 1}, distinct: nil} = exists_query assert [{^type, inner_combination}] = inner_query.combinations assert inner_query.select.expr != 1 @@ -1178,9 +1180,13 @@ defmodule Ecto.RepoTest do ph_key = {:placeholder, :uuid_key} entries = [%{bid: ph_key, str: ph_key}] - assert_raise ArgumentError, ~r/a placeholder key can only be used with columns of the same type/, fn -> - TestRepo.insert_all(MySchemaWithBinaryId, entries, placeholders: placeholders) - end + assert_raise ArgumentError, + ~r/a placeholder key can only be used with columns of the same type/, + fn -> + TestRepo.insert_all(MySchemaWithBinaryId, entries, + placeholders: placeholders + ) + end end test "Repo.insert_all raises when placeholder key is used with invalid types" do @@ -2265,7 +2271,11 @@ defmodule Ecto.RepoTest do test "includes conflict target in :replace_all when replace_changed is false" do fields = [:map, :array, :z, :yyy, :x, :id] - TestRepo.insert(%MySchema{id: 1}, on_conflict: :replace_all, conflict_target: [:id], replace_changed: false) + TestRepo.insert(%MySchema{id: 1}, + on_conflict: :replace_all, + conflict_target: [:id], + replace_changed: false + ) assert_received {:insert, %{source: "my_schema", on_conflict: {^fields, [], [:id]}}} end @@ -2474,6 +2484,16 @@ defmodule Ecto.RepoTest do :ok end + test "configures the query cache size", config do + %{cache: cache} = Ecto.Repo.Registry.lookup(config.test) + assert [{_, 10_000}] = :ets.lookup(cache, :__query_cache_max_size__) + + name = :"#{config.test}_bounded" + {:ok, _pid} = TestRepo.start_link(name: name, query_cache_max_size: 42) + %{cache: cache} = Ecto.Repo.Registry.lookup(name) + assert [{_, 42}] = :ets.lookup(cache, :__query_cache_max_size__) + end + test "puts the dynamic repo in pdict" do assert is_pid(TestRepo.get_dynamic_repo()) @@ -2687,24 +2707,25 @@ defmodule Ecto.RepoTest do end test "update with on_writable_violation: :warn saves changes for writable: :always, ignores changes for writable: :insert/:never, and logs a warning" do - log = capture_log(fn -> - %{always: 10, never: nil, insert: nil} = - %MySchemaWritableWarn{id: 1} - |> Ecto.Changeset.change(%{always: 10, never: 11, insert: 12}) - |> TestRepo.update!() - - assert_received {:update, %{changes: [always: 10]}} - end) + log = + capture_log(fn -> + %{always: 10, never: nil, insert: nil} = + %MySchemaWritableWarn{id: 1} + |> Ecto.Changeset.change(%{always: 10, never: 11, insert: 12}) + |> TestRepo.update!() + + assert_received {:update, %{changes: [always: 10]}} + end) assert log =~ ~r""" - you are attempting to write to the field :insert of #{inspect(__MODULE__.MySchemaWritableWarn)} but - the `:writable` option of this field indicates the field should not be written to during an update. - """ + you are attempting to write to the field :insert of #{inspect(__MODULE__.MySchemaWritableWarn)} but + the `:writable` option of this field indicates the field should not be written to during an update. + """ assert log =~ ~r""" - you are attempting to write to the field :never of #{inspect(__MODULE__.MySchemaWritableWarn)} but - the `:writable` option of this field indicates the field should not be written to during an update. - """ + you are attempting to write to the field :never of #{inspect(__MODULE__.MySchemaWritableWarn)} but + the `:writable` option of this field indicates the field should not be written to during an update. + """ end test "update with on_writable_violation: :raise saves changes for writable: :always and raises for changes for writable: :insert/:never" do @@ -2816,40 +2837,42 @@ defmodule Ecto.RepoTest do end test "insert with surfaced changes and on_writable_violation: :warn saves changes for writable: :always/:insert, ignores changes for writable: :never, and logs a warning" do - log = capture_log(fn -> - # For surfaced changes from the underlying struct, the value in the returned struct is - # maintained even though the underlying write was not performed, as opposed to "normal" changes - # provided via a changeset. - %{always: 10, never: 11, insert: 12} = - %MySchemaWritableWarn{id: 1, always: 10, never: 11, insert: 12} - |> Ecto.Changeset.change(%{}) - |> TestRepo.insert!() - - assert_received {:insert, %{fields: inserted_fields}} - assert Enum.sort(inserted_fields) == [always: 10, id: 1, insert: 12] - end) + log = + capture_log(fn -> + # For surfaced changes from the underlying struct, the value in the returned struct is + # maintained even though the underlying write was not performed, as opposed to "normal" changes + # provided via a changeset. + %{always: 10, never: 11, insert: 12} = + %MySchemaWritableWarn{id: 1, always: 10, never: 11, insert: 12} + |> Ecto.Changeset.change(%{}) + |> TestRepo.insert!() + + assert_received {:insert, %{fields: inserted_fields}} + assert Enum.sort(inserted_fields) == [always: 10, id: 1, insert: 12] + end) assert log =~ ~r""" - you are attempting to write to the field :never of #{inspect(__MODULE__.MySchemaWritableWarn)} but - the `:writable` option of this field indicates the field should not be written to during an insert. - """ + you are attempting to write to the field :never of #{inspect(__MODULE__.MySchemaWritableWarn)} but + the `:writable` option of this field indicates the field should not be written to during an insert. + """ end test "insert with on_writable_violation: :warn saves changes for writable: :always/:insert, ignores changes for writable: :never, and logs a warning" do - log = capture_log(fn -> - %{always: 10, never: nil, insert: 12} = - %MySchemaWritableWarn{id: 1} - |> Ecto.Changeset.change(%{always: 10, never: 11, insert: 12}) - |> TestRepo.insert!() - - assert_received {:insert, %{fields: inserted_fields}} - assert Enum.sort(inserted_fields) == [always: 10, id: 1, insert: 12] - end) + log = + capture_log(fn -> + %{always: 10, never: nil, insert: 12} = + %MySchemaWritableWarn{id: 1} + |> Ecto.Changeset.change(%{always: 10, never: 11, insert: 12}) + |> TestRepo.insert!() + + assert_received {:insert, %{fields: inserted_fields}} + assert Enum.sort(inserted_fields) == [always: 10, id: 1, insert: 12] + end) assert log =~ ~r""" - you are attempting to write to the field :never of #{inspect(__MODULE__.MySchemaWritableWarn)} but - the `:writable` option of this field indicates the field should not be written to during an insert. - """ + you are attempting to write to the field :never of #{inspect(__MODULE__.MySchemaWritableWarn)} but + the `:writable` option of this field indicates the field should not be written to during an insert. + """ end test "insert with surfaced changes and on_writable_violation: :raise saves changes for writable: :always/:insert and raises for changes for writable: :never" do