fix: disallow generated columns in shapes (#2507)

Closes #2448 

Generated columns cannot be included in a publication by PostgreSQL, so
we disallow them in shapes. When doing `CREATE PUBLICATION p FOR TABLE
t` without a column list, generated columns are omitted. When doing an
explicit `FOR TABLE t (id)` (if `id` is generated), then it says `ERROR:
cannot use generated column "id" in publication column list`.

We could, later, allow generated columns based on the subset of where
clause functions we support by filling the stored values on our side
too, but that's a way off.

Author: icehaunter

.changeset/forty-buttons-accept.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+fix: disallow generated columns in shapes

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

@@ -88,6 +88,7 @@ defmodule Electric.Postgres.Inspector.DirectInspector do
       attndims as array_dimensions,
       atttypmod as type_mod,
       attnotnull as not_null,
+      attgenerated != '' as is_generated,
       pg_type.typname as type,
       pg_type.typtype as type_kind, -- e.g. an enum is kind 'e'
       elem_pg_type.typname as array_type, -- type of the element inside the array or nil if it's not an array

packages/sync-service/lib/electric/shapes/shape.ex

@@ -174,14 +174,19 @@ defmodule Electric.Shapes.Shape do
           [String.t(), ...] | nil
         ) ::
           {:ok, [String.t(), ...] | nil} | {:error, {:columns, [String.t()]}}
-  defp validate_selected_columns(column_info, _pk_cols, nil) do
-    {:ok, Enum.map(column_info, & &1.name)}
+  defp validate_selected_columns(column_info, pk_cols, nil) do
+    validate_selected_columns(column_info, pk_cols, Enum.map(column_info, & &1.name))
   end
 
   defp validate_selected_columns(column_info, pk_cols, columns_to_select) do
     missing_pk_cols = pk_cols -- columns_to_select
     invalid_cols = columns_to_select -- Enum.map(column_info, & &1.name)
 
+    generated_cols =
+      column_info
+      |> Enum.filter(&(&1.is_generated and &1.name in columns_to_select))
+      |> Enum.map(& &1.name)
+
     cond do
       missing_pk_cols != [] ->
         {:error,
@@ -197,6 +202,13 @@ defmodule Electric.Shapes.Shape do
             "The following columns could not be found: #{invalid_cols |> Enum.join(", ")}"
           ]}}
 
+      generated_cols != [] ->
+        {:error,
+         {:columns,
+          [
+            "The following columns are generated and cannot be included in replication: #{generated_cols |> Enum.join(", ")}"
+          ]}}
+
       true ->
         {:ok, Enum.sort(columns_to_select)}
     end

packages/sync-service/test/electric/plug/delete_shape_plug_test.exs

@@ -26,7 +26,8 @@ defmodule Electric.Plug.DeleteShapePlugTest do
   @test_pg_id "12345"
 
   def load_column_info({"public", "users"}, _),
-    do: {:ok, [%{name: "id", type: "int8", pk_position: 0, type_id: {20, -1}}]}
+    do:
+      {:ok, [%{name: "id", type: "int8", pk_position: 0, type_id: {20, -1}, is_generated: false}]}
 
   def load_relation(tbl, _),
     do: Support.StubInspector.load_relation(tbl, nil)

packages/sync-service/test/electric/plug/router_test.exs

@@ -621,6 +621,56 @@ defmodule Electric.Plug.RouterTest do
              ] = Jason.decode!(conn.resp_body)
     end
 
