Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 59 additions & 3 deletions impl/datascript.ml
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,37 @@ let normalize_datom_for_schema = Db_impl.normalize_datom_for_schema

let refresh_db_indexes = Db_impl.refresh_indexes
let refresh_db_indexes_with_added_datoms = Db_impl.refresh_indexes_with_added_datoms
let refresh_db_indexes_with_tx_data = Db_impl.refresh_indexes_with_tx_data

(* Shared per-transaction memo (see Transact.tx_memo); populated through the
memo-aware refresh_db_indexes_with_tx_data below and consulted by
add_active_datom_with_report_db. Only read while apply_tx has it active. *)
let transact_tx_memo : Transact.tx_memo =
{ active = false; base_max_eid = 0; datoms = Hashtbl.create 8 }

let tx_memo_apply tx_data =
if transact_tx_memo.active then
List.iter
(fun d ->
if d.e > transact_tx_memo.base_max_eid
&& d.e <= Db_impl.max_allocatable_entity_id
then
let key = d.e, d.a in
let bucket =
Option.value (Hashtbl.find_opt transact_tx_memo.datoms key)
~default:[]
in
let bucket =
if d.added then d :: bucket
else
List.filter (fun x -> Util.compare_value x.v d.v <> 0) bucket
in
Hashtbl.replace transact_tx_memo.datoms key bucket)
tx_data

let refresh_db_indexes_with_tx_data db tx_data =
let db = Db_impl.refresh_indexes_with_tx_data db tx_data in
tx_memo_apply tx_data;
db

let empty_db ?(schema = []) ?storage () =
Db_impl.empty_db db_core_context ~schema ?storage ()
Expand Down Expand Up @@ -485,16 +515,41 @@ let add_active_datom_with_report_db ?(allow_tuple = false) ?(validate_value = tr
| Some existing when existing.e <> d.e ->
invalid_arg "unique constraint"
| Some _ | None -> ());
(* tx-fresh entities (e allocated inside this transaction) cannot exist
in the pre-tx index, so their same-fact and entity/attr lookups are
answered from the tx memo instead of descending the index trees. *)
(* Transaction entities live above max_allocatable_entity_id and can
already carry index datoms from earlier transactions, so they are
never memo-fresh. *)
let tx_fresh_bucket =
if transact_tx_memo.active && d.e > transact_tx_memo.base_max_eid
&& d.e <= Db_impl.max_allocatable_entity_id
then
Some
(Option.value
(Hashtbl.find_opt transact_tx_memo.datoms (d.e, d.a))
~default:[])
else None
in
let same_fact_exists =
find_eavt_exact db d.e d.a d.v |> Option.is_some
match tx_fresh_bucket with
| Some bucket ->
List.exists (fun existing -> compare_value existing.v d.v = 0) bucket
| None -> find_eavt_exact db d.e d.a d.v |> Option.is_some
in
if same_fact_exists then
db, []
else
let tx_data =
match cardinality schema_db d.a with
| Many -> [ d ]
| One -> sorted_retractions tx (entity_attr_datoms_db db d.e d.a) @ [ d ]
| One ->
let existing =
match tx_fresh_bucket with
| Some bucket -> bucket
| None -> entity_attr_datoms_db db d.e d.a
in
sorted_retractions tx existing @ [ d ]
in
(* Index decisions use schema_db (the current schema, which may have been
updated by schema datoms earlier in this transaction); the working db
Expand Down Expand Up @@ -755,6 +810,7 @@ let transact_apply_context : Transact_impl.apply_context =
; refresh_db_indexes_with_added_datoms
; refresh_db_indexes_with_tx_data
; refresh_db_identity
; tx_memo = transact_tx_memo
}

let apply_tx tx_ops db =
Expand Down
36 changes: 28 additions & 8 deletions impl/schema.ml
Original file line number Diff line number Diff line change
Expand Up @@ -47,15 +47,35 @@ let schema_attr_is_tuple = function
| Some { tuple_attrs = Some _; _ } -> true
| _ -> false

