Pure OCaml B-tree implementation for persistent storage
Fix merlint E005/E010: split long functions and flatten deep nesting
Refactored 21 functions across 17 files to bring them under merlint
thresholds. Pure structural refactoring — no behavior changes.
Packages: irmin, merlint, monopam, ocaml-btree, ocaml-cdm, ocaml-git,
ocaml-globe, ocaml-kepler, ocaml-osv, ocaml-sigstore, ocaml-stix, prune.
+63
-64
+63
-64
lib/table.ml
+63
-64
lib/table.ml
···
47
47
follow first_overflow remaining;
48
48
Buffer.contents buf
49
49
50
+
(* Write cell header (payload_size varint + rowid varint) into [cell] *)
51
+
let write_cell_header cell ~payload_size_varint ~rowid_varint =
52
+
Bytes.blit_string payload_size_varint 0 cell 0
53
+
(String.length payload_size_varint);
54
+
Bytes.blit_string rowid_varint 0 cell
55
+
(String.length payload_size_varint)
56
+
(String.length rowid_varint)
57
+
58
+
(* Write overflow chain pages and return the array of page numbers *)
59
+
let write_overflow_chain ~pager ~data ~local ~usable =
60
+
let payload_size = String.length data in
61
+
let overflow_payload_size = usable - 4 in
62
+
let remaining = payload_size - local in
63
+
let num_overflow_pages =
64
+
(remaining + overflow_payload_size - 1) / overflow_payload_size
65
+
in
66
+
let overflow_pages =
67
+
Array.init num_overflow_pages (fun _ -> Pager.allocate pager)
68
+
in
69
+
let data_off = ref local in
70
+
for i = 0 to num_overflow_pages - 1 do
71
+
let page_buf = Bytes.create usable in
72
+
let next =
73
+
if i < num_overflow_pages - 1 then overflow_pages.(i + 1) else 0
74
+
in
75
+
Page.set_u32_be page_buf 0 next;
76
+
let chunk = min (payload_size - !data_off) overflow_payload_size in
77
+
Bytes.blit_string data !data_off page_buf 4 chunk;
78
+
data_off := !data_off + chunk;
79
+
Pager.write pager overflow_pages.(i) (Bytes.unsafe_to_string page_buf)
80
+
done;
81
+
overflow_pages
82
+
50
83
(* Encode a table leaf cell, spilling to overflow pages as needed *)
51
84
let encode_table_leaf_cell ~pager ~rowid ~data =
52
85
let usable = Pager.page_size pager in
···
61
94
if payload_size <= max_local then begin
62
95
(* Fits inline - no overflow needed *)
63
96
let cell = Bytes.create (header_len + payload_size) in
64
-
Bytes.blit_string payload_size_varint 0 cell 0
65
-
(String.length payload_size_varint);
66
-
Bytes.blit_string rowid_varint 0 cell
67
-
(String.length payload_size_varint)
68
-
(String.length rowid_varint);
97
+
write_cell_header cell ~payload_size_varint ~rowid_varint;
69
98
Bytes.blit_string data 0 cell header_len payload_size;
70
99
Bytes.unsafe_to_string cell
71
100
end
···
73
102
(* Need overflow pages *)
74
103
let k = min_local + ((payload_size - min_local) mod (usable - 4)) in
75
104
let local = if k <= max_local then k else min_local in
76
-
let overflow_payload_size = usable - 4 in
77
-
(* Write overflow chain *)
78
-
let remaining = payload_size - local in
79
-
let num_overflow_pages =
80
-
(remaining + overflow_payload_size - 1) / overflow_payload_size
81
-
in
82
-
(* Allocate all overflow pages first so we know the page numbers *)
83
-
let overflow_pages =
84
-
Array.init num_overflow_pages (fun _ -> Pager.allocate pager)
85
-
in
86
-
(* Write data to overflow pages *)
87
-
let data_off = ref local in
88
-
for i = 0 to num_overflow_pages - 1 do
89
-
let page_buf = Bytes.create usable in
90
-
let next =
91
-
if i < num_overflow_pages - 1 then overflow_pages.(i + 1) else 0
92
-
in
93
-
Page.set_u32_be page_buf 0 next;
94
-
let chunk = min (payload_size - !data_off) overflow_payload_size in
95
-
Bytes.blit_string data !data_off page_buf 4 chunk;
96
-
data_off := !data_off + chunk;
97
-
Pager.write pager overflow_pages.(i) (Bytes.unsafe_to_string page_buf)
98
-
done;
105
+
let overflow_pages = write_overflow_chain ~pager ~data ~local ~usable in
99
106
(* Build cell: header + local payload + 4-byte overflow pointer *)
100
107
let cell = Bytes.create (header_len + local + 4) in
101
-
Bytes.blit_string payload_size_varint 0 cell 0
102
-
(String.length payload_size_varint);
103
-
Bytes.blit_string rowid_varint 0 cell
104
-
(String.length payload_size_varint)
105
-
(String.length rowid_varint);
108
+
write_cell_header cell ~payload_size_varint ~rowid_varint;
106
109
Bytes.blit_string data 0 cell header_len local;
107
110
Page.set_u32_be cell (header_len + local) overflow_pages.(0);
108
111
Bytes.unsafe_to_string cell
···
385
388
(* After split, insert via full tree traversal *)
386
389
insert t ~rowid data
387
390
end
388
-
else begin
391
+
else
389
392
(* Degenerate case: page has 0 or 1 cells and the new cell doesn't fit
390
393
alongside them. Allocate a fresh page for the new cell and propagate
391
394
a separator up. *)
392
-
let page_size = Pager.page_size t.pager in
393
-
let new_page_num = Pager.allocate t.pager in
394
-
let new_buf = Page.init ~page_size ~kind:Page.Leaf_table in
395
-
let new_header = Page.parse_header (Bytes.unsafe_to_string new_buf) 0 in
396
-
ignore
397
-
(write_and_insert_cell new_buf ~header:new_header ~kind:Page.Leaf_table
398
-
~index:0 ~cell);
399
-
Pager.write t.pager new_page_num (Bytes.unsafe_to_string new_buf);
400
-
(* Determine separator: the existing page's last rowid separates the pages.
401
-
We need left page keys < separator, right page keys >= separator. *)
402
-
if header.Page.cell_count = 0 then
403
-
(* Existing page is empty. New cell goes to new page. Separator = rowid. *)
395
+
insert_into_leaf_degenerate t page_num page ~header ~rowid ~cell
396
+
~parent_stack
397
+
398
+
and insert_into_leaf_degenerate t page_num page ~header ~rowid ~cell
399
+
~parent_stack =
400
+
let page_size = Pager.page_size t.pager in
401
+
let new_page_num = Pager.allocate t.pager in
402
+
let new_buf = Page.init ~page_size ~kind:Page.Leaf_table in
403
+
let new_header = Page.parse_header (Bytes.unsafe_to_string new_buf) 0 in
404
+
ignore
405
+
(write_and_insert_cell new_buf ~header:new_header ~kind:Page.Leaf_table
406
+
~index:0 ~cell);
407
+
Pager.write t.pager new_page_num (Bytes.unsafe_to_string new_buf);
408
+
if header.Page.cell_count = 0 then
409
+
propagate_split t ~parent_stack ~left_page:page_num ~separator_rowid:rowid
410
+
~right_page:new_page_num
411
+
else begin
412
+
let usable = usable_size t in
413
+
let ptrs = Page.cell_pointers page 0 header in
414
+
let existing_cell, _ =
415
+
Cell.parse_table_leaf page ptrs.(0) ~usable_size:usable
416
+
in
417
+
if rowid > existing_cell.Cell.rowid then
404
418
propagate_split t ~parent_stack ~left_page:page_num ~separator_rowid:rowid
405
419
~right_page:new_page_num
406
-
else begin
407
-
(* Existing page has 1 cell. Figure out which page should be left/right. *)
408
-
let usable = usable_size t in
409
-
let ptrs = Page.cell_pointers page 0 header in
410
-
let existing_cell, _ =
411
-
Cell.parse_table_leaf page ptrs.(0) ~usable_size:usable
412
-
in
413
-
if rowid > existing_cell.Cell.rowid then
414
-
(* New cell (larger rowid) goes right, existing stays left.
415
-
Separator = new cell's rowid so routing sends >= rowid to right. *)
416
-
propagate_split t ~parent_stack ~left_page:page_num
417
-
~separator_rowid:rowid ~right_page:new_page_num
418
-
else
419
-
(* New cell (smaller rowid) goes left, existing stays right.
420
-
Separator = existing cell's rowid. *)
421
-
propagate_split t ~parent_stack ~left_page:new_page_num
422
-
~separator_rowid:existing_cell.Cell.rowid ~right_page:page_num
423
-
end
420
+
else
421
+
propagate_split t ~parent_stack ~left_page:new_page_num
422
+
~separator_rowid:existing_cell.Cell.rowid ~right_page:page_num
424
423
end
425
424
426
425
and propagate_split t ~parent_stack ~left_page ~separator_rowid ~right_page =