feat: speed up poller by not re-encoding user data (#1605)

record, old_record and columns are relayed without decoding and encoding
their jsonb representation

Author: edgurgel

lib/extensions/postgres_cdc_rls/replication_poller.ex

@@ -186,7 +186,7 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
   defp handle_list_changes_result(
          {:ok,
           %Postgrex.Result{
-            columns: ["wal", "is_rls_enabled", "subscription_ids", "errors"] = columns,
+            columns: columns,
             rows: [_ | _] = rows,
             num_rows: rows_count
           }},
@@ -260,75 +260,75 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
   end
 
   def generate_record([
-        {"wal",
-         %{
-           "type" => "INSERT" = type,
-           "schema" => schema,
-           "table" => table
-         } = wal},
-        {"is_rls_enabled", _},
+        {"type", "INSERT" = type},
+        {"schema", schema},
+        {"table", table},
+        {"columns", columns},
+        {"record", record},
+        {"old_record", _},
+        {"commit_timestamp", commit_timestamp},
         {"subscription_ids", subscription_ids},
         {"errors", errors}
       ])
       when is_list(subscription_ids) do
     %NewRecord{
-      columns: Map.get(wal, "columns", []),
-      commit_timestamp: Map.get(wal, "commit_timestamp"),
+      columns: Jason.Fragment.new(columns),
+      commit_timestamp: commit_timestamp,
       errors: convert_errors(errors),
       schema: schema,
       table: table,
       type: type,
       subscription_ids: MapSet.new(subscription_ids),
-      record: Map.get(wal, "record", %{})
+      record: Jason.Fragment.new(record)
     }
   end
 
   def generate_record([
-        {"wal",
-         %{
-           "type" => "UPDATE" = type,
-           "schema" => schema,
-           "table" => table
-         } = wal},
-        {"is_rls_enabled", _},
+        {"type", "UPDATE" = type},
+        {"schema", schema},
+        {"table", table},
+        {"columns", columns},
+        {"record", record},
+        {"old_record", old_record},
+        {"commit_timestamp", commit_timestamp},
         {"subscription_ids", subscription_ids},
         {"errors", errors}
       ])
       when is_list(subscription_ids) do
     %UpdatedRecord{
-      columns: Map.get(wal, "columns", []),
-      commit_timestamp: Map.get(wal, "commit_timestamp"),
+      columns: Jason.Fragment.new(columns),
+      commit_timestamp: commit_timestamp,
       errors: convert_errors(errors),
       schema: schema,
       table: table,
       type: type,
       subscription_ids: MapSet.new(subscription_ids),
-      old_record: Map.get(wal, "old_record", %{}),
-      record: Map.get(wal, "record", %{})
+      old_record: Jason.Fragment.new(old_record),
+      record: Jason.Fragment.new(record)
     }
   end
 
   def generate_record([
-        {"wal",
-         %{
-           "type" => "DELETE" = type,
-           "schema" => schema,
-           "table" => table
-         } = wal},
-        {"is_rls_enabled", _},
+        {"type", "DELETE" = type},
+        {"schema", schema},
+        {"table", table},
+        {"columns", columns},
+        {"record", _},
+        {"old_record", old_record},
+        {"commit_timestamp", commit_timestamp},
         {"subscription_ids", subscription_ids},
         {"errors", errors}
       ])
       when is_list(subscription_ids) do
     %DeletedRecord{
-      columns: Map.get(wal, "columns", []),
-      commit_timestamp: Map.get(wal, "commit_timestamp"),
+      columns: Jason.Fragment.new(columns),
+      commit_timestamp: commit_timestamp,
       errors: convert_errors(errors),
       schema: schema,
       table: table,
       type: type,
       subscription_ids: MapSet.new(subscription_ids),
-      old_record: Map.get(wal, "old_record", %{})
+      old_record: Jason.Fragment.new(old_record)
     }
   end
 

lib/extensions/postgres_cdc_rls/replications.ex

@@ -72,7 +72,18 @@ defmodule Extensions.PostgresCdcRls.Replications do
   def list_changes(conn, slot_name, publication, max_changes, max_record_bytes) do
     query(
       conn,
-      "select * from realtime.list_changes($1, $2, $3, $4)",
+      """
+      SELECT wal->>'type' as type,
+             wal->>'schema' as schema,
+             wal->>'table' as table,
+             wal->>'columns' as columns,
+             wal->>'record' as record,
+             wal->>'old_record' as old_record,
+             wal->>'commit_timestamp' as commit_timestamp,
+             subscription_ids,
+             errors
+      FROM realtime.list_changes($1, $2, $3, $4)
+      """,
       [
         publication,
         slot_name,

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.59.1",
+      version: "2.60.0",
       elixir: "~> 1.18",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,