mirror of
https://github.com/ash-project/ash_postgres.git
synced 2024-09-20 13:33:50 +12:00
a5b1bc8a3f
fixes: #127
706 lines
19 KiB
Elixir
706 lines
19 KiB
Elixir
defmodule AshPostgres.Join do
|
|
@moduledoc false
|
|
import Ecto.Query, only: [from: 2, subquery: 1]
|
|
|
|
alias Ash.Query.{BooleanExpression, Not, Ref}
|
|
|
|
@known_inner_join_operators [
|
|
Eq,
|
|
GreaterThan,
|
|
GreaterThanOrEqual,
|
|
In,
|
|
LessThanOrEqual,
|
|
LessThan,
|
|
NotEq
|
|
]
|
|
|> Enum.map(&Module.concat(Ash.Query.Operator, &1))
|
|
|
|
@known_inner_join_functions [
|
|
Ago,
|
|
Contains
|
|
]
|
|
|> Enum.map(&Module.concat(Ash.Query.Function, &1))
|
|
|
|
@known_inner_join_predicates @known_inner_join_functions ++ @known_inner_join_operators
|
|
|
|
def join_all_relationships(
|
|
query,
|
|
filter,
|
|
relationship_paths \\ nil,
|
|
path \\ [],
|
|
source \\ nil
|
|
) do
|
|
relationship_paths =
|
|
relationship_paths ||
|
|
filter
|
|
|> Ash.Filter.relationship_paths()
|
|
|> Enum.map(fn path ->
|
|
if can_inner_join?(path, filter) do
|
|
{:inner,
|
|
AshPostgres.Join.relationship_path_to_relationships(
|
|
filter.resource,
|
|
path
|
|
)}
|
|
else
|
|
{:left,
|
|
AshPostgres.Join.relationship_path_to_relationships(
|
|
filter.resource,
|
|
path
|
|
)}
|
|
end
|
|
end)
|
|
|
|
Enum.reduce_while(relationship_paths, {:ok, query}, fn
|
|
{_join_type, []}, {:ok, query} ->
|
|
{:cont, {:ok, query}}
|
|
|
|
{join_type, [relationship | rest_rels]}, {:ok, query} ->
|
|
source = source || relationship.source
|
|
|
|
current_path = path ++ [relationship]
|
|
|
|
current_join_type = join_type
|
|
|
|
if has_binding?(source, Enum.reverse(current_path), query, current_join_type) do
|
|
{:cont, {:ok, query}}
|
|
else
|
|
case join_relationship(
|
|
query,
|
|
relationship,
|
|
Enum.map(path, & &1.name),
|
|
current_join_type,
|
|
source,
|
|
filter
|
|
) do
|
|
{:ok, joined_query} ->
|
|
joined_query_with_distinct = add_distinct(relationship, join_type, joined_query)
|
|
|
|
case join_all_relationships(
|
|
joined_query_with_distinct,
|
|
filter,
|
|
[{join_type, rest_rels}],
|
|
current_path,
|
|
source
|
|
) do
|
|
{:ok, query} ->
|
|
{:cont, {:ok, query}}
|
|
|
|
{:error, error} ->
|
|
{:halt, {:error, error}}
|
|
end
|
|
|
|
{:error, error} ->
|
|
{:halt, {:error, error}}
|
|
end
|
|
end
|
|
end)
|
|
end
|
|
|
|
def relationship_path_to_relationships(resource, path, acc \\ [])
|
|
def relationship_path_to_relationships(_resource, [], acc), do: Enum.reverse(acc)
|
|
|
|
def relationship_path_to_relationships(resource, [relationship | rest], acc) do
|
|
relationship = Ash.Resource.Info.relationship(resource, relationship)
|
|
|
|
relationship_path_to_relationships(relationship.destination, rest, [relationship | acc])
|
|
end
|
|
|
|
def maybe_get_resource_query(
|
|
resource,
|
|
relationship,
|
|
root_query,
|
|
path \\ [],
|
|
bindings \\ nil,
|
|
start_binding \\ nil
|
|
) do
|
|
resource
|
|
|> Ash.Query.new(nil, base_filter?: false)
|
|
|> Ash.Query.set_context(%{data_layer: %{start_bindings_at: start_binding}})
|
|
|> Ash.Query.set_context((bindings || root_query.__ash_bindings__).context)
|
|
|> Ash.Query.set_context(relationship.context)
|
|
|> case do
|
|
%{valid?: true} = query ->
|
|
ash_query = query
|
|
|
|
initial_query = %{
|
|
AshPostgres.DataLayer.resource_to_query(resource, nil)
|
|
| prefix: Map.get(root_query, :prefix)
|
|
}
|
|
|
|
case Ash.Query.data_layer_query(query,
|
|
initial_query: initial_query
|
|
) do
|
|
{:ok, query} ->
|
|
query =
|
|
query
|
|
|> do_base_filter(
|
|
root_query,
|
|
ash_query,
|
|
resource,
|
|
path,
|
|
bindings
|
|
)
|
|
|> do_relationship_filter(
|
|
relationship.filter,
|
|
root_query,
|
|
ash_query,
|
|
resource,
|
|
path,
|
|
bindings
|
|
)
|
|
|
|
{:ok, query}
|
|
|
|
{:error, error} ->
|
|
{:error, error}
|
|
end
|
|
|
|
query ->
|
|
{:error, query}
|
|
end
|
|
end
|
|
|
|
defp do_relationship_filter(query, nil, _, _, _, _, _), do: query
|
|
|
|
defp do_relationship_filter(
|
|
query,
|
|
relationship_filter,
|
|
root_query,
|
|
ash_query,
|
|
resource,
|
|
path,
|
|
bindings
|
|
) do
|
|
filter =
|
|
resource
|
|
|> Ash.Filter.parse!(
|
|
relationship_filter,
|
|
ash_query.aggregates,
|
|
ash_query.calculations,
|
|
ash_query.context
|
|
)
|
|
|
|
dynamic =
|
|
if bindings do
|
|
filter = Ash.Filter.move_to_relationship_path(filter, path)
|
|
|
|
AshPostgres.Expr.dynamic_expr(root_query, filter, bindings, true)
|
|
else
|
|
AshPostgres.Expr.dynamic_expr(query, filter, query.__ash_bindings__, true)
|
|
end
|
|
|
|
{:ok, query} = join_all_relationships(query, filter)
|
|
from(row in query, where: ^dynamic)
|
|
end
|
|
|
|
defp do_base_filter(query, root_query, ash_query, resource, path, bindings) do
|
|
case Ash.Resource.Info.base_filter(resource) do
|
|
nil ->
|
|
query
|
|
|
|
filter ->
|
|
filter =
|
|
resource
|
|
|> Ash.Filter.parse!(
|
|
filter,
|
|
ash_query.aggregates,
|
|
ash_query.calculations,
|
|
ash_query.context
|
|
)
|
|
|
|
dynamic =
|
|
if bindings do
|
|
filter = Ash.Filter.move_to_relationship_path(filter, path)
|
|
|
|
AshPostgres.Expr.dynamic_expr(root_query, filter, bindings, true)
|
|
else
|
|
AshPostgres.Expr.dynamic_expr(query, filter, query.__ash_bindings__, true)
|
|
end
|
|
|
|
from(row in query, where: ^dynamic)
|
|
end
|
|
end
|
|
|
|
def set_join_prefix(join_query, query, resource) do
|
|
if Ash.Resource.Info.multitenancy_strategy(resource) == :context do
|
|
%{
|
|
join_query
|
|
| prefix: query.prefix || AshPostgres.DataLayer.Info.schema(resource) || "public"
|
|
}
|
|
else
|
|
%{
|
|
join_query
|
|
| prefix:
|
|
AshPostgres.DataLayer.Info.schema(resource) ||
|
|
AshPostgres.DataLayer.Info.repo(resource).config()[:default_prefix] ||
|
|
"public"
|
|
}
|
|
end
|
|
end
|
|
|
|
defp can_inner_join?(path, expr, seen_an_or? \\ false)
|
|
|
|
defp can_inner_join?(path, %{expression: expr}, seen_an_or?),
|
|
do: can_inner_join?(path, expr, seen_an_or?)
|
|
|
|
defp can_inner_join?(_path, expr, _seen_an_or?) when expr in [nil, true, false], do: true
|
|
|
|
defp can_inner_join?(path, %BooleanExpression{op: :and, left: left, right: right}, seen_an_or?) do
|
|
can_inner_join?(path, left, seen_an_or?) || can_inner_join?(path, right, seen_an_or?)
|
|
end
|
|
|
|
defp can_inner_join?(path, %BooleanExpression{op: :or, left: left, right: right}, _) do
|
|
can_inner_join?(path, left, true) && can_inner_join?(path, right, true)
|
|
end
|
|
|
|
defp can_inner_join?(
|
|
_,
|
|
%Not{},
|
|
_
|
|
) do
|
|
false
|
|
end
|
|
|
|
defp can_inner_join?(
|
|
search_path,
|
|
%struct{__operator__?: true, left: %Ref{relationship_path: relationship_path}},
|
|
seen_an_or?
|
|
)
|
|
when search_path == relationship_path and struct in @known_inner_join_predicates do
|
|
not seen_an_or?
|
|
end
|
|
|
|
defp can_inner_join?(
|
|
search_path,
|
|
%struct{__operator__?: true, right: %Ref{relationship_path: relationship_path}},
|
|
seen_an_or?
|
|
)
|
|
when search_path == relationship_path and struct in @known_inner_join_predicates do
|
|
not seen_an_or?
|
|
end
|
|
|
|
defp can_inner_join?(
|
|
search_path,
|
|
%struct{__function__?: true, arguments: arguments},
|
|
seen_an_or?
|
|
)
|
|
when struct in @known_inner_join_predicates do
|
|
if Enum.any?(arguments, &match?(%Ref{relationship_path: ^search_path}, &1)) do
|
|
not seen_an_or?
|
|
else
|
|
true
|
|
end
|
|
end
|
|
|
|
defp can_inner_join?(_, _, _), do: false
|
|
|
|
defp has_binding?(resource, candidate_path, %{__ash_bindings__: _} = query, type) do
|
|
Enum.any?(query.__ash_bindings__.bindings, fn
|
|
{_, %{path: path, source: source, type: ^type}} ->
|
|
Ash.SatSolver.synonymous_relationship_paths?(resource, path, candidate_path, source)
|
|
|
|
_ ->
|
|
false
|
|
end)
|
|
end
|
|
|
|
defp has_binding?(_, _, _, _), do: false
|
|
|
|
defp add_distinct(relationship, _join_type, joined_query) do
|
|
if !joined_query.__ash_bindings__.in_group? && relationship.cardinality == :many &&
|
|
!joined_query.distinct do
|
|
if joined_query.group_bys && joined_query.group_bys != [] do
|
|
joined_query
|
|
else
|
|
from(row in joined_query,
|
|
distinct: ^Ash.Resource.Info.primary_key(joined_query.__ash_bindings__.resource)
|
|
)
|
|
end
|
|
else
|
|
joined_query
|
|
end
|
|
end
|
|
|
|
defp join_relationship(
|
|
query,
|
|
relationship,
|
|
path,
|
|
join_type,
|
|
source,
|
|
filter
|
|
) do
|
|
case Map.get(query.__ash_bindings__.bindings, path) do
|
|
%{type: existing_join_type} when join_type != existing_join_type ->
|
|
raise "unreachable?"
|
|
|
|
nil ->
|
|
do_join_relationship(
|
|
query,
|
|
relationship,
|
|
path,
|
|
join_type,
|
|
source,
|
|
filter
|
|
)
|
|
|
|
_ ->
|
|
{:ok, query}
|
|
end
|
|
end
|
|
|
|
defp do_join_relationship(
|
|
query,
|
|
%{manual: {module, opts}} = relationship,
|
|
path,
|
|
kind,
|
|
source,
|
|
filter
|
|
) do
|
|
full_path = path ++ [relationship.name]
|
|
initial_ash_bindings = query.__ash_bindings__
|
|
|
|
binding_data = %{type: kind, path: full_path, source: source}
|
|
|
|
query = AshPostgres.DataLayer.add_binding(query, binding_data)
|
|
|
|
used_calculations =
|
|
Ash.Filter.used_calculations(
|
|
filter,
|
|
relationship.destination,
|
|
full_path
|
|
)
|
|
|
|
used_aggregates =
|
|
filter
|
|
|> AshPostgres.Aggregate.used_aggregates(
|
|
relationship.destination,
|
|
used_calculations,
|
|
full_path
|
|
)
|
|
|> Enum.map(fn aggregate ->
|
|
%{aggregate | load: aggregate.name}
|
|
end)
|
|
|
|
use_root_query_bindings? = Enum.empty?(used_aggregates)
|
|
|
|
root_bindings =
|
|
if use_root_query_bindings? do
|
|
query.__ash_bindings__
|
|
end
|
|
|
|
case maybe_get_resource_query(
|
|
relationship.destination,
|
|
relationship,
|
|
query,
|
|
full_path,
|
|
root_bindings
|
|
) do
|
|
{:error, error} ->
|
|
{:error, error}
|
|
|
|
{:ok, relationship_destination} ->
|
|
relationship_destination =
|
|
relationship_destination
|
|
|> Ecto.Queryable.to_query()
|
|
|> set_join_prefix(query, relationship.destination)
|
|
|
|
binding_kind = kind
|
|
|
|
current_binding =
|
|
Enum.find_value(initial_ash_bindings.bindings, 0, fn {binding, data} ->
|
|
if data.type == binding_kind && data.path == path do
|
|
binding
|
|
end
|
|
end)
|
|
|
|
relationship_destination =
|
|
case used_aggregates do
|
|
[] ->
|
|
relationship_destination
|
|
|
|
_ ->
|
|
subquery(relationship_destination)
|
|
end
|
|
|
|
case module.ash_postgres_join(
|
|
query,
|
|
opts,
|
|
current_binding,
|
|
initial_ash_bindings.current,
|
|
kind,
|
|
relationship_destination
|
|
) do
|
|
{:ok, query} ->
|
|
AshPostgres.Aggregate.add_aggregates(
|
|
query,
|
|
used_aggregates,
|
|
relationship.destination,
|
|
false
|
|
)
|
|
end
|
|
end
|
|
rescue
|
|
e in UndefinedFunctionError ->
|
|
if e.function == :ash_postgres_join do
|
|
reraise """
|
|
Cannot join to a manual relationship #{inspect(module)} that does not implement the `AshPostgres.ManualRelationship` behaviour.
|
|
""",
|
|
__STACKTRACE__
|
|
else
|
|
reraise e, __STACKTRACE__
|
|
end
|
|
end
|
|
|
|
defp do_join_relationship(
|
|
query,
|
|
%{type: :many_to_many} = relationship,
|
|
path,
|
|
kind,
|
|
source,
|
|
filter
|
|
) do
|
|
join_relationship =
|
|
Ash.Resource.Info.relationship(relationship.source, relationship.join_relationship)
|
|
|
|
join_path = Enum.reverse([join_relationship.name | path])
|
|
|
|
full_path = path ++ [relationship.name]
|
|
|
|
initial_ash_bindings = query.__ash_bindings__
|
|
|
|
binding_data = %{type: kind, path: full_path, source: source}
|
|
|
|
query =
|
|
query
|
|
|> AshPostgres.DataLayer.add_binding(%{
|
|
path: join_path,
|
|
type: :left,
|
|
source: source
|
|
})
|
|
|> AshPostgres.DataLayer.add_binding(binding_data)
|
|
|
|
used_calculations =
|
|
Ash.Filter.used_calculations(
|
|
filter,
|
|
relationship.destination,
|
|
full_path
|
|
)
|
|
|
|
used_aggregates =
|
|
filter
|
|
|> AshPostgres.Aggregate.used_aggregates(
|
|
relationship.destination,
|
|
used_calculations,
|
|
full_path
|
|
)
|
|
|> Enum.map(fn aggregate ->
|
|
%{aggregate | load: aggregate.name}
|
|
end)
|
|
|
|
use_root_query_bindings? = Enum.empty?(used_aggregates)
|
|
|
|
root_bindings =
|
|
if use_root_query_bindings? do
|
|
query.__ash_bindings__
|
|
end
|
|
|
|
with {:ok, relationship_through} <-
|
|
maybe_get_resource_query(
|
|
relationship.through,
|
|
join_relationship,
|
|
query,
|
|
join_path,
|
|
root_bindings
|
|
),
|
|
{:ok, relationship_destination} <-
|
|
maybe_get_resource_query(
|
|
relationship.destination,
|
|
relationship,
|
|
query,
|
|
path,
|
|
root_bindings
|
|
) do
|
|
relationship_through =
|
|
relationship_through
|
|
|> Ecto.Queryable.to_query()
|
|
|> set_join_prefix(query, relationship.through)
|
|
|
|
relationship_destination =
|
|
relationship_destination
|
|
|> Ecto.Queryable.to_query()
|
|
|> set_join_prefix(query, relationship.destination)
|
|
|
|
binding_kind = kind
|
|
|
|
current_binding =
|
|
Enum.find_value(initial_ash_bindings.bindings, 0, fn {binding, data} ->
|
|
if data.type == binding_kind && data.path == path do
|
|
binding
|
|
end
|
|
end)
|
|
|
|
relationship_destination =
|
|
case used_aggregates do
|
|
[] ->
|
|
relationship_destination
|
|
|
|
_ ->
|
|
subquery(relationship_destination)
|
|
end
|
|
|
|
query =
|
|
case kind do
|
|
:inner ->
|
|
from([{row, current_binding}] in query,
|
|
join: through in ^relationship_through,
|
|
as: ^initial_ash_bindings.current,
|
|
on:
|
|
field(row, ^relationship.source_attribute) ==
|
|
field(through, ^relationship.source_attribute_on_join_resource),
|
|
join: destination in ^relationship_destination,
|
|
as: ^(initial_ash_bindings.current + 1),
|
|
on:
|
|
field(destination, ^relationship.destination_attribute) ==
|
|
field(through, ^relationship.destination_attribute_on_join_resource)
|
|
)
|
|
|
|
_ ->
|
|
from([{row, current_binding}] in query,
|
|
left_join: through in ^relationship_through,
|
|
as: ^initial_ash_bindings.current,
|
|
on:
|
|
field(row, ^relationship.source_attribute) ==
|
|
field(through, ^relationship.source_attribute_on_join_resource),
|
|
left_join: destination in ^relationship_destination,
|
|
as: ^(initial_ash_bindings.current + 1),
|
|
on:
|
|
field(destination, ^relationship.destination_attribute) ==
|
|
field(through, ^relationship.destination_attribute_on_join_resource)
|
|
)
|
|
end
|
|
|
|
AshPostgres.Aggregate.add_aggregates(
|
|
query,
|
|
used_aggregates,
|
|
relationship.destination,
|
|
false
|
|
)
|
|
end
|
|
end
|
|
|
|
defp do_join_relationship(
|
|
query,
|
|
relationship,
|
|
path,
|
|
kind,
|
|
source,
|
|
filter
|
|
) do
|
|
full_path = path ++ [relationship.name]
|
|
initial_ash_bindings = query.__ash_bindings__
|
|
|
|
binding_data = %{type: kind, path: full_path, source: source}
|
|
|
|
query = AshPostgres.DataLayer.add_binding(query, binding_data)
|
|
|
|
used_calculations =
|
|
Ash.Filter.used_calculations(
|
|
filter,
|
|
relationship.destination,
|
|
full_path
|
|
)
|
|
|
|
used_aggregates =
|
|
filter
|
|
|> AshPostgres.Aggregate.used_aggregates(
|
|
relationship.destination,
|
|
used_calculations,
|
|
full_path
|
|
)
|
|
|> Enum.map(fn aggregate ->
|
|
%{aggregate | load: aggregate.name}
|
|
end)
|
|
|
|
use_root_query_bindings? = Enum.empty?(used_aggregates)
|
|
|
|
root_bindings =
|
|
if use_root_query_bindings? do
|
|
query.__ash_bindings__
|
|
end
|
|
|
|
case maybe_get_resource_query(
|
|
relationship.destination,
|
|
relationship,
|
|
query,
|
|
full_path,
|
|
root_bindings
|
|
) do
|
|
{:error, error} ->
|
|
{:error, error}
|
|
|
|
{:ok, relationship_destination} ->
|
|
relationship_destination =
|
|
relationship_destination
|
|
|> Ecto.Queryable.to_query()
|
|
|> set_join_prefix(query, relationship.destination)
|
|
|
|
binding_kind = kind
|
|
|
|
current_binding =
|
|
Enum.find_value(initial_ash_bindings.bindings, 0, fn {binding, data} ->
|
|
if data.type == binding_kind && data.path == path do
|
|
binding
|
|
end
|
|
end)
|
|
|
|
relationship_destination =
|
|
case used_aggregates do
|
|
[] ->
|
|
relationship_destination
|
|
|
|
_ ->
|
|
subquery(relationship_destination)
|
|
end
|
|
|
|
query =
|
|
case {kind, Map.get(relationship, :no_attributes?)} do
|
|
{:inner, true} ->
|
|
from([{row, current_binding}] in query,
|
|
join: destination in ^relationship_destination,
|
|
as: ^initial_ash_bindings.current
|
|
)
|
|
|
|
{_, true} ->
|
|
from([{row, current_binding}] in query,
|
|
left_join: destination in ^relationship_destination,
|
|
as: ^initial_ash_bindings.current
|
|
)
|
|
|
|
{:inner, _} ->
|
|
from([{row, current_binding}] in query,
|
|
join: destination in ^relationship_destination,
|
|
as: ^initial_ash_bindings.current,
|
|
on:
|
|
field(row, ^relationship.source_attribute) ==
|
|
field(destination, ^relationship.destination_attribute)
|
|
)
|
|
|
|
_ ->
|
|
from([{row, current_binding}] in query,
|
|
left_join: destination in ^relationship_destination,
|
|
as: ^initial_ash_bindings.current,
|
|
on:
|
|
field(row, ^relationship.source_attribute) ==
|
|
field(destination, ^relationship.destination_attribute)
|
|
)
|
|
end
|
|
|
|
AshPostgres.Aggregate.add_aggregates(
|
|
query,
|
|
used_aggregates,
|
|
relationship.destination,
|
|
false
|
|
)
|
|
end
|
|
end
|
|
end
|