Packages
ash_postgres
1.0.0-pre.3
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)
def add_aggregates(query, [], _, _), do: {:ok, query}
def add_aggregates(query, aggregates, resource, select?) do
resource = resource
query = AshPostgres.DataLayer.default_bindings(query, resource)
result =
Enum.reduce_while(aggregates, {:ok, query, []}, fn aggregate, {:ok, query, dynamics} ->
if aggregate.query && !aggregate.query.valid? do
{:halt, {:error, aggregate.query.errors}}
else
query_and_binding =
case AshPostgres.DataLayer.get_binding(
resource,
aggregate.relationship_path,
query,
:aggregate
) do
nil ->
relationship =
Ash.Resource.Info.relationship(resource, aggregate.relationship_path)
if relationship.type == :many_to_many do
subquery = aggregate_subquery(relationship, aggregate, query)
case AshPostgres.Join.join_all_relationships(
query,
nil,
[
{{:aggregate, aggregate.name, subquery},
AshPostgres.Join.relationship_path_to_relationships(
resource,
aggregate.relationship_path
)}
]
) do
{:ok, new_query} ->
{:ok,
{new_query,
AshPostgres.DataLayer.get_binding(
resource,
aggregate.relationship_path,
new_query,
:aggregate
)}}
{:error, error} ->
{:error, error}
end
else
subquery = aggregate_subquery(relationship, aggregate, query)
case AshPostgres.Join.join_all_relationships(
query,
nil,
[
{{:aggregate, aggregate.name, subquery},
AshPostgres.Join.relationship_path_to_relationships(
resource,
aggregate.relationship_path
)}
]
) do
{:ok, new_query} ->
{:ok,
{new_query,
AshPostgres.DataLayer.get_binding(
resource,
aggregate.relationship_path,
new_query,
:aggregate
)}}
{:error, error} ->
{:error, error}
end
end
binding ->
{:ok, {query, binding}}
end
case query_and_binding do
{:ok, {query, binding}} ->
query_with_aggregate_binding =
put_in(
query.__ash_bindings__.aggregates,
Map.put(query.__ash_bindings__.aggregates, aggregate.name, binding)
)
query_with_aggregate_defs =
put_in(
query_with_aggregate_binding.__ash_bindings__.aggregate_defs,
Map.put(
query_with_aggregate_binding.__ash_bindings__.aggregate_defs,
aggregate.name,
aggregate
)
)
new_query =
query_with_aggregate_defs
|> add_aggregate_to_subquery(resource, aggregate, binding)
if select? do
dynamic = select_dynamic(resource, query, aggregate)
{:cont, {:ok, new_query, [{aggregate.load, aggregate.name, dynamic} | dynamics]}}
else
{:cont, {:ok, new_query, dynamics}}
end
{:error, error} ->
{:halt, {:error, error}}
end
end
end)
case result do
{:ok, query, dynamics} ->
{:ok, add_aggregate_selects(query, dynamics)}
{:error, error} ->
{:error, error}
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)
query =
if query.select do
query
else
Ecto.Query.select_merge(query, %{})
end
{exprs, new_params, count} =
Enum.reduce(
in_body,
{[], Enum.reverse(query.select.params), Enum.count(query.select.params)},
fn {load, _, dynamic}, {exprs, params, count} ->
{expr, new_params, count} =
Ecto.Query.Builder.Dynamic.partially_expand(
:select,
query,
dynamic,
params,
count
)
{[{load, expr} | exprs], new_params, count}
end
)
query = %{
query
| select: %{
query.select
| expr: {:merge, [], [query.select.expr, {:%{}, [], Enum.reverse(exprs)}]}
}
}
add_aggregates_in_aggregates(query, in_aggregates, new_params, count)
end
defp add_aggregates_in_aggregates(query, [], params, _count),
do: %{query | select: %{query.select | params: Enum.reverse(params)}}
defp add_aggregates_in_aggregates(
%{select: %{expr: expr} = select} = query,
in_aggregates,
params,
count
) do
{exprs, new_params, _} =
Enum.reduce(
in_aggregates,
{[], params, count},
fn {_, name, dynamic}, {exprs, params, count} ->
{expr, new_params, count} =
Ecto.Query.Builder.Dynamic.partially_expand(
:select,
query,
dynamic,
params,
count
)
{[{name, expr} | exprs], new_params, count}
end
)
%{
query
| select: %{
select
| expr: {:merge, [], [expr, {:%{}, [], [aggregates: {:%{}, [], Enum.reverse(exprs)}]}]},
params: Enum.reverse(new_params)
}
}
end
def agg_subquery_for_lateral_join(current_binding, query, subquery, relationship) do
{dest_binding, dest_field} =
case relationship.type do
:many_to_many ->
{1, relationship.source_attribute_on_join_resource}
_ ->
{0, relationship.destination_attribute}
end
inner_sub =
if Map.get(relationship, :no_attributes?) do
subquery
else
from(destination in subquery,
where:
field(as(^dest_binding), ^dest_field) ==
field(parent_as(^current_binding), ^relationship.source_attribute)
)
end
from(sub in subquery(inner_sub), [])
|> AshPostgres.Join.set_join_prefix(query, relationship.destination)
end
defp select_dynamic(resource, query, aggregate) do
binding =
AshPostgres.DataLayer.get_binding(resource, aggregate.relationship_path, query, :aggregate)
type = AshPostgres.Types.parameterized_type(aggregate.type, [])
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 add_aggregate_to_subquery(query, resource, aggregate, binding) do
new_joins =
List.update_at(query.joins, binding - 1, fn join ->
aggregate_query =
if aggregate.authorization_filter do
{:ok, filtered} =
AshPostgres.DataLayer.filter(
join.source.from.source.query,
aggregate.authorization_filter,
Ash.Resource.Info.related(resource, aggregate.relationship_path)
)
filtered
else
join.source.from.source.query
end
{:ok, aggregate_query} =
AshPostgres.Aggregate.add_aggregates(
aggregate_query,
Map.values(aggregate.query.aggregates || %{}),
Ash.Resource.Info.related(resource, aggregate.relationship_path),
false
)
{:ok, aggregate_query} =
if aggregate.query && aggregate.query.filter do
AshPostgres.Join.join_all_relationships(
aggregate_query,
aggregate.query.filter
)
else
{:ok, aggregate_query}
end
new_aggregate_query = add_subquery_aggregate_select(aggregate_query, aggregate, resource)
put_in(join.source.from.source.query, new_aggregate_query)
end)
%{
query
| joins: new_joins
}
end
def used_aggregates(filter, relationship, 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: relationship.destination,
aggregates: %{},
calculations: %{},
public?: false
}
) do
{:ok, hydrated} ->
Ash.Filter.used_aggregates(hydrated)
_ ->
[]
end
end
)
end
def add_subquery_aggregate_select(query, %{kind: :first} = aggregate, _resource) do
query = AshPostgres.DataLayer.default_bindings(query, aggregate.resource)
key = aggregate.field
type = AshPostgres.Types.parameterized_type(aggregate.type, [])
field =
if aggregate.query && aggregate.query.sort && aggregate.query.sort != [] do
sort_expr = AshPostgres.Sort.order_to_ecto(aggregate.query.sort)
question_marks = Enum.map(sort_expr, fn _ -> " ? " end)
field = Ecto.Query.dynamic([{^0, row}], field(row, ^key))
{: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,
AshPostgres.Types.parameterized_type(aggregate.type, [])
)
else
Ecto.Query.dynamic(
[row],
fragment("array_agg(?)", field(row, ^key))
)
end
filtered =
if aggregate.query && aggregate.query.filter &&
not match?(%Ash.Filter{expression: nil}, aggregate.query.filter) do
expr =
AshPostgres.Expr.dynamic_expr(
query,
aggregate.query.filter,
query.__ash_bindings__,
false,
AshPostgres.Types.parameterized_type(aggregate.type, [])
)
Ecto.Query.dynamic(filter(^field, ^expr))
else
field
end
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, %{kind: :list} = aggregate, _resource) do
query = AshPostgres.DataLayer.default_bindings(query, aggregate.resource)
key = aggregate.field
type = AshPostgres.Types.parameterized_type(aggregate.type, [])
field =
if aggregate.query && aggregate.query.sort && aggregate.query.sort != [] do
sort_expr = AshPostgres.Sort.order_to_ecto(aggregate.query.sort)
question_marks = Enum.map(sort_expr, fn _ -> " ? " end)
field = Ecto.Query.dynamic([row], field(row, ^key))
{: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,
AshPostgres.Types.parameterized_type(aggregate.type, [])
)
else
Ecto.Query.dynamic(
[row],
fragment("array_agg(?)", field(row, ^key))
)
end
filtered =
if aggregate.query && aggregate.query.filter &&
not match?(%Ash.Filter{expression: nil}, aggregate.query.filter) do
expr =
AshPostgres.Expr.dynamic_expr(
query,
aggregate.query.filter,
query.__ash_bindings__,
false,
AshPostgres.Types.parameterized_type(aggregate.type, [])
)
Ecto.Query.dynamic(filter(^field, ^expr))
else
field
end
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, %{kind: kind} = aggregate, resource)
when kind in [:count, :sum] do
query = AshPostgres.DataLayer.default_bindings(query, aggregate.resource)
key = aggregate.field || List.first(Ash.Resource.Info.primary_key(resource))
type = AshPostgres.Types.parameterized_type(aggregate.type, [])
field =
case kind do
:count ->
Ecto.Query.dynamic([row], count(field(row, ^key)))
:sum ->
Ecto.Query.dynamic([row], sum(field(row, ^key)))
end
filtered =
if aggregate.query && aggregate.query.filter &&
not match?(%Ash.Filter{expression: nil}, aggregate.query.filter) do
expr =
AshPostgres.Expr.dynamic_expr(
query,
aggregate.query.filter,
query.__ash_bindings__,
false,
AshPostgres.Types.parameterized_type(aggregate.type, [])
)
Ecto.Query.dynamic(filter(^field, ^expr))
else
field
end
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 select_or_merge(query, aggregate_name, casted) do
query =
if query.select do
query
else
Ecto.Query.select(query, %{})
end
{expr, new_params, _} =
Ecto.Query.Builder.Dynamic.partially_expand(
:select,
query,
casted,
Enum.reverse(query.select.params),
Enum.count(query.select.params)
)
%{
query
| select: %{
query.select
| expr: {:merge, [], [query.select.expr, {:%{}, [], [{aggregate_name, expr}]}]},
params: Enum.reverse(new_params)
}
}
end
defp aggregate_subquery(%{type: :many_to_many} = relationship, aggregate, root_query) do
destination =
case AshPostgres.Join.maybe_get_resource_query(
relationship.destination,
relationship,
root_query
) do
{:ok, query} ->
query
_ ->
relationship.destination
end
join_relationship =
Ash.Resource.Info.relationship(relationship.source, relationship.join_relationship)
through =
case AshPostgres.Join.maybe_get_resource_query(
relationship.through,
join_relationship,
root_query
) do
{:ok, query} ->
query
_ ->
relationship.through
end
query =
from(destination in destination,
join: through in ^through,
as: ^1,
on:
field(through, ^relationship.destination_attribute_on_join_resource) ==
field(destination, ^relationship.destination_attribute),
group_by: field(through, ^relationship.source_attribute_on_join_resource)
)
query_tenant = aggregate.query && aggregate.query.tenant
root_tenant = root_query.prefix
if Ash.Resource.Info.multitenancy_strategy(relationship.destination) &&
(root_tenant ||
query_tenant) do
Ecto.Query.put_query_prefix(
query,
query_tenant || root_tenant || AshPostgres.DataLayer.Info.schema(relationship.destination)
)
else
%{
query
| prefix:
AshPostgres.DataLayer.Info.schema(relationship.destination) ||
AshPostgres.DataLayer.Info.repo(relationship.destination).config()[:default_prefix] ||
"public"
}
end
end
defp aggregate_subquery(relationship, aggregate, root_query) do
destination =
case AshPostgres.Join.maybe_get_resource_query(
relationship.destination,
relationship,
root_query
) do
{:ok, query} ->
query
_ ->
relationship.destination
end
query =
from(row in destination,
group_by: ^relationship.destination_attribute,
select: %{}
)
query_tenant = aggregate.query && aggregate.query.tenant
root_tenant = root_query.prefix
if Ash.Resource.Info.multitenancy_strategy(relationship.destination) &&
(root_tenant ||
query_tenant) do
Ecto.Query.put_query_prefix(
query,
query_tenant || root_tenant || AshPostgres.DataLayer.Info.schema(relationship.destination)
)
else
%{
query
| prefix:
AshPostgres.DataLayer.Info.schema(relationship.destination) ||
AshPostgres.DataLayer.Info.repo(relationship.destination).config()[:default_prefix] ||
"public"
}
end
end
end