Packages
electric
1.3.2
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/shape_cache/pure_file_storage/action_file.ex
defmodule Electric.ShapeCache.PureFileStorage.ActionFile do
@moduledoc false
alias Electric.Replication.LogOffset
alias Electric.ShapeCache.PureFileStorage.KeyIndex
alias Electric.ShapeCache.PureFileStorage.LogFile
alias Electric.Utils
defp base_entry({_, label, offset, _type, entry_start, _}, action),
do: base_entry(offset, label, entry_start, action)
defp base_entry(offset, label, entry_start, action) when action in [:keep, :skip] do
action_tag = if action == :keep, do: ?k, else: ?s
# 26 = 16 (offset) + 1 (label) + 8 (entry_start) + 1 (keep)
<<26::32, LogOffset.to_int128(offset)::binary, label::8, entry_start::64, action_tag::8>>
end
defp compaction_entry({_, label, offset, _type, entry_start, _} = item, other_positions) do
# 28 + 17p = 16 (offset) + 1 (label) + 8 (entry_start) + 1 (compaction) + 2 (positions_len) + positions_len * (8 (pos) + 8 (len) + 1 label)
[
<<28 + 17 * (length(other_positions) + 1)::32, LogOffset.to_int128(offset)::binary,
label::8, entry_start::64, ?c::8, length(other_positions) + 1::16>>,
Utils.list_reverse_map([infer_json_position(item) | other_positions], fn {label, pos, size} ->
<<label::8, pos::64, size::64>>
end)
]
end
defp infer_json_position({{key_size, _key}, label, _offset, _type, entry_start, json_size}) do
{label, LogFile.expected_json_position(entry_start, key_size), json_size}
end
def create_from_key_index(key_index_path, action_file_path)
when is_binary(key_index_path) and is_binary(action_file_path) do
KeyIndex.stream_for_actions(key_index_path)
|> Stream.chunk_by(&elem(&1, 0))
|> Stream.map(fn chunk ->
# Chunk contains all operations for a given key in order
chunk
|> Enum.chunk_by(&elem(&1, 3))
|> Enum.map(fn
# Keep any single operation, since inserts/deletes won't be duplicated, and one update can't be compacted
[{_, label, offset, _, entry_start, _}] ->
base_entry(offset, label, entry_start, :keep)
# If more than one, then it's definitely an update
updates ->
updates_to_actions(updates)
end)
end)
|> Stream.into(File.stream!(action_file_path, [:delayed_write]))
|> Stream.run()
:ok =
:file_sorter.sort([to_charlist(action_file_path)], to_charlist(action_file_path),
format: fn <<tx::64, op::64>> <> _ -> {tx, op} end
)
action_file_path
end
# acc format: {positions_len, positions, actions}
defp updates_to_actions(updates, acc \\ {0, [], []})
# We don't care about order being reversed because it's going to be sorted.
defp updates_to_actions([], {_, _, acc}), do: acc
# The compaction target is either last one, or after we hit 65535 updates. Technically makes it suboptimal,
# but saves us a lot of memory because the position list will take up at most 65535 * 16 = 1048560 bytes ~ 1MB of memory,
# as opposed to 65536MB if we allow int32 positions.
defp updates_to_actions([item | rest], {total_positions, positions, actions})
when rest == []
when total_positions > 65534 do
actions = [compaction_entry(item, positions) | actions]
updates_to_actions(rest, {0, [], actions})
end
defp updates_to_actions([item | rest], {total_positions, all_positions, actions}) do
updates_to_actions(
rest,
{total_positions + 1, [infer_json_position(item) | all_positions],
[base_entry(item, :skip) | actions]}
)
end
def stream(action_file_path) do
Stream.resource(
fn -> File.open!(action_file_path, [:read, :raw, :read_ahead]) end,
fn file ->
case IO.binread(file, 30) do
<<26::32, tx::64, op::64, label::8, entry_start::64, action_tag::8>> ->
{[{{tx, op}, label, entry_start, if(action_tag == ?k, do: :keep, else: :skip)}], file}
<<_::32, tx::64, op::64, label::8, entry_start::64, ?c::8>> ->
<<count::16>> = IO.binread(file, 2)
offsets =
for <<label::8, pos::64, size::64 <- IO.binread(file, 17 * count)>>,
do: {label, pos, size}
{[{{tx, op}, label, entry_start, {:compact, offsets}}], file}
:eof ->
{:halt, file}
end
end,
&File.close/1
)
end
end