diff --git a/docs/design.md b/docs/design.md index 2ab9f3b..539fb64 100644 --- a/docs/design.md +++ b/docs/design.md @@ -159,6 +159,16 @@ Not every value near the DB should become an ordered persistent index. Moving these to PSS would not improve parity with upstream DataScript and would make serialization and equality behavior more complex. +## Storage Tail + +`conn.storage_tail` holds one `datom list` batch per committed transaction, +stored newest-first so appending a transaction's `tx_data` is O(1). The +batches are flattened back to transaction order only where the flat tail is +consumed: `store_tail` writes, `storage_tail_datom_count`, the compaction +threshold check, and the `Conn.storage_tail` accessor. The persisted +`Storage_tail` payload keeps upstream's transaction-order-then-in-transaction +datom order. + ## Persistent Sorted Set Dependency The repo depends on `persistent_sorted_set_ocaml` through an opam pin to diff --git a/impl/conn.ml b/impl/conn.ml index 491c246..125fb81 100644 --- a/impl/conn.ml +++ b/impl/conn.ml @@ -5,7 +5,10 @@ type t = ; mutable listeners : (string * (tx_report -> unit)) list ; mutable next_listener_id : int ; storage : storage option - ; mutable storage_tail : datom list list + ; (* per-transaction datom batches, newest first: consing each report's + tx_data is O(1); reversed back to on-disk tx order only when the + tail is written or read out *) + mutable storage_tail : datom list list } type creation_context = @@ -70,7 +73,7 @@ let make ?storage ?(storage_tail = []) db = | None -> db | Some _ -> { db with storage_ref = storage } in - { db; listeners = []; next_listener_id = 0; storage; storage_tail } + { db; listeners = []; next_listener_id = 0; storage; storage_tail = List.rev storage_tail } let create (context : creation_context) ?schema ?storage () = let db = context.empty_db ?schema ?storage () in @@ -97,7 +100,7 @@ let db conn = conn.db store/notify side effects, for bookkeeping fields like :max-tx. *) let update_db conn f = conn.db <- f conn.db -let storage_tail conn = conn.storage_tail +let storage_tail conn = List.rev conn.storage_tail let is_conn (_ : t) = true @@ -145,13 +148,13 @@ let transact (context : transact_context) ?(tx_meta = []) conn tx_data = | None -> () | Some storage -> if report.tx_data <> [] then begin - let tail = conn.storage_tail @ [ report.tx_data ] in + let tail = report.tx_data :: conn.storage_tail in if context.storage_tail_datom_count tail > context.storage_tail_compaction_threshold report.db_after then begin conn.db <- context.store ~storage report.db_after; conn.storage_tail <- [] end else begin conn.storage_tail <- tail; - context.store_tail storage conn.storage_tail + context.store_tail storage (List.rev conn.storage_tail) end end); notify_listeners conn report; @@ -173,13 +176,13 @@ let apply_report (context : transact_context) conn (report : tx_report) = | None -> () | Some storage -> if report.tx_data <> [] then begin - let tail = conn.storage_tail @ [ report.tx_data ] in + let tail = report.tx_data :: conn.storage_tail in if context.storage_tail_datom_count tail > context.storage_tail_compaction_threshold db_after then begin conn.db <- context.store ~storage db_after; conn.storage_tail <- [] end else begin conn.storage_tail <- tail; - context.store_tail storage conn.storage_tail + context.store_tail storage (List.rev conn.storage_tail) end end); let report = { report with db_after } in diff --git a/impl/datascript.ml b/impl/datascript.ml index 09a4de1..f2b522a 100644 --- a/impl/datascript.ml +++ b/impl/datascript.ml @@ -903,11 +903,14 @@ let persist_transact_tail ~tx_meta db tx_data = match db.storage_ref with | None -> db | Some storage -> - let tail = restore_tail_groups storage @ [ tx_data ] in - if storage_tail_datom_count tail > storage_tail_compaction_threshold db then + let groups = restore_tail_groups storage in + if + storage_tail_datom_count groups + List.length tx_data + > storage_tail_compaction_threshold db + then store ~storage db else begin - store_tail storage tail; + store_tail storage (groups @ [ tx_data ]); db end diff --git a/impl/storage.ml b/impl/storage.ml index b07b1ba..8d93bbc 100644 --- a/impl/storage.ml +++ b/impl/storage.ml @@ -280,7 +280,7 @@ let tail_compaction_threshold (db : db) = (PSet.settings db.eavt_index).branching_factor let tail_datom_count tail = - tail |> List.concat |> List.length + List.fold_left (fun count group -> count + List.length group) 0 tail let restore_root_snapshot storage = match storage.storage_restore root_address with diff --git a/test/test_storage.ml b/test/test_storage.ml index f4eefc7..a2fd7ac 100644 --- a/test/test_storage.ml +++ b/test/test_storage.ml @@ -379,6 +379,70 @@ let test_storage__test_conn () = (datoms restored_db Eavt ()) | None -> failwith "conn_from_datoms should store attached datoms" +let tail_groups storage = + match storage.storage_restore Storage.tail_address with + | Some (Storage_tail groups) -> groups + | _ -> failwith "storage tail address missing" + +let flatten_tail_groups groups = + List.map (fun group -> List.map (fun d -> d.e, d.a, d.v) group) groups + +let test_storage__test_conn_tail_write_order () = + let storage = memory_storage () in + let conn = create_conn ~storage () in + let assert_tail label conn expected = + let stored = flatten_tail_groups (tail_groups storage) in + if stored <> expected then + failf "%s: stored tail order mismatch: %s" label + (String.concat ";" (List.map (fun g -> string_of_int (List.length g)) stored)); + if flatten_tail_groups (Conn.storage_tail conn) <> expected then + failwith (label ^ ": conn storage_tail accessor order mismatch") + in + ignore (transact_conn conn [ Add (Entity_id 1, "a", String "1") ]); + ignore + (transact_conn conn + [ Add (Entity_id 2, "a", String "2"); Add (Entity_id 2, "b", String "3") ]); + assert_tail "two txs" conn + [ [ 1, "a", String "1" ] + ; [ 2, "a", String "2"; 2, "b", String "3" ] + ]; + (* a restored conn seeds its tail from disk and keeps appending in tx + order; the on-disk order must survive the internal reversal *) + let conn = + match restore_conn storage with + | Some conn -> conn + | None -> failwith "restore_conn should return a conn" + in + ignore (transact_conn conn [ Add (Entity_id 3, "a", String "4") ]); + assert_tail "restored conn" conn + [ [ 1, "a", String "1" ] + ; [ 2, "a", String "2"; 2, "b", String "3" ] + ; [ 3, "a", String "4" ] + ]; + (* crossing the compaction threshold clears the stored tail *) + ignore + (transact_conn conn + (List.init 40 (fun i -> Add (Entity_id (i + 10), "a", String (string_of_int i))))); + (match tail_groups storage with + | [] -> () + | _ -> failwith "compaction should clear the stored tail"); + ignore (transact_conn conn [ Add (Entity_id 99, "a", String "9") ]); + assert_tail "post-compaction" conn [ [ 99, "a", String "9" ] ]; + assert_equal_triples + "all datoms queryable after compaction" + [ 1, "a", String "1"; 2, "a", String "2"; 2, "b", String "3" + ; 3, "a", String "4"; 99, "a", String "9" ] + (datoms (conn_db conn) Eavt ~e:1 () + @ datoms (conn_db conn) Eavt ~e:2 () + @ datoms (conn_db conn) Eavt ~e:3 () + @ datoms (conn_db conn) Eavt ~e:99 () + |> List.sort compare) + +let test_storage__test_tail_datom_count () = + let d tx e = datom ~tx ~e ~a:"a" ~v:(String "v") () in + assert_equal_int "tail_datom_count sums group sizes" 4 + (Storage.tail_datom_count [ [ d 1 1 ]; [ d 2 2; d 2 3; d 2 4 ]; [] ]) + let test_storage__test_db_with_tail () = let db = empty_db ~schema:[ "block/updated-at", indexed; "block/uuid", unique_identity ] () @@ -525,4 +589,6 @@ let () = test_storage__test_transact_after_restore_uses_index_slices (); test_storage__test_conn_repeated_transacts_store_incrementally (); test_storage__test_conn (); + test_storage__test_conn_tail_write_order (); + test_storage__test_tail_datom_count (); test_storage__test_db_with_tail ();