feat: Support generated columns in Postgres 18 (#3297)

Closes #3220

## Overview of changes

I've made the following core changes to support this:

- When creating a new publication, if on PG18 and above, always set
`publish_generated_columns = 'stored'`, see [PG
docs](https://www.postgresql.org/docs/current/sql-createpublication.html#SQL-CREATEPUBLICATION-PARAMS-WITH-PUBLISH-GENERATED-COLUMNS)
for details
- If we find an existing publication, whether owned or not, we do not
alter it.
- For people upgrading Electric on a PG18 db, they will have to manually
set the publication parameter or somehow reset the electric
slot/publication/timeline to get this feature. I could add a step to
always enable it if we can, but ideally in a separate PR.
- In the `PublicationManager`, before each update, we check the
publication status which now includes whether generated columns are
replicated
- We cache the result, and we initialise with an optimistic assumption
that generated columns are supported
- In PubMan, we now keep track of which shape handles use generated
columns, and in case the configuration changes we do an O(N) scan over
our tracked shapes and schedule removals for any shape that used
generated columns
- Slow scan is fine, it's just a list traversal that happens _only_ if
the publication configuration is disabled at runtime, which might be the
case for manually configured publications but never for our own owned
publication.
- Any shapes pending creation that require generated columns would also
receive an error reply
- Once the configuration result is cached, shape creations immediately
get an error answer from the publication manager until a check confirms
that generated columns are supported
- The Inspector now loads `supported_features` from Postgres, so at the
validation stage we can determine if we should block generated column
shapes the same way we have until now, or whether we should let them
through and let the PublicationManager determine if they can be created.
- We will now get different errors if you are on e.g. PG17 and request a
shape with generated columns, and if you are on PG18 and above. In the
latter you now receive:
  ```
Publication "{pub_name}" does not publish generated columns. This is a
feature introduced in PostgreSQL 18 and requires setting the publication
parameter 'publish_generated_columns' to 'stored'. Alternatively, you
can exclude them from the shape by explicitly listing which columns to
fetch in the 'columns' query param.
  ```
- Refactored the Inspector ETS table name to `inspector_table` rather
than `relation_table` for more generality as it now holds more Postgres
inspection information - I think it makes sense to reuse the Inspector
for this to keep the infrastructure decoupled, this is purely
information about what Postgres supports as part of a validation step.
- Parse PublicationManager error, which gets turned into a
`SnapshotError`, at the API level to return 400 in case the feature is
supported in the database but not configured.
- Some minor typespec fixes and cleanups of unnecessary info (e.g.
`can_alter_publication?` flag in ConnMan - the pubman will determine it
by itself at runtime)


## Potential alternatives
- Rather than return errors from the publication manager, we can just
clean up shapes when the configuration changes and set the flag
appropriately somewhere so that we continue to reject generated columns
in the same way at the shape creation validation stage
- PRO: simplify handling of cases by introducing a sort of "supported
features" table for the validation to lookup
- PRO: by separating into a separate "supported features" table we could
offload it from the inspector (although it would still need to check
postgres somehow, so I don't like this)
- CON: I think it's important to separate when a feature is not
available and when it is not enabled by choice - by letting the
publication manager reply, we are notifying developers that they _could_
use generated columns if they wanted to if they are using PG18 and
above.

Author: msfstef

packages/sync-service/dev/docker-compose-otel.yml

@@ -3,7 +3,7 @@ name: 'electric_dev_otel'
 
 services:
   postgres:
-    image: postgres:16-alpine
+    image: postgres:18-alpine
     environment:
       POSTGRES_DB: electric
       POSTGRES_USER: postgres

packages/sync-service/dev/docker-compose.yml

@@ -3,7 +3,7 @@ name: 'electric_dev'
 
 services:
   postgres:
-    image: postgres:17-alpine
+    image: postgres:18-alpine
     environment:
       POSTGRES_DB: electric
       POSTGRES_USER: postgres
@@ -22,7 +22,7 @@ services:
       - -c
       - config_file=/etc/postgresql.conf
   postgres2:
-    image: postgres:17-alpine
+    image: postgres:18-alpine
     environment:
       POSTGRES_DB: electric
       POSTGRES_USER: postgres

packages/sync-service/lib/electric/connection/manager.ex

@@ -103,9 +103,6 @@ defmodule Electric.Connection.Manager do
       :pg_system_identifier,
       # PostgreSQL timeline ID
       :pg_timeline_id,
-      # Capability flag that is set during replication client initialization and shows whether
-      # the PG role has the necessary privilege to alter the PG publication.
-      :can_alter_publication?,
       # User setting that determines whether the table publishing is to be automatically
       # managed by the stack or whether it's the user's responsibility.
       :manual_table_publishing?,
@@ -225,10 +222,6 @@ defmodule Electric.Connection.Manager do
     GenServer.cast(manager, :replication_client_created_new_slot)
   end
 
-  def replication_client_has_insufficient_privilege(manager) do
-    GenServer.cast(manager, :replication_client_has_insufficient_privilege)
-  end
-
   def replication_client_ready_to_stream(manager) do
     GenServer.cast(manager, :replication_client_ready_to_stream)
   end
@@ -315,7 +308,6 @@ defmodule Electric.Connection.Manager do
         stack_events_registry: Keyword.fetch!(opts, :stack_events_registry),
         tweaks: Keyword.fetch!(opts, :tweaks),
         persistent_kv: Keyword.fetch!(opts, :persistent_kv),
-        can_alter_publication?: true,
         manual_table_publishing?: Keyword.get(opts, :manual_table_publishing?, false),
         max_shapes: Keyword.fetch!(opts, :max_shapes),
         expiry_batch_size: Keyword.fetch!(opts, :expiry_batch_size)
@@ -534,7 +526,6 @@ defmodule Electric.Connection.Manager do
       pool_opts: state.pool_opts,
       replication_opts: state.replication_opts,
       tweaks: state.tweaks,
-      can_alter_publication?: state.can_alter_publication?,
       manual_table_publishing?: state.manual_table_publishing?,
       persistent_kv: state.persistent_kv,
       max_shapes: state.max_shapes,
@@ -807,16 +798,6 @@ defmodule Electric.Connection.Manager do
     {:noreply, %{state | purge_all_shapes?: true}}
   end
 
-  def handle_cast(
-        :replication_client_has_insufficient_privilege,
-        %State{
-          current_phase: :connection_setup,
-          current_step: {:start_replication_client, :configuring_connection}
-        } = state
-      ) do
-    {:noreply, %{state | can_alter_publication?: false}}
-  end
-
   def handle_cast(
         :replication_client_ready_to_stream,
         %State{

packages/sync-service/lib/electric/connection/manager/supervisor.ex

@@ -59,7 +59,6 @@ defmodule Electric.Connection.Manager.Supervisor do
       {Electric.Replication.PublicationManager,
        stack_id: stack_id,
        publication_name: Keyword.fetch!(replication_opts, :publication_name),
-       can_alter_publication?: Keyword.fetch!(opts, :can_alter_publication?),
        manual_table_publishing?: Keyword.fetch!(opts, :manual_table_publishing?),
        db_pool: Electric.Connection.Manager.admin_pool(stack_id),
        update_debounce_timeout: Keyword.get(tweaks, :publication_alter_debounce_ms, 0),

packages/sync-service/lib/electric/db_configuration_error.ex

@@ -18,6 +18,16 @@ defmodule Electric.DbConfigurationError do
     }
   end
 
+  def publication_missing_generated_columns(pub_name) do
+    %Electric.DbConfigurationError{
+      type: :publication_missing_generated_columns,
+      message:
+        "Publication #{Utils.quote_name(pub_name)} does not publish generated columns." <>
+          " This is a feature introduced in PostgreSQL 18 and requires setting the publication parameter 'publish_generated_columns' to 'stored'." <>
+          " Alternatively, you can exclude them from the shape by explicitly listing which columns to fetch in the 'columns' query param."
+    }
+  end
+
   def publication_not_owned(pub_name) do
     %Electric.DbConfigurationError{
       type: :publication_not_owned,

packages/sync-service/lib/electric/postgres/configuration.ex

@@ -21,6 +21,12 @@ defmodule Electric.Postgres.Configuration do
                | :misconfigured_replica_identity}
         }
 
+  @type publication_status :: %{
+          can_alter_publication?: boolean(),
+          publishes_all_operations?: boolean(),
+          publishes_generated_columns?: boolean()
+        }
+
   @typep changed_relation ::
            {
              Electric.relation_id(),
@@ -32,41 +38,44 @@ defmodule Electric.Postgres.Configuration do
   @typep publication_relation :: {Electric.relation_id(), Electric.relation(), <<_::8>>}
 
   @doc """
-  Check whether the publication with the given name exists, is configured to
-  publish all required operations, and is owned by the current user.
+  Check whether the publication with the given name exists, and return its status.
+
+  The status includes whether the publication is owned, whether it publishes all operations,
+  and whether it publishes generated columns (if supported by the Postgres version).
   """
-  @spec check_publication_status!(Postgrex.conn(), String.t()) :: :ok
+  @spec check_publication_status!(Postgrex.conn(), String.t()) ::
+          publication_status() | :not_found
   def check_publication_status!(conn, publication_name) do
+    # note: we could do a separate query to get the PG version, unsure which is more efficient
+    # but avoiding having to know the PG version in advance keeps things contained and simple
     query =
       """
       SELECT
-        pubinsert AND pubupdate AND pubdelete AND pubtruncate as all_operations,
-        pg_get_userbyid(p.pubowner) = current_role as is_owned
-      FROM pg_publication as p WHERE pubname = $1
+        pg_get_userbyid(p.pubowner) = current_role as can_alter_publication,
+        pubinsert AND pubupdate AND pubdelete AND pubtruncate as publishes_all_operations,
+        CASE WHEN current_setting('server_version_num')::int >= 180000
+            THEN (to_jsonb(p) ->> 'pubgencols') = 's'
+            ELSE FALSE
+        END AS publishes_generated_columns
+      FROM pg_publication as p WHERE pubname = $1;
       """
 
     Postgrex.query!(conn, query, [publication_name])
-    |> one_or_nil()
     |> case do
-      %{"all_operations" => true, "is_owned" => true} ->
-        :ok
-
-      %{"is_owned" => false} ->
-        raise Electric.DbConfigurationError.publication_not_owned(publication_name)
-
-      %{"all_operations" => false} ->
-        raise Electric.DbConfigurationError.publication_missing_operations(publication_name)
+      %Postgrex.Result{
+        rows: [[can_alter_publication, publishes_all_operations, publishes_generated_columns]]
+      } ->
+        %{
+          can_alter_publication?: can_alter_publication,
+          publishes_all_operations?: publishes_all_operations,
+          publishes_generated_columns?: publishes_generated_columns
+        }
 
-      _ ->
-        raise Electric.DbConfigurationError.publication_missing(publication_name)
+      %Postgrex.Result{num_rows: 0} ->
+        :not_found
     end
   end
 
-  defp one_or_nil(%Postgrex.Result{rows: [row], columns: cols}),
-    do: Enum.zip(cols, row) |> Map.new()
-
-  defp one_or_nil(_), do: nil
-
   @doc """
   Check whether the state of the publication relations in the database matches the sets of
   filters passed into the function.

packages/sync-service/lib/electric/postgres/inspector.ex

@@ -28,6 +28,10 @@ defmodule Electric.Postgres.Inspector do
           children: nil | [relation(), ...]
         }
 
+  @type supported_features :: %{
+          supports_generated_column_replication: boolean()
+        }
+
   @callback load_relation_oid(relation(), opts :: term()) ::
               {:ok, Electric.oid_relation()}
               | :table_not_found
@@ -43,6 +47,10 @@ defmodule Electric.Postgres.Inspector do
               | :table_not_found
               | {:error, String.t() | :connection_not_available}
 
+  @callback load_supported_features(opts :: term()) ::
+              {:ok, supported_features()}
+              | {:error, String.t() | :connection_not_available}
+
   @callback clean(relation_id(), opts :: term()) :: :ok
 
   @callback list_relations_with_stale_cache(opts :: term()) ::
@@ -91,6 +99,15 @@ defmodule Electric.Postgres.Inspector do
     module.load_column_info(relation_id, opts)
   end
 
+  @doc """
+  Load the supported features on the target database using a provided inspector.
+  """
+  @spec load_supported_features(inspector()) ::
+          {:ok, supported_features()} | {:error, String.t() | :connection_not_available}
+  def load_supported_features({module, opts}) do
+    module.load_supported_features(opts)
+  end
+
   @doc """
   Clean up all information about a given relation using a provided inspector.
   """

packages/sync-service/lib/electric/postgres/inspector/direct_inspector.ex

@@ -230,6 +230,17 @@ defmodule Electric.Postgres.Inspector.DirectInspector do
   defp parse_type_kind("r"), do: :range
   defp parse_type_kind("m"), do: :multirange
 
+  @impl Electric.Postgres.Inspector
+  def load_supported_features(conn) do
+    with {:ok, %{rows: [[pg_version]]}} <-
+           Postgrex.query(conn, "SELECT current_setting('server_version_num')::int", []) do
+      {:ok,
+       %{
+         supports_generated_column_replication: pg_version >= 180_000
+       }}
+    end
+  end
+
   @impl Electric.Postgres.Inspector
   def clean(_, _), do: :ok
 end