Packages
electric
1.4.12
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/postgres/snapshot_query.ex
defmodule Electric.Postgres.SnapshotQuery do
alias Electric.SnapshotError
alias Electric.Shapes.Shape
alias Electric.Telemetry.OpenTelemetry
@type pg_snapshot() ::
{xmin :: pos_integer(), xmax :: pos_integer(), xip_list :: [pos_integer()]}
@doc """
Execute a snapshot query for a shape in a isolated readonly transaction.
This function operates on two callbacks: `snapshot_info_fn` and `query_fn`.
`snapshot_info_fn` is called with the shape handle, the pg_snapshot, and the lsn as soon
as the snapshot information for the started transaction is available.
`query_fn` is called with the connection, the pg_snapshot, and the lsn and
is expected to do all the work querying and dealing with the results.
The query function is executed within a transaction, so it shouldn't return
a stream (as it will fail to be read after the transaction is committed), but
rather should execute all desired side-effects or materialize the results.
Query will be executed within a REPEATABLE READ READ ONLY transaction, with
correct display settings set.
Options:
- `:query_fn` - the function to execute the query.
- `:snapshot_info_fn` - the function to call with the snapshot information.
- `:stack_id` - the stack id for this shape.
"""
@spec execute_for_shape(Postgrex.conn(), Shape.handle(), Shape.t(), [option]) ::
{:ok, result} | {:error, any()}
when result: term(),
option:
{:snapshot_info_fn, (Shape.handle(), pg_snapshot, pos_integer() -> any())}
| {:query_fn, (Postgrex.conn(), pg_snapshot, pos_integer() -> result)}
| {:stack_id, Electric.stack_id()},
pg_snapshot:
{xmin :: pos_integer(), xmax :: pos_integer(), xip_list :: [pos_integer()]}
def execute_for_shape(pool, shape_handle, shape, opts) do
query_fn = Access.fetch!(opts, :query_fn)
snapshot_info_fn = Access.fetch!(opts, :snapshot_info_fn)
shape_attrs = shape_attrs(shape_handle, shape)
stack_id = Access.fetch!(opts, :stack_id)
Postgrex.transaction(
pool,
fn conn ->
ctx = %{
conn: conn,
stack_id: stack_id,
span_attrs: shape_attrs,
query_reason: Access.get(opts, :query_reason, "initial_snapshot")
}
query!(ctx, "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY",
span_name: "shape_snapshot.start_readonly_txn"
)
[%{rows: [[pg_snapshot, lsn]]}] =
query!(ctx, "SELECT pg_current_snapshot(), pg_current_wal_lsn()",
span_name: "shape_snapshot.get_pg_snapshot"
)
snapshot_info_fn.(shape_handle, pg_snapshot, lsn)
query!(ctx, Electric.Postgres.display_settings(),
span_name: "shape_snapshot.set_display_settings"
)
query_fn.(conn, pg_snapshot, lsn)
end,
timeout: :infinity
)
catch
:exit, {_, {DBConnection.Holder, :checkout, _}} ->
raise SnapshotError.connection_not_available()
end
defp shape_attrs(shape_handle, shape) do
[
"shape.handle": shape_handle,
"shape.root_table": shape.root_table,
"shape.where": if(not is_nil(shape.where), do: shape.where.query, else: nil)
]
end
@spec query!(map(), String.t() | [String.t()], Keyword.t()) :: [Postgrex.Result.t(), ...]
defp query!(
%{conn: conn, stack_id: stack_id, span_attrs: span_attrs},
query_or_queries,
opts
) do
OpenTelemetry.with_span(
Keyword.fetch!(opts, :span_name),
span_attrs,
stack_id,
fn ->
query_or_queries
|> List.wrap()
|> Enum.map(fn query -> Postgrex.query!(conn, query, Keyword.get(opts, :params, [])) end)
end
)
end
end