Aggregate only the related items of the items of a page

guides/fields/computed-fields.md

@@ -66,17 +66,15 @@ e.g. `expr(author.name)`. `Backpex.Adapters.Ecto` joins a subquery of the values
 duplicate items. Aggregates that only differ in their `:value`, e.g. the title and the likes of the top post, share
 one subquery. The index view sorts by aggregates and searches them like any other computed value.
 
-The subquery of an aggregate is not limited to the items of the page or of a filter: on every read, it aggregates the
-related items of all items. If that is too slow for a very large table of related items, a native value (see below)
-can select a correlated subquery instead, which only reads the related items of one item:
-
-```elixir
-post_count:
-  Backpex.Expr.native(
-    Backpex.Adapters.Ecto,
-    dynamic([user: u], fragment("(SELECT count(*) FROM posts WHERE posts.user_id = ?)", u.id))
-  )
-```
+With a limit, e.g. for a page of the index view or a single item, the subqueries only aggregate the related items of
+the items the query returns: the adapter first finds the primary keys of these items (the common table expression
+`backpex_page`), so the cost of an aggregate grows with the page, not with the table. An index on the foreign key of
+the relationship, e.g. on `posts.user_id`, lets the database find the related items of a page.
+
+An aggregate that decides which items the query returns, as the filter, the search or the sort reads it, is aggregated
+over the related items of all items first. So is every aggregate of a query without a limit, e.g. an export, and of a
+metric. On a very large table of related items, sorting or filtering by an aggregate takes as long as aggregating the
+whole table.
 
 ## Values of the database
 

guides/upgrading/v0.25.md

@@ -373,9 +373,9 @@ LiveResource that return expressions (see `Backpex.Expr`) and work with every ad
 | joins for a `:select` or under a `:custom_alias` | paths in expressions, e.g. `expr(author.last_name)` |
 | a SQL fragment | `Backpex.Expr.native/3`, with `uses: [:author]` for the joins of paths |
 
