fix: don't pass the extra pagination row to read hooks

lib/ash.ex

@@ -2970,14 +2970,25 @@ defmodule Ash do
           run: (Ash.DataLayer.data_layer_query() ->
                   {:ok, list(Ash.Resource.Record.t()) | Ash.Page.page() | no_return}
                   | {:error, Ash.Error.t()}),
-          load: (list(Ash.Resource.Record.t()) | Ash.Page.page() ->
+          load: (Ash.Query.t(), list(Ash.Resource.Record.t()) | Ash.Page.page() ->
                    {:ok, list(Ash.Resource.Record.t()) | Ash.Page.page()}
                    | {:error, Ash.Error.t()})
         }
 
   @doc """
   Gets the full query and any runtime calculations that would be loaded
 
+  ## Pagination
+
+  When the query is paginated, `run` returns an `Ash.Page.Offset` or
+  `Ash.Page.Keyset` rather than a list, otherwise a list of records.
+
+  Pass whatever `run` returned to `load`, which loads relationships,
+  calculations and load-through attributes on the records and returns the same
+  shape it was given. `load` also accepts the raw rows of a data layer query
+  you executed yourself and builds the page from them. Passing `page.results`
+  instead of the page loses `more?`.
+
   ## Examples
 
       iex> query = MyApp.Post |> Ash.Query.filter(published: true)

lib/ash/actions/read/read.ex

@@ -818,6 +818,7 @@ defmodule Ash.Actions.Read do
                          query
                        ),
                      :ok <- validate_get(results, query.action, query),
+                     {query, results} <- drop_pagination_extra(query, results, opts),
                      results <- add_keysets(query, results, query.sort),
                      {:ok, results} <- run_authorize_results(query, results),
                      {:ok, results, after_notifications} <- run_after_action(query, results),
@@ -1068,78 +1069,57 @@ defmodule Ash.Actions.Read do
              :ok <- validate_combinations(query, calculations_at_runtime, query.load),
              {:ok, data_layer_query} <-
                Ash.Query.data_layer_query(query, data_layer_calculations: data_layer_calculations) do
+          # `Ash.page/2` reruns the read with these, so it must produce a page.
+          page_opts = Keyword.delete(opts, :data_layer_query?)
+
           {:ok,
            %{
              query: data_layer_query,
              ash_query: query,
              load: fn query_ran, data ->
-               with {:ok, data} <-
-                      load_through_attributes(
-                        data,
-                        %{
-                          query_ran
-                          | calculations: Map.new(calculations_in_query, &{&1.name, &1})
-                        },
-                        query.domain,
-                        opts[:actor],
-                        opts[:tracer],
-                        opts[:authorize?]
-                      ),
-                    {:ok, data} <-
-                      load_relationships(data, query, opts),
-                    {:ok, data} <-
-                      Ash.Actions.Read.Calculations.run(
-                        data,
+               case data do
+                 %struct{results: results} = page
+                 when struct in [Ash.Page.Offset, Ash.Page.Keyset] ->
+                   with {:ok, results} <-
+                          load_data_layer_results(
+                            results,
+                            query_ran,
+                            query,
+                            initial_query,
+                            calculations_at_runtime,
+                            calculations_in_query,
+                            opts
+                          ) do
+                     {:ok, %{page | results: results}}
+                   end
+
+                 results when is_list(results) ->
+                   # Raw rows from a data layer query the caller ran themselves.
+                   with {:ok, results} <-
+                          load_data_layer_results(
+                            results,
+                            query_ran,
+                            query,
+                            initial_query,
+                            calculations_at_runtime,
+                            calculations_in_query,
+                            opts
+                          ),
+                        {:ok, resolved_count} <- count.() do
+                     {:ok,
+                      add_page(
+                        results,
+                        query.action,
+                        resolved_count,
+                        query.sort,
+                        initial_query,
                         query,
-                        calculations_at_runtime,
-                        calculations_in_query
-                      ),
-                    {:ok, data} <-
-                      load_through_attributes(
-                        data,
-                        %{
-                          query
-                          | calculations: Map.new(calculations_at_runtime, &{&1.name, &1}),
-                            load_through: Map.delete(query.load_through || %{}, :attribute)
-                        },
-                        query.domain,
-                        opts[:actor],
-                        opts[:tracer],
-                        opts[:authorize?],
-                        false
-                      ) do
-                 data
-                 |> Helpers.restrict_field_access(query)
-                 |> add_tenant(query)
-                 |> attach_fields(nil, initial_query, query, false)
-                 |> cleanup_field_auth(query)
-                 |> add_page(
-                   query.action,
-                   count,
-                   query.sort,
-                   initial_query,
-                   query,
-                   opts
-                 )
-               else
-                 {:error, %Ash.Query{errors: errors} = query} ->
-                   {:error, Ash.Error.to_error_class(errors, query: query)}
-
-                 {:error,
-                  %Ash.Error.Forbidden.Placeholder{
-                    authorizer: authorizer
-                  }} ->
-                   error =
-                     Ash.Authorizer.exception(
-                       authorizer,
-                       :forbidden,
-                       query_ran.context[:private][:authorizer_state][authorizer]
-                     )
-
-                   {:error, Ash.Error.to_error_class(error)}
-
-                 {:error, error} ->
-                   {:error, Ash.Error.to_error_class(error, query: query)}
+                        page_opts
+                      )}
+                   else
+                     {:error, error} ->
+                       {:error, Ash.Error.to_error_class(error, query: query)}
+                   end
                end
              end,
              run: fn data_layer_query ->
@@ -1162,12 +1142,24 @@ defmodule Ash.Actions.Read do
                         query
                       ),
                     :ok <- validate_get(results, query.action, query),
+                    {query, results} <- drop_pagination_extra(query, results, opts),
                     results <- add_keysets(query, results, query.sort),
                     {:ok, results} <- run_authorize_results(query, results),
-                    {:ok, results, after_notifications} <- run_after_action(query, results) do
+                    {:ok, results, after_notifications} <- run_after_action(query, results),
+                    {:ok, resolved_count} <- count.() do
                  notify_or_store(query, before_notifications ++ after_notifications, notify?)
 
-                 {:ok, add_tenant(results, query)}
+                 {:ok,
+                  results
+                  |> add_tenant(query)
+                  |> add_page(
+                    query.action,
+                    resolved_count,
+                    query.sort,
+                    initial_query,
+                    query,
+                    page_opts
+                  )}
                else
                  {%{valid?: false} = query, before_notifications} ->
                    notify_or_store(query, before_notifications, notify?)
@@ -1307,6 +1299,75 @@ defmodule Ash.Actions.Read do
     "__paginated_#{relationship_name}_count__"
   end
 
+  # Everything `Ash.read/2` does to records after they come back from the data
+  # layer, minus paging.
+  defp load_data_layer_results(
+         records,
+         query_ran,
+         query,
+         initial_query,
+         calculations_at_runtime,
+         calculations_in_query,
+         opts
+       ) do
+    with {:ok, records} <-
+           load_through_attributes(
+             records,
+             %{query_ran | calculations: Map.new(calculations_in_query, &{&1.name, &1})},
+             query.domain,
+             opts[:actor],
+             opts[:tracer],
+             opts[:authorize?]
+           ),
+         {:ok, records} <- load_relationships(records, query, opts),
+         {:ok, records} <-
+           Ash.Actions.Read.Calculations.run(
+             records,
+             query,
+             calculations_at_runtime,
+             calculations_in_query
+           ),
+         {:ok, records} <-
+           load_through_attributes(
+             records,
+             %{
+               query
+               | calculations: Map.new(calculations_at_runtime, &{&1.name, &1}),
+                 load_through: Map.delete(query.load_through || %{}, :attribute)
+             },
+             query.domain,
+             opts[:actor],
+             opts[:tracer],
+             opts[:authorize?],
+             false
+           ) do
+      records =
+        records
+        |> Helpers.restrict_field_access(query)
+        |> add_tenant(query)
+        |> attach_fields(nil, initial_query, query, false)
+        |> cleanup_field_auth(query)
+
+      {:ok, records}
+    else
+      {:error, %Ash.Query{errors: errors} = query} ->
+        {:error, Ash.Error.to_error_class(errors, query: query)}
+
+      {:error, %Ash.Error.Forbidden.Placeholder{authorizer: authorizer}} ->
+        error =
+          Ash.Authorizer.exception(
+            authorizer,
+            :forbidden,
+            query_ran.context[:private][:authorizer_state][authorizer]
+          )
+
+        {:error, Ash.Error.to_error_class(error)}
+
+      {:error, error} ->
+        {:error, Ash.Error.to_error_class(error, query: query)}
+    end
+  end
+
   @doc false
   def cleanup_field_auth(records, query, top_level? \\ true)
 
@@ -2966,23 +3027,18 @@ defmodule Ash.Actions.Read do
   @doc false
   def add_page(data, action, count, sort, original_query, new_query, opts) do
     cond do
-      opts[:skip_pagination?] ->
-        data
-
-      action.pagination == false ->
+      not paginated?(original_query, action, opts) ->
         data
 
-      original_query.page == false ->
-        data
-
-      opts[:return_unpaged?] && original_query.page[:limit] ->
-        Ash.Page.Unpaged.new(data, opts)
-
-      original_query.page[:limit] ->
-        to_page(data, action, count, sort, original_query, new_query, opts)
+      opts[:return_unpaged?] ->
+        Ash.Page.Unpaged.new(
+          data,
+          opts,
+          new_query.context[:pagination_more_by_source] || %{}
+        )
 
       true ->
-        data
+        to_page(data, action, count, sort, original_query, new_query, opts)
     end
   end
 
@@ -3014,7 +3070,30 @@ defmodule Ash.Actions.Read do
         last_record = List.last(data)
         not is_nil(last_record) && not is_nil(last_record.__metadata__[:keyset])
       else
-        not Enum.empty?(rest)
+        # `drop_pagination_extra/3` already removed the extra row and put `more?`
+        # on the context. Raw rows given to `load` skip it, so there the extra
+        # row is still in `data` and ends up in `rest`.
+        cond do
+          is_boolean(new_query.context[:pagination_more?]) ->
+            new_query.context[:pagination_more?]
+
+          is_map(new_query.context[:pagination_more_by_source]) ->
+            # A relationship load: one answer per `__lateral_join_source__`.
+            case data do
+              [record | _] ->
+                Map.get(
+                  new_query.context[:pagination_more_by_source],
+                  record.__lateral_join_source__,
+                  false
+                )
+
+              [] ->
+                false
+            end
+
+          true ->
+            not Enum.empty?(rest)
+        end
       end
 
     if page_opts[:offset] do
@@ -3058,6 +3137,55 @@ defmodule Ash.Actions.Read do
     end
   end
 
+  # Drops the extra row fetched to determine `more?` and records `more?` on the
+  # query context, so `to_page/7` can build the page without it.
+  defp drop_pagination_extra(query, results, opts) do
+    cond do
+      not paginated?(query, query.action, opts) ->
+        {query, results}
+
+      match?(%{data_layer: %{lateral_join_source: {_, _}}}, query.context) ->
+        # A lateral join fetches `limit + 1` rows per source record.
+        drop_pagination_extra_per_parent(query, results, query.page[:limit])
+
+      true ->
+        {results, more?} = take_page(results, query.page[:limit])
+        {Ash.Query.set_context(query, %{pagination_more?: more?}), results}
+    end
+  end
+
+  # `more?` per `__lateral_join_source__`, carried to the per-parent `to_page/7`
+  # call via `Ash.Page.Unpaged.more_by_source`. Rows of one parent all carry the
+  # same source value, so no key normalization is needed here.
+  defp drop_pagination_extra_per_parent(query, results, limit) do
+    groups =
+      results
+      |> Enum.with_index()
+      |> Enum.group_by(fn {record, _index} -> record.__lateral_join_source__ end)
+
+    more_by_source =
+      Map.new(groups, fn {source, records} -> {source, length(records) > limit} end)
+
+    results =
+      groups
+      |> Enum.flat_map(fn {_source, records} -> Enum.take(records, limit) end)
+      |> Enum.sort_by(&elem(&1, 1))
+      |> Enum.map(&elem(&1, 0))
+
+    {%{query | context: Map.put(query.context, :pagination_more_by_source, more_by_source)},
+     results}
+  end
+
+  defp take_page(results, limit) do
+    {results, extra} = Enum.split(results, limit)
+    {results, extra != []}
+  end
+
+  defp paginated?(%{page: page_opts}, action, opts) do
+    not (opts[:skip_pagination?] || action.pagination == false ||
+           page_opts in [nil, false] || is_nil(page_opts[:limit]))
+  end
+
   defp remove_already_selected(fields, %struct{results: results})
        when struct in [Ash.Page.Keyset, Ash.Page.Offset],
        do: remove_already_selected(fields, results)