(* Schema assoc lists are immutable, so physical equality on the list is a
valid staleness check: any schema update produces a new list (same trick
as Schema_access.schema_attr). This runs once per index datom insert, so
the per-schema result is precomputed for every attr instead of scanned. *)
let avet_tbl_schema : schema ref = ref []
let avet_tbl : (attr, bool) Hashtbl.t ref = ref (Hashtbl.create 0)

let schema_attr_is_avet_accessible schema attr =
attr = "db/ident"
|| schema_attr_is_tuple (schema_attr_by_name schema attr)
||
match schema_attr_by_name schema attr with
| Some { value_type = Some RefType; _ }
| Some { unique = Some _; _ }
| Some { indexed = true; _ } -> true
| _ -> false
if attr = "db/ident" then true
else begin
if not (schema == !avet_tbl_schema) then begin
let tbl = Hashtbl.create (List.length schema) in
List.iter
(fun (attr, spec) ->
if not (Hashtbl.mem tbl attr) then
Hashtbl.add tbl attr
(schema_attr_is_tuple (Some spec)
||
match spec with
| { value_type = Some RefType; _ }
| { unique = Some _; _ }
| { indexed = true; _ } -> true
| _ -> false))
schema;
avet_tbl := tbl;
avet_tbl_schema := schema
end;
Option.value (Hashtbl.find_opt !avet_tbl attr) ~default:false
end

let schema_has_no_history schema attr =
match List.assoc_opt attr schema with
Expand Down
87 changes: 66 additions & 21 deletions impl/transact.ml
Original file line number Diff line number Diff line change
Expand Up @@ -396,6 +396,20 @@ let remap_tempid_entity old_e new_e tempids =
}


(* Per-transaction memo of datoms added to entities allocated inside this
transaction (base_max_eid < e <= max_allocatable_entity_id): they cannot
exist in the pre-tx index, so same-fact and entity/attr lookups for them
can be answered from this table instead of descending the index trees.
Bulk seed transactions (imports) consist almost entirely of such datoms.
Transaction entities (e >= tx0) are excluded: they are ids, not freshly
allocated entities, and may already carry index datoms from earlier
transactions. *)
type tx_memo =
{ mutable active : bool
; mutable base_max_eid : entity_id
; datoms : (entity_id * attr, datom list) Hashtbl.t
}

type apply_context =
{ resolve_context : context
; is_filtered : db -> bool
Expand Down Expand Up @@ -430,10 +444,38 @@ type apply_context =
; refresh_db_indexes_with_added_datoms : db -> datom list -> db
; refresh_db_indexes_with_tx_data : db -> datom list -> db
; refresh_db_identity : db -> db
; tx_memo : tx_memo
}

