Packages
ash_postgres
1.3.2
2.10.0
2.9.1
2.9.0
2.8.0
2.7.0
2.6.32
2.6.31
2.6.30
2.6.29
2.6.28
2.6.27
2.6.26
2.6.25
2.6.24
2.6.23
2.6.22
2.6.21
2.6.20
2.6.19
2.6.18
2.6.17
2.6.16
2.6.15
2.6.14
2.6.13
2.6.12
2.6.11
2.6.10
2.6.9
2.6.8
2.6.7
2.6.6
2.6.5
2.6.4
2.6.3
2.6.2
2.6.1
2.6.0
2.5.22
2.5.21
2.5.20
2.5.19
2.5.18
2.5.17
2.5.16
2.5.15
2.5.14
2.5.13
2.5.12
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.4.21
2.4.20
2.4.19
2.4.18
2.4.17
2.4.16
2.4.15
2.4.14
2.4.13
2.4.12
2.4.11
2.4.10
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.1
2.3.0
2.2.5
2.2.4
2.2.3
2.2.2
2.2.1
2.2.0
2.1.19
2.1.18
2.1.17
2.1.15
2.1.14
2.1.13
2.1.12
2.1.11
2.1.10
2.1.9
2.1.8
2.1.6
2.1.5
2.1.4
2.1.3
2.1.2
2.1.1
2.1.0
2.0.12
2.0.11
2.0.10
2.0.9
2.0.8
2.0.7
2.0.6
2.0.5
2.0.4
2.0.3
2.0.2
2.0.1
2.0.0
2.0.0-rc.15
2.0.0-rc.14
2.0.0-rc.13
2.0.0-rc.12
2.0.0-rc.11
2.0.0-rc.10
2.0.0-rc.9
2.0.0-rc.8
2.0.0-rc.7
2.0.0-rc.6
2.0.0-rc.5
2.0.0-rc.4
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
1.5.30
1.5.29
1.5.28
1.5.27
1.5.26
1.5.25
1.5.24
1.5.23
1.5.22
1.5.21
1.5.20
1.5.19
1.5.18
1.5.17
1.5.16
1.5.15
1.5.14
1.5.13
1.5.12
1.5.11
1.5.10
1.5.9
1.5.8
1.5.7
1.5.6
1.5.5
1.5.4
1.5.3
1.5.2
1.5.1
1.5.0
1.4.0
1.3.68
1.3.67
1.3.66
1.3.65
1.3.64
1.3.63
1.3.62
1.3.61
1.3.60
1.3.59
1.3.58
1.3.56
1.3.55
1.3.54
1.3.53
1.3.52
1.3.51
1.3.50
1.3.49
1.3.48
1.3.47
1.3.46
1.3.45
1.3.44
1.3.43
1.3.42
1.3.41
1.3.40
1.3.39
1.3.38
1.3.37
1.3.36
1.3.35
1.3.34
1.3.33
1.3.32
1.3.31
1.3.30
1.3.29
1.3.28
1.3.26
1.3.25
1.3.24
1.3.23
1.3.22
1.3.21
1.3.20
1.3.19
1.3.18
1.3.17
1.3.16
1.3.15
1.3.14
1.3.12
1.3.11
1.3.10
1.3.9
1.3.8
1.3.6
1.3.5
1.3.4
1.3.3
1.3.2
1.3.1
1.3.0
1.3.0-rc.4
1.3.0-rc.3
1.3.0-rc.2
1.3.0-rc.0
1.2.6
1.2.5
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.2.0-rc.1
1.2.0-rc.0
1.1.3
1.1.2
1.1.1
1.1.0
1.0.0
1.0.0-rc.9
1.0.0-rc.8
1.0.0-rc.7
1.0.0-rc.6
1.0.0-rc.5
1.0.0-rc.4
1.0.0-rc.3
1.0.0-rc.2
1.0.0-rc.1
1.0.0-rc.0
1.0.0-pre.3
1.0.0-pre.2
1.0.0-pre.1
1.0.0-pre.0
0.43.0
0.42.0-rc.7
0.42.0-rc.6
0.42.0-rc.5
0.42.0-rc.4
0.42.0-rc.3
0.42.0-rc.1
0.42.0-rc.0
0.41.7
0.41.6
0.41.5
0.41.4
0.41.3
0.41.2
0.41.1
0.41.0-rc0
retired
0.41.0-rc.9
0.41.0-rc.8
0.41.0-rc.7
0.41.0-rc.6
0.41.0-rc.5
0.41.0-rc.4
0.41.0-rc.3
0.41.0-rc.2
0.41.0-rc.1
0.41.0-rc.0
0.40.11
0.40.10
0.40.9
0.40.8
0.40.7
0.40.6
0.40.5
0.40.4
0.40.3
0.40.2
0.40.1
0.40.0-rc5
0.40.0-rc4
0.40.0-rc3
0.40.0-rc2
0.40.0-rc1
0.39.0-rc0
0.38.11
0.38.10
0.38.9
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.8
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.5
0.35.4
0.35.3
0.35.1
0.34.5
0.34.4
0.34.3
0.34.2
0.34.1
0.34.0
0.33.1
0.33.0
0.32.2
0.32.1
0.32.0
0.31.1
0.30.1
0.29.6
0.29.5
0.29.4
0.29.3
0.29.2
0.29.1
0.28.1
0.28.0
0.27.0
0.26.2
0.26.1
0.26.0
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.0
0.23.2
0.23.1
0.23.0
0.22.1
0.22.0
0.21.0
0.20.1
0.20.0
0.19.0
0.18.0
0.17.0
0.16.1
0.15.0
0.14.0
0.13.0
0.12.1
0.12.0
0.11.2
0.10.0
0.9.0
0.8.0
0.7.0
0.6.0
0.5.0
0.4.0
0.3.0
0.2.1
0.2.0
0.1.4
0.1.3
0.1.2
0.1.1
0.1.1-rc.2
0.1.1-rc.1
0.1.1-rc.0
0.1.0
The PostgreSQL data layer for Ash Framework
Current section
Files
Jump to
Current section
Files
lib/aggregate.ex
defmodule AshPostgres.Aggregate do
@moduledoc false
import Ecto.Query, only: [from: 2, subquery: 1]
require Ecto.Query
def add_aggregates(query, aggregates, resource, select? \\ true, source_binding \\ nil)
def add_aggregates(query, [], _, _, _), do: {:ok, query}
def add_aggregates(query, aggregates, resource, select?, source_binding) do
query = AshPostgres.DataLayer.default_bindings(query, resource)
aggregates =
Enum.reject(aggregates, fn aggregate ->
Map.has_key?(query.__ash_bindings__.aggregate_defs, aggregate.name)
end)
query =
query
|> Map.update!(:__ash_bindings__, fn bindings ->
bindings
|> Map.update!(:aggregate_defs, fn aggregate_defs ->
Map.merge(aggregate_defs, Map.new(aggregates, &{&1.name, &1}))
end)
end)
source_binding =
source_binding ||
query.__ash_bindings__.bindings
|> Enum.reject(fn
{_, %{aggregates: _}} ->
true
_ ->
false
end)
|> Enum.map(&elem(&1, 0))
|> Enum.max()
result =
aggregates
|> Enum.reject(&already_added?(&1, query.__ash_bindings__))
|> Enum.group_by(& &1.relationship_path)
|> Enum.flat_map(fn {path, aggregates} ->
{can_group, cant_group} = Enum.split_with(aggregates, &can_group?(resource, &1))
[{path, can_group}] ++ Enum.map(cant_group, &{path, [&1]})
end)
|> Enum.filter(fn
{_, []} ->
false
_ ->
true
end)
|> Enum.reduce_while({:ok, query, []}, fn {[first_relationship | relationship_path],
aggregates},
{:ok, query, dynamics} ->
first_relationship = Ash.Resource.Info.relationship(resource, first_relationship)
is_single? = Enum.count_until(aggregates, 2) == 1
with {:ok, agg_root_query} <-
AshPostgres.Join.maybe_get_resource_query(
first_relationship.destination,
first_relationship,
query
),
agg_root_query <-
Map.update!(agg_root_query, :__ash_bindings__, &Map.put(&1, :in_group?, true)),
{:ok, joined} <-
join_all_relationships(
agg_root_query,
aggregates,
relationship_path,
first_relationship,
is_single?
),
{:ok, filtered} <-
maybe_filter_subquery(
joined,
first_relationship,
relationship_path,
aggregates,
is_single?,
source_binding
),
with_subquery_select <-
select_all_aggregates(
aggregates,
filtered,
relationship_path,
query,
is_single?,
Ash.Resource.Info.related(first_relationship.destination, relationship_path)
),
query <-
join_subquery(
query,
with_subquery_select,
first_relationship,
relationship_path,
aggregates,
source_binding
) do
if select? do
new_dynamics =
Enum.map(
aggregates,
&{&1.load, &1.name,
select_dynamic(resource, query, &1, query.__ash_bindings__.current - 1)}
)
{:cont, {:ok, query, new_dynamics ++ dynamics}}
else
{:cont, {:ok, query, dynamics}}
end
end
end)
case result do
{:ok, query, dynamics} ->
{:ok, add_aggregate_selects(query, dynamics)}
{:error, error} ->
{:error, error}
end
end
defp already_added?(aggregate, bindings) do
Enum.any?(bindings.bindings, fn
{_, %{type: :aggregate, aggregates: aggregates}} ->
aggregate in aggregates
_ ->
false
end)
end
defp maybe_filter_subquery(
agg_query,
first_relationship,
relationship_path,
[aggregate | _rest],
false,
source_binding
) do
apply_agg_authorization_filter(
agg_query,
aggregate,
relationship_path,
first_relationship,
source_binding
)
end
defp maybe_filter_subquery(
agg_query,
first_relationship,
relationship_path,
[aggregate],
true,
source_binding
) do
with {:ok, agg_query} <-
apply_agg_query(
agg_query,
aggregate,
relationship_path,
first_relationship,
source_binding
) do
apply_agg_authorization_filter(
agg_query,
aggregate,
relationship_path,
first_relationship,
source_binding
)
end
end
defp apply_agg_query(
agg_query,
aggregate,
relationship_path,
first_relationship,
source_binding
) do
if has_filter?(aggregate.query) do
filter =
Ash.Filter.move_to_relationship_path(
aggregate.query.filter,
relationship_path
)
|> Map.put(:resource, first_relationship.destination)
used_calculations =
Ash.Filter.used_calculations(
filter,
first_relationship.destination,
relationship_path
)
related = Ash.Resource.Info.related(first_relationship.destination, relationship_path)
used_calculations =
case Ash.Resource.Info.calculation(related, aggregate.field) do
%{name: name, calculation: {module, opts}, type: type, constraints: constraints} ->
{:ok, new_calc} = Ash.Query.Calculation.new(name, module, opts, {type, constraints})
if new_calc in used_calculations do
used_calculations
else
[
new_calc
| used_calculations
]
end
nil ->
used_calculations
end
used_aggregates =
used_aggregates(
filter,
first_relationship.destination,
used_calculations,
relationship_path
)
case add_aggregates(
agg_query,
used_aggregates,
first_relationship.destination,
false,
source_binding
) do
{:ok, agg_query} ->
AshPostgres.DataLayer.filter(agg_query, filter, agg_query.__ash_bindings__.resource)
other ->
other
end
else
{:ok, agg_query}
end
end
defp apply_agg_authorization_filter(
agg_query,
aggregate,
relationship_path,
first_relationship,
source_binding
) do
filter =
if is_nil(aggregate.authorization_filter) do
nil
else
Ash.Filter.move_to_relationship_path(
aggregate.authorization_filter,
relationship_path
)
|> Map.put(:resource, first_relationship.destination)
end
filter =
if is_nil(Map.get(aggregate, :join_filter)) do
filter
else
Ash.Filter.add_to_filter(filter, aggregate.join_filter)
end
used_calculations =
Ash.Filter.used_calculations(
filter,
first_relationship.destination,
relationship_path
)
used_aggregates =
used_aggregates(
filter,
first_relationship.destination,
used_calculations,
relationship_path
)
case add_aggregates(
agg_query,
used_aggregates,
first_relationship.destination,
false,
source_binding
) do
{:ok, agg_query} ->
AshPostgres.DataLayer.filter(agg_query, filter, first_relationship.destination)
other ->
other
end
end
defp join_subquery(
query,
subquery,
%{manual: {module, opts}} = first_relationship,
_relationship_path,
aggregates,
source_binding
) do
field = first_relationship.destination_attribute
new_subquery =
from(row in subquery,
select_merge: map(row, ^[field]),
group_by: field(row, ^first_relationship.destination_attribute),
distinct: true
)
{:ok, subquery} =
module.ash_postgres_subquery(
opts,
source_binding,
subquery.__ash_bindings__.current - 1,
new_subquery
)
subquery = AshPostgres.Join.set_join_prefix(subquery, query, first_relationship.destination)
query =
from(row in query,
left_lateral_join: sub in subquery(subquery_if_distinct(subquery)),
as: ^query.__ash_bindings__.current
)
AshPostgres.DataLayer.add_binding(
query,
%{
path: [],
type: :aggregate,
aggregates: aggregates
}
)
end
defp join_subquery(
query,
subquery,
%{type: :many_to_many, join_relationship: join_relationship, source: source} =
first_relationship,
_relationship_path,
aggregates,
source_binding
) do
join_relationship_struct = Ash.Resource.Info.relationship(source, join_relationship)
{:ok, through} =
AshPostgres.Join.maybe_get_resource_query(
join_relationship_struct.destination,
join_relationship_struct,
query,
[],
nil,
subquery.__ash_bindings__.current
)
field = first_relationship.source_attribute_on_join_resource
subquery =
from(sub in subquery,
join: through in ^through,
as: ^subquery.__ash_bindings__.current,
on:
field(through, ^first_relationship.destination_attribute_on_join_resource) ==
field(sub, ^first_relationship.destination_attribute),
select_merge: map(through, ^[field]),
group_by: field(through, ^first_relationship.source_attribute_on_join_resource),
distinct: field(through, ^first_relationship.source_attribute_on_join_resource),
where:
field(
parent_as(^source_binding),
^first_relationship.source_attribute
) ==
field(
through,
^first_relationship.source_attribute_on_join_resource
)
)
subquery = AshPostgres.Join.set_join_prefix(subquery, query, first_relationship.destination)
query =
from(row in query,
left_lateral_join: agg in subquery(subquery_if_distinct(subquery)),
as: ^query.__ash_bindings__.current
)
AshPostgres.DataLayer.add_binding(
query,
%{
path: [],
type: :aggregate,
aggregates: aggregates
}
)
end
defp join_subquery(
query,
subquery,
first_relationship,
_relationship_path,
aggregates,
source_binding
) do
field = first_relationship.destination_attribute
if source_binding == 2 && first_relationship.source_attribute == :user_id do
IO.puts("""
DEBUG INFO:
#{inspect(first_relationship, pretty: true)}
#{inspect(subquery, pretty: true)}
#{inspect(query.__ash_bindings__, pretty: true)}
""")
end
subquery =
from(row in subquery,
group_by: field(row, ^first_relationship.destination_attribute),
select_merge: map(row, ^[field]),
where:
field(parent_as(^source_binding), ^first_relationship.source_attribute) ==
field(as(^0), ^first_relationship.destination_attribute)
)
subquery = AshPostgres.Join.set_join_prefix(subquery, query, first_relationship.destination)
query =
from(row in query,
left_lateral_join: agg in subquery(subquery_if_distinct(subquery)),
as: ^query.__ash_bindings__.current
)
AshPostgres.DataLayer.add_binding(
query,
%{
path: [],
type: :aggregate,
aggregates: aggregates
}
)
end
defp subquery_if_distinct(%{distinct: nil} = query), do: query
defp subquery_if_distinct(subquery) do
from(row in subquery(subquery),
select: row
)
end
defp select_all_aggregates(aggregates, joined, relationship_path, _query, is_single?, resource) do
Enum.reduce(aggregates, joined, fn aggregate, joined ->
add_subquery_aggregate_select(joined, relationship_path, aggregate, resource, is_single?)
end)
end
defp join_all_relationships(
agg_root_query,
_aggregates,
relationship_path,
first_relationship,
_is_single?
) do
if Enum.empty?(relationship_path) do
{:ok, agg_root_query}
else
AshPostgres.Join.join_all_relationships(
agg_root_query,
nil,
[],
[
{:inner,
AshPostgres.Join.relationship_path_to_relationships(
first_relationship.destination,
relationship_path
)}
],
[],
nil
)
end
end
defp can_group?(_, %{kind: :list}), do: false
defp can_group?(resource, aggregate) do
can_group_kind?(aggregate, resource) && !has_exists?(aggregate) &&
!references_relationships?(aggregate)
end
# We can potentially optimize this. We don't have to prevent aggregates that reference
# relationships from joining, we can
# 1. group up the ones that do join relationships by the relationships they join
# 2. potentially group them all up that join to relationships and just join to all the relationships
# but this method is predictable and easy so we're starting by just not grouping them
defp references_relationships?(aggregate) do
!!Ash.Filter.find(aggregate.query && aggregate.query.filter, fn
%Ash.Query.Ref{relationship_path: relationship_path} when relationship_path != [] ->
true
_ ->
false
end)
end
defp can_group_kind?(aggregate, resource) do
if aggregate.kind == :first do
related = Ash.Resource.Info.related(resource, aggregate.relationship_path)
case Ash.Resource.Info.field(related, aggregate.field).type do
{:array, _} ->
false
_ ->
true
end
else
true
end
end
defp has_exists?(aggregate) do
!!Ash.Filter.find(aggregate.query && aggregate.query.filter, fn
%Ash.Query.Exists{} -> true
_ -> false
end)
end
defp add_aggregate_selects(query, dynamics) do
{in_aggregates, in_body} =
Enum.split_with(dynamics, fn {load, _name, _dynamic} -> is_nil(load) end)
aggs =
in_body
|> Map.new(fn {load, _, dynamic} ->
{load, dynamic}
end)
aggs =
if Enum.empty?(in_aggregates) do
aggs
else
Map.put(
aggs,
:aggregates,
Map.new(in_aggregates, fn {_, name, dynamic} ->
{name, dynamic}
end)
)
end
Ecto.Query.select_merge(query, ^aggs)
end
defp select_dynamic(_resource, _query, aggregate, binding) do
type = AshPostgres.Types.parameterized_type(aggregate.type, aggregate.constraints)
field =
if type do
Ecto.Query.dynamic(
type(
field(as(^binding), ^aggregate.name),
^type
)
)
else
Ecto.Query.dynamic(field(as(^binding), ^aggregate.name))
end
coalesced =
if is_nil(aggregate.default_value) do
field
else
if type do
Ecto.Query.dynamic(
coalesce(
^field,
type(
^aggregate.default_value,
^type
)
)
)
else
Ecto.Query.dynamic(
coalesce(
^field,
^aggregate.default_value
)
)
end
end
if type do
Ecto.Query.dynamic(type(^coalesced, ^type))
else
coalesced
end
end
defp has_filter?(nil), do: false
defp has_filter?(%{filter: nil}), do: false
defp has_filter?(%{filter: %Ash.Filter{expression: nil}}), do: false
defp has_filter?(%{filter: %Ash.Filter{}}), do: true
defp has_filter?(_), do: false
defp has_sort?(nil), do: false
defp has_sort?(%{sort: nil}), do: false
defp has_sort?(%{sort: []}), do: false
defp has_sort?(%{sort: _}), do: true
defp has_sort?(_), do: false
def used_aggregates(filter, resource, used_calculations, path) do
Ash.Filter.used_aggregates(filter, path) ++
Enum.flat_map(
used_calculations,
fn calculation ->
case Ash.Filter.hydrate_refs(
calculation.module.expression(calculation.opts, calculation.context),
%{
resource: resource,
aggregates: %{},
calculations: %{},
public?: false
}
) do
{:ok, hydrated} ->
Ash.Filter.used_aggregates(hydrated)
_ ->
[]
end
end
)
end
def add_subquery_aggregate_select(
query,
relationship_path,
%{kind: :first} = aggregate,
resource,
is_single?
) do
query = AshPostgres.DataLayer.default_bindings(query, aggregate.resource)
ref = %Ash.Query.Ref{
attribute: Ash.Resource.Info.field(resource, aggregate.field),
relationship_path: relationship_path,
resource: resource
}
type = AshPostgres.Types.parameterized_type(aggregate.type, aggregate.constraints)
binding =
AshPostgres.DataLayer.get_binding(
query.__ash_bindings__.resource,
relationship_path,
query,
[:left, :inner, :root]
)
field = AshPostgres.Expr.dynamic_expr(query, ref, query.__ash_bindings__, false)
sorted =
if has_sort?(aggregate.query) do
{:ok, sort_expr} =
AshPostgres.Sort.sort(
query,
aggregate.query.sort,
Ash.Resource.Info.related(resource, relationship_path),
relationship_path,
binding,
true
)
question_marks = Enum.map(sort_expr, fn _ -> " ? " end)
{:ok, expr} =
AshPostgres.Functions.Fragment.casted_new(
["array_agg(? ORDER BY #{question_marks})", field] ++ sort_expr
)
AshPostgres.Expr.dynamic_expr(query, expr, query.__ash_bindings__, false)
else
Ecto.Query.dynamic(
[row],
fragment("array_agg(?)", ^field)
)
end
filtered = filter_field(sorted, query, aggregate, relationship_path, is_single?)
value = Ecto.Query.dynamic(fragment("(?)[1]", ^filtered))
with_default =
if aggregate.default_value do
if type do
Ecto.Query.dynamic(coalesce(^value, type(^aggregate.default_value, ^type)))
else
Ecto.Query.dynamic(coalesce(^value, ^aggregate.default_value))
end
else
value
end
casted =
if type do
Ecto.Query.dynamic(type(^with_default, ^type))
else
with_default
end
select_or_merge(query, aggregate.name, casted)
end
def add_subquery_aggregate_select(
query,
relationship_path,
%{kind: :list} = aggregate,
resource,
is_single?
) do
query = AshPostgres.DataLayer.default_bindings(query, aggregate.resource)
type = AshPostgres.Types.parameterized_type(aggregate.type, aggregate.constraints)
binding =
AshPostgres.DataLayer.get_binding(
query.__ash_bindings__.resource,
relationship_path,
query,
[:left, :inner, :root]
)
ref = %Ash.Query.Ref{
attribute: Ash.Resource.Info.field(resource, aggregate.field),
relationship_path: relationship_path,
resource: resource
}
field = AshPostgres.Expr.dynamic_expr(query, ref, query.__ash_bindings__, false)
sorted =
if has_sort?(aggregate.query) do
{:ok, sort_expr} =
AshPostgres.Sort.sort(
query,
aggregate.query.sort,
Ash.Resource.Info.related(resource, relationship_path),
relationship_path,
binding,
true
)
question_marks = Enum.map(sort_expr, fn _ -> " ? " end)
{:ok, expr} =
AshPostgres.Functions.Fragment.casted_new(
["array_agg(? ORDER BY #{question_marks})", field] ++ sort_expr
)
AshPostgres.Expr.dynamic_expr(query, expr, query.__ash_bindings__, false)
else
Ecto.Query.dynamic(
[row],
fragment("array_agg(?)", ^field)
)
end
filtered = filter_field(sorted, query, aggregate, relationship_path, is_single?)
with_default =
if aggregate.default_value do
if type do
Ecto.Query.dynamic(coalesce(^filtered, type(^aggregate.default_value, ^type)))
else
Ecto.Query.dynamic(coalesce(^filtered, ^aggregate.default_value))
end
else
filtered
end
cast =
if type do
Ecto.Query.dynamic(type(^with_default, ^{:array, type}))
else
with_default
end
select_or_merge(query, aggregate.name, cast)
end
def add_subquery_aggregate_select(
query,
relationship_path,
%{kind: kind} = aggregate,
resource,
is_single?
)
when kind in [:count, :sum, :avg, :max, :min, :custom] do
query = AshPostgres.DataLayer.default_bindings(query, aggregate.resource)
ref = %Ash.Query.Ref{
attribute:
Ash.Resource.Info.field(
resource,
aggregate.field || List.first(Ash.Resource.Info.primary_key(resource))
),
relationship_path: relationship_path,
resource: resource
}
field =
if kind == :custom do
# we won't use this if its custom so don't try to make one
nil
else
AshPostgres.Expr.dynamic_expr(query, ref, query.__ash_bindings__, false)
end
type = AshPostgres.Types.parameterized_type(aggregate.type, aggregate.constraints)
binding =
AshPostgres.DataLayer.get_binding(
query.__ash_bindings__.resource,
relationship_path,
query,
[:left, :inner, :root]
)
field =
case kind do
:count ->
Ecto.Query.dynamic([row], count(^field))
:sum ->
Ecto.Query.dynamic([row], sum(^field))
:avg ->
Ecto.Query.dynamic([row], avg(^field))
:max ->
Ecto.Query.dynamic([row], max(^field))
:min ->
Ecto.Query.dynamic([row], min(^field))
:custom ->
{module, opts} = aggregate.implementation
module.dynamic(opts, binding)
end
filtered = filter_field(field, query, aggregate, relationship_path, is_single?)
with_default =
if aggregate.default_value do
if type do
Ecto.Query.dynamic(coalesce(^filtered, type(^aggregate.default_value, ^type)))
else
Ecto.Query.dynamic(coalesce(^filtered, ^aggregate.default_value))
end
else
filtered
end
cast =
if type do
Ecto.Query.dynamic(type(^with_default, ^type))
else
with_default
end
select_or_merge(query, aggregate.name, cast)
end
defp filter_field(field, _query, _aggregate, _relationship_path, true) do
field
end
defp filter_field(field, query, aggregate, relationship_path, _is_single?) do
if has_filter?(aggregate.query) do
filter =
Ash.Filter.move_to_relationship_path(
aggregate.query.filter,
relationship_path
)
expr =
AshPostgres.Expr.dynamic_expr(
query,
filter,
query.__ash_bindings__,
false,
AshPostgres.Types.parameterized_type(aggregate.type, aggregate.constraints)
)
Ecto.Query.dynamic(filter(^field, ^expr))
else
field
end
end
defp select_or_merge(query, aggregate_name, casted) do
query =
if query.select do
query
else
Ecto.Query.select(query, %{})
end
Ecto.Query.select_merge(query, ^%{aggregate_name => casted})
end
end