-The subquery of `Backpex.Expr.aggregate/3` aggregates the related items of all items on every read, not only the ones
-of the page. A correlated subquery in a `:select`, which only reads the related items of one item, can stay a SQL
-fragment in `Backpex.Expr.native/3`, see the [computed fields](computed-fields.md) guide.
+Like a correlated subquery in a `:select`, the subquery of `Backpex.Expr.aggregate/3` only reads the related items of
+the items of a page. Sorting or filtering by it aggregates the related items of all items, see the
+[computed fields](computed-fields.md) guide.
 
 ```elixir
 # Before

integration_test/options/computed_test.exs

@@ -135,6 +135,31 @@ defmodule Backpex.Integration.Options.ComputedTest do
     assert Resource.count(query) == {:ok, 18}
   end
 
+  test "a page reads the same aggregates as all items", %{bob: bob} do
+    insert(:post, user: bob, title: "Only", likes: 5, published: true, tags: [build(:tag, name: "ash")])
+    carl = insert(:user, username: "carl", posts: [])
+    insert(:post, user: carl, title: "Top", likes: 7, tags: [build(:tag, name: "beam"), build(:tag, name: "otp")])
+
+    stats = fn users ->
+      Enum.map(users, &Map.take(&1, [:username | Keyword.keys(UserStatsLive.computed(%{}))]))
+    end
+
+    for sort <- [[username: :asc], [total_likes: :desc, username: :asc], [top_post: :asc]],
+        filter <- [nil, expr(^Expr.aggregate(:count, :posts) > 0), expr(username != "ann")] do
+      query = [] |> Resource.items_query(%{}, UserStatsLive) |> Query.filter(filter) |> Query.sort(sort)
+      {:ok, all} = Resource.read(query)
+
+      pages =
+        Enum.flat_map(0..2, fn page ->
+          {:ok, users} = query |> Query.offset(page) |> Query.limit(1) |> Resource.read()
+          users
+        end)
+
+      assert stats.(pages) == stats.(all)
+      assert {:ok, ^pages} = query |> Query.limit(10) |> Resource.read()
+    end
+  end
+
   test "computes a correlated subquery of the related items as a native value" do
     post_count =
       Expr.native(

lib/backpex/adapters/ecto.ex

@@ -94,6 +94,9 @@ defmodule Backpex.Adapters.Ecto do
   alias Backpex.Expr.Ref
   alias Backpex.Query
 
+  # The name of the common table expression of the primary keys of the items of a page, see `apply_page/3`.
+  @page "backpex_page"
+
   @doc false
   def default_changeset(item, attrs, _metadata), do: Ecto.Changeset.cast(item, attrs, [])
 
@@ -542,10 +545,17 @@ defmodule Backpex.Adapters.Ecto do
     select_merge(ecto_query, ^Map.new(values))
   end
 
-  # The expressions of a query that may read aggregates, which `Backpex.Adapters.Ecto.Where.join_aggregates/2` joins
+  # The expressions of a query that may read aggregates, which `Backpex.Adapters.Ecto.Where.join_aggregates/3` joins
   # before any of them is built, so that all aggregates of a group share a join: the computed values, the filter, the
   # values of related items that sorts and the search refer to, and the conditions of the search.
-  defp aggregate_expressions(%Query{computed: computed, filter: filter, search: search} = query) do
+  defp aggregate_expressions(%Query{computed: computed} = query) do
+    [filter | related] = condition_expressions(query)
+
+    [filter | Map.values(computed) ++ related]
+  end
+
+  # The expressions of `aggregate_expressions/1` that decide which items a query returns and in which order.
+  defp condition_expressions(%Query{filter: filter, search: search} = query) do
     sort_targets = query |> Query.sorts() |> Enum.map(fn {ref, _direction} -> ref end)
     search_targets = if search, do: search.targets, else: []
 
@@ -555,7 +565,7 @@ defmodule Backpex.Adapters.Ecto do
         condition -> [condition]
       end)
 
-    [filter | Map.values(computed) ++ related]
+    [filter | related]
   end
 
   # --- QUERIES
@@ -597,15 +607,74 @@ defmodule Backpex.Adapters.Ecto do
   end
 
   defp apply_query(ecto_query, %Query{} = query, schema, expressions) do
-    {ecto_query, joined} = Where.join_aggregates(ecto_query, aggregate_expressions(query) ++ expressions)
+    # Further expressions read their aggregates for all items of the query.
+    if expressions == [] and page_aggregates?(ecto_query, query, schema) do
+      apply_page(ecto_query, query, schema)
+    else
+      {ecto_query, joined} = Where.join_aggregates(ecto_query, aggregate_expressions(query) ++ expressions)
+
+      ecto_query =
+        ecto_query
+        |> merge_computed(query, joined)
+        |> apply_conditions(query, schema, joined)
+
+      {ecto_query, joined}
+    end
+  end
+
+  defp apply_conditions(ecto_query, %Query{} = query, schema, joined) do
+    ecto_query
+    |> Where.where_expr(query.filter, joined)
+    |> apply_search(query, schema, joined)
+    |> apply_sorts(query, schema, joined)
+    |> apply_offset_limit(query)
+  end
+
+  # Whether the query selects aggregates for a limited number of items, which `apply_page/3` then aggregates the
+  # related items of. The query of a page finds its items by their primary key, so it needs a query that returns every
+  # item once, e.g. not an options query with a join.
+  defp page_aggregates?(%Ecto.Query{} = ecto_query, %Query{limit: limit, computed: computed}, schema) do
+    is_integer(limit) and is_atom(schema) and match?([_primary_key], schema.__schema__(:primary_key)) and
+      match?(
+        %Ecto.Query{
+          joins: [],
+          select: nil,
+          distinct: nil,
+          group_bys: [],
+          havings: [],
+          combinations: [],
+          limit: nil,
+          offset: nil,
+          with_ctes: nil
+        },
+        ecto_query
+      ) and Enum.any?(computed, fn {_name, expression} -> Where.aggregates?(expression) end)
+  end
+
+  # The items of a page with their aggregates. The common table expression `backpex_page` finds the primary keys of the
+  # items of the page: it applies the filter, the search, the sorts, the offset and the limit, and joins only the
+  # aggregates these read, over all items. The query then selects the items of these keys in the order of the sorts,
+  # and the aggregates it selects only aggregate the related items of these keys. So the cost of an aggregate that is
+  # only selected grows with the page, not with the table.
+  defp apply_page(ecto_query, %Query{computed: computed} = query, schema) do
+    primary_key = get_primary_key_field(schema)
+
+    {keys, joined} = ecto_query |> exclude(:preload) |> Where.join_aggregates(condition_expressions(query))
+
+    keys =
+      keys
+      |> apply_conditions(query, schema, joined)
+      |> select([item], %{key: field(item, ^primary_key)})
+
+    page = from(page in @page, select: page.key)
+    {ecto_query, joined} = Where.join_aggregates(ecto_query, Map.values(computed), of: page)
 
     ecto_query =
       ecto_query
+      |> with_cte(@page, as: ^keys)
+      |> where([item], field(item, ^primary_key) in subquery(page))
       |> merge_computed(query, joined)
-      |> Where.where_expr(query.filter, joined)
-      |> apply_search(query, schema, joined)
       |> apply_sorts(query, schema, joined)
-      |> apply_offset_limit(query)
 
     {ecto_query, joined}
   end

lib/backpex/adapters/ecto/where.ex

@@ -6,9 +6,11 @@ defmodule Backpex.Adapters.Ecto.Where do
   # joins, named after the path, so every expression that uses the same path shares the join. `exists` becomes a
   # subquery of the primary keys of the items with a matching related item, so it never duplicates items.
   #
-  # An aggregate is a left join of a subquery of the values of the related items. `join_aggregates/2` joins the
+  # An aggregate is a left join of a subquery of the values of the related items. `join_aggregates/3` joins the
   # aggregates of expressions and returns the joined aggregates: a map of `{binding, aggregate}` to the name of the join
-  # and the key of the value in it. `where_expr/3` and `value/3` take this map and join the aggregates it lacks.
+  # and the key of the value in it. `where_expr/3` and `value/3` take this map and join the aggregates it lacks. The
+  # subquery aggregates the related items of all items, or with `:of` only the ones of the items with the given
+  # primary keys, e.g. the items of a page.
 
   import Ecto.Query
 
@@ -38,7 +40,7 @@ defmodule Backpex.Adapters.Ecto.Where do
 
   @doc """
   Adds the expression to the query as a `where` clause. Keeps the query for a `nil` expression. `aggregates` are the
-  joined aggregates of `join_aggregates/2`.
+  joined aggregates of `join_aggregates/3`.
   """
   def where_expr(query, expression, aggregates \\ %{})
 
@@ -77,7 +79,7 @@ defmodule Backpex.Adapters.Ecto.Where do
   @doc """
   Returns the query with the joins the expression needs and a dynamic expression of its value, e.g. of a computed
   value: a field, `concat`, a condition or a native expression. `aggregates` are the joined aggregates of
-  `join_aggregates/2`.
+  `join_aggregates/3`.
   """
   def value(query, expression, aggregates \\ %{}) do
     query = Ecto.Queryable.to_query(query)
@@ -104,15 +106,24 @@ defmodule Backpex.Adapters.Ecto.Where do
   Returns the query with one left join for every group of aggregates of the expressions that share the relationships,
   the kind, the `:where` and the `:sort`, e.g. the title and the likes of the post with the most likes, and the joined
   aggregates. `where_expr/3` and `value/3` take them to read the values of the aggregates from the joins.
+
+  The subqueries aggregate the related items of all items. With `of: keys`, a query of primary keys of items, they only
+  aggregate the related items of these items, so the values of all other items of the query are `nil`.
   """
-  def join_aggregates(query, expressions) do
+  def join_aggregates(query, expressions, opts \\ []) do
     query = Ecto.Queryable.to_query(query)
 
-    join_missing(query, {:root, schema(query)}, %{}, expressions)
+    join_missing(query, {:root, schema(query)}, %{}, expressions, Keyword.get(opts, :of))
   end
 
-  # Joins the aggregates of the expressions that are not joined yet, one join for every group.
-  defp join_missing(query, {binding_name, schema} = binding, joined, expressions) do
+  @doc """
+  Returns whether `join_aggregates/3` joins an aggregate for the expression.
+  """
+  def aggregates?(expression), do: aggregates(expression) != []
+
+  # Joins the aggregates of the expressions that are not joined yet, one join for every group. `of` is a query of the
+  # primary keys of the items to aggregate the related items of, or `nil` for all items.
+  defp join_missing(query, {binding_name, schema} = binding, joined, expressions, of \\ nil) do
     expressions
     |> Enum.flat_map(&aggregates/1)
     |> Enum.uniq()
@@ -122,7 +133,7 @@ defmodule Backpex.Adapters.Ecto.Where do
       name = free_aggregate_name!(query)
       keys = value_keys!(aggregates)
       primary_key = primary_key!(schema)
-      subquery = aggregate_subquery(aggregates, keys, schema, primary_key)
+      subquery = aggregates |> aggregate_subquery(keys, schema, primary_key) |> aggregate_of(of, primary_key)
 
       joined =
         aggregates
@@ -236,6 +247,10 @@ defmodule Backpex.Adapters.Ecto.Where do
     aggregate_select(subquery, aggregate, target, select, primary_key)
   end
 
+  defp aggregate_of(subquery, nil = _of, _primary_key), do: subquery
+
+  defp aggregate_of(subquery, of, primary_key), do: where(subquery, [item], field(item, ^primary_key) in subquery(of))
+
   defp aggregate_value(%Aggregate{kind: :count}, subquery, _target, _joined), do: {subquery, dynamic(count())}
 
   defp aggregate_value(%Aggregate{kind: :first, value: value}, subquery, target, joined),

test/adapters/ecto_test.exs

@@ -1074,6 +1074,126 @@ defmodule Backpex.Adapters.EctoTest do
       end
     end
 
+    test "aggregates only the related items of the items of a page for a query with a limit" do
+      defmodule PagedAggregatePostLive do
+        @moduledoc false
+        use FakeLiveResource, schema: Backpex.Adapters.EctoTest.TestPost
+
+        def computed(_assigns) do
+          [
+            tag_count: Expr.aggregate(:count, :tags),
+            first_tag: Expr.aggregate(:first, :tags, value: expr(name), sort: [name: :asc]),
+            first_tag_id: Expr.aggregate(:first, :tags, value: expr(id), sort: [name: :asc])
+          ]
+        end
+      end
+
+      query =
+        []
+        |> Resource.items_query(%{}, PagedAggregatePostLive)
+        |> Query.filter(expr(title == "Hello"))
+        |> Query.sort(title: :asc)
+        |> Query.offset(30)
+        |> Query.limit(15)
+        |> ecto_query()
+
+      # The page finds its primary keys without any aggregate, as the filter and the sort read none.
+      assert %{with_ctes: %{queries: [{"backpex_page", _opts, page}]}} = query
+      assert %{joins: [], wheres: [_filter], order_bys: [_sort], limit: %{}, offset: %{}} = page
+      assert Macro.to_string(page.select.expr) == "%{key: &0.id()}"
+
+      # The query selects the items of these keys in the order of the sort, and one join for every group of aggregates
+      # aggregates the related items of these keys.
+      assert %{limit: nil, offset: nil, order_bys: [_sort], wheres: [%{subqueries: [%{query: keys}]}]} = query
+      assert %{from: %{source: {"backpex_page", nil}}} = keys
+      assert [_count, _first] = query.joins
+
+      for %{qual: :left, source: %Ecto.SubQuery{query: aggregate}} <- query.joins do
+        assert [%{expr: where, subqueries: [%{query: ^keys}]}] = aggregate.wheres
+        assert Macro.to_string(where) == "&0.id() in {:subquery, 0}"
+      end
+
+      assert %{select: %{expr: {:merge, _meta, [_item, {:%{}, _map_meta, values}]}}} = query
+      assert values |> Keyword.keys() |> Enum.sort() == [:first_tag, :first_tag_id, :tag_count]
+    end
+
+    test "aggregates the related items of all items for what decides the items of a page" do
+      defmodule SortedAggregatePostLive do
+        @moduledoc false
+        use FakeLiveResource, schema: Backpex.Adapters.EctoTest.TestPost
+
+        def computed(_assigns) do
+          [
+            tag_count: Expr.aggregate(:count, :tags),
+            first_tag: Expr.aggregate(:first, :tags, value: expr(name), sort: [name: :asc])
+          ]
+        end
+      end
+
+      first_tag_id = Expr.aggregate(:first, :tags, value: expr(id), sort: [name: :asc])
+
+      query =
+        []
+        |> Resource.items_query(%{}, SortedAggregatePostLive)
+        |> Query.filter(expr(^first_tag_id > 1))
+        |> Query.sort(tag_count: :desc)
+        |> Query.limit(15)
+        |> ecto_query()
+
+      # The page joins the aggregates of the filter and of the sort over all items, but not the value it only selects.
+      assert %{with_ctes: %{queries: [{"backpex_page", _opts, page}]}} = query
+      assert [first, count] = Enum.sort_by(page.joins, &is_nil(&1.source.query.distinct))
+      assert %{wheres: [], select: %{expr: {:%{}, _meta, [_key, _first_tag_id]}}} = first.source.query
+      assert %{wheres: [], group_bys: [_id]} = count.source.query
+      assert %{wheres: [_filter], order_bys: [_sort], limit: %{}} = page
+
+      # The query sorts the items of the page by the aggregate of these items.
+      assert [_count, _first] = query.joins
+      assert Enum.all?(query.joins, &match?([_page], &1.source.query.wheres))
+      assert %{order_bys: [_sort], limit: nil} = query
+    end
+
+    test "aggregates the related items of all items without a limit, without selected aggregates or with a join" do
+      # All items, e.g. of an export.
+      tag_count = Expr.aggregate(:count, :tags)
+      query = PostLive |> Query.new() |> Query.compute(:tag_count, tag_count) |> Query.offset(15) |> ecto_query()
+
+      assert %{with_ctes: nil, joins: [%{source: %{query: %{wheres: []}}}], offset: %{}} = query
+
+      # An aggregate that is only filtered by.
+      query = PostLive |> Query.new() |> Query.filter(expr(^tag_count > 1)) |> Query.limit(15) |> ecto_query()
+
+      assert %{with_ctes: nil, joins: [%{source: %{query: %{wheres: []}}}], limit: %{}} = query
+
+      # An options query with a join may return an item more than once, which the primary keys of a page would not.
+      options_query = fn query, _assigns -> join(query, :inner, [post_tag], tag in assoc(post_tag, :tag)) end
+
+      query =
+        PostLive
+        |> Query.new()
+        |> Query.from([:post_tags])
+        |> EctoAdapter.put_field_options({:post_tags, %{options_query: options_query}})
+        |> Query.compute(:post_count, Expr.aggregate(:count, :post))
+        |> Query.limit(15)
+        |> ecto_query()
+
+      assert %{with_ctes: nil, joins: [_tag, %{source: %{query: %{wheres: []}}}], limit: %{}} = query
+    end
+
+    test "aggregates the related items of all items for the aggregates of a metric" do
+      Process.put({FakeRepo, :one}, 3)
+
+      assert {:ok, %{tags: 3}} =
+               PostLive
+               |> Query.new()
+               |> Query.compute(:first_tag, Expr.aggregate(:first, :tags, value: expr(name), sort: [name: :asc]))
+               |> Query.limit(15)
+               |> Query.aggregate(:tags, {:sum, Expr.aggregate(:count, :tags)})
+               |> EctoAdapter.aggregate()
+
+      assert_received {:repo, :one, %{with_ctes: nil, joins: [_count, _first]}}
+    end
+
     test "joins the aggregates of an exists and of an aggregate in their subqueries" do
       tag_count = Expr.aggregate(:count, :tags)
       tagged_post_count = Expr.aggregate(:count, :post, where: expr(^tag_count > 1))