let apply_tx context tx_ops db =
if context.is_filtered db then invalid_arg "filtered db is read-only";
let tx_memo = context.tx_memo in
(* The memo record is shared by every transaction, so a nested transaction
(a Call fn that transacts) or an exception raised mid-tx must not leak
this transaction's state into the outer one: save any enclosing state
and restore it when this transaction ends or aborts. *)
let saved_memo =
if tx_memo.active then
Some (tx_memo.base_max_eid, Hashtbl.copy tx_memo.datoms)
else None
in
let restore_memo () =
match saved_memo with
| Some (base_max_eid, datoms) ->
Hashtbl.reset tx_memo.datoms;
Hashtbl.iter (Hashtbl.replace tx_memo.datoms) datoms;
tx_memo.base_max_eid <- base_max_eid
| None ->
(* The outermost transaction is over: the memo table is dead state
until the next transaction and would otherwise retain every datom
of the transaction that just ran. *)
tx_memo.active <- false;
Hashtbl.reset tx_memo.datoms
in
Fun.protect ~finally:restore_memo @@ fun () ->
tx_memo.base_max_eid <- db.max_eid;
Hashtbl.reset tx_memo.datoms;
tx_memo.active <- true;
(* Mid-tx schema refresh is incremental: schema datoms appended to tx_data
are recorded here and each refresh folds only the entities touched by
the datoms seen since the previous refresh, instead of rescanning all
Expand Down Expand Up @@ -1851,24 +1893,27 @@ let apply_tx context tx_ops db =
; tx_fns = !current_tx_fns
}
in
( (match fast_tx_data with
| Some _ ->
db_after
|> (fun db -> context.refresh_db_indexes_with_tx_data db tx_data)
|> context.refresh_db_identity
| None ->
db_after
|> (fun db ->
if List.exists tx_op_affects_schema tx_ops then
{ datoms with
schema = db.schema
; max_eid = db.max_eid
; max_tx = db.max_tx
; tx_fns = db.tx_fns
}
else
context.refresh_db_indexes_with_tx_data db tx_data)
|> context.refresh_db_identity)
, tempid_map_order tempids
, tx_data
)
let result =
( (match fast_tx_data with
| Some _ ->
db_after
|> (fun db -> context.refresh_db_indexes_with_tx_data db tx_data)
|> context.refresh_db_identity
| None ->
db_after
|> (fun db ->
if List.exists tx_op_affects_schema tx_ops then
{ datoms with
schema = db.schema
; max_eid = db.max_eid
; max_tx = db.max_tx
; tx_fns = db.tx_fns
}
else
context.refresh_db_indexes_with_tx_data db tx_data)
|> context.refresh_db_identity)
, tempid_map_order tempids
, tx_data
)
in
result
13 changes: 13 additions & 0 deletions impl/transact.mli
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,18 @@ val remap_datom_entity : context -> entity_id -> entity_id -> datom -> datom
val remap_resolved_tx_value : context -> entity_id -> entity_id -> tx_value -> tx_value
val remap_tempid_entity : entity_id -> entity_id -> tempid_map -> tempid_map

(* Per-transaction memo of datoms added to entities allocated inside this
transaction (base_max_eid < e <= max_allocatable_entity_id): they cannot
exist in the pre-tx index, so same-fact and entity/attr lookups for them
can be answered from this table instead of descending the index trees.
Transaction entities (e >= tx0) are excluded: they may already carry
index datoms from earlier transactions. *)
type tx_memo =
{ mutable active : bool
; mutable base_max_eid : entity_id
; datoms : (entity_id * attr, datom list) Hashtbl.t
}

type apply_context =
{ resolve_context : context
; is_filtered : db -> bool
Expand Down Expand Up @@ -82,6 +94,7 @@ type apply_context =
; refresh_db_indexes_with_added_datoms : db -> datom list -> db
; refresh_db_indexes_with_tx_data : db -> datom list -> db
; refresh_db_identity : db -> db
; tx_memo : tx_memo
}

val apply_tx : apply_context -> tx_op list -> db -> db * (string * entity_id) list * datom list
55 changes: 31 additions & 24 deletions impl/util.ml
Original file line number Diff line number Diff line change
Expand Up @@ -110,37 +110,44 @@ let split_keyword keyword =
let name = String.sub keyword (index + 1) (String.length keyword - index - 1) in
namespace, name

(* compare [off..off+len) slices of two strings without allocation *)
let compare_slice (s : string) (soff : int) (slen : int) (t : string)
(toff : int) (tlen : int) : int =
let rec loop i =
if i >= slen && i >= tlen then 0
else if i >= slen then -1
else if i >= tlen then 1
else
let c = Char.compare s.[soff + i] t.[toff + i] in
if c <> 0 then c else loop (i + 1)
in
loop 0

