diff --git a/config/config.exs b/config/config.exs index ef2a0fbc..08863ea2 100644 --- a/config/config.exs +++ b/config/config.exs @@ -27,6 +27,7 @@ if Mix.env() == :test do config :elixir, :time_zone_database, Tz.TimeZoneDatabase config :ash_postgres, AshPostgres.TestRepo, log: false config :ash_postgres, AshPostgres.TestNoSandboxRepo, log: false + config :ash_postgres, AshPostgres.TemporalTestRepo, log: false config :ash, :validate_domain_resource_inclusion?, false config :ash, :validate_domain_config_inclusion?, false @@ -47,11 +48,13 @@ if Mix.env() == :test do AshPostgres.Test.OperatorExpression.IntListType ] + pg_port = String.to_integer(System.get_env("PG_PORT", "5432")) + config :ash_postgres, AshPostgres.TestRepo, username: "postgres", database: "ash_postgres_test", hostname: "localhost", - port: 5433, + port: pg_port, pool: Ecto.Adapters.SQL.Sandbox, types: AshPostgres.Test.PostgrexTypes @@ -60,7 +63,7 @@ if Mix.env() == :test do password: "postgres", database: "ash_postgres_dev_test", hostname: "localhost", - port: 5433, + port: pg_port, migration_primary_key: [name: :id, type: :binary_id], pool: Ecto.Adapters.SQL.Sandbox @@ -73,7 +76,7 @@ if Mix.env() == :test do username: "postgres", database: "ash_postgres_test", hostname: "localhost", - port: 5433 + port: pg_port # sobelow_skip ["Config.Secrets"] config :ash_postgres, AshPostgres.TestNoSandboxRepo, password: "postgres" @@ -81,8 +84,24 @@ if Mix.env() == :test do config :ash_postgres, AshPostgres.TestNoSandboxRepo, migration_primary_key: [name: :id, type: :binary_id] + config :ash_postgres, AshPostgres.TemporalTestRepo, + username: "postgres", + password: "postgres", + database: "ash_postgres_test", + hostname: "localhost", + port: pg_port, + pool: Ecto.Adapters.SQL.Sandbox, + types: AshPostgres.Test.PostgrexTypes + + # Temporal resources need PostgreSQL 18+, so their repo is only migrated when testing + # against it. An unset `PG_VERSION` means the newest (see `AshPostgres.TestRepo`). + {pg_version, _} = Integer.parse(System.get_env("PG_VERSION", "19")) + temporal_repos = if pg_version >= 18, do: [AshPostgres.TemporalTestRepo], else: [] + config :ash_postgres, - ecto_repos: [AshPostgres.TestRepo, AshPostgres.DevTestRepo, AshPostgres.TestNoSandboxRepo], + ecto_repos: + [AshPostgres.TestRepo, AshPostgres.DevTestRepo, AshPostgres.TestNoSandboxRepo] ++ + temporal_repos, ash_domains: [ AshPostgres.Test.Domain, AshPostgres.MultitenancyTest.Domain, diff --git a/documentation/topics/advanced/temporal-resources.md b/documentation/topics/advanced/temporal-resources.md new file mode 100644 index 00000000..2bc50e69 --- /dev/null +++ b/documentation/topics/advanced/temporal-resources.md @@ -0,0 +1,85 @@ + + +# Temporal Resources + +Temporal resources keep every version of every record, each valid for a period of time. +How to define and use them is covered in Ash's +[temporal resources guide](https://hexdocs.pm/ash/temporal-resources.html). This page covers +what AshPostgres needs and does for them. + +## Requirements + +Temporal resources need **PostgreSQL 18 or later**, and the repo has to say so: + +```elixir +def min_pg_version do + %Version{major: 18, minor: 0, patch: 0} +end +``` + +They also need the `btree_gist` extension, because the temporal primary key is backed by a +GiST exclusion constraint: + +```elixir +def installed_extensions do + ["ash-functions", "btree_gist"] +end +``` + +A temporal resource whose repo doesn't declare both fails to compile. + +## What the migration generator creates + +For a temporal resource, `mix ash_postgres.generate_migrations` creates: + +- the period column, a `tstzrange` +- `PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)`, so a record can have many versions, but never + two valid at the same time +- `PERIOD` foreign keys for relationships between temporal resources +- identities as exclusion constraints, so they're unique at every instant rather than across + all of time + +## How writes split versions + +A write at an instant changes only the part of each version from that instant onward, and +keeps the rest as it was. SQL:2011 calls this `UPDATE/DELETE ... FOR PORTION OF`. + +That was planned for PostgreSQL 19, but it was reverted before release, because concurrent +writes under `READ COMMITTED` could silently lose part of an update. So AshPostgres splits +versions itself, in a single statement: it writes the slice the write covers, and puts back +the rest of the version as new rows. Once PostgreSQL ships `FOR PORTION OF` natively, +AshPostgres will use it. + +## Concurrent writes + +Under `READ COMMITTED`, each temporal write locks the versions it's about to split, and checks +that they're the versions it expected. If another transaction changed one of them in the +meantime, the write does nothing and runs again, holding the locks it already took. Two upserts +that race to create the same record are retried the same way. + +After 25 conflicts in a row, which takes many transactions writing the same slice of the same +record at once, the action fails with `AshPostgres.Temporal.WriteConflict`. Nothing was +written, so retrying it is safe. + +Under `REPEATABLE READ` or `SERIALIZABLE`, PostgreSQL raises a serialization failure for these +cases itself, and retrying is up to you. + +## Upgrading from earlier versions + +Migrations generated by earlier versions of AshPostgres only created the temporal primary key, +temporal identities and `PERIOD` foreign keys on PostgreSQL 19, and fell back to plain keys +otherwise. If you ran them on PostgreSQL 18, your tables have the plain keys. Check with: + +```sql +SELECT conname, pg_get_constraintdef(oid) +FROM pg_constraint +WHERE conrelid = 'your_table'::regclass; +``` + +If the primary key doesn't include `WITHOUT OVERLAPS`, write a migration that drops it and +adds `PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)`, and do the same for identities and +`PERIOD` foreign keys. diff --git a/lib/data_layer.ex b/lib/data_layer.ex index aa097cbc..a3138770 100644 --- a/lib/data_layer.ex +++ b/lib/data_layer.ex @@ -2018,7 +2018,7 @@ defmodule AshPostgres.DataLayer do if changeset.context[:data_layer][:use_atomic_update_data?] && Enum.count_until(results, 2) == 1 do - # For temporal resources `FOR PORTION OF` truncates `valid_at` to + # For temporal resources the split truncates `valid_at` to # the affected slice; include it so the returned record reflects # the slice rather than the pre-split range. temporal_attrs = @@ -2416,13 +2416,25 @@ defmodule AshPostgres.DataLayer do @impl true def bulk_create(resource, stream, options) do if options[:upsert?] && Ash.Resource.Info.temporal?(resource) do - # Temporal upsert (single or bulk): one atomic `FOR PORTION OF`-update-or-insert CTE + # Temporal upsert (single or bulk): one atomic split-or-insert CTE # per `as_of` (see `AshPostgres.Temporal.upsert_all/5`). changesets = Enum.to_list(stream) repo = AshSql.dynamic_repo(resource, AshPostgres.SqlImplementation, Enum.at(changesets, 0)) keys = options[:upsert_keys] || Ash.Resource.Info.primary_key(resource) - AshPostgres.Temporal.upsert_all(repo, resource, changesets, keys, options[:upsert_fields]) + try do + AshPostgres.Temporal.upsert_all(repo, resource, changesets, keys, options[:upsert_fields]) + rescue + e -> + changeset = Ash.Changeset.new(resource) + + handle_raised_error( + e, + __STACKTRACE__, + {:bulk_create, ecto_changeset(changeset.data, changeset, :create, repo, false)}, + resource + ) + end else do_bulk_create(resource, stream, options) end @@ -2921,7 +2933,7 @@ defmodule AshPostgres.DataLayer do end) end - # The `FOR PORTION OF` portion for a temporal mutation. A range on the changeset + # The portion a temporal mutation splits out. A range on the changeset # names it outright and is the only place a period survives; otherwise the query's # threaded `as_of` (set for both single and bulk operations), the changeset's # `as_of` (single), or the wall clock each give `[as_of, ∞)`. @@ -4394,7 +4406,7 @@ defmodule AshPostgres.DataLayer do |> then(fn query -> # A temporal update targets the period valid at `as_of` (the pkey filter # alone matches every period of the id), and `update_query` splits it via - # FOR PORTION OF. Scope to that period with `valid_at @> as_of`. + # splitting it at as_of. Scope to that period with `valid_at @> as_of`. if Ash.Resource.Info.temporal?(resource) do {:ok, query} = set_as_of( diff --git a/lib/migration_generator/operation.ex b/lib/migration_generator/operation.ex index 19761331..5f217a2b 100644 --- a/lib/migration_generator/operation.ex +++ b/lib/migration_generator/operation.ex @@ -1140,11 +1140,9 @@ defmodule AshPostgres.MigrationGenerator.Operation do period -> # Period-aware uniqueness: the same key values may recur across non-overlapping - # periods (history), but never within a single instant. PG19+ enforces this with - # a `WITHOUT OVERLAPS`-style GiST exclusion (`key WITH =`, `period WITH &&`, - # needs `btree_gist`). Version-guarded so the migration is portable — on servers - # older than PG19 (where temporal isn't supported) it degrades to a plain unique - # index rather than failing. + # periods (history), but never within a single instant. Enforced with a + # `WITHOUT OVERLAPS`-style GiST exclusion (`key WITH =`, `period WITH &&`, + # needs `btree_gist`). predicate = cond do base_filter && where -> "(#{where}) AND (#{base_filter})" @@ -1168,13 +1166,7 @@ defmodule AshPostgres.MigrationGenerator.Operation do body = "ADD CONSTRAINT \\\"#{index_name}\\\" EXCLUDE USING gist (#{exclude_cols})#{where_sql}" - """ - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= 190_000 do - #{alter_table(table, schema, multitenancy, body)} - else - #{plain} - end - """ + alter_table(table, schema, multitenancy, body) end end @@ -1205,13 +1197,7 @@ defmodule AshPostgres.MigrationGenerator.Operation do _period -> body = "DROP CONSTRAINT IF EXISTS \\\"#{index_name}\\\"" - """ - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= 190_000 do - #{alter_table(table, schema, multitenancy, body)} - else - #{plain} - end - """ + alter_table(table, schema, multitenancy, body) end end @@ -1435,18 +1421,9 @@ defmodule AshPostgres.MigrationGenerator.Operation do # Temporal primary key: the period column is checked for non-overlap # rather than equality (SQL:2011 / PG18+, `WITHOUT OVERLAPS`, backed by a # GiST exclusion constraint requiring `btree_gist` — enforced via - # `installed_extensions`, see VerifyTemporal). Guarded at migration time so - # the migration is portable: on servers older than PG19 it degrades to a - # plain primary key rather than failing. - temporal_keys = plain_keys <> ", #{op.temporal.attribute} WITHOUT OVERLAPS" - - """ - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= 190_000 do - #{alter_for.(temporal_keys)} - else - #{alter_for.(plain_keys)} - end - """ + # `installed_extensions`, see VerifyTemporal). Needs PG18+, which VerifyTemporal + # requires of the repo's `min_pg_version/0`. + alter_for.(plain_keys <> ", #{op.temporal.attribute} WITHOUT OVERLAPS") <> "\n" else alter_for.(plain_keys) <> "\n" end @@ -1545,12 +1522,10 @@ defmodule AshPostgres.MigrationGenerator.Operation do "\\\"#{op.destination_table}\\\"" end - # PERIOD foreign keys need PG18+; guarded at migration time so the migration - # is portable (skipped entirely on servers older than PG19). + # PERIOD foreign keys need PG18+, which VerifyTemporal requires of the repo's + # `min_pg_version/0`. """ - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= 190_000 do - execute("ALTER TABLE \\\"#{op.table}\\\" ADD CONSTRAINT #{name} FOREIGN KEY (#{op.column}, PERIOD #{op.source_period}) REFERENCES #{destination} (#{op.destination_attribute}, PERIOD #{op.destination_period})") - end + execute("ALTER TABLE \\\"#{op.table}\\\" ADD CONSTRAINT #{name} FOREIGN KEY (#{op.column}, PERIOD #{op.source_period}) REFERENCES #{destination} (#{op.destination_attribute}, PERIOD #{op.destination_period})") """ end diff --git a/lib/temporal.ex b/lib/temporal.ex index 29987d23..416f9ae7 100644 --- a/lib/temporal.ex +++ b/lib/temporal.ex @@ -4,21 +4,60 @@ defmodule AshPostgres.Temporal do @moduledoc false - # Emits `UPDATE/DELETE ... FOR PORTION OF FROM $1 TO NULL ...` - # for temporal resources. Postgres applies the change only to the slice of each - # matched row from `as_of` onward, truncating the prior row at `as_of` (its - # before-portion is preserved as a new row). A temporal write is always - # `[as_of, ∞)`, so `TO` is always `NULL` (the open/unbounded end) and `FROM` is - # the changeset's `as_of`. `TO NULL` (not `TO 'infinity'`) is essential: a current - # row's stored upper is unbounded (`NULL`), and `'infinity'` is a real value that - # an unbounded range *contains* — splitting `TO 'infinity'` would leave a literal - # `'infinity'` upper (which Postgrex can't decode) plus a junk `[infinity, )` row. + # Splits periods for temporal writes. A write changes only the slice of each matched + # version that falls inside its *portion*: `[as_of, ∞)` for an instant, or the period + # itself when `as_of` names one. The rest of the version is kept as it was, as one or + # two *leftover* rows. # - # Ecto has no `FOR PORTION OF`, and PG requires the clause BETWEEN the table and - # the `AS` alias (`UPDATE t AS a FOR PORTION OF ...` is a syntax error). So we - # render the statement with Ecto (`to_sql`, with the bound as `$1` via a counter - # offset), splice the clause in after the table, and run it via `repo.query!`. - # Mirrors the approach in `AshPostgres.Merge`. + # That's SQL:2011's `UPDATE/DELETE ... FOR PORTION OF`, which Postgres doesn't have (it + # was reverted from 19 before release, over the same READ COMMITTED lost updates handled + # below). Once Postgres ships it natively, we'll use it. Until then we do it by hand, in + # one statement, using PG18's `RETURNING old`: + # + # WITH "__fpo" AS ( + # UPDATE t AS s0 SET ..., valid_at = s0.valid_at * + # WHERE (...) AND (s0.valid_at && ) + # RETURNING ..., old AS "__old" + # ), "__leftovers" AS ( + # INSERT INTO t (, valid_at) + # SELECT (f."__old")., ..., l.r + # FROM "__fpo" f, unnest(multirange((f."__old").valid_at) - multirange()) l(r) + # ) + # SELECT ... FROM "__fpo" + # + # A delete is the same, except it removes the whole version and the leftovers put back + # what was outside the portion. + # + # The portion's upper is `NULL` (unbounded), never `'infinity'`. A current version's + # stored upper is unbounded, and `'infinity'` is a real value that an unbounded range + # *contains*, so splitting at it would leave a literal `'infinity'` upper (which + # Postgrex can't decode) plus a junk `[infinity, )` row. + # + # Concurrency: under READ COMMITTED, two writes to one record can lose an update. The + # second blocks on the version the first is splitting and rechecks it once the first + # commits, but by then the version may no longer match (so it's skipped), and the + # leftovers the first inserted are newer than the second's snapshot (so they're never + # seen). This is what got `FOR PORTION OF` reverted. + # + # So each statement first locks the versions it's about to split, and compares them + # with the versions its snapshot matched (see `gate_ctes/2`). Locking waits out any + # concurrent write and then sees the version as that write left it, so if the two + # differ, a concurrent write got in the way. The writes are gated on that, so the + # statement writes nothing and we just run it again, with a snapshot that sees what the + # other write committed. No savepoint needed: a savepoint per write would cost two round + # trips and a subtransaction, and enough subtransactions in one transaction slow down + # every snapshot on the server. + # + # When there's nothing to lock (an upsert of a record that doesn't exist yet, or into a + # gap), the `WITHOUT OVERLAPS` primary key catches a concurrent insert instead: our insert + # waits for it and then conflicts. The insert skips that key (`ON CONFLICT ON CONSTRAINT + # DO NOTHING`) and we retry just the skipped keys. + # + # Under REPEATABLE READ and SERIALIZABLE, Postgres raises a serialization failure for + # these cases itself. + + # How many times a write runs before giving up on getting past concurrent writes. + @attempts 25 @doc """ Atomic temporal upsert for one or more changesets. @@ -29,25 +68,29 @@ defmodule AshPostgres.Temporal do WITH src() AS (VALUES (...), (...)), upd AS ( - UPDATE t FOR PORTION OF FROM $as_of TO NULL - SET = src., ... FROM src + UPDATE t SET = src., ..., = t. * [as_of, ∞) + FROM src WHERE t. = src. AND t. @> $as_of -- period AT as_of - RETURNING t.* + AND NOT -- see `gate_ctes/2` + RETURNING new AS "__new", old AS "__old" ), + "__leftovers" AS (INSERT INTO t ... the part of each matched period before as_of), ins AS ( INSERT INTO t (, ) SELECT src., FROM src - WHERE NOT EXISTS (SELECT 1 FROM upd WHERE upd. = src.) - RETURNING * + WHERE NOT + AND NOT EXISTS (SELECT 1 FROM upd WHERE (upd."__new"). = src.) + ON CONFLICT ON CONSTRAINT DO NOTHING -- lost a race: retried + RETURNING ... ) - SELECT * FROM upd UNION ALL SELECT * FROM ins + SELECT (upd."__new").* FROM upd UNION ALL SELECT * FROM ins - Per changeset: match the period valid AT `as_of` and split it via `FOR PORTION OF` - (future periods, which don't contain `as_of`, are untouched); if none, insert a new - period bounded to the next one — or unbounded above when there is no later period. + Per changeset: match the period valid AT `as_of` and split it there (future periods, + which don't contain `as_of`, are untouched); if none, insert a new period bounded to + the next one — or unbounded above when there is no later period. - All changesets sharing an `as_of` run in one statement (the `FOR PORTION OF` bound is a - single parameter — it can't be a per-row column). Changesets with differing `as_of`s are + All changesets sharing an `as_of` run in one statement (the split point is a single + parameter — it can't be a per-row column). Changesets with differing `as_of`s are grouped and run as one statement each, inside a transaction. Returns `{:ok, records}`. """ def upsert_all(repo, resource, changesets, upsert_keys, upsert_fields \\ nil) do @@ -62,30 +105,43 @@ defmodule AshPostgres.Temporal do end end) - run = fn -> - Enum.flat_map(groups, fn {as_of, group} -> - upsert_group(repo, resource, group, upsert_keys, upsert_fields, as_of) - end) - end + prefix = get_in(hd(changesets).context, [:data_layer, :schema]) + qtable = quote_table(prefix, AshPostgres.DataLayer.Info.table(resource)) + col_types = column_types(repo, qtable) records = - if map_size(groups) > 1 do - {:ok, records} = repo.transaction(run) - records - else - run.() - end + transaction(repo, fn -> + Enum.flat_map(groups, fn {as_of, group} -> + upsert_group( + repo, + resource, + qtable, + col_types, + group, + upsert_keys, + upsert_fields, + as_of + ) + end) + end) {:ok, records} end - defp upsert_group(repo, resource, changesets, upsert_keys, upsert_fields, as_of) do + defp upsert_group( + repo, + resource, + qtable, + col_types, + changesets, + upsert_keys, + upsert_fields, + as_of, + attempt \\ 1 + ) do attribute = Ash.Resource.Info.temporal_attribute(resource) - prefix = get_in(hd(changesets).context, [:data_layer, :schema]) - qtable = quote_table(prefix, AshPostgres.DataLayer.Info.table(resource)) tname = quote_name(AshPostgres.DataLayer.Info.table(resource)) - qattr = quote_name(attribute) - adapter = repo.__adapter__() + qattr = quote_name(source_col(resource, attribute)) # The columns to write — every changeset attribute in the group except the temporal # range, which we build from `as_of`. @@ -107,68 +163,43 @@ defmodule AshPostgres.Temporal do |> Kernel.--(upsert_keys) |> Kernel.--([attribute]) - col_types = column_types(repo, qtable) qsrc = fn field -> quote_name(source_col(resource, field)) end - col_count = length(insert_cols) - # `VALUES` rows + their params. `$1` is `as_of`; row i, column j is `$(2 + i*c + j)`. - # The first row carries `::type` casts so Postgres knows each `src` column's type. - {rows_sql, row_params} = - changesets - |> Enum.with_index() - |> Enum.map(fn {changeset, ri} -> - insert_cols - |> Enum.with_index() - |> Enum.map(fn {field, ci} -> - n = 2 + ri * col_count + ci + # `$1` is `as_of`, the `VALUES` params follow it. + {src_clause, row_params} = + values_source(repo, resource, col_types, changesets, insert_cols, 2) - {:ok, dumped} = - Ecto.Type.adapter_dump( - adapter, - resource.__schema__(:type, field), - Map.get(changeset.attributes, field) - ) - - placeholder = - if ri == 0 do - "$#{n}::#{Map.fetch!(col_types, to_string(source_col(resource, field)))}" - else - "$#{n}" - end - - {placeholder, dumped} - end) - |> Enum.unzip() - |> then(fn {phs, vals} -> {"(" <> Enum.join(phs, ", ") <> ")", vals} end) - end) - |> Enum.unzip() - - params = [as_of | List.flatten(row_params)] - - src_clause = - "src(#{Enum.map_join(insert_cols, ", ", qsrc)}) AS (VALUES #{Enum.join(rows_sql, ", ")})" + params = [as_of | row_params] key_match = fn left -> Enum.map_join(upsert_keys, " AND ", fn k -> "#{left}.#{qsrc.(k)} = src.#{qsrc.(k)}" end) end has_update? = set_fields != [] + range = "tstzrange($1::timestamptz, NULL)" + pkey = quote_name("#{AshPostgres.DataLayer.Info.table(resource)}_pkey") + columns = Enum.map(stored_columns(resource), "e_name/1) + upsert_cols = Enum.map(upsert_keys, qsrc) - upd_cte = + {gate_ctes, attempted} = if has_update? do set_sql = Enum.map_join(set_fields, ", ", fn f -> "#{qsrc.(f)} = src.#{qsrc.(f)}" end) - "upd AS (UPDATE #{qtable} FOR PORTION OF #{qattr} FROM $1::timestamptz TO NULL " <> - "SET #{set_sql} FROM src " <> - "WHERE #{key_match.(tname)} AND #{tname}.#{qattr} @> $1::timestamptz RETURNING #{tname}.*)" - end - - not_exists = - if has_update? do - "NOT EXISTS (SELECT 1 FROM upd WHERE #{key_match.("upd")})" + matched = + "FROM #{qtable} t JOIN src ON #{key_match.("t")} AND t.#{qattr} @> $1::timestamptz" + + {gate_ctes(matched, "t") <> + "upd AS (UPDATE #{qtable} SET #{set_sql}, #{qattr} = #{tname}.#{qattr} * #{range} " <> + "FROM src " <> + "WHERE #{key_match.(tname)} AND #{tname}.#{qattr} @> $1::timestamptz " <> + ~s|AND NOT (SELECT conflict FROM "__conflict") | <> + ~s|RETURNING new AS "__new", old AS "__old"), | <> + leftovers_cte(resource, qtable, "upd", range) <> ", ", + ~s|NOT EXISTS (SELECT 1 FROM upd WHERE #{key_match.(~s|(upd."__new")|)})|} else - "NOT EXISTS (SELECT 1 FROM #{qtable} ex WHERE #{key_match.("ex")} " <> - "AND ex.#{qattr} @> $1::timestamptz)" + {~s|"__conflict" AS (SELECT false AS conflict), |, + "NOT EXISTS (SELECT 1 FROM #{qtable} ex WHERE #{key_match.("ex")} " <> + "AND ex.#{qattr} @> $1::timestamptz)"} end # Bound the inserted range to this key's next period (gap-fill), else unbounded above @@ -182,24 +213,102 @@ defmodule AshPostgres.Temporal do insert_select_sql = Enum.map_join(insert_cols, ", ", fn f -> "src.#{qsrc.(f)}" end) <> ", " <> range_sql + # A key that loses an insert race to a concurrent write of the same record hits the + # `WITHOUT OVERLAPS` primary key. It's skipped rather than raised, and retried below. + # Only the primary key is an arbiter, so an identity violation still raises. ins_cte = - "ins AS (INSERT INTO #{qtable} (#{insert_cols_sql}) " <> - "SELECT #{insert_select_sql} FROM src WHERE #{not_exists} RETURNING *)" + ~s|"__attempted" AS (SELECT * FROM src WHERE NOT (SELECT conflict FROM "__conflict") | <> + "AND #{attempted}), " <> + "ins AS (INSERT INTO #{qtable} (#{insert_cols_sql}) " <> + ~s|SELECT #{insert_select_sql} FROM "__attempted" src | <> + "ON CONFLICT ON CONSTRAINT #{pkey} DO NOTHING " <> + "RETURNING #{Enum.join(columns, ", ")})" + + written = + if has_update? do + ~s|SELECT #{Enum.map_join(columns, ", ", &~s|(upd."__new").#{&1}|)}, 'ok' AS "__status" | <> + "FROM upd UNION ALL SELECT #{Enum.join(columns, ", ")}, 'ok' FROM ins" + else + ~s|SELECT #{Enum.join(columns, ", ")}, 'ok' AS "__status" FROM ins| + end + + skipped_columns = + Enum.map_join(columns, ", ", fn column -> + if column in upsert_cols, do: "src.#{column}", else: "NULL" + end) sql = - if has_update? do - "WITH #{src_clause}, #{upd_cte}, #{ins_cte} SELECT * FROM upd UNION ALL SELECT * FROM ins" + "WITH #{src_clause}, #{gate_ctes}#{ins_cte} " <> + written <> + ~s| UNION ALL SELECT #{skipped_columns}, 'skipped' FROM "__attempted" src | <> + "WHERE NOT EXISTS (SELECT 1 FROM ins WHERE #{key_match.("ins")}) " <> + "UNION ALL SELECT #{Enum.map_join(columns, ", ", fn _ -> "NULL" end)}, 'conflict' " <> + ~s|FROM "__conflict" WHERE conflict| + + {columns, %{"ok" => written, "skipped" => skipped, "conflict" => conflict}} = + by_status(repo.query!(sql, params)) + + retry = fn retried -> + if attempt < @attempts do + upsert_group( + repo, + resource, + qtable, + col_types, + retried, + upsert_keys, + upsert_fields, + as_of, + attempt + 1 + ) else - "WITH #{src_clause}, #{ins_cte} SELECT * FROM ins" + raise AshPostgres.Temporal.WriteConflict, resource: resource, attempts: @attempts end + end + + cond do + # Nothing was written, so run the whole group again. + conflict != [] -> + retry.(changesets) + + # Every other key was written, so run just the skipped ones again. + skipped != [] -> + skipped_keys = + repo + |> load_rows(resource, columns, skipped) + |> MapSet.new(fn record -> Enum.map(upsert_keys, &Map.get(record, &1)) end) + + retried = + Enum.filter(changesets, fn changeset -> + Enum.map(upsert_keys, &Map.get(changeset.attributes, &1)) in skipped_keys + end) + + records(repo, resource, columns, written, changesets, upsert_keys) ++ + retry.(retried) - result = repo.query!(sql, params) + true -> + records(repo, resource, columns, written, changesets, upsert_keys) + end + end + defp records(repo, resource, columns, rows, changesets, upsert_keys) do repo - |> load_rows(resource, result.columns, result.rows || []) + |> load_rows(resource, columns, rows) |> tag_bulk_refs(changesets, upsert_keys) end + # Splits a result's rows by their `"__status"` column, which is dropped. + defp by_status(%{columns: columns, rows: rows}) do + index = Enum.find_index(columns, &(&1 == "__status")) + + groups = + rows + |> Enum.group_by(&Enum.at(&1, index), &List.delete_at(&1, index)) + |> then(&Map.merge(%{"ok" => [], "skipped" => [], "conflict" => []}, &1)) + + {List.delete_at(columns, index), groups} + end + # Ash's bulk action correlates each returned record back to its changeset via # `__metadata__.bulk_action_ref`. Our `UNION` returns rows out of input order, so we # re-attach the ref by matching each record's upsert-key values to its changeset. @@ -227,6 +336,47 @@ defmodule AshPostgres.Temporal do end end + # `src() AS (VALUES ...)` over the changesets, with its params numbered from + # `first`. The first row carries `::type` casts so Postgres knows each column's type. + defp values_source(repo, resource, col_types, changesets, cols, first) do + adapter = repo.__adapter__() + col_count = length(cols) + + {rows_sql, row_params} = + changesets + |> Enum.with_index() + |> Enum.map(fn {changeset, ri} -> + cols + |> Enum.with_index() + |> Enum.map(fn {field, ci} -> + n = first + ri * col_count + ci + + {:ok, dumped} = + Ecto.Type.adapter_dump( + adapter, + resource.__schema__(:type, field), + Map.get(changeset.attributes, field) + ) + + placeholder = + if ri == 0 do + "$#{n}::#{Map.fetch!(col_types, to_string(source_col(resource, field)))}" + else + "$#{n}" + end + + {placeholder, dumped} + end) + |> Enum.unzip() + |> then(fn {phs, vals} -> {"(" <> Enum.join(phs, ", ") <> ")", vals} end) + end) + |> Enum.unzip() + + columns = Enum.map_join(cols, ", ", "e_name(source_col(resource, &1))) + + {"src(#{columns}) AS (VALUES #{Enum.join(rows_sql, ", ")})", List.flatten(row_params)} + end + # The Postgres type of each column (e.g. `"integer"`, `"tstzrange"`) so `VALUES` params # can be cast — otherwise they default to `text` and join/compare against typed columns. defp column_types(repo, qtable) do @@ -244,91 +394,219 @@ defmodule AshPostgres.Temporal do defp portion(%Ash.Range{} = period), do: period defp portion(instant), do: %Ash.Range{lower: instant, upper: nil, bounds: :"[)"} - @doc "Run an UPDATE ... FOR PORTION OF. Returns `{count, loaded_records | nil}`." + @doc "Run a temporal UPDATE over `as_of`'s portion. Returns `{count, loaded_records | nil}`." def update_all(repo, query, resource, as_of) do run(repo, :update_all, query, resource, as_of) end - @doc "Run a DELETE ... FOR PORTION OF. Returns `{count, loaded_records | nil}`." + @doc "Run a temporal DELETE over `as_of`'s portion. Returns `{count, loaded_records | nil}`." def delete_all(repo, query, resource, as_of) do run(repo, :delete_all, query, resource, as_of) end defp run(repo, kind, query, resource, as_of) do - attribute = Ash.Resource.Info.temporal_attribute(resource) + query = Map.delete(query, :__ash_bindings__) returning? = not is_nil(query.select) %Ash.Range{lower: lower, upper: upper} = portion(as_of) # Render with the portion's bounds reserved as $1 (and $2 when it has an upper). counter = if upper, do: 2, else: 1 + {sql, params} = Ecto.Adapters.SQL.to_sql(kind, repo, query, counter: counter) - {sql, params} = - Ecto.Adapters.SQL.to_sql(kind, repo, Map.delete(query, :__ash_bindings__), counter: counter) - - sql = splice_for_portion_of(sql, kind, attribute, not is_nil(upper)) - sql = if returning?, do: first_versions(sql, resource), else: sql + range = "tstzrange($1::timestamptz, #{if upper, do: "$2::timestamptz", else: "NULL"})" + sql = split_portion(sql, kind, resource, range, returning?) bounds = if upper, do: [lower, upper], else: [lower] - result = repo.query!(sql, bounds ++ params) + result = write(repo, resource, sql, bounds ++ params) if returning? do - rows = result.rows || [] - {length(rows), load_rows(repo, resource, result.columns, rows)} + {length(result.rows), load_rows(repo, resource, result.columns, result.rows)} else - {result.num_rows, nil} + {result.rows |> hd() |> hd(), nil} end end - # A range can carve several versions of one record; the write returns the first of them. - defp first_versions(sql, resource) do - keys = - resource - |> Ash.Resource.Info.primary_key() - |> Enum.map_join(", ", "e_name(source_col(resource, &1))) + # Rewrites Ecto's `UPDATE`/`DELETE` into the split described in the moduledoc. + defp split_portion(sql, kind, resource, range, returning?) do + [_, qtable, alias_token] = + Regex.run(~r/^(?:UPDATE|DELETE FROM) ((?:"[^"]+"\.)?"[^"]+") AS (\w+) /, sql) - period = quote_name(source_col(resource, Ash.Resource.Info.temporal_attribute(resource))) + period_col = quote_name(source_col(resource, temporal_attribute(resource))) + period = alias_token <> "." <> period_col - "WITH first_versions AS (#{sql}) SELECT DISTINCT ON (#{keys}) * FROM first_versions " <> - "ORDER BY #{keys}, lower(#{period})" - end + {dml, returned} = + case find_top_level(sql, " RETURNING ") do + nil -> {sql, nil} + pos -> {String.slice(sql, 0, pos), String.slice(sql, (pos + 11)..-1//1)} + end - # Insert ` FOR PORTION OF "" FROM $1::timestamptz TO …` immediately after the - # table name and before the ` AS ` that Ecto emits. The upper is `$2` when - # `as_of` names a period, else `NULL` — see the moduledoc for why not `'infinity'`. - defp splice_for_portion_of(sql, kind, attribute, upper?) do - {head, rest} = split_on_set_or_where(sql, kind) - alias_token = head |> String.trim() |> String.split() |> List.last() - table_part = String.replace_suffix(head, " AS #{alias_token}", "") + dml = + case kind do + :update_all -> + set_end = + [find_top_level(dml, " FROM "), find_top_level(dml, " WHERE ")] + |> Enum.reject(&is_nil/1) + |> Enum.min(fn -> String.length(dml) end) - to = if upper?, do: "$2::timestamptz", else: "NULL" + String.slice(dml, 0, set_end) <> + ", #{period_col} = #{period} * #{range}" <> String.slice(dml, set_end..-1//1) - clause = - " FOR PORTION OF " <> - quote_name(attribute) <> " FROM $1::timestamptz TO " <> to + :delete_all -> + dml + end + + overlaps = "(#{period} && #{range})" + + dml = + case find_top_level(dml, " WHERE ") do + nil -> + dml <> " WHERE " <> overlaps + + pos -> + String.slice(dml, 0, pos) <> + " WHERE (" <> String.slice(dml, (pos + 7)..-1//1) <> ") AND " <> overlaps + end + + # The versions the snapshot matched: the same table, joins and filter, as a SELECT. + {from_keyword, from_pos} = + case kind do + :update_all -> {" FROM ", find_top_level(dml, " FROM ")} + :delete_all -> {" USING ", find_top_level(dml, " USING ")} + end + + where_pos = find_top_level(dml, " WHERE ") + + joined = + if from_pos && from_pos < where_pos do + ", " <> String.slice(dml, (from_pos + String.length(from_keyword))..(where_pos - 1)//1) + else + "" + end + + matched = "FROM #{qtable} AS #{alias_token}#{joined}" <> String.slice(dml, where_pos..-1//1) + + returning = Enum.join(List.wrap(returned) ++ [~s|old AS "__old"|], ", ") + + select = + if returning? do + # A range can carve several versions of one record; the write returns the first. + columns = returned_columns(returned, alias_token) + + keys = + resource + |> Ash.Resource.Info.primary_key() + |> Enum.map_join(", ", &"f.#{quote_name(source_col(resource, &1))}") + + ~s|(SELECT DISTINCT ON (#{keys}) #{Enum.map_join(columns, ", ", &"f.#{&1}")}, | <> + ~s|'ok' AS "__status" FROM "__fpo" f ORDER BY #{keys}, lower(f.#{period_col})) | <> + "UNION ALL SELECT #{Enum.map_join(columns, ", ", &"n.#{&1}")}, 'conflict' " <> + ~s|FROM (SELECT (NULL::#{qtable}).*) n, "__conflict" WHERE conflict| + else + ~s|SELECT count(*), | <> + ~s|(SELECT CASE WHEN conflict THEN 'conflict' ELSE 'ok' END FROM "__conflict") | <> + ~s|AS "__status" FROM "__fpo"| + end - table_part <> clause <> " AS " <> alias_token <> rest + "WITH " <> + gate_ctes(matched, alias_token) <> + ~s|"__fpo" AS (#{dml} AND NOT (SELECT conflict FROM "__conflict") RETURNING #{returning}), | <> + leftovers_cte(resource, qtable, ~s|"__fpo"|, range) <> " " <> select end - # `head` is everything up to (and including the keyword we split on); `rest` - # is the keyword + remainder. We keep the keyword on `rest`. - defp split_on_set_or_where(sql, :update_all), do: split_keeping(sql, " SET ") + # `"__cand"` is the versions `matched` (a `FROM ... WHERE ...`) finds in the snapshot, and + # `"__locked"` is the same versions, locked. Locking waits out any concurrent write to a + # version and then sees it as that write left it: changed (a newer row) or gone. So the + # two differ exactly when a concurrent write got in the way, and `"__conflict"` says so. + # The writes are gated on it, so a conflicting statement writes nothing. + defp gate_ctes(matched, alias_token) do + ~s|"__cand" AS (SELECT #{alias_token}.ctid #{matched}), | <> + ~s|"__locked" AS (SELECT #{alias_token}.ctid #{matched} FOR UPDATE OF #{alias_token}), | <> + ~s|"__conflict" AS (SELECT (| <> + ~s|EXISTS (SELECT 1 FROM "__cand" c WHERE c.ctid NOT IN (SELECT ctid FROM "__locked")) OR | <> + ~s|EXISTS (SELECT 1 FROM "__locked" l WHERE l.ctid NOT IN (SELECT ctid FROM "__cand"))| <> + ~s|) AS conflict), | + end + + # Runs a split, and runs it again while it reports a conflict (see the moduledoc). A + # conflicting statement wrote nothing, so there's nothing to undo first. + defp write(repo, resource, sql, params, attempt \\ 1) do + case by_status(repo.query!(sql, params)) do + # Retried in a transaction, so the versions this attempt locked stay locked and can't + # move again under the next one. + {_, %{"conflict" => [_ | _]}} when attempt < @attempts -> + transaction(repo, fn -> write(repo, resource, sql, params, attempt + 1) end) + + {_, %{"conflict" => [_ | _]}} -> + raise AshPostgres.Temporal.WriteConflict, resource: resource, attempts: @attempts - defp split_on_set_or_where(sql, :delete_all) do - case find_top_level(sql, " WHERE ") do - nil -> split_keeping(sql, " RETURNING ") - _ -> split_keeping(sql, " WHERE ") + {columns, %{"ok" => rows}} -> + %{columns: columns, rows: rows} end end - defp split_keeping(sql, keyword) do - case find_top_level(sql, keyword) do + # Ecto returns plain columns (`s0."id", s0."name"`); select them by name. + defp returned_columns(returned, alias_token) do + returned + |> split_top_level(", ") + |> Enum.map(fn item -> + case Regex.run(~r/^#{alias_token}\.("[^"]+")$/, item) do + [_, column] -> + column + + nil -> + raise ArgumentError, + "Expected a plain column in temporal RETURNING, got: #{inspect(item)}" + end + end) + end + + defp split_top_level(sql, separator) do + case find_top_level(sql, separator) do nil -> - raise ArgumentError, "Expected #{keyword} in rendered temporal SQL: #{inspect(sql)}" + [sql] pos -> - {String.slice(sql, 0, pos), String.slice(sql, pos..-1//1)} + [ + String.slice(sql, 0, pos) + | split_top_level(String.slice(sql, (pos + String.length(separator))..-1//1), separator) + ] + end + end + + # Puts back what each version in `from` had outside the portion: every column copied + # from `"__old"`, over `old - portion` (zero, one or two ranges). + defp leftovers_cte(resource, qtable, from, range) do + attribute = temporal_attribute(resource) + period = quote_name(source_col(resource, attribute)) + + columns = + resource + |> stored_columns() + |> Enum.reject(&(&1 == source_col(resource, attribute))) + |> Enum.map("e_name/1) + + ~s|"__leftovers" AS (INSERT INTO #{qtable} (#{Enum.join(columns ++ [period], ", ")}) | <> + "OVERRIDING SYSTEM VALUE " <> + ~s|SELECT #{Enum.map_join(columns, ", ", &~s|(f."__old").#{&1}|)}, l.r | <> + ~s|FROM #{from} f, unnest(multirange((f."__old").#{period}) - multirange(#{range})) AS l(r))| + end + + defp stored_columns(resource) do + resource.__schema__(:fields) + |> Enum.map(&resource.__schema__(:field_source, &1)) + end + + # A partly retried upsert must commit or roll back as a whole. + defp transaction(repo, fun) do + if repo.in_transaction?() do + fun.() + else + {:ok, result} = repo.transaction(fun) + result end end + defp temporal_attribute(resource), do: Ash.Resource.Info.temporal_attribute(resource) + defp load_rows(repo, resource, columns, rows) do source = AshPostgres.DataLayer.Info.table(resource) keys = Enum.map(columns, &String.to_existing_atom/1) diff --git a/lib/temporal/write_conflict.ex b/lib/temporal/write_conflict.ex new file mode 100644 index 00000000..d0ef431c --- /dev/null +++ b/lib/temporal/write_conflict.ex @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: 2024 ash_postgres contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshPostgres.Temporal.WriteConflict do + @moduledoc """ + Used when a temporal write keeps losing to concurrent writes to the same records. + + A temporal write that finds the versions it's splitting were changed by another + transaction in the meantime runs again. This is raised when it has run `attempts` times + and still conflicted, which takes many transactions writing the same slice of the same + records at once. Retrying the action is safe: nothing was written. + """ + + use Splode.Error, fields: [:resource, :attempts], class: :framework + + def message(error) do + "Temporal write to #{inspect(error.resource)} conflicted with concurrent writes to the " <> + "same records #{error.attempts} times in a row, so nothing was written. " <> + "Retrying the action is safe." + end +end diff --git a/lib/transformers/validate_temporal_references.ex b/lib/transformers/validate_temporal_references.ex index dda1ee96..7f3331ec 100644 --- a/lib/transformers/validate_temporal_references.ex +++ b/lib/transformers/validate_temporal_references.ex @@ -6,9 +6,9 @@ defmodule AshPostgres.Transformers.ValidateTemporalReferences do @moduledoc false # A temporal relationship (`temporal_keys` with both a source and destination period # attribute) is backed by a Postgres `PERIOD` foreign key. PostgreSQL only supports - # `NO ACTION` for those — "PostgreSQL supports temporal foreign keys with action - # NO ACTION, but not RESTRICT, CASCADE, SET NULL, or SET DEFAULT" (PG19 docs, 5.7 - # Temporal Tables). So reject an `on_delete`/`on_update` referential action on such a + # `NO ACTION` for those, not RESTRICT, CASCADE, SET NULL, or SET DEFAULT ("unsupported + # ON DELETE action for foreign key constraint using PERIOD"). So reject an + # `on_delete`/`on_update` referential action on such a # relationship's `references` — it can't be expressed at the database level and must be # handled in the application (e.g. an `Ash.Resource.Change.CascadeDestroy`). # diff --git a/lib/verifiers/verify_temporal.ex b/lib/verifiers/verify_temporal.ex index e7811bda..a719e312 100644 --- a/lib/verifiers/verify_temporal.ex +++ b/lib/verifiers/verify_temporal.ex @@ -4,41 +4,76 @@ defmodule AshPostgres.Verifiers.VerifyTemporal do @moduledoc false - # A temporal resource's `WITHOUT OVERLAPS` primary key is backed by a GiST - # exclusion constraint, which requires the `btree_gist` extension. Ensure the - # repo declares it in `installed_extensions/0` so migrations install it. + # A temporal resource needs: + # + # - PostgreSQL 18+, for `WITHOUT OVERLAPS` primary keys, `PERIOD` foreign keys and + # `RETURNING old`. Migrations emit that DDL unconditionally, so the repo has to + # declare it with `min_pg_version/0`. + # - The `btree_gist` extension, since the `WITHOUT OVERLAPS` primary key is backed by + # a GiST exclusion constraint. The repo has to declare it in `installed_extensions/0` + # so migrations install it. use Spark.Dsl.Verifier alias Spark.Dsl.Verifier def verify(dsl) do - if is_nil(Ash.Resource.Info.temporal_strategy(dsl)) do + repo = AshPostgres.DataLayer.Info.repo(dsl, :mutate) + + if is_nil(Ash.Resource.Info.temporal_strategy(dsl)) or is_nil(repo) or + not Code.ensure_loaded?(repo) do :ok else - repo = AshPostgres.DataLayer.Info.repo(dsl, :mutate) - - with true <- not is_nil(repo) and Code.ensure_loaded?(repo), - true <- function_exported?(repo, :installed_extensions, 0), - false <- "btree_gist" in repo.installed_extensions() do - resource = Verifier.get_persisted(dsl, :module) - - raise Spark.Error.DslError, - module: resource, - message: """ - Temporal resource #{inspect(resource)} requires the `btree_gist` extension. - - Its `WITHOUT OVERLAPS` primary key is backed by a GiST exclusion constraint, - which needs `btree_gist`. Add it to your repo's `installed_extensions/0`: - - ```elixir - def installed_extensions do - ["btree_gist"] - end - ``` - """, - path: [:temporal] - else - _ -> :ok + resource = Verifier.get_persisted(dsl, :module) + + with :ok <- verify_min_pg_version(resource, repo) do + verify_btree_gist(resource, repo) end end end + + defp verify_min_pg_version(resource, repo) do + with true <- function_exported?(repo, :min_pg_version, 0), + %Version{major: major} = version when major < 18 <- repo.min_pg_version() do + raise Spark.Error.DslError, + module: resource, + message: """ + Temporal resource #{inspect(resource)} requires PostgreSQL 18 or later. + + `#{inspect(repo)}.min_pg_version/0` is #{version}. Temporal resources rely on + `WITHOUT OVERLAPS` primary keys, `PERIOD` foreign keys and `RETURNING old`, + which arrived in PostgreSQL 18. If your database is on 18 or later, say so: + + ```elixir + def min_pg_version do + %Version{major: 18, minor: 0, patch: 0} + end + ``` + """, + path: [:temporal] + else + _ -> :ok + end + end + + defp verify_btree_gist(resource, repo) do + with true <- function_exported?(repo, :installed_extensions, 0), + false <- "btree_gist" in repo.installed_extensions() do + raise Spark.Error.DslError, + module: resource, + message: """ + Temporal resource #{inspect(resource)} requires the `btree_gist` extension. + + Its `WITHOUT OVERLAPS` primary key is backed by a GiST exclusion constraint, + which needs `btree_gist`. Add it to your repo's `installed_extensions/0`: + + ```elixir + def installed_extensions do + ["btree_gist"] + end + ``` + """, + path: [:temporal] + else + _ -> :ok + end + end end diff --git a/mix.exs b/mix.exs index d792f91b..cbb8342f 100644 --- a/mix.exs +++ b/mix.exs @@ -120,6 +120,7 @@ defmodule AshPostgres.MixProject do "documentation/topics/advanced/schema-based-multitenancy.md", "documentation/topics/advanced/using-multiple-repos.md", "documentation/topics/advanced/upserts.md", + "documentation/topics/advanced/temporal-resources.md", {"documentation/dsls/DSL-AshPostgres.DataLayer.md", search_data: Spark.Docs.search_data_for(AshPostgres.DataLayer)}, "CHANGELOG.md" @@ -189,8 +190,7 @@ defmodule AshPostgres.MixProject do [ {:ash, ash_version("~> 3.34")}, {:spark, "~> 2.3 and >= 2.3.4"}, - {:ash_sql, - ash_sql_version("~> 0.8")}, + {:ash_sql, ash_sql_version("~> 0.8")}, {:igniter, "~> 0.6 and >= 0.6.29", optional: true}, {:ecto_sql, "~> 3.13"}, {:ecto, "~> 3.13"}, diff --git a/test/support/repo_case.ex b/test/support/repo_case.ex index 8a971ea5..f7ced59c 100644 --- a/test/support/repo_case.ex +++ b/test/support/repo_case.ex @@ -21,10 +21,17 @@ defmodule AshPostgres.RepoCase do end setup tags do - :ok = Sandbox.checkout(AshPostgres.TestRepo) + # `AshPostgres.TemporalTestRepo` is only started on PostgreSQL 18+. + repos = + [AshPostgres.TestRepo, AshPostgres.TemporalTestRepo] + |> Enum.filter(&Process.whereis/1) - if !tags[:async] do - Sandbox.mode(AshPostgres.TestRepo, {:shared, self()}) + for repo <- repos do + :ok = Sandbox.checkout(repo) + + if !tags[:async] do + Sandbox.mode(repo, {:shared, self()}) + end end :ok diff --git a/test/support/temporal/domain.ex b/test/support/temporal/domain.ex index 23b30c02..81189fc3 100644 --- a/test/support/temporal/domain.ex +++ b/test/support/temporal/domain.ex @@ -10,6 +10,7 @@ defmodule AshPostgres.Test.Temporal.Domain do resource AshPostgres.Test.Temporal.Subscription do define(:add_seat_if_basic, action: :add_seat_if_basic, get_by: [:id]) end + resource(AshPostgres.Test.Temporal.Tier) resource(AshPostgres.Test.Temporal.Event) resource(AshPostgres.Test.Temporal.Member) diff --git a/test/support/temporal/event.ex b/test/support/temporal/event.ex index a76a7e96..2285cc7c 100644 --- a/test/support/temporal/event.ex +++ b/test/support/temporal/event.ex @@ -10,7 +10,7 @@ defmodule AshPostgres.Test.Temporal.Event do postgres do table("event") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end attributes do diff --git a/test/support/temporal/member.ex b/test/support/temporal/member.ex index c7f0afc0..b0e29c90 100644 --- a/test/support/temporal/member.ex +++ b/test/support/temporal/member.ex @@ -10,7 +10,7 @@ defmodule AshPostgres.Test.Temporal.Member do postgres do table("member") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end temporal do diff --git a/test/support/temporal/subscription.ex b/test/support/temporal/subscription.ex index 17d17446..67f55911 100644 --- a/test/support/temporal/subscription.ex +++ b/test/support/temporal/subscription.ex @@ -10,7 +10,7 @@ defmodule AshPostgres.Test.Temporal.Subscription do postgres do table("subscription") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end temporal do @@ -87,7 +87,7 @@ defmodule AshPostgres.Test.Temporal.Subscription do accept([:tier]) end - # Atomic arithmetic update — exercised over FOR PORTION OF. + # Atomic arithmetic update — exercised over a period split. update :add_seat do require_atomic?(true) change(atomic_update(:seats, expr(seats + 1))) diff --git a/test/support/temporal/tier.ex b/test/support/temporal/tier.ex index 4695103f..beebc7cf 100644 --- a/test/support/temporal/tier.ex +++ b/test/support/temporal/tier.ex @@ -10,7 +10,7 @@ defmodule AshPostgres.Test.Temporal.Tier do postgres do table("tier") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end temporal do diff --git a/test/support/temporal_test_repo.ex b/test/support/temporal_test_repo.ex new file mode 100644 index 00000000..fc1c003a --- /dev/null +++ b/test/support/temporal_test_repo.ex @@ -0,0 +1,31 @@ +# SPDX-FileCopyrightText: 2019 ash_postgres contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshPostgres.TemporalTestRepo do + @moduledoc false + use AshPostgres.Repo, + otp_app: :ash_postgres + + # Temporal resources need PostgreSQL 18+ (see `AshPostgres.Verifiers.VerifyTemporal`), + # while the rest of the suite also runs on older servers. So they get a repo of their + # own, in the same database, that's only migrated and started when testing against 18+ + # (see `config/config.exs` and `test/test_helper.exs`). On older servers the resources + # still compile, and their tests are excluded by the `:postgres_18` tag. + # + # See `AshPostgres.TestPaths` for why the migrations live outside `priv/`. + def init(type, config) do + {:ok, config} = super(type, config) + {:ok, AshPostgres.TestPaths.put_paths(config, "temporal_test_repo")} + end + + def min_pg_version, do: %Version{major: 18, minor: 0, patch: 0} + + def installed_extensions, do: ["ash-functions", "citext", "btree_gist"] + + def prefer_transaction?, do: false + + def prefer_transaction_for_atomic_updates?, do: false + + def all_tenants, do: [] +end diff --git a/test/temporal_concurrency_test.exs b/test/temporal_concurrency_test.exs new file mode 100644 index 00000000..d5b8b82f --- /dev/null +++ b/test/temporal_concurrency_test.exs @@ -0,0 +1,185 @@ +# SPDX-FileCopyrightText: 2019 ash_postgres contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshPostgres.TemporalConcurrencyTest do + @moduledoc """ + Concurrent temporal writes under READ COMMITTED. A write that blocks on a version + another transaction is splitting must still apply once that transaction commits, even + though the version it found has moved and the leftovers are newer than its snapshot. + + These need real, committed transactions on separate connections, so they run outside + the sandbox and clean up after themselves. + """ + use ExUnit.Case, async: false + + @moduletag :temporal + @moduletag :postgres_18 + + require Ash.Query + alias AshPostgres.Test.Temporal.Subscription + alias AshPostgres.TemporalTestRepo + alias Ecto.Adapters.SQL.Sandbox + + @jan1 ~U[2026-01-01 00:00:00.000000Z] + @jan15 ~U[2026-01-15 00:00:00.000000Z] + @feb1 ~U[2026-02-01 00:00:00.000000Z] + + setup do + :ok = Sandbox.checkout(TemporalTestRepo, sandbox: false) + cleanup() + + TemporalTestRepo.query!( + "INSERT INTO tier (id, name, valid_at) VALUES (100, 'basic', tstzrange('2026-01-01', NULL))" + ) + + TemporalTestRepo.query!(""" + INSERT INTO subscription (id, tier, tier_id, seats, valid_at) VALUES + (101, 'orig', 100, 1, tstzrange('2026-01-01', NULL)), + (102, 'orig', 100, 1, tstzrange('2026-01-01', NULL)) + """) + + on_exit(fn -> + :ok = Sandbox.checkout(TemporalTestRepo, sandbox: false) + cleanup() + end) + end + + defp cleanup do + TemporalTestRepo.query!("DELETE FROM subscription WHERE id IN (101, 102, 103)") + TemporalTestRepo.query!("DELETE FROM tier WHERE id = 100") + end + + # Runs `fun` in a transaction on its own connection and holds it open until released. + defp hold_open(fun) do + parent = self() + + task = + Task.async(fn -> + :ok = Sandbox.checkout(TemporalTestRepo, sandbox: false) + + TemporalTestRepo.transaction(fn -> + fun.() + send(parent, :written) + + receive do + :commit -> :ok + end + end) + end) + + assert_receive :written, 5_000 + task + end + + # Starts `fun` on its own connection, lets it block on the held transaction, then + # commits that transaction so `fun` resumes. + defp race(held, fun) do + task = + Task.async(fn -> + :ok = Sandbox.checkout(TemporalTestRepo, sandbox: false) + fun.() + end) + + Process.sleep(300) + send(held.pid, :commit) + Task.await(held) + Task.await(task) + end + + defp change_tier(id, tier, as_of) do + Subscription + |> Ash.Query.filter(id == ^id) + |> Ash.Query.as_of(as_of) + |> Ash.read_one!() + |> Ash.Changeset.for_update(:change_tier, %{tier: tier}) + |> Ash.Changeset.as_of(as_of) + |> Ash.update!() + end + + defp add_seat_to_all(as_of) do + Subscription + |> Ash.Query.filter(id in [101, 102]) + |> Ash.Query.as_of(as_of) + |> Ash.bulk_update!(:add_seat, %{}, strategy: :atomic) + end + + defp upsert(id, tier, as_of) do + Subscription + |> Ash.Changeset.for_create(:create, %{id: id, tier: tier, tier_id: 100, seats: 1}) + |> Ash.Changeset.as_of(as_of) + |> Ash.create!(upsert?: true) + end + + defp timeline(id, column) do + TemporalTestRepo.query!( + "SELECT lower(valid_at), upper(valid_at), #{column} FROM subscription " <> + "WHERE id = $1 ORDER BY lower(valid_at)", + [id] + ).rows + end + + test "an update isn't lost when the version it found is split under it" do + held = hold_open(fn -> change_tier(101, "a", @feb1) end) + race(held, fn -> change_tier(101, "b", @jan15) end) + + assert [[@jan1, @jan15, "orig"], [@jan15, @feb1, "b"], [@feb1, nil, "a"]] = + timeline(101, "tier") + end + + test "an update isn't lost when the version it found is ended under it" do + held = + hold_open(fn -> + Subscription + |> Ash.Query.filter(id == 101) + |> Ash.Query.as_of(@feb1) + |> Ash.bulk_destroy!(:expire, %{}, strategy: :atomic) + end) + + race(held, fn -> change_tier(101, "b", @jan15) end) + + assert [[@jan1, @jan15, "orig"], [@jan15, @feb1, "b"]] = timeline(101, "tier") + end + + test "a bulk update isn't lost on any record whose version is split under it" do + held = hold_open(fn -> add_seat_to_all(@feb1) end) + race(held, fn -> add_seat_to_all(@jan15) end) + + for id <- [101, 102] do + assert [[@jan1, @jan15, 1], [@jan15, @feb1, 2], [@feb1, nil, 2]] = timeline(id, "seats") + end + end + + test "two upserts inserting the same new record both apply" do + held = hold_open(fn -> upsert(103, "a", @jan1) end) + race(held, fn -> upsert(103, "b", @feb1) end) + + assert [[@jan1, @feb1, "a"], [@feb1, nil, "b"]] = timeline(103, "tier") + end + + test "a bulk upsert retries just the record that lost an insert race" do + held = hold_open(fn -> upsert(103, "a", @jan1) end) + + result = + race(held, fn -> + Ash.bulk_create!( + [ + %{id: 101, tier: "b", tier_id: 100, seats: 1}, + %{id: 103, tier: "b", tier_id: 100, seats: 1} + ], + Subscription, + :create, + upsert?: true, + upsert_fields: [:tier], + as_of: @feb1, + return_records?: true, + return_errors?: true + ) + end) + + assert %Ash.BulkResult{status: :success, records: records} = result + assert records |> Enum.map(& &1.id) |> Enum.sort() == [101, 103] + assert [[@jan1, @feb1, "orig"], [@feb1, nil, "b"]] = timeline(101, "tier") + assert [[@jan1, @feb1, "a"], [@feb1, nil, "b"]] = timeline(103, "tier") + end +end diff --git a/test/temporal_identity_test.exs b/test/temporal_identity_test.exs index b3693a9c..2956d86e 100644 --- a/test/temporal_identity_test.exs +++ b/test/temporal_identity_test.exs @@ -3,12 +3,12 @@ # SPDX-License-Identifier: MIT defmodule AshPostgres.TemporalIdentityTest do - @moduledoc "Identities on temporal resources: period-aware uniqueness + as_of-anchored eager checks (PG19)." + @moduledoc "Identities on temporal resources: period-aware uniqueness + as_of-anchored eager checks (PG18+)." use AshPostgres.RepoCase, async: false @moduletag :temporal - @moduletag :postgres_19 + @moduletag :postgres_18 - alias AshPostgres.TestRepo + alias AshPostgres.TemporalTestRepo defmodule Domain do @moduledoc false @@ -25,7 +25,7 @@ defmodule AshPostgres.TemporalIdentityTest do postgres do table("temporal_id_thing") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end temporal do @@ -70,7 +70,7 @@ defmodule AshPostgres.TemporalIdentityTest do setup do # Mirror what the migration generator now emits for an identity on a temporal # resource: a period-aware GiST exclusion rather than a plain unique index. - TestRepo.query!(""" + TemporalTestRepo.query!(""" CREATE TABLE temporal_id_thing ( id integer NOT NULL, name text, @@ -95,7 +95,7 @@ defmodule AshPostgres.TemporalIdentityTest do test "an identity allows the same value across non-overlapping periods (history)" do # bounded historical "x" in [jan, mar) (the prior period of entity 1) - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO temporal_id_thing (id, name, valid_at) VALUES (1, 'x', tstzrange('2026-01-01','2026-03-01','[)'))" ) @@ -104,7 +104,9 @@ defmodule AshPostgres.TemporalIdentityTest do assert {:ok, _} = create(%{id: 1, name: "x"}, @mar) rows = - TestRepo.query!("SELECT name FROM temporal_id_thing WHERE id = 1 ORDER BY lower(valid_at)").rows + TemporalTestRepo.query!( + "SELECT name FROM temporal_id_thing WHERE id = 1 ORDER BY lower(valid_at)" + ).rows assert rows == [["x"], ["x"]] end @@ -118,7 +120,7 @@ defmodule AshPostgres.TemporalIdentityTest do near_future = DateTime.add(now, 365, :day) far_future = DateTime.add(now, 730, :day) - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO temporal_id_thing (id, name, valid_at) VALUES (9, 'z', tstzrange($1, $2, '[)'))", [past, near_future] ) @@ -131,7 +133,7 @@ defmodule AshPostgres.TemporalIdentityTest do test "eager_check is anchored at as_of: a create conflicting at that instant is rejected" do # "z" valid across [jan, jun) - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO temporal_id_thing (id, name, valid_at) VALUES (9, 'z', tstzrange('2026-01-01','2026-06-01','[)'))" ) diff --git a/test/temporal_migration_generator_test.exs b/test/temporal_migration_generator_test.exs index 6b5e91db..73ec74cd 100644 --- a/test/temporal_migration_generator_test.exs +++ b/test/temporal_migration_generator_test.exs @@ -3,10 +3,10 @@ # SPDX-License-Identifier: MIT defmodule AshPostgres.TemporalMigrationGeneratorTest do - @moduledoc "Asserts the migration generator emits temporal DDL (PG19)." + @moduledoc "Asserts the migration generator emits temporal DDL (PG18+)." use AshPostgres.RepoCase, async: false @moduletag :temporal - @moduletag :postgres_19 + @moduletag :postgres_18 @moduletag :tmp_dir defmodule GenTier do @@ -15,7 +15,7 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do postgres do table("gen_tier") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end temporal do @@ -54,7 +54,7 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do postgres do table("gen_sub") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end temporal do @@ -96,7 +96,7 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do postgres do table("gen_a_badge") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end temporal do @@ -146,7 +146,7 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do } end - test "emits range column, WITHOUT OVERLAPS PK, PERIOD FK and btree_gist, version-guarded", %{ + test "emits range column, WITHOUT OVERLAPS PK, PERIOD FK and btree_gist", %{ snapshot_path: snapshot_path, migration_path: migration_path } do @@ -168,20 +168,20 @@ defmodule AshPostgres.TemporalMigrationGeneratorTest do # range period column assert migration =~ ":tstzrange" - # temporal WITHOUT OVERLAPS primary key, version-guarded with a plain-PK fallback + # temporal WITHOUT OVERLAPS primary key, with no fallback: the repo's + # `min_pg_version/0` guarantees PostgreSQL 18+, so nothing is checked at migration time assert migration =~ "ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)" - assert migration =~ "ADD PRIMARY KEY (id)" - assert migration =~ "server_version_num" - assert migration =~ "190_000" + refute migration =~ "ADD PRIMARY KEY (id)" + refute migration =~ "server_version_num" - # temporal PERIOD foreign key (also version-guarded) + # temporal PERIOD foreign key assert migration =~ "FOREIGN KEY (tier_id, PERIOD valid_at)" assert migration =~ "PERIOD valid_at)" # identity on a temporal resource -> period-aware exclusion (WITHOUT OVERLAPS), - # version-guarded with a plain unique-index fallback on pre-PG19 servers. + # never a plain unique index assert migration =~ "EXCLUDE USING gist (name WITH =, valid_at WITH &&)" - assert migration =~ ~r/create unique_index\(:gen_tier, \[:name\]/ + refute migration =~ ~r/create unique_index\(:gen_tier, \[:name\]/ # GiST has no citext operator class, so citext keys compare as citext does assert migration =~ "EXCLUDE USING gist ((lower(email::text)) WITH =, valid_at WITH &&)" diff --git a/test/temporal_test.exs b/test/temporal_test.exs index 2dd2e0bb..6428bc6f 100644 --- a/test/temporal_test.exs +++ b/test/temporal_test.exs @@ -4,10 +4,10 @@ defmodule AshPostgres.TemporalTest do @moduledoc """ - End-to-end temporal (bitemporal) tests against Postgres 19. + End-to-end temporal (bitemporal) tests against PostgreSQL 18+. Exercises `as_of` time-travel reads, `now()` anchoring (filters/calcs/atomics), - the `Ash.Type.Range` <-> `Postgrex.Range` bridge, `FOR PORTION OF` mutations, + the `Ash.Type.Range` <-> `Postgrex.Range` bridge, period-splitting mutations, `range_overlaps/2`, and `temporal_keys` relationships. Tagged `:temporal`; run with: @@ -17,12 +17,12 @@ defmodule AshPostgres.TemporalTest do use AshPostgres.RepoCase, async: false @moduletag :temporal - @moduletag :postgres_19 + @moduletag :postgres_18 require Ash.Query require Ash.Expr alias AshPostgres.Test.Temporal.{Event, Member, Subscription, Tier} - alias AshPostgres.TestRepo + alias AshPostgres.TemporalTestRepo @jan15 ~U[2026-01-15 00:00:00.000000Z] @mar1 ~U[2026-03-01 00:00:00.000000Z] @@ -30,17 +30,17 @@ defmodule AshPostgres.TemporalTest do setup do # tier first: the temporal PERIOD foreign key requires the referenced tier # period to exist (and cover the subscription's period) before insert. - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO tier (id, name, valid_at) VALUES (10, 'basic', tstzrange('2026-01-01','2026-06-01','[)'))" ) - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO subscription (id, tier, tier_id, seats, activated_at, valid_at) VALUES (1, 'bronze', 10, 3, '2026-02-15', tstzrange('2026-01-01','2026-02-01','[)')), (1, 'gold', 10, 5, '2026-02-15', tstzrange('2026-02-01','2026-04-01','[)')) """) - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO event (id, name, created) VALUES (1, 'early', '2026-01-01'), (2, 'late', '2026-02-01') """) @@ -51,14 +51,14 @@ defmodule AshPostgres.TemporalTest do defp tiers(records), do: records |> Enum.map(& &1.tier) |> Enum.sort() defp subscription_timeline(id) do - TestRepo.query!( + TemporalTestRepo.query!( "SELECT tier, lower(valid_at), upper(valid_at) FROM subscription WHERE id = $1 ORDER BY lower(valid_at)", [id] ).rows end defp seats_timeline(id) do - TestRepo.query!( + TemporalTestRepo.query!( "SELECT tier, seats, lower(valid_at) FROM subscription WHERE id = $1 ORDER BY lower(valid_at)", [id] ).rows @@ -67,7 +67,7 @@ defmodule AshPostgres.TemporalTest do describe "default as_of (reads are current-state by default)" do test "get / read without as_of returns the currently-valid period, not all history" do # id=8: a past period + a current open-ended one (tier_id NULL avoids the PERIOD FK) - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO subscription (id, tier, tier_id, valid_at) VALUES (8, 'old', NULL, tstzrange('2020-01-01','2021-01-01','[)')), (8, 'current', NULL, tstzrange('2021-01-01', NULL, '[)')) @@ -102,7 +102,7 @@ defmodule AshPostgres.TemporalTest do describe "sorting through a temporal relationship" do setup do # tier 30 changes name across two adjacent periods; tier 40 is constant. - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO tier (id, name, valid_at) VALUES (30, 'mango', tstzrange('2026-01-01','2026-03-01','[)')), (30, 'apple', tstzrange('2026-03-01','2026-06-01','[)')), @@ -111,7 +111,7 @@ defmodule AshPostgres.TemporalTest do # Each subscription spans the whole window (sub 5's span covers BOTH of tier 30's # periods — the as_of join must pick the single covering tier row, not both). - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO subscription (id, tier, tier_id, valid_at) VALUES (5, 'sub5', 30, tstzrange('2026-01-01','2026-06-01','[)')), (6, 'sub6', 40, tstzrange('2026-01-01','2026-06-01','[)')) @@ -173,13 +173,13 @@ defmodule AshPostgres.TemporalTest do describe "Ash.load carries as_of (like tenant)" do setup do # tier 50 changes name across periods; subscription 7 spans both. - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO tier (id, name, valid_at) VALUES (50, 'old', tstzrange('2026-01-01','2026-03-01','[)')), (50, 'new', tstzrange('2026-03-01','2026-06-01','[)')) """) - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO subscription (id, tier, tier_id, valid_at) VALUES (7, 'sub7', 50, tstzrange('2026-01-01','2026-06-01','[)')) """) @@ -287,14 +287,14 @@ defmodule AshPostgres.TemporalTest do end end - describe "FOR PORTION OF preserves an unbounded (NULL) upper" do + describe "splitting preserves an unbounded (NULL) upper" do test "updating a current row yields [as_of, ∞) with no 'infinity' upper or junk row" do # open-ended "current" row ([2026-01-01, ∞)); tier_id NULL avoids the PERIOD FK - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO subscription (id, tier, tier_id, seats, valid_at) VALUES (20, 'x', NULL, 1, tstzrange('2026-01-01', NULL, '[)'))" ) - # An atomic FOR PORTION OF update at @mar1 splits [jan,∞) -> [jan,mar1) + [mar1,∞). + # An atomic update at @mar1 splits [jan,∞) -> [jan,mar1) + [mar1,∞). # Returning the new slice previously crashed (literal 'infinity' upper); now it's NULL. updated = Subscription @@ -307,13 +307,13 @@ defmodule AshPostgres.TemporalTest do # exactly two contiguous periods, the later one unbounded, and NO [infinity, ) junk row rows = - TestRepo.query!( + TemporalTestRepo.query!( "SELECT upper_inf(valid_at), seats FROM subscription WHERE id = 20 ORDER BY lower(valid_at)" ).rows assert rows == [[false, 1], [true, 2]] - assert TestRepo.query!( + assert TemporalTestRepo.query!( "SELECT count(*) FROM subscription WHERE id = 20 AND lower(valid_at) = 'infinity'" ).rows == [[0]] end @@ -357,7 +357,7 @@ defmodule AshPostgres.TemporalTest do test "reads the rows valid at the current instant" do # An open-ended ([2020, ∞)) row that always covers the wall clock. # `tier_id` is NULL so it isn't subject to the temporal PERIOD FK. - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO subscription (id, tier, tier_id, valid_at) VALUES (9, 'current', NULL, tstzrange('2020-01-01', NULL, '[)'))" ) @@ -467,7 +467,7 @@ defmodule AshPostgres.TemporalTest do end end - describe "FOR PORTION OF mutations" do + describe "period-splitting mutations" do test "update splits the row at as_of, applying the new value forward" do [gold] = Subscription |> Ash.Query.filter(id == 1) |> Ash.Query.as_of(@mar1) |> Ash.read!() @@ -721,7 +721,7 @@ defmodule AshPostgres.TemporalTest do end test "an atomic update through an exists at the current instant leaves later versions alone" do - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO subscription (id, tier, seats, valid_at) VALUES (2, 'now', 1, tstzrange(now() - interval '1 day', now() + interval '1 day', '[)')), (2, 'later', 9, tstzrange(now() + interval '1 day', NULL, '[)')) @@ -1010,7 +1010,7 @@ defmodule AshPostgres.TemporalTest do describe "aggregates respect as_of" do test "a count over a temporal has_many counts only periods valid at as_of" do # add a second subscription (id=3) for tier 10, valid only [Mar,May) - TestRepo.query!(""" + TemporalTestRepo.query!(""" INSERT INTO subscription (id, tier, tier_id, seats, valid_at) VALUES (3, 'silver', 10, 1, tstzrange('2026-03-01','2026-05-01','[)')) """) @@ -1036,7 +1036,7 @@ defmodule AshPostgres.TemporalTest do describe "bulk_create" do test "distinct keys succeed; same key at the same as_of raises the exclusion constraint" do # open-ended tier so each create's `[as_of, ∞)` satisfies the PERIOD FK - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO tier (id, name, valid_at) VALUES (20, 'open', tstzrange('2026-01-01', NULL, '[)'))" ) @@ -1087,7 +1087,7 @@ defmodule AshPostgres.TemporalTest do end end - describe "temporal upsert (atomic FOR-PORTION-OF-or-insert CTE)" do + describe "temporal upsert (atomic split-or-insert CTE)" do test "match: splits the period valid at as_of, applying new values forward" do # id=1: bronze [jan,feb), gold [feb,apr). Upsert at mar1 (inside gold). assert {:ok, result} = @@ -1108,7 +1108,7 @@ defmodule AshPostgres.TemporalTest do end test "match: future periods (that don't contain as_of) are untouched" do - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO subscription (id, tier, tier_id, seats, valid_at) VALUES (1,'future',10,1,tstzrange('2026-05-01','2026-06-01','[)'))" ) @@ -1124,13 +1124,13 @@ defmodule AshPostgres.TemporalTest do |> Ash.create(upsert?: true) assert [["future", 1]] = - TestRepo.query!( + TemporalTestRepo.query!( "SELECT tier, seats FROM subscription WHERE id=1 AND lower(valid_at)='2026-05-01'" ).rows end test "no match: inserts a new period gap-filled to infinity" do - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO tier (id, name, valid_at) VALUES (20, 'open', tstzrange('2026-01-01', NULL, '[)'))" ) @@ -1143,14 +1143,28 @@ defmodule AshPostgres.TemporalTest do assert result.tier == "new" assert [["new", "2026-01-15 00:00:00+00", nil]] = - TestRepo.query!( + TemporalTestRepo.query!( "SELECT tier, lower(valid_at)::text, upper(valid_at) FROM subscription WHERE id=2" ).rows end + test "an identity violation still raises (only the primary key is retried)" do + TemporalTestRepo.query!( + "INSERT INTO member (id, email, valid_at) VALUES (1, 'taken', tstzrange('2026-01-01', NULL))" + ) + + assert {:error, %Ash.Error.Invalid{} = error} = + Member + |> Ash.Changeset.for_create(:create, %{id: 2, email: "taken"}) + |> Ash.Changeset.as_of(@mar1) + |> Ash.create(upsert?: true) + + assert Exception.message(error) =~ "email" + end + test "bulk: upserts many keys in one statement (match splits, miss inserts)" do # open-ended tier so the new id=5 row's [mar1, ∞) satisfies the PERIOD FK - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO tier (id, name, valid_at) VALUES (20, 'open', tstzrange('2026-01-01', NULL, '[)'))" ) @@ -1176,9 +1190,9 @@ defmodule AshPostgres.TemporalTest do end describe "bulk_update / update_many" do - # The `FOR PORTION OF FROM` bound is sourced from the query context as_of + # The split point is sourced from the query context as_of # (via `temporal_from_bound`), which carries it for both single and bulk. - test "each matched row is split at as_of (FOR PORTION OF, set-wide)" do + test "each matched row is split at as_of (set-wide)" do Subscription |> Ash.Query.filter(id == 1) |> Ash.Query.as_of(@mar1) @@ -1195,7 +1209,7 @@ defmodule AshPostgres.TemporalTest do describe "bulk_destroy" do # FROM bound sourced from query context as_of (via `temporal_from_bound`). - test "each matched row's validity is truncated at as_of (FOR PORTION OF DELETE)" do + test "each matched row's validity is truncated at as_of (temporal DELETE)" do Subscription |> Ash.Query.filter(id == 1) |> Ash.Query.as_of(@mar1) @@ -1209,7 +1223,7 @@ defmodule AshPostgres.TemporalTest do end describe "temporal PERIOD foreign keys reject referential actions" do - test "on_delete on a temporal relationship fails to compile (PG19: NO ACTION only)" do + test "on_delete on a temporal relationship fails to compile (PERIOD FKs: NO ACTION only)" do assert_raise Spark.Error.DslError, ~r/PERIOD|NO ACTION/, fn -> defmodule BadTemporalRef do use Ash.Resource, @@ -1219,7 +1233,7 @@ defmodule AshPostgres.TemporalTest do postgres do table("bad_temporal_ref") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) references do reference(:tier_record, on_delete: :delete) @@ -1255,6 +1269,63 @@ defmodule AshPostgres.TemporalTest do end end + describe "a write that keeps conflicting" do + test "fails with an Ash framework error that says retrying is safe" do + error = + Ash.Error.to_error_class( + AshPostgres.Temporal.WriteConflict.exception(resource: Subscription, attempts: 25) + ) + + assert %Ash.Error.Framework{errors: [%AshPostgres.Temporal.WriteConflict{}]} = error + assert Exception.message(error) =~ "conflicted with concurrent writes" + assert Exception.message(error) =~ "Retrying the action is safe" + end + end + + describe "a temporal resource's repo must target PostgreSQL 18+" do + defmodule PrePg18Repo do + @moduledoc false + use AshPostgres.Repo, otp_app: :ash_postgres + + def min_pg_version, do: %Version{major: 17, minor: 0, patch: 0} + def installed_extensions, do: ["ash-functions", "btree_gist"] + end + + # Verifiers run after a module is verified, which doesn't happen synchronously for one + # defined in a test, so this runs the verifier on the resource directly. + test "a repo whose min_pg_version is below 18 fails to compile" do + # Spark reports the verifier's error on its own once the module is verified. + ExUnit.CaptureIO.capture_io(:stderr, fn -> + defmodule TemporalOnPrePg18 do + use Ash.Resource, + domain: nil, + validate_domain_inclusion?: false, + data_layer: AshPostgres.DataLayer + + postgres do + table("temporal_on_pre_pg18") + repo(PrePg18Repo) + end + + temporal do + strategy(:context) + attribute(:valid_at) + end + + attributes do + attribute(:id, :integer, primary_key?: true, allow_nil?: false, public?: true) + end + end + end) + + assert_raise Spark.Error.DslError, ~r/requires PostgreSQL 18 or later/, fn -> + AshPostgres.Verifiers.VerifyTemporal.verify( + AshPostgres.TemporalTest.TemporalOnPrePg18.spark_dsl_config() + ) + end + end + end + describe "the period attribute is not settable as input" do test "a resource that accepts the temporal attribute fails to compile" do assert_raise Spark.Error.DslError, ~r/must not be accepted as input/, fn -> @@ -1266,7 +1337,7 @@ defmodule AshPostgres.TemporalTest do postgres do table("bad_temporal") - repo(AshPostgres.TestRepo) + repo(AshPostgres.TemporalTestRepo) end temporal do @@ -1313,7 +1384,7 @@ defmodule AshPostgres.TemporalTest do # and the tier itself ended at mar1 assert [["2026-01-01 00:00:00+00", "2026-03-01 00:00:00+00"]] = - TestRepo.query!( + TemporalTestRepo.query!( "SELECT lower(valid_at)::text, upper(valid_at)::text FROM tier WHERE id = 10" ).rows end @@ -1321,7 +1392,7 @@ defmodule AshPostgres.TemporalTest do describe "cascade destroy carries the destroy's as_of" do defp tier_timeline do - TestRepo.query!( + TemporalTestRepo.query!( "SELECT lower(valid_at)::text, upper(valid_at)::text FROM tier WHERE id = 10 ORDER BY lower(valid_at)" ).rows end @@ -1410,7 +1481,7 @@ defmodule AshPostgres.TemporalTest do # the subscription's [jan15, ∞) (the temporal PERIOD FK requires it). A child written # at "now" would be [now, ∞) and fail the FK / show the wrong lower bound. assert [["2026-01-15 00:00:00+00", nil]] = - TestRepo.query!( + TemporalTestRepo.query!( "SELECT lower(valid_at)::text, upper(valid_at) FROM tier WHERE id = 30" ).rows end @@ -1443,7 +1514,7 @@ defmodule AshPostgres.TemporalTest do describe "recorded_at is the wall clock, not as_of" do defp recorded_timeline(id) do - TestRepo.query!( + TemporalTestRepo.query!( "SELECT tier, recorded_at FROM subscription WHERE id = $1 ORDER BY lower(valid_at)", [id] ).rows @@ -1461,7 +1532,7 @@ defmodule AshPostgres.TemporalTest do end test "a back-dated create is recorded now" do - TestRepo.query!( + TemporalTestRepo.query!( "INSERT INTO tier (id, name, valid_at) VALUES (20, 'open', tstzrange('2026-01-01', NULL, '[)'))" ) diff --git a/test/test_helper.exs b/test/test_helper.exs index 82928b22..2687a430 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -32,6 +32,11 @@ AshPostgres.TestRepo.start_link() AshPostgres.DevTestRepo.start_link() AshPostgres.TestNoSandboxRepo.start_link() +# Only migrated on PostgreSQL 18+ (see `AshPostgres.TemporalTestRepo`). +if pg_version >= 18 do + AshPostgres.TemporalTestRepo.start_link() +end + format_sql_query = try do case System.shell("which pg_format") do diff --git a/test_priv/resource_snapshots/test_repo/event/20260627032019.json b/test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json similarity index 89% rename from test_priv/resource_snapshots/test_repo/event/20260627032019.json rename to test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json index a07709a8..f6f152e9 100644 --- a/test_priv/resource_snapshots/test_repo/event/20260627032019.json +++ b/test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json @@ -43,14 +43,14 @@ "custom_indexes": [], "custom_statements": [], "has_create_action": false, - "hash": "564460F32319359E8E086E341F05793F960D4A5AD72521313CFFA05C7CBDDD1A", + "hash": "EA6BD551D1DD7BB8A2CD2B99E3009290FF8F1AE54431BBF8699E289B65148F58", "identities": [], "multitenancy": { "attribute": null, "global": null, "strategy": null }, - "repo": "Elixir.AshPostgres.TestRepo", + "repo": "Elixir.AshPostgres.TemporalTestRepo", "schema": null, "table": "event", "temporal": null diff --git a/test_priv/resource_snapshots/test_repo/event/20260627032019.json.license b/test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json.license similarity index 100% rename from test_priv/resource_snapshots/test_repo/event/20260627032019.json.license rename to test_priv/resource_snapshots/temporal_test_repo/event/20261004073135.json.license diff --git a/test_priv/resource_snapshots/temporal_test_repo/extensions.json b/test_priv/resource_snapshots/temporal_test_repo/extensions.json new file mode 100644 index 00000000..c958e4d1 --- /dev/null +++ b/test_priv/resource_snapshots/temporal_test_repo/extensions.json @@ -0,0 +1,8 @@ +{ + "ash_functions_version": 7, + "installed": [ + "ash-functions", + "citext", + "btree_gist" + ] +} \ No newline at end of file diff --git a/test_priv/resource_snapshots/test_repo/event/20260824043711.json.license b/test_priv/resource_snapshots/temporal_test_repo/extensions.json.license similarity index 100% rename from test_priv/resource_snapshots/test_repo/event/20260824043711.json.license rename to test_priv/resource_snapshots/temporal_test_repo/extensions.json.license diff --git a/test_priv/resource_snapshots/test_repo/member/20260926192339.json b/test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json similarity index 92% rename from test_priv/resource_snapshots/test_repo/member/20260926192339.json rename to test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json index 90497a1f..7e9cb5a8 100644 --- a/test_priv/resource_snapshots/test_repo/member/20260926192339.json +++ b/test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json @@ -43,7 +43,7 @@ "custom_indexes": [], "custom_statements": [], "has_create_action": true, - "hash": "82A6392870511BE46BFCBE836F3527E4D4969B8C418C08DA4FC7D34D423850B1", + "hash": "D2D92A32204963CA53D91F4E82EA30039FFEBDF4CCF3D62641344C50E8610746", "identities": [ { "all_tenants?": false, @@ -65,7 +65,7 @@ "global": null, "strategy": null }, - "repo": "Elixir.AshPostgres.TestRepo", + "repo": "Elixir.AshPostgres.TemporalTestRepo", "schema": null, "table": "member", "temporal": { diff --git a/test_priv/resource_snapshots/test_repo/member/20260926192339.json.license b/test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json.license similarity index 100% rename from test_priv/resource_snapshots/test_repo/member/20260926192339.json.license rename to test_priv/resource_snapshots/temporal_test_repo/member/20261004073136.json.license diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json b/test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json similarity index 95% rename from test_priv/resource_snapshots/test_repo/subscription/20260927024130.json rename to test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json index 841f3658..9de1f143 100644 --- a/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json +++ b/test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json @@ -116,14 +116,14 @@ "custom_indexes": [], "custom_statements": [], "has_create_action": true, - "hash": "FA1ADDB90BF620ACD6E8BEFFE489DD56995132A8BE5F05281DFAC3C5858087BB", + "hash": "0D9D101FE4C593F8EFFC03DC84F5B4249ECEC5C08E2F699816D74C9D37FE3F0E", "identities": [], "multitenancy": { "attribute": null, "global": null, "strategy": null }, - "repo": "Elixir.AshPostgres.TestRepo", + "repo": "Elixir.AshPostgres.TemporalTestRepo", "schema": null, "table": "subscription", "temporal": { diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260627032021.json.license b/test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json.license similarity index 100% rename from test_priv/resource_snapshots/test_repo/subscription/20260627032021.json.license rename to test_priv/resource_snapshots/temporal_test_repo/subscription/20261004073137.json.license diff --git a/test_priv/resource_snapshots/test_repo/tier/20260824043712.json b/test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json similarity index 90% rename from test_priv/resource_snapshots/test_repo/tier/20260824043712.json rename to test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json index 11549929..9ab54f2e 100644 --- a/test_priv/resource_snapshots/test_repo/tier/20260824043712.json +++ b/test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json @@ -43,14 +43,14 @@ "custom_indexes": [], "custom_statements": [], "has_create_action": true, - "hash": "94A1C6C459EB720CC502D358935D76F39DD1D36D15A5868B67BCA4A16860BA1E", + "hash": "4C05418764E40BBAC82177FD5660CB06CD04C36A81D47C40D53A1FBFA02BCF00", "identities": [], "multitenancy": { "attribute": null, "global": null, "strategy": null }, - "repo": "Elixir.AshPostgres.TestRepo", + "repo": "Elixir.AshPostgres.TemporalTestRepo", "schema": null, "table": "tier", "temporal": { diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260824043713.json.license b/test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json.license similarity index 100% rename from test_priv/resource_snapshots/test_repo/subscription/20260824043713.json.license rename to test_priv/resource_snapshots/temporal_test_repo/tier/20261004073138.json.license diff --git a/test_priv/resource_snapshots/test_repo/event/20260824043711.json b/test_priv/resource_snapshots/test_repo/event/20260824043711.json deleted file mode 100644 index 13fa9083..00000000 --- a/test_priv/resource_snapshots/test_repo/event/20260824043711.json +++ /dev/null @@ -1,57 +0,0 @@ -{ - "attributes": [ - { - "allow_nil?": false, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": true, - "references": null, - "scale": null, - "size": null, - "source": "id", - "type": "bigint" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "name", - "type": "text" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "created", - "type": "utc_datetime_usec" - } - ], - "base_filter": null, - "check_constraints": [], - "create_table_options": null, - "custom_indexes": [], - "custom_statements": [], - "has_create_action": false, - "hash": "E7967071B2D8054BFE5787408ED2DE625F213CCD28C592078468D413C2BB5B8C", - "identities": [], - "multitenancy": { - "attribute": null, - "global": null, - "strategy": null - }, - "repo": "Elixir.AshPostgres.TestRepo", - "schema": null, - "table": "event", - "temporal": null -} \ No newline at end of file diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260627032021.json b/test_priv/resource_snapshots/test_repo/subscription/20260627032021.json deleted file mode 100644 index f4b21c3c..00000000 --- a/test_priv/resource_snapshots/test_repo/subscription/20260627032021.json +++ /dev/null @@ -1,119 +0,0 @@ -{ - "attributes": [ - { - "allow_nil?": false, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": true, - "references": null, - "scale": null, - "size": null, - "source": "id", - "type": "bigint" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "tier", - "type": "text" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": { - "deferrable": false, - "destination_attribute": "id", - "destination_attribute_default": null, - "destination_attribute_generated": null, - "index?": false, - "match_type": null, - "match_with": null, - "multitenancy": { - "attribute": null, - "global": null, - "strategy": null - }, - "name": "subscription_tier_id_fkey", - "on_delete": null, - "on_update": null, - "primary_key?": true, - "schema": "public", - "table": "tier", - "temporal_period": [ - "valid_at", - "valid_at" - ] - }, - "scale": null, - "size": null, - "source": "tier_id", - "type": "bigint" - }, - { - "allow_nil?": true, - "default": "0", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "seats", - "type": "bigint" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "activated_at", - "type": "utc_datetime_usec" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "valid_at", - "type": "tstzrange" - } - ], - "base_filter": null, - "check_constraints": [], - "create_table_options": null, - "custom_indexes": [], - "custom_statements": [], - "has_create_action": true, - "hash": "7B9A78F8010D1D9C80CE76721335887B4C40C4554C82C93692332AA6DF7A84D6", - "identities": [], - "multitenancy": { - "attribute": null, - "global": null, - "strategy": null - }, - "repo": "Elixir.AshPostgres.TestRepo", - "schema": null, - "table": "subscription", - "temporal": { - "attribute": "valid_at", - "strategy": "context" - } -} \ No newline at end of file diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260824043713.json b/test_priv/resource_snapshots/test_repo/subscription/20260824043713.json deleted file mode 100644 index e1c66e20..00000000 --- a/test_priv/resource_snapshots/test_repo/subscription/20260824043713.json +++ /dev/null @@ -1,121 +0,0 @@ -{ - "attributes": [ - { - "allow_nil?": false, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": true, - "references": null, - "scale": null, - "size": null, - "source": "id", - "type": "bigint" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "tier", - "type": "text" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": { - "deferrable": false, - "destination_attribute": "id", - "destination_attribute_default": null, - "destination_attribute_generated": null, - "index?": false, - "index_where": null, - "match_tenant?": false, - "match_type": null, - "match_with": null, - "multitenancy": { - "attribute": null, - "global": null, - "strategy": null - }, - "name": "subscription_tier_id_fkey", - "on_delete": null, - "on_update": null, - "primary_key?": true, - "schema": "public", - "table": "tier", - "temporal_period": [ - "valid_at", - "valid_at" - ] - }, - "scale": null, - "size": null, - "source": "tier_id", - "type": "bigint" - }, - { - "allow_nil?": true, - "default": "0", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "seats", - "type": "bigint" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "activated_at", - "type": "utc_datetime_usec" - }, - { - "allow_nil?": false, - "default": "nil", - "generated?": true, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "valid_at", - "type": "tstzrange" - } - ], - "base_filter": null, - "check_constraints": [], - "create_table_options": null, - "custom_indexes": [], - "custom_statements": [], - "has_create_action": true, - "hash": "BB2835F7273E3D9FC169BC7715F0E8363991B560D5C23FCFB3CF9D7A41B2FAE5", - "identities": [], - "multitenancy": { - "attribute": null, - "global": null, - "strategy": null - }, - "repo": "Elixir.AshPostgres.TestRepo", - "schema": null, - "table": "subscription", - "temporal": { - "attribute": "valid_at", - "strategy": "context" - } -} \ No newline at end of file diff --git a/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json.license b/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json.license deleted file mode 100644 index c5bd9989..00000000 --- a/test_priv/resource_snapshots/test_repo/subscription/20260927024130.json.license +++ /dev/null @@ -1,3 +0,0 @@ -SPDX-FileCopyrightText: 2019 ash_postgres contributors - -SPDX-License-Identifier: MIT diff --git a/test_priv/resource_snapshots/test_repo/tier/20260627032020.json b/test_priv/resource_snapshots/test_repo/tier/20260627032020.json deleted file mode 100644 index ea494646..00000000 --- a/test_priv/resource_snapshots/test_repo/tier/20260627032020.json +++ /dev/null @@ -1,60 +0,0 @@ -{ - "attributes": [ - { - "allow_nil?": false, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": true, - "references": null, - "scale": null, - "size": null, - "source": "id", - "type": "bigint" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "name", - "type": "text" - }, - { - "allow_nil?": true, - "default": "nil", - "generated?": false, - "precision": null, - "primary_key?": false, - "references": null, - "scale": null, - "size": null, - "source": "valid_at", - "type": "tstzrange" - } - ], - "base_filter": null, - "check_constraints": [], - "create_table_options": null, - "custom_indexes": [], - "custom_statements": [], - "has_create_action": false, - "hash": "D29673CAB290355F8F1081580F32CA27552F9B5B68ECCA0E31EA453FD9426BD2", - "identities": [], - "multitenancy": { - "attribute": null, - "global": null, - "strategy": null - }, - "repo": "Elixir.AshPostgres.TestRepo", - "schema": null, - "table": "tier", - "temporal": { - "attribute": "valid_at", - "strategy": "context" - } -} \ No newline at end of file diff --git a/test_priv/resource_snapshots/test_repo/tier/20260627032020.json.license b/test_priv/resource_snapshots/test_repo/tier/20260627032020.json.license deleted file mode 100644 index c5bd9989..00000000 --- a/test_priv/resource_snapshots/test_repo/tier/20260627032020.json.license +++ /dev/null @@ -1,3 +0,0 @@ -SPDX-FileCopyrightText: 2019 ash_postgres contributors - -SPDX-License-Identifier: MIT diff --git a/test_priv/resource_snapshots/test_repo/tier/20260824043712.json.license b/test_priv/resource_snapshots/test_repo/tier/20260824043712.json.license deleted file mode 100644 index c5bd9989..00000000 --- a/test_priv/resource_snapshots/test_repo/tier/20260824043712.json.license +++ /dev/null @@ -1,3 +0,0 @@ -SPDX-FileCopyrightText: 2019 ash_postgres contributors - -SPDX-License-Identifier: MIT diff --git a/test_priv/temporal_test_repo/migrations/20261004073133_migrate_resources_extensions_1.exs b/test_priv/temporal_test_repo/migrations/20261004073133_migrate_resources_extensions_1.exs new file mode 100644 index 00000000..ecdbae3d --- /dev/null +++ b/test_priv/temporal_test_repo/migrations/20261004073133_migrate_resources_extensions_1.exs @@ -0,0 +1,136 @@ +# SPDX-FileCopyrightText: 2019 ash_postgres contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshPostgres.TemporalTestRepo.Migrations.MigrateResourcesExtensions1 do + @moduledoc """ + Installs any extensions that are mentioned in the repo's `installed_extensions/0` callback + + This file was autogenerated with `mix ash_postgres.generate_migrations` + """ + + use Ecto.Migration + + def up do + execute(""" + CREATE OR REPLACE FUNCTION ash_elixir_or(left BOOLEAN, in right ANYCOMPATIBLE, out f1 ANYCOMPATIBLE) + AS $$ SELECT COALESCE(NULLIF($1, FALSE), $2) $$ + LANGUAGE SQL + SET search_path = '' + IMMUTABLE; + """) + + execute(""" + CREATE OR REPLACE FUNCTION ash_elixir_or(left ANYCOMPATIBLE, in right ANYCOMPATIBLE, out f1 ANYCOMPATIBLE) + AS $$ SELECT COALESCE($1, $2) $$ + LANGUAGE SQL + SET search_path = '' + IMMUTABLE; + """) + + execute(""" + CREATE OR REPLACE FUNCTION ash_elixir_and(left BOOLEAN, in right ANYCOMPATIBLE, out f1 ANYCOMPATIBLE) AS $$ + SELECT CASE + WHEN $1 IS TRUE THEN $2 + ELSE $1 + END $$ + LANGUAGE SQL + SET search_path = '' + IMMUTABLE; + """) + + execute(""" + CREATE OR REPLACE FUNCTION ash_elixir_and(left ANYCOMPATIBLE, in right ANYCOMPATIBLE, out f1 ANYCOMPATIBLE) AS $$ + SELECT CASE + WHEN $1 IS NOT NULL THEN $2 + ELSE $1 + END $$ + LANGUAGE SQL + SET search_path = '' + IMMUTABLE; + """) + + execute(""" + CREATE OR REPLACE FUNCTION ash_trim_whitespace(arr text[]) + RETURNS text[] AS $$ + DECLARE + start_index INT = 1; + end_index INT = array_length(arr, 1); + BEGIN + WHILE start_index <= end_index AND arr[start_index] = '' LOOP + start_index := start_index + 1; + END LOOP; + + WHILE end_index >= start_index AND arr[end_index] = '' LOOP + end_index := end_index - 1; + END LOOP; + + IF start_index > end_index THEN + RETURN ARRAY[]::text[]; + ELSE + RETURN arr[start_index : end_index]; + END IF; + END; $$ + LANGUAGE plpgsql + SET search_path = '' + IMMUTABLE; + """) + + execute(""" + CREATE OR REPLACE FUNCTION ash_raise_error(json_data jsonb) + RETURNS BOOLEAN AS $$ + BEGIN + -- Raise an error with the provided JSON data. + -- The JSON object is converted to text for inclusion in the error message. + RAISE EXCEPTION 'ash_error: %', json_data::text; + RETURN NULL; + END; + $$ LANGUAGE plpgsql + STABLE + SET search_path = ''; + """) + + execute(""" + CREATE OR REPLACE FUNCTION ash_raise_error(json_data jsonb, type_signal ANYCOMPATIBLE) + RETURNS ANYCOMPATIBLE AS $$ + BEGIN + -- Raise an error with the provided JSON data. + -- The JSON object is converted to text for inclusion in the error message. + RAISE EXCEPTION 'ash_error: %', json_data::text; + RETURN NULL; + END; + $$ LANGUAGE plpgsql + STABLE + SET search_path = ''; + """) + + execute(""" + CREATE OR REPLACE FUNCTION ash_required(value ANYCOMPATIBLE, payload jsonb) + RETURNS ANYCOMPATIBLE AS $$ + BEGIN + IF value IS NULL THEN + RAISE EXCEPTION 'ash_error: %', payload::text; + END IF; + + RETURN value; + END; + $$ LANGUAGE plpgsql + STABLE + SET search_path = ''; + """) + + execute("CREATE EXTENSION IF NOT EXISTS \"citext\"") + execute("CREATE EXTENSION IF NOT EXISTS \"btree_gist\"") + end + + def down do + # Uncomment this if you actually want to uninstall the extensions + # when this migration is rolled back: + execute( + "DROP FUNCTION IF EXISTS uuid_generate_v7(), ash_raise_error(jsonb), ash_raise_error(jsonb, ANYCOMPATIBLE), ash_elixir_and(BOOLEAN, ANYCOMPATIBLE), ash_elixir_and(ANYCOMPATIBLE, ANYCOMPATIBLE), ash_elixir_or(ANYCOMPATIBLE, ANYCOMPATIBLE), ash_elixir_or(BOOLEAN, ANYCOMPATIBLE), ash_trim_whitespace(text[]), ash_required(ANYCOMPATIBLE, jsonb)" + ) + + # execute("DROP EXTENSION IF EXISTS \"citext\"") + # execute("DROP EXTENSION IF EXISTS \"btree_gist\"") + end +end diff --git a/test_priv/test_repo/migrations/20260824043710_migrate_resources78.exs b/test_priv/temporal_test_repo/migrations/20261004073134_migrate_resources1.exs similarity index 51% rename from test_priv/test_repo/migrations/20260824043710_migrate_resources78.exs rename to test_priv/temporal_test_repo/migrations/20261004073134_migrate_resources1.exs index 3c1ee750..c5704ce0 100644 --- a/test_priv/test_repo/migrations/20260824043710_migrate_resources78.exs +++ b/test_priv/temporal_test_repo/migrations/20261004073134_migrate_resources1.exs @@ -2,7 +2,7 @@ # # SPDX-License-Identifier: MIT -defmodule AshPostgres.TestRepo.Migrations.MigrateResources78 do +defmodule AshPostgres.TemporalTestRepo.Migrations.MigrateResources1 do @moduledoc """ Updates resources based on their most recent snapshots. @@ -18,18 +18,17 @@ defmodule AshPostgres.TestRepo.Migrations.MigrateResources78 do add(:created, :utc_datetime_usec) end - create table(:tier, primary_key: false) do + create table(:member, primary_key: false) do add(:id, :bigint, null: false) - add(:name, :text) + add(:email, :citext) add(:valid_at, :tstzrange, null: false) end - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= - 190_000 do - execute("ALTER TABLE \"tier\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)") - else - execute("ALTER TABLE \"tier\" ADD PRIMARY KEY (id)") - end + execute("ALTER TABLE \"member\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)") + + execute( + "ALTER TABLE \"member\" ADD CONSTRAINT \"member_unique_email_index\" EXCLUDE USING gist ((lower(email::text)) WITH =, valid_at WITH &&)" + ) create table(:subscription, primary_key: false) do add(:id, :bigint, null: false) @@ -40,27 +39,42 @@ defmodule AshPostgres.TestRepo.Migrations.MigrateResources78 do add(:valid_at, :tstzrange, null: false) end - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= - 190_000 do - execute("ALTER TABLE \"subscription\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)") - else - execute("ALTER TABLE \"subscription\" ADD PRIMARY KEY (id)") - end + execute("ALTER TABLE \"subscription\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)") - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= - 190_000 do - execute( - "ALTER TABLE \"subscription\" ADD CONSTRAINT subscription_tier_id_fkey FOREIGN KEY (tier_id, PERIOD valid_at) REFERENCES \"public\".\"tier\" (id, PERIOD valid_at)" + alter table(:subscription) do + add(:recorded_at, :utc_datetime_usec, + null: false, + default: fragment("(now() AT TIME ZONE 'utc')") ) end + + create table(:tier, primary_key: false) do + add(:id, :bigint, null: false) + add(:name, :text) + add(:valid_at, :tstzrange, null: false) + end + + execute("ALTER TABLE \"tier\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)") + + execute( + "ALTER TABLE \"subscription\" ADD CONSTRAINT subscription_tier_id_fkey FOREIGN KEY (tier_id, PERIOD valid_at) REFERENCES \"public\".\"tier\" (id, PERIOD valid_at)" + ) end def down do execute("ALTER TABLE \"subscription\" DROP CONSTRAINT subscription_tier_id_fkey") + drop(table(:tier)) + + alter table(:subscription) do + remove(:recorded_at) + end + drop(table(:subscription)) - drop(table(:tier)) + execute("ALTER TABLE \"member\" DROP CONSTRAINT IF EXISTS \"member_unique_email_index\"") + + drop(table(:member)) drop(table(:event)) end diff --git a/test_priv/test_repo/migrations/20260926192338_migrate_resources76.exs b/test_priv/test_repo/migrations/20260926192338_migrate_resources76.exs deleted file mode 100644 index 3987daf2..00000000 --- a/test_priv/test_repo/migrations/20260926192338_migrate_resources76.exs +++ /dev/null @@ -1,48 +0,0 @@ -# SPDX-FileCopyrightText: 2019 ash_postgres contributors -# -# SPDX-License-Identifier: MIT - -defmodule AshPostgres.TestRepo.Migrations.MigrateResources76 do - @moduledoc """ - Updates resources based on their most recent snapshots. - - This file was autogenerated with `mix ash_postgres.generate_migrations` - """ - - use Ecto.Migration - - def up do - create table(:member, primary_key: false) do - add(:id, :bigint, null: false) - add(:email, :citext) - add(:valid_at, :tstzrange, null: false) - end - - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= - 190_000 do - execute("ALTER TABLE \"member\" ADD PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)") - else - execute("ALTER TABLE \"member\" ADD PRIMARY KEY (id)") - end - - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= - 190_000 do - execute( - "ALTER TABLE \"member\" ADD CONSTRAINT \"member_unique_email_index\" EXCLUDE USING gist ((lower(email::text)) WITH =, valid_at WITH &&)" - ) - else - create(unique_index(:member, [:email], name: "member_unique_email_index")) - end - end - - def down do - if repo().query!("SHOW server_version_num").rows |> hd() |> hd() |> String.to_integer() >= - 190_000 do - execute("ALTER TABLE \"member\" DROP CONSTRAINT IF EXISTS \"member_unique_email_index\"") - else - drop_if_exists(unique_index(:member, [:email], name: "member_unique_email_index")) - end - - drop(table(:member)) - end -end diff --git a/test_priv/test_repo/migrations/20260927024129_migrate_resources77.exs b/test_priv/test_repo/migrations/20260927024129_migrate_resources77.exs deleted file mode 100644 index 06892aae..00000000 --- a/test_priv/test_repo/migrations/20260927024129_migrate_resources77.exs +++ /dev/null @@ -1,28 +0,0 @@ -# SPDX-FileCopyrightText: 2019 ash_postgres contributors -# -# SPDX-License-Identifier: MIT - -defmodule AshPostgres.TestRepo.Migrations.MigrateResources77 do - @moduledoc """ - Updates resources based on their most recent snapshots. - - This file was autogenerated with `mix ash_postgres.generate_migrations` - """ - - use Ecto.Migration - - def up do - alter table(:subscription) do - add(:recorded_at, :utc_datetime_usec, - null: false, - default: fragment("(now() AT TIME ZONE 'utc')") - ) - end - end - - def down do - alter table(:subscription) do - remove(:recorded_at) - end - end -end