diff --git a/components/electric/lib/electric/replication/postgres/client.ex b/components/electric/lib/electric/replication/postgres/client.ex index ee6fd1570f..fe482f7ff8 100644 --- a/components/electric/lib/electric/replication/postgres/client.ex +++ b/components/electric/lib/electric/replication/postgres/client.ex @@ -165,6 +165,34 @@ defmodule Electric.Replication.Postgres.Client do |> :epgsql.start_replication(:erlang.binary_to_list(slot), handler, [], ~c"0/0", opts) end + @doc """ + Explicitly set those configuration parameters that affect formatting of values of certain types. + + By setting those parameters for the current session we're safe-guarding against non-standard configuration being used + in the Postgres database cluster or even the specific database Electric is configured to replicate from. + + The parameters we're interested in are: + + * `bytea_output` - determines how Postgres encodes bytea values. It can use either Hex- or Escape-based encoding. + * `DateStyle` - determines how Postgres interprets date values. + * `TimeZone` - affects the time zone offset Postgres uses for timestamptz and timetz values. + * `extra_float_digits` - determines whether floating-point values are rounded or are encoded precisely. + """ + def set_display_settings_for_replication(conn) do + results = + :epgsql.squery( + conn, + """ + SET bytea_output = 'hex'; + SET DateStyle = 'ISO, DMY'; + SET TimeZone = 'UTC'; + SET extra_float_digits = 1; + """ + ) + + :ok = Enum.each(results, &({:ok, [], []} = &1)) + end + @doc """ Confirm successful processing of a WAL segment. diff --git a/components/electric/lib/electric/replication/postgres/logical_replication_producer.ex b/components/electric/lib/electric/replication/postgres/logical_replication_producer.ex index 840c2a88bf..075e474d15 100644 --- a/components/electric/lib/electric/replication/postgres/logical_replication_producer.ex +++ b/components/electric/lib/electric/replication/postgres/logical_replication_producer.ex @@ -88,6 +88,7 @@ defmodule Electric.Replication.Postgres.LogicalReplicationProducer do with {:ok, conn} <- Client.connect(conn_opts), {:ok, _} <- Client.create_slot(conn, slot), + :ok <- Client.set_display_settings_for_replication(conn), :ok <- Client.start_replication(conn, publication, slot, self()) do Logger.metadata(pg_producer: origin) Logger.info("Starting replication from #{origin}") diff --git a/components/electric/test/electric/replication/postgres/logical_replication_producer_test.exs b/components/electric/test/electric/replication/postgres/logical_replication_producer_test.exs index a9e6903086..4efbcf04cf 100644 --- a/components/electric/test/electric/replication/postgres/logical_replication_producer_test.exs +++ b/components/electric/test/electric/replication/postgres/logical_replication_producer_test.exs @@ -16,7 +16,8 @@ defmodule Electric.Replication.Postgres.LogicalReplicationProducerTest do [ connect: fn _ -> {:ok, :conn} end, start_replication: fn :conn, _, _, _ -> :ok end, - create_slot: fn :conn, name -> {:ok, name} end + create_slot: fn :conn, name -> {:ok, name} end, + set_display_settings_for_replication: fn _ -> :ok end ]}, {Connectors, [:passthrough], [