Packages
electric
1.3.4
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.10
1.6.9
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.16
1.4.16-beta-1
1.4.15
1.4.14
1.4.13
1.4.12
1.4.11
1.4.10
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.4
1.3.3
1.3.2
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.14
1.1.13
1.1.12
1.1.11
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
retired
1.1.5
retired
1.1.4
retired
1.1.3
retired
1.1.2
1.1.1
1.1.0
1.0.24
1.0.23
1.0.22
1.0.21
1.0.20
1.0.19
1.0.18
1.0.17
1.0.15
1.0.13
1.0.12
1.0.11
1.0.10
1.0.9
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-beta.23
1.0.0-beta.22
1.0.0-beta.20
1.0.0-beta.19
1.0.0-beta.18
1.0.0-beta.17
1.0.0-beta.16
1.0.0-beta.15
1.0.0-beta.14
1.0.0-beta.13
1.0.0-beta.12
1.0.0-beta.11
1.0.0-beta.10
1.0.0-beta.9
1.0.0-beta.8
1.0.0-beta.7
1.0.0-beta.6
1.0.0-beta.5
1.0.0-beta.4
1.0.0-beta.3
1.0.0-beta.2
1.0.0-beta.1
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.5.2
0.4.4
Postgres sync engine. Sync little subsets of your Postgres data into local apps and services.
Current section
Files
Jump to
Current section
Files
lib/electric/shapes/partial_modes.ex
defmodule Electric.Shapes.PartialModes do
alias Electric.Shapes.Shape
alias Electric.Postgres.Lsn
alias Electric.Shapes.Querying
alias Electric.Connection.Manager
alias Electric.Postgres.SnapshotQuery
def query_subset(shape_handle, %Shape{} = shape, subset, opts) do
pool = Manager.pool_name(opts[:stack_id], :snapshot)
mark = Enum.random(0..(2 ** 31 - 1))
headers = %{snapshot_mark: mark}
SnapshotQuery.execute_for_shape(pool, shape_handle, shape,
snapshot_info_fn: fn _, pg_snapshot, lsn ->
send(self(), {:pg_snapshot_info, pg_snapshot, lsn})
end,
query_fn: fn conn, _, _ ->
Querying.query_subset(conn, opts[:stack_id], shape_handle, shape, subset, headers)
|> Enum.to_list()
end,
stack_id: opts[:stack_id],
query_reason: "subset_query"
)
|> case do
{:ok, result} ->
metadata =
receive do
{:pg_snapshot_info, pg_snapshot, lsn} -> make_metadata(pg_snapshot, lsn, mark)
after
0 ->
raise "failed to execute snapshot query for shape #{shape_handle}: missing pg_snapshot_info"
end
{:ok, {metadata, result}}
{:error, error} ->
{:error, error}
end
rescue
e in Querying.QueryError ->
{:error, {:where, e.message}}
end
defp make_metadata({xmin, xmax, xip_list}, lsn, mark) do
%{
xmin: xmin,
xmax: xmax,
xip_list: xip_list,
database_lsn: to_string(Lsn.to_integer(lsn)),
snapshot_mark: mark
}
end
@doc """
Asynchronous version of query_move_in that doesn't block on snapshot.
Sends {:pg_snapshot_known, name, snapshot} immediately when snapshot is known.
Sends {:query_move_in_complete, name, key_set, snapshot} when query completes.
"""
def query_move_in_async(supervisor, shape_handle, %Shape{} = shape, where, opts) do
consumer_pid = Access.fetch!(opts, :consumer_pid)
pool = Manager.pool_name(opts[:stack_id], :snapshot)
results_fn = Access.fetch!(opts, :results_fn)
:telemetry.execute([:electric, :subqueries, :move_in_triggered], %{count: 1}, %{
stack_id: opts[:stack_id]
})
Task.Supervisor.start_child(supervisor, fn ->
try do
SnapshotQuery.execute_for_shape(pool, shape_handle, shape,
stack_id: opts[:stack_id],
query_reason: "move_in_query",
snapshot_info_fn: fn _, pg_snapshot, _ ->
# Send snapshot notification immediately instead of blocking
send(consumer_pid, {:pg_snapshot_known, opts[:move_in_name], pg_snapshot})
end,
query_fn: fn conn, pg_snapshot, _ ->
result =
Querying.query_move_in(conn, opts[:stack_id], shape_handle, shape, where)
|> results_fn.(pg_snapshot)
{key_set, snapshot} = result
send(consumer_pid, {:query_move_in_complete, opts[:move_in_name], key_set, snapshot})
end
)
rescue
error ->
send(consumer_pid, {:query_move_in_error, opts[:move_in_name], error, __STACKTRACE__})
end
end)
:ok
end
def query_move_in(supervisor, shape_handle, %Shape{} = shape, where, opts) do
parent = self()
pool = Manager.pool_name(opts[:stack_id], :snapshot)
results_fn = Access.fetch!(opts, :results_fn)
Task.Supervisor.start_child(supervisor, fn ->
try do
SnapshotQuery.execute_for_shape(pool, shape_handle, shape,
stack_id: opts[:stack_id],
query_reason: "move_in_query",
snapshot_info_fn: fn _, pg_snapshot, _ ->
send(parent, {:pg_snapshot_info, pg_snapshot})
end,
query_fn: fn conn, _, _ ->
result =
Querying.query_move_in(conn, opts[:stack_id], shape_handle, shape, where)
|> results_fn.()
send(parent, {:query_move_in_complete, opts[:move_in_name], result})
end
)
rescue
error ->
send(parent, {:query_move_in_error, opts[:move_in_name], error, __STACKTRACE__})
end
end)
receive do
{:query_move_in_error, _, error, stacktrace} ->
# {:error, error, stacktrace}
reraise(error, stacktrace)
{:pg_snapshot_info, pg_snapshot} ->
# {:ok, pg_snapshot}
pg_snapshot
end
end
end