this repo has no description
Update manifest after compaction
+53
-11
+11
-8
lib/xks/compaction.ex
+11
-8
lib/xks/compaction.ex
···
11
11
@type encoded_data_block :: {:encoded_data_block, iolist, versioned_key, versioned_key, [tuple]}
12
12
# {:data_block, i, block_index, block_checksum, first_key, last_key, index_entries}
13
13
@type data_block :: {:data_block, non_neg_integer, non_neg_integer, Blocks.checksum, versioned_key, versioned_key, [tuple]}
14
-
# {:table, last_block_index, last_block_checksum, first_key, last_key}
14
+
# {:table, last_block_index, last_block_checksum, start_key, end_key}
15
15
@type table :: {:table, non_neg_integer, Blocks.checksum, versioned_key, versioned_key}
16
16
17
17
# KiB
···
50
50
51
51
@table_metadata_size (4 + 4 + 4)
52
52
53
+
@spec compact(XKS.t, [term]) :: [table]
53
54
def compact(%XKS{} = xks, tables) do
54
55
merge_state =
55
56
tables
56
57
|> Enum.map(fn {_id, mt} -> Merge.iterator_for_memtable(mt, "") end)
57
58
|> Merge.new()
58
59
59
-
_tables = do_compact(xks, merge_state, [], nil)
60
-
|> dbg()
61
-
62
-
:ok
60
+
# Note: output tables are returned in reverse order
61
+
do_compact(xks, merge_state, [], nil)
63
62
end
64
63
65
64
defp do_compact(_xks, :empty, table_acc, initial_sub) do
···
82
81
data_blocks = Enum.reverse(data_blocks_reversed)
83
82
# TODO: unused data block indexes are never freed
84
83
85
-
{:data_block, _i, _index, _checksum, table_first_key, _last_key, _entries} = hd(data_blocks)
86
-
{:data_block, _i, _index, _checksum, _first_key, table_last_key, _entries} = hd(data_blocks_reversed)
84
+
{:data_block, _i, _index, _checksum, {fk_key, fk_version}, _last_key, _entries} = hd(data_blocks)
85
+
{:data_block, _i, _index, _checksum, _first_key, {lk_key, lk_version}, _entries} = hd(data_blocks_reversed)
86
+
# Copy out start/end keys so that they do not reference blocks
87
+
start_key = {:binary.copy(fk_key), fk_version}
88
+
# Note: end_key is exclusive
89
+
end_key = {:binary.copy(lk_key <> "\x00"), lk_version}
87
90
88
91
data_block_addresses =
89
92
Enum.map(data_blocks, fn {:data_block, _i, index, checksum, _fk, _lk, _entries} ->
···
108
111
trailer_block_data = encode_trailer_block(block_addresses, index_slots_data, index_slot_count)
109
112
trailer_block_checksum = Blocks.write(xks.block_store, trailer_block_index, trailer_block_data)
110
113
111
-
table = {:table, trailer_block_index, trailer_block_checksum, table_first_key, table_last_key}
114
+
table = {:table, trailer_block_index, trailer_block_checksum, start_key, end_key}
112
115
{table, next_sub, state}
113
116
end
114
117
+17
-1
lib/xks/manifest.ex
+17
-1
lib/xks/manifest.ex
···
1
1
defmodule Hobbes.XKS.Manifest do
2
2
@type t :: :ets.table
3
3
4
-
alias Hobbes.XKS.Memtable
4
+
alias Hobbes.XKS.{Blocks, Memtable}
5
5
6
6
@memtable_level -1
7
7
···
54
54
55
55
_ -> acc
56
56
end
57
+
end
58
+
59
+
@spec insert_table(t, non_neg_integer, non_neg_integer, {non_neg_integer, Blocks.checksum}, {binary, non_neg_integer}, {binary, non_neg_integer}) :: :ok
60
+
def insert_table(manifest, epoch, level, block_address, first_key, last_key) do
61
+
{block_index, block_checksum} = block_address
62
+
{fk_key, fk_version} = first_key
63
+
{lk_key, lk_version} = last_key
64
+
65
+
id = next_table_id(manifest)
66
+
:ets.insert(manifest, {{level, fk_key, fk_version, epoch}, {lk_key, lk_version, block_index, block_checksum, id}})
67
+
:ok
68
+
end
69
+
70
+
@doc false
71
+
def dump(manifest) do
72
+
:ets.tab2list(manifest)
57
73
end
58
74
end
+25
-2
lib/xks/xks.ex
+25
-2
lib/xks/xks.ex
···
116
116
117
117
@spec compact_memtables(t) :: :ok
118
118
def compact_memtables(%XKS{} = xks) do
119
-
memtables = Manifest.list_memtables(xks.manifest)
120
-
Compaction.compact(xks, memtables)
119
+
manifest = xks.manifest
120
+
121
+
input_tables = Manifest.list_memtables(manifest)
122
+
output_tables = Compaction.compact(xks, input_tables)
123
+
124
+
epoch = :atomics.get(xks.epoch_atomic, 1)
125
+
# We can't increment the atomic until after the memtable has been updated
126
+
epoch = epoch + 1
127
+
128
+
# Delete input tables
129
+
Enum.each(input_tables, fn {id, _memtable} ->
130
+
Manifest.delete_memtable(manifest, epoch, id)
131
+
end)
132
+
133
+
# Insert output tables
134
+
level = 1
135
+
Enum.each(output_tables, fn {:table, block_index, block_checksum, first_key, last_key} ->
136
+
Manifest.insert_table(manifest, epoch, level, {block_index, block_checksum}, first_key, last_key)
137
+
end)
138
+
139
+
# Update epoch *after* all writes are complete so that readers cannot observe torn writes
140
+
:atomics.put(xks.epoch_atomic, 1, epoch)
141
+
142
+
dbg Manifest.dump(manifest)
143
+
121
144
:ok
122
145
end
123
146