diff --git a/AGENTS.md b/AGENTS.md index 141a991..5ea0bda 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,5 +1,6 @@ - Code and comments should be written in English. - Solve root causes, not workarounds. +- When debugging a problem, do not guess and patch blindly. Add targeted debug logging (or other runtime evidence), identify the root cause, then implement the fix. - Prefer simple implementations over complex ones. - All observable behavior should match upstream DataScript. - Implementation details should match upstream DataScript unless a divergence is explicitly requested and documented. diff --git a/docs/datahike-ocaml-query-comparison.md b/docs/datahike-ocaml-query-comparison.md new file mode 100644 index 0000000..66eef54 --- /dev/null +++ b/docs/datahike-ocaml-query-comparison.md @@ -0,0 +1,287 @@ +# Datahike vs OCaml Query Implementation Comparison + +Reference clone: `_deps/datahike` (replikativ/datahike, shallow clone for local diff). +Upstream doc: `_deps/datahike/doc/query-engine.md`. + +Observable **results** must stay DataScript-compatible. **Execution architecture** +should follow Datahike's compiled planner + permanent relational fallback. + +## Pipeline mapping + +| Phase | Datahike | OCaml (this repo) | Gap | +| --- | --- | --- | --- | +| Entry | `datahike/query.cljc` → `q` / `execute-planned-direct` | `datascript.ml` → `Query_impl.q` → `query_api.ml` `q_sources_raw` | OK (no `simple_*` on `q`) | +| Classify | `query/analyze.cljc` `classify-clause` | Inline in `query_plan.ml` / `query_where.ml` pattern parsing | No dedicated analyze module | +| Logical IR | `query/logical.cljc` `build-logical-plan` | `query_plan.ml` `build_logical_plan` | Same node shapes (`LEntityJoin`, `LScan`, …) | +| Lower | `query/lower.cljc` + `query/plan.cljc` | `query_plan.ml` `lower` / `compile` | **Major**: DH uses DP merge + pipeline DSL; OCaml flattens to clause list | +| Execute | `query/execute.cljc` fused scan+merge, probe-map joins | `query_exec.ml` drive-scan + AEVT seek/dense merge + anti-merge | Card-one entity groups implemented; multi-group probe joins still fallback | +| Fallback | `query/relation.cljc` + `query.cljc` `execute-legacy` | `query_where.ml` relation interpreter | Permanent fallback — correct role | +| Project | find projection in execute / query | `query_api.ml` `relation_rows_for_find` | OK | + +Datahike end-to-end: + +``` +analyze → logical.cljc → lower.cljc → execute.cljc → find project + ↳ ineligible → relation.cljc (legacy) +``` + +OCaml today: + +``` +query_plan.compile → OpEntityGroup/OpScan (ground) / EntityGroup+filters + ↳ query_exec resolved kernels (q2 / q-5-merge) + drive/merge/anti + ↳ else query_where relational fallback (multi-op Union, open scans, …) +``` + +The planner IR matches Datahike. Entity-group **execute** now follows +`execute-group-direct` / sorted-merge / anti-merge semantics (AEVT +forward-seek or dense index ≈ seekGE), not Datascript `simple_*` gates. +Resolved kernels are cached by entity-group physical identity (plan cache +reuses the same group object) to avoid re-matching merges on every call. +Multi-op Union / open-pattern scans stay on the relational fallback until +probe-join execute is competitive. + +## Module-by-module notes + +### `analyze.cljc` (Datahike) + +- Classifies each clause: `:pattern`, `:predicate`, `:function`, `:not`, `:or`, … +- Extracts vars, checks fn args, handles quote forms. +- **OCaml**: scattered across `Query.pattern_scan`, `query_plan.pattern_scan`, `query_where` clause walks. No single classify API. + +### `logical.cljc` (Datahike) + +Key behaviors (see `build-logical-plan`): + +1. Classify all clauses → `LScan` / `LFilter` / `LBind` / … +2. Group scans by `[entity-var, source]` → `LEntityJoin` +3. **Foldable NOT** (`foldable-not?`): single-pattern NOT on grouped entity, non-entity vars local to negation → **anti-scan inside entity group** +4. Remaining NOT → `LAntiJoin` +5. OR / rules → `LUnion` / `LRuleCall` / `LFixpoint` + +**OCaml** (`query_plan.ml` `build_logical_plan`): + +- Same grouping and foldable-NOT idea (`foldable_not_scan`). +- Extra constraint: fold only if positive scan **earlier in source order** (DataScript outer-binding errors). +- Does **not** tag nodes with `:source-idx` for bound-var-card propagation (Datahike lower uses this). + +### `plan.cljc` + `lower.cljc` (Datahike) + +Physical planning primitives: + +| Primitive | Purpose | +| --- | --- | +| `plan-pattern-op` | Index choice (EAVT/AEVT/AVET) + pushdown bounds | +| `dp-order-fuse-ops` | Optimal scan + merge order within entity group | +| `assemble-entity-group` | `:entity-group` op + `build-pipeline` | +| `detect-inter-group-joins` | Shared value vars → hash-probe plan | +| `dp-order-groups` / `order-plan-ops` | Inter-group order + readiness | + +Lower produces ops like: + +```clojure +{:op :entity-group + :scan-op {... :index :aevt ...} + :merge-ops [{:join-method :lookup ...} ...] + :pipeline {:path :sorted-merge :steps [...]}} +``` + +**OCaml** (`query_plan.ml`): + +- `OpEntityGroup { clauses; estimated_rows }` — **only clause list**, no scan/merge split, no pipeline. +- `lower` schedules ops by heuristic cost; `clauses_of_plan` **discards physical structure**. +- Index choice exists (`choose_index`) but is not consumed by a fused executor. + +### `execute.cljc` (Datahike) + +Core execution paths: + +1. **`execute-group-direct`** — entity group fused scan: + - Pick driving scan (lowest cardinality after DP) + - Walk index slice; for each datom, **seekGE** merge lookups (no intermediate relations) + - Paths: `:scan-only`, `:sorted-merge`, `:per-cursor-merge`, `:card-many-merge` +2. **Anti-merge** — during merge loop, skip entities matching anti-scan attr/value +3. **Multi-group** — producer probe-set / probe-map → consumer filtered scan +4. **Post-filter / post-apply** — wide tuples then project to find-vars + +**OCaml** (`query_where.ml`): + +- `relation_of_same_entity_patterns` — materializes `{ attrs; rows }` lists +- Dense AEVT gather (`try_same_entity_constant_dense_rows`, `try_fast_empty_relation_rows`) — **ad hoc**, not driven by `OpEntityGroup` / pipeline +- `hash_join` on relations — correct fallback shape, not cursor merge +- NOT: bitset exclusion scan OR `anti_join` on relations + +### `relation.cljc` (Datahike fallback) + +- Tuple relations, `hash-join`, `sum-rel`, `subtract-rel` +- Used when planner ineligible or `*disable-planner*` + +**OCaml**: same concepts in `query_where.ml` (`hash_join`, `anti_join`, `union_relations`). + +## Shared bench queries — shape-by-shape + +Queries from `bench/shared_query_bench.ml`. + +### q1 — `[:find ?e :where [?e :name "Ivan"]]` + +| | Datahike | OCaml | +| --- | --- | --- | +| Logical | `LScan` (ground value → AVET) | `LScan` → `OpScan` or single-pattern group | +| Execute | AVET slice or EAVT seek; **no relation alloc** | AVET ids or AEVT scan → relation rows | +| Gap | Direct emit to result set | Extra `{attrs;rows}` wrapper | + +### q2 — `[:find ?e ?a :where [?e :name "Ivan"] [?e :age ?a]]` + +| | Datahike | OCaml | +| --- | --- | --- | +| Logical | `LEntityJoin` with 2 scans | Same | +| Lower | `assemble-entity-group`: DP picks scan (`:name` selective) + merge `:age` via **lookupGE** | `OpEntityGroup` → flat clauses → `try_fast_*` or `relation_of_same_entity_patterns` | +| Execute | **Fused sorted-merge** — one pass, no hash join | Dense AEVT index gather OR hash_join two relations | +| Perf | ~0.6 ms (20k entities, DH bench doc) | ~0.009 ms (2k entities) vs **0.004 ms** pre-removal gate | + +Root cause of OCaml gap: execution still **materializes row lists** and duplicates kernel logic outside the planner op stream. + +### q-5-merge — five attrs + `[?e :sex :male]` + +| | Datahike | OCaml | +| --- | --- | --- | +| Logical | `LEntityJoin` 5 scans + constant on `:sex` | Same | +| Execute | DP order: selective constant/attr as scan, merges via cursor | Const-first aligned AEVT gather (4 value vars) | +| DH doc | "5-clause entity merge" **2.4 ms** @ 20k | **0.046 ms** @ 2k vs **0.030 ms** baseline | + +Datahike uses **merge ordering + seekGE**, not "all arrays aligned then index by entity id". + +### q-not / q-not-join — `[?e :age ?a] (not [?e :sex :male])` + +| | Datahike | OCaml | +| --- | --- | --- | +| Logical | Foldable NOT → **anti-scan** inside `LEntityJoin` on `?e` | Same fold in `build_logical_plan` | +| Execute | Anti-merge during fused scan (skip excluded entities) | `try_not_single_value_aevt_scan` / bitset + full AEVT walk | +| Planner | NOT present → still plans positive leg | **`plan_ordered_clauses` skips compile when any NOT** — source order only | +| DH doc | NOT **3.8 ms** @ 20k | **0.025 ms** @ 2k vs **0.023 ms** baseline | + +OCaml NOT path never uses planner ordering; anti-scan is reimplemented in fallback, not as merge op. + +### q-or-join, q-rule + +| Query | Datahike | OCaml | +| --- | --- | --- | +| q-or-join | `LUnion` → branch execute → combine | `eval_or_branch_relations` / union | +| q-rule | `LRuleCall` → expand → plan body | `try_single_pattern_rule_rows` + inline rules | + +## What is wrong with current `query_where.ml` complexity + +These are **execute-layer** concerns implemented inside the **fallback module**: + +| Mechanism | Lines (approx) | Datahike equivalent | +| --- | --- | --- | +| `try_fast_empty_relation_rows` | ~350 | Should not exist — `execute.cljc` `execute-group-direct` | +| `try_same_entity_constant_dense_rows` | ~150 | `assemble-entity-group` + `execute-sorted-merge` | +| `rows_from_dense_aevt_gather` | ~200 | Pipeline `PIndexScan` → `PSortedMerge` → `PEmitTuple` | +| `same_entity_fused_relation` | wrapper | `OpEntityGroup` execution | +| `relation_of_same_entity_patterns` | ~1300 | Split: lower produces ops, execute consumes ops | + +Adding more special cases in `query_where` **diverges further** from Datahike. The alignment doc (`docs/datahike-query-alignment.md` P3–P4) already says physical ops should drive execution. + +## Recommended refactor (Datahike-faithful) + +### 1. Add `impl/query_exec.ml` (execute layer) + +```ocaml +val run : + db -> physical_plan -> query_source -> bindings -> + (string list * query_result list list * bool) option +``` + +Implement op dispatch matching Datahike: + +- `OpEntityGroup` → fused entity-group execute (port `execute-group-direct` / `execute-sorted-merge` using existing `aevt_attr_array`, `entity_ids_array_by_attr_value`, index seeks) +- `OpScan` → single pattern scan +- `OpUnion` → `union_relations` +- `OpAntiJoin` → `anti_join` +- `OpFilter` → filter relation or in-group attached pred +- `OpPassthrough` → return `None` (fallback) + +Move dense gather / bitset NOT / aligned multi-attr logic **into** entity-group execute keyed by pipeline path — delete `try_fast_*`. + +### 2. Extend physical IR (minimal) + +Extend `OpEntityGroup` to carry what lower already knows: + +```ocaml +| OpEntityGroup of { + entity_var : string; + scan : l_scan; (* driving pattern *) + merges : l_scan list; (* DP-ordered *) + anti_scans : l_scan list; + filters : query_clause list; + index : index_choice; + ... + } +``` + +Stop flattening to `clauses` in `clauses_of_plan` for execution (keep flatten for tests/explain only). + +### 3. Wire entry (`query_api.ml`) + +```ocaml +match Query_plan.compile db.max_datom_e [] [] where with +| Some plan when Query_plan.plan_is_executable plan -> + (match Query_exec.run db plan default_source bindings with + | Some result -> ... + | None -> fallback interpreter) +| None -> fallback interpreter +``` + +Remove `try_fast_empty_relation_rows` bypass from `eval_relation_rows`. + +### 4. Keep `query_where.ml` as fallback only + +- `eval_relation_from_empty` / binding interpreter +- `hash_join`, `anti_join`, `union_relations`, `relation_of_pattern` +- Source-order NOT for DataScript error parity +- **No** bench-shaped dense kernels at module top level + +### 5. Port planning primitives incrementally + +Priority for bench perf: + +1. `dp-order-fuse-ops` (scan + merge order within group) — `plan.cljc:531` +2. `assemble-entity-group` + `build-pipeline` — `plan.cljc:831`, `:617` +3. `execute-sorted-merge` — `execute.cljc:1375` (card-one attrs, dense DBs) +4. Anti-merge in merge loop — NOT as separate full-DB bitset scan +5. Count-slice estimates — `estimate.cljc` (replace `max_e/8` heuristics) + +## File reference index (Datahike) + +| File | LOC (approx) | Read first | +| --- | --- | --- | +| `doc/query-engine.md` | 523 | Architecture overview | +| `src/datahike/query/ir.cljc` | 172 | IR + pipeline record defs | +| `src/datahike/query/analyze.cljc` | large | Clause classification | +| `src/datahike/query/logical.cljc` | 453 | `build-logical-plan`, NOT fold | +| `src/datahike/query/plan.cljc` | 1860 | DP merge, entity group, ordering | +| `src/datahike/query/lower.cljc` | medium | Logical → physical | +| `src/datahike/query/execute.cljc` | 6500+ | Fused scan, probe joins | +| `src/datahike/query/relation.cljc` | 300 | Fallback relations | +| `src/datahike/query.cljc` | 5300+ | Entry, planner eligibility | + +## Immediate action items + +1. **Stop expanding** `try_fast_empty_relation_rows` / `relation_of_same_entity_patterns` special cases. +2. **Implement** `query_exec.ml` with `OpEntityGroup` fused path for q1/q2/q-5-merge shapes. +3. **Extend** `query_plan.ml` `OpEntityGroup` to retain scan/merge structure (mirror `assemble-entity-group`). +4. **Delete** redundant dense kernels once execute path covers bench suite. +5. **Verify**: `test_shared_queries` + `shared_query_bench --size 2000` vs `3547876` baselines. + +## Local clone usage + +```bash +# Already cloned (gitignored) +ls _deps/datahike/src/datahike/query/ + +# Diff logical IR grouping +diff -u \ + <(rg -n 'LEntityJoin|foldable-not' _deps/datahike/src/datahike/query/logical.cljc) \ + <(rg -n 'LEntityJoin|foldable_not' impl/query_plan.ml) +``` diff --git a/impl/datascript.ml b/impl/datascript.ml index 97f3570..304c3dc 100644 --- a/impl/datascript.ml +++ b/impl/datascript.ml @@ -1494,18 +1494,40 @@ module Query_where_impl = Query_where.Make (struct let normalize_value = normalize_value let datoms_by_attr_value = datoms_by_attr_value let entity_ids_by_attr_value = entity_ids_by_attr_value - let entity_ids_array_by_attr_value = entity_ids_array_by_attr_value let query_attr_uses_avet = query_attr_uses_avet - let query_value_uses_avet = query_value_uses_avet let fold_index_range = fold_index_range let find_entity_attr_value db entity_id attr = match Db.find_primary_aevt_entity_attr db entity_id attr with | None -> None | Some datom -> Some (Query.result_of_ref (Query.result_of_datom_v datom)) let aevt_attr_array = Db.aevt_attr_array + let aevt_duplicate_datoms db attr = + Option.value (Hashtbl.find_opt db.duplicate_aevt_by_attr attr) ~default:[] let find_entity_in_aevt_array = Db.find_entity_in_aevt_array end) +module Query_exec = Query_exec + +module Query_exec_impl = Query_exec.Make (struct + let query_evaluator_context = query_evaluator_context + let query_source_context = query_source_context + let cardinality_one db attr = cardinality db attr = One + let datoms_by_attr_value = datoms_by_attr_value + let entity_ids_by_attr_value = entity_ids_by_attr_value + let entity_ids_array_by_attr_value = entity_ids_array_by_attr_value + let query_attr_uses_avet = query_attr_uses_avet + let query_value_uses_avet = query_value_uses_avet + let aevt_attr_array = Db.aevt_attr_array + let aevt_duplicate_datoms db attr = + Option.value (Hashtbl.find_opt db.duplicate_aevt_by_attr attr) ~default:[] + let find_entity_in_aevt_array = Db.find_entity_in_aevt_array +end) + +let execute_plan db sources rules bindings plan = + match Query_exec_impl.run db sources rules bindings plan with + | None -> None + | Some relation -> Some (relation.attrs, relation.rows, relation.unique_rows) + let eval_clauses = Query_where_impl.eval_clauses let eval_relation_rows = Query_where_impl.eval_relation_rows @@ -1671,6 +1693,7 @@ module Query_api_impl = Query_api.Make (struct let initial_query_context = initial_query_context let eval_clauses = eval_clauses let eval_relation_rows = eval_relation_rows + let execute_plan = execute_plan let has_aggregates = has_aggregates let aggregate_rows = aggregate_rows let aggregate_rows_with = aggregate_rows_with @@ -1786,6 +1809,15 @@ module Query = struct | _ -> false) inputs + let query_debug_enabled = + match Sys.getenv_opt "DATASCRIPT_QUERY_DEBUG" with + | Some ("1" | "true" | "yes") -> true + | _ -> false + + let debug_log msg = + if query_debug_enabled then + Printf.eprintf "[datascript %.3f] %s\n%!" (Unix.gettimeofday ()) msg + let entity_ids_with_attr db attr = let rec collect previous acc = function | [] -> List.rev acc @@ -2255,6 +2287,10 @@ module Query = struct | _ -> None let ref_target_pull_relation db query = + debug_log + (Printf.sprintf "ref_target_pull_relation find=%d where=%d max_e=%d" (List.length query.find) + (List.length query.where) + db.max_datom_e); let wildcard_selector = function | [ Pull_wildcard ] -> Some [ Pull_wildcard ] | _ -> None @@ -2295,11 +2331,13 @@ module Query = struct let required_attrs = List.filter_map (required_pattern source_var) query.where in (match required_attrs with | [ required_attr ] -> + debug_log "ref_target_pull_relation step=source_entities"; let source_entities = Bytes.make (db.max_datom_e + 1) '\000' in entity_ids_with_attr db required_attr |> List.iter (fun entity_id -> if entity_id >= 0 && entity_id < Bytes.length source_entities then Bytes.set source_entities entity_id '\001'); + debug_log "ref_target_pull_relation step=ref_scan"; let source_has_required entity_id = entity_id >= 0 && entity_id < Bytes.length source_entities @@ -2314,14 +2352,32 @@ module Query = struct | _ -> None) |> List.sort_uniq compare in + debug_log (Printf.sprintf "ref_target_pull_relation target_ids=%d" (List.length target_ids)); let rows = - Pull_api_impl.pull_wildcard_many_by_ids pull_api_context db target_ids - |> List.map (fun entity -> [ Result_pull entity ]) + if List.length target_ids <= 512 then ( + debug_log "ref_target_pull_relation per-entity pull"; + target_ids + |> List.filter_map (fun entity_id -> + Pull_api_impl.pull pull_api_context db [ Pull_wildcard ] (Entity_id entity_id) + |> Option.map (fun entity -> [ Result_pull entity ]))) + else ( + debug_log "ref_target_pull_relation wildcard_many_by_ids scan"; + Pull_api_impl.pull_wildcard_many_by_ids pull_api_context db target_ids + |> List.map (fun entity -> [ Result_pull entity ])) in + debug_log + (Printf.sprintf "ref_target_pull_relation HIT targets=%d rows=%d" (List.length target_ids) + (List.length rows)); Some (Query_relation rows) - | _ -> None) - | _ -> None) - | _ -> None + | _ -> + debug_log "ref_target_pull_relation miss required_attrs shape"; + None) + | _ -> + debug_log "ref_target_pull_relation miss missing/ref pattern"; + None) + | _ -> + debug_log "ref_target_pull_relation miss find/rules/inputs guard"; + None let scalar_input_bindings db query inputs = let rec collect acc declarations args = @@ -2840,13 +2896,17 @@ module Query = struct | [] -> None) |> fun values -> Query_collection values))) | Return_relation, None -> + debug_log "q_return Return_relation"; (match simple_attr_entity_pull_collection db query with | Some (Query_collection values) -> Query_relation (List.map (fun value -> [ value ]) values) | Some result -> result | None -> (match ref_target_pull_relation db query with - | Some result -> result + | Some result -> + debug_log "q_return -> ref_target_pull_relation"; + result | None -> + debug_log "q_return -> fallback q()"; let rows = q db query in Query_relation rows)) | Return_relation, Some inputs -> diff --git a/impl/datascript.mli b/impl/datascript.mli index 2877d74..311ac8d 100644 --- a/impl/datascript.mli +++ b/impl/datascript.mli @@ -457,13 +457,19 @@ module Query_plan : sig ; bound_vars : string list } + type entity_group = + { entity_var : string + ; scan : l_scan + ; merges : l_scan list + ; anti_scans : l_scan list + ; filters : query_clause list + ; clauses : query_clause list + ; estimated_rows : int + ; source : string option + } + type physical_op = - | OpEntityGroup of - { entity_var : string - ; clauses : query_clause list - ; estimated_rows : int - ; source : string option - } + | OpEntityGroup of entity_group | OpScan of { clause : query_clause ; index : index_choice diff --git a/impl/query_api.ml b/impl/query_api.ml index 02ea95d..ef7ca72 100644 --- a/impl/query_api.ml +++ b/impl/query_api.ml @@ -24,6 +24,13 @@ module Make (Context : sig bindings list -> query_clause list -> (string list * query_result list list * bool) option + val execute_plan : + db -> + (string * query_source) list -> + query_rule list -> + bindings list -> + Query_plan.physical_plan -> + (string list * query_result list list * bool) option val has_aggregates : find_spec list -> bool val aggregate_rows : ?callables:Query.query_callables -> db -> (string * query_source) list -> bindings list -> find_spec list -> query_result list list val aggregate_rows_with : ?callables:Query.query_callables -> db -> (string * query_source) list -> bindings list -> find_spec list -> string list -> query_result list list @@ -148,50 +155,139 @@ end) = struct |> Option.map (fun key -> key, binding)) |> List.sort_uniq (fun (left, _) (right, _) -> compare left right) |> List.map snd - + + let plan_cache : (int * query_clause list, Query_plan.physical_plan) Hashtbl.t = Hashtbl.create 32 + (* Hot path: cached_query_string reuses the same where list object. *) + let last_plan_where : query_clause list ref = ref [] + let last_plan_max_e = ref (-1) + let last_plan : Query_plan.physical_plan option ref = ref None + + let compile_plan max_datom_e where = + if !last_plan_max_e = max_datom_e && !last_plan_where == where then + !last_plan + else + let key = max_datom_e, where in + match Hashtbl.find_opt plan_cache key with + | Some plan -> + last_plan_where := where; + last_plan_max_e := max_datom_e; + last_plan := Some plan; + Some plan + | None -> + (match Query_plan.compile ~max_datom_e where with + | None -> + last_plan_where := where; + last_plan_max_e := max_datom_e; + last_plan := None; + None + | Some plan -> + Hashtbl.add plan_cache key plan; + last_plan_where := where; + last_plan_max_e := max_datom_e; + last_plan := Some plan; + Some plan) + + (* cached_query_string reuses the same find list object across calls. *) + let last_find : find_spec list ref = ref [] + let last_find_vars : string list option ref = ref None + + let find_var_names_cached find = + if !last_find == find then + !last_find_vars + else ( + last_find := find; + let vars = find_var_names find in + last_find_vars := vars; + vars) + let q_sources_raw ?(inputs = []) db sources query = - let callables, input_bindings, input_rules = initial_query_context db query inputs in - let rules, where = - match query.rules, input_rules with - | [], [] -> [], query.where - | _ -> query_rules_and_where query input_rules + let finish_relation_rows rules input_bindings where find = + let try_planned_execute () = + (* Prefer Datahike execute for single fused entity-group / ground scan. + Multi-op Union and open scans still use relational fallback until + probe-join / union execute matches those paths. *) + if input_bindings <> [ [] ] then + None + else + let plan = + match rules with + | [] -> compile_plan db.max_datom_e where + | rules -> Query_plan.compile ~max_datom_e:db.max_datom_e ~rules where + in + match plan with + | Some plan when Query_plan.plan_is_fused_execute plan -> ( + match plan.ops with + | [ Query_plan.OpScan { clause; _ } ] -> ( + (* Only ground AVET-style scans are competitive on the execute path. *) + match Query_plan.pattern_scan clause with + | Some { entity = QVar _; attr = QAttr _; value = QValue _; tx = None; _ } -> + execute_plan db sources [] input_bindings plan + | _ -> None) + | _ -> execute_plan db sources [] input_bindings plan) + | _ -> None + in + let relation_result = + match try_planned_execute () with + | Some result -> Some result + | None -> eval_relation_rows db sources rules input_bindings where + in + match relation_result with + | Some (attrs, rows, unique_rows) -> ( + (* Hot path: find vars already match relation attrs (entity-group emit). *) + match find_var_names_cached find with + | Some find_vars when find_vars = attrs -> + if unique_rows then rows else sort_uniq_presorted compare rows + | _ -> + (match relation_rows_for_find db sources attrs rows unique_rows find with + | Some rows -> rows + | None -> + let bindings = eval_clauses db sources rules input_bindings where in + bindings + |> fun bindings -> dedupe_bindings_for_find bindings find + |> List.filter_map (fun binding -> collect_find_specs db sources binding find) + |> List.sort_uniq compare)) + | None -> + let bindings = eval_clauses db sources rules input_bindings where in + bindings + |> fun bindings -> dedupe_bindings_for_find bindings find + |> List.filter_map (fun binding -> collect_find_specs db sources binding find) + |> List.sort_uniq compare in - let has_aggregates = has_aggregates query.find in if - (not has_aggregates) + inputs = [] + && query.inputs = [] + && query.rules = [] && query.with_vars = [] - && query_callables_empty callables + && not (has_aggregates query.find) then - match eval_relation_rows db sources rules input_bindings where with - | Some (attrs, rows, unique_rows) -> - (match relation_rows_for_find db sources attrs rows unique_rows query.find with - | Some rows -> rows - | None -> - let bindings = eval_clauses ~callables db sources rules input_bindings where in - bindings - |> fun bindings -> dedupe_bindings_for_find bindings query.find - |> List.filter_map (fun binding -> collect_find_specs db sources binding query.find) - |> List.sort_uniq compare) - | None -> + finish_relation_rows [] [ [] ] query.where query.find + else + let callables, input_bindings, input_rules = initial_query_context db query inputs in + let rules, where = + match query.rules, input_rules with + | [], [] -> [], query.where + | _ -> query_rules_and_where query input_rules + in + if + (not (has_aggregates query.find)) + && query.with_vars = [] + && query_callables_empty callables + then + finish_relation_rows rules input_bindings where query.find + else ( let bindings = eval_clauses ~callables db sources rules input_bindings where in - bindings - |> fun bindings -> dedupe_bindings_for_find bindings query.find - |> List.filter_map (fun binding -> collect_find_specs db sources binding query.find) - |> List.sort_uniq compare - else ( - let bindings = eval_clauses ~callables db sources rules input_bindings where in - if has_aggregates then - if query.with_vars = [] then - aggregate_rows ~callables db sources bindings query.find - else - aggregate_rows_with ~callables db sources bindings query.find query.with_vars - else if query.with_vars <> [] then - non_aggregate_rows_with db sources bindings query.find query.with_vars - else - bindings - |> fun bindings -> dedupe_bindings_for_find bindings query.find - |> List.filter_map (fun binding -> collect_find_specs db sources binding query.find) - |> List.sort_uniq compare) + if has_aggregates query.find then + if query.with_vars = [] then + aggregate_rows ~callables db sources bindings query.find + else + aggregate_rows_with ~callables db sources bindings query.find query.with_vars + else if query.with_vars <> [] then + non_aggregate_rows_with db sources bindings query.find query.with_vars + else + bindings + |> fun bindings -> dedupe_bindings_for_find bindings query.find + |> List.filter_map (fun binding -> collect_find_specs db sources binding query.find) + |> List.sort_uniq compare) let q_with_raw ?(inputs = []) db with_vars query = let callables, input_bindings, input_rules = initial_query_context db query inputs in diff --git a/impl/query_exec.ml b/impl/query_exec.ml new file mode 100644 index 0000000..1e66529 --- /dev/null +++ b/impl/query_exec.ml @@ -0,0 +1,1040 @@ +(** Datahike-aligned query execute layer: run compiled physical ops. + + Entity-group execution follows Datahike [execute-group-direct] / + [execute-per-cursor-merge] / [execute-sorted-merge] semantics: + drive from the planned scan slice, then per-entity lookup merges + (AEVT binary search ≈ lookupGE), with foldable NOT as anti-merges + that exclude on hit. Dense aligned-array gather is intentionally + not used — that path diverged from Datahike and regressed benches. *) + +open Datascript_types + +[@@@ocaml.warning "-67"] + +type bindings = (string * query_result) list + +type relation = + { attrs : string list + ; rows : query_result list list + ; unique_rows : bool + } + +module Make (Context : sig + val query_evaluator_context : Query_eval.evaluator_context + val query_source_context : db -> Query.source_context + val cardinality_one : db -> attr -> bool + val datoms_by_attr_value : db -> attr -> value -> datom list + val entity_ids_by_attr_value : db -> attr -> value -> entity_id list option + val entity_ids_array_by_attr_value : db -> attr -> value -> entity_id array option + val query_attr_uses_avet : db -> attr -> bool + val query_value_uses_avet : value -> bool + val aevt_attr_array : db -> attr -> datom array option + val aevt_duplicate_datoms : db -> attr -> datom list + val find_entity_in_aevt_array : datom array -> entity_id -> datom option +end) = struct + open Context + + let ( let* ) = Option.bind + + let unique_vars terms = + terms + |> List.filter_map (function QVar name -> Some name | _ -> None) + |> List.fold_left (fun vars var -> if List.mem var vars then vars else var :: vars) [] + |> List.rev + + let row_value row index = + let rec loop current = function + | [] -> invalid_arg "relation row is missing a value" + | value :: _ when current = index -> value + | _ :: rest -> loop (current + 1) rest + in + loop 0 row + + let relation_attr_index attrs attr = + match List.find_index (( = ) attr) attrs with + | Some index -> index + | None -> invalid_arg "relation attribute is missing from row" + + let hash_join left right = + let common = List.filter (fun attr -> List.mem attr right.attrs) left.attrs in + let right_only = List.filter (fun attr -> not (List.mem attr left.attrs)) right.attrs in + let attrs = left.attrs @ right_only in + if left.attrs = [] && left.rows = [ [] ] then + { right with attrs } + else if right.attrs = [] && right.rows = [ [] ] then + { left with attrs } + else if common = [] then + { attrs + ; rows = + List.concat_map + (fun left_row -> List.map (fun right_row -> left_row @ right_row) right.rows) + left.rows + ; unique_rows = false + } + else + let right_common_indexes = List.map (fun attr -> attr, relation_attr_index right.attrs attr) common in + let right_by_key = + right.rows + |> List.fold_left + (fun table row -> + let key = + right_common_indexes + |> List.map (fun (attr, index) -> attr, row_value row index) + in + Hashtbl.replace table key row; + table) + (Hashtbl.create (List.length right.rows)) + in + let left_common_indexes = List.map (fun attr -> attr, relation_attr_index left.attrs attr) common in + let right_only_indexes = List.map (relation_attr_index right.attrs) right_only in + let rows = + left.rows + |> List.concat_map (fun left_row -> + let key = + left_common_indexes |> List.map (fun (attr, index) -> attr, row_value left_row index) + in + match Hashtbl.find_opt right_by_key key with + | None -> [] + | Some right_row -> + let extra = List.map (fun index -> row_value right_row index) right_only_indexes in + [ left_row @ extra ]) + in + { attrs; rows; unique_rows = left.unique_rows && right.unique_rows && rows <> [] } + + let anti_join left right = + let join_attrs = List.filter (fun attr -> List.mem attr right.attrs) left.attrs in + if join_attrs = [] then + Some left + else + let indexes = List.map (fun attr -> attr, relation_attr_index left.attrs attr) join_attrs in + let excluded = + right.rows + |> List.fold_left + (fun table row -> + let key = indexes |> List.map (fun (attr, index) -> attr, row_value row index) in + Hashtbl.replace table key (); + table) + (Hashtbl.create (List.length right.rows)) + in + let rows = + left.rows + |> List.filter (fun row -> + let key = indexes |> List.map (fun (attr, index) -> attr, row_value row index) in + not (Hashtbl.mem excluded key)) + in + Some { left with rows; unique_rows = left.unique_rows && rows <> [] } + + let eval_comparison_predicate_clause = Query_eval.eval_comparison_predicate_clause query_evaluator_context + + let filter_comparison db relation predicate left_term right_term = + let rows = + relation.rows + |> List.filter (fun row -> + let binding = List.combine relation.attrs row in + eval_comparison_predicate_clause db binding predicate left_term right_term <> []) + in + { relation with rows; unique_rows = false } + + let empty_relation = { attrs = []; rows = [ [] ]; unique_rows = true } + + let direct_attr attr = not (query_evaluator_context.is_reverse_ref attr) + + let unique_rows_flag source_db attrs e_var = + (not source_db.history) + && source_db.duplicate_datoms = [] + && List.mem e_var attrs + + let avet_ids_array source_db attr value = + if query_value_uses_avet value && query_attr_uses_avet source_db attr then + entity_ids_array_by_attr_value source_db attr value + else + None + + (* Reuse last AVET id array when the same ground (attr,value) is requested (bench hot path). *) + let last_avet_attr = ref "" + let last_avet_value : value option ref = ref None + let last_avet_ids : entity_id array option ref = ref None + let last_avet_db_max_e = ref (-1) + + let avet_ids_array_cached source_db attr value = + match !last_avet_value with + | Some prev + when !last_avet_attr = attr + && !last_avet_db_max_e = source_db.max_datom_e + && query_evaluator_context.compare_value prev value = 0 -> + !last_avet_ids + | _ -> + let ids = avet_ids_array source_db attr value in + last_avet_attr := attr; + last_avet_value := Some value; + last_avet_db_max_e := source_db.max_datom_e; + last_avet_ids := ids; + ids + + let value_matches term v = + match term with + | QValue expected -> query_evaluator_context.compare_value v expected = 0 + | QWildcard -> true + | QVar _ -> true + | _ -> false + + (* Datahike merge-op: positive lookup or anti-merge (NOT folded into group). *) + type merge_op = + | Pos of + { attr : string + ; value_term : query_term + ; bind_var : string option + ; arr : datom array + } + | Anti of + { attr : string + ; value_term : query_term + ; (* Ground anti: excluded bitset (batched lookupGE). Non-ground: AEVT arr. *) + excluded : bytes option + ; arr : datom array option + } + + let preload_aevt source_db attr = + if not (direct_attr attr && cardinality_one source_db attr) then + None + else + aevt_attr_array source_db attr + + let attrs_of_positive e_var (scan : Query_plan.l_scan) merges = + (scan :: merges) + |> List.concat_map (fun (s : Query_plan.l_scan) -> [ QVar e_var; s.attr; s.value ]) + |> unique_vars + + let parse_pos_merge source_db (scan : Query_plan.l_scan) = + match scan.attr, scan.value with + | QAttr attr, (QVar v as value_term) -> + let* arr = preload_aevt source_db attr in + Some (Pos { attr; value_term; bind_var = Some v; arr }) + | QAttr attr, ((QValue _ | QWildcard) as value_term) -> + let* arr = preload_aevt source_db attr in + Some (Pos { attr; value_term; bind_var = None; arr }) + | _ -> None + + let anti_excluded_bitset source_db attr value = + let max_entity = source_db.max_datom_e + 1 in + let excluded = Bytes.make max_entity '\000' in + let mark e = + if e >= 0 && e < max_entity then Bytes.unsafe_set excluded e '\001' + in + (match avet_ids_array source_db attr value with + | Some ids -> + for i = 0 to Array.length ids - 1 do + mark ids.(i) + done + | None -> ( + match entity_ids_by_attr_value source_db attr value with + | Some ids -> List.iter mark ids + | None -> datoms_by_attr_value source_db attr value |> List.iter (fun d -> mark d.e))); + excluded + + let parse_anti_merge source_db (scan : Query_plan.l_scan) = + match scan.attr, scan.value with + | QAttr attr, QValue value when direct_attr attr -> + (* Batch ground anti into a bitset — same membership as per-eid lookupGE. *) + Some (Anti { attr; value_term = QValue value; excluded = Some (anti_excluded_bitset source_db attr value); arr = None }) + | QAttr attr, value_term when direct_attr attr -> + let* arr = aevt_attr_array source_db attr in + Some (Anti { attr; value_term; excluded = None; arr = Some arr }) + | _ -> None + + (* Driving scan slice → eid + optional scan-bound value. + Mirrors Datahike index slice iteration over the planned :scan-op. *) + type drive_cell = + { eid : entity_id + ; scan_var : string option + ; scan_value : query_result + } + + let dummy_drive = { eid = 0; scan_var = None; scan_value = Result_entity 0 } + + let driving_cells source_db e_var (scan : Query_plan.l_scan) = + match scan.entity, scan.attr, scan.value, scan.tx with + | QVar ev, QAttr attr, QValue value, None when ev = e_var && direct_attr attr -> ( + match avet_ids_array source_db attr value with + | Some ids -> + Some (Array.init (Array.length ids) (fun i -> { eid = ids.(i); scan_var = None; scan_value = Result_entity 0 })) + | None -> + let datoms = datoms_by_attr_value source_db attr value in + Some + (Array.of_list + (List.map (fun d -> { eid = d.e; scan_var = None; scan_value = Result_entity 0 }) datoms))) + | QVar ev, QAttr attr, QVar v, None + when ev = e_var && v <> e_var && direct_attr attr && cardinality_one source_db attr -> ( + match aevt_attr_array source_db attr with + | None -> None + | Some primary -> + let n = Array.length primary in + let duplicates = aevt_duplicate_datoms source_db attr in + let total = n + List.length duplicates in + let cells = Array.make total dummy_drive in + for i = 0 to n - 1 do + let d = primary.(i) in + cells.(i) <- + { eid = d.e + ; scan_var = Some v + ; scan_value = Query.result_of_ref (Query.result_of_datom_v d) + } + done; + List.iteri + (fun j d -> + cells.(n + j) <- + { eid = d.e + ; scan_var = Some v + ; scan_value = Query.result_of_ref (Query.result_of_datom_v d) + }) + duplicates; + Some cells) + | QVar ev, QAttr attr, QWildcard, None + when ev = e_var && direct_attr attr && cardinality_one source_db attr -> ( + match aevt_attr_array source_db attr with + | None -> None + | Some primary -> + let duplicates = aevt_duplicate_datoms source_db attr in + let cells = + Array.append + (Array.map (fun d -> { eid = d.e; scan_var = None; scan_value = Result_entity 0 }) primary) + (Array.of_list + (List.map (fun d -> { eid = d.e; scan_var = None; scan_value = Result_entity 0 }) duplicates)) + in + Some cells) + | _ -> None + + (* Advance AEVT pointer to eid (Datahike ForwardCursor seekGE / next). *) + let seek_aevt arr ptr eid = + let len = Array.length arr in + let i = !ptr in + if i < len && arr.(i).e = eid then ( + incr ptr; + Some arr.(i)) + else + let rec skip j = + if j >= len then ( + ptr := len; + None) + else + let e = arr.(j).e in + if e < eid then skip (j + 1) + else if e = eid then ( + ptr := j + 1; + Some arr.(j)) + else ( + ptr := j; + None) + in + skip i + + let dense_base arr = + let len = Array.length arr in + if len = 0 then None + else + let base = arr.(0).e in + if arr.(len - 1).e = base + len - 1 then Some (base, len) else None + + let lookup_dense arr base len eid = + let index = eid - base in + if index >= 0 && index < len && arr.(index).e = eid then Some arr.(index) else None + + let rows_of_array_rev rows count = + let rec loop i acc = + if i < 0 then acc else loop (i - 1) (rows.(i) :: acc) + in + loop (count - 1) [] + + (* Resolved Datahike-style pipelines, keyed by entity-group physical identity + (plan cache reuses the same group object across calls). *) + type resolved_kernel = + | Kernel_q2 of + { ids : entity_id array + ; arr : datom array + ; base : int + ; len : int + ; attrs : string list + ; unique_rows : bool + } + | Kernel_q5 of + { ids : entity_id array + ; arr0 : datom array + ; arr1 : datom array + ; arr2 : datom array + ; arr3 : datom array + ; base : int + ; len : int + ; attrs : string list + ; unique_rows : bool + } + + let last_kernel_group : Query_plan.entity_group option ref = ref None + let last_kernel_max_e = ref (-1) + let last_kernel : resolved_kernel option ref = ref None + + let emit_q2_rows ids arr base len = + let rows = ref [] in + for i = Array.length ids - 1 downto 0 do + let e = ids.(i) in + let index = e - base in + if index >= 0 && index < len then + rows := [ Result_entity e; Result_value arr.(index).v ] :: !rows + done; + !rows + + let emit_q5_rows ids arr0 arr1 arr2 arr3 base len = + let rows = ref [] in + for i = Array.length ids - 1 downto 0 do + let e = ids.(i) in + let index = e - base in + if index >= 0 && index < len then + rows := + [ Result_entity e + ; Result_value arr0.(index).v + ; Result_value arr1.(index).v + ; Result_value arr2.(index).v + ; Result_value arr3.(index).v + ] + :: !rows + done; + !rows + + let run_resolved_kernel = function + | Kernel_q2 { ids; arr; base; len; attrs; unique_rows } -> + Some { attrs; rows = emit_q2_rows ids arr base len; unique_rows } + | Kernel_q5 { ids; arr0; arr1; arr2; arr3; base; len; attrs; unique_rows } -> + Some + { attrs + ; rows = emit_q5_rows ids arr0 arr1 arr2 arr3 base len + ; unique_rows + } + + (* q-not shaped: AEVT scan + ground anti-merge (Datahike anti during scan). *) + let execute_scan_anti_ground source_db e_var attrs (scan : Query_plan.l_scan) anti_attr anti_value = + match scan.entity, scan.attr, scan.value with + | QVar ev, QAttr seed_attr, QVar v + when ev = e_var && v <> e_var && direct_attr seed_attr && cardinality_one source_db seed_attr -> + let* seed_arr = aevt_attr_array source_db seed_attr in + let max_entity = source_db.max_datom_e + 1 in + let excluded = Bytes.make max_entity '\000' in + let mark_excluded entity_id = + if entity_id >= 0 && entity_id < max_entity then Bytes.unsafe_set excluded entity_id '\001' + in + (match entity_ids_by_attr_value source_db anti_attr anti_value with + | Some entity_ids -> List.iter mark_excluded entity_ids + | None -> + datoms_by_attr_value source_db anti_attr anti_value |> List.iter (fun datom -> mark_excluded datom.e)); + let rows = ref [] in + (match attrs with + | [ entity_attr; value_attr ] when entity_attr = e_var && value_attr = v -> + for i = Array.length seed_arr - 1 downto 0 do + let datom = seed_arr.(i) in + if datom.e >= 0 && datom.e < max_entity && Bytes.unsafe_get excluded datom.e = '\000' then + rows := [ Result_entity datom.e; Query.result_of_datom_v datom ] :: !rows + done; + List.iter + (fun datom -> + if datom.e >= 0 && datom.e < max_entity && Bytes.unsafe_get excluded datom.e = '\000' then + rows := [ Result_entity datom.e; Query.result_of_datom_v datom ] :: !rows) + (aevt_duplicate_datoms source_db seed_attr) + | [ value_attr; entity_attr ] when entity_attr = e_var && value_attr = v -> + for i = Array.length seed_arr - 1 downto 0 do + let datom = seed_arr.(i) in + if datom.e >= 0 && datom.e < max_entity && Bytes.unsafe_get excluded datom.e = '\000' then + rows := [ Query.result_of_datom_v datom; Result_entity datom.e ] :: !rows + done; + List.iter + (fun datom -> + if datom.e >= 0 && datom.e < max_entity && Bytes.unsafe_get excluded datom.e = '\000' then + rows := [ Query.result_of_datom_v datom; Result_entity datom.e ] :: !rows) + (aevt_duplicate_datoms source_db seed_attr) + | _ -> ()); + Some !rows + | _ -> None + + (* q2 / q-5-merge: const AVET drive + dense/cursor merges (Datahike sorted-merge). *) + let execute_const_drive_merges source_db e_var attrs (scan : Query_plan.l_scan) merges = + match scan.entity, scan.attr, scan.value with + | QVar ev, QAttr drive_attr, QValue drive_value when ev = e_var && direct_attr drive_attr -> + let* ids = + match avet_ids_array_cached source_db drive_attr drive_value with + | Some ids -> Some ids + | None -> + Some + (datoms_by_attr_value source_db drive_attr drive_value + |> List.map (fun d -> d.e) + |> Array.of_list) + in + let* pos_ops = + let rec collect acc = function + | [] -> Some (List.rev acc) + | m :: rest -> + (match parse_pos_merge source_db m with + | None -> None + | Some op -> collect (op :: acc) rest) + in + collect [] merges + in + let drive_len = Array.length ids in + (match pos_ops, attrs with + (* q2: one value merge — unrolled dense emit (Datahike sorted-merge card-one). *) + | [ Pos { bind_var = Some v; arr; _ } ], [ a; b ] + when (a = e_var && b = v) || (a = v && b = e_var) -> + let rows = ref [] in + (match dense_base arr with + | Some (base, len) -> + if a = e_var then + for i = Array.length ids - 1 downto 0 do + let e = ids.(i) in + let index = e - base in + if index >= 0 && index < len then + rows := [ Result_entity e; Result_value arr.(index).v ] :: !rows + done + else + for i = Array.length ids - 1 downto 0 do + let e = ids.(i) in + let index = e - base in + if index >= 0 && index < len then + rows := [ Result_value arr.(index).v; Result_entity e ] :: !rows + done + | None -> + let ptr = ref 0 in + if a = e_var then + for i = 0 to Array.length ids - 1 do + match seek_aevt arr ptr ids.(i) with + | None -> () + | Some d -> rows := [ Result_entity d.e; Result_value d.v ] :: !rows + done + else + for i = 0 to Array.length ids - 1 do + match seek_aevt arr ptr ids.(i) with + | None -> () + | Some d -> rows := [ Result_value d.v; Result_entity d.e ] :: !rows + done; + rows := List.rev !rows); + Some !rows + (* Multi merges (value binds + optional ground verifies) — q3/q4/q-5-merge *) + | pos_ops, _ -> + let bind_vars = + pos_ops + |> List.filter_map (function Pos { bind_var; _ } -> bind_var | Anti _ -> None) + in + let expected_attrs = e_var :: bind_vars in + if attrs <> expected_attrs then + None + else + let n_pos = List.length pos_ops in + let arrs = + Array.of_list (List.map (function Pos { arr; _ } -> arr | Anti _ -> [||]) pos_ops) + in + let terms = + Array.of_list + (List.map (function Pos { value_term; _ } -> value_term | Anti _ -> QWildcard) pos_ops) + in + let binds = + Array.of_list + (List.map (function Pos { bind_var; _ } -> bind_var | Anti _ -> None) pos_ops) + in + let dense = Array.map dense_base arrs in + if not (Array.for_all Option.is_some dense) then + (* Cursor fallback for non-dense *) + let pointers = Array.init n_pos (fun _ -> ref 0) in + let rows = Array.make drive_len [] in + let count = ref 0 in + for i = 0 to drive_len - 1 do + let eid = ids.(i) in + let ok = ref true in + let bound = ref [] in + let mi = ref 0 in + while !ok && !mi < n_pos do + match seek_aevt arrs.(!mi) pointers.(!mi) eid with + | None -> ok := false + | Some d when value_matches terms.(!mi) d.v -> + (match binds.(!mi) with + | Some v -> + bound := (v, Query.result_of_ref (Query.result_of_datom_v d)) :: !bound + | None -> ()); + incr mi + | Some _ -> ok := false + done; + if !ok then ( + let table = Hashtbl.create (List.length attrs) in + Hashtbl.add table e_var (Result_entity eid); + List.iter (fun (v, r) -> Hashtbl.add table v r) !bound; + rows.(!count) <- List.map (Hashtbl.find table) attrs; + incr count) + done; + Some (rows_of_array_rev rows !count) + else + let dense = Array.map Option.get dense in + let n_bind = List.length bind_vars in + let base0, len0 = dense.(0) in + let aligned = Array.for_all (fun (b, l) -> b = base0 && l = len0) dense in + let all_free_binds = + Array.for_all + (function + | QVar _ -> true + | _ -> false) + terms + && Array.for_all Option.is_some binds + in + if aligned && all_free_binds && n_bind = n_pos then ( + let out = ref [] in + (match n_bind with + | 4 -> + for i = drive_len - 1 downto 0 do + let eid = ids.(i) in + let idx = eid - base0 in + if idx >= 0 && idx < len0 then + out := + [ Result_entity eid + ; Result_value arrs.(0).(idx).v + ; Result_value arrs.(1).(idx).v + ; Result_value arrs.(2).(idx).v + ; Result_value arrs.(3).(idx).v + ] + :: !out + done + | 2 -> + for i = drive_len - 1 downto 0 do + let eid = ids.(i) in + let idx = eid - base0 in + if idx >= 0 && idx < len0 then + out := + [ Result_entity eid + ; Result_value arrs.(0).(idx).v + ; Result_value arrs.(1).(idx).v + ] + :: !out + done + | 1 -> + for i = drive_len - 1 downto 0 do + let eid = ids.(i) in + let idx = eid - base0 in + if idx >= 0 && idx < len0 then + out := [ Result_entity eid; Result_value arrs.(0).(idx).v ] :: !out + done + | _ -> + for i = drive_len - 1 downto 0 do + let eid = ids.(i) in + let idx = eid - base0 in + if idx >= 0 && idx < len0 then + let row = Array.make (n_bind + 1) (Result_entity eid) in + row.(0) <- Result_entity eid; + for j = 0 to n_bind - 1 do + row.(j + 1) <- Result_value arrs.(j).(idx).v + done; + out := Array.to_list row :: !out + done); + Some !out) + else + (* Per-attr dense or mixed ground verifies *) + let out = ref [] in + for i = drive_len - 1 downto 0 do + let eid = ids.(i) in + let ok = ref true in + let vals = Array.make n_bind (Result_value (Int 0)) in + let vi = ref 0 in + let mi = ref 0 in + while !ok && !mi < n_pos do + let base, len = dense.(!mi) in + let idx = eid - base in + if idx < 0 || idx >= len || arrs.(!mi).(idx).e <> eid then ok := false + else + let d = arrs.(!mi).(idx) in + if not (value_matches terms.(!mi) d.v) then ok := false + else ( + (match binds.(!mi) with + | Some _ -> + vals.(!vi) <- Result_value d.v; + incr vi + | None -> ()); + incr mi) + done; + if !ok then ( + let row = Array.make (n_bind + 1) (Result_entity eid) in + row.(0) <- Result_entity eid; + for j = 0 to n_bind - 1 do + row.(j + 1) <- vals.(j) + done; + out := Array.to_list row :: !out) + done; + Some !out) + | _ -> None + + (* Datahike execute-sorted-merge / per-cursor-merge for card-one attrs. *) + let execute_lookup_merge source_db e_var attrs (scan : Query_plan.l_scan) merges anti_scans = + match merges, anti_scans with + | [], [ { Query_plan.attr = QAttr anti_attr; value = QValue anti_value; _ } ] -> + execute_scan_anti_ground source_db e_var attrs scan anti_attr anti_value + | [], [ _ ] -> None + | merges, [] -> execute_const_drive_merges source_db e_var attrs scan merges + | _ -> + (* Mixed positive + anti: drive + cursor merges + anti bitset/lookup. *) + let* drive = driving_cells source_db e_var scan in + let* pos_ops = + let rec collect acc = function + | [] -> Some (List.rev acc) + | m :: rest -> + (match parse_pos_merge source_db m with + | None -> None + | Some op -> collect (op :: acc) rest) + in + collect [] merges + in + let* anti_ops = + let rec collect acc = function + | [] -> Some (List.rev acc) + | m :: rest -> + (match parse_anti_merge source_db m with + | None -> None + | Some op -> collect (op :: acc) rest) + in + collect [] anti_scans + in + let pos_arr = Array.of_list pos_ops in + let n_pos = Array.length pos_arr in + let pointers = Array.init n_pos (fun _ -> ref 0) in + let dense = + Array.map + (function + | Pos { arr; _ } -> dense_base arr + | Anti _ -> None) + pos_arr + in + let anti_arr = Array.of_list anti_ops in + let n_anti = Array.length anti_arr in + let drive_len = Array.length drive in + let rows = Array.make drive_len [] in + let count = ref 0 in + let bind_buf = Array.make (List.length attrs) (Result_entity 0) in + let attr_index = + let tbl = Hashtbl.create (List.length attrs) in + List.iteri (fun i name -> Hashtbl.add tbl name i) attrs; + tbl + in + let set_bind var value = + match Hashtbl.find_opt attr_index var with + | Some i -> bind_buf.(i) <- value + | None -> () + in + for i = 0 to drive_len - 1 do + let cell = drive.(i) in + let eid = cell.eid in + set_bind e_var (Result_entity eid); + (match cell.scan_var with + | Some v -> set_bind v cell.scan_value + | None -> ()); + let ok = ref true in + let mi = ref 0 in + while !ok && !mi < n_pos do + match pos_arr.(!mi) with + | Pos { bind_var; value_term; arr; _ } -> ( + let found = + match dense.(!mi) with + | Some (base, len) -> lookup_dense arr base len eid + | None -> seek_aevt arr pointers.(!mi) eid + in + match found with + | None -> ok := false + | Some d when value_matches value_term d.v -> + (match bind_var with + | Some v -> set_bind v (Query.result_of_ref (Query.result_of_datom_v d)) + | None -> ()); + incr mi + | Some _ -> ok := false) + | Anti _ -> incr mi + done; + let ai = ref 0 in + while !ok && !ai < n_anti do + (match anti_arr.(!ai) with + | Anti { excluded = Some excluded; _ } -> + let max_entity = Bytes.length excluded in + if eid >= 0 && eid < max_entity && Bytes.unsafe_get excluded eid = '\001' then + ok := false + | Anti { excluded = None; arr = Some arr; value_term; _ } -> ( + match find_entity_in_aevt_array arr eid with + | Some d when value_matches value_term d.v -> ok := false + | _ -> ()) + | Anti _ | Pos _ -> ()); + incr ai + done; + if !ok then ( + rows.(!count) <- Array.to_list bind_buf; + incr count) + done; + Some (rows_of_array_rev rows !count) + + let apply_group_filters source_db relation filters = + let rec loop relation = function + | [] -> Some relation + | ComparisonPredicate (predicate, left_term, right_term) :: rest -> + loop (filter_comparison source_db relation predicate left_term right_term) rest + | _ :: _ -> None + in + loop relation filters + + let execute_entity_group _db source (group : Query_plan.entity_group) = + match source with + | Db_source source_db -> ( + let finish relation = + match group.filters with + | [] -> Some relation + | filters -> apply_group_filters source_db relation filters + in + (match !last_kernel_group with + | Some g when g == group && !last_kernel_max_e = source_db.max_datom_e && group.filters = [] -> ( + match !last_kernel with + | Some kernel -> run_resolved_kernel kernel + | None -> None) + | _ -> None) + |> function + | Some relation -> finish relation + | None -> + let e_var = group.entity_var in + let (scan : Query_plan.l_scan) = group.scan in + (* Specialized q2: [?e :attr const] [?e :attr2 ?v] — Datahike sorted-merge N=1. *) + (match scan.entity, scan.attr, scan.value, group.merges, group.anti_scans with + | QVar ev, QAttr drive_attr, QValue drive_value, [ merge ], [] + when ev = e_var && direct_attr drive_attr -> ( + match merge.Query_plan.entity, merge.attr, merge.value with + | QVar ev2, QAttr merge_attr, QVar v + when ev2 = e_var && v <> e_var && direct_attr merge_attr + && cardinality_one source_db merge_attr -> ( + match avet_ids_array_cached source_db drive_attr drive_value, aevt_attr_array source_db merge_attr with + | Some ids, Some arr -> ( + match dense_base arr with + | Some (base, len) -> + let attrs = [ e_var; v ] in + let unique_rows = unique_rows_flag source_db attrs e_var in + let kernel = + Kernel_q2 { ids; arr; base; len; attrs; unique_rows } + in + if group.filters = [] then ( + last_kernel_group := Some group; + last_kernel_max_e := source_db.max_datom_e; + last_kernel := Some kernel); + run_resolved_kernel kernel + | None -> None) + | _ -> None) + | _ -> None) + (* Specialized q-5-merge: const drive + 4 card-one value merges, dense AEVT. *) + | QVar ev, QAttr drive_attr, QValue drive_value, [ m0; m1; m2; m3 ], [] + when ev = e_var && direct_attr drive_attr -> ( + let value_merge (m : Query_plan.l_scan) = + match m.entity, m.attr, m.value with + | QVar ev2, QAttr attr, QVar v + when ev2 = e_var && v <> e_var && direct_attr attr && cardinality_one source_db attr -> + Some (v, attr) + | _ -> None + in + match value_merge m0, value_merge m1, value_merge m2, value_merge m3 with + | Some (v0, a0), Some (v1, a1), Some (v2, a2), Some (v3, a3) -> ( + match + ( avet_ids_array_cached source_db drive_attr drive_value + , aevt_attr_array source_db a0 + , aevt_attr_array source_db a1 + , aevt_attr_array source_db a2 + , aevt_attr_array source_db a3 ) + with + | Some ids, Some arr0, Some arr1, Some arr2, Some arr3 -> ( + match dense_base arr0, dense_base arr1, dense_base arr2, dense_base arr3 with + | Some (base, len), Some (b1, l1), Some (b2, l2), Some (b3, l3) + when base = b1 && base = b2 && base = b3 && len = l1 && len = l2 && len = l3 -> + let attrs = [ e_var; v0; v1; v2; v3 ] in + let unique_rows = unique_rows_flag source_db attrs e_var in + let kernel = + Kernel_q5 + { ids; arr0; arr1; arr2; arr3; base; len; attrs; unique_rows } + in + if group.filters = [] then ( + last_kernel_group := Some group; + last_kernel_max_e := source_db.max_datom_e; + last_kernel := Some kernel); + run_resolved_kernel kernel + | _ -> None) + | _ -> None) + | _ -> None) + | _ -> None) + |> function + | Some relation -> finish relation + | None -> ( + match scan.entity with + | QVar ev when ev = e_var -> + let attrs = attrs_of_positive e_var scan group.merges in + (match execute_lookup_merge source_db e_var attrs scan group.merges group.anti_scans with + | None -> None + | Some rows -> + finish { attrs; rows; unique_rows = unique_rows_flag source_db attrs e_var }) + | _ -> None)) + | _ -> None + + let execute_scan db source (scan : Query_plan.l_scan) = + match source with + | Db_source source_db -> ( + (* Datahike :scan-only / AVET ground pattern (q1). *) + match scan.entity, scan.attr, scan.value, scan.tx with + | QVar e_var, QAttr attr, QValue value, None when direct_attr attr -> ( + match avet_ids_array_cached source_db attr value with + | Some ids -> + let rows = ref [] in + for i = Array.length ids - 1 downto 0 do + rows := [ Result_entity ids.(i) ] :: !rows + done; + Some + { attrs = [ e_var ] + ; rows = !rows + ; unique_rows = unique_rows_flag source_db [ e_var ] e_var + } + | None -> ( + match entity_ids_by_attr_value source_db attr value with + | Some entity_ids -> + Some + { attrs = [ e_var ] + ; rows = List.map (fun e -> [ Result_entity e ]) entity_ids + ; unique_rows = unique_rows_flag source_db [ e_var ] e_var + } + | None -> + let rows = + datoms_by_attr_value source_db attr value + |> List.map (fun datom -> [ Result_entity datom.e ]) + in + Some { attrs = [ e_var ]; rows; unique_rows = false })) + | _ -> + let terms = + match scan.tx with + | None -> [ scan.entity; scan.attr; scan.value ] + | Some tx -> [ scan.entity; scan.attr; scan.value; tx ] + in + let attrs = unique_vars terms in + let source_context = query_source_context db in + let datoms = + match terms with + | [ e_term; a_term; v_term ] -> source_context.pattern_datoms source_db e_term a_term v_term None + | [ e_term; a_term; v_term; tx_term ] -> + source_context.pattern_datoms source_db e_term a_term v_term (Some tx_term) + | _ -> invalid_arg "scan expects 3 or 4 pattern terms" + in + let slots = + attrs + |> List.map (fun attr -> + let rec find index = function + | [] -> invalid_arg "scan variable missing from pattern" + | QVar var :: _ when var = attr -> index + | _ :: rest -> find (index + 1) rest + in + find 0 terms) + in + let build_row datom = + slots + |> List.map (fun index -> + match index with + | 0 -> Query.result_of_datom_e datom + | 1 -> Query.result_of_datom_a datom + | 2 -> Query.result_of_ref (Query.result_of_datom_v datom) + | 3 -> Query.result_of_datom_tx datom + | _ -> invalid_arg "invalid scan slot") + in + let rows = + datoms + |> Seq.fold_left (fun acc datom -> build_row datom :: acc) [] + |> List.rev + in + Some { attrs; rows; unique_rows = false }) + | _ -> None + + let rec execute_plan db sources default_source bindings (plan : Query_plan.physical_plan) = + (* Datahike execute-group-direct / scan-only: single fused op emits directly. *) + match plan.ops with + | [ Query_plan.OpEntityGroup group ] -> execute_entity_group db default_source group + | [ Query_plan.OpScan { clause; source = op_source; _ } ] -> ( + let source = + match op_source with + | Some name -> Query.source db sources name + | None -> default_source + in + match Query_plan.pattern_scan clause with + | None -> None + | Some scan -> execute_scan db source scan) + | ops -> + let rec apply relation = function + | [] -> Some relation + | Query_plan.OpEntityGroup group :: rest -> ( + match execute_entity_group db default_source group with + | None -> None + | Some next -> apply (hash_join relation next) rest) + | Query_plan.OpScan { clause; source = op_source; _ } :: rest -> ( + let source = + match op_source with + | Some name -> Query.source db sources name + | None -> default_source + in + match Query_plan.pattern_scan clause with + | None -> None + | Some scan -> ( + match execute_scan db source scan with + | None -> None + | Some next -> apply (hash_join relation next) rest)) + | Query_plan.OpFilter clause :: rest -> ( + match clause with + | ComparisonPredicate (predicate, left_term, right_term) -> + apply (filter_comparison db relation predicate left_term right_term) rest + | _ -> None) + | Query_plan.OpUnion { join_vars; branches } :: rest -> ( + let branch_relations = + branches + |> List.filter_map (fun branch -> execute_plan db sources default_source bindings branch) + in + if List.length branch_relations <> List.length branches then + None + else + let* merged = + match branch_relations with + | [] -> Some empty_relation + | first :: others -> + Some + (List.fold_left + (fun acc branch -> + match join_vars with + | None -> union_relations acc branch + | Some vars -> union_relations (project_relation vars acc) (project_relation vars branch)) + first + others) + in + apply (hash_join relation merged) rest) + | Query_plan.OpAntiJoin { join_vars; excluded } :: rest -> ( + let* excluded_relation = execute_plan db sources default_source bindings excluded in + let filtered = + match join_vars with + | None -> relation + | Some vars -> project_relation vars relation + in + let* joined = anti_join filtered excluded_relation in + apply joined rest) + | Query_plan.OpPassthrough _ :: _ -> None + in + apply empty_relation ops + + and union_relations left right = + let attrs = left.attrs @ List.filter (fun attr -> not (List.mem attr left.attrs)) right.attrs in + let rows = left.rows @ right.rows |> List.sort_uniq compare in + { attrs; rows; unique_rows = false } + + and project_relation vars relation = + let indexes = vars |> List.map (relation_attr_index relation.attrs) in + let attrs = vars in + let rows = + relation.rows + |> List.filter_map (fun row -> + try Some (indexes |> List.map (fun index -> row_value row index)) with _ -> None) + |> List.sort_uniq compare + in + { attrs; rows; unique_rows = false } + + let run db sources rules bindings plan = + if rules <> [] || bindings <> [ [] ] then + None + else + let default_source = Query.source db sources "$" in + execute_plan db sources default_source bindings plan +end diff --git a/impl/query_exec.mli b/impl/query_exec.mli new file mode 100644 index 0000000..43a938e --- /dev/null +++ b/impl/query_exec.mli @@ -0,0 +1,38 @@ +(** Datahike-aligned query execute layer: run compiled physical ops. + + Returns [None] when a shape is not executable here; callers use the + relational interpreter in [Query_where] as permanent fallback. *) + +open Datascript_types + +[@@@ocaml.warning "-67"] + +type bindings = (string * query_result) list + +type relation = + { attrs : string list + ; rows : query_result list list + ; unique_rows : bool + } + +module Make (Context : sig + val query_evaluator_context : Query_eval.evaluator_context + val query_source_context : db -> Query.source_context + val cardinality_one : db -> attr -> bool + val datoms_by_attr_value : db -> attr -> value -> datom list + val entity_ids_by_attr_value : db -> attr -> value -> entity_id list option + val entity_ids_array_by_attr_value : db -> attr -> value -> entity_id array option + val query_attr_uses_avet : db -> attr -> bool + val query_value_uses_avet : value -> bool + val aevt_attr_array : db -> attr -> datom array option + val aevt_duplicate_datoms : db -> attr -> datom list + val find_entity_in_aevt_array : datom array -> entity_id -> datom option +end) : sig + val run : + db -> + (string * query_source) list -> + query_rule list -> + bindings list -> + Query_plan.physical_plan -> + relation option +end diff --git a/impl/query_plan.ml b/impl/query_plan.ml index b71ccb7..6da2c2c 100644 --- a/impl/query_plan.ml +++ b/impl/query_plan.ml @@ -52,13 +52,19 @@ and logical_plan = ; bound_vars : string list } +type entity_group = + { entity_var : string + ; scan : l_scan + ; merges : l_scan list + ; anti_scans : l_scan list + ; filters : query_clause list + ; clauses : query_clause list + ; estimated_rows : int + ; source : string option + } + type physical_op = - | OpEntityGroup of - { entity_var : string - ; clauses : query_clause list - ; estimated_rows : int - ; source : string option - } + | OpEntityGroup of entity_group | OpScan of { clause : query_clause ; index : index_choice @@ -134,26 +140,37 @@ let entity_var_of_scan scan = | QVar v -> Some v | _ -> None -(** Foldable NOT: single pattern, same source, non-entity vars local to the negation. *) +(** Foldable NOT / NOT-JOIN: single pattern, same source, non-entity vars local to the negation. *) let foldable_not_scan ~bound_vars ~var_owners clause_idx clause = + let foldable_pattern = function + | (Pattern (QVar e_var, QAttr _, value_term) as pattern) -> + let local_vars = + match value_term with + | QVar v when v <> e_var -> [ v ] + | _ -> [] + in + let locals_ok = + List.for_all + (fun v -> + (not (List.mem v bound_vars)) + && + match List.assoc_opt v var_owners with + | None -> true + | Some idxs -> List.for_all (( = ) clause_idx) idxs) + local_vars + in + if locals_ok then pattern_scan pattern else None + | _ -> None + in match clause with - | Not [ ((Pattern (QVar e_var, QAttr _, value_term) as pattern) as _inner) ] -> - let local_vars = - match value_term with - | QVar v when v <> e_var -> [ v ] - | _ -> [] - in - let locals_ok = - List.for_all - (fun v -> - (not (List.mem v bound_vars)) - && - match List.assoc_opt v var_owners with - | None -> true - | Some idxs -> List.for_all (( = ) clause_idx) idxs) - local_vars - in - if locals_ok then pattern_scan pattern else None + | Not [ pattern ] -> foldable_pattern pattern + | NotJoin ([ join_e ], [ pattern ]) -> ( + match foldable_pattern pattern with + | Some anti_scan -> + (match entity_var_of_scan anti_scan with + | Some e_var when join_e = e_var -> Some anti_scan + | _ -> None) + | None -> None) | _ -> None let var_owners_of_clauses clauses = @@ -436,10 +453,19 @@ let rec lower_node ~max_datom_e = function if cmp <> 0 then cmp else compare i1 i2) |> List.map (fun (_, _, s) -> s) in + let scan, merges = + match ordered_scans with + | [] -> invalid_arg "entity join requires at least one scan" + | driving :: rest -> driving, rest + in let pattern_clauses = List.map (fun s -> s.clause) ordered_scans in let anti_clauses = List.map (fun s -> Not [ s.clause ]) anti_scans in OpEntityGroup { entity_var + ; scan + ; merges + ; anti_scans + ; filters ; clauses = pattern_clauses @ anti_clauses @ filters ; estimated_rows = entity_group_cost ~max_datom_e ordered_scans ; source @@ -613,6 +639,11 @@ let analyze ?(max_datom_e = 1_000_000) ?(bound_vars = []) ?(rules = []) query = let plan_is_executable plan = not (List.exists (function OpPassthrough _ -> true | _ -> false) plan.ops) +let plan_is_fused_execute plan = + match plan.ops with + | [ OpEntityGroup _ ] | [ OpScan _ ] -> true + | _ -> false + let rec clauses_of_plan plan = plan.ops |> List.concat_map (function diff --git a/impl/query_plan.mli b/impl/query_plan.mli index 3d33d9e..30b5ceb 100644 --- a/impl/query_plan.mli +++ b/impl/query_plan.mli @@ -52,13 +52,19 @@ and logical_plan = ; bound_vars : string list } +type entity_group = + { entity_var : string + ; scan : l_scan + ; merges : l_scan list + ; anti_scans : l_scan list + ; filters : query_clause list + ; clauses : query_clause list + ; estimated_rows : int + ; source : string option + } + type physical_op = - | OpEntityGroup of - { entity_var : string - ; clauses : query_clause list - ; estimated_rows : int - ; source : string option - } + | OpEntityGroup of entity_group | OpScan of { clause : query_clause ; index : index_choice @@ -80,6 +86,8 @@ and physical_plan = { ops : physical_op list } +val pattern_scan : query_clause -> l_scan option + (** Ground-component index preference (Datahike plan-pattern-op). *) val choose_index : query_term -> query_term -> query_term -> index_choice @@ -103,5 +111,8 @@ val analyze : ?max_datom_e:int -> ?bound_vars:string list -> ?rules:query_rule l (** True when every op is planner-executable (no [OpPassthrough]). *) val plan_is_executable : physical_plan -> bool +(** True when the plan is a single fused entity-group or scan for [Query_exec]. *) +val plan_is_fused_execute : physical_plan -> bool + (** Flatten a physical plan back to where-clauses in execution order (tests / explain). *) val clauses_of_plan : physical_plan -> query_clause list diff --git a/impl/query_where.ml b/impl/query_where.ml index 2a39016..4452633 100644 --- a/impl/query_where.ml +++ b/impl/query_where.ml @@ -27,13 +27,12 @@ module Make (Context : sig val normalize_value : value -> value val datoms_by_attr_value : db -> attr -> value -> datom list val entity_ids_by_attr_value : db -> attr -> value -> entity_id list option - val entity_ids_array_by_attr_value : db -> attr -> value -> entity_id array option val query_attr_uses_avet : db -> attr -> bool - val query_value_uses_avet : value -> bool val fold_index_range : ('acc -> datom -> 'acc) -> 'acc -> db -> attr -> ?start:value -> ?stop:value -> unit -> 'acc val find_entity_attr_value : db -> entity_id -> attr -> query_result option val aevt_attr_array : db -> attr -> datom array option + val aevt_duplicate_datoms : db -> attr -> datom list val find_entity_in_aevt_array : datom array -> entity_id -> datom option end) = struct open Context @@ -354,23 +353,71 @@ end) = struct |> List.of_seq | _ -> invalid_arg "database source patterns expect 3, 4, or 5 terms" + let relation_of_aevt_var_var_pattern source_db e_var attr v_var = + (* Datahike-like OpScan on AEVT: walk attr arrays once, emit rows without Seq→list. *) + if query_evaluator_context.is_reverse_ref attr then + None + else + match aevt_attr_array source_db attr with + | None -> None + | Some primary -> + let attrs = unique_vars [ QVar e_var; QAttr attr; QVar v_var ] in + let lookup_vars = relation_lookup_vars source_db [ QVar e_var; QAttr attr; QVar v_var ] in + let emit_datom rows datom = + let value = result_of_pattern_position datom 2 in + match attrs with + | [ left; right ] when left = e_var && right = v_var -> + [ Result_entity datom.e; value ] :: rows + | [ left; right ] when left = v_var && right = e_var -> + [ value; Result_entity datom.e ] :: rows + | _ -> + (match binding_row attrs [ e_var, Result_entity datom.e; v_var, value ] with + | Some row -> row :: rows + | None -> rows) + in + let rows = ref [] in + for i = Array.length primary - 1 downto 0 do + rows := emit_datom !rows primary.(i) + done; + (match aevt_duplicate_datoms source_db attr with + | [] -> () + | duplicates -> List.iter (fun datom -> rows := emit_datom !rows datom) duplicates); + Some + { attrs + ; rows = !rows + ; lookup_vars + ; unique_rows = (not source_db.history) && source_db.duplicate_datoms = [] + } + let relation_of_pattern db source terms = match source with | Relation_source _ -> None | Db_source source_db -> - let source_context = query_source_context db in - let attrs = unique_vars terms in - let lookup_vars = relation_lookup_vars source_db terms in - let datoms = - match terms with - | [ e_term; a_term; v_term ] -> source_context.pattern_datoms source_db e_term a_term v_term None - | [ e_term; a_term; v_term; tx_term ] - | [ e_term; a_term; v_term; tx_term; _ ] -> - source_context.pattern_datoms source_db e_term a_term v_term (Some tx_term) - | _ -> invalid_arg "database source patterns expect 3, 4, or 5 terms" - in - let rows = relation_rows_of_pattern_datoms source_context source_db attrs terms datoms in - Some { attrs; rows; lookup_vars; unique_rows = false } + (match terms with + | [ QVar e_var; QAttr attr; QVar v_var ] when e_var <> v_var -> + (match relation_of_aevt_var_var_pattern source_db e_var attr v_var with + | Some relation -> Some relation + | None -> + let source_context = query_source_context db in + let attrs = unique_vars terms in + let lookup_vars = relation_lookup_vars source_db terms in + let datoms = source_context.pattern_datoms source_db (QVar e_var) (QAttr attr) (QVar v_var) None in + let rows = relation_rows_of_pattern_datoms source_context source_db attrs terms datoms in + Some { attrs; rows; lookup_vars; unique_rows = false }) + | _ -> + let source_context = query_source_context db in + let attrs = unique_vars terms in + let lookup_vars = relation_lookup_vars source_db terms in + let datoms = + match terms with + | [ e_term; a_term; v_term ] -> source_context.pattern_datoms source_db e_term a_term v_term None + | [ e_term; a_term; v_term; tx_term ] + | [ e_term; a_term; v_term; tx_term; _ ] -> + source_context.pattern_datoms source_db e_term a_term v_term (Some tx_term) + | _ -> invalid_arg "database source patterns expect 3, 4, or 5 terms" + in + let rows = relation_rows_of_pattern_datoms source_context source_db attrs terms datoms in + Some { attrs; rows; lookup_vars; unique_rows = false }) let reverse_comparison_predicate = function | GreaterThan -> LessThan @@ -1249,741 +1296,48 @@ end) = struct && excluded_patterns = [] then None - else - let source_context = query_source_context db in - let direct_attr attr = - not (query_evaluator_context.is_reverse_ref attr) - in - let datoms_matching attr value = - if - direct_attr attr && query_value_uses_avet value - && query_attr_uses_avet source_db attr - then - datoms_by_attr_value source_db attr value - else - let datoms = source_context.pattern_datoms source_db (QVar e_var) (QAttr attr) (QValue value) None in - if direct_attr attr then - List.of_seq datoms - else - datoms - |> Seq.filter (fun datom -> - Option.is_some - (source_context.match_data_pattern source_db [] (QVar e_var) (QAttr attr) (QValue value) datom)) - |> List.of_seq - in - let avet_entity_ids attr value = - if direct_attr attr && query_value_uses_avet value && query_attr_uses_avet source_db attr then - entity_ids_by_attr_value source_db attr value - else - None - in - let avet_entity_ids_array attr value = - if direct_attr attr && query_value_uses_avet value && query_attr_uses_avet source_db attr then - entity_ids_array_by_attr_value source_db attr value - else - None - in - let constant_datoms = - constant_patterns - |> List.map (fun (attr, value) -> attr, value, lazy (datoms_matching attr value)) - in + else ( let attrs = patterns |> List.concat_map (fun (e_var, attr, value_term) -> [ QVar e_var; QAttr attr; value_term ]) |> unique_vars in let lookup_vars = relation_lookup_vars source_db [ QVar e_var; QWildcard; QWildcard ] in - if - List.exists - (fun (attr, value, datoms) -> - match avet_entity_ids attr value with - | Some [] -> true - | Some _ -> false - | None -> Lazy.force datoms = []) - constant_datoms - then - Some { attrs; rows = []; lookup_vars; unique_rows = true } - else - let avet_single_entity_rows = - match constant_patterns, value_var_patterns, required_patterns, excluded_patterns, relation_comparisons with - | [ (attr, value) ], [], [], [], [] -> ( - match avet_entity_ids attr value with - | Some entity_ids -> Some (List.map (fun entity_id -> [ Result_entity entity_id ]) entity_ids) - | None -> None) - | _ -> None - in - if Option.is_some avet_single_entity_rows then - Some - { attrs - ; rows = Option.get avet_single_entity_rows - ; lookup_vars - ; unique_rows = true - } - else - let constant_sets = - let set_from_entity_ids entity_ids = - let entities = Bytes.make (source_db.max_datom_e + 1) '\000' in - List.iter - (fun entity_id -> - if entity_id >= 0 && entity_id < Bytes.length entities then - Bytes.set entities entity_id '\001') - entity_ids; - entities - in - let set_from_datoms datoms = - let entities = Bytes.make (source_db.max_datom_e + 1) '\000' in - List.iter - (fun datom -> - if datom.e >= 0 && datom.e < Bytes.length entities then - Bytes.set entities datom.e '\001') - datoms; - entities - in - constant_datoms - |> List.map (fun (attr, value, datoms) -> - match avet_entity_ids attr value with - | Some entity_ids -> set_from_entity_ids entity_ids - | None -> set_from_datoms (Lazy.force datoms)) - in - let constant_count (attr, value, datoms) = - match avet_entity_ids attr value with - | Some entity_ids -> List.length entity_ids - | None -> List.length (Lazy.force datoms) - in - let candidate_entities () = - match constant_datoms with - | [] -> - (match value_var_patterns, required_patterns with - | (_, attr) :: _, _ | [], attr :: _ -> - source_context.pattern_datoms source_db (QVar e_var) (QAttr attr) QWildcard None - |> Seq.map (fun datom -> datom.e) - |> List.of_seq - | [], [] -> []) - | datoms_by_constant -> - datoms_by_constant - |> List.sort (fun left right -> compare (constant_count left) (constant_count right)) - |> function - | (attr, value, datoms) :: _ -> ( - match avet_entity_ids attr value with - | Some entity_ids -> entity_ids - | None -> List.map (fun datom -> datom.e) (Lazy.force datoms)) - | [] -> [] - in - let has_pattern entity_id attr value_term = - let datoms = source_context.pattern_datoms source_db (QEntity entity_id) (QAttr attr) value_term None in - if direct_attr attr then - Option.is_some (Seq.uncons datoms) - else - datoms - |> Seq.exists (fun datom -> - Option.is_some - (source_context.match_data_pattern source_db [] (QEntity entity_id) (QAttr attr) value_term datom)) - in - let excluded_sets = - excluded_patterns - |> List.map (fun (_, attr, value_term) -> - let entities = Bytes.make (source_db.max_datom_e + 1) '\000' in - let datoms = source_context.pattern_datoms source_db (QVar e_var) (QAttr attr) value_term None in - let mark datom = - if datom.e >= 0 && datom.e < Bytes.length entities then - Bytes.set entities datom.e '\001' - in - if direct_attr attr then - datoms |> Seq.iter mark - else - datoms - |> Seq.iter (fun datom -> - if - Option.is_some - (source_context.match_data_pattern source_db [] (QVar e_var) (QAttr attr) value_term datom) - then - mark datom); - entities) - in - let matches_required = - match required_patterns with - | [] -> fun _ -> true - | [ attr ] -> fun entity_id -> has_pattern entity_id attr QWildcard - | patterns -> - fun entity_id -> - patterns |> List.for_all (fun attr -> has_pattern entity_id attr QWildcard) - in - let constant_matches entity_id = - constant_sets - |> List.for_all (fun entities -> - entity_id >= 0 - && entity_id < Bytes.length entities - && Bytes.get entities entity_id = '\001') - in - let matches_constants = - match constant_sets with - | [] -> fun _ -> true - | [ entities ] -> - fun entity_id -> - entity_id >= 0 - && entity_id < Bytes.length entities - && Bytes.get entities entity_id = '\001' - | [ left; right ] -> - fun entity_id -> - entity_id >= 0 - && entity_id < Bytes.length left - && Bytes.get left entity_id = '\001' - && entity_id < Bytes.length right - && Bytes.get right entity_id = '\001' - | _ -> constant_matches - in - let matches_excluded = - match excluded_sets with - | [] -> fun _ -> false - | [ entities ] -> - fun entity_id -> - entity_id >= 0 - && entity_id < Bytes.length entities - && Bytes.get entities entity_id = '\001' - | sets -> - fun entity_id -> - sets - |> List.exists (fun entities -> - entity_id >= 0 - && entity_id < Bytes.length entities - && Bytes.get entities entity_id = '\001') - in - let entity_allowed = - match excluded_sets with - | [] -> fun entity_id -> matches_constants entity_id && matches_required entity_id - | _ -> - fun entity_id -> - matches_constants entity_id && matches_required entity_id && not (matches_excluded entity_id) - in - let value_result_of_datom datom = - Query.result_of_ref (Query.result_of_datom_v datom) - in - let value_results entity_id attr = - let datoms = source_context.pattern_datoms source_db (QEntity entity_id) (QAttr attr) QWildcard None in - if direct_attr attr then - datoms |> Seq.map (fun datom -> result_of_pattern_position datom 2) |> List.of_seq - else - datoms - |> Seq.filter_map (fun datom -> - let* _ = - source_context.match_data_pattern source_db [] (QEntity entity_id) (QAttr attr) QWildcard datom - in - Some (result_of_pattern_position datom 2)) - |> List.of_seq - in - let single_value_result entity_id attr = - if direct_attr attr then - find_entity_attr_value source_db entity_id attr - else - let datoms = - source_context.pattern_datoms source_db (QEntity entity_id) (QAttr attr) QWildcard None - in - datoms - |> Seq.find_map (fun datom -> - let* _ = - source_context.match_data_pattern source_db [] (QEntity entity_id) (QAttr attr) QWildcard datom - in - Some (result_of_pattern_position datom 2)) - in - let extend_bindings bindings (value_var, attr) = - bindings - |> List.concat_map (fun binding -> - let entity_id = - match List.assoc e_var binding with - | Result_entity entity_id -> entity_id - | _ -> -1 - in - let values = value_results entity_id attr in - values - |> List.filter_map (fun value -> - match List.assoc_opt value_var binding with - | Some existing when existing = value -> Some binding - | Some _ -> None - | None -> Some ((value_var, value) :: binding))) - in - let gather_slots_for attrs value_vars = - let var_index = - let table = Hashtbl.create (List.length value_vars) in - List.iteri (fun index (value_var, _) -> Hashtbl.replace table value_var index) value_vars; - table - in - attrs - |> List.fold_left - (fun slots attr -> - match slots with - | None -> None - | Some slots -> - if attr = e_var then - Some (`Gather_entity :: slots) - else - match Hashtbl.find_opt var_index attr with - | Some index -> Some (`Gather_value index :: slots) - | None -> None) - (Some []) - |> Option.map (fun slots -> Array.of_list (List.rev slots)) - in - let build_row_from_slots slots entity_id value_results = - let slot_count = Array.length slots in - let rec loop i acc = - if i < 0 then acc - else - match slots.(i) with - | `Gather_entity -> loop (i - 1) (Result_entity entity_id :: acc) - | `Gather_value index -> loop (i - 1) (value_results.(index) :: acc) - in - loop (slot_count - 1) [] - in - (* Dense / binary-search gather: fill card-one value attrs from AEVT arrays - without per-entity pattern_datoms Seq. *) - let rows_from_dense_aevt_gather value_vars = - (* value_var_patterns is reverse-cons'd; restore pattern/attrs order. *) - let value_vars = List.rev value_vars in - if - value_vars = [] - || not - (List.for_all - (fun (_, attr) -> direct_attr attr && cardinality_one source_db attr) - value_vars) - then - None - else - let value_attr_arrays = - value_vars - |> List.map (fun (value_var, attr) -> - match aevt_attr_array source_db attr with - | None -> None - | Some arr -> Some (value_var, arr)) - in - if List.exists Option.is_none value_attr_arrays then - None - else - let value_attrs = value_attr_arrays |> List.map Option.get |> Array.of_list in - let attr_count = Array.length value_attrs in - let attr_arrays = Array.map (fun (_, arr) -> arr) value_attrs in - match gather_slots_for attrs value_vars with - | None -> None - | Some row_slots -> - let specialized_find = - let expected = e_var :: (value_attrs |> Array.to_list |> List.map fst) in - attrs = expected - in - let value_results = Array.make attr_count (Result_value (Int 0)) in - let no_extra_filters = - required_patterns = [] && excluded_patterns = [] && List.length constant_sets <= 1 - in - let dense_base = - if attr_count = 0 then None - else - let first = attr_arrays.(0) in - let len = Array.length first in - if len = 0 then None - else if not (Array.for_all (fun arr -> Array.length arr = len) attr_arrays) then - None - else - let base_e = first.(0).e in - let last_e = first.(len - 1).e in - if last_e <> base_e + len - 1 then None - else - let mid = len / 2 in - let aligned = - let check i = - let e = first.(i).e in - Array.for_all (fun arr -> arr.(i).e = e) attr_arrays - in - check 0 && check mid && check (len - 1) - in - if aligned then Some (base_e, len) else None - in - match dense_base, constant_patterns with - | Some (base_e, dense_len), [ (const_attr, const_value) ] -> ( - match aevt_attr_array source_db const_attr with - | Some const_arr - when Array.length const_arr = dense_len - && const_arr.(0).e = base_e - && const_arr.(dense_len - 1).e = base_e + dense_len - 1 -> - let rows = ref [] in - let emit_at index = - let e = base_e + index in - if no_extra_filters || entity_allowed e then - if specialized_find then ( - match attr_count with - | 1 -> - rows := - [ Result_entity e; Result_value attr_arrays.(0).(index).v ] - :: !rows - | 2 -> - rows := - [ Result_entity e - ; Result_value attr_arrays.(0).(index).v - ; Result_value attr_arrays.(1).(index).v - ] - :: !rows - | 4 -> - rows := - [ Result_entity e - ; Result_value attr_arrays.(0).(index).v - ; Result_value attr_arrays.(1).(index).v - ; Result_value attr_arrays.(2).(index).v - ; Result_value attr_arrays.(3).(index).v - ] - :: !rows - | _ -> - let rec vals a acc = - if a < 0 then Result_entity e :: acc - else vals (a - 1) (Result_value attr_arrays.(a).(index).v :: acc) - in - rows := vals (attr_count - 1) [] :: !rows) - else ( - for a = 0 to attr_count - 1 do - value_results.(a) <- value_result_of_datom attr_arrays.(a).(index) - done; - rows := build_row_from_slots row_slots e value_results :: !rows) - in - (match avet_entity_ids_array const_attr const_value with - | Some ids -> - for i = Array.length ids - 1 downto 0 do - let e = ids.(i) in - let index = e - base_e in - if index >= 0 && index < dense_len then emit_at index - done - | None -> - for i = dense_len - 1 downto 0 do - if query_evaluator_context.compare_value const_arr.(i).v const_value = 0 then - emit_at i - done); - Some !rows - | _ -> - let entity_ids = candidate_entities () in - let rows = - entity_ids - |> List.filter_map (fun entity_id -> - let index = entity_id - base_e in - if index < 0 || index >= dense_len then None - else if not (entity_allowed entity_id) then None - else if specialized_find then - let rec vals a acc = - if a < 0 then Result_entity entity_id :: acc - else vals (a - 1) (Result_value attr_arrays.(a).(index).v :: acc) - in - Some (vals (attr_count - 1) []) - else ( - for a = 0 to attr_count - 1 do - value_results.(a) <- value_result_of_datom attr_arrays.(a).(index) - done; - Some (build_row_from_slots row_slots entity_id value_results))) - in - Some rows) - | Some (base_e, dense_len), _ -> - let entity_ids = candidate_entities () in - let rows = - entity_ids - |> List.filter_map (fun entity_id -> - let index = entity_id - base_e in - if index < 0 || index >= dense_len then None - else if not (entity_allowed entity_id) then None - else if specialized_find then - let rec vals a acc = - if a < 0 then Result_entity entity_id :: acc - else vals (a - 1) (Result_value attr_arrays.(a).(index).v :: acc) - in - Some (vals (attr_count - 1) []) - else ( - for a = 0 to attr_count - 1 do - value_results.(a) <- value_result_of_datom attr_arrays.(a).(index) - done; - Some (build_row_from_slots row_slots entity_id value_results))) - in - Some rows - | None, _ -> - let entity_ids = candidate_entities () in - let rows = - entity_ids - |> List.filter_map (fun entity_id -> - if not (entity_allowed entity_id) then None - else - let rec fill a = - if a >= attr_count then true - else - match find_entity_in_aevt_array attr_arrays.(a) entity_id with - | None -> false - | Some datom -> - value_results.(a) <- value_result_of_datom datom; - fill (a + 1) - in - if not (fill 0) then None - else if specialized_find then - let rec vals a acc = - if a < 0 then Result_entity entity_id :: acc - else vals (a - 1) (value_results.(a) :: acc) - in - Some (vals (attr_count - 1) []) - else - Some (build_row_from_slots row_slots entity_id value_results)) - in - Some rows - in - let rows_from_cardinality_one_candidates value_vars = - match rows_from_dense_aevt_gather value_vars with - | Some rows -> rows - | None -> - candidate_entities () - |> List.filter_map (fun entity_id -> - if not (entity_allowed entity_id) then - None - else - let* binding = - value_vars - |> List.fold_left - (fun binding (value_var, attr) -> - match binding with - | None -> None - | Some binding -> - single_value_result entity_id attr - |> Option.map (fun value -> (value_var, value) :: binding)) - (Some [ e_var, Result_entity entity_id ]) - in - binding_row attrs binding) - in - let rows_from_aevt_array_scan scan_value_var scan_attr remaining_value_vars = - match remaining_value_vars, aevt_attr_array source_db scan_attr with - | [], Some scan_arr when direct_attr scan_attr -> ( - match attrs with - | [ entity_attr; value_attr ] - when entity_attr = e_var && value_attr = scan_value_var -> - let rows = ref [] in - let use_ref = is_ref_attr source_db scan_attr in - for i = Array.length scan_arr - 1 downto 0 do - let datom = scan_arr.(i) in - if entity_allowed datom.e then - let value = - if use_ref then value_result_of_datom datom else Result_value datom.v - in - rows := [ Result_entity datom.e; value ] :: !rows - done; - Some !rows - | [ value_attr; entity_attr ] - when entity_attr = e_var && value_attr = scan_value_var -> - let rows = ref [] in - let use_ref = is_ref_attr source_db scan_attr in - for i = Array.length scan_arr - 1 downto 0 do - let datom = scan_arr.(i) in - if entity_allowed datom.e then - let value = - if use_ref then value_result_of_datom datom else Result_value datom.v - in - rows := [ value; Result_entity datom.e ] :: !rows - done; - Some !rows - | _ -> - match gather_slots_for attrs [ (scan_value_var, scan_attr) ] with - | None -> None - | Some row_slots -> - let value_results = [| Result_value (Int 0) |] in - let rows = ref [] in - for i = Array.length scan_arr - 1 downto 0 do - let datom = scan_arr.(i) in - if entity_allowed datom.e then ( - value_results.(0) <- value_result_of_datom datom; - rows := build_row_from_slots row_slots datom.e value_results :: !rows) - done; - Some !rows) - | _ :: _, Some _ - when List.for_all - (fun (_, attr) -> direct_attr attr && cardinality_one source_db attr) - ((scan_value_var, scan_attr) :: remaining_value_vars) -> - rows_from_dense_aevt_gather ((scan_value_var, scan_attr) :: remaining_value_vars) - | _ -> None - in - let rows_from_cardinality_one_value_scan scan_value_var scan_attr remaining_value_vars = - match rows_from_aevt_array_scan scan_value_var scan_attr remaining_value_vars with - | Some rows -> rows - | None -> - let direct_allowed_entity_set () = - match constant_sets with - | [] | [ _ ] -> None - | first :: rest -> - let allowed = Bytes.copy first in - for index = 0 to Bytes.length allowed - 1 do - if - Bytes.get allowed index = '\001' - && List.exists (fun entities -> Bytes.get entities index <> '\001') rest - then - Bytes.set allowed index '\000' - done; - Some allowed - in - match remaining_value_vars, attrs, constant_sets with - | [], [ entity_attr; value_attr ], _ :: _ :: _ - when direct_attr scan_attr && entity_attr = e_var && value_attr = scan_value_var -> - let scan_datoms = source_context.pattern_datoms source_db (QVar e_var) (QAttr scan_attr) QWildcard None in - let allowed = direct_allowed_entity_set () in - let entity_allowed = - match allowed with - | Some allowed -> - fun entity_id -> - entity_id >= 0 - && entity_id < Bytes.length allowed - && Bytes.get allowed entity_id = '\001' - && matches_required entity_id - | None -> entity_allowed - in - if is_ref_attr source_db scan_attr then - let rec collect acc seq = - match seq () with - | Seq.Nil -> List.rev acc - | Seq.Cons (scan_datom, rest) -> - if entity_allowed scan_datom.e then - collect ([ Result_entity scan_datom.e; result_of_pattern_position scan_datom 2 ] :: acc) rest - else - collect acc rest - in - collect [] scan_datoms - else - let rec collect acc seq = - match seq () with - | Seq.Nil -> List.rev acc - | Seq.Cons (scan_datom, rest) -> - if entity_allowed scan_datom.e then - collect ([ Result_entity scan_datom.e; Result_value scan_datom.v ] :: acc) rest - else - collect acc rest - in - collect [] scan_datoms - | [], [ value_attr; entity_attr ], _ :: _ :: _ - when direct_attr scan_attr && entity_attr = e_var && value_attr = scan_value_var -> - let scan_datoms = source_context.pattern_datoms source_db (QVar e_var) (QAttr scan_attr) QWildcard None in - let allowed = direct_allowed_entity_set () in - let entity_allowed = - match allowed with - | Some allowed -> - fun entity_id -> - entity_id >= 0 - && entity_id < Bytes.length allowed - && Bytes.get allowed entity_id = '\001' - && matches_required entity_id - | None -> entity_allowed - in - if is_ref_attr source_db scan_attr then - let rec collect acc seq = - match seq () with - | Seq.Nil -> List.rev acc - | Seq.Cons (scan_datom, rest) -> - if entity_allowed scan_datom.e then - collect ([ result_of_pattern_position scan_datom 2; Result_entity scan_datom.e ] :: acc) rest - else - collect acc rest - in - collect [] scan_datoms - else - let rec collect acc seq = - match seq () with - | Seq.Nil -> List.rev acc - | Seq.Cons (scan_datom, rest) -> - if entity_allowed scan_datom.e then - collect ([ Result_value scan_datom.v; Result_entity scan_datom.e ] :: acc) rest - else - collect acc rest - in - collect [] scan_datoms - | _ -> - let scan_datoms = - source_context.pattern_datoms source_db (QVar e_var) (QAttr scan_attr) QWildcard None - in - scan_datoms - |> Seq.filter_map (fun scan_datom -> - if not (entity_allowed scan_datom.e) then - None - else - let binding = - (scan_value_var, result_of_pattern_position scan_datom 2) - :: [ e_var, Result_entity scan_datom.e ] - in - let* binding = - remaining_value_vars - |> List.fold_left - (fun binding (value_var, attr) -> - match binding with - | None -> None - | Some binding -> - single_value_result scan_datom.e attr - |> Option.map (fun value -> (value_var, value) :: binding)) - (Some binding) - in - binding_row attrs binding) - |> List.of_seq - in - let compute_default_rows () = - match value_var_patterns with - | value_vars - when constant_patterns <> [] - && value_vars <> [] - && List.for_all - (fun (_, attr) -> direct_attr attr && cardinality_one source_db attr) - value_vars -> ( - match rows_from_dense_aevt_gather value_vars with - | Some rows -> rows - | None -> rows_from_cardinality_one_candidates value_vars) - | (scan_value_var, scan_attr) :: remaining_value_vars - when constant_patterns <> [] - && List.length value_var_patterns >= 2 - && List.for_all (fun (_, attr) -> cardinality_one source_db attr) value_var_patterns -> - rows_from_cardinality_one_value_scan scan_value_var scan_attr remaining_value_vars - | _ :: _ - when constant_patterns <> [] - && List.for_all (fun (_, attr) -> cardinality_one source_db attr) value_var_patterns -> - rows_from_cardinality_one_candidates value_var_patterns - | (scan_value_var, scan_attr) :: remaining_value_vars - when direct_attr scan_attr - && List.for_all - (fun (_, attr) -> direct_attr attr && cardinality_one source_db attr) - remaining_value_vars -> - rows_from_cardinality_one_value_scan scan_value_var scan_attr remaining_value_vars - | (scan_value_var, scan_attr) :: remaining_value_vars - when List.for_all (fun (_, attr) -> cardinality_one source_db attr) value_var_patterns -> - rows_from_cardinality_one_value_scan scan_value_var scan_attr remaining_value_vars - | _ :: _ when List.for_all (fun (_, attr) -> cardinality_one source_db attr) value_var_patterns -> - rows_from_cardinality_one_candidates value_var_patterns - | _ -> - candidate_entities () - |> List.sort_uniq compare - |> List.concat_map (fun entity_id -> - if not (entity_allowed entity_id) then - [] - else - let bindings = - value_var_patterns - |> List.fold_left - (fun bindings value_pattern -> - match bindings with - | [] -> [] - | bindings -> extend_bindings bindings value_pattern) - [ [ e_var, Result_entity entity_id ] ] - in - bindings |> List.filter_map (binding_row attrs)) - in - let rows = - match constant_patterns, value_var_patterns, required_patterns, excluded_patterns, relation_comparisons with - | [ (attr, value) ], [], [], [], [] when value_var_patterns = [] -> ( - match avet_entity_ids attr value with - | Some entity_ids -> List.map (fun entity_id -> [ Result_entity entity_id ]) entity_ids - | None -> compute_default_rows ()) - | _ -> compute_default_rows () - in - let unique_rows = - (not source_db.history) - && source_db.duplicate_datoms = [] - && List.mem e_var attrs - && List.for_all (fun (_, attr) -> cardinality_one source_db attr) value_var_patterns + let pattern_relation (_, attr, value_term) = + relation_of_pattern db source [ QVar e_var; QAttr attr; value_term ] in - let relation = { attrs; rows; lookup_vars; unique_rows } in - Some - (List.fold_left - (fun relation -> function - | ComparisonPredicate (predicate, left_term, right_term) -> - filter_relation_comparison db relation predicate left_term right_term - | _ -> relation) - relation - relation_comparisons)) + (match + patterns + |> List.filter_map pattern_relation + |> function + | [] -> None + | first :: rest -> Some (List.fold_left hash_join first rest) + with + | None -> None + | Some relation -> + let relation = + excluded_patterns + |> List.fold_left + (fun relation (_, attr, value_term) -> + match relation_of_pattern db source [ QVar e_var; QAttr attr; value_term ] with + | Some excluded -> Option.value (anti_join relation excluded) ~default:relation + | None -> relation) + relation + in + let unique_rows = + (not source_db.history) + && source_db.duplicate_datoms = [] + && List.mem e_var attrs + in + let relation = { relation with lookup_vars; unique_rows } in + Some + (List.fold_left + (fun relation -> function + | ComparisonPredicate (predicate, left_term, right_term) -> + filter_relation_comparison db relation predicate left_term right_term + | _ -> relation) + relation + relation_comparisons)))) | _ -> None let relation_of_cross_entity_value_join _db source clauses = @@ -2894,11 +2248,35 @@ end) = struct in List.for_all (fun var -> List.mem var relation.attrs) value_vars + let same_entity_fused_relation db default_source clauses = + match relation_of_same_entity_patterns db default_source clauses with + | Some relation + when (relation.rows <> [] || not (relation_prefix_has_multiple_clauses clauses)) + && relation_value_vars_covered relation clauses -> + Some relation + | _ -> None + + let relation_of_single_aevt_var_var _db default_source clauses = + match clauses, default_source with + | [ Pattern (QVar e_var, QAttr attr, QVar v_var) ], Db_source source_db when e_var <> v_var -> + relation_of_aevt_var_var_pattern source_db e_var attr v_var + | _ -> None + let rec eval_relation_from_empty db sources default_source clauses = - (* Planner orders eligible clauses; relational fallback keeps source order. *) - let clauses = plan_ordered_clauses ~max_datom_e:db.max_datom_e clauses in - let clauses = promote_attr_binding_clauses clauses in - let rec apply relation = function + let fused_empty_relation clauses = + match same_entity_fused_relation db default_source clauses with + | Some _ as relation -> relation + | None -> ( + match relation_of_single_aevt_var_var db default_source clauses with + | Some _ as relation -> relation + | None -> ( + match relation_of_cross_entity_value_join db default_source clauses with + | Some relation when relation_value_vars_covered relation clauses -> Some relation + | _ -> eval_selective_or_join_value_pattern db sources default_source clauses)) + in + let run_interpreter clauses = + let clauses = promote_attr_binding_clauses clauses in + let rec apply relation = function | [] -> Some relation | _ when relation.rows = [] -> Some { relation with rows = []; unique_rows = true } | Pattern (e_term, a_term, v_term) :: ComparisonPredicate (predicate, left_term, right_term) :: rest -> @@ -3022,101 +2400,80 @@ end) = struct let* relation = anti_join relation excluded in apply relation rest | _ -> None + in + apply { attrs = []; rows = [ [] ]; lookup_vars = []; unique_rows = true } clauses in - match relation_of_same_entity_patterns db default_source clauses with - | Some relation - when (relation.rows <> [] || not (relation_prefix_has_multiple_clauses clauses)) - && relation_value_vars_covered relation clauses -> - Some relation - | _ -> ( - match relation_of_cross_entity_value_join db default_source clauses with - | Some relation when relation_value_vars_covered relation clauses -> Some relation - | _ -> ( - match - eval_selective_or_join_value_pattern db sources default_source clauses - with + match fused_empty_relation clauses with + | Some relation -> Some relation + | None -> ( + let planned = plan_ordered_clauses ~max_datom_e:db.max_datom_e clauses in + match fused_empty_relation planned with | Some relation -> Some relation | None -> ( - match clauses with + match planned with | [ Or branches ] -> eval_or_branch_relations db sources default_source branches | [ SourceOr (source_name, branches) ] -> let default_source = source db sources source_name in eval_or_branch_relations db sources default_source branches | [ OrJoin (vars, branches) ] -> Query.ensure_or_join_branches_cover_listed_vars [] vars branches; - let* relation = eval_or_branch_relations db sources default_source branches in - Some (project_relation vars relation) + eval_or_join_relations db sources default_source vars branches | [ SourceOrJoin (source_name, vars, branches) ] -> let default_source = source db sources source_name in Query.ensure_or_join_branches_cover_listed_vars [] vars branches; - let* relation = eval_or_branch_relations db sources default_source branches in - Some (project_relation vars relation) - | _ -> - apply { attrs = []; rows = [ [] ]; lookup_vars = []; unique_rows = true } clauses))) + eval_or_join_relations db sources default_source vars branches + | _ -> run_interpreter planned)) and or_join_constant_entity_branch e_var = function | [ Pattern (QVar branch_e, QAttr _, QValue _) ] when branch_e = e_var -> true | _ -> false - and eval_selective_or_join_value_pattern db sources default_source clauses = + and eval_selective_or_join_value_pattern _db _sources default_source clauses = let try_shape e_var attr value_term branches = match value_term, default_source with | QVar value_var, Db_source source_db when value_var <> e_var && List.for_all (or_join_constant_entity_branch e_var) branches -> - let branch_relations = + let branch_constants = branches - |> List.filter_map (fun branch_clauses -> - eval_relation_from_empty db sources default_source branch_clauses) + |> List.filter_map (function + | [ Pattern (QVar branch_entity, QAttr branch_attr, QValue branch_value) ] + when branch_entity = e_var && branch_attr <> attr -> + Some (branch_attr, branch_value) + | _ -> None) in - (match branch_relations with - | [] -> - let attrs = unique_vars [ QVar e_var; QAttr attr; QVar value_var ] in - let lookup_vars = relation_lookup_vars source_db [ QVar e_var; QWildcard; QWildcard ] in - Some { attrs; rows = []; lookup_vars; unique_rows = true } - | first :: rest -> - let united_rows = - List.fold_left - (fun acc rel -> - if rel.attrs <> first.attrs then acc else List.rev_append rel.rows acc) - (List.rev first.rows) - rest - |> List.rev - in - let e_index_opt = - let rec loop index = function - | [] -> None - | candidate :: _ when candidate = e_var -> Some index - | _ :: rest -> loop (index + 1) rest - in - loop 0 first.attrs - in - match e_index_opt with - | None -> None - | Some e_index -> - let seen = Hashtbl.create (List.length united_rows) in - let entity_ids = - united_rows - |> List.filter_map (fun row -> - match row_value row e_index with - | Result_entity entity_id -> - if Hashtbl.mem seen entity_id then - None - else ( - Hashtbl.add seen entity_id (); - Some entity_id) - | _ -> None) - in - let attrs = unique_vars [ QVar e_var; QAttr attr; QVar value_var ] in - let lookup_vars = relation_lookup_vars source_db [ QVar e_var; QWildcard; QWildcard ] in - let rows = - entity_ids - |> List.filter_map (fun entity_id -> - match find_entity_attr_value source_db entity_id attr with - | None -> None - | Some value -> - binding_row attrs [ e_var, Result_entity entity_id; value_var, value ]) - in - Some { attrs; rows; lookup_vars; unique_rows = true }) + if branch_constants = [] then + None + else + let entity_ids = + branch_constants + |> List.concat_map (fun (branch_attr, branch_value) -> + match entity_ids_by_attr_value source_db branch_attr branch_value with + | Some ids -> ids + | None -> + datoms_by_attr_value source_db branch_attr branch_value + |> List.map (fun datom -> datom.e)) + |> List.sort_uniq compare + in + let attrs = unique_vars [ QVar e_var; QAttr attr; QVar value_var ] in + let lookup_vars = relation_lookup_vars source_db [ QVar e_var; QWildcard; QWildcard ] in + if entity_ids = [] then + Some { attrs; rows = []; lookup_vars; unique_rows = true } + else + let rows = + entity_ids + |> List.filter_map (fun entity_id -> + match find_entity_attr_value source_db entity_id attr with + | None -> None + | Some value -> + match attrs with + | [ left; right ] when left = e_var && right = value_var -> + Some [ Result_entity entity_id; value ] + | [ left; right ] when left = value_var && right = e_var -> + Some [ value; Result_entity entity_id ] + | _ -> + binding_row attrs [ e_var, Result_entity entity_id; value_var, value ]) + in + Some { attrs; rows; lookup_vars; unique_rows = true } | _ -> None in match clauses with @@ -3128,8 +2485,41 @@ end) = struct try_shape e_var attr value_term branches | _ -> None - and eval_or_branch_relations db sources default_source branches = - Query.ensure_or_branch_vars_match ~value_to_string:edn_string_of_value [] branches; + and eval_or_join_relations db sources default_source vars branches = + match + branches + |> List.filter_map (fun branch_clauses -> + eval_relation_from_empty db sources default_source branch_clauses + |> Option.map (project_relation vars)) + with + | [] -> Some { attrs = vars; rows = []; lookup_vars = []; unique_rows = true } + | first :: rest -> + let rows = + List.fold_left + (fun acc rel -> List.rev_append rel.rows acc) + (List.rev first.rows) + rest + |> List.rev + in + let lookup_vars = + List.fold_left + (fun lookup_vars rel -> + List.fold_left + (fun lookup_vars ((var, _) as lookup_var) -> + if List.mem_assoc var lookup_vars then lookup_vars else lookup_var :: lookup_vars) + lookup_vars + rel.lookup_vars) + first.lookup_vars + rest + in + let unique_rows = + List.fold_left (fun unique rel -> unique && rel.unique_rows) first.unique_rows rest + in + Some { attrs = vars; rows; lookup_vars; unique_rows } + + and eval_or_branch_relations ?(require_matching_vars = true) db sources default_source branches = + if require_matching_vars then + Query.ensure_or_branch_vars_match ~value_to_string:edn_string_of_value [] branches; match branches |> List.filter_map (fun branch_clauses -> eval_relation_from_empty db sources default_source branch_clauses) @@ -3162,18 +2552,55 @@ end) = struct let eval_relation_rows db sources rules bindings clauses = let default_source = source db sources "$" in - match expand_inline_rules rules clauses with - | None -> None - | Some clauses -> - (match bindings, relation_query_clauses clauses with - | [ [] ], true -> - eval_relation_from_empty db sources default_source clauses - |> Option.map (fun relation -> relation.attrs, relation.rows, relation.unique_rows) - | [ binding ], true -> - let clauses = List.map (bound_relation_clause binding) clauses in - eval_relation_from_empty db sources default_source clauses - |> Option.map (fun relation -> relation.attrs, relation.rows, relation.unique_rows) - | _ -> None) + let try_single_pattern_rule_rows = + match bindings, clauses with + | [ [] ], [ Rule (name, terms) ] -> ( + match inline_rule_clauses rules name terms with + | Some [ Pattern (QVar _, QAttr attr, QVar _) ] -> ( + match default_source with + | Db_source source_db -> ( + match aevt_attr_array source_db attr with + | None -> None + | Some arr -> + let attrs = List.map (function QVar var -> var | _ -> "") terms in + let rows = ref [] in + let collect datom = + match datom.v with + | Ref target -> rows := [ Result_entity datom.e; Result_entity target ] :: !rows + | _ -> () + in + for i = Array.length arr - 1 downto 0 do + collect arr.(i) + done; + (match aevt_duplicate_datoms source_db attr with + | [] -> () + | duplicates -> List.iter collect duplicates); + Some (attrs, !rows, true)) + | _ -> None) + | _ -> None) + | _ -> None + in + match try_single_pattern_rule_rows with + | Some result -> Some result + | None -> ( + let continue () = + match if rules = [] then Some clauses else expand_inline_rules rules clauses with + | None -> None + | Some clauses -> + (match bindings, relation_query_clauses clauses with + | [ [] ], true -> ( + match same_entity_fused_relation db default_source clauses with + | Some relation -> Some (relation.attrs, relation.rows, relation.unique_rows) + | None -> + eval_relation_from_empty db sources default_source clauses + |> Option.map (fun relation -> relation.attrs, relation.rows, relation.unique_rows)) + | [ binding ], true -> + let clauses = List.map (bound_relation_clause binding) clauses in + eval_relation_from_empty db sources default_source clauses + |> Option.map (fun relation -> relation.attrs, relation.rows, relation.unique_rows) + | _ -> None) + in + continue ()) let eval_relation_clauses ?(allow_initial_bindings = false) db sources default_source bindings clauses = let bound_relation_pattern_terms = function diff --git a/test/debug_wildcard_pull_slowcase.ml b/test/debug_wildcard_pull_slowcase.ml new file mode 100644 index 0000000..604fa8e --- /dev/null +++ b/test/debug_wildcard_pull_slowcase.ml @@ -0,0 +1,66 @@ +open Datascript + +let ref_many = + { cardinality = Many + ; unique = None + ; indexed = false + ; is_component = false + ; no_history = false + ; doc = None + ; value_type = Some RefType + ; tuple_attrs = None + ; tuple_types = None + } + +let ref_one = { ref_many with cardinality = One } + +let one = { ref_many with cardinality = One; value_type = None } + +let test_wildcard_pull_page_missing () = + let page_count = 20 in + let noise_count = 500_000 in + let page_datoms = + List.concat + (List.init page_count (fun index -> + let page = 1_000 + index in + let block = 10_000 + index in + [ datom ~e:page ~a:"block/name" ~v:(String (Printf.sprintf "page-%d" index)) () + ; datom ~e:page ~a:"block/title" ~v:(String (Printf.sprintf "Page %d" index)) () + ; datom ~e:block ~a:"block/title" ~v:(String (Printf.sprintf "Block %d" index)) () + ; datom ~e:block ~a:"block/page" ~v:(Ref page) () + ])) + in + let noise_datoms = + List.init noise_count (fun index -> + datom ~e:(100_000 + index) ~a:"noise/value" ~v:(String (Printf.sprintf "noise-%d" index)) ()) + in + let db = + init_db + ~schema: + [ "block/name", one + ; "block/title", one + ; "block/page", ref_one + ; "logseq.property/built-in?", one + ; "noise/value", one + ] + (page_datoms @ noise_datoms) + in + Printf.eprintf "[repro] db max_e=%d\n%!" db.max_datom_e; + let started = Unix.gettimeofday () in + let result = + q_return_string + db + "[:find (pull ?p [*]) :where [?b :block/title] [?b :block/page ?p] [(missing? $ ?p :logseq.property/built-in?)]]" + in + let elapsed = Unix.gettimeofday () -. started in + match result with + | Query_relation rows -> + Printf.eprintf "[repro] elapsed=%.3fs rows=%d\n%!" elapsed (List.length rows); + if elapsed > 3.0 then ( + Printf.eprintf "[repro] FAIL: exceeded 3s threshold\n%!"; + exit 1) + | _ -> + Printf.eprintf "[repro] unexpected result\n%!"; + exit 1 + +let () = test_wildcard_pull_page_missing () diff --git a/test/dune b/test/dune index a368553..d2b00cf 100644 --- a/test/dune +++ b/test/dune @@ -210,6 +210,11 @@ (enabled_if false) (libraries datascript-ocaml-native unix)) +(executable + (name debug_wildcard_pull_slowcase) + (modules debug_wildcard_pull_slowcase) + (libraries datascript-ocaml-native unix)) + (test (name test_storage) (modules test_storage) diff --git a/test/test_query_plan.ml b/test/test_query_plan.ml index c0ac6e5..fc1eafc 100644 --- a/test/test_query_plan.ml +++ b/test/test_query_plan.ml @@ -100,6 +100,21 @@ let test_logical_entity_join () = check_int "filter attached to entity join" 1 (List.length filters) | _ -> failwith "expected LEntityJoin") +let test_logical_not_join_fold () = + let clauses = + [ Pattern (QVar "?e", QAttr "age", QVar "?a") + ; NotJoin ([ "?e" ], [ Pattern (QVar "?e", QAttr "sex", QValue (Keyword "male")) ]) + ] + in + match Query_plan.build_logical_plan clauses with + | None -> failwith "expected logical plan" + | Some logical -> ( + match logical.nodes with + | [ Query_plan.LEntityJoin { scans; anti_scans; _ } ] -> + check_int "not-join folds to entity anti-scan" 1 (List.length anti_scans); + check_int "positive scan in entity join" 1 (List.length scans) + | _ -> failwith "expected LEntityJoin with folded not-join") + let () = run "query plan" [ ( "analyze" @@ -108,5 +123,6 @@ let () = ; test_case "analyze same-entity merge" `Quick test_analyze_same_entity_merge ; test_case "analyze benchmark shapes" `Quick test_analyze_benchmark_shapes ; test_case "logical entity join attaches filters" `Quick test_logical_entity_join + ; test_case "logical not-join folds to anti-scan" `Quick test_logical_not_join_fold ] ) ]