+    @generated_pk_table "generated_pk_table"
+    @tag with_sql: [
+           "CREATE TABLE #{@generated_pk_table} (val JSONB NOT NULL, id uuid PRIMARY KEY GENERATED ALWAYS AS ((val->>'id')::uuid) STORED)"
+         ]
+    test "returns an error when trying to select a generated column", %{opts: opts} do
+      # When selecting all columns
+      conn = conn("GET", "/v1/shape?table=#{@generated_pk_table}&offset=-1") |> Router.call(opts)
+      assert %{status: 400} = conn
+
+      assert Jason.decode!(conn.resp_body) ==
+               %{
+                 "errors" => %{
+                   "columns" => [
+                     "The following columns are generated and cannot be included in replication: id"
+                   ]
+                 },
+                 "message" => "Invalid request"
+               }
+
+      # When selecting a single column but PK is generated
+      conn =
+        conn("GET", "/v1/shape?table=#{@generated_pk_table}&offset=-1&columns=val")
+        |> Router.call(opts)
+
+      assert %{status: 400} = conn
+
+      assert Jason.decode!(conn.resp_body) ==
+               %{
+                 "errors" => %{"columns" => ["Must include all primary key columns, missing: id"]},
+                 "message" => "Invalid request"
+               }
+
+      # When selecting a generated column explicitly
+      conn =
+        conn("GET", "/v1/shape?table=#{@generated_pk_table}&offset=-1&columns=id,val")
+        |> Router.call(opts)
+
+      assert %{status: 400} = conn
+
+      assert Jason.decode!(conn.resp_body) ==
+               %{
+                 "errors" => %{
+                   "columns" => [
+                     "The following columns are generated and cannot be included in replication: id"
+                   ]
+                 },
+                 "message" => "Invalid request"
+               }
+    end
+
     @tag with_sql: [
            "CREATE TABLE wide_table (id BIGINT PRIMARY KEY, value1 TEXT NOT NULL, value2 TEXT NOT NULL, value3 TEXT NOT NULL)",
            "INSERT INTO wide_table VALUES (1, 'test value 1', 'test value 1', 'test value 1')"

packages/sync-service/test/electric/plug/serve_shape_plug_test.exs

@@ -43,8 +43,22 @@ defmodule Electric.Plug.ServeShapePlugTest do
     do:
       {:ok,
        [
-         %{name: "id", type: "int8", type_id: {20, 1}, pk_position: 0, array_dimensions: 0},
-         %{name: "value", type: "text", type_id: {28, 1}, pk_position: nil, array_dimensions: 0}
+         %{
+           name: "id",
+           type: "int8",
+           type_id: {20, 1},
+           pk_position: 0,
+           array_dimensions: 0,
+           is_generated: false
+         },
+         %{
+           name: "value",
+           type: "text",
+           type_id: {28, 1},
+           pk_position: nil,
+           array_dimensions: 0,
+           is_generated: false
+         }
        ]}
 
   def load_column_info(_, _),

packages/sync-service/test/electric/replication/shape_log_collector_test.exs

@@ -117,7 +117,7 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
           {:ok, %{id: 1234, schema: "public", name: "test_table", parent: nil, children: nil}}
       end)
       |> stub(:load_column_info, fn {"public", "test_table"}, _ ->
-        {:ok, [%{pk_position: 0, name: "id"}]}
+        {:ok, [%{pk_position: 0, name: "id", is_generated: false}]}
       end)
       |> allow(self(), ctx.server)
 
@@ -158,7 +158,7 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
           {:ok, %{id: 1234, schema: "public", name: "test_table", parent: nil, children: nil}}
       end)
       |> stub(:load_column_info, fn {"public", "test_table"}, _ ->
-        {:ok, [%{pk_position: 0, name: "id"}]}
+        {:ok, [%{pk_position: 0, name: "id", is_generated: false}]}
       end)
       |> allow(self(), ctx.server)
 
@@ -295,7 +295,7 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
       {:ok, %{id: 1234, schema: "public", name: "test_table", parent: nil, children: nil}}
     end)
     |> stub(:load_column_info, fn {"public", "test_table"}, _ ->
-      {:ok, [%{pk_position: 0, name: "id"}]}
+      {:ok, [%{pk_position: 0, name: "id", is_generated: false}]}
     end)
     |> allow(self(), pid)
 

packages/sync-service/test/electric/shape_cache/shape_status_test.exs

@@ -205,12 +205,12 @@ defmodule Electric.ShapeCache.ShapeStatusTest do
     do:
       {:ok,
        [
-         %{name: "id", type: :int8, type_id: {1, 1}, pk_position: 0},
-         %{name: "value", type: :text, type_id: {2, 2}, pk_position: nil}
+         %{name: "id", type: :int8, type_id: {1, 1}, pk_position: 0, is_generated: false},
+         %{name: "value", type: :text, type_id: {2, 2}, pk_position: nil, is_generated: false}
        ]}
 
   def load_column_info({"public", "table"}, _),
-    do: {:ok, [%{name: "id", type: :int8, pk_position: 0}]}
+    do: {:ok, [%{name: "id", type: :int8, type_id: {1, 1}, pk_position: 0, is_generated: false}]}
 
   def load_relation(tbl, _),
     do: StubInspector.load_relation(tbl, nil)

packages/sync-service/test/electric/shape_cache_test.exs

@@ -44,8 +44,14 @@ defmodule Electric.ShapeCacheTest do
   @zero_offset LogOffset.last_before_real_offsets()
 
   @stub_inspector StubInspector.new([
-                    %{name: "id", type: "int8", type_id: {20, 1}, pk_position: 0},
-                    %{name: "value", type: "text", type_id: {25, 1}}
+                    %{
+                      name: "id",
+                      type: "int8",
+                      type_id: {20, 1},
+                      pk_position: 0,
+                      is_generated: false
+                    },
+                    %{name: "value", type: "text", type_id: {25, 1}, is_generated: false}
                   ])
 
   # {xmin, xmax, xip_list}

packages/sync-service/test/electric/shapes/api_test.exs

@@ -35,8 +35,22 @@ defmodule Electric.Shapes.ApiTest do
   def load_column_info({"public", "users"}, _) do
     {:ok,
      [
-       %{name: "id", type: "int8", type_id: {20, 1}, pk_position: 0, array_dimensions: 0},
-       %{name: "value", type: "text", type_id: {28, 1}, pk_position: nil, array_dimensions: 0}
+       %{
+         name: "id",
+         type: "int8",
+         type_id: {20, 1},
+         pk_position: 0,
+         array_dimensions: 0,
+         is_generated: false
+       },
+       %{
+         name: "value",
+         type: "text",
+         type_id: {28, 1},
+         pk_position: nil,
+         array_dimensions: 0,
+         is_generated: false
+       }
      ]}
   end