Packages
electric
1.0.9
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/file_storage/chunk_index.ex
defmodule Electric.ShapeCache.FileStorage.ChunkIndex do
@moduledoc false
alias Electric.Replication.LogOffset
alias Electric.ShapeCache.LogChunker
alias Electric.Utils
alias Electric.ShapeCache.FileStorage.LogFile
# 16 bytes offset + 8 bytes position + 16 bytes offset + 8 bytes position = 48 bytes
@chunk_entry_size 48
@doc """
Write a chunk index from the stream of log items to the given path.
A chunk index serves two purposes: it acts as a sparse index for the log file
and chunks are used to align client reads to benefit CDN cache hits.
The format of the file is:
<<first_offset::128, first_position::64, last_offset::128, last_position::64>>
Fixed byte width entries give us an opportunity to use binary search.
"""
@spec write_from_stream(
Enumerable.t(LogFile.log_item_with_sizes()),
path :: String.t(),
chunk_size :: non_neg_integer
) :: Enumerable.t(LogFile.log_item_with_sizes())
def write_from_stream(stream, path, chunk_size) do
Utils.stream_add_side_effect(
stream,
# agg is {file, write_position, byte_count, last_seen_offset}
fn -> {File.open!(path, [:write, :raw]), 0, 0, nil} end,
fn {offset, _, _, _, _, json_size, _} = line,
{file, write_position, byte_count, last_seen_offset} ->
# Start the chunk if there's no last offset
if is_nil(last_seen_offset),
do: IO.binwrite(file, <<LogFile.offset(offset)::binary, write_position::64>>)
position_after_write = LogFile.expected_position(write_position, line)
# We're counting bytes only on JSON payloads that are actually sent to the client
case LogChunker.fit_into_chunk(json_size, byte_count, chunk_size) do
{:ok, new_size} ->
{file, position_after_write, new_size, offset}
{:threshold_exceeded, 0} ->
# Chunk ended, finish writing the entry
IO.binwrite(file, <<LogFile.offset(offset)::binary, position_after_write::64>>)
{file, position_after_write, 0, nil}
end
end,
fn {file, pos, _, last_offset} = acc ->
# Finish writing the last entry if there is one
if not is_nil(last_offset),
do: IO.binwrite(file, <<LogFile.offset(last_offset)::binary, pos::64>>)
acc
end,
&File.close(elem(&1, 0))
)
end
@doc """
For a given chunk index, find the chunk that contains the first
offset greater than the given one.
Returns the max offset of the found chunk and reading boundaries for the log file.
"""
@spec fetch_chunk(path :: String.t(), LogOffset.t()) ::
{:ok, max_offset :: LogOffset.t(),
{start_position :: non_neg_integer, end_position :: non_neg_integer}}
| :error
def fetch_chunk(chunk_file_path, %LogOffset{} = exclusive_min_offset) do
file = File.open!(chunk_file_path, [:read, :raw])
{:ok, size} = :file.position(file, :eof)
try do
case do_binary_search(file, 0, div(size, @chunk_entry_size) - 1, exclusive_min_offset) do
{:ok, max_offset, start_pos, end_pos} -> {:ok, max_offset, {start_pos, end_pos}}
nil -> :error
end
after
File.close(file)
end
end
defp do_binary_search(file, left, right, %LogOffset{} = target)
when left <= right do
mid = div(left + right, 2)
{:ok, <<_min_tx::64, _min_op::64, start_pos::64, max_tx::64, max_op::64, end_pos::64>>} =
:file.pread(file, mid * @chunk_entry_size, @chunk_entry_size)
max_offset = LogOffset.new(max_tx, max_op)
case {LogOffset.compare(target, max_offset), mid} do
{:lt, mid} when mid > 0 ->
# Target is less than max_offset, this chunk might be the answer
# but let's check if there's a better one in the left half
do_binary_search(file, left, mid - 1, target) || {:ok, max_offset, start_pos, end_pos}
{:lt, _} ->
{:ok, max_offset, start_pos, end_pos}
{_, mid} when mid < right ->
# Target is equal to / greater than max_offset, need to look in right half
do_binary_search(file, mid + 1, right, target)
_ ->
# Target is greater than max_offset but we're at the end
nil
end
end
defp do_binary_search(_file, _left, _right, _target), do: nil
end