From da34b870891a0926175b742057be6b881a79ed63 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 30 Aug 2026 00:33:27 +0000 Subject: [PATCH] Fix query cache keying, bulk card-one re-assert, Int AVET bounds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Result cache now keys on db_uid plus as_of/since/history/filter_pred so shared query ASTs cannot cross-hit across DBs or temporal/filter views. Restore live-fact semantics in the O(n) bulk tx_data path for same-tx card-one re-asserts, guard Int n±1 AVET bounds at min/max_int, and add regression tests covering cache isolation, bulk updates, and range edges. Co-authored-by: Tienson Qin --- docs/design-tx-window-index.md | 3 + impl/datascript.mli | 3 +- impl/query_api.ml | 68 ++++++--- impl/query_exec.ml | 6 +- impl/query_where.ml | 6 +- impl/transact.ml | 22 ++- test/dune | 10 ++ test/test_bulk_card_one.ml | 96 +++++++++++++ test/test_query_result_cache.ml | 238 ++++++++++++++++++++++++++++++++ 9 files changed, 421 insertions(+), 31 deletions(-) create mode 100644 test/test_bulk_card_one.ml create mode 100644 test/test_query_result_cache.ml diff --git a/docs/design-tx-window-index.md b/docs/design-tx-window-index.md index 6e0cbae..53a35d7 100644 --- a/docs/design-tx-window-index.md +++ b/docs/design-tx-window-index.md @@ -236,6 +236,9 @@ to AEVT/AVET + tx filter (correct, slower). ### Implemented (this branch) +Shipped vs design: sections below this subsection (Mode B, TEAV alternatives, +app-level “recent N days” product plans) remain design notes unless listed here. + - `| Tave` index; key order `tx | a | v | e | added` (BLOB KV on LMDB `ds/tave` + SQLite `ds_tave`) - Written on every append alongside EAVT/AEVT/(AVET when indexed) - `fold_tave_range` / `prune_tave_before`; retention default 30 days diff --git a/impl/datascript.mli b/impl/datascript.mli index ec1d49d..6e89d7c 100644 --- a/impl/datascript.mli +++ b/impl/datascript.mli @@ -611,7 +611,8 @@ val last_query_exec_path : unit -> query_exec_path (** Run [f] with fused [Query_exec] disabled so [q] uses the relational fallback. *) val with_force_relation_fallback : (unit -> 'a) -> 'a -(** Datalevin-style query result cache (general; keyed by db epoch + physical query). +(** Datalevin-style query result cache (general; keyed by db uid, temporal/filter + view, epoch, and physical query/inputs). Enabled by default; set [DATASCRIPT_QUERY_RESULT_CACHE=0] to disable. Set [DATASCRIPT_QUERY_DEBUG=1] for plan/exec/cache stderr traces. *) val clear_query_result_cache : unit -> unit diff --git a/impl/query_api.ml b/impl/query_api.ml index bf470c5..88c09fa 100644 --- a/impl/query_api.ml +++ b/impl/query_api.ml @@ -19,8 +19,10 @@ let with_force_relation_fallback f = force_relation_fallback := true; Fun.protect ~finally:(fun () -> force_relation_fallback := previous) f -(* Datalevin-style result cache: general, keyed by db epoch + physical query/inputs. - Avoid structural hashing of [query] — where clauses may embed function values. *) +(* Datalevin-style result cache: keyed by db identity + temporal/filter view + + epoch + physical query/inputs. Avoid structural hashing of [query] — where + clauses may embed function values. [filter_pred] is compared by physical + identity (closures are not structurally comparable). *) let result_cache_enabled = match Sys.getenv_opt "DATASCRIPT_QUERY_RESULT_CACHE" with | Some ("0" | "false" | "no" | "off") -> ref false @@ -32,12 +34,17 @@ let query_debug_enabled = | _ -> false type result_cache_entry = - { max_e : entity_id - ; max_tx : int - ; query : query - ; inputs : query_arg list - ; path : query_exec_path - ; rows : query_result list list + { cache_db_uid : int + ; cache_max_e : entity_id + ; cache_max_tx : int + ; cache_as_of_tx : tx option + ; cache_since_tx : tx option + ; cache_history : bool + ; cache_filter_pred : (datom -> bool) option + ; cache_query : query + ; cache_inputs : query_arg list + ; cache_path : query_exec_path + ; cache_rows : query_result list list } let result_cache_entries : result_cache_entry list ref = ref [] @@ -65,33 +72,52 @@ let path_label = function | Relation_fallback -> "relation" | Binding_interpreter -> "bindings" +let same_filter_pred left right = + match left, right with + | None, None -> true + | Some a, Some b -> a == b + | _ -> false + +let same_result_cache_key entry db query inputs = + entry.cache_db_uid = db.db_uid + && entry.cache_max_e = db.max_datom_e + && entry.cache_max_tx = db.max_tx + && entry.cache_as_of_tx = db.as_of_tx + && entry.cache_since_tx = db.since_tx + && entry.cache_history = db.history + && same_filter_pred entry.cache_filter_pred db.filter_pred + && entry.cache_query == query + && entry.cache_inputs == inputs + let lookup_result_cache db query inputs = if (not !result_cache_enabled) || !force_relation_fallback then None else List.find_map (fun entry -> - if - entry.max_e = db.max_datom_e - && entry.max_tx = db.max_tx - && entry.query == query - && entry.inputs == inputs - then Some (entry.path, entry.rows) + if same_result_cache_key entry db query inputs then + Some (entry.cache_path, entry.cache_rows) else None) !result_cache_entries let store_result_cache db query inputs path rows = if !result_cache_enabled && not !force_relation_fallback then ( let entry = - { max_e = db.max_datom_e; max_tx = db.max_tx; query; inputs; path; rows } + { cache_db_uid = db.db_uid + ; cache_max_e = db.max_datom_e + ; cache_max_tx = db.max_tx + ; cache_as_of_tx = db.as_of_tx + ; cache_since_tx = db.since_tx + ; cache_history = db.history + ; cache_filter_pred = db.filter_pred + ; cache_query = query + ; cache_inputs = inputs + ; cache_path = path + ; cache_rows = rows + } in let rest = List.filter - (fun e -> - not - (e.query == query - && e.inputs == inputs - && e.max_e = entry.max_e - && e.max_tx = entry.max_tx)) + (fun e -> not (same_result_cache_key e db query inputs)) !result_cache_entries in let rec take n = function diff --git a/impl/query_exec.ml b/impl/query_exec.ml index 260c8d2..d50ed62 100644 --- a/impl/query_exec.ml +++ b/impl/query_exec.ml @@ -746,7 +746,7 @@ end) = struct let avet_index_start predicate threshold = match predicate, threshold with - | GreaterThan, Int n -> Some (Int (n + 1)) + | GreaterThan, Int n when n < max_int -> Some (Int (n + 1)) | GreaterOrEqual, value | GreaterThan, value -> Some value | _ -> None @@ -812,7 +812,9 @@ end) = struct (function | ComparisonPredicate (predicate, left, right) -> ( match range_predicate_for_var value_var predicate left right with - | Some (GreaterThan, Int _) | Some (LessThan, Int _) -> false + (* Tightened Int bounds (n±1) are exact only when they do not overflow. *) + | Some (GreaterThan, Int n) when n < max_int -> false + | Some (LessThan, Int n) when n > min_int -> false | Some _ -> true | None -> true) | _ -> false) diff --git a/impl/query_where.ml b/impl/query_where.ml index c2402f0..c7d81f3 100644 --- a/impl/query_where.ml +++ b/impl/query_where.ml @@ -1063,7 +1063,7 @@ end) = struct let avet_index_start predicate threshold = match predicate, threshold with - | GreaterThan, Int n -> Some (Int (n + 1)) + | GreaterThan, Int n when n < max_int -> Some (Int (n + 1)) | GreaterOrEqual, value | GreaterThan, value -> Some value | _ -> None @@ -1078,7 +1078,9 @@ end) = struct (function | ComparisonPredicate (predicate, left, right) -> ( match range_predicate_for_var value_var predicate left right with - | Some (GreaterThan, Int _) | Some (LessThan, Int _) -> false + (* Tightened Int bounds (n±1) are exact only when they do not overflow. *) + | Some (GreaterThan, Int n) when n < max_int -> false + | Some (LessThan, Int n) when n > min_int -> false | Some _ -> true | None -> true) | _ -> false) diff --git a/impl/transact.ml b/impl/transact.ml index aa73fad..301d51d 100644 --- a/impl/transact.ml +++ b/impl/transact.ml @@ -1479,7 +1479,8 @@ let apply_tx context tx_ops db = else (* Build tx_data in O(n): [acc @ pieces] was O(n²) and dominated Share SQLite bulk loads (20k entities ≈ 100k facts). Index same-tx (e,a) - facts for card-one retraction without scanning the full acc list. *) + pieces (newest-first) and mirror [existing_attr_datoms] semantics: + from_db excludes in-tx retracts; from_acc is live added facts only. *) let tx_data = let by_ea = Hashtbl.create (List.length facts) in let acc_rev = @@ -1488,11 +1489,22 @@ let apply_tx context tx_ops db = let d = { fact with v = context.resolve_context.normalize_value fact.v } in - let from_db = context.existing_entity_attr_datoms db d.e d.a in + let same_tx_pieces = + Option.value (Hashtbl.find_opt by_ea (d.e, d.a)) ~default:[] + in + let retracted_in_tx ex = + List.exists + (fun pd -> not pd.added && context.same_fact pd ex) + same_tx_pieces + in + let from_db = + context.existing_entity_attr_datoms db d.e d.a + |> List.filter (fun ex -> not (retracted_in_tx ex)) + in let from_acc = - match Hashtbl.find_opt by_ea (d.e, d.a) with - | None -> [] - | Some ds -> List.rev ds + same_tx_pieces + |> List.filter (fun pd -> pd.added && not (retracted_in_tx pd)) + |> List.rev in let existing = from_db @ from_acc in let same_fact_exists = List.exists (context.same_fact d) existing in diff --git a/test/dune b/test/dune index 1c44e07..310f831 100644 --- a/test/dune +++ b/test/dune @@ -49,6 +49,16 @@ (modules test_query_exec_parity) (libraries datascript-ocaml-native test_support alcotest)) +(test + (name test_query_result_cache) + (modules test_query_result_cache) + (libraries datascript-ocaml-native test_support alcotest)) + +(test + (name test_bulk_card_one) + (modules test_bulk_card_one) + (libraries datascript-ocaml-native test_support alcotest)) + (test (name test_shared_queries) (modules test_shared_queries) diff --git a/test/test_bulk_card_one.ml b/test/test_bulk_card_one.ml new file mode 100644 index 0000000..a990833 --- /dev/null +++ b/test/test_bulk_card_one.ml @@ -0,0 +1,96 @@ +(** Bulk card-one tx_data: same-tx re-assert after update must keep the final value. *) + +open Alcotest +open Datascript +open Test_alcotest_support + +let indexed = + { cardinality = One + ; unique = None + ; indexed = true + ; is_component = false + ; no_history = false + ; doc = None + ; value_type = None + ; tuple_attrs = None + ; tuple_types = None + } + +let int_ages db e = + datoms db Eavt ~e ~a:"age" () + |> List.of_seq + |> List.map (fun d -> match d.v with Int n -> n | _ -> -1) + |> List.sort compare + +let test_bulk_same_tx_reassert_original_after_update () = + (* Regression for O(n) by_ea path: from_db must ignore in-tx retracts, and + from_acc must only include live added facts. Otherwise age 20→30→20 drops + the final assert because from_db still looks like age=20 is live. *) + let db = + empty_db ~schema:[ "age", indexed; "name", indexed ] () + |> db_with + [ Add (Entity_id 1, "name", String "Alice") + ; Add (Entity_id 1, "age", Int 20) + ] + in + let report = + transact db + [ Add (Entity_id 1, "age", Int 30); Add (Entity_id 1, "age", Int 20) ] + in + check_int_list "live age is final re-assert 20" [ 20 ] (int_ages report.db_after 1); + let asserted_ages = + report.tx_data + |> List.filter (fun d -> d.a = "age" && d.added) + |> List.map (fun d -> match d.v with Int n -> n | _ -> -1) + in + check_int_list "tx_data asserts both intermediate and final ages" [ 30; 20 ] asserted_ages; + let final_assert_present = + List.exists + (fun d -> d.a = "age" && d.added && d.v = Int 20) + report.tx_data + in + check_bool "final age=20 assert present in tx_data" true final_assert_present + +let test_bulk_same_tx_multiple_updates () = + let db = + empty_db ~schema:[ "age", indexed ] () + |> db_with [ Add (Entity_id 1, "age", Int 1) ] + in + let report = + transact db + [ Add (Entity_id 1, "age", Int 2) + ; Add (Entity_id 1, "age", Int 3) + ; Add (Entity_id 1, "age", Int 4) + ] + in + check_int_list "live age is last write" [ 4 ] (int_ages report.db_after 1) + +let test_bulk_new_entity_card_one_updates () = + (* Bulk path with tempids: same-tx card-one updates on a newly allocated entity. *) + let report = + transact + (empty_db ~schema:[ "age", indexed; "name", indexed ] ()) + [ Entity + { db_id = Some (Temp_id "a") + ; attrs = + [ "name", One_value (String "Ada") + ; "age", One_value (Int 10) + ] + } + ; Add (Temp_id "a", "age", Int 11) + ; Add (Temp_id "a", "age", Int 10) + ] + in + match List.assoc_opt "a" report.tempids with + | None -> fail "tempid a should resolve" + | Some e -> check_int_list "new entity ends at age 10" [ 10 ] (int_ages report.db_after e) + +let () = + run "bulk_card_one" + [ ( "same_tx" + , [ test_case "reassert original after update" `Quick + test_bulk_same_tx_reassert_original_after_update + ; test_case "multiple updates keep last" `Quick test_bulk_same_tx_multiple_updates + ; test_case "new entity tempid updates" `Quick test_bulk_new_entity_card_one_updates + ] ) + ] diff --git a/test/test_query_result_cache.ml b/test/test_query_result_cache.ml new file mode 100644 index 0000000..796794b --- /dev/null +++ b/test/test_query_result_cache.ml @@ -0,0 +1,238 @@ +(** Query result cache: key must distinguish db identity and temporal/filter views. *) + +open Alcotest +open Datascript +open Test_alcotest_support + +let indexed = + { cardinality = One + ; unique = None + ; indexed = true + ; is_component = false + ; no_history = false + ; doc = None + ; value_type = None + ; tuple_attrs = None + ; tuple_types = None + } + +let schema = [ "name", indexed; "age", indexed ] + +let cell_digest = function + | Result_entity e -> "e:" ^ string_of_int e + | Result_value (String s) -> "s:" ^ s + | Result_value (Int i) -> "i:" ^ string_of_int i + | Result_value (Keyword k) -> "k:" ^ k + | _ -> "?" + +let rows_digest rows = + rows + |> List.map (fun row -> String.concat "," (List.map cell_digest row)) + |> List.sort String.compare + |> String.concat "|" + +let find_name_age = + parse_query_string + {|[:find ?e ?name ?age :where [?e :name ?name] [?e :age ?age]]|} + +let find_name = + parse_query_string {|[:find ?e ?name :where [?e :name ?name]]|} + +let with_cache_on f = + clear_query_result_cache (); + with_query_result_cache true f + +let db_alice_bob () = + empty_db ~schema () + |> db_with + [ Add (Entity_id 1, "name", String "Alice") + ; Add (Entity_id 1, "age", Int 25) + ; Add (Entity_id 2, "name", String "Bob") + ; Add (Entity_id 2, "age", Int 35) + ] + +let db_carol () = + empty_db ~schema () + |> db_with + [ Add (Entity_id 1, "name", String "Carol") + ; Add (Entity_id 1, "age", Int 40) + ] + +let test_cache_hit_same_db () = + with_cache_on (fun () -> + let db = db_alice_bob () in + let first = q db find_name_age in + check_bool "cold miss" false (last_query_cache_hit ()); + let second = q db find_name_age in + check_bool "warm hit" true (last_query_cache_hit ()); + check string "same rows" (rows_digest first) (rows_digest second)) + +let test_cache_separates_distinct_dbs () = + with_cache_on (fun () -> + let db_a = db_alice_bob () in + let db_b = db_carol () in + (* Same physical query AST + matching max_e/max_tx shapes must not cross-hit. *) + let dig_a = rows_digest (q db_a find_name_age) in + check_bool "db_a cold" false (last_query_cache_hit ()); + let dig_b = rows_digest (q db_b find_name_age) in + check_bool "db_b must miss despite shared query AST" false (last_query_cache_hit ()); + check string "db_a digest" "e:1,s:Alice,i:25|e:2,s:Bob,i:35" dig_a; + check string "db_b digest" "e:1,s:Carol,i:40" dig_b; + let dig_a_again = rows_digest (q db_a find_name_age) in + check_bool "db_a warm hit" true (last_query_cache_hit ()); + check string "db_a still Alice/Bob" dig_a dig_a_again) + +let test_cache_separates_as_of () = + with_cache_on (fun () -> + let db0 = db_alice_bob () in + let tx0 = basis_tx db0 in + let db1 = db_with [ Add (Entity_id 1, "age", Int 26) ] db0 in + let past = as_of tx0 db1 in + let dig_current = rows_digest (q db1 find_name_age) in + let dig_past = rows_digest (q past find_name_age) in + check_bool "as_of must miss after current" false (last_query_cache_hit ()); + check string "current age 26" "e:1,s:Alice,i:26|e:2,s:Bob,i:35" dig_current; + check string "as_of age 25" "e:1,s:Alice,i:25|e:2,s:Bob,i:35" dig_past; + ignore (q db1 find_name_age); + check_bool "current warm" true (last_query_cache_hit ()); + ignore (q past find_name_age); + check_bool "as_of warm" true (last_query_cache_hit ())) + +let test_cache_separates_since () = + with_cache_on (fun () -> + let db0 = + empty_db ~schema () + |> db_with [ Add (Entity_id 1, "name", String "Alice") ] + in + let tx0 = basis_tx db0 in + let db1 = + db_with + [ Add (Entity_id 2, "name", String "Bob") + ; Add (Entity_id 2, "age", Int 30) + ] + db0 + in + let delta = since tx0 db1 in + let dig_full = rows_digest (q db1 find_name) in + let dig_since = rows_digest (q delta find_name) in + check_bool "since must miss after full db" false (last_query_cache_hit ()); + check string "full names" "e:1,s:Alice|e:2,s:Bob" dig_full; + check string "since only Bob" "e:2,s:Bob" dig_since) + +let test_cache_separates_history () = + with_cache_on (fun () -> + let db = + empty_db ~schema () + |> db_with [ Add (Entity_id 1, "name", String "Alice"); Add (Entity_id 1, "age", Int 20) ] + |> db_with [ Add (Entity_id 1, "age", Int 30) ] + in + let hist = history db in + let dig_current = rows_digest (q db find_name_age) in + let dig_hist = rows_digest (q hist find_name_age) in + check_bool "history must miss after current" false (last_query_cache_hit ()); + check string "current only live age" "e:1,s:Alice,i:30" dig_current; + (* History view must not reuse the current-view cache entry. *) + check_int "history digest differs or is at least as long" 1 + (if dig_hist <> dig_current || String.length dig_hist >= String.length dig_current + then 1 + else 0); + ignore (q db find_name_age); + check_bool "current warm" true (last_query_cache_hit ()); + ignore (q hist find_name_age); + check_bool "history warm" true (last_query_cache_hit ())) + +let test_cache_separates_filter () = + with_cache_on (fun () -> + let db = db_alice_bob () in + let only_e1 = filter db (fun _ d -> d.e = 1) in + let only_e2 = filter db (fun _ d -> d.e = 2) in + let dig_full = rows_digest (q db find_name) in + let dig_e1 = rows_digest (q only_e1 find_name) in + check_bool "filter e1 must miss after full" false (last_query_cache_hit ()); + let dig_e2 = rows_digest (q only_e2 find_name) in + check_bool "filter e2 must miss after filter e1" false (last_query_cache_hit ()); + check string "full" "e:1,s:Alice|e:2,s:Bob" dig_full; + check string "filtered e=1" "e:1,s:Alice" dig_e1; + check string "filtered e=2" "e:2,s:Bob" dig_e2; + ignore (q only_e1 find_name); + check_bool "filter e1 warm" true (last_query_cache_hit ())) + +let test_cache_toggle_and_clear () = + with_cache_on (fun () -> + let db = db_alice_bob () in + ignore (q db find_name); + ignore (q db find_name); + check_bool "hit before clear" true (last_query_cache_hit ()); + clear_query_result_cache (); + ignore (q db find_name); + check_bool "miss after clear" false (last_query_cache_hit ()); + with_query_result_cache false (fun () -> + ignore (q db find_name); + check_bool "disabled never hits" false (last_query_cache_hit ()); + ignore (q db find_name); + check_bool "disabled second call" false (last_query_cache_hit ())); + ignore (q db find_name); + check_bool "re-enabled can hit" true (last_query_cache_hit ())) + +let test_int_range_at_max_int_does_not_overflow () = + with_cache_on (fun () -> + let db = + empty_db ~schema:[ "age", indexed ] () + |> db_with + [ Add (Entity_id 1, "age", Int max_int) + ; Add (Entity_id 2, "age", Int (max_int - 1)) + ] + in + let q_gt = + parse_query_string + {|[:find ?e :where [?e :age ?age] [(> ?age ?t)]]|} + in + let dig = + rows_digest + (q ~inputs:[ Arg_scalar (Result_value (Int (max_int - 1))) ] db q_gt) + in + check string "> max_int-1 finds only max_int" "e:1" dig; + let dig_none = + rows_digest (q ~inputs:[ Arg_scalar (Result_value (Int max_int)) ] db q_gt) + in + check string "> max_int finds nothing" "" dig_none) + +let test_int_range_at_min_int_does_not_overflow () = + with_cache_on (fun () -> + let db = + empty_db ~schema:[ "age", indexed ] () + |> db_with + [ Add (Entity_id 1, "age", Int min_int) + ; Add (Entity_id 2, "age", Int (min_int + 1)) + ] + in + let q_lt = + parse_query_string + {|[:find ?e :where [?e :age ?age] [(< ?age ?t)]]|} + in + let dig = + rows_digest + (q ~inputs:[ Arg_scalar (Result_value (Int (min_int + 1))) ] db q_lt) + in + check string "< min_int+1 finds only min_int" "e:1" dig; + let dig_none = + rows_digest (q ~inputs:[ Arg_scalar (Result_value (Int min_int)) ] db q_lt) + in + check string "< min_int finds nothing" "" dig_none) + +let () = + run "query_result_cache" + [ ( "cache" + , [ test_case "hit same db" `Quick test_cache_hit_same_db + ; test_case "separate distinct dbs" `Quick test_cache_separates_distinct_dbs + ; test_case "separate as_of" `Quick test_cache_separates_as_of + ; test_case "separate since" `Quick test_cache_separates_since + ; test_case "separate history" `Quick test_cache_separates_history + ; test_case "separate filter" `Quick test_cache_separates_filter + ; test_case "toggle and clear" `Quick test_cache_toggle_and_clear + ] ) + ; ( "avet_int_bounds" + , [ test_case "max_int" `Quick test_int_range_at_max_int_does_not_overflow + ; test_case "min_int" `Quick test_int_range_at_min_int_does_not_overflow + ] ) + ]