# SPDX-FileCopyrightText: 2019 ash_postgres contributors # # SPDX-License-Identifier: MIT defmodule AshPostgres.DataLayer do require Ecto.Query require Ash.Expr @manage_tenant %Spark.Dsl.Section{ name: :manage_tenant, describe: """ Configuration for the behavior of a resource that manages a tenant """, examples: [ """ manage_tenant do template ["organization_", :id] create? true update? false end """ ], schema: [ template: [ type: {:wrap_list, {:or, [:string, :atom]}}, required: true, doc: """ A template that will cause the resource to create/manage the specified schema. """ ], create?: [ type: :boolean, default: true, doc: "Whether or not to automatically create a tenant when a record is created" ], update?: [ type: :boolean, default: true, doc: "Whether or not to automatically update the tenant name if the record is udpated" ] ] } @index %Spark.Dsl.Entity{ name: :index, describe: """ Add an index to be managed by the migration generator. """, examples: [ "index [\"column\", \"column2\"], unique: true, where: \"thing = TRUE\"" ], target: AshPostgres.CustomIndex, schema: AshPostgres.CustomIndex.schema(), transform: {AshPostgres.CustomIndex, :transform, []}, args: [:fields] } @custom_indexes %Spark.Dsl.Section{ name: :custom_indexes, describe: """ A section for configuring indexes to be created by the migration generator. In general, prefer to use `identities` for simple unique constraints. This is a tool to allow for declaring more complex indexes. """, examples: [ """ custom_indexes do index [:column1, :column2], unique: true, where: "thing = TRUE" end """ ], entities: [ @index ] } @statement %Spark.Dsl.Entity{ name: :statement, describe: """ Add a custom statement for migrations. """, examples: [ """ statement :pgweb_idx do up "CREATE INDEX pgweb_idx ON pgweb USING GIN (to_tsvector('english', title || ' ' || body));" down "DROP INDEX pgweb_idx;" end """ ], target: AshPostgres.Statement, schema: AshPostgres.Statement.schema(), args: [:name] } @custom_statements %Spark.Dsl.Section{ name: :custom_statements, describe: """ A section for configuring custom statements to be added to migrations. Changing custom statements may require manual intervention, because Ash can't determine what order they should run in (i.e if they depend on table structure that you've added, or vice versa). As such, any `down` statements we run for custom statements happen first, and any `up` statements happen last. Additionally, when changing a custom statement, we must make some assumptions, i.e that we should migrate the old structure down using the previously configured `down` and recreate it. This may not be desired, and so what you may end up doing is simply modifying the old migration and deleting whatever was generated by the migration generator. As always: read your migrations after generating them! """, examples: [ """ custom_statements do # the name is used to detect if you remove or modify the statement statement :pgweb_idx do up "CREATE INDEX pgweb_idx ON pgweb USING GIN (to_tsvector('english', title || ' ' || body));" down "DROP INDEX pgweb_idx;" end end """ ], entities: [ @statement ] } @reference %Spark.Dsl.Entity{ name: :reference, describe: """ Configures the reference for a relationship in resource migrations. Keep in mind that multiple relationships can theoretically involve the same destination and foreign keys. In those cases, you only need to configure the `reference` behavior for one of them. Any conflicts will result in an error, across this resource and any other resources that share a table with this one. For this reason, instead of adding a reference configuration for `:nothing`, its best to just leave the configuration out, as that is the default behavior if *no* relationship anywhere has configured the behavior of that reference. """, examples: [ "reference :post, on_delete: :delete, on_update: :update, name: \"comments_to_posts_fkey\"" ], args: [:relationship], target: AshPostgres.Reference, schema: AshPostgres.Reference.schema() } @references %Spark.Dsl.Section{ name: :references, describe: """ A section for configuring the references (foreign keys) in resource migrations. This section is only relevant if you are using the migration generator with this resource. Otherwise, it has no effect. """, examples: [ """ references do reference :post, on_delete: :delete, on_update: :update, name: "comments_to_posts_fkey" end """ ], entities: [@reference], schema: [ polymorphic_on_delete: [ type: {:or, [ {:one_of, [:delete, :nilify, :nothing, :restrict]}, {:tagged_tuple, :nilify, {:wrap_list, :atom}} ]}, doc: "For polymorphic resources, configures the on_delete behavior of the automatically generated foreign keys to source tables." ], polymorphic_on_update: [ type: {:one_of, [:update, :nilify, :nothing, :restrict]}, doc: "For polymorphic resources, configures the on_update behavior of the automatically generated foreign keys to source tables." ], polymorphic_name: [ type: :string, doc: "For polymorphic resources, then index name to use for the foreign key to the source table." ] ] } @check_constraint %Spark.Dsl.Entity{ name: :check_constraint, describe: """ Add a check constraint to be validated. If a check constraint exists on the table but not in this section, and it produces an error, a runtime error will be raised. Provide a list of attributes instead of a single attribute to add the message to multiple attributes. By adding the `check` option, the migration generator will include it when generating migrations. """, examples: [ """ check_constraint :price, "price_must_be_positive", check: "price > 0", message: "price must be positive" """ ], args: [:attribute, :name], target: AshPostgres.CheckConstraint, schema: AshPostgres.CheckConstraint.schema() } @check_constraints %Spark.Dsl.Section{ name: :check_constraints, describe: """ A section for configuring the check constraints for a given table. This can be used to automatically create those check constraints, or just to provide message when they are raised """, examples: [ """ check_constraints do check_constraint :price, "price_must_be_positive", check: "price > 0", message: "price must be positive" end """ ], entities: [@check_constraint] } @references %Spark.Dsl.Section{ name: :references, describe: """ A section for configuring the references (foreign keys) in resource migrations. This section is only relevant if you are using the migration generator with this resource. Otherwise, it has no effect. """, examples: [ """ references do reference :post, on_delete: :delete, on_update: :update, name: "comments_to_posts_fkey" end """ ], entities: [@reference], schema: [ polymorphic_on_delete: [ type: {:or, [ {:one_of, [:delete, :nilify, :nothing, :restrict]}, {:tagged_tuple, :nilify, {:wrap_list, :atom}} ]}, doc: "For polymorphic resources, configures the on_delete behavior of the automatically generated foreign keys to source tables." ], polymorphic_on_update: [ type: {:one_of, [:update, :nilify, :nothing, :restrict]}, doc: "For polymorphic resources, configures the on_update behavior of the automatically generated foreign keys to source tables." ] ] } @postgres %Spark.Dsl.Section{ name: :postgres, describe: """ Postgres data layer configuration """, sections: [ @custom_indexes, @custom_statements, @manage_tenant, @references, @check_constraints ], modules: [ :repo ], examples: [ """ postgres do repo MyApp.Repo table "organizations" end """ ], schema: [ repo: [ type: {:or, [{:behaviour, Ecto.Repo}, {:fun, 2}]}, required: true, doc: "The repo that will be used to fetch your data. See the `AshPostgres.Repo` documentation for more. Can also be a function that takes a resource and a type `:read | :mutate` and returns the repo" ], migrate?: [ type: :boolean, default: true, doc: "Whether or not to include this resource in the generated migrations with `mix ash.generate_migrations`" ], storage_types: [ type: :keyword_list, default: [], doc: "A keyword list of attribute names to the ecto type that should be used for that attribute. Only necessary if you need to override the defaults." ], migration_types: [ type: :keyword_list, default: [], doc: "A keyword list of attribute names to the ecto migration type that should be used for that attribute. Only necessary if you need to override the defaults." ], migration_defaults: [ type: :keyword_list, default: [], doc: """ A keyword list of attribute names to the ecto migration default that should be used for that attribute. The string you use will be placed verbatim in the migration. Use fragments like `fragment(\\\\"now()\\\\")`, or for `nil`, use `\\\\"nil\\\\"`. """ ], calculations_to_sql: [ type: :keyword_list, doc: "A keyword list of calculations and their SQL representation. Used when creating unique indexes for identities over calculations" ], identity_wheres_to_sql: [ type: :keyword_list, doc: "A keyword list of identity names and the SQL representation of their `where` clause. See `AshPostgres.DataLayer.Info.identity_wheres_to_sql/1` for more details." ], base_filter_sql: [ type: :string, doc: "A raw sql version of the base_filter, e.g `representative = true`. Required if trying to create a unique constraint on a resource with a base_filter" ], simple_join_first_aggregates: [ type: {:list, :atom}, default: [], doc: """ A list of `:first` type aggregate names that can be joined to using a simple join. Use when you have a `:first` aggregate that uses a to-many relationship , but your `filter` statement ensures that there is only one result. Optimizes the generated query. """ ], skip_unique_indexes: [ type: {:wrap_list, :atom}, default: [], doc: "Skip generating unique indexes when generating migrations" ], unique_index_names: [ type: {:list, {:or, [{:tuple, [{:list, :atom}, :string]}, {:tuple, [{:list, :atom}, :string, :string]}]}}, default: [], doc: """ A list of unique index names that could raise errors that are not configured in identities, or an mfa to a function that takes a changeset and returns the list. In the format `{[:affected, :keys], "name_of_constraint"}` or `{[:affected, :keys], "name_of_constraint", "custom error message"}` """ ], exclusion_constraint_names: [ type: :any, default: [], doc: """ A list of exclusion constraint names that could raise errors. Must be in the format `{:affected_key, "name_of_constraint"}` or `{:affected_key, "name_of_constraint", "custom error message"}` """ ], identity_index_names: [ type: :any, default: [], doc: """ A keyword list of identity names to the unique index name that they should use when being managed by the migration generator. """ ], foreign_key_names: [ type: {:list, {:or, [ {:tuple, [{:or, [:atom, :string]}, :string]}, {:tuple, [{:or, [:atom, :string]}, :string, :string]} ]}}, default: [], doc: """ A list of foreign keys that could raise errors, or an mfa to a function that takes a changeset and returns a list. In the format: `{:key, "name_of_constraint"}` or `{:key, "name_of_constraint", "custom error message"}` """ ], migration_ignore_attributes: [ type: {:list, :atom}, default: [], doc: """ A list of attributes that will be ignored when generating migrations. """ ], table: [ type: :string, doc: """ The table to store and read the resource from. If this is changed, the migration generator will not remove the old table. """ ], schema: [ type: :string, doc: """ The schema that the table is located in. Schema-based multitenancy will supercede this option. If this is changed, the migration generator will not remove the old schema. """ ], polymorphic?: [ type: :boolean, default: false, doc: """ Declares this resource as polymorphic. See the [polymorphic resources guide](/documentation/topics/resources/polymorphic-resources.md) for more. """ ], create_table_options: [ type: :string, doc: """ Options passed to ecto's table/2 in the create migration. See the [Ecto.Migration.table/2](https://hexdocs.pm/ecto_sql/Ecto.Migration.html#table/2) documentation for more information. """ ] ] } @behaviour Ash.DataLayer @sections [@postgres] @moduledoc """ A postgres data layer that leverages Ecto's postgres capabilities. """ use Spark.Dsl.Extension, sections: @sections, verifiers: [ AshPostgres.Verifiers.PreventMultidimensionalArrayAggregates, AshPostgres.Verifiers.ValidateReferences, AshPostgres.Verifiers.ValidateCheckConstraints, AshPostgres.Verifiers.PreventAttributeMultitenancyAndNonFullMatchType, AshPostgres.Verifiers.EnsureTableOrPolymorphic, AshPostgres.Verifiers.ValidateIdentityIndexNames ] def migrate(args) do Mix.Task.reenable("ash_postgres.migrate") Mix.Task.run("ash_postgres.migrate", args) end def rollback(args) do {opts, _, _} = OptionParser.parse(args, switches: [ repo: :string ], aliases: [r: :repo] ) repos = AshPostgres.Mix.Helpers.repos!(opts, args) show_for_repo? = Enum.count_until(repos, 2) == 2 for repo <- repos do {:ok, _, _} = Ecto.Migrator.with_repo(repo, fn repo -> for_repo = if show_for_repo? do " for repo #{inspect(repo)}" else "" end migrations_path = AshPostgres.Mix.Helpers.migrations_path([], repo) tenant_migrations_path = AshPostgres.Mix.Helpers.tenant_migrations_path([], repo) if "--tenants" not in args do current_migrations = Ecto.Query.from(row in "schema_migrations", select: row.version ) |> repo.all() |> Enum.map(&to_string/1) files = migrations_path |> Path.join("**/*.exs") |> Path.wildcard() |> Enum.sort() |> Enum.reverse() |> Enum.filter(fn file -> Enum.any?(current_migrations, &String.starts_with?(Path.basename(file), &1)) end) |> Enum.take(20) |> Enum.map(&String.trim_leading(&1, migrations_path)) |> Enum.map(&String.trim_leading(&1, "/")) indexed = files |> Enum.with_index() |> Enum.map(fn {file, index} -> "#{index + 1}: #{file}" end) to = Mix.shell().prompt( """ How many migrations should be rolled back#{for_repo}? (default: 0) Last 20 migration names, with the input you must provide to rollback up to *and including* that migration: #{Enum.join(indexed, "\n")} Rollback to: """ |> String.trim_trailing() ) |> String.trim() |> case do "" -> nil "0" -> nil n -> try do files |> Enum.at(String.to_integer(n) - 1) rescue _ -> reraise "Required an integer value, got: #{n}", __STACKTRACE__ end |> String.split("_", parts: 2) |> Enum.at(0) |> String.to_integer() end if to do Mix.Task.run( "ash_postgres.rollback", args ++ ["-r", inspect(repo), "--to", to_string(to)] ) Mix.Task.reenable("ash_postgres.rollback") end end tenant_files = tenant_migrations_path |> Path.join("**/*.exs") |> Path.wildcard() |> Enum.sort() |> Enum.reverse() if !Enum.empty?(tenant_files) do first_tenant = repo.all_tenants() |> Enum.at(0) if first_tenant do current_tenant_migrations = Ecto.Query.from(row in "schema_migrations", select: row.version ) |> repo.all(prefix: first_tenant) |> Enum.map(&to_string/1) tenant_files = tenant_files |> Enum.filter(fn file -> Enum.any?( current_tenant_migrations, &String.starts_with?(Path.basename(file), &1) ) end) |> Enum.take(20) |> Enum.map(&String.trim_leading(&1, tenant_migrations_path)) |> Enum.map(&String.trim_leading(&1, "/")) indexed = tenant_files |> Enum.with_index() |> Enum.map(fn {file, index} -> "#{index + 1}: #{file}" end) to = Mix.shell().prompt( """ How many _tenant_ migrations should be rolled back#{for_repo}? (default: 0) IMPORTANT: we are assuming that all of your tenants have all had the same migrations run. If each tenant may be in a different state: *abort this command and roll them back individually*. To do so, use the `--only-tenants` option to `mix ash_postgres.rollback`. Last 20 migration names, with the input you must provide to rollback up to *and including* that migration: #{Enum.join(indexed, "\n")} Rollback to: """ |> String.trim_trailing() ) |> String.trim() |> case do "" -> nil "0" -> nil n -> try do tenant_files |> Enum.at(String.to_integer(n) - 1) rescue _ -> reraise "Required an integer value, got: #{n}", __STACKTRACE__ end |> String.split("_", parts: 2) |> Enum.at(0) |> String.to_integer() end if to do Mix.Task.run( "ash_postgres.rollback", args ++ ["--tenants", "-r", inspect(repo), "--to", to_string(to)] ) Mix.Task.reenable("ash_postgres.rollback") end end end end) end end def codegen(args) do Mix.Task.reenable("ash_postgres.generate_migrations") Mix.Task.run("ash_postgres.generate_migrations", args) end def setup(args) do # TODO: take args that we care about Mix.Task.run("ash_postgres.create", args) Mix.Task.run("ash_postgres.migrate", args) [] |> AshPostgres.Mix.Helpers.repos!(args) |> Enum.all?(&(not has_tenant_migrations?(&1))) |> case do true -> :ok _ -> Mix.Task.run("ash_postgres.migrate", ["--tenant" | args]) end end def tear_down(args) do # TODO: take args that we care about Mix.Task.run("ash_postgres.drop", args) end defp has_tenant_migrations?(repo) do [] |> AshPostgres.Mix.Helpers.tenant_migrations_path(repo) |> Path.join("**/*.exs") |> Path.wildcard() |> Enum.empty?() end import Ecto.Query, only: [from: 2, subquery: 1] @impl true def prefer_transaction?(resource) do AshPostgres.DataLayer.Info.repo(resource, :mutate).prefer_transaction?() end @impl true def prefer_transaction_for_atomic_updates?(resource) do AshPostgres.DataLayer.Info.repo(resource, :mutate).prefer_transaction_for_atomic_updates?() end @impl true def can?(_, :async_engine), do: true def can?(_, :combine), do: true def can?(_, {:combine, _}), do: true def can?(_, :bulk_create), do: true def can?(_, :bulk_upsert_return_skipped), do: true def can?(_, :action_select), do: true def can?(resource, :update_query) do # We can't currently support updating a record from a query # if that record manages a tenant on update !AshPostgres.DataLayer.Info.manage_tenant_update?(resource) end def can?(_, :destroy_query), do: true def can?(resource, :update_many) do # `update_many` is implemented with a single SQL MERGE, which (with RETURNING + merge_action) # requires PostgreSQL 17, and the same tenant-management restriction as `update_query`. AshPostgres.DataLayer.Info.pg_version_matches?(resource, ">= 17.0.0") && !AshPostgres.DataLayer.Info.manage_tenant_update?(resource) end def can?(_, {:lock, :for_update}), do: true def can?(_, :composite_types), do: true def can?(_, {:lock, string}), do: string |> String.upcase() |> can_lock?() def can?(_, :transact), do: true def can?(_, :composite_primary_key), do: true def can?(resource, {:atomic, :update}), do: not AshPostgres.DataLayer.Info.repo(resource, :mutate).disable_atomic_actions?() def can?(resource, {:atomic, :upsert}), do: not AshPostgres.DataLayer.Info.repo(resource, :mutate).disable_atomic_actions?() def can?(resource, {:atomic, :create}), do: not AshPostgres.DataLayer.Info.repo(resource, :mutate).disable_atomic_actions?() def can?(_, :upsert), do: true def can?(_, :changeset_filter), do: true def can?(resource, {:join, other_resource}) do data_layer = Ash.DataLayer.data_layer(resource) other_data_layer = Ash.DataLayer.data_layer(other_resource) data_layer == other_data_layer and AshPostgres.DataLayer.Info.repo(resource, :read) == AshPostgres.DataLayer.Info.repo(other_resource, :read) end def can?(resource, {:lateral_join, resources}) do repo = AshPostgres.DataLayer.Info.repo(resource, :read) data_layer = Ash.DataLayer.data_layer(resource) data_layer == __MODULE__ && Enum.all?(resources, fn resource -> Ash.DataLayer.data_layer(resource) == data_layer && AshPostgres.DataLayer.Info.repo(resource, :read) == repo end) end def can?(_, :boolean_filter), do: true def can?(_, {:aggregate, type}) when type in [:count, :sum, :first, :list, :avg, :max, :min, :exists, :custom], do: true def can?(_, {:aggregate, :unrelated}), do: true def can?(_, {:exists, :unrelated}), do: true def can?(_, :aggregate_filter), do: true def can?(_, :aggregate_sort), do: true def can?(_, :calculate), do: true def can?(_, :expression_calculation), do: true def can?(_, :expression_calculation_sort), do: true def can?(_, :create), do: true def can?(_, :select), do: true def can?(_, :read), do: true def can?(resource, action) when action in ~w[update destroy]a do resource |> Ash.Resource.Info.primary_key() |> Enum.any?() end def can?(_, :filter), do: true def can?(_, :limit), do: true def can?(_, :offset), do: true def can?(_, :multitenancy), do: true def can?(_, {:filter_relationship, %{manual: {module, _}}}) do Spark.implements_behaviour?(module, AshPostgres.ManualRelationship) end def can?(_, {:filter_relationship, _}), do: true def can?(_, {:aggregate_relationship, %{manual: {module, _}}}) do Spark.implements_behaviour?(module, AshPostgres.ManualRelationship) end def can?(_, {:aggregate_relationship, _}), do: true def can?(_, :timeout), do: true def can?(resource, :expr_error), do: not AshPostgres.DataLayer.Info.repo(resource, :mutate).disable_expr_error?() def can?(resource, :required_error) do not AshPostgres.DataLayer.Info.repo(resource, :mutate).disable_expr_error?() && "ash-functions" in AshPostgres.DataLayer.Info.repo(resource, :read).installed_extensions() && "ash-functions" in AshPostgres.DataLayer.Info.repo(resource, :mutate).installed_extensions() end def can?(resource, {:filter_expr, %Ash.Query.Function.Error{}}) do not AshPostgres.DataLayer.Info.repo(resource, :mutate).disable_expr_error?() && "ash-functions" in AshPostgres.DataLayer.Info.repo(resource, :read).installed_extensions() && "ash-functions" in AshPostgres.DataLayer.Info.repo(resource, :mutate).installed_extensions() end def can?(_, {:filter_expr, _}), do: true def can?(_, :nested_expressions), do: true def can?(_, {:query_aggregate, _}), do: true def can?(_, :sort), do: true def can?(_, :distinct_sort), do: true def can?(_, :distinct), do: true def can?(_, {:sort, _}), do: true def can?(_, :through_relationship), do: true def can?(_, _), do: false @locks [ "FOR UPDATE", "FOR NO KEY UPDATE", "FOR SHARE", "FOR KEY SHARE" ] for lock <- @locks do defp can_lock?(unquote(lock)), do: true for suffix <- ["NOWAIT", "SKIP LOCKED"] do defp can_lock?(unquote("#{lock} #{suffix}")), do: true end end defp can_lock?(_), do: false @impl true def in_transaction?(resource) do AshPostgres.DataLayer.Info.repo(resource, :mutate).in_transaction?() end @impl true def limit(query, nil, _), do: {:ok, query} def limit(query, limit, _resource) do {:ok, from(row in query, limit: ^limit)} end @impl true def source(resource) do AshPostgres.DataLayer.Info.table(resource) || "" end @impl true def set_context(resource, data_layer_query, context) do AshSql.Query.set_context(resource, data_layer_query, AshPostgres.SqlImplementation, context) end @impl true def offset(query, nil, _), do: query def offset(%{offset: old_offset} = query, 0, _resource) when old_offset in [0, nil] do {:ok, query} end def offset(query, offset, _resource) do {:ok, from(row in query, offset: ^offset)} end @impl true def return_query(query, resource) do query |> AshSql.Bindings.default_bindings(resource, AshPostgres.SqlImplementation) |> AshSql.Query.return_query(resource) end @impl true def run_query(query, resource) do query = AshSql.Bindings.default_bindings(query, resource, AshPostgres.SqlImplementation) if AshPostgres.DataLayer.Info.polymorphic?(resource) && no_table?(query) do raise_table_error!(resource, :read) else repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, query) with_savepoint(repo, query, fn -> repo.all( query, AshSql.repo_opts(repo, AshPostgres.SqlImplementation, nil, nil, resource) ) |> AshSql.Query.remap_mapped_fields(query) |> then(fn results -> if query.__ash_bindings__.context[:data_layer][:combination_of_queries?] do Enum.map(results, fn result -> struct(resource, result) |> Map.put(:__meta__, %Ecto.Schema.Metadata{ state: :loaded }) end) else results end end) |> then(&{:ok, &1}) end) end rescue e -> handle_raised_error(e, __STACKTRACE__, query, resource) end defp no_table?(%{from: %{source: {"", _}}}), do: true defp no_table?(_), do: false @impl true def functions(resource) do config = AshPostgres.DataLayer.Info.repo(resource, :mutate).config() functions = [ AshPostgres.Functions.Like, AshPostgres.Functions.ILike, AshPostgres.Functions.Binding, AshPostgres.Functions.PostgresIn ] functions = [Ash.Query.Function.RequiredError | functions] functions = if "pg_trgm" in (config[:installed_extensions] || []) do functions ++ [ AshPostgres.Functions.TrigramSimilarity ] else functions end if "vector" in (config[:installed_extensions] || []) do functions ++ [ AshPostgres.Functions.VectorCosineDistance, AshPostgres.Functions.VectorL2Distance ] else functions end end @impl true def combination_acc(query), do: AshSql.Query.combination_acc(query) @impl true def run_aggregate_query(original_query, aggregates, resource) do AshSql.AggregateQuery.run_aggregate_query( original_query, aggregates, resource, AshPostgres.SqlImplementation ) end @impl true def set_tenant(resource, query, tenant) do if Ash.Resource.Info.multitenancy_strategy(resource) == :context && tenant do {:ok, Map.put(Ecto.Query.put_query_prefix(query, to_string(tenant)), :__tenant__, tenant)} else {:ok, query} end end @impl true def run_aggregate_query_with_lateral_join( query, aggregates, root_data, destination_resource, path ) do {can_group, cant_group} = aggregates |> Enum.split_with(&AshSql.Aggregate.can_group?(destination_resource, &1, query)) |> case do {[one], cant_group} -> {[], [one | cant_group]} {can_group, cant_group} -> {can_group, cant_group} end case lateral_join_query( query, root_data, path ) do {:ok, lateral_join_query} -> source_resource = path |> Enum.at(0) |> elem(0) |> Map.get(:resource) subquery = from(row in subquery(lateral_join_query), as: ^0, select: %{}) subquery = AshSql.Bindings.default_bindings( subquery, source_resource, AshPostgres.SqlImplementation ) {global_filter, can_group} = AshSql.Aggregate.extract_shared_filters(can_group) original_subquery = subquery subquery = case global_filter do {:ok, global_filter} -> filter(subquery, global_filter, destination_resource) :error -> {:ok, subquery} end case subquery do {:error, error} -> {:error, error} {:ok, subquery} -> query = Enum.reduce( can_group, subquery, fn agg, subquery -> has_exists? = Ash.Filter.find(agg.query && agg.query.filter, fn %Ash.Query.Exists{} -> true _ -> false end) first_relationship = Ash.Resource.Info.relationship( source_resource, agg.relationship_path |> Enum.at(0) ) AshSql.Aggregate.add_subquery_aggregate_select( subquery, agg.relationship_path |> Enum.drop(1), agg, destination_resource, has_exists?, first_relationship ) end ) result = case can_group do [] -> %{} _ -> repo = AshSql.dynamic_repo(source_resource, AshPostgres.SqlImplementation, query) repo.one( query, AshSql.repo_opts( repo, AshPostgres.SqlImplementation, nil, nil, source_resource ) ) end {:ok, AshSql.AggregateQuery.add_single_aggs( result, source_resource, original_subquery, cant_group, AshPostgres.SqlImplementation )} end {:error, error} -> {:error, error} end end @impl true def run_query_with_lateral_join( query, root_data, _destination_resource, path ) do {calculations_require_rewrite, aggregates_require_rewrite, query} = AshSql.Query.rewrite_nested_selects(query) case lateral_join_query( query, root_data, path ) do {:ok, lateral_join_query} -> source_resource = path |> Enum.at(0) |> elem(0) |> Map.get(:resource) repo = AshSql.dynamic_repo(source_resource, AshPostgres.SqlImplementation, lateral_join_query) # patching strange behavior that sets `take` to this empty list even though I'm not telling it to lateral_join_query = case lateral_join_query do %{select: %{take: %{0 => {:map, []}}}} -> put_in(lateral_join_query.select.take, %{}) _ -> lateral_join_query end results = repo.all( lateral_join_query, AshSql.repo_opts(repo, AshPostgres.SqlImplementation, nil, nil, source_resource) ) |> AshSql.Query.remap_mapped_fields( query, calculations_require_rewrite, aggregates_require_rewrite ) {:ok, results} {:error, error} -> {:error, error} end end defp lateral_join_query( query, root_data, [{source_query, source_attribute, destination_attribute, relationship}] = path ) do source_query = Ash.Query.new(source_query) base_query = if query.__ash_bindings__[:__order__?] do from(row in query, select_merge: %{__order__: over(row_number(), :order)} ) else query end base_query = if Map.get(relationship, :from_many?) do from(row in base_query, limit: 1) else base_query end base_query = if Map.get(relationship, :offset) do from(row in base_query, offset: ^relationship.offset) else base_query end base_query = cond do Map.get(relationship, :manual) -> {module, opts} = relationship.manual case module.ash_postgres_subquery(opts, 0, 0, base_query) do {:ok, subquery} -> subquery {:error, error} -> {:error, error} subquery -> subquery end Map.get(relationship, :no_attributes?) -> base_query true -> from(destination in base_query, where: field(destination, ^destination_attribute) == field(parent_as(^0), ^source_attribute) ) end subquery = base_query |> set_subquery_prefix(source_query, relationship.destination) |> subquery() source_pkey = Ash.Resource.Info.primary_key(source_query.resource) case lateral_join_source_query(query, source_query, root_data, path) do {:ok, data_layer_query} -> source_values = Enum.map(root_data, &Map.get(&1, source_attribute)) data_layer_query = if Map.get(relationship, :manual) || Map.get(relationship, :no_attributes?) do data_layer_query else source_filter = case source_pkey do [] -> Ecto.Query.dynamic([source], field(source, ^source_attribute) in ^source_values) [field] -> values = Enum.map(root_data, &Map.get(&1, field)) Ecto.Query.dynamic([source], field(source, ^field) in ^values) fields -> Enum.reduce(root_data, nil, fn record, acc -> row_match = Enum.reduce(fields, nil, fn field, acc -> if is_nil(acc) do Ecto.Query.dynamic( [source], field(source, ^field) == ^Map.get(record, field) ) else Ecto.Query.dynamic( [source], field(source, ^field) == ^Map.get(record, field) and ^acc ) end end) if is_nil(acc) do row_match else Ecto.Query.dynamic(^row_match or ^acc) end end) end from(source in data_layer_query, where: ^source_filter ) end data_layer_query = data_layer_query |> Ecto.Query.exclude(:distinct) |> Ecto.Query.exclude(:select) if query.__ash_bindings__[:__order__?] do {:ok, from(source in data_layer_query, inner_lateral_join: destination in subquery(get_subquery(subquery)), on: true, order_by: destination.__order__, select: merge(destination, %{__lateral_join_source__: map(source, ^source_pkey)}), distinct: true )} else {:ok, from(source in data_layer_query, inner_lateral_join: destination in subquery(get_subquery(subquery)), on: true, select: merge(destination, %{__lateral_join_source__: map(source, ^source_pkey)}), distinct: true )} end {:error, error} -> {:error, error} end end defp lateral_join_query( query, root_data, [ {source_query, source_attribute, source_attribute_on_join_resource, relationship} | rest ] = path ) do source_query = Ash.Query.new(source_query) source_values = Enum.map(root_data, &Map.get(&1, source_attribute)) source_pkey = Ash.Resource.Info.primary_key(source_query.resource) case lateral_join_source_query(query, source_query, root_data, path) do {:ok, data_layer_query} -> data_layer_query = Ecto.Query.exclude(data_layer_query, :select) case rest do [ {through_resource, destination_attribute_on_join_resource, destination_attribute, through_relationship} ] -> through_binding = Map.get(query, :__ash_bindings__)[:current] through_resource |> Ash.Query.new() |> Ash.Query.put_context(:data_layer, %{ start_bindings_at: through_binding }) |> Ash.Query.set_context(Map.get(through_relationship, :context)) |> Ash.Query.do_filter(Map.get(through_relationship, :filter)) |> then(fn q -> # For through-list paths, the first relationship's filter applies to the through table # For many_to_many, the relationship filter is for the destination, not the through table if !is_atom(Map.get(relationship, :through)) || is_nil(Map.get(relationship, :through)) do Ash.Query.do_filter(q, Map.get(relationship, :filter), parent_stack: [relationship.source] ) else q end end) |> Ash.Query.sort(Map.get(through_relationship, :sort)) |> then(fn q -> if Map.get(through_relationship, :limit) do Ash.Query.limit(q, Map.get(through_relationship, :limit)) else q end end) |> Ash.Query.set_tenant(source_query.tenant) |> set_lateral_join_prefix(query) |> case do %{valid?: true} = through_query -> through_query |> Ash.Query.data_layer_query() query -> {:error, query} end |> case do {:ok, through_query} -> through_query = Ecto.Query.exclude(through_query, :select) has_subqueries_in_wheres? = Enum.any?(through_query.wheres, fn w -> w.subqueries != [] end) needs_subquery? = through_query.limit != nil || through_query.order_bys != [] || (through_query.joins && through_query.joins != []) || has_subqueries_in_wheres? || (Ash.Resource.Info.multitenancy_strategy(through_resource.resource) == :context && source_query.tenant) through_query = if needs_subquery? do through_query = from(through in through_query, where: field(through, ^source_attribute_on_join_resource) == field(parent_as(^0), ^source_attribute) ) subquery( set_subquery_prefix( through_query, source_query, through_resource.resource ) ) else set_subquery_prefix(through_query, source_query, through_resource.resource) end if query.__ash_bindings__[:__order__?] do subquery = if needs_subquery? do subquery( from( destination in query, select_merge: %{__order__: over(row_number(), :order)}, join: through in ^through_query, as: ^through_binding, on: field(through, ^destination_attribute_on_join_resource) == field(destination, ^destination_attribute) ) |> set_subquery_prefix( source_query, relationship.destination ) ) else subquery( from( destination in query, select_merge: %{__order__: over(row_number(), :order)}, join: through in ^through_query, as: ^through_binding, on: field(through, ^destination_attribute_on_join_resource) == field(destination, ^destination_attribute), where: field(through, ^source_attribute_on_join_resource) == field( parent_as(^0), ^source_attribute ) ) |> set_subquery_prefix( source_query, relationship.destination ) ) end {:ok, from(source in data_layer_query, where: field(source, ^source_attribute) in ^source_values, inner_lateral_join: destination in subquery(get_subquery(subquery)), on: true, select: destination, select_merge: %{__lateral_join_source__: map(source, ^source_pkey)}, order_by: destination.__order__, distinct: true )} else subquery = if needs_subquery? do subquery( from( destination in query, join: through in ^through_query, as: ^through_binding, on: field(through, ^destination_attribute_on_join_resource) == field(destination, ^destination_attribute) ) |> set_subquery_prefix( source_query, relationship.destination ) ) else subquery( from( destination in query, join: through in ^through_query, as: ^through_binding, on: field(through, ^destination_attribute_on_join_resource) == field(destination, ^destination_attribute), where: field(through, ^source_attribute_on_join_resource) == field( parent_as(^0), ^source_attribute ) ) |> set_subquery_prefix( source_query, relationship.destination ) ) end data_layer_query = Ecto.Query.exclude(data_layer_query, :distinct) {:ok, from(source in data_layer_query, where: field(source, ^source_attribute) in ^source_values, inner_lateral_join: destination in subquery(get_subquery(subquery)), on: true, select: destination, select_merge: %{__lateral_join_source__: map(source, ^source_pkey)}, distinct: true )} end {:error, error} -> {:error, error} end rest -> case build_through_queries(query, source_query, relationship, rest) do {:ok, through_queries} -> data_layer_query = Ecto.Query.exclude(data_layer_query, :distinct) base_subquery = if query.__ash_bindings__[:__order__?] do from(destination in query, select_merge: %{__order__: over(row_number(), :order)} ) else query end final_subquery = rest |> Enum.zip(through_queries) |> Enum.reverse() |> Enum.reduce( {base_subquery, 0}, fn {{_through_res, src_attr, dst_attr, _through_rel}, {through_ecto_q, through_module}}, {sub_q, offset} -> binding = Map.get(query, :__ash_bindings__)[:current] + offset is_outermost? = offset == length(rest) - 1 prepared_through = if is_outermost? do correlated = from(through in through_ecto_q, where: field(through, ^source_attribute_on_join_resource) == field(parent_as(^0), ^source_attribute) ) subquery(set_subquery_prefix(correlated, source_query, through_module)) else through_ecto_q |> set_subquery_prefix(source_query, through_module) |> subquery() end new_sub_q = if offset == 0 do from(dest in sub_q, join: through in ^prepared_through, as: ^binding, on: field(through, ^src_attr) == field(dest, ^dst_attr) ) else prev_binding = binding - 1 from(dest in sub_q, join: through in ^prepared_through, as: ^binding, on: field(through, ^src_attr) == field(as(^prev_binding), ^dst_attr) ) end {new_sub_q, offset + 1} end ) |> elem(0) |> set_subquery_prefix(source_query, relationship.destination) |> subquery() result_query = if query.__ash_bindings__[:__order__?] do from(source in data_layer_query, where: field(source, ^source_attribute) in ^source_values, inner_lateral_join: destination in subquery(get_subquery(final_subquery)), on: true, select: destination, select_merge: %{__lateral_join_source__: map(source, ^source_pkey)}, order_by: destination.__order__, distinct: true ) else from(source in data_layer_query, where: field(source, ^source_attribute) in ^source_values, inner_lateral_join: destination in subquery(get_subquery(final_subquery)), on: true, select: destination, select_merge: %{__lateral_join_source__: map(source, ^source_pkey)}, distinct: true ) end {:ok, result_query} {:error, error} -> {:error, error} end end {:error, error} -> {:error, error} end end defp build_through_queries(query, source_query, prev_rel, [{resource, _, _, rel}]) do resource |> Ash.Query.new() |> Ash.Query.put_context(:data_layer, %{start_bindings_at: 0}) |> Ash.Query.do_filter(Map.get(prev_rel, :filter)) |> Ash.Query.set_tenant(source_query.tenant) |> set_lateral_join_prefix(query) |> case do %{valid?: true} = tq -> case Ash.Query.data_layer_query(tq) do {:ok, ecto_query} -> result = [{Ecto.Query.exclude(ecto_query, :select), rel.source}] {:ok, result} {:error, error} -> {:error, error} end invalid_query -> {:error, invalid_query} end end defp build_through_queries(query, source_query, prev_rel, [{_, _, _, through_rel} = head | rest]) do with {:ok, rel_query} <- build_through_queries(query, source_query, prev_rel, [head]), {:ok, rest_queries} <- build_through_queries(query, source_query, through_rel, rest) do {:ok, rel_query ++ rest_queries} end end defp lateral_join_source_query( %{ __ash_bindings__: %{ lateral_join_source_query: lateral_join_source_query } }, source_query, _root_data, _path ) when not is_nil(lateral_join_source_query) do {:ok, lateral_join_source_query |> set_subquery_prefix(source_query, lateral_join_source_query.__ash_bindings__.resource)} end defp lateral_join_source_query(query, source_query, root_data, path) do source_query.resource |> Ash.Query.set_context(%{:data_layer => source_query.context[:data_layer]}) |> Ash.Query.set_context(%{ :data_layer => Map.put( source_query.context[:data_layer] || %{}, :no_inner_join?, true ) |> Map.delete(:lateral_join_source) }) |> Ash.Query.set_tenant(source_query.tenant) |> filter_for_records(root_data) |> set_lateral_join_prefix(query) |> case do %{valid?: true} = query -> relationship = path |> List.first() |> elem(3) {:ok, expr} = Ash.Filter.hydrate_refs(relationship.filter, %{ resource: relationship.destination, parent_stack: [relationship.source] }) parent_expr = AshSql.Join.parent_expr(expr) used_aggregates = Ash.Filter.used_aggregates(parent_expr, []) with {:ok, query} <- Ash.Query.data_layer_query(query) do AshSql.Aggregate.add_aggregates( query, used_aggregates, relationship.source, false, query.__ash_bindings__.root_binding ) end query -> {:error, query} end end defp filter_for_records(query, records) do keys = case Ash.Resource.Info.primary_key(query.resource) do [] -> case Ash.Resource.Info.identities(query.resource) do [%{keys: keys} | _] -> keys _ -> [] end pkey -> pkey end expr = case keys do [] -> raise "Cannot use lateral joins with a resource that has no primary key and no identities" [key] -> Ash.Expr.expr(^Ash.Expr.ref(key) in ^Enum.map(records, &Map.get(&1, key))) keys -> Enum.reduce(records, Ash.Expr.expr(false), fn record, filter_expr -> all_keys_match_expr = Enum.reduce(keys, Ash.Expr.expr(true), fn key, key_expr -> Ash.Expr.expr(^key_expr and ^Ash.Expr.ref(key) == ^Map.get(record, key)) end) Ash.Expr.expr(^filter_expr or ^all_keys_match_expr) end) end Ash.Query.do_filter(query, expr) end defp get_subquery(%Ecto.Query{} = query), do: query defp get_subquery(%Ecto.SubQuery{query: query}), do: query @doc false def set_subquery_prefix(data_layer_query, source_query, resource) do repo = AshPostgres.DataLayer.Info.repo(resource, :mutate) config = repo.config() case data_layer_query do %{__ash_bindings__: %{context: %{data_layer: %{schema: schema}}}} when not is_nil(schema) -> data_layer_query _ -> query_tenant = case source_query do %{__tenant__: tenant} -> tenant %{tenant: tenant} -> Ash.ToTenant.to_tenant(tenant, resource) _ -> nil end if Ash.Resource.Info.multitenancy_strategy(resource) == :context do %{ data_layer_query | prefix: query_tenant || AshPostgres.DataLayer.Info.schema(resource) || config[:default_prefix] || "public" } else %{ data_layer_query | prefix: AshPostgres.DataLayer.Info.schema(resource) || config[:default_prefix] || "public" } end end end defp set_lateral_join_prefix(ash_query, query) do if Ash.Resource.Info.multitenancy_strategy(ash_query.resource) == :context do Ash.Query.set_tenant(ash_query, query.prefix) else ash_query end end @impl true def resource_to_query(resource, domain) do AshSql.Query.resource_to_query(resource, AshPostgres.SqlImplementation, domain) end @impl true def combination_of(combination_of, resource, domain) do AshSql.Query.combination_of(combination_of, resource, domain, AshPostgres.SqlImplementation) end @impl true def update_query(query, changeset, resource, options) do repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, changeset) ecto_changeset = case changeset.data do %Ash.Changeset.OriginalDataNotAvailable{} -> changeset.resource.__struct__() data -> data end |> Map.update!(:__meta__, &Map.put(&1, :source, table(resource, changeset))) |> ecto_changeset(changeset, :update, repo, true) case bulk_updatable_query( query, resource, changeset.atomics, options[:calculations] || [], changeset.context ) do {:error, error} -> {:error, error} {:ok, query} -> try do repo_opts = AshSql.repo_opts( repo, AshPostgres.SqlImplementation, changeset.timeout, changeset.tenant, changeset.resource ) case AshSql.Atomics.query_with_atomics( resource, query, changeset.filter, changeset.atomics, ecto_changeset.changes, [] ) do {:empty, query} -> if options[:return_records?] do if changeset.context[:data_layer][:use_atomic_update_data?] do case query.__ash_bindings__ do %{expression_accumulator: %AshSql.Expr.ExprInfo{has_error?: true}} -> # if the query could produce an error # we must run it even if we will just be returning the original data. repo.all(query, repo_opts) _ -> :ok end {:ok, [changeset.data]} else {:ok, repo.all(query, repo_opts)} end else :ok end {:ok, query} -> query = if options[:return_records?] do {:ok, query} = if options[:action_select] do query |> Ecto.Query.exclude(:select) |> Ecto.Query.select([row], struct(row, ^options[:action_select])) |> add_calculations(options[:calculations] || [], resource) else query |> Ecto.Query.exclude(:select) |> Ecto.Query.select([row], row) |> add_calculations(options[:calculations] || [], resource) end query else Ecto.Query.exclude(query, :select) end {_, results} = with_savepoint(repo, query, fn -> repo.update_all( Map.delete(query, :__ash_bindings__), [], repo_opts ) end) if options[:return_records?] do results = AshSql.Query.remap_mapped_fields(results, query) if changeset.context[:data_layer][:use_atomic_update_data?] && Enum.count_until(results, 2) == 1 do modifying = Map.keys(changeset.attributes) ++ Keyword.keys(changeset.atomics) ++ Ash.Resource.Info.primary_key(resource) result = hd(results) Map.merge(changeset.data, Map.take(result, modifying)) |> Map.update!(:aggregates, &Map.merge(&1, result.aggregates)) |> Map.update!(:calculations, &Map.merge(&1, result.calculations)) |> then(&{:ok, [&1]}) else {:ok, results} end else :ok end {:error, error} -> {:error, error} end rescue e -> handle_raised_error( e, __STACKTRACE__, {:ecto_changeset, :update, ecto_changeset}, resource ) end end end defp bulk_updatable_query(query, resource, atomics, calculations, context, type \\ :update) do Enum.reduce_while(atomics, {:ok, query}, fn {_, expr}, {:ok, query} -> used_aggregates = Ash.Filter.used_aggregates(expr, []) with {:ok, query} <- AshSql.Join.join_all_relationships( query, %Ash.Filter{ resource: resource, expression: expr }, left_only?: true ), {:ok, query} <- AshSql.Aggregate.add_aggregates(query, used_aggregates, resource, false, 0) do {:cont, {:ok, query}} else {:error, error} -> {:halt, {:error, error}} end end) |> case do {:ok, query} -> requires_adding_inner_join? = case type do :update -> # could potentially optimize this to avoid the subquery by shuffling free # inner joins to the top of the query has_inner_join_to_start? = case Enum.at(query.joins, 0) do nil -> false %{qual: :inner} -> true _ -> false end cond do has_inner_join_to_start? -> false Enum.any?(query.joins, &(&1.qual != :inner)) -> true Enum.any?(atomics ++ calculations, fn {_, expr} -> Ash.Filter.list_refs(expr) |> Enum.any?(&(&1.relationship_path != [])) end) -> true true -> false end :destroy -> Enum.any?(query.joins, &(&1.qual != :inner)) || Enum.any?(atomics ++ calculations, fn {_, expr} -> expr |> Ash.Filter.list_refs() |> Enum.any?(&(&1.relationship_path != [])) end) end has_exists? = Enum.any?(atomics, fn {_key, expr} -> Ash.Filter.find(expr, fn %Ash.Query.Exists{} -> true _ -> false end) end) needs_to_join? = requires_adding_inner_join? || query.distinct || query.limit || query.offset || has_exists? || query.combinations != [] query = if needs_to_join? do root_query = Ecto.Query.exclude(query, :select) root_query_result = cond do query.limit || query.offset -> with {:ok, root_query} <- AshSql.Atomics.select_atomics(resource, root_query, atomics) do {:ok, from(row in Ecto.Query.subquery(root_query), []), root_query.__ash_bindings__.expression_accumulator, atomics != []} end !Enum.empty?(query.joins) || has_exists? -> with root_query <- Ecto.Query.exclude(root_query, :order_by), {:ok, root_query} <- AshSql.Atomics.select_atomics(resource, root_query, atomics) do {:ok, from(row in Ecto.Query.subquery(root_query), []), root_query.__ash_bindings__.expression_accumulator, atomics != []} end true -> {:ok, Ecto.Query.exclude(root_query, :order_by), Map.get(root_query, :__ash_bindings__).expression_accumulator, false} end case root_query_result do {:ok, root_query, acc, selected_atomics?} -> dynamic = Enum.reduce(Ash.Resource.Info.primary_key(resource), nil, fn pkey, dynamic -> if dynamic do Ecto.Query.dynamic( [row, joining], field(row, ^pkey) == field(joining, ^pkey) and ^dynamic ) else Ecto.Query.dynamic( [row, joining], field(row, ^pkey) == field(joining, ^pkey) ) end end) faked_query = from(row in query.from.source, as: ^0, inner_join: limiter in ^root_query, as: ^1, on: ^dynamic ) |> AshSql.Bindings.default_bindings( query.__ash_bindings__.resource, AshPostgres.SqlImplementation, context ) |> AshSql.Bindings.merge_expr_accumulator(acc) |> then(fn query -> if selected_atomics? do Map.update!(query, :__ash_bindings__, &Map.put(&1, :atomics_in_binding, 1)) else query end end) {:ok, faked_query} {:error, error} -> {:error, error} end else {:ok, query |> Ecto.Query.exclude(:select) |> Ecto.Query.exclude(:order_by)} end case query do {:ok, query} -> Enum.reduce_while(calculations, {:ok, query}, fn {_, expr}, {:ok, query} -> used_aggregates = Ash.Filter.used_aggregates(expr, []) with {:ok, query} <- AshSql.Join.join_all_relationships( query, %Ash.Filter{ resource: resource, expression: expr }, left_only?: true ), {:ok, query} <- AshSql.Aggregate.add_aggregates(query, used_aggregates, resource, false, 0) do {:cont, {:ok, query}} else {:error, error} -> {:halt, {:error, error}} end end) {:error, error} -> {:error, error} end {:error, error} -> {:error, error} end end @impl true def destroy_query(query, changeset, resource, options) do repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, changeset) ecto_changeset = case changeset.data do %Ash.Changeset.OriginalDataNotAvailable{} -> changeset.resource.__struct__() data -> data end |> Map.update!(:__meta__, &Map.put(&1, :source, table(resource, changeset))) |> ecto_changeset(changeset, :delete, repo, true) case bulk_updatable_query( query, resource, [], options[:calculations] || [], changeset.context, :destroy ) do {:error, error} -> {:error, error} {:ok, query} -> try do repo_opts = AshSql.repo_opts( repo, AshPostgres.SqlImplementation, changeset.timeout, changeset.tenant, changeset.resource ) query = if options[:return_records?] do {:ok, query} = case options[:action_select] do nil -> query |> Ecto.Query.exclude(:select) |> Ecto.Query.select([row], row) |> add_calculations(options[:calculations] || [], resource) action_select -> query |> Ecto.Query.exclude(:select) |> Ecto.Query.select([row], struct(row, ^action_select)) |> add_calculations(options[:calculations] || [], resource) end query else Ecto.Query.exclude(query, :select) end # query = Ecto.Query.exclude(query, :distinct) {_, results} = with_savepoint(repo, query, fn -> repo.delete_all( query, repo_opts ) end) if options[:return_records?] do {:ok, AshSql.Query.remap_mapped_fields(results, query)} else :ok end rescue e -> handle_raised_error( e, __STACKTRACE__, {:ecto_changeset, :delete, ecto_changeset}, resource ) end end end @impl true def calculate(resource, expressions, context) do ash_query = resource |> Ash.Query.new() |> Map.put(:context, context) {:ok, query} = Ash.Query.data_layer_query(ash_query) query = AshSql.Bindings.default_bindings(query, resource, AshPostgres.SqlImplementation) try do {dynamics, query} = Enum.reduce(expressions, {[], query}, fn expression, {dynamics, query} -> {dynamic, acc} = AshSql.Expr.dynamic_expr(query, expression, query.__ash_bindings__) dynamic = case dynamic do %Ecto.Query.DynamicExpr{} -> dynamic other -> Ecto.Query.dynamic(^other) end {[dynamic | dynamics], AshSql.Bindings.merge_expr_accumulator(query, acc)} end) dynamics = dynamics |> Enum.with_index() |> Map.new(fn {dynamic, index} -> {index, dynamic} end) query = Ecto.Query.from(row in fragment("(VALUES(1))"), select: ^dynamics) |> Map.put(:__ash_bindings__, query.__ash_bindings__) repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, ash_query) with_savepoint(repo, query, fn -> {:ok, repo.one(query) |> Enum.sort_by(&elem(&1, 0)) |> Enum.map(&elem(&1, 1)) |> Enum.reverse()} end) rescue e -> handle_raised_error(e, __STACKTRACE__, query, resource) end end @impl true def bulk_create(resource, stream, options) do changesets = Enum.to_list(stream) repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, Enum.at(changesets, 0)) opts = AshSql.repo_opts(repo, AshPostgres.SqlImplementation, nil, options[:tenant], resource) opts = if options.return_records? do returning = case options[:action_select] do nil -> true [] -> Ash.Resource.Info.primary_key(resource) fields -> fields end Keyword.put(opts, :returning, returning) else opts end source = resolve_source(resource, Enum.at(changesets, 0)) # On PostgreSQL 17+ we implement upserts with `MERGE` instead of `INSERT ... ON CONFLICT`, # which lets us report per-row whether each record was inserted or updated (via # `merge_action()`), surfaced as `:upsert_action` metadata. Below 17, when explicitly # disabled, or for upserts whose keys cannot be matched by `MERGE`, we fall back to the # `ON CONFLICT` path. use_merge? = options[:upsert?] && merge_upsert?(resource, options) try do opts = if options[:upsert?] && !use_merge? do # Ash groups changesets by atomics before dispatching them to the data layer # this means that all changesets have the same atomics %{atomics: atomics, filter: filter} = Enum.at(changesets, 0) query = from(row in source, as: ^0) query = query |> AshSql.Bindings.default_bindings(resource, AshPostgres.SqlImplementation) upsert_set = upsert_set( resource, changesets, options[:upsert_keys] || Ash.Resource.Info.primary_key(resource), options ) on_conflict = case AshSql.Atomics.query_with_atomics( resource, query, filter, atomics, %{}, upsert_set ) do {:empty, _query} -> raise "Cannot upsert with no fields to specify in the upsert statement. This can only happen on resources without a primary key." {:ok, query} -> query {:error, error} -> raise Ash.Error.to_ash_error(error) end opts |> Keyword.put(:on_conflict, on_conflict) |> Keyword.put( :conflict_target, conflict_target( resource, options[:identity], options[:upsert_keys] || Ash.Resource.Info.primary_key(resource) ) ) else opts end create_atomics = Map.get(Enum.at(changesets, 0), :create_atomics, []) atomic_insert_values = if create_atomics != [] do # Hydrate expressions to convert Ash.Query.Call structs to proper function structs create_atomics = Enum.map(create_atomics, fn {key, expr} -> case Ash.Filter.hydrate_refs(expr, %{resource: resource, public?: false}) do {:ok, hydrated_expr} -> {key, hydrated_expr} {:error, error} -> raise Ash.Error.to_ash_error(error) end end) query = from(row in source, as: ^0) query = query |> AshSql.Bindings.default_bindings(resource, AshPostgres.SqlImplementation) case AshSql.Atomics.atomics_to_insert_values(resource, query, create_atomics) do {:ok, values} -> values {:error, error} -> raise Ash.Error.to_ash_error(error) end else %{} end ecto_changesets = Enum.map(changesets, fn cs -> Map.merge(cs.attributes, atomic_insert_values) end) opts = if schema = Enum.at(changesets, 0).context[:data_layer][:schema] do Keyword.put(opts, :prefix, schema) else opts end result = if use_merge? do merge_upsert(repo, resource, source, changesets, ecto_changesets, opts, options) else with_savepoint(repo, opts[:on_conflict], fn -> repo.insert_all(source, ecto_changesets, opts) end) end identity = options[:identity] keys = Map.get(identity || %{}, :keys) || Ash.Resource.Info.primary_key(resource) # if it's single the return_skipped_upsert? is handled at the # call site https://github.com/ash-project/ash_postgres/blob/0b21d4a99cc3f6d8676947e291ac9b9d57ad6e2e/lib/data_layer.ex#L3046-L3046 result = if options[:upsert?] && options[:return_skipped_upsert?] && !opts[:single?] do [changeset | _] = changesets results_by_identity = result |> elem(1) |> List.wrap() |> Enum.into(%{}, fn r -> {Map.take(r, keys), r} end) skipped_filter = changesets |> Enum.filter(fn changeset -> not Map.has_key?( results_by_identity, Map.take(changeset.attributes, keys) ) end) |> Enum.map(fn changeset -> changeset.attributes |> Map.take(keys) |> Keyword.new() end) skipped_upserts = case skipped_filter do [] -> # No skipped records to query for %{} skipped_filter -> ash_query = resource |> Ash.Query.do_filter(or: skipped_filter) |> then(fn query when is_nil(identity) or is_nil(identity.where) -> query query -> Ash.Query.do_filter(query, identity.where) end) |> Ash.Query.set_tenant(changeset.tenant) with {:ok, ecto_query} <- Ash.Query.data_layer_query(ash_query), {:ok, results} <- run_query(ecto_query, resource) do results |> Enum.map(fn result -> Ash.Resource.put_metadata(result, :upsert_skipped, true) end) |> Enum.reduce(%{}, fn r, acc -> Map.put(acc, Map.take(r, keys), r) end) end end results = changesets |> Enum.map(fn changeset -> identity = changeset.attributes |> Map.take(keys) Map.get(results_by_identity, identity, Map.get(skipped_upserts, identity)) end) |> Enum.filter(& &1) {length(results), results} else result end case result do {_, nil} -> :ok {_, results} -> if options[:single?] do Enum.each(results, &maybe_create_tenant!(resource, &1)) {:ok, results} else results_by_identity = results |> Enum.into(%{}, fn r -> {Map.take(r, keys), r} end) results = if opts[:upsert?] do changesets |> Enum.map(fn changeset -> identity = changeset.attributes |> Map.take(keys) result_for_changeset = Map.get(results_by_identity, identity) if result_for_changeset do if !opts[:upsert?] do maybe_create_tenant!(resource, result_for_changeset) end # Compatibility fallback Ash.Resource.put_metadata( result_for_changeset, :bulk_action_ref, changeset.context[:bulk_create][:ref] ) end end) |> Enum.filter(& &1) else results |> Enum.zip(changesets) |> Enum.map(fn {result, changeset} -> if !opts[:upsert?] do maybe_create_tenant!(resource, result) end # Compatibility fallback Ash.Resource.put_metadata( result, :bulk_action_ref, changeset.context[:bulk_create][:ref] ) end) end {:ok, results} end end rescue e -> changeset = case source do {table, resource} -> resource |> Ash.Changeset.new() |> Ash.Changeset.put_context(:data_layer, %{table: table}) resource -> resource |> Ash.Changeset.new() end handle_raised_error( e, __STACKTRACE__, {:bulk_create, ecto_changeset(changeset.data, changeset, :create, repo, false)}, resource ) end end @impl true @doc false # Updates many records, each with its own changes, as a single SQL MERGE (PostgreSQL 17+). # Each changeset targets one record by primary key (`changeset.data`); `changeset.attributes` # are this record's changes and `changeset.atomics` are shared across the batch. Records that # are no longer present are simply not returned (`WHEN NOT MATCHED THEN DO NOTHING`); the action # layer diffs the returned records against the changesets to surface NotFound/Stale errors. def update_many(resource, changesets, options) do changesets = Enum.to_list(changesets) changeset = Enum.at(changesets, 0) repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, changeset) opts = AshSql.repo_opts(repo, AshPostgres.SqlImplementation, nil, options[:tenant], resource) opts = if options[:return_records?] do returning = case options[:action_select] do nil -> true [] -> Ash.Resource.Info.primary_key(resource) fields -> fields end Keyword.put(opts, :returning, returning) else opts end source = resolve_source(resource, changeset) pkey = Ash.Resource.Info.primary_key(resource) %{atomics: atomics, filter: filter} = changeset try do # Each entry carries the primary key (for matching) plus this record's changed attributes. entries = Enum.map(changesets, fn cs -> Map.merge(Map.take(cs.data, pkey), cs.attributes) end) {set_values, present_columns} = update_many_set(resource, changesets) base_query = from(row in source, as: ^0) |> AshSql.Bindings.default_bindings(resource, AshPostgres.SqlImplementation) set_query = case AshSql.Atomics.query_with_atomics( resource, base_query, nil, atomics, %{}, set_values ) do {:empty, _query} -> raise "Cannot `update_many` with no changes." {:ok, query} -> query {:error, error} -> raise Ash.Error.to_ash_error(error) end when_matched_condition_query = if filter, do: merge_filter_query(resource, source, filter) {on_query, source_fields} = build_merge_on_query(resource, source, pkey, nil, changeset) table = case source do {table, _resource} -> table _ -> AshPostgres.DataLayer.Info.table(resource) end opts = if schema = changeset.context[:data_layer][:schema] do Keyword.put(opts, :prefix, schema) else opts end savepoint_query = combine_expression_accumulators(set_query, [when_matched_condition_query, on_query]) result = with_savepoint(repo, savepoint_query, fn -> AshPostgres.Merge.merge_all(repo, resource: resource, table: table, prefix: opts[:prefix], entries: entries, on_query: on_query, source_fields: source_fields, extra_source_columns: present_columns, set_query: set_query, when_matched_condition_query: when_matched_condition_query, on_not_matched: :do_nothing, returning: opts[:returning], report_action?: true ) end) case result do {_, nil} -> :ok {_, records} -> {:ok, records} end rescue e -> handle_raised_error( e, __STACKTRACE__, {:bulk_create, ecto_changeset(changeset.data, changeset, :update, repo, false)}, resource ) end end # Single pass over the batch: an attribute set by *every* changeset is copied straight from the # source (`col = EXCLUDED.col`); an attribute set by only *some* is gated by a per-row present? # flag (`col = CASE WHEN EXCLUDED."col__present" THEN EXCLUDED.col ELSE t.col END`) so a record # that didn't set it keeps its current value (and a record that set it to NULL still gets NULL). # sobelow_skip ["DOS.StringToAtom"] defp update_many_set(resource, changesets) do total = length(changesets) counts = Enum.reduce(changesets, %{}, fn changeset, counts -> Enum.reduce(Map.keys(changeset.attributes), counts, fn field, counts -> Map.update(counts, field, 1, &(&1 + 1)) end) end) Enum.reduce(counts, {[], []}, fn {field, count}, {set_values, present_columns} -> source_col = to_string(get_source_for_upsert_field(field, resource)) if count == total do dynamic = Ecto.Query.dynamic([], fragment("EXCLUDED.?", identifier(^source_col))) {[{field, dynamic} | set_values], present_columns} else present_name = String.to_atom(source_col <> "__present") present_ref = to_string(present_name) dynamic = Ecto.Query.dynamic( [row], fragment( "CASE WHEN EXCLUDED.? THEN EXCLUDED.? ELSE ? END", identifier(^present_ref), identifier(^source_col), field(row, ^field) ) ) present_column = %{ name: present_name, type: :boolean, values: Enum.map(changesets, &Map.has_key?(&1.attributes, field)) } {[{field, dynamic} | set_values], [present_column | present_columns]} end end) end defp with_savepoint( repo, %{ __ash_bindings__: %{ expression_accumulator: %AshSql.Expr.ExprInfo{has_error?: true} } }, fun ) do if repo.in_transaction?() do savepoint_id = "a" <> (Ash.UUID.generate() |> String.replace("-", "_")) repo.query!("SAVEPOINT #{savepoint_id}") result = try do {:ok, fun.()} rescue e in Postgrex.Error -> case e do %Postgrex.Error{ postgres: %{ code: :raise_exception, message: "ash_error:" <> _, severity: "ERROR" } } -> repo.query!("ROLLBACK TO #{savepoint_id}") # This kind of exception won't trigger # a rollback {:exception, e, __STACKTRACE__} _ -> {:exception, e, __STACKTRACE__} end end case result do {:exception, e, stacktrace} -> reraise e, stacktrace {:ok, result} -> repo.query!("RELEASE #{savepoint_id}") result end else fun.() end end defp with_savepoint(_repo, _acc, fun) do fun.() end defp upsert_set(resource, changesets, keys, options) do attributes_changing_anywhere = changesets |> Enum.flat_map(&Map.keys(&1.attributes)) |> Enum.uniq() update_defaults = update_defaults(resource) # We can't reference EXCLUDED if at least one of the changesets in the stream is not # changing the value (and we wouldn't want to even if we could as it would be unnecessary) upsert_fields = (options[:upsert_fields] || []) |> Enum.filter(&(&1 in attributes_changing_anywhere)) fields_to_upsert = upsert_fields -- (Keyword.keys(Enum.at(changesets, 0).atomics) -- keys) fields_to_upsert = case fields_to_upsert do [] -> keys fields_to_upsert -> # Include fields with update_defaults (e.g. update_timestamp) # even if they aren't in the changeset attributes or upsert_fields. # These fields should always be refreshed when an upsert modifies fields. # Can be disabled via touch_update_defaults?: false in the changeset # context (either in [:private] or [:data_layer]) or via options map touch_update_defaults? = Map.get(options, :touch_update_defaults?, true) && Enum.at(changesets, 0).context[:private][:touch_update_defaults?] != false && Enum.at(changesets, 0).context[:data_layer][:touch_update_defaults?] != false if touch_update_defaults? do update_default_fields = update_defaults |> Keyword.keys() |> Enum.reject(&(&1 in fields_to_upsert or &1 in keys)) fields_to_upsert ++ update_default_fields else fields_to_upsert end end fields_to_upsert |> Enum.uniq() |> Enum.map(fn upsert_field -> # for safety, we check once more at the end that all values in # upsert_fields are names of attributes. This is because # below we use `identifier/1` to bring them into the query if is_nil(resource.__schema__(:type, upsert_field)) do raise "Only attribute names can be used in upsert_fields" end case Keyword.fetch(update_defaults, upsert_field) do {:ok, default} -> if upsert_field in upsert_fields do {upsert_field, Ecto.Query.dynamic( [], fragment( "COALESCE(EXCLUDED.?, ?)", identifier(^to_string(get_source_for_upsert_field(upsert_field, resource))), ^default ) )} else {upsert_field, default} end :error -> {upsert_field, Ecto.Query.dynamic( [], fragment( "EXCLUDED.?", identifier(^to_string(get_source_for_upsert_field(upsert_field, resource))) ) )} end end) end defp merge_upsert?(resource, _options) do Application.get_env(:ash_postgres, :upsert_with_merge?, true) != false && AshPostgres.DataLayer.Info.pg_version_matches?(resource, ">= 17.0.0") end # Implements an upsert as a single `MERGE` statement (PostgreSQL 17+). Returns the same # `{count, records | nil}` shape as `repo.insert_all/3` so the downstream result handling # in `bulk_create/3` is shared between both paths. defp merge_upsert(repo, resource, source, changesets, ecto_changesets, opts, options) do changeset = Enum.at(changesets, 0) %{atomics: atomics, filter: filter} = changeset upsert_keys = options[:upsert_keys] || Ash.Resource.Info.primary_key(resource) base_query = from(row in source, as: ^0) |> AshSql.Bindings.default_bindings(resource, AshPostgres.SqlImplementation) upsert_set = upsert_set(resource, changesets, upsert_keys, options) set_query = case AshSql.Atomics.query_with_atomics(resource, base_query, nil, atomics, %{}, upsert_set) do {:empty, _query} -> raise "Cannot upsert with no fields to specify in the upsert statement. This can only happen on resources without a primary key." {:ok, query} -> query {:error, error} -> raise Ash.Error.to_ash_error(error) end # The upsert condition (which may reference the incoming row via `EXCLUDED`) becomes # `WHEN MATCHED AND ...`. when_matched_condition_query = if filter do merge_filter_query(resource, source, filter) end # The full `ON` clause: each key matched as ` = `, combined # with the resource's base_filter and the identity's `where` (reproducing partial-unique-index # matching semantics). `source_fields` are the attributes the incoming side references and which # therefore must exist on the `EXCLUDED` source even if these entries don't set them. {on_query, source_fields} = build_merge_on_query(resource, source, upsert_keys, options[:identity], changeset) table = case source do {table, _resource} -> table _ -> AshPostgres.DataLayer.Info.table(resource) end # Any of the rendered sub-queries (the SET expressions, the upsert condition, or the ON # clause) can contain ash expressions that raise at runtime. Their `has_error?` flags live # on separate query structs, so fold them together to decide whether to savepoint. savepoint_query = combine_expression_accumulators(set_query, [when_matched_condition_query, on_query]) with_savepoint(repo, savepoint_query, fn -> AshPostgres.Merge.merge_all(repo, resource: resource, table: table, prefix: opts[:prefix], entries: ecto_changesets, on_query: on_query, source_fields: source_fields, set_query: set_query, when_matched_condition_query: when_matched_condition_query, on_not_matched: :insert, returning: opts[:returning], report_action?: true ) end) end # Returns `base_query` with its `has_error?` accumulator flag set if any of `base_query` or # `others` flagged a potentially-raising expression, so a single `with_savepoint/3` call # covers every sub-query rendered into the MERGE statement. defp combine_expression_accumulators(base_query, others) do has_error? = [base_query | others] |> Enum.reject(&is_nil/1) |> Enum.any?(& &1.__ash_bindings__.expression_accumulator.has_error?) put_in(base_query.__ash_bindings__.expression_accumulator.has_error?, has_error?) end # Builds a `SELECT 1 ... WHERE ` query whose rendered WHERE clause AshPostgres.Merge # extracts and splices into a MERGE condition. Built from the same `from(row in source, as: ^0)` # shape as the SET query so Ecto emits a consistent target alias across all clauses. defp merge_filter_query(resource, source, filter) do query = from(row in source, as: ^0) |> AshSql.Bindings.default_bindings(resource, AshPostgres.SqlImplementation) case AshSql.Filter.filter(query, filter, resource) do {:ok, query} -> Ecto.Query.select(query, 1) {:error, error} -> raise Ash.Error.to_ash_error(error) end end # Builds the rendered `ON` query for a MERGE upsert: a key-match expression for each upsert key, # combined with the identity's `where` (and the resource base_filter, applied by `Ash.Query.new/1`). # Also returns the attributes the incoming (`EXCLUDED`) side references, which must be present on # the source. defp build_merge_on_query(resource, source, keys, identity, changeset) do nils_distinct? = if identity, do: identity.nils_distinct?, else: true identity_where = identity && identity.where {match_expressions, source_fields} = Enum.map_reduce(keys, [], fn key, acc -> {expression, fields} = key_match_expression(resource, key, nils_distinct?, changeset) {expression, fields ++ acc} end) match_expression = Enum.reduce(match_expressions, fn expression, acc -> Ash.Expr.expr(^acc and ^expression) end) filter = resource |> Ash.Query.new() |> Ash.Query.do_filter(match_expression) |> then(fn query -> if identity_where, do: Ash.Query.do_filter(query, identity_where), else: query end) |> Map.get(:filter) {merge_filter_query(resource, source, filter), Enum.uniq(source_fields)} end # ` == `. For `nils_distinct?: false` # identities (NULLS NOT DISTINCT) two NULL keys must also match, so widen to an # `IS NOT DISTINCT FROM` equivalent. Returns the match expression and the attributes its # incoming side references. defp key_match_expression(resource, key, nils_distinct?, changeset) do {lhs, rhs, source_fields} = key_match_sides(resource, key, changeset) expression = if nils_distinct? do Ash.Expr.expr(^lhs == ^rhs) else Ash.Expr.expr(^lhs == ^rhs or (is_nil(^lhs) and is_nil(^rhs))) end {expression, source_fields} end # Produces `{existing-row expr, incoming-row expr, referenced source attributes}` for a key. The # incoming side references the conflicting row via `upsert_conflict/1` (rendered as `EXCLUDED.col`). # Calculation keys are resolved to a proper `Ash.Query.Calculation` (with context) and matched as # expressions on both sides. defp key_match_sides(resource, key, changeset) do case Ash.Resource.Info.field(resource, key) do %Ash.Resource.Attribute{} = attribute -> {%Ash.Query.Ref{attribute: attribute, relationship_path: [], resource: resource}, %Ash.Query.UpsertConflict{attribute: key}, [key]} %Ash.Resource.Calculation{calculation: {module, opts}} = resource_calculation -> {:ok, calculation} = Ash.Query.Calculation.new( resource_calculation.name, module, opts, resource_calculation.type, resource_calculation.constraints ) calculation = Ash.Actions.Read.add_calc_context( calculation, changeset.context[:private][:actor], changeset.context[:private][:authorize?], changeset.tenant, changeset.context[:private][:tracer], changeset.domain, resource, parent_stack: [] ) resolved = calculation.module.expression(calculation.opts, calculation.context) lhs = %Ash.Query.Ref{attribute: calculation, relationship_path: [], resource: resource} rhs = rewrite_to_incoming_row(resolved) {lhs, rhs, calculation_source_fields(resolved)} other -> raise ArgumentError, "Cannot use #{inspect(key)} as a MERGE upsert key: unsupported field #{inspect(other && other.__struct__)}" end end # Rewrites a calculation's (local) attribute references to reference the incoming/conflicting row. defp rewrite_to_incoming_row(expression) do Ash.Filter.map(expression, fn %Ash.Query.Ref{relationship_path: [], attribute: attribute} -> %Ash.Query.UpsertConflict{attribute: merge_field_name(attribute)} %Ash.Query.Ref{} = ref -> raise ArgumentError, "Calculation MERGE upsert key references a relationship (#{inspect(ref.relationship_path)}), which cannot be matched in a single MERGE statement" other -> other end) end # The local attributes a calculation key depends on; they must exist on the `EXCLUDED` source. defp calculation_source_fields(expression) do expression |> Ash.Filter.list_refs() |> Enum.filter(&(&1.relationship_path == [])) |> Enum.map(&merge_field_name(&1.attribute)) end defp merge_field_name(%{name: name}), do: name defp merge_field_name(name) when is_atom(name), do: name @doc false def get_source_for_upsert_field(field, resource) do case Ash.Resource.Info.attribute(resource, field) do %{source: source} when not is_nil(source) -> source _ -> field end end @impl true def create(resource, changeset) do changeset = %{ changeset | data: Map.update!( changeset.data, :__meta__, &Map.put(&1, :source, table(resource, changeset)) ) } case bulk_create(resource, [changeset], %{ single?: true, tenant: Map.get(changeset, :to_tenant, changeset.tenant), action_select: changeset.action_select, return_records?: true }) do {:ok, [result]} -> {:ok, result} {:error, error} -> {:error, error} {:error, :no_rollback, error} -> {:error, :no_rollback, error} end end defp maybe_create_tenant!(resource, result) do if AshPostgres.DataLayer.Info.manage_tenant_create?(resource) do tenant_name = tenant_name(resource, result) AshPostgres.MultiTenancy.create_tenant!( tenant_name, AshPostgres.DataLayer.Info.repo(resource, :read) ) else :ok end end defp maybe_update_tenant(resource, changeset, result) do if AshPostgres.DataLayer.Info.manage_tenant_update?(resource) do changing_tenant_name? = resource |> AshPostgres.DataLayer.Info.manage_tenant_template() |> Enum.filter(&is_atom/1) |> Enum.any?(&Ash.Changeset.changing_attribute?(changeset, &1)) if changing_tenant_name? do old_tenant_name = tenant_name(resource, changeset.data) new_tenant_name = tenant_name(resource, result) AshPostgres.MultiTenancy.rename_tenant( AshPostgres.DataLayer.Info.repo(resource, :read), old_tenant_name, new_tenant_name ) end end :ok end defp tenant_name(resource, result) do resource |> AshPostgres.DataLayer.Info.manage_tenant_template() |> Enum.map_join(fn item -> if is_binary(item) do item else result |> Map.get(item) |> to_string() end end) end defp ecto_changeset(record, changeset, type, repo, table_error?) do attributes = changeset.resource |> Ash.Resource.Info.attributes() |> Enum.map(& &1.name) attributes_to_change = Enum.reject(attributes, fn attribute -> Keyword.has_key?(changeset.atomics, attribute) end) ecto_changeset = record |> to_ecto() |> set_table(changeset, type, table_error?) |> Ecto.Changeset.cast(%{}, []) |> force_changes(Map.take(changeset.attributes, attributes_to_change)) |> add_configured_foreign_key_constraints(record.__struct__, repo) |> add_unique_indexes(record.__struct__, changeset, repo) |> add_check_constraints(record.__struct__, repo) |> add_exclusion_constraints(record.__struct__, repo) case type do :create -> ecto_changeset |> add_my_foreign_key_constraints(record.__struct__, repo) type when type in [:upsert, :update] -> ecto_changeset |> add_my_foreign_key_constraints(record.__struct__, repo) |> add_related_foreign_key_constraints(record.__struct__, repo) :delete -> ecto_changeset |> add_related_foreign_key_constraints(record.__struct__, repo) end end defp force_changes(changeset, changes) do Enum.reduce(changes, changeset, fn {key, value}, changeset -> Ecto.Changeset.force_change(changeset, key, value) end) end defp handle_raised_error( %Ecto.StaleEntryError{changeset: %{data: %resource{}, filters: filters}}, stacktrace, context, resource ) do handle_raised_error( Ash.Error.Changes.StaleRecord.exception(resource: resource, filter: filters), stacktrace, context, resource ) end defp handle_raised_error( %Postgrex.Error{ postgres: %{ code: :lock_not_available, message: message } }, stacktrace, context, resource ) do handle_raised_error( Ash.Error.Invalid.Unavailable.exception( resource: resource, source: inspect(context, pretty: true), reason: message ), stacktrace, context, resource ) end defp handle_raised_error( %Postgrex.Error{ postgres: %{ code: :raise_exception, message: "ash_error: \"" <> json, severity: "ERROR" } }, _, _, _ ) do %{"exception" => exception, "input" => input} = json |> String.trim_trailing("\"") |> String.replace("\\\"", "\"") |> Jason.decode!() exception = Module.concat([exception]) {:error, :no_rollback, Ash.Error.from_json(exception, input)} end defp handle_raised_error( %Postgrex.Error{ postgres: %{ code: :raise_exception, message: "ash_error: " <> json, severity: "ERROR" } }, _, _, _ ) do %{"exception" => exception, "input" => input} = Jason.decode!(json) exception = Module.concat([exception]) {:error, :no_rollback, Ash.Error.from_json(exception, input)} end defp handle_raised_error( %Postgrex.Error{} = error, stacktrace, {:bulk_create, fake_changeset}, resource ) do case Ecto.Adapters.Postgres.Connection.to_constraints(error, []) do [] -> {:error, Ash.Error.to_ash_error(error, stacktrace)} constraints -> {:error, fake_changeset |> constraints_to_errors(:insert, constraints, resource, error) |> Ash.Error.to_ash_error()} end end defp handle_raised_error( %Postgrex.Error{} = error, stacktrace, {:ecto_changeset, action, changeset}, resource ) when action in [:insert, :update, :delete] do handle_postgrex_error(error, stacktrace, changeset, resource, action) end defp handle_raised_error(%Ecto.Query.CastError{} = e, stacktrace, context, resource) do handle_raised_error( Ash.Error.Query.InvalidFilterValue.exception(value: e.value, context: context), stacktrace, context, resource ) end defp handle_raised_error( %Postgrex.Error{} = error, stacktrace, changeset, resource ) do handle_postgrex_error(error, stacktrace, changeset, resource, :insert) end defp handle_raised_error(error, stacktrace, _ecto_changeset, _resource) do {:error, Ash.Error.to_ash_error(error, stacktrace)} end defp handle_postgrex_error(error, stacktrace, changeset, resource, action) do case Ecto.Adapters.Postgres.Connection.to_constraints(error, []) do [] -> constraints = maybe_foreign_key_violation_constraints(error) if constraints != [] do {:error, changeset |> constraints_to_errors(action, constraints, resource, error) |> Ash.Error.to_ash_error()} else {:error, Ash.Error.to_ash_error(error, stacktrace)} end constraints -> {:error, changeset |> constraints_to_errors(action, constraints, resource, error) |> Ash.Error.to_ash_error()} end end defp maybe_foreign_key_violation_constraints(%Postgrex.Error{postgres: postgres}) when is_map(postgres) do code = postgres[:code] || postgres["code"] constraint = postgres[:constraint] || postgres["constraint"] if code in ["23503", 23_503, :foreign_key_violation] and is_binary(constraint) do [{:foreign_key, constraint}] else [] end end defp maybe_foreign_key_violation_constraints(_), do: [] defp constraints_to_errors( %{constraints: user_constraints} = changeset, action, constraints, resource, error ) do Enum.map(constraints, fn {type, constraint} -> user_constraint = Enum.find(user_constraints, fn c -> case {c.type, c.constraint, c.match} do {^type, ^constraint, :exact} -> true {^type, cc, :suffix} -> String.ends_with?(constraint, cc) {^type, cc, :prefix} -> String.starts_with?(constraint, cc) {^type, %Regex{} = r, _match} -> Regex.match?(r, constraint) _ -> false end end) case user_constraint do %{field: field, error_message: error_message, type: type, constraint: constraint} -> identities = Ash.Resource.Info.identities(resource) table = AshPostgres.DataLayer.Info.table(resource) identity = Enum.find(identities, fn identity -> "#{table}_#{identity.name}_index" == constraint end) field_names = if identity, do: identity.field_names || [field], else: [field] Enum.map(field_names, fn field_name -> Ash.Error.Changes.InvalidAttribute.exception( field: field_name, message: error_message, private_vars: [ constraint: constraint, constraint_type: type, detail: error.postgres.detail ] ) end) nil -> Ecto.ConstraintError.exception( action: action, type: type, constraint: constraint, changeset: changeset ) end end) end defp set_table(record, changeset, operation, table_error?) do if AshPostgres.DataLayer.Info.polymorphic?(record.__struct__) do table = changeset.context[:data_layer][:table] || AshPostgres.DataLayer.Info.table(record.__struct__) record = if table do Ecto.put_meta(record, source: table) else if table_error? do raise_table_error!(changeset.resource, operation) else record end end prefix = changeset.context[:data_layer][:schema] || AshPostgres.DataLayer.Info.schema(record.__struct__) if prefix do Ecto.put_meta(record, prefix: table) else record end else record end end def from_ecto({:ok, result}), do: {:ok, from_ecto(result)} def from_ecto({:error, _} = other), do: other def from_ecto(nil), do: nil def from_ecto(value) when is_list(value) do Enum.map(value, &from_ecto/1) end def from_ecto(%resource{} = record) do if Spark.Dsl.is?(resource, Ash.Resource) do empty = struct(resource) resource |> Ash.Resource.Info.relationships() |> Enum.reduce(record, fn relationship, record -> case Map.get(record, relationship.name) do %Ecto.Association.NotLoaded{} -> Map.put(record, relationship.name, Map.get(empty, relationship.name)) value -> Map.put(record, relationship.name, from_ecto(value)) end end) else record end end def from_ecto(other), do: other def to_ecto(nil), do: nil def to_ecto(value) when is_list(value) do Enum.map(value, &to_ecto/1) end def to_ecto(%resource{} = record) do if Spark.Dsl.is?(resource, Ash.Resource) do resource |> Ash.Resource.Info.relationships() |> Enum.reduce(record, fn relationship, record -> value = case Map.get(record, relationship.name) do %Ash.NotLoaded{} -> %Ecto.Association.NotLoaded{ __field__: relationship.name, __cardinality__: relationship.cardinality } value -> to_ecto(value) end Map.put(record, relationship.name, value) end) else record end end def to_ecto(other), do: other defp add_check_constraints(changeset, resource, repo) do resource |> AshPostgres.DataLayer.Info.check_constraints() |> Enum.reduce(changeset, fn constraint, changeset -> constraint.attribute |> List.wrap() |> Enum.reduce(changeset, fn attribute, changeset -> case repo.default_constraint_match_type(:check, constraint.name) do {:regex, regex} -> Ecto.Changeset.check_constraint(changeset, attribute, name: regex, message: constraint.message || "is invalid", match: :exact ) match -> Ecto.Changeset.check_constraint(changeset, attribute, name: constraint.name, message: constraint.message || "is invalid", match: match ) end end) end) end defp add_exclusion_constraints(changeset, resource, repo) do resource |> AshPostgres.DataLayer.Info.exclusion_constraint_names() |> Enum.reduce(changeset, fn constraint, changeset -> case constraint do {key, name} -> case repo.default_constraint_match_type(:check, name) do {:regex, regex} -> Ecto.Changeset.exclusion_constraint(changeset, key, name: regex, match: :exact ) match -> Ecto.Changeset.exclusion_constraint(changeset, key, name: name, match: match ) end {key, name, message} -> case repo.default_constraint_match_type(:check, name) do {:regex, regex} -> Ecto.Changeset.exclusion_constraint(changeset, key, name: regex, message: message, match: :exact ) match -> Ecto.Changeset.exclusion_constraint(changeset, key, name: name, message: message, match: match ) end end end) end defp add_related_foreign_key_constraints(changeset, resource, repo) do # TODO: this doesn't guarantee us to get all of them, because if something is related to this # schema and there is no back-relation, then this won't catch it's foreign key constraints resource |> Ash.Resource.Info.relationships() |> Enum.map(& &1.destination) |> Enum.uniq() |> Enum.flat_map(fn related -> related |> Ash.Resource.Info.relationships() |> Enum.filter(&(&1.destination == resource)) |> Enum.map(&Map.take(&1, [:source, :source_attribute, :destination_attribute, :name])) end) |> Enum.reduce(changeset, fn %{ source: source, source_attribute: source_attribute, destination_attribute: destination_attribute, name: relationship_name }, changeset -> case AshPostgres.DataLayer.Info.reference(resource, relationship_name) do %{name: name} when not is_nil(name) -> case repo.default_constraint_match_type(:foreign, name) do {:regex, regex} -> Ecto.Changeset.foreign_key_constraint(changeset, destination_attribute, name: regex, message: "would leave records behind", match: :exact ) match -> Ecto.Changeset.foreign_key_constraint(changeset, destination_attribute, name: name, message: "would leave records behind", match: match ) end _ -> name = "#{AshPostgres.DataLayer.Info.table(source)}_#{source_attribute}_fkey" case repo.default_constraint_match_type(:foreign, name) do {:regex, regex} -> Ecto.Changeset.foreign_key_constraint(changeset, destination_attribute, name: regex, message: "would leave records behind", match: :exact ) match -> Ecto.Changeset.foreign_key_constraint(changeset, destination_attribute, name: name, message: "would leave records behind", match: match ) end end end) end defp add_my_foreign_key_constraints(changeset, resource, repo) do resource |> Ash.Resource.Info.relationships() |> Enum.reduce(changeset, fn relationship, changeset -> # Check if there's a custom reference name defined in the DSL name = case AshPostgres.DataLayer.Info.reference(resource, relationship.name) do %{name: custom_name} when not is_nil(custom_name) -> custom_name _ -> "#{AshPostgres.DataLayer.Info.table(resource)}_#{relationship.source_attribute}_fkey" end case repo.default_constraint_match_type(:foreign, name) do {:regex, regex} -> Ecto.Changeset.foreign_key_constraint(changeset, relationship.source_attribute, name: regex, match: :exact ) match -> Ecto.Changeset.foreign_key_constraint(changeset, relationship.source_attribute, name: name, match: match ) end end) end defp add_configured_foreign_key_constraints(changeset, resource, repo) do resource |> AshPostgres.DataLayer.Info.foreign_key_names() |> case do {m, f, a} -> List.wrap(apply(m, f, [changeset | a])) value -> List.wrap(value) end |> Enum.reduce(changeset, fn {key, name}, changeset -> case repo.default_constraint_match_type(:foreign, name) do {:regex, regex} -> Ecto.Changeset.foreign_key_constraint(changeset, key, name: regex, match: :exact ) match -> Ecto.Changeset.foreign_key_constraint(changeset, key, name: name, match: match ) end {key, name, message}, changeset -> case repo.default_constraint_match_type(:foreign, name) do {:regex, regex} -> Ecto.Changeset.foreign_key_constraint(changeset, key, name: regex, message: message, match: :exact ) match -> Ecto.Changeset.foreign_key_constraint(changeset, key, name: name, message: message, match: match ) end end) end defp add_unique_indexes(changeset, resource, ash_changeset, repo) do table = table(resource, ash_changeset) pkey = Ash.Resource.Info.primary_key(resource) changeset = resource |> Ash.Resource.Info.identities() |> Enum.reduce(changeset, fn identity, changeset -> name = AshPostgres.DataLayer.Info.identity_index_names(resource)[identity.name] || "#{table}_#{identity.name}_index" opts = case repo.default_constraint_match_type(:unique, name) do {:regex, regex} -> if Map.get(identity, :message) do [name: regex, message: identity.message, match: :exact] else [name: regex, match: :exact] end index_match_type -> if Map.get(identity, :message) do [name: name, message: identity.message, match: index_match_type] else [name: name, match: index_match_type] end end fields = case identity.keys do [] -> pkey keys -> keys end Ecto.Changeset.unique_constraint(changeset, fields, opts) end) changeset = resource |> AshPostgres.DataLayer.Info.custom_indexes() |> Enum.reduce(changeset, fn index, changeset -> name = index.name || AshPostgres.CustomIndex.name(table, index) opts = case repo.default_constraint_match_type(:custom, name) do {:regex, regex} -> if Map.get(index, :message) do [name: regex, message: index.message, match: :exact] else [name: regex, match: :exact] end index_match_type -> if Map.get(index, :message) do [name: name, message: index.message, match: index_match_type] else [name: name, match: index_match_type] end end fields = if index.error_fields do case index.error_fields do [] -> pkey fields -> fields end else case Enum.filter(index.fields, &is_atom/1) do [] -> pkey fields -> fields end end Ecto.Changeset.unique_constraint(changeset, fields, opts) end) names = resource |> AshPostgres.DataLayer.Info.unique_index_names() |> case do {m, f, a} -> List.wrap(apply(m, f, [changeset | a])) value -> List.wrap(value) end names = case Ash.Resource.Info.primary_key(resource) do [] -> names fields -> if table = table(resource, ash_changeset) do [{fields, table <> "_pkey"} | names] else [] end end Enum.reduce(names, changeset, fn {keys, name}, changeset -> case repo.default_constraint_match_type(:unique, name) do {:regex, regex} -> Ecto.Changeset.unique_constraint(changeset, List.wrap(keys), name: regex, match: :exact ) match -> Ecto.Changeset.unique_constraint(changeset, List.wrap(keys), name: name, match: match ) end {keys, name, message}, changeset -> case repo.default_constraint_match_type(:unique, name) do {:regex, regex} -> Ecto.Changeset.unique_constraint(changeset, List.wrap(keys), name: regex, message: message, match: :exact ) match -> Ecto.Changeset.unique_constraint(changeset, List.wrap(keys), name: name, message: message, match: match ) end end) end @impl true def upsert(resource, changeset, keys, identity) do if AshPostgres.DataLayer.Info.manage_tenant_update?(resource) do {:error, "Cannot currently upsert a resource that owns a tenant"} else keys = keys || Ash.Resource.Info.primary_key(keys) touch_update_defaults? = changeset.context[:private][:touch_update_defaults?] != false update_defaults = update_defaults(resource) explicitly_changing_attributes = changeset.attributes |> Map.keys() |> then(fn attrs -> if touch_update_defaults? do Enum.concat(attrs, Keyword.keys(update_defaults)) else attrs end end) |> Kernel.--(Map.get(changeset, :defaults, [])) |> Kernel.--(keys) upsert_fields = changeset.context[:private][:upsert_fields] || explicitly_changing_attributes case bulk_create(resource, [changeset], %{ single?: true, upsert?: true, tenant: changeset.tenant, identity: identity, upsert_keys: keys, action_select: changeset.action_select, upsert_fields: upsert_fields, touch_update_defaults?: touch_update_defaults?, return_records?: true }) do {:ok, []} -> key_filters = Enum.map(keys, fn key -> {key, Ash.Changeset.get_attribute(changeset, key) || Map.get(changeset.params, key) || Map.get(changeset.params, to_string(key))} end) ash_query = resource |> Ash.Query.do_filter(and: [key_filters]) |> then(fn query when is_nil(identity) or is_nil(identity.where) -> query query -> Ash.Query.do_filter(query, identity.where) end) |> Ash.Query.set_tenant(changeset.tenant) {:ok, {:upsert_skipped, ash_query, fn -> with {:ok, ecto_query} <- Ash.Query.data_layer_query(ash_query), {:ok, [result]} <- run_query(ecto_query, resource) do {:ok, Ash.Resource.put_metadata(result, :upsert_skipped, true)} end end}} {:ok, [result]} -> {:ok, result} {:error, :no_rollback, error} -> {:error, :no_rollback, error} {:error, error} -> {:error, error} end end end defp conflict_target(resource, identity, keys) do identity_where = case identity do %{name: name, where: where} when not is_nil(where) -> AshPostgres.DataLayer.Info.identity_where_to_sql(resource, name) || raise(""" Must provide an entry for :#{identity.name} in `postgres.identity_wheres_to_sql` to use it as an upsert_identity. See https://hexdocs.pm/ash_postgres/AshPostgres.DataLayer.Info.html#identity_wheres_to_sql/1 for an example. """) _ -> nil end base_filter_sql = case Ash.Resource.Info.base_filter(resource) do nil -> nil _base_filter -> AshPostgres.DataLayer.Info.base_filter_sql(resource) || raise """ Cannot use upserts with resources that have a base_filter without also adding `base_filter_sql` in the postgres section. """ end where = case {base_filter_sql, identity_where} do {nil, nil} -> nil {base_filter_sql, nil} -> " WHERE (#{base_filter_sql})" {nil, identity_where} -> " WHERE (#{identity_where})" {base_filter_sql, identity_where} -> " WHERE ((#{base_filter_sql}) AND (#{identity_where}))" end if is_nil(where) && Enum.all?(keys, &Ash.Resource.Info.attribute(resource, &1)) do keys else sources = sources_to_sql(resource, keys) {:unsafe_fragment, "(" <> Enum.join(sources, ", ") <> ")#{where}"} end end defp sources_to_sql(resource, keys) do Enum.map(keys, fn key -> case Ash.Resource.Info.field(resource, key) do %Ash.Resource.Attribute{source: source, name: name} -> ~s("#{source || name}") %Ash.Resource.Calculation{name: name} -> if sql = AshPostgres.DataLayer.Info.calculation_to_sql(resource, name) do "(" <> sql <> ")" else raise ArgumentError, "Calculation #{inspect(key)} used in `AshPostgres.DataLayer` conflict target must have its sql defined in `calculations_to_sql`" end _other -> raise ArgumentError, "Unsupported field #{inspect(key)} used in `AshPostgres.DataLayer` conflict target" end end) end defp update_defaults(resource) do attributes = resource |> Ash.Resource.Info.attributes() |> Enum.reject(&is_nil(&1.update_default)) attributes |> static_defaults() |> Enum.concat(lazy_matching_defaults(attributes)) |> Enum.concat(lazy_non_matching_defaults(attributes)) end defp static_defaults(attributes) do attributes |> Enum.reject(&get_default_fun(&1)) |> Enum.map(&{&1.name, &1.update_default}) end defp lazy_non_matching_defaults(attributes) do attributes |> Enum.filter(&(!&1.match_other_defaults? && get_default_fun(&1))) |> Enum.map(fn attribute -> default_value = case attribute.update_default do function when is_function(function) -> function.() {m, f, a} when is_atom(m) and is_atom(f) and is_list(a) -> apply(m, f, a) end {:ok, default_value} = Ash.Type.cast_input(attribute.type, default_value, attribute.constraints) {attribute.name, default_value} end) end defp lazy_matching_defaults(attributes) do attributes |> Enum.filter(&(&1.match_other_defaults? && get_default_fun(&1))) |> Enum.group_by(&{&1.update_default, &1.type, &1.constraints}) |> Enum.flat_map(fn {{default_fun, type, constraints}, attributes} -> default_value = case default_fun do function when is_function(function) -> function.() {m, f, a} when is_atom(m) and is_atom(f) and is_list(a) -> apply(m, f, a) end {:ok, default_value} = Ash.Type.cast_input(type, default_value, constraints) Enum.map(attributes, &{&1.name, default_value}) end) end defp get_default_fun(attribute) do if is_function(attribute.update_default) or match?({_, _, _}, attribute.update_default) do attribute.update_default end end @impl true def update(resource, changeset) do source = resolve_source(resource, changeset) query = from(row in source, as: ^0) |> AshSql.Bindings.default_bindings( resource, AshPostgres.SqlImplementation, changeset.context ) |> pkey_filter(changeset.data) |> then(fn query -> if changeset.tenant do set_tenant(resource, query, changeset.tenant) |> elem(1) else query end end) |> then(fn query -> Map.put( query, :__ash_bindings__, Map.put_new(query.__ash_bindings__, :tenant, changeset.tenant) ) end) changeset = Ash.Changeset.set_context(changeset, %{ data_layer: %{ use_atomic_update_data?: true } }) case update_query(query, changeset, resource, %{ return_records?: true, action_select: changeset.action_select, calculations: [] }) do {:ok, []} -> {:error, Ash.Error.Changes.StaleRecord.exception( resource: resource, filter: changeset.filter )} {:ok, [record]} -> maybe_update_tenant(resource, changeset, record) {:ok, record} {:error, error} -> {:error, error} {:error, :no_rollback, error} -> {:error, :no_rollback, error} end end defp pkey_filter(query, %resource{} = record) do pkey = record |> Map.take(Ash.Resource.Info.primary_key(resource)) |> Map.to_list() Ecto.Query.where(query, ^pkey) end @impl true def destroy(resource, %{data: record} = changeset) do source = resolve_source(resource, changeset) query = from(row in source, as: ^0) |> AshSql.Bindings.default_bindings( resource, AshPostgres.SqlImplementation, changeset.context ) |> pkey_filter(record) repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, changeset) with {:ok, query} <- filter(query, changeset.filter, resource) do ecto_changeset = case changeset.data do %Ash.Changeset.OriginalDataNotAvailable{} -> changeset.resource.__struct__() data -> data end |> Map.update!(:__meta__, &Map.put(&1, :source, table(resource, changeset))) |> ecto_changeset(changeset, :delete, repo, true) case bulk_updatable_query( query, resource, [], [], changeset.context, :destroy ) do {:error, error} -> {:error, error} {:ok, query} -> try do repo_opts = AshSql.repo_opts( repo, AshPostgres.SqlImplementation, changeset.timeout, changeset.tenant, changeset.resource ) query = Ecto.Query.exclude(query, :select) with_savepoint(repo, query, fn -> repo.delete_all( query, repo_opts ) |> case do {0, _} -> {:error, Ash.Error.Changes.StaleRecord.exception( resource: resource, filter: changeset.filter )} _ -> :ok end end) rescue e -> handle_raised_error( e, __STACKTRACE__, {:ecto_changeset, :delete, ecto_changeset}, resource ) end end end end @impl true def lock(query, :for_update, _) do if query.distinct do new_query = Ecto.Query.lock(%{query | distinct: nil}, [{^0, a}], fragment("FOR UPDATE OF ?", a)) q = from(row in subquery(new_query), []) {:ok, %{q | distinct: query.distinct}} else {:ok, Ecto.Query.lock(query, [{^0, a}], fragment("FOR UPDATE OF ?", a))} end end for lock <- @locks do frag = "#{lock} OF ?" def lock(query, unquote(lock), _) do {:ok, Ecto.Query.lock(query, [{^0, a}], fragment(unquote(frag), a))} end frag = "#{lock} OF ? NOWAIT" new_lock = "#{lock} NOWAIT" def lock(query, unquote(new_lock), _) do {:ok, Ecto.Query.lock(query, [{^0, a}], fragment(unquote(frag), a))} end frag = "#{lock} OF ? SKIP LOCKED" new_lock = "#{lock} SKIP LOCKED" def lock(query, unquote(new_lock), _) do {:ok, Ecto.Query.lock(query, [{^0, a}], fragment(unquote(frag), a))} end end @impl true def sort(query, sort, _resource) do query = maybe_subquery_upgrade(query, {:sort, sort}) {:ok, Map.update!( query, :__ash_bindings__, &Map.put(&1, :sort, sort) )} end @impl true def select(query, select, _resource) do query = maybe_subquery_upgrade(query, {:select, select}) query = put_in(query.__ash_bindings__[:select], select) if query.__ash_bindings__.context[:data_layer][:combination_query?] || query.__ash_bindings__.context[:data_layer][:combination_of_queries?] do binding = query.__ash_bindings__.root_binding {query, select} = if field_set = query.__ash_bindings__[:already_selected] do {query, select -- field_set} else {from(row in Ecto.Query.exclude(query, :select), select: %{}), select} end Enum.reduce(select, query, fn field, query -> from(row in query, select_merge: %{^field => field(as(^binding), ^field)}) end) |> then(&{:ok, &1}) else {:ok, from(row in query, select: struct(row, ^Enum.uniq(select)))} end end @impl true def distinct_sort(query, sort, _) when sort in [nil, []] do {:ok, query} end def distinct_sort(query, sort, _) do query = maybe_subquery_upgrade(query, {:distinct_sort, sort}) {:ok, Map.update!(query, :__ash_bindings__, &Map.put(&1, :distinct_sort, sort))} end # If the order by does not match the initial sort clause, then we use a subquery # to limit to only distinct rows. This may not perform that well, so we may need # to come up with alternatives here. @impl true def distinct(query, distinct, resource) do query = maybe_subquery_upgrade(query, {:distinct, distinct}) AshSql.Distinct.distinct(query, distinct, resource) end @impl true def filter(query, filter, resource, opts \\ []) do query = maybe_subquery_upgrade(query, {:filter, filter}) used_aggregates = Ash.Filter.used_aggregates(filter, []) query = AshSql.Bindings.default_bindings(query, resource, AshPostgres.SqlImplementation) query |> AshSql.Join.join_all_relationships(filter, opts) |> case do {:ok, query} -> query |> AshSql.Aggregate.add_aggregates( used_aggregates, resource, false, query.__ash_bindings__.root_binding ) |> case do {:ok, query} -> {:ok, AshSql.Filter.add_filter_expression(query, filter)} {:error, error} -> {:error, error} end {:error, error} -> {:error, error} end end defp maybe_subquery_upgrade( %{__ash_bindings__: %{subquery_upgrade?: true}} = query, _ ) do query end defp maybe_subquery_upgrade(query, type) do fieldset = query.__ash_bindings__.context[:data_layer][:combination_fieldset] if query.__ash_bindings__.context[:data_layer][:combination_of_queries?] && fieldset do requires_join? = case type do {:filter, contents} -> Enum.any?( Ash.Filter.list_refs(contents), &(&1.relationship_path != [] || &1.attribute.name not in fieldset) ) {:calculations, calculations} -> Enum.any?(calculations, fn {_, expr} -> Enum.any?( Ash.Filter.list_refs(expr), &(&1.relationship_path != [] || &1.attribute.name not in fieldset) ) end) {sort, sorts} when sort in [:sort, :distinct, :distinct_sort] -> Enum.any?(sorts, fn {atom, _} when is_atom(atom) -> atom not in fieldset {%Ash.Query.Calculation{} = calc, _} -> calc.opts |> calc.module.expression(calc.context) |> Ash.Filter.hydrate_refs(%{ resource: query.__ash_bindings__.resource, parent_stack: query.__ash_bindings__[:parent_resources] || [], public?: false }) |> Ash.Filter.list_refs() |> Enum.any?(&(&1.relationship_path != [] || &1.attribute.name not in fieldset)) _ -> true end) {:select, select} -> Enum.any?(select, &(&1 not in fieldset)) end resource = query.__ash_bindings__.resource if requires_join? do primary_key = Ash.Resource.Info.primary_key(query.__ash_bindings__.resource) if primary_key != [] && primary_key -- fieldset == [] do dynamic = Enum.reduce(primary_key, nil, fn key, expr -> if is_nil(expr) do Ecto.Query.dynamic([l, r], field(l, ^key) == field(r, ^key)) else Ecto.Query.dynamic([l, r], field(l, ^key) == field(r, ^key) and ^expr) end end) default_select = MapSet.to_list( Ash.Resource.Info.selected_by_default_attribute_names( query.__ash_bindings__.resource ) ) query_with_select = from(sub in Ecto.Query.exclude(query, :select), join: row in ^query.__ash_bindings__.resource, # why doesn't `.root_binding` work the way I expect it to here? on: ^dynamic, select: map(row, ^default_select), select_merge: map(sub, ^fieldset) ) from(row in subquery(query_with_select), as: ^0) |> AshSql.Bindings.default_bindings(resource, AshPostgres.SqlImplementation) |> Map.update!( :__ash_bindings__, &Map.merge(&1, %{ already_selected: fieldset, select: query.__ash_bindings__[:select], select_calculations: query.__ash_bindings__[:select_calculations], subquery_upgrade?: true, sort: query.__ash_bindings__[:sort], context: query.__ash_bindings__.context }) ) else raise """ Unsupported combination query. Combinations must select the primary key if referencing any fields that are *not* selected by the combinations in filter, sort & distinct. """ end else query end else query end end @impl true def add_aggregates(query, aggregates, _resource) do {:ok, Map.update!(query, :__ash_bindings__, fn bindings -> Map.put(bindings, :load_aggregates, aggregates) end)} end @impl true def add_calculations(query, calculations, resource, select? \\ true) do query = maybe_subquery_upgrade(query, {:calculations, calculations}) AshSql.Calculation.add_calculations( query, calculations, resource, query.__ash_bindings__.root_binding, select? ) end def add_known_binding(query, data, known_binding) do bindings = query.__ash_bindings__.bindings new_ash_bindings = %{ query.__ash_bindings__ | bindings: Map.put(bindings, known_binding, data) } %{query | __ash_bindings__: new_ash_bindings} end @impl true def transaction(resource, func, timeout \\ nil, reason \\ %{type: :custom, metadata: %{}}) do repo = case reason[:data_layer_context] do %{repo: repo} when not is_nil(repo) -> repo _ -> AshPostgres.DataLayer.Info.repo(resource, :mutate) end func = fn -> repo.on_transaction_begin(reason) func.() end if timeout do repo.transaction(func, timeout: timeout) else repo.transaction(func) end end if Code.ensure_loaded?(Igniter) do def install(igniter, module, Ash.Resource, _path, argv) do table_name = module |> Module.split() |> List.last() |> Macro.underscore() |> Igniter.Inflex.pluralize() {options, _, _} = OptionParser.parse(argv, switches: [repo: :string]) repo = case options[:repo] do nil -> Igniter.Project.Module.module_name(igniter, "Repo") repo -> Igniter.Project.Module.parse(repo) end igniter |> Spark.Igniter.set_option(module, [:postgres, :table], table_name) |> Spark.Igniter.set_option(module, [:postgres, :repo], repo) end def install(igniter, _, _, _), do: igniter end @impl true def rollback(resource, term) do AshPostgres.DataLayer.Info.repo(resource, :mutate).rollback(term) end defp table(resource, changeset) do changeset.context[:data_layer][:table] || AshPostgres.DataLayer.Info.table(resource) end defp raise_table_error!(resource, operation) do if AshPostgres.DataLayer.Info.polymorphic?(resource) do raise """ Could not determine table for #{operation} on #{inspect(resource)}. Polymorphic resources require that the `data_layer[:table]` context is provided. See the guide on polymorphic resources for more information. """ else raise """ Could not determine table for #{operation} on #{inspect(resource)}. """ end end defp resolve_source(resource, changeset) do if table = changeset.context[:data_layer][:table] do {table, resource} else resource end end end