Packages
ash
3.27.4
3.29.3
3.29.2
3.29.1
3.29.0
3.28.0
3.27.8
3.27.7
3.27.6
3.27.5
3.27.4
3.27.3
3.27.2
3.27.1
3.27.0
3.26.0
3.25.2
3.25.1
3.25.0
retired
3.24.7
3.24.6
3.24.5
3.24.4
3.24.3
3.24.2
3.24.1
3.24.0
3.23.1
3.23.0
3.22.2
3.22.1
3.22.0
3.21.3
3.21.2
3.21.1
3.21.0
3.20.0
3.19.3
3.19.2
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.0
3.15.0
3.14.1
3.14.0
retired
3.13.2
3.13.1
3.13.0
3.12.0
3.11.3
3.11.2
3.11.1
3.11.0
3.10.1
3.10.0
3.9.0
3.8.0
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
retired
3.6.3
retired
3.6.2
3.6.1
3.6.0
3.5.43
3.5.42
3.5.41
3.5.40
3.5.39
3.5.38
3.5.37
3.5.36
3.5.35
3.5.34
3.5.33
3.5.32
3.5.31
3.5.30
3.5.29
3.5.28
3.5.27
3.5.26
3.5.25
3.5.24
3.5.23
3.5.22
3.5.21
3.5.20
3.5.19
3.5.18
3.5.17
3.5.16
3.5.15
3.5.14
3.5.13
3.5.12
3.5.11
3.5.10
3.5.9
3.5.8
3.5.7
3.5.6
3.5.5
3.5.4
3.5.3
3.5.2
3.5.1
3.5.0
3.4.74
retired
3.4.73
3.4.72
3.4.71
3.4.70
3.4.69
3.4.68
3.4.67
3.4.66
3.4.65
3.4.64
3.4.63
3.4.62
3.4.61
3.4.60
3.4.59
3.4.58
3.4.57
3.4.56
3.4.55
3.4.54
3.4.53
3.4.52
3.4.51
3.4.50
3.4.49
3.4.48
3.4.47
3.4.46
3.4.45
3.4.44
3.4.43
3.4.42
3.4.41
3.4.40
3.4.39
3.4.38
3.4.37
3.4.36
3.4.35
3.4.34
3.4.33
3.4.32
3.4.31
3.4.30
3.4.29
3.4.28
3.4.27
3.4.26
3.4.25
3.4.24
3.4.23
3.4.22
3.4.21
3.4.20
3.4.19
3.4.18
3.4.17
3.4.16
3.4.15
3.4.14
3.4.13
3.4.12
3.4.11
3.4.10
3.4.9
3.4.8
3.4.7
3.4.6
3.4.5
3.4.4
3.4.3
3.4.2
3.4.1
3.4.0
3.3.3
3.3.2
3.3.1
3.3.0
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.16
3.0.15
3.0.14
3.0.13
3.0.12
3.0.11
3.0.10
3.0.9
3.0.8
3.0.7
3.0.6
3.0.5
3.0.4
3.0.3
3.0.2
3.0.1
3.0.0
3.0.0-rc.46
3.0.0-rc.45
3.0.0-rc.44
3.0.0-rc.43
3.0.0-rc.42
3.0.0-rc.41
3.0.0-rc.40
3.0.0-rc.39
3.0.0-rc.38
3.0.0-rc.37
3.0.0-rc.36
3.0.0-rc.35
3.0.0-rc.34
3.0.0-rc.33
3.0.0-rc.32
3.0.0-rc.31
3.0.0-rc.29
3.0.0-rc.28
3.0.0-rc.27
3.0.0-rc.26
3.0.0-rc.25
3.0.0-rc.24
3.0.0-rc.23
3.0.0-rc.22
3.0.0-rc.21
3.0.0-rc.20
3.0.0-rc.19
3.0.0-rc.18
3.0.0-rc.17
3.0.0-rc.16
3.0.0-rc.15
3.0.0-rc.14
3.0.0-rc.13
3.0.0-rc.12
3.0.0-rc.11
3.0.0-rc.10
3.0.0-rc.9
3.0.0-rc.8
3.0.0-rc.7
3.0.0-rc.6
3.0.0-rc.5
3.0.0-rc.4
3.0.0-rc.3
3.0.0-rc.1
3.0.0-rc.0
2.21.15
2.21.14
2.21.13
2.21.12
2.21.11
2.21.10
2.21.9
2.21.8
2.21.7
2.21.6
2.21.5
2.21.4
2.21.3
2.21.2
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.14
2.19.13
2.19.12
2.19.11
2.19.10
2.19.9
2.19.8
2.19.7
2.19.6
2.19.5
2.19.4
2.19.3
retired
2.19.2
retired
2.19.1
retired
2.19.0
retired
2.18.2
2.18.1
2.18.0
2.17.24
2.17.23
2.17.22
2.17.21
2.17.20
2.17.19
2.17.18
2.17.17
2.17.16
2.17.15
2.17.14
2.17.13
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.1
2.16.0
2.15.20
2.15.19
2.15.18
2.15.17
2.15.16
2.15.15
2.15.14
2.15.13
2.15.12
2.15.11
2.15.10
2.15.9
2.15.8
2.15.7
2.15.6
2.15.5
2.15.4
2.15.2
2.15.1
2.15.0
2.14.21
2.14.20
2.14.19
2.14.18
2.14.17
2.14.16
2.14.15
2.14.14
2.14.13
2.14.12
2.14.11
2.14.10
2.14.9
2.14.8
2.14.7
2.14.6
2.14.5
2.14.4
2.14.3
2.14.2
2.14.1
2.14.0
2.13.4
retired
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.11
2.11.10
2.11.9
2.11.8
2.11.7
2.11.6
2.11.5
2.11.4
2.11.3
2.11.2
2.11.1
2.11.0
2.11.0-rc.3
2.11.0-rc.2
2.11.0-rc.1
2.11.0-rc.0
2.10.2
2.10.1
2.10.0
2.9.29
2.9.28
2.9.27
2.9.26
2.9.25
2.9.24
2.9.23
2.9.22
2.9.21
2.9.20
2.9.19
2.9.18
2.9.17
2.9.16
2.9.15
2.9.14
2.9.13
2.9.12
2.9.11
2.9.10
2.9.9
2.9.8
2.9.7
2.9.6
2.9.5
2.9.4
2.9.3
2.9.2
2.9.1
2.9.0
2.8.1
2.8.0
2.7.1
2.7.0
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.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.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.5.0-rc.6
2.5.0-rc.5
2.5.0-rc.4
2.5.0-rc.3
2.5.0-rc.2
2.5.0-rc.1
2.5.0-rc.0
2.4.30
2.4.29
2.4.28
2.4.27
2.4.26
2.4.25
2.4.24
2.4.23
2.4.22
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.0
2.2.0
2.1.0
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
2.0.0-pre.8
2.0.0-pre.7
2.0.0-pre.6
2.0.0-pre.5
2.0.0-pre.4
2.0.0-pre.3
2.0.0-pre.2
2.0.0-pre.1
2.0.0-pre.0
1.53.3
1.53.2
1.53.0
1.52.0-rc.22
1.52.0-rc.21
1.52.0-rc.20
1.52.0-rc.19
1.52.0-rc.18
1.52.0-rc.17
1.52.0-rc.16
1.52.0-rc.15
1.52.0-rc.14
1.52.0-rc.13
1.52.0-rc.12
1.52.0-rc.11
1.52.0-rc.10
1.52.0-rc.9
1.52.0-rc.8
1.52.0-rc.7
1.52.0-rc.6
1.52.0-rc.5
1.52.0-rc.4
1.52.0-rc.3
1.52.0-rc.2
1.52.0-rc.1
1.52.0-rc.0
1.51.2
1.51.1
retired
1.51.0
1.50.21
1.50.20
1.50.19
1.50.18
1.50.17
1.50.16
1.50.15
1.50.14
1.50.13
1.50.12
1.50.11
1.50.10
1.50.9
1.50.8
1.50.7
1.50.6
1.50.5
1.50.4
1.50.3
1.50.2
1.50.1
1.50.0
1.49.0
1.48.0-rc.30
1.48.0-rc.29
1.48.0-rc.28
1.48.0-rc.27
1.48.0-rc.26
1.48.0-rc.25
1.48.0-rc.24
1.48.0-rc.23
1.48.0-rc.22
1.48.0-rc.21
1.48.0-rc.20
1.48.0-rc.19
1.48.0-rc.18
1.48.0-rc.17
1.48.0-rc.16
1.48.0-rc.15
1.48.0-rc.14
1.48.0-rc.13
1.48.0-rc.12
1.48.0-rc.11
1.48.0-rc.10
1.48.0-rc.9
1.48.0-rc.8
1.48.0-rc.7
1.48.0-rc.6
1.48.0-rc.5
1.48.0-rc.4
1.48.0-rc.3
1.48.0-rc.2
1.48.0-rc.1
1.48.0-rc.0
1.47.12
1.47.11
1.47.10
1.47.9
1.47.8
1.47.7
1.47.6
1.47.5
1.47.4
1.47.3
1.47.2
1.47.1
1.47.0
1.46.13
1.46.12
1.46.11
1.46.10
1.46.9
1.46.8
1.46.7
1.46.6
1.46.5
1.46.4
1.46.3
1.46.2
1.46.1
1.46.0
1.45.0-rc9
1.45.0-rc8
1.45.0-rc7
1.45.0-rc6
1.45.0-rc5
1.45.0-rc4
1.45.0-rc3
1.45.0-rc20
1.45.0-rc2
1.45.0-rc19
1.45.0-rc18
1.45.0-rc17
1.45.0-rc16
1.45.0-rc15
1.45.0-rc14
1.45.0-rc13
1.45.0-rc12
1.45.0-rc11
1.45.0-rc10
1.45.0-rc1
1.45.0-rc0
1.44.13
1.44.12
1.44.11
1.44.10
1.44.9
1.44.8
1.44.7
1.44.6
1.44.5
1.44.4
1.44.3
1.44.2
1.44.1
1.44.0
1.43.12
1.43.11
1.43.10
1.43.9
1.43.8
1.43.7
1.43.6
1.43.5
1.43.4
1.43.3
1.43.2
1.43.1
1.43.0
1.42.0
1.41.12
1.41.11
1.41.10
1.41.9
1.41.8
1.41.7
1.41.6
1.41.5
1.41.4
1.41.3
1.41.2
1.41.1
1.41.0
1.40.0
1.39.7
1.39.6
1.39.5
1.39.4
1.39.3
1.39.2
1.39.1
1.39.0
1.38.0
1.37.2
1.37.1
1.37.0
1.36.22
1.36.21
1.36.19
1.36.18
1.36.17
1.36.16
1.36.15
1.36.14
1.36.13
1.36.12
1.36.11
1.36.10
1.36.9
1.36.8
1.36.7
1.36.6
1.36.5
1.36.4
1.36.3
1.36.2
1.36.0
1.35.1
1.35.0
1.34.9
1.34.8
1.34.7
1.34.6
1.34.5
1.34.4
1.34.3
1.34.2
1.34.1
1.34.0
1.33.0
1.32.2
1.32.1
1.32.0
1.31.1
1.31.0
1.30.2
1.30.1
1.29.0-rc1
1.29.0-rc0
1.28.1
1.28.0
1.27.1
1.27.0
1.26.13
1.26.12
1.26.11
1.26.10
1.26.9
1.26.8
1.26.7
1.26.6
1.26.5
1.26.4
1.26.2
1.26.1
1.26.0
1.25.8
1.25.7
1.25.6
1.25.5
1.25.4
1.25.3
1.25.2
1.25.1
1.25.0
1.24.2
1.24.1
1.24.0
1.23.3
1.23.2
1.23.1
1.23.0
1.22.1
1.22.0
1.20.1
1.20.0
1.19.1
1.19.0
1.18.1
1.18.0
1.17.1
1.17.0
1.16.2
1.15.1
1.15.0
1.14.0
1.13.4
1.13.3
1.13.2
1.13.1
1.13.0
1.12.0
1.11.1
1.11.0
1.10.0
1.9.0
1.8.0
1.7.0
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.1
1.4.0
1.3.1
1.3.0
1.2.1
1.2.0
1.1.3
1.1.2
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
0.13.1
0.13.0
0.12.0
0.10.0
0.9.1
0.9.0
0.8.0
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.0
0.3.0
0.2.0
0.1.9
0.1.8
0.1.3
0.1.1
0.1.0
A declarative, extensible framework for building Elixir applications.
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/ash/actions/read/relationships.ex
# SPDX-FileCopyrightText: 2019 ash contributors <https://github.com/ash-project/ash/graphs/contributors>
#
# SPDX-License-Identifier: MIT
defmodule Ash.Actions.Read.Relationships do
@moduledoc false
require Ash.Query
import Ash.Expr
require Logger
def load([], _query, _lazy?, _reuse_values?) do
{:ok, []}
end
def load(record, query, lazy?, reuse_values?) when not is_list(record) do
case load([record], query, lazy?, reuse_values?) do
{:ok, [record]} -> {:ok, record}
{:error, error} -> {:error, error}
end
end
def load(records, %{load: load}, _lazy?, _reuse_values?) when load in [%{}, [], nil] do
{:ok, records}
end
def load(records, query, lazy?, reuse_values?) do
query.load
|> with_related_queries(query, records, lazy?)
|> fetch_related_records(query, records, reuse_values?)
|> attach_related_records(records)
end
defp attach_related_records(relationships_queries_and_related_records, records) do
Enum.reduce_while(relationships_queries_and_related_records, {:ok, records}, fn
{relationship, related_query, {:ok, related_records}}, {:ok, records} ->
{:cont,
{:ok, do_attach_related_records(records, relationship, related_records, related_query)}}
{relationship, _related_query, {:error, %Ash.Error.Forbidden{} = error}}, _ ->
if relationship.allow_forbidden_field? do
{:cont,
{:ok,
Enum.map(records, fn record ->
Map.put(record, relationship.name, %Ash.ForbiddenField{
type: :relationship,
field: relationship.name
})
end)}}
else
{:halt, {:error, Ash.Error.set_path(error, relationship.name)}}
end
{relationship, _related_query, {:error, error}}, _ ->
{:halt, {:error, Ash.Error.set_path(error, relationship.name)}}
{:__exception__, error, stacktrace}, _ ->
{:halt, {:error, Ash.Error.to_ash_error(error, stacktrace)}}
end)
end
defp fetch_related_records(batch, query, records, reuse_values?, acc \\ [])
defp fetch_related_records([], _query, _records, _reuse_values?, acc) do
Enum.map(acc, fn
{a, b, %Task{} = task} ->
{a, b, Task.await(task, :infinity)}
%Task{} = task ->
Task.await(task, :infinity)
other ->
other
end)
end
defp fetch_related_records([first | rest], query, records, reuse_values?, acc) do
result =
case first do
{relationship, {:lazy, query}} ->
{relationship, {:lazy, query},
lazy_related_records(records, relationship, query, Enum.empty?(rest), reuse_values?)}
{relationship, %{valid?: true} = related_query} ->
do_fetch_related_records(query, records, relationship, related_query, Enum.empty?(rest))
{relationship, %{errors: errors} = related_query} ->
{relationship, related_query, {:error, errors}}
end
fetch_related_records(rest, query, records, reuse_values?, [result | acc])
end
defp lazy_related_records(records, relationship, related_query, last?, reuse_values?) do
related_records_with_lazy_join_source =
if Ash.Resource.Info.primary_key_simple_equality?(relationship.destination) do
primary_key = Ash.Resource.Info.primary_key(relationship.source)
records
|> Enum.reduce(%{}, fn record, acc ->
record_pkey = Map.take(record, primary_key)
record
|> Map.get(relationship.name)
|> case do
%Ash.NotLoaded{} ->
[]
%Ash.ForbiddenField{} ->
[]
record_or_records ->
List.wrap(record_or_records)
end
|> Stream.with_index()
|> Stream.map(fn {record, index} ->
%{record | __order__: index}
end)
|> Enum.reduce(acc, fn related_record, acc ->
related_record_pkey =
Map.take(related_record, Ash.Resource.Info.primary_key(relationship.destination))
if Map.has_key?(acc, related_record_pkey) do
Map.update!(
acc,
related_record_pkey,
fn related_record ->
Map.update!(related_record, :__metadata__, fn metadata ->
Map.update!(metadata, :__lazy_join_sources__, &MapSet.put(&1, record_pkey))
end)
end
)
else
related_record =
Ash.Resource.set_metadata(related_record, %{
__lazy_join_sources__: MapSet.new([record_pkey])
})
Map.put(
acc,
related_record_pkey,
related_record
)
end
end)
end)
|> Stream.map(&elem(&1, 1))
|> Enum.sort_by(& &1.__order__)
else
primary_key = Ash.Resource.Info.primary_key(relationship.destination)
records
|> Enum.reduce([], fn record, acc ->
record_pkey = Map.take(record, primary_key)
record
|> Map.get(relationship.name)
|> case do
%Ash.NotLoaded{} ->
[]
%Ash.ForbiddenField{} ->
[]
record_or_records ->
List.wrap(record_or_records)
end
|> Enum.reduce(acc, fn related_record, acc ->
{acc, updated?} =
Enum.reduce(acc, {acc, false}, fn existing_record, {acc, updated?} ->
if relationship.destination.primary_key_matches?(existing_record, related_record) do
related_record =
Map.update!(related_record, :__metadata__, fn metadata ->
Map.update(
metadata,
:__lazy_join_sources__,
[record_pkey],
&[record_pkey | &1]
)
end)
{[related_record | acc], true}
else
{acc, updated?}
end
end)
if updated? do
acc
else
related_record =
Ash.Resource.set_metadata(related_record, %{
__lazy_join_sources__: MapSet.new([record_pkey])
})
[related_record | acc]
end
end)
|> Enum.reverse()
end)
end
Ash.Actions.Read.AsyncLimiter.async_or_inline(
related_query,
Ash.Context.to_opts(related_query.context),
last?,
fn ->
Ash.load(related_records_with_lazy_join_source, related_query,
lazy?: true,
reuse_values?:
reuse_values? || related_query.context[:private][:reuse_values?] || false,
domain: related_query.domain,
actor: related_query.context.private[:actor],
tenant: related_query.tenant,
authorize?: related_query.context.private[:authorize?]
)
end
)
end
defp with_related_queries(load, query, records, lazy?) do
Enum.map(load, fn {relationship_name, related_query} ->
lazy? = lazy? || related_query.context[:private][:lazy?]
if lazy? && Ash.Resource.loaded?(records, relationship_name, lists: :any) do
relationship = Ash.Resource.Info.relationship(query.resource, relationship_name)
related_query =
case related_query do
[] -> Ash.Query.new(relationship.destination)
query -> query
end
domain = Ash.Domain.Info.related_domain(related_query, relationship, query.domain)
{read_action_name, arguments} =
case related_query do
%Ash.Query{action: %{name: name}, arguments: arguments} ->
{name, arguments}
_ ->
read_action_name =
relationship.read_action ||
Ash.Resource.Info.primary_action!(relationship.destination, :read).name
{read_action_name, %{}}
end
{relationship,
{:lazy,
related_query
|> Map.put(
:domain,
Ash.Domain.Info.related_domain(related_query, relationship, query.domain)
)
|> Ash.Query.set_tenant(query.tenant)
|> Ash.Query.set_context(%{
private: %{async_limiter: query.context[:private][:async_limiter]},
shared: query.context[:shared] || %{}
})
|> Ash.Query.set_context(Map.take(query.context, [:shared]))
|> Ash.Query.for_read(
read_action_name,
arguments,
domain: domain,
authorize?: query.context[:private][:authorize?],
actor: query.context[:private][:actor],
tracer: query.context[:private][:tracer]
)}}
else
related_query(relationship_name, records, related_query, query)
end
end)
end
@doc false
def related_query(relationship_name, records, related_query, query) do
relationship = Ash.Resource.Info.relationship(query.resource, relationship_name)
{read_action, arguments} =
case related_query do
%Ash.Query{action: read_action, arguments: arguments} when not is_nil(read_action) ->
{read_action, arguments}
_ ->
read_action =
if relationship.read_action do
Ash.Resource.Info.action(relationship.destination, relationship.read_action)
else
Ash.Resource.Info.primary_action!(relationship.destination, :read)
end
{read_action, %{}}
end
domain = Ash.Domain.Info.related_domain(related_query, relationship, query.domain)
parent_stack =
[
query.resource | Ash.Actions.Read.parent_stack_from_context(query.context)
]
related_query =
related_query
|> Ash.Query.set_context(%{
private: %{
loading_relationships?: true,
async_limiter: query.context[:private][:async_limiter]
}
})
|> Ash.Query.set_tenant(query.tenant || related_query.tenant)
|> Ash.Query.set_context(Map.take(query.context, [:shared]))
|> Ash.Actions.Read.for_read(read_action, nil, arguments,
domain: domain,
authorize?: query.context[:private][:authorize?],
actor: query.context[:private][:actor],
tracer: query.context[:private][:tracer]
)
|> then(fn query ->
cond do
relationship.cardinality == :one && Map.get(relationship, :from_many?) ->
Ash.Query.limit(query, 1)
Map.get(relationship, :limit) ->
Ash.Query.limit(query, relationship.limit)
true ->
query
end
end)
|> apply_relationship_offset(relationship)
|> Ash.Query.sort(relationship.sort)
|> Ash.Query.default_sort(relationship.default_sort)
|> Ash.Query.do_filter(relationship.filter, parent_stack: parent_stack)
|> Ash.Query.set_context(relationship.context)
|> hydrate_refs(query.context[:private][:actor], relationship.source)
|> with_lateral_join_query(query, relationship, records)
if !related_query.context[:data_layer][:lateral_join_source] &&
related_query.distinct not in [[], nil] do
raise ArgumentError, message: "Cannot yet use `distinct` when loading related records"
end
{relationship, related_query}
end
defp with_lateral_join_query(related_query, source_query, relationship, records) do
if lateral_join?(related_query, source_query, relationship, records) do
lateral_join_source_path =
cond do
is_list(Map.get(relationship, :through)) ->
through_path = Map.get(relationship, :through)
expanded_relationships = expand_through_path(relationship.source, through_path)
Enum.reduce_while(
expanded_relationships,
{:ok, [], relationship.source},
fn relationship, {:ok, acc, current_resource} ->
entry_query =
if acc == [] do
clear_lateral_join_source(source_query)
else
current_resource
|> Ash.Query.new()
|> Ash.Query.set_tenant(source_query.tenant)
|> Ash.Query.set_context(%{
accessing_from: %{
source: current_resource,
name: relationship.name
},
shared: source_query.context[:shared] || %{}
})
|> Ash.Query.for_read(
relationship.read_action ||
Ash.Resource.Info.primary_action!(current_resource, :read).name,
%{},
authorize?: source_query.context[:private][:authorize?],
actor: source_query.context[:private][:actor],
tenant: source_query.tenant,
tracer: source_query.context[:private][:tracer],
domain: relationship.domain || related_query.domain
)
|> clear_lateral_join_source()
end
%{
source_attribute: source_attribute,
destination_attribute: destination_attribute
} = relationship
if acc != [] && source_query.context[:private][:authorize?] do
case Ash.can(
entry_query,
source_query.context[:private][:actor],
return_forbidden_error?: true,
pre_flight?: false,
alter_source?: true,
run_queries?: false,
base_query: entry_query
) do
{:ok, true} ->
entry =
{entry_query, source_attribute, destination_attribute, relationship}
{:cont, {:ok, acc ++ [entry], relationship.destination}}
{:ok, true, authorized_query} ->
entry =
{authorized_query, source_attribute, destination_attribute, relationship}
{:cont, {:ok, acc ++ [entry], relationship.destination}}
{:ok, false, error} ->
{:halt, {:error, Ash.Error.set_path(error, relationship.name)}}
{:error, error} ->
{:halt, {:error, Ash.Error.set_path(error, relationship.name)}}
end
else
entry = {entry_query, source_attribute, destination_attribute, relationship}
{:cont, {:ok, acc ++ [entry], relationship.destination}}
end
end
)
|> case do
{:ok, path, _} -> {:ok, path}
{:error, error} -> {:error, error}
end
relationship.type == :many_to_many ->
join_relationship =
Ash.Resource.Info.relationship(
source_query.resource,
relationship.join_relationship
)
through_query =
relationship.through
|> Ash.Query.do_filter(join_relationship.filter)
|> Ash.Query.set_context(%{
accessing_from: %{
source: relationship.source,
name: relationship.join_relationship
},
shared: source_query.context[:shared] || %{}
})
|> Ash.Query.for_read(
join_relationship.read_action ||
Ash.Resource.Info.primary_action!(relationship.through, :read).name,
%{},
authorize?: source_query.context[:private][:authorize?],
actor: source_query.context[:private][:actor],
tenant: source_query.tenant,
tracer: source_query.context[:private][:tracer],
domain: join_relationship.domain || related_query.domain
)
|> Ash.Query.select([
relationship.source_attribute_on_join_resource,
relationship.destination_attribute_on_join_resource
])
|> hydrate_refs(source_query.context[:private][:actor], relationship.source)
if source_query.context[:private][:authorize?] do
case Ash.can(
through_query,
source_query.context[:private][:actor],
return_forbidden_error?: true,
pre_flight?: false,
alter_source?: true,
run_queries?: false,
base_query: through_query
) do
{:ok, true} ->
{:ok,
[
{clear_lateral_join_source(source_query), relationship.source_attribute,
relationship.source_attribute_on_join_resource, relationship},
{clear_lateral_join_source(through_query),
relationship.destination_attribute_on_join_resource,
relationship.destination_attribute, join_relationship}
]}
{:ok, true, authorized_through_query} ->
{:ok,
[
{clear_lateral_join_source(source_query), relationship.source_attribute,
relationship.source_attribute_on_join_resource, relationship},
{clear_lateral_join_source(authorized_through_query),
relationship.destination_attribute_on_join_resource,
relationship.destination_attribute, join_relationship}
]}
{:ok, false, error} ->
{:error, Ash.Error.set_path(error, join_relationship.name)}
{:error, error} ->
{:error, Ash.Error.set_path(error, join_relationship.name)}
end
else
{:ok,
[
{clear_lateral_join_source(source_query), relationship.source_attribute,
relationship.source_attribute_on_join_resource, relationship},
{clear_lateral_join_source(through_query),
relationship.destination_attribute_on_join_resource,
relationship.destination_attribute, join_relationship}
]}
end
true ->
{:ok,
[
{clear_lateral_join_source(source_query), relationship.source_attribute,
relationship.destination_attribute, relationship}
]}
end
case lateral_join_source_path do
{:ok, lateral_join_source_path} ->
Ash.Query.set_context(related_query, %{
data_layer: %{
lateral_join_source: {records, lateral_join_source_path}
}
})
{:error, error} ->
Ash.Query.add_error(related_query, error)
end
else
related_query
end
end
defp hydrate_refs(query, actor, parent) do
query.filter
|> Ash.Expr.fill_template(
actor: actor,
tenant: query.to_tenant,
args: %{},
context: query.context
)
|> Ash.Filter.hydrate_refs(%{
resource: query.resource,
parent_stack: [parent],
public?: false
})
|> case do
{:ok, hydrated} ->
%{query | filter: hydrated}
{:error, error} ->
Ash.Query.add_error(query, error)
end
end
defp clear_lateral_join_source(source_query) do
Ash.Query.unset(source_query, [:load, :select, :sort])
end
defp do_fetch_related_records(
query,
records,
%{manual: {module, opts}} = relationship,
related_query,
last?
) do
parent_stack = [
query.resource | Ash.Actions.Read.parent_stack_from_context(query.context)
]
Ash.Actions.Read.AsyncLimiter.async_or_inline(
related_query,
Ash.Context.to_opts(related_query.context),
last?,
fn ->
result =
Ash.Resource.ManualRelationship.load(
module,
records,
opts,
%Ash.Resource.ManualRelationship.Context{
relationship: relationship,
source_context: query.context,
query:
related_query
|> Ash.Query.sort(relationship.sort)
|> Ash.Query.do_filter(relationship.filter)
|> Map.put(:load, [])
|> Ash.Query.set_context(%{
accessing_from: %{source: relationship.source, name: relationship.name},
parent_stack: parent_stack
}),
actor: related_query.context[:private][:actor],
authorize?: related_query.context[:private][:authorize?],
tracer: related_query.context[:private][:tracer],
domain: related_query.domain,
tenant: related_query.tenant
}
)
|> case do
{:ok, result} -> {:ok, result}
{:error, error} -> {:error, error}
end
|> case do
{:ok, result_records} ->
result_records
|> normalize_manual_results(records, relationship)
|> Enum.with_index()
|> Enum.flat_map(fn {value, index} ->
value
|> List.wrap()
|> Enum.map(&Ash.Resource.put_metadata(&1, :manual_key, index))
end)
|> Ash.load(related_query,
domain: related_query.domain,
actor: related_query.context[:private][:actor],
authorize?: related_query.context[:private][:authorize?],
tenant: related_query.tenant
)
|> case do
{:ok, results} ->
{:ok, regroup_manual_results(results, relationship, Enum.count(records))}
{:error, error} ->
{:error, error}
end
{:error, error} ->
{:error, error}
end
{relationship, related_query, result}
end
)
end
defp do_fetch_related_records(
query,
records,
%{no_attributes?: true} = relationship,
related_query,
last?
) do
parent_stack = [
query.resource | Ash.Actions.Read.parent_stack_from_context(query.context)
]
# Check if we need per-record iteration (parent expressions without lateral join support)
has_parent_expr? = has_parent_expr?(relationship, query.context, query.domain)
through_resources =
case Map.get(relationship, :through) do
nil ->
[]
through when is_list(through) ->
expanded = expand_through_path(relationship.source, through)
Enum.map(expanded, & &1.destination)
through ->
[through]
end
resources =
([relationship.source] ++ through_resources ++ [relationship.destination])
|> Enum.reject(&is_nil/1)
can_lateral_join? =
Ash.DataLayer.data_layer_can?(relationship.source, {:lateral_join, resources})
needs_per_record_iteration? = has_parent_expr? && !can_lateral_join?
Ash.Actions.Read.AsyncLimiter.async_or_inline(
related_query,
Ash.Context.to_opts(related_query.context),
last?,
fn ->
result =
if needs_per_record_iteration? do
# Iterate per-record, resolving parent expressions for each
do_per_record_no_attributes_load(
records,
relationship,
related_query,
parent_stack,
query
)
else
related_query
|> select_destination_attribute(relationship)
|> Ash.Query.set_context(%{
accessing_from: %{source: relationship.source, name: relationship.name},
parent_stack: parent_stack
})
|> then(fn query ->
if relationship.cardinality == :one && Map.get(relationship, :from_many?) do
Ash.Query.limit(query, 1)
else
query
end
end)
|> apply_relationship_offset(relationship)
|> Ash.Actions.Read.unpaginated_read(nil,
authorize_with: relationship.authorize_read_with
)
end
{relationship, related_query, result}
end
)
end
defp do_fetch_related_records(
query,
_records,
relationship,
%{context: %{data_layer: %{lateral_join_source: {_, _}}}} = related_query,
last?
) do
parent_stack = [
query.resource | Ash.Actions.Read.parent_stack_from_context(query.context)
]
Ash.Actions.Read.AsyncLimiter.async_or_inline(
related_query,
Ash.Context.to_opts(related_query.context),
last?,
fn ->
result =
related_query
|> select_destination_attribute(relationship)
|> Ash.Query.set_context(%{
accessing_from: %{source: relationship.source, name: relationship.name},
parent_stack: parent_stack
})
|> Ash.Actions.Read.read_and_return_unpaged(nil,
authorize_with: relationship.authorize_read_with
)
{relationship, related_query, result}
end
)
end
defp do_fetch_related_records(
query,
records,
%{type: :many_to_many} = relationship,
related_query,
last?
) do
record_ids =
Enum.map(records, fn record ->
Map.get(record, relationship.source_attribute)
end)
join_relationship =
Ash.Resource.Info.relationship(relationship.source, relationship.join_relationship)
join_query =
relationship.through
|> Ash.Query.filter(^ref(relationship.source_attribute_on_join_resource) in ^record_ids)
|> Ash.Query.do_filter(join_relationship.filter)
|> Ash.Query.set_context(%{
accessing_from: %{source: relationship.source, name: relationship.join_relationship},
parent_stack: [query.resource | Ash.Actions.Read.parent_stack_from_context(query.context)],
shared: query.context[:shared] || %{}
})
|> Ash.Query.select(
[
relationship.source_attribute_on_join_resource,
relationship.destination_attribute_on_join_resource
] ++ Ash.Resource.Info.primary_key(join_relationship.destination)
)
|> Ash.Query.for_read(
join_relationship.read_action ||
Ash.Resource.Info.primary_action!(relationship.through, :read).name,
%{},
authorize?: related_query.context[:private][:authorize?],
actor: related_query.context[:private][:actor],
tracer: related_query.context[:private][:tracer],
tenant: related_query.tenant,
domain:
Ash.Domain.Info.related_domain(
related_query,
join_relationship,
related_query.domain
)
)
|> Ash.Query.sort(join_relationship.sort)
|> Ash.Query.set_context(join_relationship.context)
|> then(fn q ->
if join_relationship.limit do
Ash.Query.limit(q, join_relationship.limit)
else
q
end
end)
parent_stack = [
query.resource | Ash.Actions.Read.parent_stack_from_context(query.context)
]
Ash.Actions.Read.AsyncLimiter.async_or_inline(
related_query,
Ash.Context.to_opts(related_query.context),
last?,
fn ->
case Ash.Actions.Read.unpaginated_read(join_query, nil) do
{:ok, join_records} ->
{join_id_mapping, destination_ids} =
Enum.reduce(join_records, {%{}, MapSet.new()}, fn join_record,
{mapping, destination_ids} ->
destination_value =
Map.get(join_record, relationship.destination_attribute_on_join_resource)
source_value =
Map.get(join_record, relationship.source_attribute_on_join_resource)
new_destination_ids = MapSet.put(destination_ids, destination_value)
new_mapping =
Map.update(
mapping,
destination_value,
[
source_value
],
&[source_value | &1]
)
{new_mapping, new_destination_ids}
end)
related_query =
if related_query.page do
if Ash.Actions.Sort.sorting_on_identity?(related_query) do
related_query
else
Ash.Query.sort(
related_query,
Ash.Resource.Info.primary_key(related_query.resource)
)
end
else
related_query
end
related_query
|> select_destination_attribute(relationship)
|> Ash.Query.sort(relationship.sort)
|> Ash.Query.do_filter(relationship.filter)
|> Ash.Query.filter(^ref(relationship.destination_attribute) in ^destination_ids)
|> Ash.Query.set_context(%{
accessing_from: %{source: relationship.source, name: relationship.name},
parent_stack: parent_stack
})
|> Map.put(:page, nil)
|> Ash.Actions.Read.unpaginated_read(nil,
authorize_with: relationship.authorize_read_with
)
|> case do
{:ok, records} ->
{relationship, related_query,
{:ok,
Enum.flat_map(records, fn record ->
Enum.map(
join_id_mapping[Map.get(record, relationship.destination_attribute)] || [],
fn lateral_join_source ->
Map.put(
record,
:__lateral_join_source__,
lateral_join_source
)
end
)
end)}}
{:error, error} ->
{relationship, related_query, {:error, error}}
end
{:error, error} ->
{relationship, related_query, {:error, error}}
end
end
)
end
defp do_fetch_related_records(query, records, relationship, related_query, last?) do
destination_attributes = Enum.map(records, &Map.get(&1, relationship.source_attribute))
parent_stack = [
query.resource | Ash.Actions.Read.parent_stack_from_context(query.context)
]
Ash.Actions.Read.AsyncLimiter.async_or_inline(
related_query,
Ash.Context.to_opts(related_query.context),
last?,
fn ->
result =
related_query
|> select_destination_attribute(relationship)
|> Ash.Query.filter(^ref(relationship.destination_attribute) in ^destination_attributes)
|> Ash.Query.unset([:limit, :offset, :distinct, :distinct_sort, :page])
|> Ash.Query.set_context(%{
accessing_from: %{source: relationship.source, name: relationship.name},
parent_stack: parent_stack
})
|> Ash.Actions.Read.unpaginated_read(nil,
authorize_with: relationship.authorize_read_with
)
{relationship, related_query, result}
end
)
end
defp do_per_record_no_attributes_load(
records,
relationship,
related_query,
parent_stack,
_source_query
) do
primary_key = Ash.Resource.Info.primary_key(relationship.source)
records
|> Enum.reduce_while({:ok, []}, fn record, {:ok, acc} ->
# Resolve parent expressions using this record's values
resolved_filter =
if related_query.filter do
resolve_parent_in_filter(related_query.filter, record, relationship.source)
else
nil
end
per_record_query =
related_query
|> Map.put(:filter, resolved_filter)
|> select_destination_attribute(relationship)
|> Ash.Query.set_context(%{
accessing_from: %{source: relationship.source, name: relationship.name},
parent_stack: parent_stack
})
|> then(fn query ->
if relationship.cardinality == :one && Map.get(relationship, :from_many?) do
Ash.Query.limit(query, 1)
else
query
end
end)
|> apply_relationship_offset(relationship)
case Ash.Actions.Read.unpaginated_read(per_record_query, nil,
authorize_with: relationship.authorize_read_with
) do
{:ok, results} ->
# Tag results with the source record's primary key
tagged_results =
Enum.map(results, fn result ->
Map.put(result, :__lateral_join_source__, Map.take(record, primary_key))
end)
{:cont, {:ok, tagged_results ++ acc}}
{:error, error} ->
{:halt, {:error, error}}
end
end)
end
defp resolve_parent_in_filter(filter, parent_record, source_resource) do
Ash.Filter.map(filter, fn
%Ash.Query.Parent{expr: expr} ->
case Ash.Expr.eval(expr, record: parent_record, resource: source_resource) do
{:ok, value} -> value
_ -> nil
end
other ->
other
end)
end
defp regroup_manual_results(records, %{cardinality: :many}, count) do
indexed_records =
Enum.group_by(records, & &1.__metadata__.manual_key, &delete_manual_key/1)
Enum.map(0..max(count - 1, 0), fn index ->
Map.get(indexed_records, index, [])
end)
end
defp regroup_manual_results(records, %{cardinality: :one}, count) do
indexed_records =
Map.new(records, fn record ->
{record.__metadata__.manual_key, delete_manual_key(record)}
end)
Enum.map(0..max(count - 1, 0), fn index ->
Map.get(indexed_records, index)
end)
end
defp delete_manual_key(record) do
Map.update!(record, :__metadata__, &Map.delete(&1, :manual_key))
end
defp normalize_manual_results(result_records, [%resource{} | _] = records, relationship)
when is_map(result_records) do
default =
case relationship.cardinality do
:one ->
nil
:many ->
[]
end
if Ash.Resource.Info.primary_key_simple_equality?(resource) do
pkey = Ash.Resource.Info.primary_key(resource)
single_match? =
case pkey do
[_] -> true
_ -> false
end
Enum.map(records, fn record ->
value =
if single_match? do
case Map.fetch(result_records, Map.get(record, Enum.at(pkey, 0))) do
{:ok, value} -> {:ok, value}
:error -> Map.fetch(result_records, Map.take(record, pkey))
end
else
Map.fetch(result_records, Map.take(record, pkey))
end
case value do
{:ok, result} ->
result
:error ->
default
end
end)
else
pkey = Ash.Resource.Info.primary_key(resource)
case pkey do
[pkey_key] ->
Enum.map(records, fn record ->
pkey_values = Map.take(record, pkey)
value =
Enum.find_value(result_records, fn {key, value} ->
if is_map(key) do
if resource.primary_key_matches?(key, pkey_values) do
{:ok, value}
end
else
if resource.primary_key_matches?(%{pkey_key => key}, pkey_values) do
{:ok, value}
end
end
end) || :error
case value do
{:ok, result} ->
result
:error ->
default
end
end)
_pkeys ->
Enum.map(records, fn record ->
pkey_values = Map.take(record, pkey)
value =
Enum.find_value(result_records, fn {key, value} ->
if resource.primary_key_matches?(key, pkey_values) do
{:ok, value}
end
end) || :error
case value do
{:ok, result} ->
result
:error ->
default
end
end)
end
end
end
defp normalize_manual_results(result_records, _records, _relationship) do
result_records
end
defp select_destination_attribute(related_query, relationship) do
if Map.get(relationship, :no_attributes?) ||
!Ash.Resource.Info.attribute(
relationship.destination,
relationship.destination_attribute
) do
related_query
else
Ash.Query.ensure_selected(related_query, [relationship.destination_attribute])
end
end
defp do_attach_related_records(
[%resource{} | _] = records,
relationship,
related_records,
{:lazy, _related_query}
) do
pkey = Ash.Resource.Info.primary_key(resource)
pkey_to_key = pkey_normalizer(resource, pkey)
if pkey_to_key do
records_by_pkey =
Enum.reduce(related_records, %{}, fn related, acc ->
Enum.reduce(related.__metadata__.__lazy_join_sources__, acc, fn source, acc ->
Map.update(acc, pkey_to_key.(source), [related], &[related | &1])
end)
end)
Enum.map(records, fn record ->
record_pkey = pkey_to_key.(Map.take(record, pkey))
related = Enum.reverse(Map.get(records_by_pkey, record_pkey, []))
related =
case relationship.cardinality do
:many ->
related
:one ->
case related do
[related_record] ->
related_record
[] ->
nil
[related_record | _] ->
# 4.0
Logger.warning("""
Got more than one result while loading relationship `#{inspect(relationship.source)}.#{relationship.name}`.
In the future this will be an error. If you have a `has_one` relationship that could produce multiple
related records, you must specify `from_many? true` on the relationship, *or* specify a `sort` (which
implicitly sets `from_many?` to `true`.
""")
related_record
end
end
case Map.get(record, relationship.name) do
%Ash.ForbiddenField{} -> record
%Ash.NotLoaded{} -> record
_ -> Map.put(record, relationship.name, related)
end
end)
else
Enum.map(records, fn record ->
related =
Enum.filter(related_records, fn related_record ->
Enum.any?(
related_record.__metadata__.__lazy_join_sources__,
&resource.primary_key_matches?(&1, record)
)
end)
related =
case relationship.cardinality do
:many ->
related
:one ->
case related do
[related_record] ->
related_record
[] ->
nil
[related_record | _] ->
# 4.0
Logger.warning("""
Got more than one result while loading relationship `#{inspect(relationship.source)}.#{relationship.name}`.
In the future this will be an error. If you have a `has_one` relationship that could produce multiple
related records, you must specify `from_many? true` on the relationship, *or* specify a `sort` (which
implicitly sets `from_many?` to `true`.
""")
related_record
end
end
case Map.get(record, relationship.name) do
%Ash.ForbiddenField{} -> record
%Ash.NotLoaded{} -> record
_ -> Map.put(record, relationship.name, related)
end
end)
end
end
defp do_attach_related_records(
[_ | _] = records,
relationship,
%Ash.Page.Unpaged{} = unpaged,
%{context: %{data_layer: %{lateral_join_source: {_records, lateral_join_source_path}}}} =
related_query
) do
%Ash.Page.Unpaged{
related_records: related_records,
opts: opts
} = unpaged
attach_fun =
if relationship.cardinality == :many do
fn record, relationship_name, value ->
count_key =
Ash.Actions.Read.paginated_relationship_count_aggregate_name(relationship.name)
# Retrieve the count (if present) while deleting it from the record aggregates
{count, record} = pop_in(record.aggregates[count_key])
# We scope the lateral join to the specific record, so that next runs of rerun
# just fetch the entries related to this record
related_query =
Ash.Query.set_context(related_query, %{
data_layer: %{lateral_join_source: {[record], lateral_join_source_path}}
})
page =
Ash.Actions.Read.to_page(
value,
related_query.action,
count,
related_query.sort,
related_query,
related_query,
opts
)
attach_related(record, relationship_name, page)
end
else
&attach_related/3
end
attach_lateral_join_related_records(
records,
relationship,
related_records,
attach_fun
)
end
defp do_attach_related_records(
[_ | _] = records,
relationship,
related_records,
%{context: %{data_layer: %{lateral_join_source: {_, _}}}}
) do
attach_lateral_join_related_records(records, relationship, related_records)
end
defp do_attach_related_records(
records,
%{type: :many_to_many} = relationship,
related_records,
related_query
) do
do_attach_related_records(
records,
relationship,
related_records,
Ash.Query.set_context(related_query, %{data_layer: %{lateral_join_source: {nil, nil}}})
)
|> Enum.map(fn row ->
Map.update!(row, relationship.name, fn related_records ->
apply_runtime_query_operations(
row,
relationship,
related_records,
related_query
)
end)
end)
end
defp do_attach_related_records(
records,
%{manual: {_module, _opts}} = relationship,
values,
_related_query
) do
default =
case relationship.cardinality do
:one -> nil
:many -> []
end
records
|> Enum.zip_with(values, fn
record, nil ->
Map.put(record, relationship.name, default)
record, value ->
Map.put(record, relationship.name, value)
end)
end
defp do_attach_related_records(
records,
%{no_attributes?: true} = relationship,
related_records,
_related_query
) do
# Check if related records have __lateral_join_source__ SET (not just the key existing)
# All Ash resource structs have the __lateral_join_source__ key defined,
# but the value is nil unless it was actually set by a lateral join operation
has_lateral_join_source? =
case related_records do
[first | _] -> not is_nil(first.__lateral_join_source__)
_ -> false
end
if has_lateral_join_source? do
# Use lateral join attachment logic
attach_lateral_join_related_records(records, relationship, related_records)
else
# Original behavior: assign all related records to all source records
Enum.map(records, fn record ->
case relationship.cardinality do
:many ->
Map.put(record, relationship.name, related_records)
:one ->
related_record =
case related_records do
[related_record] ->
related_record
[] ->
nil
[related_record | _] ->
# 4.0
Logger.warning("""
Got more than one result while loading relationship `#{inspect(relationship.source)}.#{relationship.name}`.
In the future this will be an error. If you have a `has_one` relationship that could produce multiple
related records, you must specify `from_many? true` on the relationship, *or* specify a `sort` (which
implicitly sets `from_many?` to `true`.
""")
related_record
end
Map.put(record, relationship.name, related_record)
end
end)
end
end
defp do_attach_related_records(records, relationship, related_records, related_query) do
attribute = Ash.Resource.Info.attribute(relationship.source, relationship.source_attribute)
to_key = key_fn_for_type(attribute.type)
related =
if to_key do
Enum.group_by(related_records, &to_key.(Map.get(&1, relationship.destination_attribute)))
else
related_records
end
default =
if relationship.cardinality == :one do
nil
else
[]
end
if to_key do
Enum.map(records, fn record ->
value = to_key.(Map.get(record, relationship.source_attribute))
if relationship.cardinality == :many do
Map.put(
record,
relationship.name,
apply_runtime_query_operations(
record,
relationship,
Map.get(related, value) || default,
related_query
)
)
else
related_records =
case Map.fetch(related, value) do
:error ->
[]
{:ok, records} ->
List.wrap(records)
end
related_records =
apply_runtime_query_operations(
record,
relationship,
related_records,
related_query
)
related_record =
case related_records do
[] ->
nil
[single] ->
single
[single | _] ->
# 4.0
Logger.warning("""
Got more than one result while loading relationship `#{inspect(relationship.source)}.#{relationship.name}`.
In the future this will be an error. If you have a `has_one` relationship that could produce multiple
related records, you must specify `from_many? true` on the relationship, *or* specify a `sort` (which
implicitly sets `from_many?` to `true`.
""")
single
end
Map.put(
record,
relationship.name,
related_record
)
end
end)
else
Enum.map(records, fn record ->
value = Map.get(record, relationship.source_attribute)
if relationship.cardinality == :many do
related
|> Enum.filter(fn result ->
destination_value = Map.get(result, relationship.destination_attribute)
Ash.Type.equal?(attribute.type, value, destination_value)
end)
|> then(fn result ->
result =
apply_runtime_query_operations(record, relationship, {:ok, result}, related_query)
put_result(record, result, relationship.name, default)
end)
else
related
|> Enum.find_value(:error, fn result ->
destination_value = Map.get(result, relationship.destination_attribute)
if Ash.Type.equal?(attribute.type, value, destination_value) do
{:ok, result}
end
end)
|> then(fn result ->
result =
apply_runtime_query_operations(record, relationship, result, related_query)
put_result(record, result, relationship.name, default)
end)
end
end)
end
end
defp apply_runtime_query_operations(record, relationship, {:ok, value}, related_query) do
{:ok, apply_runtime_query_operations(record, relationship, value, related_query)}
end
defp apply_runtime_query_operations(_record, _relationship, :error, _related_query), do: :error
defp apply_runtime_query_operations(_record, _relationship, empty, _related_query)
when empty in [nil, []],
do: empty
defp apply_runtime_query_operations(record, relationship, value, related_query)
when not is_list(value) do
record |> apply_runtime_query_operations(relationship, [value], related_query) |> Enum.at(0)
end
defp apply_runtime_query_operations(record, relationship, value, related_query) do
value
|> apply_runtime_offset(related_query)
|> apply_runtime_limit(related_query)
|> apply_runtime_pagination(record, relationship, related_query)
end
defp apply_runtime_pagination(
value,
source_record,
relationship,
%{page: page_opts} = related_query
)
when not is_nil(page_opts) do
pagination_type =
cond do
related_query.action.pagination.keyset? && (page_opts[:before] || page_opts[:after]) ->
:keyset
page_opts[:offset] ->
:offset
related_query.action.pagination.offset? && related_query.action.pagination.keyset? ->
:offset
related_query.action.pagination.offset? ->
:offset
true ->
:keyset
end
count = Map.get(source_record.aggregates, "__paginated_#{relationship.name}_count__")
limit =
page_opts[:limit] ||
related_query.action.pagination.default_limit ||
related_query.action.pagination.max_page_size || 250
value =
if pagination_type == :keyset && (page_opts[:before] || page_opts[:after]) do
after_or_before =
if page_opts[:before] do
:before
else
:after
end
# Apparently dialyzer hated me trying to write this as a *with* statement???
# giving up and making it a case statement because I don't have time for this shit
case Ash.Page.Keyset.filter(
related_query,
page_opts[:before] || page_opts[:after],
related_query.sort,
after_or_before
) do
{:ok, filter} ->
case Ash.Query.do_filter(related_query.resource, filter) do
%{valid?: true} = query ->
case Ash.Filter.Runtime.filter_matches(
related_query.domain,
value,
query.filter,
tenant: query.tenant,
actor: query.actor
) do
{:ok, value} ->
value
{:error, error} ->
raise Ash.Error.to_ash_error(error)
end
%{valid?: false, errors: errors} ->
raise Ash.Error.to_ash_error(errors)
end
{:error, error} ->
raise Ash.Error.to_ash_error(error)
end
else
value
end
{value, rest} =
value
|> Enum.drop(page_opts[:offset] || 0)
|> Enum.split(limit)
more? = !Enum.empty?(rest)
case pagination_type do
:offset ->
Ash.Page.Offset.new(value, count, related_query, more?, [])
:keyset ->
Ash.Page.Keyset.new(value, count, related_query.sort, related_query, more?, [])
end
end
defp apply_runtime_pagination(value, _, _, _) do
value
end
defp apply_runtime_offset(value, %{offset: nil}), do: value
defp apply_runtime_offset(value, %{offset: offset}), do: Enum.drop(value, offset)
defp apply_runtime_limit(value, %{limit: nil}), do: value
defp apply_runtime_limit(value, %{limit: limit}), do: Enum.take(value, limit)
defp put_result(record, {:ok, match}, key, _default) do
Map.put(
record,
key,
match
)
end
defp put_result(record, :error, key, default) do
Map.put(record, key, default)
end
defp attach_related(record, relationship_name, value) do
Map.put(record, relationship_name, value)
end
defp attach_lateral_join_related_records(
[%resource{} | _] = records,
relationship,
related_records,
attach_fun \\ &attach_related/3
) do
source_attribute =
if Map.get(relationship, :no_attributes?) do
nil
else
Ash.Resource.Info.attribute(relationship.source, relationship.source_attribute)
end
primary_key = Ash.Resource.Info.primary_key(resource)
pkey_to_key = pkey_normalizer(resource, primary_key)
attr_to_key = source_attribute && key_fn_for_type(source_attribute.type)
# Fast path requires that we can build comparable Map keys for whichever
# shape `__lateral_join_source__` takes. The pkey side is always needed;
# the attribute side only when this relationship has a `source_attribute`.
if pkey_to_key && (is_nil(source_attribute) || attr_to_key) do
# `__lateral_join_source__` is set by the data layer at query time and
# holds *either* a pkey map (`%{id: ...}`) or a single attribute value,
# depending on how the lateral join was constructed. Dispatch on shape:
# plain map → pkey path; anything else (including structs like
# `%Ash.CiString{}`) → attribute path. This mirrors the existing
# slow-path dispatch below (`is_map(...)` check around the
# `primary_key_matches?` vs `Ash.Type.equal?` branch).
lateral_source_key = fn
m when is_map(m) and not is_struct(m) -> pkey_to_key.(m)
v -> attr_to_key.(v)
end
values =
if relationship.cardinality == :many do
Enum.group_by(related_records, &lateral_source_key.(&1.__lateral_join_source__))
else
# `Enum.reverse` + `Map.new` makes the *first* related record win on
# key collisions (Map.new keeps the last write); reversing flips that
# to first-wins, matching the prior behaviour of this code.
Map.new(
Enum.reverse(related_records),
&{lateral_source_key.(&1.__lateral_join_source__), &1}
)
end
default =
if relationship.cardinality == :many do
[]
else
nil
end
if source_attribute do
# Two lookups because `__lateral_join_source__` could have been
# populated as either shape; try pkey first, then the attribute value.
Enum.map(records, fn record ->
with :error <- Map.fetch(values, pkey_to_key.(Map.take(record, primary_key))),
:error <-
Map.fetch(
values,
attr_to_key.(Map.get(record, relationship.source_attribute))
) do
attach_fun.(record, relationship.name, default)
else
{:ok, value} ->
attach_fun.(record, relationship.name, value)
end
end)
else
Enum.map(records, fn record ->
case Map.fetch(values, pkey_to_key.(Map.take(record, primary_key))) do
{:ok, value} ->
attach_fun.(record, relationship.name, value)
:error ->
attach_fun.(record, relationship.name, default)
end
end)
end
else
Enum.map(records, fn record ->
func =
if relationship.cardinality == :one do
:find
else
:filter
end
related =
apply(Enum, func, [
related_records,
fn related_record ->
if is_map(related_record.__lateral_join_source__) do
resource.primary_key_matches?(record, related_record.__lateral_join_source__)
else
Ash.Type.equal?(
source_attribute.type,
related_record.__lateral_join_source__,
Map.get(record, relationship.source_attribute)
)
end
end
])
attach_fun.(record, relationship.name, related)
end)
end
end
defp lateral_join?(%{action: action} = query, source_query, relationship, source_data) do
if action.manual do
raise_if_parent_expr!(relationship, query, "manual actions")
false
else
{offset, limit} = offset_and_limit(query)
resources =
relationship
|> Map.get(:through)
|> case do
through when is_list(through) ->
expanded = expand_through_path(relationship.source, through)
through_resources = Enum.map(expanded, & &1.destination)
[relationship.source] ++ through_resources ++ [relationship.destination]
through ->
[relationship.source, through, relationship.destination]
end
|> Enum.reject(&is_nil/1)
has_distinct? = query.distinct not in [[], nil]
has_page? = query.page not in [nil, false]
cond do
!Ash.DataLayer.data_layer_can?(
relationship.source,
{:lateral_join, resources}
) ->
false
is_many_to_many_not_unique_on_join?(relationship, query, source_query) ->
raise_if_parent_expr!(
relationship,
query,
"many to many relationships that don't have unique constraints on their join resource attributes"
)
false
Map.get(relationship, :from_many?) ->
true
Map.get(relationship, :manual) ->
# Manual relationships handle their own data fetching and return a map,
# not records with __lateral_join_source__ fields
false
limit == 1 && is_nil(relationship.context) && is_nil(relationship.filter) &&
is_nil(relationship.sort) && relationship.cardinality != :many ->
has_parent_expr?(relationship, query.context, query.domain)
limit == 1 && (source_data == :unknown || Enum.count_until(source_data, 2) == 1) &&
relationship.type != :many_to_many ->
has_parent_expr?(relationship, query.context, query.domain)
has_parent_expr?(relationship, query.context, query.domain) ->
true
is_list(Map.get(relationship, :through)) ->
true
relationship.type == :many_to_many && !is_list(relationship.through) &&
Ash.DataLayer.prefer_lateral_join_for_many_to_many?(
Ash.DataLayer.data_layer(relationship.source)
) ->
true
limit || offset || has_distinct? || has_page? ->
true
true ->
false
end
end
end
defp expand_through_path(resource, path) do
expand_through_path(resource, path, [])
end
defp expand_through_path(_resource, [], acc), do: Enum.reverse(acc)
defp expand_through_path(resource, [name | rest], acc) do
relationship = Ash.Resource.Info.relationship(resource, name)
if is_nil(relationship) do
raise Ash.Error.Query.NoSuchRelationship,
resource: resource,
relationship: name
end
case Map.get(relationship, :through) do
through when is_list(through) ->
nested_rels = expand_through_path(resource, through, [])
destination = List.last(nested_rels).destination
expand_through_path(destination, rest, Enum.reverse(nested_rels) ++ acc)
_ ->
expand_through_path(relationship.destination, rest, [relationship | acc])
end
end
defp raise_if_parent_expr!(relationship, query, reason) do
if has_parent_expr?(relationship, query.context, query.domain) do
raise ArgumentError, "Found `parent_expr` in unsupported context: #{reason}"
end
end
@doc false
def has_parent_expr?(%{
destination: destination,
filter: filter,
sort: sort,
context: rel_context
}) do
sort =
case Ash.Actions.Sort.process(destination, sort, %{}, rel_context) do
{:ok, sort} ->
sort
_ ->
nil
end
has_parent_expr_in_sort?(sort) ||
do_has_parent_expr?(filter)
end
@doc false
def has_parent_expr?(
%{
destination: destination,
filter: filter,
source: source,
sort: sort,
context: rel_context
},
context,
domain
) do
case Ash.Actions.Sort.process(destination, sort, %{}, rel_context) do
{:ok, sort} ->
sort
_ ->
nil
end
parent_stack = [source | Ash.Actions.Read.parent_stack_from_context(context)]
has_parent_expr_in_sort?(sort) ||
filter
|> Ash.Filter.hydrate_refs(%{
parent_stack: parent_stack,
resource: destination,
public?: false
})
|> Ash.Actions.Read.add_calc_context_to_filter(
context[:private][:actor],
context[:private][:authorize],
context[:private][:tenant],
context[:private][:tracer],
domain,
destination,
expand?: true,
parent_stack: parent_stack,
source_context: context
)
|> do_has_parent_expr?()
end
defp has_parent_expr_in_sort?(sort) do
sort
|> List.wrap()
|> Enum.any?(fn
atom when is_atom(atom) ->
false
{atom, _} when is_atom(atom) ->
false
%Ash.Query.Calculation{} = calculation ->
expression =
Ash.Resource.Calculation.expression(
calculation.module,
calculation.opts,
calculation.context
)
do_has_parent_expr?(expression)
{%Ash.Query.Calculation{} = calculation, _} ->
expression =
Ash.Resource.Calculation.expression(
calculation.module,
calculation.opts,
calculation.context
)
do_has_parent_expr?(expression)
end)
end
@doc false
def do_has_parent_expr?(filter, depth \\ 0) do
not is_nil(
Ash.Filter.find(
filter,
fn
%Ash.Query.Call{name: :parent, args: [expr]} ->
if depth == 0 do
true
else
do_has_parent_expr?(expr, depth - 1)
end
%Ash.Query.Exists{expr: expr} ->
do_has_parent_expr?(expr, depth + 1)
%Ash.Query.Parent{expr: expr} ->
if depth == 0 do
true
else
do_has_parent_expr?(expr, depth - 1)
end
%Ash.Query.Ref{
attribute: %Ash.Query.Aggregate{
field: %Ash.Query.Calculation{module: module, opts: opts, context: context}
}
} ->
if Ash.Resource.Calculation.has_expression?(module) do
Ash.Resource.Calculation.expression(module, opts, context)
|> do_has_parent_expr?(depth + 1)
else
false
end
_other ->
false
end,
true,
true,
true
)
)
end
defp apply_relationship_offset(query, relationship) do
if Map.get(relationship, :offset) do
Ash.Query.offset(query, relationship.offset)
else
query
end
end
defp offset_and_limit(query) do
if query.offset == 0 do
{nil, query.limit}
else
{query.offset, query.limit}
end
end
defp is_many_to_many_not_unique_on_join?(
%{type: :many_to_many} = relationship,
query,
source_query
) do
keys =
[
relationship.source_attribute_on_join_resource,
relationship.destination_attribute_on_join_resource
]
join_keys =
if query.tenant && source_query.tenant &&
Ash.Resource.Info.multitenancy_strategy(relationship.through) == :attribute do
attr = Ash.Resource.Info.multitenancy_attribute(relationship.through)
[attr | keys]
else
keys
end
primary_key_is_join_keys? =
Enum.all?(
Ash.Resource.Info.primary_key(relationship.through),
&(&1 in join_keys)
)
is_unique_on_join_keys? =
Enum.any?(Ash.Resource.Info.identities(relationship.through), fn identity ->
is_nil(identity.where) && identity.nils_distinct? &&
Enum.all?(identity.keys, &(&1 in join_keys))
end)
not (primary_key_is_join_keys? || is_unique_on_join_keys?)
end
defp is_many_to_many_not_unique_on_join?(_, _, _), do: false
# Returns a function that converts a value of `type` into a term safe for use
# as a Map key (i.e. comparable with `==`). Callers use this to build O(n+m)
# hash-based joins instead of O(n*m) `Ash.Type.equal?/3` scans.
#
# Three outcomes:
#
# * `simple_equality?` types (string, integer, uuid, ...): returns `& &1`.
# Values already work as Map keys; no transformation needed.
# * Types that define `c:Ash.Type.to_simple_equality_comparable/1` (e.g.
# `:ci_string` returns the downcased binary): returns a closure that
# applies the callback.
# * Neither: returns `nil`. The caller's contract is that `nil` means
# "no safe Map key exists, fall back to the scan-based slow path."
defp key_fn_for_type(type) do
cond do
Ash.Type.simple_equality?(type) ->
& &1
Ash.Type.simple_equality_comparable?(type) ->
&Ash.Type.to_simple_equality_comparable(type, &1)
true ->
nil
end
end
# The pkey analogue of `key_fn_for_type/1`. Returns a function that
# normalizes a `%{field => value}` pkey map (the shape produced by
# `Map.take(record, primary_key)`) into one usable as a Map key.
defp pkey_normalizer(resource, primary_key) do
if Ash.Resource.Info.primary_key_simple_equality?(resource) do
& &1
else
Enum.reduce_while(primary_key, [], fn name, acc ->
case key_fn_for_type(Ash.Resource.Info.attribute(resource, name).type) do
nil -> {:halt, nil}
f -> {:cont, [{name, f} | acc]}
end
end)
|> case do
nil ->
nil
pairs ->
fn pkey_map ->
Map.new(pairs, fn {name, f} -> {name, f.(Map.get(pkey_map, name))} end)
end
end
end
end
end