fix: split temporal periods without FOR PORTION OF, on PostgreSQL 18

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,16 +76,32 @@ 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"
 
   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,

documentation/topics/advanced/temporal-resources.md

@@ -0,0 +1,85 @@
+<!--
+SPDX-FileCopyrightText: 2020 Zach Daniel
+
+SPDX-License-Identifier: MIT
+-->
+
+# 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.

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(

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