(* Upstream compares attr keywords with compare-keywords: no-namespace
attrs sort first, otherwise (ns, name) pairwise. Compare with a
plain string sort here would order "logseq.property.class/extends"
before "logseq.property/built-in?" ('.' < '/'), which both diverges
from the runtime index order upstream and writes a leaf order the
CLJS storage reader cannot binary-search.
Slice-wise comparison keeps the same order without allocating the
(ns, name) pair for every index compare. *)

The (ns, name) pairwise order equals a byte compare on a normalized
key: replace the last '/' by '\x00' (or prepend '\x00' when the attr
has no '/'). '\x00' sorts below every byte in an attr and below '/',
so a shorter ns that is a prefix of the other always loses to it, and
a missing ns sorts first — exactly the pairwise order. Keys are
memoized by attr content since index comparators see the same small
set of attrs over and over. *)
(* Bounded: keyword values also flow through compare_attr, so an unlimited
vocabulary would grow this table without bound; past the cap, keys are
recomputed per call. *)
let attr_key_tbl_cap = 65536
let attr_key_tbl : (attr, attr) Hashtbl.t = Hashtbl.create 256

let attr_key attr =
match Hashtbl.find_opt attr_key_tbl attr with
| Some key -> key
| None ->
let key =
match String.rindex_opt attr '/' with
| None -> "\x00" ^ attr
| Some i ->
String.init (String.length attr) (fun j ->
if j = i then '\x00' else attr.[j])
in
if Hashtbl.length attr_key_tbl < attr_key_tbl_cap then
Hashtbl.replace attr_key_tbl attr key;
key

let compare_attr left right =
let li = match String.rindex_opt left '/' with Some i -> i | None -> -1 in
let ri = match String.rindex_opt right '/' with Some i -> i | None -> -1 in
(* ns = [0, li) for li >= 0 else ""; name = [li+1, len) *)
let c = compare_slice left 0 (max li 0) right 0 (max ri 0) in
if c <> 0 then c
else
let loff, roff = li + 1, ri + 1 in
compare_slice left loff (String.length left - loff) right roff
(String.length right - roff)
if left == right || String.equal left right then 0
else String.compare (attr_key left) (attr_key right)

let rec compare_list_items_with compare_item left right =
match left, right with
Expand Down
19 changes: 19 additions & 0 deletions test/test_datascript.ml
Original file line number Diff line number Diff line change
Expand Up @@ -2314,6 +2314,24 @@ let test_transact__test_resolve_current_tx () =
[ "db/current-tx", tx0 + 1; "entity", 1 ]
report.tempids

let test_transact__test_tx_entity_cardinality () =
(* Datoms on a transaction entity created by an earlier transaction
already live in the index: a cardinality-one update on it in a later
transaction must retract the previous value — tx ids are above
max_allocatable_entity_id, not transaction-fresh entities. *)
let schema = [ "note", indexed ] in
let report1 =
transact (empty_db ~schema ()) [ Add (CurrentTx, "note", String "v1") ]
in
let tx_eid = Option.get (resolve_tempid report1.tempids "db/current-tx") in
let report2 =
transact report1.db_after [ Add (Entity_id tx_eid, "note", String "v2") ]
in
assert_equal_triples
"cardinality-one update on a past tx entity retracts the old value"
[ tx_eid, "note", String "v2" ]
(datoms report2.db_after Eavt ~e:tx_eid ~a:"note" ())

let test_current_tx_string_aliases_resolve_in_transactions () =
let report =
transact
Expand Down Expand Up @@ -17560,6 +17578,7 @@ let () =
test_transact_report_exposes_tempids ();
test_resolve_tempid_reads_tx_report_tempids ();
test_transact__test_resolve_current_tx ();
test_transact__test_tx_entity_cardinality ();
test_current_tx_string_aliases_resolve_in_transactions ();
test_current_tx_string_aliases_can_be_value_only ();
test_current_tx_colon_string_alias_resolves_in_transactions ();
Expand Down
Loading