fix: use AshSql for running aggregate queries

this came up due to a bug where `ash_sqlite` was not applying filters.
using the abstracted `AshSql.AggregateQuery` code both fixes the
bug and ensures that `AshSqlite` gets the state of the art from the
new(-ish) centralized `AshSql` code.

lib/data_layer.ex

@@ -406,7 +406,7 @@ defmodule AshSqlite.DataLayer do
     Mix.Task.run("ash_sqlite.drop", args)
   end
 
-  import Ecto.Query, only: [from: 2, subquery: 1]
+  import Ecto.Query, only: [from: 2]
 
   @impl true
   def can?(_, :async_engine), do: false
@@ -523,76 +523,12 @@ defmodule AshSqlite.DataLayer do
 
   @impl true
   def run_aggregate_query(query, aggregates, resource) do
-    {exists, aggregates} = Enum.split_with(aggregates, &(&1.kind == :exists))
-    query = AshSql.Bindings.default_bindings(query, resource, AshSqlite.SqlImplementation)
-
-    query =
-      if query.limit do
-        query =
-          query
-          |> Ecto.Query.exclude(:select)
-          |> Ecto.Query.exclude(:order_by)
-          |> Map.put(:windows, [])
-
-        from(row in subquery(query), as: ^0, select: %{})
-      else
-        query
-        |> Ecto.Query.exclude(:select)
-        |> Ecto.Query.exclude(:order_by)
-        |> Map.put(:windows, [])
-        |> Ecto.Query.select(%{})
-      end
-
-    query_before_select = query
-
-    query =
-      Enum.reduce(
-        aggregates,
-        query,
-        fn agg, query ->
-          AshSql.Aggregate.add_subquery_aggregate_select(
-            query,
-            agg.relationship_path |> Enum.drop(1),
-            agg,
-            resource,
-            true,
-            Ash.Resource.Info.relationship(resource, agg.relationship_path |> Enum.at(1))
-          )
-        end
-      )
-
-    result =
-      case aggregates do
-        [] ->
-          %{}
-
-        _ ->
-          dynamic_repo(resource, query).one(query, repo_opts(nil, nil, resource))
-      end
-
-    {:ok, add_exists_aggs(result, resource, query_before_select, exists)}
-  end
-
-  defp add_exists_aggs(result, resource, query, exists) do
-    repo = dynamic_repo(resource, query)
-    repo_opts = repo_opts(nil, nil, resource)
-
-    Enum.reduce(exists, result, fn agg, result ->
-      {:ok, filtered} =
-        case agg do
-          %{query: %{filter: filter}} when not is_nil(filter) ->
-            filter(query, filter, resource)
-
-          _ ->
-            {:ok, query}
-        end
-
-      Map.put(
-        result || %{},
-        agg.name,
-        repo.exists?(filtered, repo_opts)
-      )
-    end)
+    AshSql.AggregateQuery.run_aggregate_query(
+      query,
+      aggregates,
+      resource,
+      AshSqlite.SqlImplementation
+    )
   end
 
   @impl true

test/aggregate_test.exs

@@ -0,0 +1,19 @@
+defmodule AshSqlite.AggregatesTest do
+  use AshSqlite.RepoCase, async: false
+
+  require Ash.Query
+  alias AshSqlite.Test.Post
+
+  test "a count with a filter returns the appropriate value" do
+    Ash.Seed.seed!(%Post{title: "foo"})
+    Ash.Seed.seed!(%Post{title: "foo"})
+    Ash.Seed.seed!(%Post{title: "bar"})
+
+    count =
+      Post
+      |> Ash.Query.filter(title == "foo")
+      |> Ash.count!()
+
+    assert count == 2
+  end
+end