feat: add old values to replica=full (#2381)

Co-authored-by: Sam Willis <sam.willis@gmail.com>

Author: icehaunter

.changeset/gorgeous-rabbits-hope.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+feat: add `old_value` to updates under replica mode `full`

.changeset/twenty-sheep-swim.md

@@ -0,0 +1,5 @@
+---
+"@electric-sql/client": patch
+---
+
+feat: add `old_value` to updates under replica mode `full`

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

@@ -65,15 +65,15 @@ defmodule Electric.LogItems do
       {change.log_offset,
        %{
          key: change.key,
-         value: update_values(change, pk_cols, replica),
          headers: %{
            operation: :update,
            txids: List.wrap(txids),
            relation: Tuple.to_list(change.relation),
            lsn: change.log_offset.tx_offset,
            op_position: change.log_offset.op_offset
          }
-       }}
+       }
+       |> Map.merge(put_update_values(change, pk_cols, replica))}
     ]
   end
 
@@ -114,12 +114,26 @@ defmodule Electric.LogItems do
   defp take_pks_or_all(record, [], :default), do: record
   defp take_pks_or_all(record, pks, :default), do: Map.take(record, pks)
 
-  defp update_values(%{record: record, changed_columns: changed_columns}, pk_cols, :default) do
-    Map.take(record, Enum.concat(pk_cols, changed_columns))
+  defp put_update_values(%{record: record, changed_columns: changed_columns}, pk_cols, :default) do
+    %{value: Map.take(record, Enum.concat(pk_cols, changed_columns))}
+  end
+
+  defp put_update_values(
+         %{record: record, old_record: old_record, changed_columns: changed_columns},
+         _pk_cols,
+         :full
+       ) do
+    %{value: record, old_value: Map.take(old_record, MapSet.to_list(changed_columns))}
   end
 
-  defp update_values(%{record: record}, _pk_cols, :full) do
-    record
+  def merge_updates(u1, u2) when is_map_key(u1, "old_value") or is_map_key(u2, "old_value") do
+    %{
+      "key" => u1["key"],
+      "headers" => Map.take(u1["headers"], ["operation", "relation"]),
+      "value" => Map.merge(u1["value"], u2["value"]),
+      # When merging old values, we give preference to the older u1
+      "old_value" => Map.merge(u2["old_value"] || %{}, u1["old_value"] || %{})
+    }
   end
 
   def merge_updates(u1, u2) do

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

