Packages
ash_postgres
1.3.39
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?,
source_binding,
root_data \\ nil
)
def add_aggregates(query, [], _, _, _, _), do: {:ok, query}
def add_aggregates(query, aggregates, resource, select?, source_binding, root_data) do
case resource_aggregates_to_aggregates(resource, aggregates) do
{:ok, aggregates} ->
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)
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? = match?([_], aggregates)
cond do
is_single? &&
optimizable_first_aggregate?(resource, Enum.at(aggregates, 0)) ->
case add_first_join_aggregate(query, resource, hd(aggregates), root_data) do
{:ok, query, dynamic} ->
query =
if select? do
select_or_merge(query, hd(aggregates).name, dynamic)
else
query
end
{:cont, {:ok, query, dynamics}}
{:error, error} ->
{:halt, {:error, error}}
end
is_single? && Enum.at(aggregates, 0).kind == :exists ->
[aggregate] = aggregates
exists =
AshPostgres.Expr.dynamic_expr(
query,
%Ash.Query.Exists{path: aggregate.relationship_path, expr: true},
query.__ash_bindings__
)
{:cont, {:ok, query, [{aggregate.load, aggregate.name, exists} | dynamics]}}
true ->
with {:ok, agg_root_query} <-
AshPostgres.Join.maybe_get_resource_query(
first_relationship.destination,
first_relationship,
query,
[first_relationship.name]
),
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
end)
case result do
{:ok, query, dynamics} ->
{:ok, add_aggregate_selects(query, dynamics)}
{:error, error} ->
{:error, error}
end
{:error, error} ->
{:error, error}
end
end
defp resource_aggregates_to_aggregates(resource, aggregates) do
aggregates
|> Enum.reduce_while({:ok, []}, fn
%Ash.Query.Aggregate{} = aggregate, {:ok, aggregates} ->
{:cont, {:ok, [aggregate | aggregates]}}
aggregate, {:ok, aggregates} ->
related = Ash.Resource.Info.related(resource, aggregate.relationship_path)
read_action =
aggregate.read_action || Ash.Resource.Info.primary_action!(related, :read).name
with %{valid?: true} = aggregate_query <- Ash.Query.for_read(related, read_action),
%{valid?: true} = aggregate_query <-
Ash.Query.build(aggregate_query, filter: aggregate.filter, sort: aggregate.sort) do
Ash.Query.Aggregate.new(
resource,
aggregate.name,
aggregate.kind,
path: aggregate.relationship_path,
query: aggregate_query,
field: aggregate.field,
default: aggregate.default,
filterable?: aggregate.filterable?,
type: aggregate.type,
constraints: aggregate.constraints,
implementation: aggregate.implementation,
uniq?: aggregate.uniq?,
read_action:
aggregate.read_action ||
Ash.Resource.Info.primary_action!(
Ash.Resource.Info.related(resource, aggregate.relationship_path),
:read
).name,
authorize?: aggregate.authorize?
)
else
%{valid?: false, errors: errors} ->
{:error, errors}
{:error, error} ->
{:error, error}
end
|> case do
{:ok, aggregate} ->
{:cont, {:ok, [aggregate | aggregates]}}
{:error, error} ->
{:halt, {:error, error}}
end
end)
end
defp add_first_join_aggregate(query, resource, aggregate, root_data) do
{resource, path} =
case root_data do
{resource, path} ->
{resource, path}
_ ->
{resource, []}
end
case AshPostgres.Join.join_all_relationships(
query,
nil,
[],
[
{:left,
AshPostgres.Join.relationship_path_to_relationships(
resource,
path ++ aggregate.relationship_path
)}
]
) do
{:ok, query} ->
binding =
AshPostgres.DataLayer.get_binding(
resource,
aggregate.relationship_path,
query,
[:left, :inner]
)
{:ok, query, Ecto.Query.dynamic(field(as(^binding), ^aggregate.field))}
{: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,
[_aggregate1, _aggregate2 | _rest],
false,
_source_binding
) do
{:ok, agg_query}
end
defp maybe_filter_subquery(
agg_query,
first_relationship,
relationship_path,
[aggregate],
true,
source_binding
) do
apply_agg_query(
agg_query,
aggregate,
relationship_path,
first_relationship,
source_binding
)
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 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,
on: true
)
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,
[join_relationship],
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,
on: true
)
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
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,
on: true
)
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: :exists}), do: false
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
if array_type?(resource, aggregate) || optimizable_first_aggregate?(resource, aggregate) do
false
else
true
end
else
true
end
end
@doc false
def optimizable_first_aggregate?(resource, %{
name: name,
kind: :first,
relationship_path: relationship_path
}) do
name in AshPostgres.DataLayer.Info.simple_join_first_aggregates(resource) &&
single_path?(resource, relationship_path)
end
def optimizable_first_aggregate?(_, _), do: false
defp array_type?(resource, aggregate) do
related = Ash.Resource.Info.related(resource, aggregate.relationship_path)
case Ash.Resource.Info.field(related, aggregate.field).type do
{:array, _} ->
false
_ ->
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: aggregate_field(aggregate, resource, relationship_path, query),
relationship_path: relationship_path,
resource: query.__ash_bindings__.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(
query.__ash_bindings__.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: aggregate_field(aggregate, resource, relationship_path, query),
relationship_path: relationship_path,
resource: query.__ash_bindings__.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(
query.__ash_bindings__.resource,
relationship_path
),
relationship_path,
binding,
true
)
question_marks = Enum.map(sort_expr, fn _ -> " ? " end)
distinct =
if Map.get(aggregate, :uniq?) do
"DISTINCT "
else
""
end
{:ok, expr} =
AshPostgres.Functions.Fragment.casted_new(
["array_agg(#{distinct}? ORDER BY #{question_marks})", field] ++ sort_expr
)
AshPostgres.Expr.dynamic_expr(query, expr, query.__ash_bindings__, false)
else
if Map.get(aggregate, :uniq?) do
Ecto.Query.dynamic(
[row],
fragment("array_agg(DISTINCT ?)", ^field)
)
else
Ecto.Query.dynamic(
[row],
fragment("array_agg(?)", ^field)
)
end
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, ^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: aggregate_field(aggregate, resource, relationship_path, query),
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 ->
if Map.get(aggregate, :uniq?) do
Ecto.Query.dynamic([row], count(^field, :distinct))
else
Ecto.Query.dynamic([row], count(^field))
end
: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
defp single_path?(_, []), do: true
defp single_path?(resource, [relationship | rest]) do
relationship = Ash.Resource.Info.relationship(resource, relationship)
relationship.type == :belongs_to && single_path?(relationship.destination, rest)
end
defp aggregate_field(aggregate, resource, _relationship_path, query) do
case Ash.Resource.Info.field(
resource,
aggregate.field || List.first(Ash.Resource.Info.primary_key(resource))
) do
%Ash.Resource.Calculation{calculation: {module, opts}} = calculation ->
calc_type =
AshPostgres.Types.parameterized_type(
calculation.type,
Map.get(calculation, :constraints, [])
)
AshPostgres.Expr.validate_type!(query, calc_type, "#{inspect(calculation.name)}")
{:ok, query_calc} =
Ash.Query.Calculation.new(
calculation.name,
module,
opts,
calculation.type,
Map.get(aggregate, :context, %{})
)
query_calc
other ->
other
end
end
end