@@ -139,6 +139,7 @@ defmodule Electric.LogItemsTest do
                  {LogOffset.first(),
                   %{
                     value: %{"pk" => "10", "hello" => "world", "test" => "new"},
+                    old_value: %{"test" => "me"},
                     key: "my_key",
                     headers: %{
                       relation: ["public", "test"],

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

@@ -618,6 +618,57 @@ defmodule Electric.Plug.RouterTest do
       assert [_] = Jason.decode!(conn.resp_body)
     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')"
+         ]
+    test "GET receives old rows in updates when replica=full", %{opts: opts, db_conn: db_conn} do
+      conn =
+        conn("GET", "/v1/shape?table=wide_table&offset=-1&replica=full") |> Router.call(opts)
+
+      assert %{status: 200} = conn
+      shape_handle = get_resp_shape_handle(conn)
+      json_body = Jason.decode!(conn.resp_body)
+
+      assert [
+               %{
+                 "value" => %{"id" => _, "value1" => _, "value2" => _, "value3" => _},
+                 "key" => key
+               }
+             ] = json_body
+
+      # Old value cannot be present on the snapshot
+      refute match?(%{"old_value" => _}, json_body)
+
+      task =
+        Task.async(fn ->
+          conn(
+            "GET",
+            "/v1/shape?table=wide_table&offset=0_0&handle=#{shape_handle}&live&replica=full"
+          )
+          |> Router.call(opts)
+        end)
+
+      Postgrex.query!(db_conn, "UPDATE wide_table SET value2 = 'test value 2' WHERE id = 1", [])
+
+      assert %{status: 200} = conn = Task.await(task)
+
+      # No extra keys should be present, so this is a pin
+      value = %{
+        "id" => "1",
+        "value2" => "test value 2",
+        "value1" => "test value 1",
+        "value3" => "test value 1"
+      }
+
+      old_value = %{
+        "value2" => "test value 1"
+      }
+
+      assert [%{"key" => ^key, "value" => ^value, "old_value" => ^old_value}, @up_to_date] =
+               Jason.decode!(conn.resp_body)
+    end
+
     @tag additional_fields: "num INTEGER NOT NULL"
     @tag with_sql: [
            "INSERT INTO serial_ids (id, num) VALUES (1, 1), (2, 10)"

packages/sync-service/test/electric/shape_cache/file_storage/compaction_test.exs

@@ -78,6 +78,51 @@ defmodule Electric.ShapeCache.FileStorage.CompactionTest do
              |> length == 4
     end
 
+    test "compacts a log file with replica mode full", %{tmp_dir: tmp_dir} do
+      log_file_path = Path.join(tmp_dir, "log_file")
+
+      log_stream =
+        [
+          ins(offset: {1, 1}, rec: [id: "key1", value: "value1"]),
+          upd(offset: {2, 1}, rec: [id: "key1", value: {"value1", "value2"}]),
+          ins(offset: {3, 1}, rec: [id: "key2", value: "value3"]),
+          upd(offset: {4, 1}, rec: [id: "key1", value: {"value2", "value new 1"}]),
+          upd(offset: {5, 1}, rec: [id: "key1", value: {"value new 1", "value new 2"}]),
+          upd(offset: {6, 1}, rec: [id: "key1", value: {"value new 2", "value new 3"}]),
+          upd(offset: {7, 1}, rec: [id: "key1", value: {"value new 3", "value new 4"}]),
+          upd(offset: {8, 1}, rec: [id: "key1", value: {"value new 4", "value new 5"}]),
+          del(offset: {9, 1}, rec: [id: "key2", value: "value"])
+        ]
+        |> TestUtils.changes_to_log_items(replica: :full)
+
+      paths = LogFile.write_log_file(log_stream, log_file_path)
+
+      assert LogFile.read_chunk(paths, %LogOffset{tx_offset: 0, op_offset: 0})
+             |> Enum.to_list()
+             |> length == 9
+
+      assert {log_file_path, chunk_index_path, key_index_path} =
+               Compaction.compact_in_place(paths, 1_000_000)
+
+      assert File.exists?(log_file_path)
+      assert File.exists?(chunk_index_path)
+      assert File.exists?(key_index_path)
+
+      assert [
+               %{"headers" => %{"operation" => "insert"}},
+               %{"headers" => %{"operation" => "insert"}},
+               %{
+                 "headers" => %{"operation" => "update"},
+                 "value" => %{"id" => "key1", "value" => "value new 5"},
+                 "old_value" => %{"value" => "value1"}
+               },
+               %{"headers" => %{"operation" => "delete"}}
+             ] =
+               LogFile.read_chunk(paths, %LogOffset{tx_offset: 0, op_offset: 0})
+               |> Enum.to_list()
+               |> Enum.map(&Jason.decode!/1)
+    end
+
     test "compacts a large enough log file full of updates (failing property)", %{
       tmp_dir: tmp_dir
     } do

packages/typescript-client/src/client.ts

@@ -69,7 +69,8 @@ export interface PostgresParams {
    * changed columns in an update.
    *
    * If it's `full` Electric will send the entire row with both changed and
-   * unchanged values.
+   * unchanged values. `old_value` will also be present on update messages,
+   * containing the previous value for changed columns.
    *
    * Setting `replica` to `full` will result in higher bandwidth
    * usage and so is not generally recommended.

packages/typescript-client/src/parser.ts

@@ -104,7 +104,12 @@ export class MessageParser<T extends Row<unknown>> {
       // is needed because there could be a column named `value`
       // and the value associated to that column will be a string or null.
       // But `typeof null === 'object'` so we need to make an explicit check.
-      if (key === `value` && typeof value === `object` && value !== null) {
+      // We also parse the `old_value`, which appears on updates when `replica=full`.
+      if (
+        (key === `value` || key === `old_value`) &&
+        typeof value === `object` &&
+        value !== null
+      ) {
         // Parse the row values
         const row = value as Record<string, Value<GetExtensions<T>>>
         Object.keys(row).forEach((key) => {

packages/typescript-client/src/types.ts

@@ -32,6 +32,7 @@ export type ControlMessage = {
 export type ChangeMessage<T extends Row<unknown> = Row> = {
   key: string
   value: T
+  old_value?: Partial<T> // Only provided for updates if `replica` is `full`
   headers: Header & { operation: Operation }
 }