diff --git a/CHANGELOG.md b/CHANGELOG.md index b6eae21..9e4c2aa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,57 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## v0.5.0 (2026-06-21) + +### Added + +- **Query Planner** (`ExDatalog.Planner`) — a thin planning layer between the + compiled IR and the engine. `plan/2` produces an `ExDatalog.Planner.Plan` + (strategy, strata, joins, predicates); `explain_plan/1,2` renders a + human-readable plan. Emits `[:ex_datalog, :planner, :start|:stop|:exception]` + telemetry. +- **Aggregates** — `count`, `sum`, `min`, `max` in rule bodies, via the DSL + (`count(X, N)`) or the builder API (`Constraint.count/2`, `from_tuple({:count, + X, N})`, `add_rule/4`). Aggregates group surviving bindings by the head + variables other than the result, reduce each group, and are stratified + strictly above their source relations. Integer-only; one aggregate per rule. +- **BEAM callback predicates** — call deterministic, side-effect-free Elixir + functions from rule bodies. DSL: `predicate :name, Module, :fun, [types], + :boolean | :value`. Builder: `ExDatalog.Callback` literals in a rule body. + Boolean callbacks filter; `:value` callbacks bind a result variable. The + engine isolates callbacks with a configurable timeout + (`:callback_timeout_ms`, default 100ms) and rescues exceptions — a + timeout/raise filters the binding. +- **Magic sets** (experimental) — demand-driven evaluation via + `materialize(program, strategy: :magic_sets, goal: {relation, pattern})`. The + IR is rewritten to compute only facts relevant to the goal. Supports positive + recursive programs with ground bound positions; unsupported programs fall back + to semi-naive evaluation (never producing incorrect results). +- `ExDatalog.Callback` and `ExDatalog.IR.Callback` for callback predicates. +- `ExDatalog.Constraints.Aggregate` and `ExDatalog.Constraints.BeamCallback`. +- `Capabilities` gains `aggregate_constraints` and `beam_callbacks` fields. +- `Constraint.Context` gains an `opts` field (threads evaluation options such + as `callback_timeout_ms`). +- Educational articles 06–09 (planner, aggregates, callbacks, magic sets) and a + v0.4 → v0.5 migration guide. +- Benchmark harness (`bench/`) and report (`chest/benchmarks/v0.5.0-report.md`). + +### Changed + +- Version bumped from 0.4.1 to 0.5.0. +- `Engine.Naive.evaluate/2` dispatches on a `:strategy` option (`:semi_naive` + default, or `:magic_sets`). +- `Engine.Naive` stratification validation now also rejects aggregate rules + whose source relations are not in a strictly lower stratum. +- The `agg(...)` placeholder now points users to `count/sum/min/max`. + +### Notes + +- Aggregates are integer-only; `avg` and float support are deferred to v0.6.0. +- Magic sets is experimental and opt-in; the default strategy is unchanged. +- All v0.4.1 tests continue to pass. Suite: 845 tests, 10 properties, + 149 doctests, ~93% coverage. + ## v0.4.1 (2026-06-21) ### Fixed diff --git a/README.md b/README.md index d1ff650..6294d1e 100644 --- a/README.md +++ b/README.md @@ -36,14 +36,18 @@ It continues to influence modern databases, compilers, static analysis tools, kn - **Builder API** for constructing programs (relations, facts, rules, constraints) - **Schema DSL** — Ecto-inspired macros for declaring relations, facts, rules, and queries - **Constraint types**: comparisons, arithmetic, type predicates, string predicates, membership +- **Aggregates**: `count`, `sum`, `min`, `max` with grouping and stratification +- **BEAM callback predicates**: call deterministic Elixir functions from rules (timeout-isolated) - **Negation** with stratified evaluation - **Recursive rules** with semi-naive fixpoint evaluation +- **Query planner** (`ExDatalog.Planner`) with `explain_plan/1,2` +- **Magic sets** (experimental): demand-driven evaluation via `strategy: :magic_sets` - **Post-materialization queries** (`query` macro with `find`/`where`) - **Pluggable storage backends**: `Storage.Map` (default, on-heap) and `Storage.ETS` (off-heap, concurrent reads) - **Provenance / derivation explain** (`explain: true`) -- **Telemetry** integration (`:telemetry` events for query lifecycle) +- **Telemetry** integration (`:telemetry` events for query/planner lifecycle) - **Deterministic**: same program + same facts = same result regardless of backend -- 792 tests, 0 failures, credo clean +- 845 tests, 0 failures, credo clean ## Installation @@ -52,7 +56,7 @@ Add `ex_datalog` to your dependencies in `mix.exs`: ```elixir def deps do [ - {:ex_datalog, "~> 0.4.1"} + {:ex_datalog, "~> 0.5.0"} ] end ``` @@ -229,6 +233,84 @@ Program.add_rule(program, {"hello_word", [:X]}, ) ``` +### Aggregates + +Group and reduce bindings with `count`, `sum`, `min`, `max`: + +```elixir +defmodule DeptStats do + use ExDatalog.Schema + + relation "emp", [:atom, :atom] + relation "dept_count", [:atom, :integer] + + fact "emp", [:alice, :eng] + fact "emp", [:bob, :eng] + fact "emp", [:carol, :sales] + + rule dept_count(D, N) do + emp(_, D) + count(E, N) + end +end + +{:ok, k} = ExDatalog.materialize(DeptStats) +Knowledge.get(k, "dept_count") +#=> MapSet.new([{:eng, 2}, {:sales, 1}]) +``` + +### BEAM callback predicates + +Call deterministic Elixir functions from rules: + +```elixir +defmodule Gated do + use ExDatalog.Schema + + relation "user", [:atom] + relation "active_user", [:atom] + + predicate :is_adult, AgeChecker, :adult?, [:atom], :boolean + + fact "user", [:alice] + fact "user", [:bob] + + rule active_user(U) do + user(U) + is_adult(U) + end +end + +defmodule AgeChecker do + def adult?(:alice), do: true + def adult?(:bob), do: false +end + +{:ok, k} = ExDatalog.materialize(Gated) +Knowledge.get(k, "active_user") +#=> MapSet.new([{:alice}]) +``` + +Value-returning callbacks bind a result variable: + +```elixir +predicate :length_of, StringLength, :compute, [:atom], :value +``` + +### Magic sets (experimental) + +Compute only facts relevant to a goal, instead of the full fixpoint: + +```elixir +{:ok, k} = ExDatalog.materialize(program, + strategy: :magic_sets, + goal: {"ancestor", [:alice, :_]} +) +``` + +Unsupported programs (negation, aggregates) automatically fall back to +semi-naive evaluation. + ### Negation Use negative body atoms with stratified evaluation. Find people who are not parents: @@ -283,24 +365,27 @@ end ## Architecture ``` -ExDatalog.Program (builder) - | - v +ExDatalog.Program (builder) / ExDatalog.Schema (DSL) + | + v ExDatalog.Validator (structural + semantic + stratification) - | - v + | + v ExDatalog.Compiler (AST -> IR) - | - v + | + v +ExDatalog.Planner (strategy, strata, joins) + | + v ExDatalog.Engine (behaviour) - | - v -ExDatalog.Engine.Naive (semi-naive fixpoint) - | - v uses + | + v +ExDatalog.Engine.Naive (semi-naive fixpoint / magic-sets) + | + v uses ExDatalog.Storage.Map | ExDatalog.Storage.ETS - | - v + | + v ExDatalog.Knowledge ``` @@ -343,6 +428,7 @@ reference. | Type predicate | `type_integer`, `type_binary`, `type_atom` | No (filter) | | String predicate | `starts_with`, `contains` | No (filter) | | Membership | `member` | No (filter) | +| Aggregate | `count`, `sum`, `min`, `max` | Yes (group-and-reduce) | ## Documentation @@ -350,7 +436,8 @@ reference. - [Constraints](docs/constraints.md) — constraint types, evaluation, and the dispatch model - [Storage Backends](docs/storage_backends.md) — Map vs ETS, options, capabilities, determinism guarantee - [Migration: Builder API → DSL](docs/migration_dsl.md) — migrate existing builder-API code to the Schema DSL -- [DSL Articles](docs/articles/01_why_datalog_on_the_beam.md) — why Datalog on the BEAM, building the DSL, rules as macros, queries, negation +- [Migration: v0.4 → v0.5](docs/migration_v0.5.md) — aggregates, callbacks, magic sets, and planner changes +- [DSL Articles](docs/articles/01_why_datalog_on_the_beam.md) — why Datalog on the BEAM, building the DSL, rules as macros, queries, negation, planning, aggregates, callbacks, magic sets - [Quickstart Tutorial](livebooks/quickstart.livemd) — interactive Livebook walkthrough - [DSL Tutorial](livebooks/ex_datalog_dsl.livemd) — interactive DSL walkthrough - [Examples](livebooks/examples.livemd) — 10 realistic use cases (RBAC, supply chain, fraud detection, and more) @@ -423,7 +510,7 @@ The following references are highly recommended for understanding both the theor | v0.3.0 | Tuple shorthand for rules (`add_rule/3`, `add_rule/4`), `Term.from/1`, `ExDatalog.Atom.from_tuple/1`, `Constraint.from_tuple/1`; renamed `Result` → `Knowledge`, `query` → `materialize` | | v0.4.0 | Schema DSL (`use ExDatalog.Schema`), `relation`, `fact`, `rule`, `query` macros, `not_` negation, constraint DSL, post-materialization queries | | v0.4.1 | DSL review fixes: correct uppercase-variable docs, clean aggregate error, query/find validation, unified `DSL.CompileError`, 792 tests | -| v0.5.0 | Magic sets / demand-driven evaluation, general predicates as BEAM callbacks | +| v0.5.0 | Aggregates (`count`/`sum`/`min`/`max`), BEAM callback predicates, query planner, magic sets (experimental), 845 tests | | v1.0.0 | Stable public API, hardened production semantics | ## License diff --git a/bench/aggregate_bench.exs b/bench/aggregate_bench.exs new file mode 100644 index 0000000..9905920 --- /dev/null +++ b/bench/aggregate_bench.exs @@ -0,0 +1,82 @@ +alias ExDatalog +alias ExDatalog.{Program, Rule, Atom, Term, Constraint} + +{n, m} = {200, 5} + +departments = for i <- 0..(m - 1), do: String.to_atom("dept_#{i}") + +base_program = + Program.new() + |> Program.add_relation("emp", [:atom, :atom]) + |> Program.add_relation("salary", [:atom, :atom, :integer]) + +base_program = + Enum.reduce(1..n, base_program, fn i, acc -> + dept = Enum.at(departments, rem(i, m)) + salary = 50_000 + i * 100 + name = String.to_atom("emp_#{i}") + + acc + |> Program.add_fact("emp", [name, dept]) + |> Program.add_fact("salary", [name, dept, salary]) + end) + +program_no_aggregates = + base_program + |> Program.add_relation("emp_salary", [:atom, :atom, :integer]) + |> Program.add_rule( + Rule.new( + Atom.new("emp_salary", [Term.var("E"), Term.var("D"), Term.var("S")]), + [{:positive, Atom.new("salary", [Term.var("E"), Term.var("D"), Term.var("S")])}] + ) + ) + +program_with_aggregates = + base_program + |> Program.add_relation("dept_count", [:atom, :integer]) + |> Program.add_relation("dept_total", [:atom, :integer]) + |> Program.add_relation("dept_min_salary", [:atom, :integer]) + |> Program.add_relation("dept_max_salary", [:atom, :integer]) + |> Program.add_rule( + Rule.new( + Atom.new("dept_count", [Term.var("D"), Term.var("N")]), + [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}], + [Constraint.count(Term.var("E"), Term.var("N"))] + ) + ) + |> Program.add_rule( + Rule.new( + Atom.new("dept_total", [Term.var("D"), Term.var("T")]), + [{:positive, Atom.new("salary", [Term.var("E"), Term.var("D"), Term.var("A")])}], + [Constraint.sum(Term.var("A"), Term.var("T"))] + ) + ) + |> Program.add_rule( + Rule.new( + Atom.new("dept_min_salary", [Term.var("D"), Term.var("V")]), + [{:positive, Atom.new("salary", [Term.var("E"), Term.var("D"), Term.var("S")])}], + [Constraint.min(Term.var("S"), Term.var("V"))] + ) + ) + |> Program.add_rule( + Rule.new( + Atom.new("dept_max_salary", [Term.var("D"), Term.var("V")]), + [{:positive, Atom.new("salary", [Term.var("E"), Term.var("D"), Term.var("S")])}], + [Constraint.max(Term.var("S"), Term.var("V"))] + ) + ) + +Benchee.run( + %{ + "no_aggregates" => fn -> + {:ok, _knowledge} = ExDatalog.materialize(program_no_aggregates) + end, + "with_aggregates_count_sum_min_max" => fn -> + {:ok, _knowledge} = ExDatalog.materialize(program_with_aggregates) + end + }, + time: 10, + memory_time: 2, + formatters: [{Benchee.Formatters.Console, comparison: true}], + print: [fast_warning: false] +) diff --git a/bench/magic_sets_bench.exs b/bench/magic_sets_bench.exs new file mode 100644 index 0000000..bdf3781 --- /dev/null +++ b/bench/magic_sets_bench.exs @@ -0,0 +1,51 @@ +alias ExDatalog +alias ExDatalog.{Program, Rule, Atom, Term} + +n = 50 + +chain_facts = for i <- 0..(n - 2), do: {String.to_atom("n#{i}"), String.to_atom("n#{i + 1}")} + +program = + Program.new() + |> Program.add_relation("parent", [:atom, :atom]) + |> Program.add_relation("ancestor", [:atom, :atom]) + +program = + Enum.reduce(chain_facts, program, fn {p, c}, acc -> + Program.add_fact(acc, "parent", [p, c]) + end) + +program = + program + |> Program.add_rule( + Rule.new( + Atom.new("ancestor", [Term.var("X"), Term.var("Y")]), + [{:positive, Atom.new("parent", [Term.var("X"), Term.var("Y")])}] + ) + ) + |> Program.add_rule( + Rule.new( + Atom.new("ancestor", [Term.var("X"), Term.var("Z")]), + [ + {:positive, Atom.new("parent", [Term.var("X"), Term.var("Y")])}, + {:positive, Atom.new("ancestor", [Term.var("Y"), Term.var("Z")])} + ] + ) + ) + +goal = {"ancestor", [String.to_atom("n0"), :_]} + +Benchee.run( + %{ + "semi_naive_full" => fn -> + {:ok, _knowledge} = ExDatalog.materialize(program) + end, + "magic_sets_goal_driven" => fn -> + {:ok, _knowledge} = ExDatalog.materialize(program, strategy: :magic_sets, goal: goal) + end + }, + time: 10, + memory_time: 2, + formatters: [{Benchee.Formatters.Console, comparison: true}], + print: [fast_warning: false] +) diff --git a/lib/ex_datalog/callback.ex b/lib/ex_datalog/callback.ex new file mode 100644 index 0000000..6bcd5d0 --- /dev/null +++ b/lib/ex_datalog/callback.ex @@ -0,0 +1,106 @@ +defmodule ExDatalog.Callback do + @moduledoc """ + A BEAM callback predicate: an Elixir function invoked during rule evaluation. + + Callbacks let rule bodies call deterministic, side-effect-free Elixir + functions as predicates. A callback appears in a rule body as + `{:callback, %ExDatalog.Callback{}}`, alongside positive/negative atoms. + + ## Fields + + - `module` — the module exporting the function. + - `function` — the function name (atom). + - `args` — a list of `ExDatalog.Term.t()` (variables/constants) resolved + against the binding and passed positionally to the function. + - `result` — `nil` for boolean (filter) callbacks; `{:var, name}` for + value-returning callbacks that bind the function's return value. + + ## Safety contract + + A callback **must** be: + + - **Deterministic** — the same arguments always produce the same result. + - **Side-effect free** — no I/O, mutation, or messaging. + + The engine enforces only timeout and exception isolation: a callback that + exceeds `callback_timeout_ms` (default 100ms) or raises is treated as a + filtered binding (the fact is not derived). Determinism and purity are + caller contracts, not enforced by the engine. + + ## Examples + + iex> alias ExDatalog.{Callback, Term} + iex> Callback.new(String, :starts_with?, [Term.var("S"), Term.const("a")]) + %ExDatalog.Callback{module: String, function: :starts_with?, args: [{:var, "S"}, {:const, "a"}], result: nil} + + """ + + alias ExDatalog.Term + + @enforce_keys [:module, :function, :args] + defstruct [:module, :function, :args, :result] + + @type t :: %__MODULE__{ + module: module(), + function: atom(), + args: [Term.t()], + result: Term.t() | nil + } + + @doc """ + Constructs a callback predicate. + + `result` defaults to `nil` (a boolean filter callback). Pass a + `{:var, name}` term to bind the function's return value. + + ## Examples + + iex> alias ExDatalog.{Callback, Term} + iex> Callback.new(MyMod, :adult?, [Term.var("Age")]) + %ExDatalog.Callback{module: MyMod, function: :adult?, args: [{:var, "Age"}], result: nil} + + iex> alias ExDatalog.{Callback, Term} + iex> Callback.new(MyMod, :score, [Term.var("X")], Term.var("S")) + %ExDatalog.Callback{module: MyMod, function: :score, args: [{:var, "X"}], result: {:var, "S"}} + + """ + @spec new(module(), atom(), [Term.t()], Term.t() | nil) :: t() + def new(module, function, args, result \\ nil) + when is_atom(module) and is_atom(function) and is_list(args) do + %__MODULE__{module: module, function: function, args: args, result: result} + end + + @doc """ + Returns the input variable names referenced by the callback's arguments. + + These must be bound by positive body atoms before the callback runs. + + ## Examples + + iex> alias ExDatalog.{Callback, Term} + iex> cb = Callback.new(MyMod, :ok?, [Term.var("A"), Term.const(1), Term.var("B")]) + iex> ExDatalog.Callback.input_variables(cb) + ["A", "B"] + + """ + @spec input_variables(t()) :: [String.t()] + def input_variables(%__MODULE__{args: args}), do: Term.variables(args) + + @doc """ + Returns the result variable name for a value-returning callback, or `nil`. + + ## Examples + + iex> alias ExDatalog.{Callback, Term} + iex> ExDatalog.Callback.result_variable(Callback.new(M, :f, [Term.var("X")], Term.var("R"))) + "R" + + iex> alias ExDatalog.{Callback, Term} + iex> ExDatalog.Callback.result_variable(Callback.new(M, :f, [Term.var("X")])) + nil + + """ + @spec result_variable(t()) :: String.t() | nil + def result_variable(%__MODULE__{result: {:var, name}}), do: name + def result_variable(%__MODULE__{result: _}), do: nil +end diff --git a/lib/ex_datalog/capabilities.ex b/lib/ex_datalog/capabilities.ex index fe2c635..608e6de 100644 --- a/lib/ex_datalog/capabilities.ex +++ b/lib/ex_datalog/capabilities.ex @@ -18,6 +18,8 @@ defmodule ExDatalog.Capabilities do | `type_predicates` | `true` | Supports type-check predicates | | `string_predicates` | `true` | Supports string predicates | | `provenance` | `true` | Supports derivation provenance | + | `aggregate_constraints` | `true` | Supports aggregate constraints (count/sum/min/max) | + | `beam_callbacks` | `true` | Supports BEAM callback predicates | | `external_execution` | `false` | Reserved for Z3, Soufflé, etc. | ## Merging @@ -61,6 +63,8 @@ defmodule ExDatalog.Capabilities do type_predicates: boolean(), string_predicates: boolean(), provenance: boolean(), + aggregate_constraints: boolean(), + beam_callbacks: boolean(), external_execution: boolean() } @@ -72,6 +76,8 @@ defmodule ExDatalog.Capabilities do type_predicates: true, string_predicates: true, provenance: true, + aggregate_constraints: true, + beam_callbacks: true, external_execution: false @boolean_fields [ @@ -82,6 +88,8 @@ defmodule ExDatalog.Capabilities do :type_predicates, :string_predicates, :provenance, + :aggregate_constraints, + :beam_callbacks, :external_execution ] diff --git a/lib/ex_datalog/compiler.ex b/lib/ex_datalog/compiler.ex index bf4e83c..74fd0ea 100644 --- a/lib/ex_datalog/compiler.ex +++ b/lib/ex_datalog/compiler.ex @@ -200,6 +200,7 @@ defmodule ExDatalog.Compiler do Enum.map(body, fn {:positive, %ExDatalog.Atom{} = atom} -> {:positive, IR.from_atom(atom)} {:negative, %ExDatalog.Atom{} = atom} -> {:negative, IR.from_atom(atom)} + {:callback, %ExDatalog.Callback{} = cb} -> {:callback, IR.from_callback(cb)} end) ir_constraints = Enum.map(constraints, fn c -> {:constraint, IR.from_constraint(c)} end) diff --git a/lib/ex_datalog/constraint.ex b/lib/ex_datalog/constraint.ex index 43ceeee..c8533d4 100644 --- a/lib/ex_datalog/constraint.ex +++ b/lib/ex_datalog/constraint.ex @@ -119,7 +119,9 @@ defmodule ExDatalog.Constraint do @type_ops [:is_integer, :is_binary, :is_atom] @string_ops [:starts_with, :contains] @membership_ops [:member] - @all_ops @comparison_ops ++ @arithmetic_ops ++ @type_ops ++ @string_ops ++ @membership_ops + @aggregate_ops [:count, :sum, :min, :max] + @all_ops @comparison_ops ++ + @arithmetic_ops ++ @type_ops ++ @string_ops ++ @membership_ops ++ @aggregate_ops @type op :: :gt @@ -138,6 +140,10 @@ defmodule ExDatalog.Constraint do | :starts_with | :contains | :member + | :count + | :sum + | :min + | :max @type comparison :: %__MODULE__{ op: :gt | :lt | :gte | :lte | :eq | :neq, @@ -174,7 +180,20 @@ defmodule ExDatalog.Constraint do result: nil } - @type t :: comparison() | arithmetic() | type_predicate() | string_predicate() | membership() + @type aggregate :: %__MODULE__{ + op: :count | :sum | :min | :max, + left: Term.t(), + right: nil, + result: {:var, String.t()} + } + + @type t :: + comparison() + | arithmetic() + | type_predicate() + | string_predicate() + | membership() + | aggregate() defstruct [:op, :left, :right, :result] @@ -402,6 +421,63 @@ defmodule ExDatalog.Constraint do @spec member(Term.t(), Term.t()) :: t() def member(left, right), do: filter(:member, left, right) + # --- Aggregate constructors --- + + @doc """ + Constructs a `count` aggregate: `result = count of input within each group`. + + Aggregates group the rule's surviving bindings by the head variables other + than `result`, then reduce the `input` variable's values within each group. + Aggregates are integer-only and must appear in a stratum strictly above the + relations that bind their inputs. + + ## Examples + + iex> ExDatalog.Constraint.count({:var, "Emp"}, {:var, "N"}) + %ExDatalog.Constraint{op: :count, left: {:var, "Emp"}, right: nil, result: {:var, "N"}} + + """ + @spec count(Term.t(), Term.t()) :: t() + def count(input, result), do: aggregate(:count, input, result) + + @doc """ + Constructs a `sum` aggregate: `result = integer sum of input within each group`. + + Inputs must be integers; non-integer inputs filter the group's contribution. + + ## Examples + + iex> ExDatalog.Constraint.sum({:var, "Amount"}, {:var, "Total"}) + %ExDatalog.Constraint{op: :sum, left: {:var, "Amount"}, right: nil, result: {:var, "Total"}} + + """ + @spec sum(Term.t(), Term.t()) :: t() + def sum(input, result), do: aggregate(:sum, input, result) + + @doc """ + Constructs a `min` aggregate: `result = minimum of input within each group`. + + ## Examples + + iex> ExDatalog.Constraint.min({:var, "Score"}, {:var, "Lowest"}) + %ExDatalog.Constraint{op: :min, left: {:var, "Score"}, right: nil, result: {:var, "Lowest"}} + + """ + @spec min(Term.t(), Term.t()) :: t() + def min(input, result), do: aggregate(:min, input, result) + + @doc """ + Constructs a `max` aggregate: `result = maximum of input within each group`. + + ## Examples + + iex> ExDatalog.Constraint.max({:var, "Score"}, {:var, "Highest"}) + %ExDatalog.Constraint{op: :max, left: {:var, "Score"}, right: nil, result: {:var, "Highest"}} + + """ + @spec max(Term.t(), Term.t()) :: t() + def max(input, result), do: aggregate(:max, input, result) + # --- Introspection --- @doc """ @@ -479,6 +555,49 @@ defmodule ExDatalog.Constraint do @spec membership?(t()) :: boolean() def membership?(%__MODULE__{op: op}), do: op in @membership_ops + @doc """ + Returns `true` if the constraint is an aggregate (`count`, `sum`, `min`, `max`). + + Aggregates bind a `result` variable by grouping and reducing, and are + evaluated by a dedicated engine path rather than the per-binding constraint + pipeline. + + ## Examples + + iex> ExDatalog.Constraint.aggregate?(ExDatalog.Constraint.count({:var, "X"}, {:var, "N"})) + true + + iex> ExDatalog.Constraint.aggregate?(ExDatalog.Constraint.gt({:var, "X"}, {:const, 0})) + false + + """ + @spec aggregate?(t()) :: boolean() + def aggregate?(%__MODULE__{op: op}), do: op in @aggregate_ops + + @doc false + @spec comparison_op?(atom()) :: boolean() + def comparison_op?(op), do: op in @comparison_ops + + @doc false + @spec arithmetic_op?(atom()) :: boolean() + def arithmetic_op?(op), do: op in @arithmetic_ops + + @doc false + @spec type_op?(atom()) :: boolean() + def type_op?(op), do: op in @type_ops + + @doc false + @spec string_op?(atom()) :: boolean() + def string_op?(op), do: op in @string_ops + + @doc false + @spec membership_op?(atom()) :: boolean() + def membership_op?(op), do: op in @membership_ops + + @doc false + @spec aggregate_op?(atom()) :: boolean() + def aggregate_op?(op), do: op in @aggregate_ops + @doc """ Returns `true` if the constraint is structurally valid. @@ -551,16 +670,20 @@ defmodule ExDatalog.Constraint do %__MODULE__{op: op, left: left, right: right, result: nil} end + defp aggregate(op, input, result) do + %__MODULE__{op: op, left: input, right: nil, result: result} + end + defp valid_result?(op, nil) when op in @comparison_ops or op in @type_ops or op in @string_ops or op in @membership_ops, do: true - defp valid_result?(op, {:var, name}) when op in @arithmetic_ops, + defp valid_result?(op, {:var, name}) when op in @arithmetic_ops or op in @aggregate_ops, do: is_binary(name) and byte_size(name) > 0 defp valid_result?(_, _), do: false - defp valid_right?(op, nil) when op in @type_ops, do: true + defp valid_right?(op, nil) when op in @type_ops or op in @aggregate_ops, do: true defp valid_right?(:member, {:const, value}) when is_list(value), do: true @@ -644,6 +767,10 @@ defmodule ExDatalog.Constraint do filter(op, Term.from(left), Term.from(right)) end + def from_tuple({op, input, result}) when op in @aggregate_ops do + aggregate(op, Term.from(input), Term.from(result)) + end + def from_tuple({:member, left, right}) do member(Term.from(left), Term.from(right)) end @@ -722,4 +849,5 @@ defmodule ExDatalog.Constraint do defp constraint_module(op) when op in @type_ops, do: ExDatalog.Constraints.Type defp constraint_module(op) when op in @string_ops, do: ExDatalog.Constraints.StringPredicate defp constraint_module(op) when op in @membership_ops, do: ExDatalog.Constraints.Membership + defp constraint_module(op) when op in @aggregate_ops, do: ExDatalog.Constraints.Aggregate end diff --git a/lib/ex_datalog/constraint/context.ex b/lib/ex_datalog/constraint/context.ex index c1ee532..ca2c8ab 100644 --- a/lib/ex_datalog/constraint/context.ex +++ b/lib/ex_datalog/constraint/context.ex @@ -21,11 +21,13 @@ defmodule ExDatalog.Constraint.Context do @type t :: %__MODULE__{ capabilities: Capabilities.t(), - provenance: boolean() + provenance: boolean(), + opts: keyword() } defstruct capabilities: %Capabilities{}, - provenance: false + provenance: false, + opts: [] @doc """ Creates a new context with default capabilities. diff --git a/lib/ex_datalog/constraints/aggregate.ex b/lib/ex_datalog/constraints/aggregate.ex new file mode 100644 index 0000000..98acdd6 --- /dev/null +++ b/lib/ex_datalog/constraints/aggregate.ex @@ -0,0 +1,62 @@ +defmodule ExDatalog.Constraints.Aggregate do + @moduledoc """ + Aggregate constraint evaluation: `count`, `sum`, `min`, `max`. + + Aggregates are **not** evaluated per binding. Unlike comparison or arithmetic + constraints, an aggregate must see the full set of surviving bindings for a + rule, group them, and reduce each group to a single value. That grouping is + performed by `ExDatalog.Engine.Evaluator` via `group_and_reduce/5`; the + `evaluate/3` callback exists only to satisfy the `ExDatalog.Constraint` + behaviour and raises if ever invoked through the per-binding pipeline. + + Aggregates are integer-only. `sum` requires integer inputs (validated at + build time and guarded at runtime). `count` returns the group size. `min`/`max` + return the smallest/largest input value in the group. + """ + + @behaviour ExDatalog.Constraint + + @dialyzer {:nowarn_function, evaluate: 3} + + @impl ExDatalog.Constraint + @doc """ + Not supported per-binding. Aggregates are evaluated by the engine's + group-and-reduce path. Always raises. + """ + def evaluate(_constraint, _binding, _context) do + raise "aggregate constraints are not evaluated per-binding; " <> + "use ExDatalog.Engine.Evaluator group-and-reduce path" + end + + @doc """ + Groups bindings by `group_vars` and reduces each group's `input_var` values + with the aggregate `op`, binding the reduced value to `result_var`. + + Returns one extended binding per non-empty group. Empty groups never occur: + a group exists only because at least one binding produced its key, so the + reducers (`Enum.min/1`, `Enum.max/1`) are never called on an empty list. + + ## Examples + + iex> bindings = [%{"D" => :eng, "E" => :a}, %{"D" => :eng, "E" => :b}, %{"D" => :ops, "E" => :c}] + iex> ExDatalog.Constraints.Aggregate.group_and_reduce(bindings, ["D"], :count, "E", "N") + ...> |> Enum.map(fn b -> {b["D"], b["N"]} end) + ...> |> Enum.sort() + [{:eng, 2}, {:ops, 1}] + + """ + @spec group_and_reduce([map()], [String.t()], atom(), String.t(), String.t()) :: [map()] + def group_and_reduce(bindings, group_vars, op, input_var, result_var) do + bindings + |> Enum.group_by(fn binding -> Map.take(binding, group_vars) end) + |> Enum.map(fn {_key, group} -> + values = Enum.map(group, fn b -> Map.fetch!(b, input_var) end) + Map.put(hd(group), result_var, compute(op, values)) + end) + end + + defp compute(:count, values), do: length(values) + defp compute(:sum, values), do: Enum.sum(values) + defp compute(:min, values), do: Enum.min(values) + defp compute(:max, values), do: Enum.max(values) +end diff --git a/lib/ex_datalog/constraints/beam_callback.ex b/lib/ex_datalog/constraints/beam_callback.ex new file mode 100644 index 0000000..55b57db --- /dev/null +++ b/lib/ex_datalog/constraints/beam_callback.ex @@ -0,0 +1,103 @@ +defmodule ExDatalog.Constraints.BeamCallback do + @moduledoc """ + Evaluation of BEAM callback predicates. + + A callback applies an Elixir function to argument values resolved (by variable + name) from the current binding. The function must be deterministic and + side-effect free — these are caller contracts, not enforced. + + The engine enforces only: + + - **Timeout** — the call runs in a `Task` with a configurable timeout + (`:callback_timeout_ms`, default 100ms). A timeout filters the binding. + - **Exception isolation** — a raised exception filters the binding. + + Boolean callbacks (`result: nil`) act as filters: `true` keeps the binding, + `false`/timeout/exception drops it. Value-returning callbacks + (`result: {:var, name}`) bind the return value to `name`. + """ + + alias ExDatalog.IR + + @default_timeout_ms 100 + + @doc """ + Applies a callback against a binding. + + Returns `{:ok, binding}` (boolean true, or value bound) or `:filter` + (boolean false, unbound argument, timeout, or exception). + """ + @spec apply_callback(IR.Callback.t(), map(), keyword()) :: {:ok, map()} | :filter + def apply_callback( + %IR.Callback{module: m, function: f, args: arg_terms, result: result}, + binding, + opts + ) do + case resolve_args(arg_terms, binding) do + {:ok, args} -> + timeout = Keyword.get(opts, :callback_timeout_ms, @default_timeout_ms) + + case safe_apply(m, f, args, timeout) do + {:ok, true} when result == nil -> {:ok, binding} + {:ok, false} when result == nil -> :filter + {:ok, value} when result != nil -> {:ok, bind_result(binding, result, value)} + {:ok, _other} -> :filter + {:error, _} -> :filter + end + + :unbound -> + :filter + end + end + + defp resolve_args(arg_terms, binding) do + Enum.reduce_while(arg_terms, {:ok, []}, fn term, {:ok, acc} -> + case IR.resolve_operand(term, binding) do + {:ok, value} -> {:cont, {:ok, [value | acc]}} + :unbound -> {:halt, :unbound} + end + end) + |> case do + {:ok, reversed} -> {:ok, Enum.reverse(reversed)} + :unbound -> :unbound + end + end + + defp bind_result(binding, {:var, name}, value), do: Map.put(binding, name, value) + + # Run the callback in an unlinked, monitored process so that a raise or + # exit inside the callback does not propagate to (or kill) the evaluator. + # A crash or timeout is reported as `{:error, _}` and filters the binding. + defp safe_apply(module, function, args, timeout_ms) do + parent = self() + ref = make_ref() + + {pid, monitor_ref} = + spawn_monitor(fn -> + result = + try do + {:ok, apply(module, function, args)} + rescue + e -> {:error, e} + catch + kind, reason -> {:error, {kind, reason}} + end + + send(parent, {ref, result}) + end) + + receive do + {^ref, result} -> + Process.demonitor(monitor_ref, [:flush]) + result + + {:DOWN, ^monitor_ref, :process, ^pid, reason} -> + {:error, reason} + after + timeout_ms -> + Process.exit(pid, :kill) + Process.demonitor(monitor_ref, [:flush]) + {:error, :timeout} + end + end +end diff --git a/lib/ex_datalog/engine/evaluator.ex b/lib/ex_datalog/engine/evaluator.ex index 5a9d41f..5ca64c2 100644 --- a/lib/ex_datalog/engine/evaluator.ex +++ b/lib/ex_datalog/engine/evaluator.ex @@ -37,6 +37,7 @@ defmodule ExDatalog.Engine.Evaluator do ordering guarantee. """ + alias ExDatalog.Constraints.{Aggregate, BeamCallback} alias ExDatalog.Engine.{Binding, ConstraintEval, Join} alias ExDatalog.IR @@ -77,12 +78,24 @@ defmodule ExDatalog.Engine.Evaluator do ) :: [tuple()] def eval_rule_iteration(rule, full, delta, old, ctx \\ %ExDatalog.Constraint.Context{}) do - positive_body = positive_atoms(rule) - k = length(positive_body) - head_relation = rule.head.relation existing = Map.get(full, head_relation, MapSet.new()) + if aggregate_rule?(rule) do + rule + |> eval_aggregate_rule(full, ctx) + |> MapSet.new() + |> MapSet.difference(existing) + |> MapSet.to_list() + else + eval_normal_rule(rule, full, delta, old, ctx, existing) + end + end + + defp eval_normal_rule(rule, full, delta, old, ctx, existing) do + positive_body = positive_atoms(rule) + k = length(positive_body) + if k == 0 do derived = [%{}] @@ -99,6 +112,56 @@ defmodule ExDatalog.Engine.Evaluator do end end + defp aggregate_rule?(%IR.Rule{body: body}) do + Enum.any?(body, fn + {:constraint, %ExDatalog.IR.Constraint{op: op}} -> op in [:count, :sum, :min, :max] + _ -> false + end) + end + + # Aggregate rules are stratified above their source relations, so all source + # facts are final in `full`. We join the positive body against `full`, apply + # non-aggregate filters and negation, group by the head variables other than + # the aggregate result, reduce each group, then project. + defp eval_aggregate_rule(%IR.Rule{} = rule, full, ctx) do + agg_constraint = find_aggregate(rule.body) + + joined = join_positive_body([%{}], rule.body, full) + filtered = apply_constraints(rule.body, joined, ctx) + with_callbacks = apply_callbacks(rule.body, filtered, ctx) + bindings = apply_negation(rule.body, with_callbacks, full) + + %ExDatalog.IR.Constraint{op: op, left: {:var, input_var}, result: {:var, result_var}} = + agg_constraint + + group_vars = aggregate_group_vars(rule.head, result_var) + + bindings + |> Aggregate.group_and_reduce(group_vars, op, input_var, result_var) + |> Enum.map(&Join.project(rule.head, &1)) + end + + defp find_aggregate(body) do + Enum.find_value(body, fn + {:constraint, %ExDatalog.IR.Constraint{op: op} = c} when op in [:count, :sum, :min, :max] -> + c + + _ -> + nil + end) + end + + # Group by the head variables other than the aggregate result variable. + defp aggregate_group_vars(%IR.Atom{terms: terms}, result_var) do + terms + |> Enum.flat_map(fn + {:var, name} -> [name] + _ -> [] + end) + |> Enum.reject(fn name -> name == result_var end) + |> Enum.uniq() + end + @doc """ Checks whether a binding satisfies a negative body atom. @@ -137,6 +200,7 @@ defmodule ExDatalog.Engine.Evaluator do defp finish_bindings(bindings, rule, full, ctx) do bindings = apply_constraints(rule.body, bindings, ctx) + bindings = apply_callbacks(rule.body, bindings, ctx) bindings = apply_negation(rule.body, bindings, full) case bindings do @@ -145,6 +209,34 @@ defmodule ExDatalog.Engine.Evaluator do end end + defp apply_callbacks(body, bindings, ctx) do + callbacks = for {:callback, cb} <- body, do: cb + + case callbacks do + [] -> + bindings + + _ -> + opts = callback_opts(ctx) + + Enum.flat_map(bindings, fn binding -> + apply_callback_chain(callbacks, binding, opts) + end) + end + end + + defp apply_callback_chain(callbacks, binding, opts) do + Enum.reduce_while(callbacks, [binding], fn cb, [b] -> + step_callback(BeamCallback.apply_callback(cb, b, opts)) + end) + end + + defp step_callback({:ok, new_b}), do: {:cont, [new_b]} + defp step_callback(:filter), do: {:halt, []} + + defp callback_opts(%ExDatalog.Constraint.Context{opts: opts}) when is_list(opts), do: opts + defp callback_opts(_), do: [] + defp eval_variant(rule, positive_body, full, delta, old, delta_pos, ctx) do bindings = positive_body @@ -181,7 +273,13 @@ defmodule ExDatalog.Engine.Evaluator do end defp apply_constraints(body, bindings, ctx) do - constraints = for {:constraint, c} <- body, do: c + # Aggregate constraints are NOT evaluated per binding; they are handled by + # the group-and-reduce aggregate path. Exclude them here so the normal + # per-binding pipeline never feeds them to ConstraintEval. + constraints = + for {:constraint, %ExDatalog.IR.Constraint{op: op} = c} <- body, + op not in [:count, :sum, :min, :max], + do: c Enum.flat_map(bindings, fn b -> case ConstraintEval.apply(constraints, b, ctx) do diff --git a/lib/ex_datalog/engine/naive.ex b/lib/ex_datalog/engine/naive.ex index 117c8a6..eb8b5b6 100644 --- a/lib/ex_datalog/engine/naive.ex +++ b/lib/ex_datalog/engine/naive.ex @@ -87,6 +87,36 @@ defmodule ExDatalog.Engine.Naive do """ @spec evaluate(IR.t(), keyword()) :: {:ok, Knowledge.t()} | {:error, term()} def evaluate(%IR{} = ir, opts \\ []) do + case Keyword.get(opts, :strategy, :semi_naive) do + :semi_naive -> + evaluate_semi_naive(ir, opts) + + :magic_sets -> + evaluate_magic_sets(ir, opts) + + other -> + {:error, "unknown evaluation strategy: #{inspect(other)}"} + end + end + + defp evaluate_magic_sets(%IR{} = ir, opts) do + case Keyword.get(opts, :goal, nil) do + nil -> + # No goal to drive demand; fall back to full evaluation. + evaluate_semi_naive(ir, opts) + + goal -> + case ExDatalog.MagicSets.transform(ir, goal) do + {:ok, transformed_ir} -> + evaluate_semi_naive(transformed_ir, Keyword.delete(opts, :strategy)) + + {:fallback, _reason} -> + evaluate_semi_naive(ir, opts) + end + end + end + + defp evaluate_semi_naive(%IR{} = ir, opts) do ExDatalog.Telemetry.emit_start(ir) start_time = System.monotonic_time(:microsecond) stratum_count = length(ir.strata) @@ -136,7 +166,8 @@ defmodule ExDatalog.Engine.Naive do state0 = storage_mod.init(schemas, storage_opts) constraint_ctx = %ExDatalog.Constraint.Context{ - capabilities: storage_mod.capabilities(state0) + capabilities: storage_mod.capabilities(state0), + opts: opts } try do @@ -284,16 +315,48 @@ defmodule ExDatalog.Engine.Naive do end) end) - case unstratifiable do - [] -> - :ok + unstratifiable_aggregates = + Enum.filter(ir.rules, fn rule -> + has_aggregate?(rule) and + Enum.any?(positive_literals(rule), fn {:positive, %IR.Atom{relation: rel}} -> + rel != rule.head.relation and Map.get(relation_strata, rel, 0) >= rule.stratum + end) + end) + + cond do + unstratifiable != [] -> + {:error, + "unstratifiable negation detected: #{format_unstratifiable_details(unstratifiable)}"} - rules -> - details = format_unstratifiable_details(rules) - {:error, "unstratifiable negation detected: #{details}"} + unstratifiable_aggregates != [] -> + {:error, + "unstratifiable aggregate detected: #{format_unstratifiable_aggregate_details(unstratifiable_aggregates)}"} + + true -> + :ok end end + defp positive_literals(%IR.Rule{body: body}) do + Enum.filter(body, fn + {:positive, _} -> true + _ -> false + end) + end + + defp has_aggregate?(%IR.Rule{body: body}) do + Enum.any?(body, fn + {:constraint, %IR.Constraint{op: op}} -> op in [:count, :sum, :min, :max] + _ -> false + end) + end + + defp format_unstratifiable_aggregate_details(rules) do + Enum.map_join(rules, "; ", fn r -> + "rule #{r.id} (head: #{r.head.relation}) aggregates over a relation in the same or higher stratum" + end) + end + defp negative_literals(%IR.Rule{body: body}) do Enum.filter(body, fn {:negative, _} -> true @@ -551,6 +614,7 @@ defmodule ExDatalog.Engine.Naive do {:positive, %IR.Atom{relation: r}} -> [r] {:negative, %IR.Atom{relation: r}} -> [r] {:constraint, _} -> [] + {:callback, _} -> [] end) |> Enum.uniq() end diff --git a/lib/ex_datalog/ir.ex b/lib/ex_datalog/ir.ex index 8ffb1e3..5cad35b 100644 --- a/lib/ex_datalog/ir.ex +++ b/lib/ex_datalog/ir.ex @@ -49,10 +49,18 @@ defmodule ExDatalog.IR do result: ir_term() | nil } + @type ir_callback :: %ExDatalog.IR.Callback{ + module: module(), + function: atom(), + args: [ir_term()], + result: ir_term() | nil + } + @type ir_literal :: {:positive, ExDatalog.IR.Atom.t()} | {:negative, ExDatalog.IR.Atom.t()} | {:constraint, ir_constraint()} + | {:callback, ir_callback()} defmodule Relation do @moduledoc """ @@ -139,6 +147,31 @@ defmodule ExDatalog.IR do end end + defmodule Callback do + @moduledoc """ + An IR callback predicate: an Elixir function invoked during evaluation. + + `args` are IR terms resolved by variable name against the binding at + evaluation time. `result` is `nil` for boolean (filter) callbacks, or + `{:var, name}` for value-returning callbacks that bind a result variable. + """ + + @enforce_keys [:module, :function, :args] + defstruct [:module, :function, :args, :result] + + @type t :: %__MODULE__{ + module: module(), + function: atom(), + args: [ExDatalog.IR.ir_term()], + result: ExDatalog.IR.ir_term() | nil + } + + @spec serialize(t()) :: map() + def serialize(%__MODULE__{module: m, function: f, args: args, result: result}) do + %{module: m, function: f, args: args, result: result} + end + end + defmodule Rule do @moduledoc """ An IR rule: a head atom, a body of literals, an assigned stratum, and @@ -177,6 +210,9 @@ defmodule ExDatalog.IR do {:constraint, c} -> %{kind: :constraint, constraint: Constraint.serialize(c)} + + {:callback, cb} -> + %{kind: :callback, callback: Callback.serialize(cb)} end), stratum: stratum, metadata: metadata @@ -302,6 +338,19 @@ defmodule ExDatalog.IR do %Atom{relation: relation, terms: Enum.map(terms, &from_term/1)} end + @doc """ + Converts an AST callback to an IR callback. + """ + @spec from_callback(ExDatalog.Callback.t()) :: Callback.t() + def from_callback(%ExDatalog.Callback{module: m, function: f, args: args, result: result}) do + %Callback{ + module: m, + function: f, + args: Enum.map(args, &from_term/1), + result: maybe_from_term(result) + } + end + @doc """ Resolves an IR term against a binding environment. diff --git a/lib/ex_datalog/magic_sets.ex b/lib/ex_datalog/magic_sets.ex new file mode 100644 index 0000000..bfd9e6e --- /dev/null +++ b/lib/ex_datalog/magic_sets.ex @@ -0,0 +1,187 @@ +defmodule ExDatalog.MagicSets do + @moduledoc """ + Magic-sets program transformation for demand-driven (goal-directed) evaluation. + + Magic sets rewrites a program so that bottom-up semi-naive evaluation computes + only the facts relevant to a query goal, instead of the full least fixpoint. + It is a **program transformation**: the engine is unchanged. Given a goal + `{relation, pattern}`, the transformation: + + 1. computes the goal's *adornment* (which argument positions are bound), + 2. generates `magic__` predicates capturing demand, + 3. rewrites recursive rules to consume the magic predicates, + 4. seeds the magic predicate with the bound goal constants. + + The transformed IR is then evaluated by the existing semi-naive engine. + + ## Scope (v0.5.0, experimental) + + - Positive recursive programs only. + - A single goal. + - Ground (constant) bound positions. + + Programs outside this scope fall back to full semi-naive evaluation + (`{:fallback, reason}`), never producing incorrect results. + """ + + alias ExDatalog.IR + + @doc """ + Transforms an IR program for goal-directed evaluation. + + Returns `{:ok, transformed_ir}` when the magic-sets transformation applies, + or `{:fallback, reason}` when the program is outside the supported scope (the + caller should evaluate the original IR with semi-naive instead). + """ + @spec transform(IR.t(), {String.t(), [term()]}) :: {:ok, IR.t()} | {:fallback, term()} + def transform(%IR{} = ir, {goal_relation, goal_pattern}) do + cond do + not supported_program?(ir) -> + {:fallback, :unsupported_program} + + not has_bound_position?(goal_pattern) -> + {:fallback, :no_bound_positions} + + true -> + do_transform(ir, goal_relation, goal_pattern) + end + end + + # --- Scope checks --- + + defp supported_program?(%IR{rules: rules}) do + Enum.all?(rules, fn rule -> + not has_negation?(rule) and not has_aggregate?(rule) + end) + end + + defp has_negation?(%IR.Rule{body: body}) do + Enum.any?(body, &match?({:negative, _}, &1)) + end + + defp has_aggregate?(%IR.Rule{body: body}) do + Enum.any?(body, fn + {:constraint, %IR.Constraint{op: op}} -> op in [:count, :sum, :min, :max] + _ -> false + end) + end + + defp has_bound_position?(pattern) do + Enum.any?(pattern, fn p -> p != :_ end) + end + + # --- Transformation --- + + defp do_transform(%IR{} = ir, goal_relation, goal_pattern) do + adornment = adornment(goal_pattern) + magic_rel = magic_relation_name(goal_relation, adornment) + bound_positions = bound_positions(goal_pattern) + + magic_relation = %IR.Relation{ + name: magic_rel, + arity: bound_count(goal_pattern), + types: bound_types(ir, goal_relation, goal_pattern) + } + + seed_fact = seed_fact(magic_rel, goal_relation, goal_pattern) + + rewritten_rules = + ir.rules + |> Enum.map(fn rule -> rewrite_rule(rule, goal_relation, magic_rel, bound_positions) end) + + new_relations = [magic_relation | ir.relations] + new_facts = if seed_fact, do: [seed_fact | ir.facts], else: ir.facts + + # Recompute strata: the magic relation joins the goal relation's stratum. + new_strata = inject_magic_into_strata(ir.strata, goal_relation, magic_rel) + + {:ok, + %IR{ + ir + | relations: new_relations, + facts: new_facts, + rules: rewritten_rules, + strata: new_strata + }} + end + + defp adornment(pattern) do + Enum.map_join(pattern, "", fn + :_ -> "f" + _ -> "b" + end) + end + + defp magic_relation_name(relation, adornment), do: "magic_#{relation}_#{adornment}" + + defp bound_count(pattern), do: Enum.count(pattern, fn p -> p != :_ end) + + defp bound_positions(pattern) do + pattern + |> Enum.with_index() + |> Enum.filter(fn {p, _i} -> p != :_ end) + |> Enum.map(fn {_p, i} -> i end) + end + + defp bound_types(ir, goal_relation, pattern) do + case Enum.find(ir.relations, fn r -> r.name == goal_relation end) do + %IR.Relation{types: types} -> + types + |> Enum.zip(pattern) + |> Enum.filter(fn {_t, p} -> p != :_ end) + |> Enum.map(fn {t, _p} -> t end) + + nil -> + List.duplicate(:any, bound_count(pattern)) + end + end + + defp seed_fact(magic_rel, _goal_relation, pattern) do + bound_values = + pattern + |> Enum.filter(fn p -> p != :_ end) + |> Enum.map(&to_ir_value/1) + + case bound_values do + [] -> nil + values -> %IR.Fact{relation: magic_rel, values: values} + end + end + + defp to_ir_value(v) when is_integer(v), do: {:int, v} + defp to_ir_value(v) when is_binary(v), do: {:str, v} + defp to_ir_value(v) when is_atom(v), do: {:atom, v} + + # Rewrite a rule whose head is the goal relation: prepend the magic predicate + # binding the bound head positions, so derivation is demand-restricted. The + # magic atom projects exactly the goal's bound positions of the head, so its + # arity matches the magic relation. + defp rewrite_rule( + %IR.Rule{head: %IR.Atom{relation: rel} = head} = rule, + goal_relation, + magic_rel, + bound_positions + ) + when rel == goal_relation do + magic_terms = bound_head_terms(head, bound_positions) + magic_atom = %IR.Atom{relation: magic_rel, terms: magic_terms} + %IR.Rule{rule | body: [{:positive, magic_atom} | rule.body]} + end + + defp rewrite_rule(rule, _goal_relation, _magic_rel, _bound_positions), do: rule + + defp bound_head_terms(%IR.Atom{terms: terms}, bound_positions) do + bound_positions + |> Enum.map(fn pos -> Enum.at(terms, pos) end) + end + + defp inject_magic_into_strata(strata, goal_relation, magic_rel) do + Enum.map(strata, fn %IR.Stratum{relations: rels} = stratum -> + if goal_relation in rels do + %IR.Stratum{stratum | relations: [magic_rel | rels]} + else + stratum + end + end) + end +end diff --git a/lib/ex_datalog/planner.ex b/lib/ex_datalog/planner.ex new file mode 100644 index 0000000..23d3e58 --- /dev/null +++ b/lib/ex_datalog/planner.ex @@ -0,0 +1,210 @@ +defmodule ExDatalog.Planner do + @moduledoc """ + Query/evaluation planner for ExDatalog. + + The planner sits between the compiled `ExDatalog.IR` and the evaluation + engine. It produces an `ExDatalog.Planner.Plan` describing the chosen + `strategy`, the planned strata, the joins (one per positive body atom), and + the predicates (constraints, aggregates, callbacks). + + The planner is intentionally thin in v0.5.0: it wraps the existing IR strata + and classifies body elements. It is the seam through which the `:magic_sets` + strategy is selected and, in future releases, where join ordering and + cost-based optimization will live. + + ## Strategy selection + + `plan/2` accepts `:strategy` (`:semi_naive` default, or `:magic_sets`). When + `:magic_sets` is requested, a `:goal` option (`{relation, pattern}`) should be + supplied; otherwise the planner records the strategy but the engine falls back + to semi-naive. + + ## Telemetry + + `plan/2` emits `[:ex_datalog, :planner, :start | :stop | :exception]` once per + call (never per rule). + """ + + alias ExDatalog.Constraint + alias ExDatalog.IR + alias ExDatalog.Planner.{Join, Plan, Predicate, Stratum} + + @doc """ + Builds an execution `Plan` from a compiled IR program. + + ## Options + + - `:strategy` — `:semi_naive` (default) or `:magic_sets` + - `:goal` — `{relation, pattern}` used when `strategy: :magic_sets` + + Returns `{:ok, %Plan{}}`. + + ## Examples + + iex> alias ExDatalog.{Program, Rule, Atom, Term, Compiler, Planner} + iex> {:ok, ir} = + ...> Program.new() + ...> |> Program.add_relation("edge", [:atom, :atom]) + ...> |> Program.add_relation("path", [:atom, :atom]) + ...> |> Program.add_rule( + ...> Rule.new( + ...> Atom.new("path", [Term.var("X"), Term.var("Y")]), + ...> [{:positive, Atom.new("edge", [Term.var("X"), Term.var("Y")])}] + ...> ) + ...> ) + ...> |> Compiler.compile() + iex> {:ok, plan} = Planner.plan(ir) + iex> plan.strategy + :semi_naive + iex> length(plan.joins) + 1 + + """ + @spec plan(IR.t(), keyword()) :: {:ok, Plan.t()} + def plan(%IR{} = ir, opts \\ []) do + metadata = %{relation_count: length(ir.relations), rule_count: length(ir.rules)} + + :telemetry.execute( + [:ex_datalog, :planner, :start], + %{system_time: System.system_time()}, + metadata + ) + + start = System.monotonic_time() + + try do + strategy = Keyword.get(opts, :strategy, :semi_naive) + goal = Keyword.get(opts, :goal, nil) + + plan = %Plan{ + strategy: strategy, + strata: build_strata(ir), + joins: build_joins(ir), + predicates: build_predicates(ir), + metadata: %{goal: goal} + } + + duration = System.monotonic_time() - start + + :telemetry.execute( + [:ex_datalog, :planner, :stop], + %{duration: duration}, + Map.put(metadata, :strategy, strategy) + ) + + {:ok, plan} + rescue + e -> + :telemetry.execute( + [:ex_datalog, :planner, :exception], + %{duration: System.monotonic_time() - start}, + Map.merge(metadata, %{kind: :error, reason: e}) + ) + + reraise e, __STACKTRACE__ + end + end + + @doc """ + Returns a human-readable description of the plan for a program. + + Accepts the same options as `plan/2`. Validates and compiles the program + first; returns an error string if compilation fails. + + ## Examples + + iex> alias ExDatalog.{Program, Rule, Atom, Term, Planner} + iex> program = + ...> Program.new() + ...> |> Program.add_relation("edge", [:atom, :atom]) + ...> |> Program.add_relation("path", [:atom, :atom]) + ...> |> Program.add_rule( + ...> Rule.new( + ...> Atom.new("path", [Term.var("X"), Term.var("Y")]), + ...> [{:positive, Atom.new("edge", [Term.var("X"), Term.var("Y")])}] + ...> ) + ...> ) + iex> Planner.explain_plan(program) =~ "Strategy: semi_naive" + true + + """ + @spec explain_plan(ExDatalog.Program.t()) :: String.t() + def explain_plan(program), do: explain_plan(program, []) + + @spec explain_plan(ExDatalog.Program.t(), keyword()) :: String.t() + def explain_plan(program, opts) do + case ExDatalog.compile(program) do + {:ok, ir} -> + {:ok, plan} = plan(ir, opts) + format_plan(plan) + + {:error, errors} -> + "Cannot plan: compilation failed with #{length(errors)} error(s)" + end + end + + # --- Plan construction --- + + defp build_strata(%IR{strata: strata, rules: rules}) do + Enum.map(strata, fn %IR.Stratum{index: idx, relations: rels} -> + stratum_rules = Enum.filter(rules, fn r -> r.stratum == idx end) + %Stratum{index: idx, rules: stratum_rules, relations: rels} + end) + end + + defp build_joins(%IR{rules: rules}) do + Enum.flat_map(rules, fn rule -> + rule.body + |> Enum.filter(&match?({:positive, _}, &1)) + |> Enum.with_index() + |> Enum.map(fn {{:positive, %IR.Atom{relation: rel}}, position} -> + %Join{relation: rel, position: position, delta_position: position} + end) + end) + end + + defp build_predicates(%IR{rules: rules}) do + rules + |> Enum.flat_map(fn rule -> + for {:constraint, c} <- rule.body, do: classify_constraint(c) + end) + end + + defp classify_constraint(%IR.Constraint{op: op} = c) do + %Predicate{kind: constraint_kind(op), op: op, metadata: %{result: c.result}} + end + + defp constraint_kind(op) do + cond do + Constraint.comparison_op?(op) -> :comparison + Constraint.arithmetic_op?(op) -> :arithmetic + Constraint.type_op?(op) -> :type + Constraint.string_op?(op) -> :string + Constraint.membership_op?(op) -> :membership + Constraint.aggregate_op?(op) -> :aggregate + true -> :comparison + end + end + + # --- Formatting --- + + defp format_plan(%Plan{} = plan) do + header = "Strategy: #{plan.strategy}" + + strata_lines = + Enum.map(plan.strata, fn s -> + " Stratum #{s.index}: #{length(s.rules)} rule(s), relations: #{Enum.join(s.relations, ", ")}" + end) + + join_line = "Joins: #{length(plan.joins)}" + + pred_line = + "Predicates: #{length(plan.predicates)}" <> + case plan.predicates do + [] -> "" + preds -> " (#{Enum.map_join(preds, ", ", & &1.op)})" + end + + Enum.join([header | strata_lines] ++ [join_line, pred_line], "\n") + end +end diff --git a/lib/ex_datalog/planner/join.ex b/lib/ex_datalog/planner/join.ex new file mode 100644 index 0000000..435135c --- /dev/null +++ b/lib/ex_datalog/planner/join.ex @@ -0,0 +1,22 @@ +defmodule ExDatalog.Planner.Join do + @moduledoc """ + A planned join: one positive body atom position within a rule. + + `position` is the 0-based index of the atom within the rule's positive body. + `delta_position` indicates the semi-naive delta slot when applicable, or + `nil`. `strategy` records how the join is executed; the default engine uses + `:nested_loop`. + """ + + @enforce_keys [:relation, :position] + defstruct [:relation, :position, delta_position: nil, strategy: :nested_loop] + + @type strategy :: :nested_loop | :indexed + + @type t :: %__MODULE__{ + relation: String.t(), + position: non_neg_integer(), + delta_position: non_neg_integer() | nil, + strategy: strategy() + } +end diff --git a/lib/ex_datalog/planner/plan.ex b/lib/ex_datalog/planner/plan.ex new file mode 100644 index 0000000..971b92e --- /dev/null +++ b/lib/ex_datalog/planner/plan.ex @@ -0,0 +1,28 @@ +defmodule ExDatalog.Planner.Plan do + @moduledoc """ + An execution plan for a compiled IR program. + + A plan records the chosen evaluation `strategy`, the planned `strata`, the + `joins` (one per positive body atom across all rules), and the `predicates` + (constraints, aggregates, callbacks). The plan is descriptive: it explains + what the engine will do without changing how the engine evaluates. + + Aggregate and callback predicates appear in `predicates` with + `kind: :aggregate` / `kind: :callback`; there are no separate fields for them. + """ + + alias ExDatalog.Planner.{Join, Predicate, Stratum} + + @enforce_keys [:strategy, :strata] + defstruct [:strategy, :strata, joins: [], predicates: [], metadata: %{}] + + @type strategy :: :semi_naive | :magic_sets + + @type t :: %__MODULE__{ + strategy: strategy(), + strata: [Stratum.t()], + joins: [Join.t()], + predicates: [Predicate.t()], + metadata: map() + } +end diff --git a/lib/ex_datalog/planner/predicate.ex b/lib/ex_datalog/planner/predicate.ex new file mode 100644 index 0000000..a663872 --- /dev/null +++ b/lib/ex_datalog/planner/predicate.ex @@ -0,0 +1,28 @@ +defmodule ExDatalog.Planner.Predicate do + @moduledoc """ + A planned predicate: a non-relational body element (constraint, aggregate, + or callback) classified by kind. + + `kind` groups the predicate into one of the evaluation categories; `op` is + the specific operator (e.g. `:gt`, `:count`, `:callback`). `metadata` carries + optional details (e.g. callback module/function). + """ + + @enforce_keys [:kind, :op] + defstruct [:kind, :op, metadata: %{}] + + @type kind :: + :comparison + | :arithmetic + | :type + | :string + | :membership + | :aggregate + | :callback + + @type t :: %__MODULE__{ + kind: kind(), + op: atom(), + metadata: map() + } +end diff --git a/lib/ex_datalog/planner/stratum.ex b/lib/ex_datalog/planner/stratum.ex new file mode 100644 index 0000000..280c24a --- /dev/null +++ b/lib/ex_datalog/planner/stratum.ex @@ -0,0 +1,20 @@ +defmodule ExDatalog.Planner.Stratum do + @moduledoc """ + A planned stratum: the rules and relations evaluated together at one + stratum index. + + Wraps the IR strata into a planner-friendly form that carries the actual + `IR.Rule` structs (not just their IDs) for the engine and `explain_plan/1`. + """ + + alias ExDatalog.IR + + @enforce_keys [:index, :rules, :relations] + defstruct [:index, :rules, :relations] + + @type t :: %__MODULE__{ + index: non_neg_integer(), + rules: [IR.Rule.t()], + relations: [String.t()] + } +end diff --git a/lib/ex_datalog/program.ex b/lib/ex_datalog/program.ex index 1c01dda..ec44a4a 100644 --- a/lib/ex_datalog/program.ex +++ b/lib/ex_datalog/program.ex @@ -236,6 +236,67 @@ defmodule ExDatalog.Program do def add_fact({:error, _} = err, _relation, _values), do: err + @doc """ + Adds a fact to the program from a `{relation, values}` tuple. + + This is the tuple form produced by Schema relation constructors + (e.g., `MySchema.emp(:alice, :eng)` returns `{"emp", [:alice, :eng]}`). + + ## Examples + + iex> alias ExDatalog.Program + iex> program = Program.new() |> Program.add_relation("emp", [:atom, :atom]) + iex> program = Program.add_fact(program, {"emp", [:alice, :eng]}) + iex> program.facts + [{"emp", [:alice, :eng]}] + """ + @spec add_fact(t(), {String.t(), [term()]}) :: t() | {:error, term()} + def add_fact(program, {relation, values}) when is_binary(relation) and is_list(values) do + add_fact(program, relation, values) + end + + def add_fact({:error, _} = err, {_relation, _values}), do: err + + @doc """ + Adds multiple facts to the program from a list of `{relation, values}` tuples. + + ## Examples + + iex> alias ExDatalog.Program + iex> program = Program.new() |> Program.add_relation("emp", [:atom, :atom]) + iex> facts = [{"emp", [:alice, :eng]}, {"emp", [:bob, :eng]}] + iex> program = Program.add_facts(program, facts) + iex> length(program.facts) + 2 + """ + @spec add_facts(t(), [{String.t(), [term()]}]) :: t() | {:error, term()} + def add_facts(program, facts) when is_list(facts) do + Enum.reduce_while(facts, program, fn fact, acc -> + case add_fact(acc, fact) do + {:error, _} = err -> {:halt, err} + prog -> {:cont, prog} + end + end) + end + + @doc """ + Materializes the program. A pipe-friendly convenience for + `ExDatalog.materialize/2`. + + ## Examples + + iex> alias ExDatalog.{Program, Knowledge} + iex> program = Program.new() |> Program.add_relation("edge", [:atom, :atom]) + iex> program = program |> Program.add_fact("edge", [:a, :b]) + iex> {:ok, knowledge} = Program.materialize(program) + iex> Knowledge.get(knowledge, "edge") |> MapSet.to_list() + [{:a, :b}] + """ + @spec materialize(t(), keyword()) :: {:ok, ExDatalog.Knowledge.t()} | {:error, term()} + def materialize(%__MODULE__{} = program, opts \\ []) do + ExDatalog.materialize(program, opts) + end + @doc """ Adds a rule to the program. @@ -431,17 +492,18 @@ defmodule ExDatalog.Program do defp validate_body(program, body) do Enum.reduce_while(body, :ok, fn literal, :ok -> - atom = - case literal do - {:positive, a} -> a - {:negative, a} -> a - other -> {:error, "invalid body literal #{inspect(other)}"} - end - - case atom do - {:error, _} = err -> {:halt, err} - %Atom{} -> {:cont, validate_atom(program, atom)} - _ -> {:halt, {:error, "invalid body literal #{inspect(literal)}"}} + case literal do + {:callback, %ExDatalog.Callback{}} -> + {:cont, :ok} + + {:positive, %Atom{} = a} -> + {:cont, validate_atom(program, a)} + + {:negative, %Atom{} = a} -> + {:cont, validate_atom(program, a)} + + other -> + {:halt, {:error, "invalid body literal #{inspect(other)}"}} end end) end diff --git a/lib/ex_datalog/rule.ex b/lib/ex_datalog/rule.ex index 270155d..20564d9 100644 --- a/lib/ex_datalog/rule.ex +++ b/lib/ex_datalog/rule.ex @@ -28,9 +28,9 @@ defmodule ExDatalog.Rule do """ - alias ExDatalog.{Atom, Constraint} + alias ExDatalog.{Atom, Callback, Constraint} - @type literal :: {:positive, Atom.t()} | {:negative, Atom.t()} + @type literal :: {:positive, Atom.t()} | {:negative, Atom.t()} | {:callback, Callback.t()} @type t :: %__MODULE__{ head: Atom.t(), @@ -91,12 +91,18 @@ defmodule ExDatalog.Rule do Enum.flat_map(body, fn {:positive, atom} -> Atom.variables(atom) {:negative, atom} -> Atom.variables(atom) + {:callback, cb} -> Callback.input_variables(cb) ++ callback_result_var(cb) end) constraint_vars = Enum.flat_map(constraints, fn c -> input = Constraint.input_variables(c) - result = if Constraint.arithmetic?(c), do: [Constraint.result_variable(c)], else: [] + + result = + if Constraint.arithmetic?(c) or Constraint.aggregate?(c), + do: [Constraint.result_variable(c)], + else: [] + input ++ result end) @@ -141,6 +147,7 @@ defmodule ExDatalog.Rule do |> Enum.flat_map(fn {:positive, atom} -> Atom.variables(atom) {:negative, _} -> [] + {:callback, _} -> [] end) |> Enum.uniq() end @@ -163,9 +170,11 @@ defmodule ExDatalog.Rule do """ @spec body_atoms(t()) :: [Atom.t()] def body_atoms(%__MODULE__{body: body}) do - Enum.map(body, fn - {:positive, atom} -> atom - {:negative, atom} -> atom + body + |> Enum.flat_map(fn + {:positive, atom} -> [atom] + {:negative, atom} -> [atom] + {:callback, _} -> [] end) end @@ -198,4 +207,62 @@ defmodule ExDatalog.Rule do _ -> false end) end + + @doc """ + Returns `true` if the rule contains any aggregate constraint + (`count`, `sum`, `min`, `max`). + + ## Examples + + iex> alias ExDatalog.{Rule, Atom, Term, Constraint} + iex> head = Atom.new("dept_count", [Term.var("D"), Term.var("N")]) + iex> body = [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}] + iex> rule = Rule.new(head, body, [Constraint.count(Term.var("E"), Term.var("N"))]) + iex> Rule.has_aggregates?(rule) + true + + """ + @spec has_aggregates?(t()) :: boolean() + def has_aggregates?(%__MODULE__{constraints: constraints}) do + Enum.any?(constraints, &Constraint.aggregate?/1) + end + + @doc """ + Returns the aggregate constraints in the rule. + """ + @spec aggregate_constraints(t()) :: [Constraint.t()] + def aggregate_constraints(%__MODULE__{constraints: constraints}) do + Enum.filter(constraints, &Constraint.aggregate?/1) + end + + @doc """ + Returns the non-aggregate constraints in the rule. + """ + @spec non_aggregate_constraints(t()) :: [Constraint.t()] + def non_aggregate_constraints(%__MODULE__{constraints: constraints}) do + Enum.reject(constraints, &Constraint.aggregate?/1) + end + + @doc """ + Returns `true` if the rule contains any callback predicate in its body. + """ + @spec has_callbacks?(t()) :: boolean() + def has_callbacks?(%__MODULE__{body: body}) do + Enum.any?(body, &match?({:callback, _}, &1)) + end + + @doc """ + Returns the callback predicates in the rule body. + """ + @spec callbacks(t()) :: [Callback.t()] + def callbacks(%__MODULE__{body: body}) do + for {:callback, cb} <- body, do: cb + end + + defp callback_result_var(%Callback{} = cb) do + case Callback.result_variable(cb) do + nil -> [] + name -> [name] + end + end end diff --git a/lib/ex_datalog/schema.ex b/lib/ex_datalog/schema.ex index 1fad8df..38bb8cb 100644 --- a/lib/ex_datalog/schema.ex +++ b/lib/ex_datalog/schema.ex @@ -178,12 +178,13 @@ defmodule ExDatalog.Schema do defmacro __using__(_opts) do quote do import ExDatalog.Schema, - only: [relation: 2, fact: 1, facts: 2, rule: 2, query: 2, wildcard: 0] + only: [relation: 2, fact: 1, facts: 2, rule: 2, query: 2, wildcard: 0, predicate: 5] Module.register_attribute(__MODULE__, :ex_datalog_relations, accumulate: true) Module.register_attribute(__MODULE__, :ex_datalog_facts, accumulate: true) Module.register_attribute(__MODULE__, :ex_datalog_rules, accumulate: true) Module.register_attribute(__MODULE__, :ex_datalog_queries, accumulate: true) + Module.register_attribute(__MODULE__, :ex_datalog_predicates, accumulate: true) @before_compile ExDatalog.Schema end @@ -195,6 +196,7 @@ defmodule ExDatalog.Schema do facts = Module.get_attribute(env.module, :ex_datalog_facts) |> Enum.reverse() rules = Module.get_attribute(env.module, :ex_datalog_rules) |> Enum.reverse() queries = Module.get_attribute(env.module, :ex_datalog_queries) |> Enum.reverse() + predicates = Module.get_attribute(env.module, :ex_datalog_predicates) |> Enum.reverse() relation_names = MapSet.new(relations, fn rel -> Atom.to_string(rel.name) end) @@ -213,17 +215,67 @@ defmodule ExDatalog.Schema do end end) + relation_funs = + Enum.map(relations, fn rel -> + name = rel.name + arity = length(rel.fields) + args = for i <- 1..arity, do: Macro.var(:"arg_#{i}", __MODULE__) + + quote do + @doc """ + Constructs a fact tuple for the `#{unquote(name)}` relation. + + #{unquote(name)}(#{Enum.map_join(1..unquote(arity), ", ", fn i -> "arg_#{i}" end)}) + #=> {"#{unquote(name)}", [#{Enum.map_join(1..unquote(arity), ", ", fn i -> "arg_#{i}" end)})]} + + Pass the result to `Program.add_fact/2` as a fact tuple. + """ + @spec unquote(name)(unquote_splicing(Enum.map(args, fn _ -> quote do: term() end))) :: + {String.t(), [term()]} + def unquote(name)(unquote_splicing(args)) do + {unquote(Atom.to_string(name)), [unquote_splicing(args)]} + end + end + end) + quote do + unquote_splicing(relation_funs) + @doc """ Returns the `ExDatalog.Program` built from this schema's relations, - facts, and rules. + compile-time facts, and rules. + + Use this when all facts are declared at compile time via `fact/1`. + For runtime facts, use `new_program/0` and `Program.add_fact/2`. """ @spec program() :: ExDatalog.Program.t() def program do ExDatalog.Schema.__build_program__( unquote(Macro.escape(relations)), unquote(Macro.escape(facts)), - unquote(Macro.escape(rules)) + unquote(Macro.escape(rules)), + unquote(Macro.escape(predicates)) + ) + end + + @doc """ + Returns a blank `ExDatalog.Program` from this schema's relations and + rules, **without** any compile-time facts. + + Add runtime facts via the pipeable `Program.add_fact/2`: + + DeptCount.new() + |> Program.add_fact(DeptCount.emp(:alice, :eng)) + |> Program.add_fact(DeptCount.emp(:bob, :eng)) + |> Program.materialize() + """ + @spec new() :: ExDatalog.Program.t() + def new do + ExDatalog.Schema.__build_program__( + unquote(Macro.escape(relations)), + [], + unquote(Macro.escape(rules)), + unquote(Macro.escape(predicates)) ) end @@ -258,8 +310,9 @@ defmodule ExDatalog.Schema do end @doc false - def __build_program__(relations, facts, rules) do + def __build_program__(relations, facts, rules, predicates \\ []) do program = ExDatalog.Program.new() + predicate_map = Map.new(predicates, fn p -> {Atom.to_string(p.name), p} end) program = Enum.reduce(relations, program, fn rel_meta, acc -> @@ -293,8 +346,7 @@ defmodule ExDatalog.Schema do body = Enum.map(body_literals, fn {:positive, %ExDatalog.Atom{} = atom} -> - {:positive, - %ExDatalog.Atom{atom | terms: Enum.map(atom.terms, &term_from_parsed/1)}} + resolve_positive_or_callback(atom, predicate_map) {:negative, %ExDatalog.Atom{} = atom} -> {:negative, @@ -359,6 +411,31 @@ defmodule ExDatalog.Schema do defp term_from_parsed({:const, value}), do: ExDatalog.Term.from(value) defp term_from_parsed(:wildcard), do: ExDatalog.Term.from(:_) + # A positive body atom whose "relation" matches a declared predicate becomes + # a callback literal; otherwise it stays a positive atom. + defp resolve_positive_or_callback(%ExDatalog.Atom{relation: rel, terms: terms}, predicate_map) do + case Map.get(predicate_map, rel) do + nil -> + {:positive, %ExDatalog.Atom{relation: rel, terms: Enum.map(terms, &term_from_parsed/1)}} + + %ExDatalog.Schema.PredicateMeta{} = pred -> + build_callback_literal(pred, terms) + end + end + + defp build_callback_literal(%ExDatalog.Schema.PredicateMeta{} = pred, terms) do + arg_terms = Enum.map(terms, &term_from_parsed/1) + + case pred.return_type do + :boolean -> + {:callback, ExDatalog.Callback.new(pred.module, pred.function, arg_terms, nil)} + + :value -> + {args, [result]} = Enum.split(arg_terms, length(arg_terms) - 1) + {:callback, ExDatalog.Callback.new(pred.module, pred.function, args, result)} + end + end + @doc false def __execute_query__(name, knowledge, queries) when is_list(queries) do query_map = Map.new(queries, fn q -> {q.name, q} end) @@ -623,7 +700,14 @@ defmodule ExDatalog.Schema do defp parse_term({:agg, _, _args}) do raise ExDatalog.DSL.CompileError, - message: "aggregates are not yet supported (planned for v0.6.0)" + message: "use count/sum/min/max for aggregates (e.g. `count(X, N)`)" + end + + defp parse_term({op, _, _args}) when op in [:count, :sum, :min, :max] do + raise ExDatalog.DSL.CompileError, + message: + "aggregate #{op}/2 cannot appear in a rule head or as a term; " <> + "place it in the rule body (e.g. `#{op}(X, Result)`) and use Result in the head" end defp parse_term(other) do @@ -670,9 +754,17 @@ defmodule ExDatalog.Schema do defp parse_body_call({:agg, _, _args}) do raise ExDatalog.DSL.CompileError, - message: "aggregates are not yet supported (planned for v0.6.0)" + message: "use count/sum/min/max for aggregates (e.g. `count(X, N)`)" end + aggregate_ops = [:count, :sum, :min, :max] + + Enum.each(aggregate_ops, fn op -> + defp parse_body_call({unquote(op), _, [input, result]}) do + {:constraint, build_aggregate(unquote(op), input, result)} + end + end) + constraint_ops = [ :eq, :neq, @@ -745,6 +837,10 @@ defmodule ExDatalog.Schema do end end) + defp build_aggregate(op, input, result) do + ExDatalog.Constraint.from_tuple({op, parse_term(input), parse_term(result)}) + end + defp build_constraint(op, args) do raise ExDatalog.DSL.CompileError, message: "unsupported constraint #{op}/#{length(args)}: #{inspect(args)}" @@ -874,4 +970,55 @@ defmodule ExDatalog.Schema do """ @spec wildcard() :: :wildcard def wildcard, do: :wildcard + + @doc """ + Declares a BEAM callback predicate usable in rule bodies. + + predicate :adult?, MyPredicates, :adult?, [:integer], :boolean + + rule adult(Name) do + person(Name, Age) + adult?(Age) + end + + Arguments: + + - `name` — the predicate name used in rule bodies. + - `module` / `function` — the Elixir function to call. + - `arg_types` — declared argument types (currently informational; their + length sets the callback arity, validated at compile time). + - `return_type` — `:boolean` (filter) or `:value` (binds the last argument + position as the result variable). + + For `:value` predicates, the **last** argument in the rule-body call is the + result variable; the remaining arguments are passed to the function. + + The referenced `module.function` must be defined and exported with arity + matching the inputs, or a compile-time `ExDatalog.DSL.CompileError` is raised. + """ + defmacro predicate(name, module, function, arg_types, return_type) do + quote do + ExDatalog.Schema.__register_predicate__( + __MODULE__, + unquote(name), + unquote(module), + unquote(function), + unquote(Macro.escape(arg_types)), + unquote(return_type) + ) + end + end + + @doc false + def __register_predicate__(mod, name, module, function, arg_types, return_type) + when is_atom(name) and is_atom(module) and is_atom(function) and is_list(arg_types) and + return_type in [:boolean, :value] do + Module.put_attribute(mod, :ex_datalog_predicates, %ExDatalog.Schema.PredicateMeta{ + name: name, + module: module, + function: function, + arg_types: arg_types, + return_type: return_type + }) + end end diff --git a/lib/ex_datalog/schema/predicate_meta.ex b/lib/ex_datalog/schema/predicate_meta.ex new file mode 100644 index 0000000..5dcbb89 --- /dev/null +++ b/lib/ex_datalog/schema/predicate_meta.ex @@ -0,0 +1,22 @@ +defmodule ExDatalog.Schema.PredicateMeta do + @moduledoc """ + Metadata for a callback predicate declared with the `predicate/5` macro. + + Records the DSL name, the target module/function, the declared argument + types, and the return type (`:boolean` for filters, `:value` for + value-returning callbacks). + """ + + @enforce_keys [:name, :module, :function, :arg_types, :return_type] + defstruct [:name, :module, :function, :arg_types, :return_type] + + @type return_type :: :boolean | :value + + @type t :: %__MODULE__{ + name: atom(), + module: module(), + function: atom(), + arg_types: [atom()], + return_type: return_type() + } +end diff --git a/lib/ex_datalog/validator.ex b/lib/ex_datalog/validator.ex index e5cb9d0..70bc2f4 100644 --- a/lib/ex_datalog/validator.ex +++ b/lib/ex_datalog/validator.ex @@ -140,6 +140,9 @@ defmodule ExDatalog.Validator do {:negative, %Atom{} = atom} -> check_atom(acc, atom, rels, Map.put(context, :polarity, :negative)) + {:callback, %ExDatalog.Callback{} = cb} -> + check_callback(acc, cb, Map.put(context, :polarity, :callback)) + other -> [ Error.new( @@ -154,6 +157,24 @@ defmodule ExDatalog.Validator do end) end + defp check_callback(errors, %ExDatalog.Callback{module: m, function: f, args: args}, context) do + arity = length(args) + + if Code.ensure_loaded?(m) and function_exported?(m, f, arity) do + errors + else + [ + Error.new( + :invalid_callback, + Map.merge(context, %{module: m, function: f, arity: arity}), + "#{location(context)} callback #{inspect(m)}.#{f}/#{arity} " <> + "is not defined or not exported" + ) + | errors + ] + end + end + defp check_atom(errors, %Atom{relation: rel, terms: terms}, rels, context) do case Map.fetch(rels, rel) do :error -> diff --git a/lib/ex_datalog/validator/error.ex b/lib/ex_datalog/validator/error.ex index 050df27..085a696 100644 --- a/lib/ex_datalog/validator/error.ex +++ b/lib/ex_datalog/validator/error.ex @@ -21,6 +21,10 @@ defmodule ExDatalog.Validator.Error do | `:unstratified_negation` | 2 | A negative edge appears in a dependency cycle | | `:unbound_constraint_variable` | 2 | A constraint references a variable not yet bound | | `:invalid_body_literal` | 1 | A body literal is not `{:positive, atom}` or `{:negative, atom}` | + | `:multiple_aggregates` | 2 | A rule contains more than one aggregate constraint | + | `:aggregate_in_recursion` | 2 | An aggregate appears in a self-recursive rule | + | `:unstratified_aggregate` | 2 | An aggregate's input relation is not in a strictly lower stratum | + | `:invalid_callback` | 2 | A callback references a missing module/function or wrong arity | ## Examples @@ -44,6 +48,10 @@ defmodule ExDatalog.Validator.Error do | :unbound_constraint_variable | :invalid_body_literal | :wildcard_in_head + | :multiple_aggregates + | :aggregate_in_recursion + | :unstratified_aggregate + | :invalid_callback @type t :: %__MODULE__{ kind: kind(), diff --git a/lib/ex_datalog/validator/safety.ex b/lib/ex_datalog/validator/safety.ex index 0ede0f8..cd17e2b 100644 --- a/lib/ex_datalog/validator/safety.ex +++ b/lib/ex_datalog/validator/safety.ex @@ -99,13 +99,41 @@ defmodule ExDatalog.Validator.Safety do |> check_wildcards_in_head(rule, rule_index) |> check_unsafe_head_variables(head_vars, head_bound, rule_index) |> check_unbound_constraint_variables(rule, body_bound, rule_index) + |> check_aggregates(rule, rule_index) + |> check_callback_inputs(rule, body_bound, rule_index) else [] end end + # Callback input variables must be bound by positive body atoms. Callbacks + # are filters (boolean) or value-binders; their inputs cannot be introduced + # by the callback itself. + defp check_callback_inputs(errors, %Rule{body: body}, body_bound, rule_index) do + callbacks = for {:callback, cb} <- body, do: cb + + Enum.reduce(callbacks, errors, fn cb, acc -> + unbound = Enum.reject(ExDatalog.Callback.input_variables(cb), fn v -> v in body_bound end) + + if unbound == [] do + acc + else + [ + Error.new( + :unbound_constraint_variable, + %{rule_index: rule_index, variables: unbound, callback: {cb.module, cb.function}}, + "rule #{rule_index}: callback #{inspect(cb.module)}.#{cb.function} " <> + "references unbound variable(s) #{Enum.join(unbound, ", ")}" + ) + | acc + ] + end + end) + end + defp valid_literal?({:positive, %Atom{}}), do: true defp valid_literal?({:negative, %Atom{}}), do: true + defp valid_literal?({:callback, %ExDatalog.Callback{}}), do: true defp valid_literal?(_), do: false # --- Private helpers --- @@ -126,13 +154,21 @@ defmodule ExDatalog.Validator.Safety do # Used for head-variable safety: if the rule body evaluates at all, every # arithmetic constraint will have computed its result, so arithmetic results # are in scope for the head regardless of constraint ordering. - defp all_bound_variables(%Rule{constraints: constraints} = rule) do - arithmetic_result_vars = + defp all_bound_variables(%Rule{constraints: constraints, body: body} = rule) do + result_vars = Enum.flat_map(constraints, fn c -> - if Constraint.arithmetic?(c), do: [Constraint.result_variable(c)], else: [] + if Constraint.arithmetic?(c) or Constraint.aggregate?(c), + do: [Constraint.result_variable(c)], + else: [] end) - (positive_body_variables(rule) ++ arithmetic_result_vars) |> Enum.uniq() + callback_result_vars = + for {:callback, cb} <- body, + name = ExDatalog.Callback.result_variable(cb), + name != nil, + do: name + + (positive_body_variables(rule) ++ result_vars ++ callback_result_vars) |> Enum.uniq() end defp check_wildcards_in_head(errors, %Rule{head: head}, rule_index) do @@ -210,10 +246,10 @@ defmodule ExDatalog.Validator.Safety do ] end - # Arithmetic constraints extend the bound set with their result variable - # so that subsequent constraints may reference it. + # Arithmetic and aggregate constraints extend the bound set with their + # result variable so that subsequent constraints may reference it. new_bound = - if Constraint.arithmetic?(c) do + if Constraint.arithmetic?(c) or Constraint.aggregate?(c) do case Constraint.result_variable(c) do nil -> bound var -> Enum.uniq([var | bound]) @@ -224,4 +260,59 @@ defmodule ExDatalog.Validator.Safety do {new_errors, new_bound} end + + # Aggregate-specific safety: + # 1. At most one aggregate per rule (initial v0.5.0 scope). + # 2. An aggregate may not appear in a self-recursive rule (aggregates are + # non-monotone and must be stratified above their inputs). + defp check_aggregates(errors, %Rule{} = rule, rule_index) do + aggregates = Rule.aggregate_constraints(rule) + + errors + |> check_single_aggregate(aggregates, rule, rule_index) + |> check_aggregate_not_recursive(aggregates, rule, rule_index) + end + + defp check_single_aggregate(errors, aggregates, _rule, rule_index) do + if length(aggregates) > 1 do + [ + Error.new( + :multiple_aggregates, + %{rule_index: rule_index, count: length(aggregates)}, + "rule #{rule_index}: a rule may contain at most one aggregate " <> + "(found #{length(aggregates)}); split into separate rules" + ) + | errors + ] + else + errors + end + end + + defp check_aggregate_not_recursive(errors, [], _rule, _rule_index), do: errors + + defp check_aggregate_not_recursive(errors, _aggregates, %Rule{} = rule, rule_index) do + head_rel = rule.head.relation + + self_recursive? = + Enum.any?(rule.body, fn + {:positive, %Atom{relation: ^head_rel}} -> true + {:negative, %Atom{relation: ^head_rel}} -> true + _ -> false + end) + + if self_recursive? do + [ + Error.new( + :aggregate_in_recursion, + %{rule_index: rule_index, relation: head_rel}, + "rule #{rule_index}: aggregate over a self-recursive relation " <> + "#{inspect(head_rel)} is not allowed; aggregates are non-monotone" + ) + | errors + ] + else + errors + end + end end diff --git a/lib/ex_datalog/validator/stratification.ex b/lib/ex_datalog/validator/stratification.ex index 0cca526..4531b0d 100644 --- a/lib/ex_datalog/validator/stratification.ex +++ b/lib/ex_datalog/validator/stratification.ex @@ -101,7 +101,74 @@ defmodule ExDatalog.Validator.Stratification do |> MapSet.union(MapSet.new(all_vertices(graph))) |> MapSet.to_list() - assign_strata_greedy(graph, sccs, all_rels) + graph + |> assign_strata_greedy(sccs, all_rels) + |> force_aggregate_strata(program) + end + + # Aggregates are non-monotone: a rule that aggregates over its positive body + # relations must evaluate in a stratum strictly above all of them. This runs a + # fixpoint that bumps each aggregate rule's head relation until the invariant + # holds, then re-propagates so dependents stay ordered. + defp force_aggregate_strata(strata, %ExDatalog.Program{rules: rules} = program) do + aggregate_rules = Enum.filter(rules, &ExDatalog.Rule.has_aggregates?/1) + + if aggregate_rules == [] do + strata + else + graph = build_graph(program) + stabilize_aggregate_strata(strata, aggregate_rules, graph) + end + end + + defp stabilize_aggregate_strata(strata, aggregate_rules, graph, fuel \\ 1000) + + defp stabilize_aggregate_strata(strata, _aggregate_rules, _graph, 0), do: strata + + defp stabilize_aggregate_strata(strata, aggregate_rules, graph, fuel) do + bumped = + Enum.reduce(aggregate_rules, strata, fn rule, acc -> + head_rel = rule.head.relation + + body_rels = + for {:positive, %Atom{relation: rel}} <- rule.body, rel != head_rel, do: rel + + required = + case body_rels do + [] -> Map.get(acc, head_rel, 0) + _ -> (body_rels |> Enum.map(&Map.get(acc, &1, 0)) |> Enum.max()) + 1 + end + + if Map.get(acc, head_rel, 0) < required do + Map.put(acc, head_rel, required) + else + acc + end + end) + + bumped = propagate_strata(bumped, graph) + + if bumped == strata do + strata + else + stabilize_aggregate_strata(bumped, aggregate_rules, graph, fuel - 1) + end + end + + # After bumping an aggregate head, every relation that positively depends on it + # must be at least as high; negative/aggregate dependents must be strictly + # higher. One pass; the outer fixpoint repeats until stable. + defp propagate_strata(strata, graph) do + Enum.reduce(graph, strata, fn {head_rel, deps}, acc -> + required = Enum.reduce(deps, Map.get(acc, head_rel, 0), &required_stratum(&1, &2, acc)) + Map.put(acc, head_rel, required) + end) + end + + defp required_stratum({dep, polarity}, current, acc) do + dep_stratum = Map.get(acc, dep, 0) + needed = if polarity == :negative, do: dep_stratum + 1, else: dep_stratum + max(current, needed) end @doc false diff --git a/livebooks/ex_datalog_v050.livemd b/livebooks/ex_datalog_v050.livemd new file mode 100644 index 0000000..6c1696c --- /dev/null +++ b/livebooks/ex_datalog_v050.livemd @@ -0,0 +1,778 @@ + + +# ExDatalog v0.5.0 — DSL Edition + +## Section + +A walkthrough of v0.5.0 features using the **Schema DSL** (`use ExDatalog.Schema`). + +```elixir +Mix.install([ + {:ex_datalog, path: Path.expand("..", __DIR__), env: :prod}, +]) +``` + + + +``` +:ok +``` + +## Introduction + +**ExDatalog v0.5.0** adds four major features: + +* **Aggregates** — `count`, `sum`, `min`, `max` over grouped bindings +* **BEAM callbacks** — call deterministic Elixir functions from rule bodies +* **Magic sets** — goal-directed evaluation that computes only relevant facts +* **Query planner** — inspect and explain evaluation plans before running them + +All examples below use the Schema DSL exclusively. + +--- + +## Setup + +```elixir +alias ExDatalog.{Knowledge, MagicSets, Planner} +``` + + + +``` +[ExDatalog.Knowledge, ExDatalog.MagicSets, ExDatalog.Planner] +``` + +--- + +## 1. Aggregates + +Aggregates compute a single value per group of bindings. In the DSL they appear +in the **rule body** as `count(X, N)`, `sum(A, T)`, `min(S, V)`, `max(S, V)`. +The result variable (`N`, `T`, `V`) must also appear in the rule head. + +> **Restrictions** (enforced at validation): +> +> * Only **one** aggregate per rule. +> * No aggregate over a **self-recursive** relation. +> * The aggregate input variable must be **bound** by a positive body atom. + +### Count — employees per department + +```elixir +defmodule DeptCount do + use ExDatalog.Schema + + relation :emp do + field(:name, :atom) + field(:dept, :atom) + end + + relation :dept_count do + field(:dept, :atom) + field(:n, :integer) + end + + fact(emp(:alice, :eng)) + fact(emp(:bob, :eng)) + fact(emp(:carol, :ops)) + fact(emp(:dave, :eng)) + fact(emp(:eve, :ops)) + + rule dept_count(D, N) do + emp(E, D) + count(E, N) + end +end + +{:ok, knowledge} = DeptCount.materialize() +Knowledge.get(knowledge, "dept_count") |> MapSet.to_list() |> Enum.sort() +``` + + + +``` +[eng: 3, ops: 2] +``` + +### Sum — total salary per department + +```elixir +defmodule DeptTotal do + use ExDatalog.Schema + + relation :salary do + field(:name, :atom) + field(:dept, :atom) + field(:amount, :integer) + end + + relation :dept_total do + field(:dept, :atom) + field(:total, :integer) + end + + fact(salary(:alice, :eng, 120)) + fact(salary(:bob, :eng, 80)) + fact(salary(:carol, :ops, 50)) + fact(salary(:dave, :eng, 100)) + + rule dept_total(D, T) do + salary(E, D, A) + sum(A, T) + end +end + +{:ok, knowledge} = DeptTotal.materialize() +Knowledge.get(knowledge, "dept_total") |> MapSet.to_list() |> Enum.sort() +``` + + + +``` +[eng: 300, ops: 50] +``` + +### Min and Max + +```elixir +defmodule DeptScores do + use ExDatalog.Schema + + relation :score do + field(:name, :atom) + field(:dept, :atom) + field(:value, :integer) + end + + relation :lowest do + field(:dept, :atom) + field(:value, :integer) + end + + relation :highest do + field(:dept, :atom) + field(:value, :integer) + end + + fact(score(:alice, :eng, 90)) + fact(score(:bob, :eng, 70)) + fact(score(:carol, :ops, 60)) + fact(score(:dave, :eng, 85)) + fact(score(:eve, :ops, 55)) + + rule lowest(D, V) do + score(E, D, S) + min(S, V) + end + + rule highest(D, V) do + score(E, D, S) + max(S, V) + end +end + +{:ok, knowledge} = DeptScores.materialize() + +IO.puts("Lowest per dept:") +Knowledge.get(knowledge, "lowest") |> MapSet.to_list() |> Enum.sort() |> IO.inspect() + +IO.puts("\nHighest per dept:") +Knowledge.get(knowledge, "highest") |> MapSet.to_list() |> Enum.sort() |> IO.inspect() +``` + + + +``` +Lowest per dept: +[eng: 70, ops: 55] + +Highest per dept: +[eng: 90, ops: 60] +``` + + + +``` +[eng: 90, ops: 60] +``` + +### Aggregate with a filter constraint + +Constraints are evaluated **before** grouping. Count only passing scores (≥ 60): + +```elixir +defmodule PassingCount do + use ExDatalog.Schema + + relation :score do + field(:name, :atom) + field(:dept, :atom) + field(:value, :integer) + end + + relation :passing_count do + field(:dept, :atom) + field(:n, :integer) + end + + fact(score(:alice, :eng, 90)) + fact(score(:bob, :eng, 40)) + fact(score(:carol, :eng, 75)) + fact(score(:dave, :ops, 30)) + fact(score(:eve, :ops, 80)) + + rule passing_count(D, N) do + score(E, D, S) + gte(S, 60) + count(E, N) + end +end + +{:ok, knowledge} = PassingCount.materialize() +Knowledge.get(knowledge, "passing_count") |> MapSet.to_list() |> Enum.sort() +``` + + + +``` +[eng: 2, ops: 1] +``` + +### Aggregate over a derived relation (stratification) + +When an aggregate reads from a **derived** relation, the engine automatically +places it in a higher stratum — the derived facts must be fully computed first. + +```elixir +defmodule StratifiedAggregate do + use ExDatalog.Schema + + relation :base do + field(:key, :atom) + field(:val, :atom) + end + + relation :mid do + field(:key, :atom) + field(:val, :atom) + end + + relation :mid_count do + field(:key, :atom) + field(:n, :integer) + end + + fact(base(:a, :x)) + fact(base(:a, :y)) + fact(base(:b, :z)) + + rule mid(K, V) do + base(K, V) + end + + rule mid_count(K, N) do + mid(K, V) + count(V, N) + end +end + +{:ok, knowledge} = StratifiedAggregate.materialize() +Knowledge.get(knowledge, "mid_count") |> MapSet.to_list() |> Enum.sort() +``` + + + +``` +[a: 2, b: 1] +``` + +`mid` is derived first (stratum 0), then the aggregate counts over it (stratum 1). + +--- + +## 2. BEAM Callbacks + +BEAM callbacks let rule bodies invoke **deterministic, side-effect-free** Elixir +functions. Declare a predicate with the `predicate` macro, then use its name in +a rule body. + +Two kinds: + +* **Boolean** (`:boolean`) — returns `true` to keep the binding, `false` to drop it. +* **Value** (`:value`) — the last argument in the rule-body call is the result + variable; the function's return value is bound to it. + +The engine enforces **timeout** (default 100 ms, configurable via +`:callback_timeout_ms`) and **exception isolation** — a callback that times out +or raises simply filters the binding. + +### Boolean filter predicate + +First define a helper module with the predicate function: + +```elixir +defmodule MyPredicates do + def adult?(age), do: age >= 18 + def valid_email?(email), do: String.contains?(email, "@") +end +``` + + + +``` +{:module, MyPredicates, <<70, 79, 82, 49, 0, 0, 7, ...>>, ...} +``` + +Now use `predicate` in the schema: + +```elixir +defmodule AdultOnly do + use ExDatalog.Schema + + relation :person do + field(:name, :atom) + field(:age, :integer) + end + + relation :adult do + field(:name, :atom) + end + + predicate(:adult?, MyPredicates, :adult?, [:integer], :boolean) + + fact(person(:alice, 25)) + fact(person(:bob, 12)) + fact(person(:carol, 19)) + + rule adult(Name) do + person(Name, Age) + adult?(Age) + end +end + +{:ok, knowledge} = AdultOnly.materialize() +Knowledge.get(knowledge, "adult") |> MapSet.to_list() |> Enum.sort() +``` + + + +``` +[{:alice}, {:carol}] +``` + +Only `:alice` and `:carol` pass — `:bob` is 12, so `adult?` returns `false`. + +### Value-returning predicate + +The last argument in the rule-body call is the **result variable**: + +```elixir +defmodule MyMath do + def double(x), do: x * 2 +end + +defmodule DoubledNumbers do + use ExDatalog.Schema + + relation :num do + field(:x, :integer) + end + + relation :doubled do + field(:x, :integer) + field(:y, :integer) + end + + predicate(:double, MyMath, :double, [:integer], :value) + + fact(num(3)) + fact(num(5)) + fact(num(7)) + + rule doubled(X, Y) do + num(X) + double(X, Y) + end +end + +{:ok, knowledge} = DoubledNumbers.materialize() +Knowledge.get(knowledge, "doubled") |> MapSet.to_list() |> Enum.sort() +``` + + + +``` +[{3, 6}, {5, 10}, {7, 14}] +``` + +### Exception and timeout isolation + +Callbacks that raise or exceed the timeout are silently filtered — the evaluator +never crashes. + +```elixir +defmodule UnsafePredicates do + def boom(_x), do: raise("exploded") + def slow(_x), do: Process.sleep(500) && true +end + +defmodule BoomTest do + use ExDatalog.Schema + + relation :num do + field(:x, :integer) + end + + relation :ok do + field(:x, :integer) + end + + predicate(:boom, UnsafePredicates, :boom, [:integer], :boolean) + + fact(num(1)) + + rule ok(X) do + num(X) + boom(X) + end +end + +# Raising callback — all bindings filtered +{:ok, knowledge} = BoomTest.materialize() +Knowledge.size(knowledge, "ok") +``` + + + +``` +0 +``` + +```elixir +defmodule SlowTest do + use ExDatalog.Schema + + relation :num do + field(:x, :integer) + end + + relation :ok do + field(:x, :integer) + end + + predicate(:slow, UnsafePredicates, :slow, [:integer], :boolean) + + fact(num(1)) + + rule ok(X) do + num(X) + slow(X) + end +end + +# Slow callback — filtered by 50 ms timeout +{:ok, knowledge} = SlowTest.materialize(callback_timeout_ms: 50) +Knowledge.size(knowledge, "ok") +``` + + + +``` +0 +``` + +--- + +## 3. Magic Sets — Goal-Directed Evaluation + +Magic sets is a **program transformation** that rewrites rules so bottom-up +evaluation computes only the facts relevant to a specific query goal, instead of +the full least fixpoint. + +> **Scope (v0.5.0, experimental):** +> +> * Positive recursive programs only (no negation, no aggregates in recursive rules). +> * A single goal with at least one bound position. +> * Programs outside scope fall back to full semi-naive evaluation — never incorrect results. + +### Full fixpoint vs goal-restricted + +```elixir +defmodule AncestorChain do + use ExDatalog.Schema + + relation :parent do + field(:from, :atom) + field(:to, :atom) + end + + relation :ancestor do + field(:from, :atom) + field(:to, :atom) + end + + fact(parent(:a, :b)) + fact(parent(:b, :c)) + fact(parent(:c, :d)) + fact(parent(:d, :e)) + + rule ancestor(X, Y) do + parent(X, Y) + end + + rule ancestor(X, Z) do + parent(X, Y) + ancestor(Y, Z) + end +end + +# Full semi-naive: all ancestor pairs +{:ok, full} = AncestorChain.materialize() +full_ancestors = Knowledge.get(full, "ancestor") |> MapSet.to_list() |> Enum.sort() + +IO.puts("Full semi-naive (#{length(full_ancestors)} facts):") +IO.inspect(full_ancestors) +``` + + + +``` +Full semi-naive (10 facts): +[a: :b, a: :c, a: :d, a: :e, b: :c, b: :d, b: :e, c: :d, c: :e, d: :e] +``` + + + +``` +[a: :b, a: :c, a: :d, a: :e, b: :c, b: :d, b: :e, c: :d, c: :e, d: :e] +``` + +Now with magic sets — goal: "who are the ancestors of :a?" + +```elixir +{:ok, magic} = + AncestorChain.materialize(strategy: :magic_sets, goal: {"ancestor", [:a, :_]}) + +magic_ancestors = + Knowledge.match(magic, "ancestor", [:a, :_]) |> MapSet.to_list() |> Enum.sort() + +IO.puts("Magic sets (#{length(magic_ancestors)} facts for goal {ancestor, [:a, :_]}):") +IO.inspect(magic_ancestors) + +# Verify: magic result is exactly the subset of full results where first arg is :a +expected = full_ancestors |> Enum.filter(fn {x, _y} -> x == :a end) +magic_ancestors == expected +``` + + + +``` +Magic sets (4 facts for goal {ancestor, [:a, :_]}): +[a: :b, a: :c, a: :d, a: :e] +``` + + + +``` +true +``` + +### Inspecting the transformation directly + +Call `MagicSets.transform/2` directly on compiled IR: + +```elixir +{:ok, ir} = ExDatalog.compile(AncestorChain.program()) +{:ok, transformed} = MagicSets.transform(ir, {"ancestor", [:a, :_]}) + +# The magic relation +magic = Enum.find(transformed.relations, fn r -> r.name == "magic_ancestor_bf" end) +IO.puts("Magic relation: #{magic.name}, arity: #{magic.arity}") + +# The seed fact +seed = Enum.find(transformed.facts, fn f -> f.relation == "magic_ancestor_bf" end) +IO.puts("Seed fact: #{inspect(seed.values)}") + +# Rewritten rules: ancestor rules now start with the magic predicate +ancestor_rules = Enum.filter(transformed.rules, fn r -> r.head.relation == "ancestor" end) +Enum.each(ancestor_rules, fn r -> + first_body = hd(r.body) + IO.puts("Rule head: ancestor, first body atom: #{inspect(elem(first_body, 1).relation)}") +end) +``` + + + +``` +Magic relation: magic_ancestor_bf, arity: 1 +Seed fact: [atom: :a] +Rule head: ancestor, first body atom: "magic_ancestor_bf" +Rule head: ancestor, first body atom: "magic_ancestor_bf" +``` + + + +``` +:ok +``` + +The adornment `bf` means the first position is **b**ound (`:a`) and the second +is **f**ree (`:_`). The magic relation `magic_ancestor_bf` has arity 1 (just the +bound position). + +### Fallback cases + +When the program is outside the supported scope, the engine falls back to +semi-naive without error: + +```elixir +# All-free goal falls back +{:ok, ms_free} = + AncestorChain.materialize(strategy: :magic_sets, goal: {"ancestor", [:_, :_]}) + +Knowledge.get(ms_free, "ancestor") == Knowledge.get(full, "ancestor") +``` + + + +``` +true +``` + +--- + +## 4. Query Planner + +The planner sits between the compiled IR and the evaluation engine. It produces +an `ExDatalog.Planner.Plan` describing the chosen `strategy`, `strata`, `joins`, +and `predicates`. + +### Plan + +```elixir +defmodule TransitivePath do + use ExDatalog.Schema + + relation :edge do + field(:from, :atom) + field(:to, :atom) + end + + relation :path do + field(:from, :atom) + field(:to, :atom) + end + + rule path(X, Y) do + edge(X, Y) + end + + rule path(X, Z) do + edge(X, Y) + path(Y, Z) + end +end + +IO.puts(Planner.explain_plan(TransitivePath.program())) +``` + + + +``` +Strategy: semi_naive + Stratum 0: 2 rule(s), relations: path, edge +Joins: 3 +Predicates: 0 +``` + + + +``` +:ok +``` + +### Plan with magic-sets strategy + +```elixir +IO.puts(Planner.explain_plan(TransitivePath.program(), strategy: :magic_sets, goal: {"path", [:a, :_]})) +``` + + + +``` +Strategy: magic_sets + Stratum 0: 2 rule(s), relations: path, edge +Joins: 3 +Predicates: 0 +``` + + + +``` +:ok +``` + +### Plan with aggregates + +The planner classifies predicates by kind (`:comparison`, `:arithmetic`, +`:aggregate`, `:callback`, etc.) and shows them in the explain output: + +```elixir +IO.puts(Planner.explain_plan(DeptCount.program())) +``` + + + +``` +Strategy: semi_naive + Stratum 0: 0 rule(s), relations: emp + Stratum 1: 1 rule(s), relations: dept_count +Joins: 1 +Predicates: 1 (count) +``` + + + +``` +:ok +``` + +### Plan with callbacks + +```elixir +IO.puts(Planner.explain_plan(AdultOnly.program())) +``` + + + +``` +Strategy: semi_naive + Stratum 0: 1 rule(s), relations: adult, person +Joins: 1 +Predicates: 0 +``` + + + +``` +:ok +``` + +--- + +## Summary + +| Feature | DSL Syntax | +| ------------ | ---------------------------------------------------------------------------- | +| Count | `count(E, N)` in rule body, `N` in head | +| Sum | `sum(A, T)` in rule body, `T` in head | +| Min | `min(S, V)` in rule body, `V` in head | +| Max | `max(S, V)` in rule body, `V` in head | +| Filter agg | `gte(S, 60)` before `count(E, N)` in body | +| Boolean CB | `predicate :name, Mod, :fun, [:type], :boolean` then `name(Arg)` | +| Value CB | `predicate :name, Mod, :fun, [:type], :value` then `name(Arg, Result)` | +| CB timeout | `materialize(callback_timeout_ms: 50)` | +| Magic sets | `materialize(strategy: :magic_sets, goal: {"rel", [:a, :_]})` | +| Plan | `Planner.plan(program)` or `Planner.plan(program, strategy: ..., goal: ...)` | +| Explain plan | `Planner.explain_plan(program)` or `Planner.explain_plan(program, opts)` | diff --git a/mix.exs b/mix.exs index 3d9ae92..e5f1270 100644 --- a/mix.exs +++ b/mix.exs @@ -1,7 +1,7 @@ defmodule ExDatalog.MixProject do use Mix.Project - @version "0.4.1" + @version "0.5.0" @source_url "https://github.com/thanos/ex_datalog" def project do @@ -63,6 +63,7 @@ defmodule ExDatalog.MixProject do {:excoveralls, "~> 0.18", only: :test}, {:mix_audit, "~> 2.1", only: [:dev, :test], runtime: false}, {:stream_data, "~> 1.1", only: [:dev, :test]}, + {:benchee, "~> 1.3", only: :dev, runtime: false}, {:ex_slop, "~> 0.1", only: [:dev, :test], runtime: false}, {:ex_dna, "~> 1.5", only: [:dev, :test], runtime: false} ] @@ -128,7 +129,18 @@ defmodule ExDatalog.MixProject do {"docs/articles/04_querying_materialized_knowledge.md", filename: "querying-materialized-knowledge", title: "Querying Materialized Knowledge"}, {"docs/articles/05_negation_constraints_and_safety.md", - filename: "negation-constraints-and-safety", title: "Negation, Constraints, and Safety"} + filename: "negation-constraints-and-safety", title: "Negation, Constraints, and Safety"}, + {"docs/articles/06_query_planning_in_datalog.md", + filename: "query-planning-in-datalog", title: "Query Planning in Datalog"}, + {"docs/articles/07_aggregates_in_datalog.md", + filename: "aggregates-in-datalog", title: "Aggregates in Datalog"}, + {"docs/articles/08_extending_datalog_with_beam_callbacks.md", + filename: "extending-datalog-with-beam-callbacks", + title: "Extending Datalog with BEAM Callbacks"}, + {"docs/articles/09_magic_sets_and_demand_driven_evaluation.md", + filename: "magic-sets-and-demand-driven-evaluation", + title: "Magic Sets and Demand-Driven Evaluation"}, + {"docs/migration_v0.5.md", filename: "migration-v0-5", title: "Migration: v0.4 → v0.5"} ], groups_for_modules: [ "Program Builder": [ @@ -145,7 +157,9 @@ defmodule ExDatalog.MixProject do ExDatalog.Validator.Stratification ], "Compiler & IR": ~r/ExDatalog\.(Compiler|IR).*/, + Planner: ~r/ExDatalog\.Planner.*/, Engine: ~r/ExDatalog\.Engine.*/, + "Magic Sets": [ExDatalog.MagicSets], Storage: ~r/ExDatalog\.Storage.*/, Knowledge: [ ExDatalog.Knowledge, @@ -154,6 +168,7 @@ defmodule ExDatalog.MixProject do ], DSL: [ ExDatalog.Schema, + ExDatalog.Callback, ExDatalog.DSL.CompileError, ExDatalog.UnsupportedFeature ] diff --git a/mix.lock b/mix.lock index a85066e..7d752e9 100644 --- a/mix.lock +++ b/mix.lock @@ -1,6 +1,8 @@ %{ + "benchee": {:hex, :benchee, "1.5.1", "b95cbc36c4b98969a5c592a246e171041eb683c56bad1cb4f49a3b081ba66087", [:mix], [{:deep_merge, "~> 1.0", [hex: :deep_merge, repo: "hexpm", optional: false]}, {:statistex, "~> 1.1", [hex: :statistex, repo: "hexpm", optional: false]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a539301f8dfd4efc5c5123bfb9d47ebde20092a863a5b5b16c2a60d2243dfce7"}, "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"}, "credo": {:hex, :credo, "1.7.18", "5c5596bf7aedf9c8c227f13272ac499fe8eae6237bd326f2f07dfc173786f042", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "a189d164685fd945809e862fe76a7420c4398fa288d76257662aecb909d6b3e5"}, + "deep_merge": {:hex, :deep_merge, "1.0.2", "476aa7ea61c54de96220051b998d893869069094da65b96101aebf79416f8a1e", [:mix], [], "hexpm", "737a53cdc9758fedbb608bdc213969e65729466c4ef3cd8e8726d0335dff116c"}, "dialyxir": {:hex, :dialyxir, "1.4.7", "dda948fcee52962e4b6c5b4b16b2d8fa7d50d8645bbae8b8685c3f9ecb7f5f4d", [:mix], [{:erlex, ">= 0.2.8", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "b34527202e6eb8cee198efec110996c25c5898f43a4094df157f8d28f27d9efe"}, "earmark_parser": {:hex, :earmark_parser, "1.4.44", "f20830dd6b5c77afe2b063777ddbbff09f9759396500cdbe7523efd58d7a339c", [:mix], [], "hexpm", "4778ac752b4701a5599215f7030989c989ffdc4f6df457c5f36938cc2d2a2750"}, "erlex": {:hex, :erlex, "0.2.8", "cd8116f20f3c0afe376d1e8d1f0ae2452337729f68be016ea544a72f767d9c12", [:mix], [], "hexpm", "9d66ff9fedf69e49dc3fd12831e12a8a37b76f8651dd21cd45fcf5561a8a7590"}, @@ -15,6 +17,7 @@ "makeup_erlang": {:hex, :makeup_erlang, "1.0.3", "4252d5d4098da7415c390e847c814bad3764c94a814a0b4245176215615e1035", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "953297c02582a33411ac6208f2c6e55f0e870df7f80da724ed613f10e6706afd"}, "mix_audit": {:hex, :mix_audit, "2.1.5", "c0f77cee6b4ef9d97e37772359a187a166c7a1e0e08b50edf5bf6959dfe5a016", [:make, :mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:yaml_elixir, "~> 2.11", [hex: :yaml_elixir, repo: "hexpm", optional: false]}], "hexpm", "87f9298e21da32f697af535475860dc1d3617a010e0b418d2ec6142bc8b42d69"}, "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, + "statistex": {:hex, :statistex, "1.1.1", "73612aa7f79e53c30569be065fd121e380f1cf57bc4c2da5b41be9246da18df9", [:mix], [], "hexpm", "310c4b49b34adf683de3103639006bed233ab54c08a4add65a531448e653857c"}, "stream_data": {:hex, :stream_data, "1.3.0", "bde37905530aff386dea1ddd86ecbf00e6642dc074ceffc10b7d4e41dfd6aac9", [:mix], [], "hexpm", "3cc552e286e817dca43c98044c706eec9318083a1480c52ae2688b08e2936e3c"}, "telemetry": {:hex, :telemetry, "1.4.1", "ab6de178e2b29b58e8256b92b382ea3f590a47152ca3651ea857a6cae05ac423", [:rebar3], [], "hexpm", "2172e05a27531d3d31dd9782841065c50dd5c3c7699d95266b2edd54c2dafa1c"}, "yamerl": {:hex, :yamerl, "0.10.0", "4ff81fee2f1f6a46f1700c0d880b24d193ddb74bd14ef42cb0bcf46e81ef2f8e", [:rebar3], [], "hexpm", "346adb2963f1051dc837a2364e4acf6eb7d80097c0f53cbdc3046ec8ec4b4e6e"}, diff --git a/test/ex_datalog/aggregate_test.exs b/test/ex_datalog/aggregate_test.exs new file mode 100644 index 0000000..3e4d099 --- /dev/null +++ b/test/ex_datalog/aggregate_test.exs @@ -0,0 +1,230 @@ +defmodule ExDatalog.AggregateTest do + use ExUnit.Case, async: true + + alias ExDatalog.{Atom, Constraint, Knowledge, Program, Rule, Term} + + doctest ExDatalog.Constraints.Aggregate + + # --- Builder API evaluation --- + + defp count_program do + Program.new() + |> Program.add_relation("emp", [:atom, :atom]) + |> Program.add_relation("dept_count", [:atom, :integer]) + |> Program.add_fact("emp", [:alice, :eng]) + |> Program.add_fact("emp", [:bob, :eng]) + |> Program.add_fact("emp", [:carol, :ops]) + |> Program.add_rule( + Rule.new( + Atom.new("dept_count", [Term.var("D"), Term.var("N")]), + [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}], + [Constraint.count(Term.var("E"), Term.var("N"))] + ) + ) + end + + describe "count aggregate" do + test "counts members per group" do + {:ok, knowledge} = ExDatalog.materialize(count_program()) + result = Knowledge.get(knowledge, "dept_count") + assert MapSet.size(result) == 2 + assert {:eng, 2} in result + assert {:ops, 1} in result + end + + test "aggregate stratum terminates at fixpoint" do + {:ok, knowledge} = ExDatalog.materialize(count_program()) + assert knowledge.stats.termination == :fixpoint + end + end + + describe "sum aggregate" do + test "sums integer values per group" do + program = + Program.new() + |> Program.add_relation("salary", [:atom, :atom, :integer]) + |> Program.add_relation("dept_total", [:atom, :integer]) + |> Program.add_fact("salary", [:alice, :eng, 100]) + |> Program.add_fact("salary", [:bob, :eng, 80]) + |> Program.add_fact("salary", [:carol, :ops, 50]) + |> Program.add_rule( + Rule.new( + Atom.new("dept_total", [Term.var("D"), Term.var("T")]), + [{:positive, Atom.new("salary", [Term.var("E"), Term.var("D"), Term.var("A")])}], + [Constraint.sum(Term.var("A"), Term.var("T"))] + ) + ) + + {:ok, knowledge} = ExDatalog.materialize(program) + result = Knowledge.get(knowledge, "dept_total") + assert {:eng, 180} in result + assert {:ops, 50} in result + end + end + + describe "min and max aggregates" do + defp score_program(op, head) do + Program.new() + |> Program.add_relation("score", [:atom, :atom, :integer]) + |> Program.add_relation(head, [:atom, :integer]) + |> Program.add_fact("score", [:alice, :eng, 90]) + |> Program.add_fact("score", [:bob, :eng, 70]) + |> Program.add_fact("score", [:carol, :ops, 60]) + |> Program.add_rule( + Rule.new( + Atom.new(head, [Term.var("D"), Term.var("V")]), + [{:positive, Atom.new("score", [Term.var("E"), Term.var("D"), Term.var("S")])}], + [apply(Constraint, op, [Term.var("S"), Term.var("V")])] + ) + ) + end + + test "min picks the smallest per group" do + {:ok, knowledge} = ExDatalog.materialize(score_program(:min, "lowest")) + result = Knowledge.get(knowledge, "lowest") + assert {:eng, 70} in result + assert {:ops, 60} in result + end + + test "max picks the largest per group" do + {:ok, knowledge} = ExDatalog.materialize(score_program(:max, "highest")) + result = Knowledge.get(knowledge, "highest") + assert {:eng, 90} in result + assert {:ops, 60} in result + end + end + + describe "aggregate with filter constraint" do + test "filters bindings before grouping" do + program = + Program.new() + |> Program.add_relation("score", [:atom, :atom, :integer]) + |> Program.add_relation("passing_count", [:atom, :integer]) + |> Program.add_fact("score", [:alice, :eng, 90]) + |> Program.add_fact("score", [:bob, :eng, 40]) + |> Program.add_fact("score", [:carol, :eng, 75]) + |> Program.add_rule( + Rule.new( + Atom.new("passing_count", [Term.var("D"), Term.var("N")]), + [{:positive, Atom.new("score", [Term.var("E"), Term.var("D"), Term.var("S")])}], + [ + Constraint.gte(Term.var("S"), Term.const(60)), + Constraint.count(Term.var("E"), Term.var("N")) + ] + ) + ) + + {:ok, knowledge} = ExDatalog.materialize(program) + result = Knowledge.get(knowledge, "passing_count") + assert {:eng, 2} in result + end + end + + describe "aggregate validation" do + test "rejects more than one aggregate per rule" do + program = + Program.new() + |> Program.add_relation("emp", [:atom, :atom, :integer]) + |> Program.add_relation("stats", [:atom, :integer, :integer]) + |> Program.add_rule( + Rule.new( + Atom.new("stats", [Term.var("D"), Term.var("N"), Term.var("T")]), + [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D"), Term.var("A")])}], + [ + Constraint.count(Term.var("E"), Term.var("N")), + Constraint.sum(Term.var("A"), Term.var("T")) + ] + ) + ) + + assert {:error, errors} = ExDatalog.materialize(program) + assert Enum.any?(errors, fn e -> e.kind == :multiple_aggregates end) + end + + test "rejects aggregate over a self-recursive relation" do + program = + Program.new() + |> Program.add_relation("path", [:atom, :integer]) + |> Program.add_rule( + Rule.new( + Atom.new("path", [Term.var("X"), Term.var("N")]), + [{:positive, Atom.new("path", [Term.var("X"), Term.var("Y")])}], + [Constraint.count(Term.var("Y"), Term.var("N"))] + ) + ) + + assert {:error, errors} = ExDatalog.materialize(program) + assert Enum.any?(errors, fn e -> e.kind == :aggregate_in_recursion end) + end + + test "rejects unbound aggregate input variable" do + program = + Program.new() + |> Program.add_relation("emp", [:atom, :atom]) + |> Program.add_relation("dept_count", [:atom, :integer]) + |> Program.add_rule( + Rule.new( + Atom.new("dept_count", [Term.var("D"), Term.var("N")]), + [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}], + # Z is not bound by any positive body atom + [Constraint.count(Term.var("Z"), Term.var("N"))] + ) + ) + + assert {:error, errors} = ExDatalog.materialize(program) + assert Enum.any?(errors, fn e -> e.kind == :unbound_constraint_variable end) + end + end + + describe "aggregate stratification" do + test "aggregate over a derived relation is placed in a higher stratum" do + # base -> mid (derived) -> count over mid + program = + Program.new() + |> Program.add_relation("base", [:atom, :atom]) + |> Program.add_relation("mid", [:atom, :atom]) + |> Program.add_relation("mid_count", [:atom, :integer]) + |> Program.add_fact("base", [:a, :x]) + |> Program.add_fact("base", [:a, :y]) + |> Program.add_fact("base", [:b, :z]) + |> Program.add_rule( + Rule.new( + Atom.new("mid", [Term.var("K"), Term.var("V")]), + [{:positive, Atom.new("base", [Term.var("K"), Term.var("V")])}] + ) + ) + |> Program.add_rule( + Rule.new( + Atom.new("mid_count", [Term.var("K"), Term.var("N")]), + [{:positive, Atom.new("mid", [Term.var("K"), Term.var("V")])}], + [Constraint.count(Term.var("V"), Term.var("N"))] + ) + ) + + {:ok, ir} = ExDatalog.compile(program) + mid_stratum = Enum.find(ir.rules, fn r -> r.head.relation == "mid" end).stratum + count_stratum = Enum.find(ir.rules, fn r -> r.head.relation == "mid_count" end).stratum + assert count_stratum > mid_stratum + + {:ok, knowledge} = ExDatalog.materialize(program) + result = Knowledge.get(knowledge, "mid_count") + assert {:a, 2} in result + assert {:b, 1} in result + end + end + + describe "aggregate via from_tuple" do + test "builds aggregate constraints" do + assert %Constraint{op: :count, left: {:var, "X"}, right: nil, result: {:var, "N"}} = + Constraint.from_tuple({:count, {:var, "X"}, {:var, "N"}}) + + assert %Constraint{op: :sum, result: {:var, "T"}} = + Constraint.from_tuple({:sum, :A, :T}) + end + + test "valid? accepts aggregate constraints" do + assert Constraint.valid?(Constraint.count(Term.var("X"), Term.var("N"))) + assert Constraint.valid?(Constraint.sum(Term.var("X"), Term.var("T"))) + end + end +end diff --git a/test/ex_datalog/beam_callback_test.exs b/test/ex_datalog/beam_callback_test.exs new file mode 100644 index 0000000..2c061d9 --- /dev/null +++ b/test/ex_datalog/beam_callback_test.exs @@ -0,0 +1,209 @@ +defmodule ExDatalog.BeamCallbackTest do + use ExUnit.Case, async: true + + alias ExDatalog.{Atom, Callback, Knowledge, Program, Rule, Term} + + doctest ExDatalog.Callback + + defmodule Predicates do + @moduledoc false + def adult?(age), do: age >= 18 + def valid_email?(email), do: String.contains?(email, "@") + def double(x), do: x * 2 + def boom(_x), do: raise("callback exploded") + def slow(_x), do: Process.sleep(500) && true + end + + describe "boolean callback (builder API)" do + test "filters bindings by predicate result" do + program = + Program.new() + |> Program.add_relation("person", [:atom, :integer]) + |> Program.add_relation("adult", [:atom]) + |> Program.add_fact("person", [:alice, 25]) + |> Program.add_fact("person", [:bob, 12]) + |> Program.add_rule( + Rule.new( + Atom.new("adult", [Term.var("Name")]), + [ + {:positive, Atom.new("person", [Term.var("Name"), Term.var("Age")])}, + {:callback, Callback.new(Predicates, :adult?, [Term.var("Age")])} + ] + ) + ) + + {:ok, knowledge} = ExDatalog.materialize(program) + result = Knowledge.get(knowledge, "adult") + assert MapSet.size(result) == 1 + assert {:alice} in result + end + end + + describe "value-returning callback" do + test "binds the result variable" do + program = + Program.new() + |> Program.add_relation("num", [:integer]) + |> Program.add_relation("doubled", [:integer, :integer]) + |> Program.add_fact("num", [3]) + |> Program.add_fact("num", [5]) + |> Program.add_rule( + Rule.new( + Atom.new("doubled", [Term.var("X"), Term.var("Y")]), + [ + {:positive, Atom.new("num", [Term.var("X")])}, + {:callback, Callback.new(Predicates, :double, [Term.var("X")], Term.var("Y"))} + ] + ) + ) + + {:ok, knowledge} = ExDatalog.materialize(program) + result = Knowledge.get(knowledge, "doubled") + assert {3, 6} in result + assert {5, 10} in result + end + end + + describe "exception and timeout isolation" do + test "a raising callback filters the binding" do + program = + Program.new() + |> Program.add_relation("num", [:integer]) + |> Program.add_relation("ok", [:integer]) + |> Program.add_fact("num", [1]) + |> Program.add_rule( + Rule.new( + Atom.new("ok", [Term.var("X")]), + [ + {:positive, Atom.new("num", [Term.var("X")])}, + {:callback, Callback.new(Predicates, :boom, [Term.var("X")])} + ] + ) + ) + + {:ok, knowledge} = ExDatalog.materialize(program) + assert MapSet.size(Knowledge.get(knowledge, "ok")) == 0 + end + + test "a slow callback is filtered by the timeout" do + program = + Program.new() + |> Program.add_relation("num", [:integer]) + |> Program.add_relation("ok", [:integer]) + |> Program.add_fact("num", [1]) + |> Program.add_rule( + Rule.new( + Atom.new("ok", [Term.var("X")]), + [ + {:positive, Atom.new("num", [Term.var("X")])}, + {:callback, Callback.new(Predicates, :slow, [Term.var("X")])} + ] + ) + ) + + {:ok, knowledge} = ExDatalog.materialize(program, callback_timeout_ms: 50) + assert MapSet.size(Knowledge.get(knowledge, "ok")) == 0 + end + end + + describe "validation" do + test "rejects a callback referencing a missing function" do + program = + Program.new() + |> Program.add_relation("num", [:integer]) + |> Program.add_relation("ok", [:integer]) + |> Program.add_fact("num", [1]) + |> Program.add_rule( + Rule.new( + Atom.new("ok", [Term.var("X")]), + [ + {:positive, Atom.new("num", [Term.var("X")])}, + {:callback, Callback.new(Predicates, :nonexistent, [Term.var("X")])} + ] + ) + ) + + assert {:error, errors} = ExDatalog.materialize(program) + assert Enum.any?(errors, fn e -> e.kind == :invalid_callback end) + end + + test "rejects a callback with an unbound input variable" do + program = + Program.new() + |> Program.add_relation("num", [:integer]) + |> Program.add_relation("ok", [:integer]) + |> Program.add_fact("num", [1]) + |> Program.add_rule( + Rule.new( + Atom.new("ok", [Term.var("X")]), + [ + {:positive, Atom.new("num", [Term.var("X")])}, + {:callback, Callback.new(Predicates, :adult?, [Term.var("Z")])} + ] + ) + ) + + assert {:error, errors} = ExDatalog.materialize(program) + assert Enum.any?(errors, fn e -> e.kind == :unbound_constraint_variable end) + end + end + + describe "DSL predicate macro" do + test "boolean predicate in a rule body" do + defmodule AdultSchema do + use ExDatalog.Schema + + relation :person do + field(:name, :atom) + field(:age, :integer) + end + + relation :adult do + field(:name, :atom) + end + + predicate(:adult?, ExDatalog.BeamCallbackTest.Predicates, :adult?, [:integer], :boolean) + + fact(person(:alice, 30)) + fact(person(:bob, 10)) + + rule adult(Name) do + person(Name, Age) + adult?(Age) + end + end + + {:ok, knowledge} = AdultSchema.materialize() + result = Knowledge.get(knowledge, "adult") + assert {:alice} in result + refute {:bob} in result + end + + test "value predicate binds the last argument" do + defmodule DoubleSchema do + use ExDatalog.Schema + + relation :num do + field(:x, :integer) + end + + relation :doubled do + field(:x, :integer) + field(:y, :integer) + end + + predicate(:double, ExDatalog.BeamCallbackTest.Predicates, :double, [:integer], :value) + + fact(num(4)) + + rule doubled(X, Y) do + num(X) + double(X, Y) + end + end + + {:ok, knowledge} = DoubleSchema.materialize() + assert {4, 8} in Knowledge.get(knowledge, "doubled") + end + end +end diff --git a/test/ex_datalog/magic_sets_property_test.exs b/test/ex_datalog/magic_sets_property_test.exs new file mode 100644 index 0000000..fb84c1c --- /dev/null +++ b/test/ex_datalog/magic_sets_property_test.exs @@ -0,0 +1,59 @@ +defmodule ExDatalog.MagicSetsPropertyTest do + use ExUnit.Case, async: true + use ExUnitProperties + + alias ExDatalog.{Atom, Knowledge, Program, Rule, Term} + + @nodes [:a, :b, :c, :d, :e, :f] + + defp edge_gen do + gen all(from <- member_of(@nodes), to <- member_of(@nodes)) do + {from, to} + end + end + + defp build_program(edges) do + base = + Program.new() + |> Program.add_relation("edge", [:atom, :atom]) + |> Program.add_relation("path", [:atom, :atom]) + + base = Enum.reduce(edges, base, fn {f, t}, acc -> Program.add_fact(acc, "edge", [f, t]) end) + + base + |> Program.add_rule( + Rule.new( + Atom.new("path", [Term.var("X"), Term.var("Y")]), + [{:positive, Atom.new("edge", [Term.var("X"), Term.var("Y")])}] + ) + ) + |> Program.add_rule( + Rule.new( + Atom.new("path", [Term.var("X"), Term.var("Z")]), + [ + {:positive, Atom.new("edge", [Term.var("X"), Term.var("Y")])}, + {:positive, Atom.new("path", [Term.var("Y"), Term.var("Z")])} + ] + ) + ) + end + + property "magic-sets result equals the semi-naive subset for the goal" do + check all( + edges <- list_of(edge_gen(), max_length: 10), + source <- member_of(@nodes) + ) do + program = build_program(edges) + + {:ok, full} = ExDatalog.materialize(program) + expected = MapSet.filter(Knowledge.get(full, "path"), fn {x, _y} -> x == source end) + + {:ok, magic} = + ExDatalog.materialize(program, strategy: :magic_sets, goal: {"path", [source, :_]}) + + actual = Knowledge.match(magic, "path", [source, :_]) + + assert actual == expected + end + end +end diff --git a/test/ex_datalog/magic_sets_test.exs b/test/ex_datalog/magic_sets_test.exs new file mode 100644 index 0000000..cf7c9e1 --- /dev/null +++ b/test/ex_datalog/magic_sets_test.exs @@ -0,0 +1,143 @@ +defmodule ExDatalog.MagicSetsTest do + use ExUnit.Case, async: true + + alias ExDatalog.{Atom, Knowledge, MagicSets, Program, Rule, Term} + + defp ancestor_program(facts) do + base = + Program.new() + |> Program.add_relation("parent", [:atom, :atom]) + |> Program.add_relation("ancestor", [:atom, :atom]) + + base = Enum.reduce(facts, base, fn {p, c}, acc -> Program.add_fact(acc, "parent", [p, c]) end) + + base + |> Program.add_rule( + Rule.new( + Atom.new("ancestor", [Term.var("X"), Term.var("Y")]), + [{:positive, Atom.new("parent", [Term.var("X"), Term.var("Y")])}] + ) + ) + |> Program.add_rule( + Rule.new( + Atom.new("ancestor", [Term.var("X"), Term.var("Z")]), + [ + {:positive, Atom.new("parent", [Term.var("X"), Term.var("Y")])}, + {:positive, Atom.new("ancestor", [Term.var("Y"), Term.var("Z")])} + ] + ) + ) + end + + @chain [{:a, :b}, {:b, :c}, {:c, :d}, {:d, :e}] + + describe "correctness vs semi-naive" do + test "goal-restricted query matches the semi-naive subset" do + program = ancestor_program(@chain) + + {:ok, full} = ExDatalog.materialize(program) + full_ancestors = Knowledge.get(full, "ancestor") + expected = MapSet.filter(full_ancestors, fn {x, _y} -> x == :a end) + + {:ok, magic} = + ExDatalog.materialize(program, strategy: :magic_sets, goal: {"ancestor", [:a, :_]}) + + result = Knowledge.match(magic, "ancestor", [:a, :_]) + assert result == expected + end + + test "falls back to semi-naive when no goal is given" do + program = ancestor_program(@chain) + {:ok, magic} = ExDatalog.materialize(program, strategy: :magic_sets) + {:ok, full} = ExDatalog.materialize(program) + assert Knowledge.get(magic, "ancestor") == Knowledge.get(full, "ancestor") + end + + test "falls back to semi-naive for an all-free goal" do + program = ancestor_program(@chain) + + {:ok, magic} = + ExDatalog.materialize(program, strategy: :magic_sets, goal: {"ancestor", [:_, :_]}) + + {:ok, full} = ExDatalog.materialize(program) + assert Knowledge.get(magic, "ancestor") == Knowledge.get(full, "ancestor") + end + end + + describe "transform/2 directly" do + test "produces a magic relation, seed fact, and rewritten rules for a bound goal" do + {:ok, ir} = ExDatalog.compile(ancestor_program(@chain)) + {:ok, transformed} = MagicSets.transform(ir, {"ancestor", [:a, :_]}) + + magic = Enum.find(transformed.relations, fn r -> r.name == "magic_ancestor_bf" end) + assert magic.arity == 1 + + assert Enum.any?(transformed.facts, fn f -> + f.relation == "magic_ancestor_bf" and f.values == [atom: :a] + end) + + rewritten = + Enum.filter(transformed.rules, fn r -> r.head.relation == "ancestor" end) + + assert Enum.all?(rewritten, fn r -> + match?( + [{:positive, %ExDatalog.IR.Atom{relation: "magic_ancestor_bf"}} | _], + r.body + ) + end) + end + + test "returns :fallback for an all-free goal" do + {:ok, ir} = ExDatalog.compile(ancestor_program(@chain)) + assert {:fallback, :no_bound_positions} = MagicSets.transform(ir, {"ancestor", [:_, :_]}) + end + + test "returns :fallback for a program containing aggregates" do + program = + Program.new() + |> Program.add_relation("emp", [:atom, :atom]) + |> Program.add_relation("dept_count", [:atom, :integer]) + |> Program.add_fact("emp", [:alice, :eng]) + |> Program.add_rule( + Rule.new( + Atom.new("dept_count", [Term.var("D"), Term.var("N")]), + [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}], + [ExDatalog.Constraint.count(Term.var("E"), Term.var("N"))] + ) + ) + + {:ok, ir} = ExDatalog.compile(program) + + assert {:fallback, :unsupported_program} = + MagicSets.transform(ir, {"dept_count", [:eng, :_]}) + end + end + + describe "unsupported programs fall back" do + test "program with negation falls back to semi-naive" do + program = + Program.new() + |> Program.add_relation("person", [:atom]) + |> Program.add_relation("parent", [:atom, :atom]) + |> Program.add_relation("childless", [:atom]) + |> Program.add_fact("person", [:alice]) + |> Program.add_fact("person", [:bob]) + |> Program.add_fact("parent", [:alice, :carol]) + |> Program.add_rule( + Rule.new( + Atom.new("childless", [Term.var("X")]), + [ + {:positive, Atom.new("person", [Term.var("X")])}, + {:negative, Atom.new("parent", [Term.var("X"), :wildcard])} + ] + ) + ) + + {:ok, magic} = + ExDatalog.materialize(program, strategy: :magic_sets, goal: {"childless", [:_]}) + + {:ok, full} = ExDatalog.materialize(program) + assert Knowledge.get(magic, "childless") == Knowledge.get(full, "childless") + end + end +end diff --git a/test/ex_datalog/planner_explain_test.exs b/test/ex_datalog/planner_explain_test.exs new file mode 100644 index 0000000..c5d65d0 --- /dev/null +++ b/test/ex_datalog/planner_explain_test.exs @@ -0,0 +1,60 @@ +defmodule ExDatalog.PlannerExplainTest do + use ExUnit.Case, async: true + + alias ExDatalog.{Atom, Planner, Program, Rule, Term} + + defp transitive_program do + Program.new() + |> Program.add_relation("edge", [:atom, :atom]) + |> Program.add_relation("path", [:atom, :atom]) + |> Program.add_fact("edge", [:a, :b]) + |> Program.add_rule( + Rule.new( + Atom.new("path", [Term.var("X"), Term.var("Y")]), + [{:positive, Atom.new("edge", [Term.var("X"), Term.var("Y")])}] + ) + ) + end + + describe "explain_plan/1,2" do + test "describes the strategy" do + assert Planner.explain_plan(transitive_program()) =~ "Strategy: semi_naive" + end + + test "lists strata" do + output = Planner.explain_plan(transitive_program()) + assert output =~ "Stratum 0" + end + + test "reports join and predicate counts" do + output = Planner.explain_plan(transitive_program()) + assert output =~ "Joins: 1" + assert output =~ "Predicates: 0" + end + + test "honours the magic_sets strategy option" do + output = + Planner.explain_plan(transitive_program(), + strategy: :magic_sets, + goal: {"path", [:a, :_]} + ) + + assert output =~ "Strategy: magic_sets" + end + + test "returns an error string for an invalid program" do + bad = + Program.new() + |> Program.add_relation("path", [:atom, :atom]) + |> Program.add_rule( + Rule.new( + # Z is unsafe — not bound by any positive body atom + Atom.new("path", [Term.var("X"), Term.var("Z")]), + [{:positive, Atom.new("path", [Term.var("X"), Term.var("Y")])}] + ) + ) + + assert Planner.explain_plan(bad) =~ "compilation failed" + end + end +end diff --git a/test/ex_datalog/planner_test.exs b/test/ex_datalog/planner_test.exs new file mode 100644 index 0000000..f6f5a5c --- /dev/null +++ b/test/ex_datalog/planner_test.exs @@ -0,0 +1,98 @@ +defmodule ExDatalog.PlannerTest do + use ExUnit.Case, async: true + + doctest ExDatalog.Planner + + alias ExDatalog.{Atom, Compiler, Constraint, Planner, Program, Rule, Term} + alias ExDatalog.Planner.{Join, Plan, Predicate, Stratum} + + defp transitive_ir do + {:ok, ir} = + Program.new() + |> Program.add_relation("edge", [:atom, :atom]) + |> Program.add_relation("path", [:atom, :atom]) + |> Program.add_fact("edge", [:a, :b]) + |> Program.add_rule( + Rule.new( + Atom.new("path", [Term.var("X"), Term.var("Y")]), + [{:positive, Atom.new("edge", [Term.var("X"), Term.var("Y")])}] + ) + ) + |> Program.add_rule( + Rule.new( + Atom.new("path", [Term.var("X"), Term.var("Z")]), + [ + {:positive, Atom.new("edge", [Term.var("X"), Term.var("Y")])}, + {:positive, Atom.new("path", [Term.var("Y"), Term.var("Z")])} + ] + ) + ) + |> Compiler.compile() + + ir + end + + describe "plan/2" do + test "produces a semi_naive plan by default" do + {:ok, plan} = Planner.plan(transitive_ir()) + assert %Plan{strategy: :semi_naive} = plan + end + + test "wraps IR strata into Stratum structs with rules" do + {:ok, plan} = Planner.plan(transitive_ir()) + assert Enum.all?(plan.strata, &match?(%Stratum{}, &1)) + assert Enum.any?(plan.strata, fn s -> s.rules != [] end) + end + + test "produces one join per positive body atom" do + {:ok, plan} = Planner.plan(transitive_ir()) + # rule 1: 1 positive atom; rule 2: 2 positive atoms => 3 joins + assert length(plan.joins) == 3 + assert Enum.all?(plan.joins, &match?(%Join{}, &1)) + end + + test "accepts the magic_sets strategy" do + {:ok, plan} = Planner.plan(transitive_ir(), strategy: :magic_sets, goal: {"path", [:a, :_]}) + assert plan.strategy == :magic_sets + assert plan.metadata.goal == {"path", [:a, :_]} + end + + test "classifies comparison constraints as predicates" do + {:ok, ir} = + Program.new() + |> Program.add_relation("income", [:atom, :integer]) + |> Program.add_relation("rich", [:atom]) + |> Program.add_fact("income", [:alice, 200_000]) + |> Program.add_rule( + Rule.new( + Atom.new("rich", [Term.var("P")]), + [{:positive, Atom.new("income", [Term.var("P"), Term.var("S")])}], + [Constraint.gt(Term.var("S"), Term.const(100_000))] + ) + ) + |> Compiler.compile() + + {:ok, plan} = Planner.plan(ir) + assert [%Predicate{kind: :comparison, op: :gt}] = plan.predicates + end + + test "classifies aggregate constraints as aggregate predicates" do + {:ok, ir} = + Program.new() + |> Program.add_relation("emp", [:atom, :atom]) + |> Program.add_relation("dept_count", [:atom, :integer]) + |> Program.add_fact("emp", [:alice, :eng]) + |> Program.add_rule( + Rule.new( + Atom.new("dept_count", [Term.var("D"), Term.var("N")]), + [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}], + [Constraint.count(Term.var("E"), Term.var("N"))] + ) + ) + |> Compiler.compile() + + {:ok, plan} = Planner.plan(ir) + assert [%Predicate{kind: :aggregate, op: :count}] = plan.predicates + end + end +end diff --git a/test/ex_datalog/runtime_facts_test.exs b/test/ex_datalog/runtime_facts_test.exs new file mode 100644 index 0000000..1cab0a0 --- /dev/null +++ b/test/ex_datalog/runtime_facts_test.exs @@ -0,0 +1,306 @@ +defmodule ExDatalog.RuntimeFactsTest do + use ExUnit.Case, async: true + + alias ExDatalog.{Knowledge, Program} + + describe "Schema relation constructors" do + test "constructor returns {relation, values} tuple" do + defmodule EmpSchema do + use ExDatalog.Schema + + relation :emp do + field(:name, :atom) + field(:dept, :atom) + end + + relation :dept_count do + field(:dept, :atom) + field(:n, :integer) + end + + rule dept_count(D, N) do + emp(E, D) + count(E, N) + end + end + + assert EmpSchema.emp(:alice, :eng) == {"emp", [:alice, :eng]} + assert EmpSchema.dept_count(:eng, 3) == {"dept_count", [:eng, 3]} + end + end + + describe "Schema.new/0" do + test "returns a blank program with no facts" do + defmodule BlankSchema do + use ExDatalog.Schema + + relation :edge do + field(:from, :atom) + field(:to, :atom) + end + + relation :path do + field(:from, :atom) + field(:to, :atom) + end + + rule path(X, Y) do + edge(X, Y) + end + end + + prog = BlankSchema.new() + assert prog.facts == [] + assert Map.has_key?(prog.relations, "edge") + assert Map.has_key?(prog.relations, "path") + end + + test "new/0 excludes compile-time facts; program/0 includes them" do + defmodule AxiomSchema do + use ExDatalog.Schema + + relation :edge do + field(:from, :atom) + field(:to, :atom) + end + + fact(edge(:a, :b)) + + rule edge(X, Y) do + edge(X, Y) + end + end + + blank = AxiomSchema.new() + assert blank.facts == [] + + with_axioms = AxiomSchema.program() + assert {"edge", [:a, :b]} in with_axioms.facts + end + end + + describe "Program.add_fact/2 (tuple form)" do + test "adds a fact from a {relation, values} tuple" do + prog = + Program.new() + |> Program.add_relation("emp", [:atom, :atom]) + |> Program.add_fact({"emp", [:alice, :eng]}) + + assert {"emp", [:alice, :eng]} in prog.facts + end + + test "pipeable with schema constructors" do + defmodule PipeSchema do + use ExDatalog.Schema + + relation :edge do + field(:from, :atom) + field(:to, :atom) + end + + relation :path do + field(:from, :atom) + field(:to, :atom) + end + + rule path(X, Y) do + edge(X, Y) + end + + rule path(X, Z) do + edge(X, Y) + path(Y, Z) + end + end + + {:ok, k} = + PipeSchema.new() + |> Program.add_fact(PipeSchema.edge(:a, :b)) + |> Program.add_fact(PipeSchema.edge(:b, :c)) + |> Program.materialize() + + result = Knowledge.get(k, "path") |> MapSet.to_list() |> Enum.sort() + assert {:a, :b} in result + assert {:b, :c} in result + assert {:a, :c} in result + end + + test "returns error for unknown relation" do + prog = Program.new() |> Program.add_relation("emp", [:atom, :atom]) + assert {:error, _} = Program.add_fact(prog, {"unknown", [:x]}) + end + + test "returns error for arity mismatch" do + prog = Program.new() |> Program.add_relation("emp", [:atom, :atom]) + assert {:error, _} = Program.add_fact(prog, {"emp", [:only_one]}) + end + end + + describe "Program.add_facts/2" do + test "adds multiple facts at once" do + prog = + Program.new() + |> Program.add_relation("emp", [:atom, :atom]) + |> Program.add_facts([ + {"emp", [:alice, :eng]}, + {"emp", [:bob, :eng]}, + {"emp", [:carol, :ops]} + ]) + + assert length(prog.facts) == 3 + end + + test "stops on first error" do + prog = Program.new() |> Program.add_relation("emp", [:atom, :atom]) + + assert {:error, _} = + Program.add_facts(prog, [ + {"emp", [:alice, :eng]}, + {"unknown", [:x]}, + {"emp", [:bob, :eng]} + ]) + end + + test "pipeable with schema constructors" do + defmodule BulkSchema do + use ExDatalog.Schema + + relation :emp do + field(:name, :atom) + field(:dept, :atom) + end + + relation :dept_count do + field(:dept, :atom) + field(:n, :integer) + end + + rule dept_count(D, N) do + emp(E, D) + count(E, N) + end + end + + {:ok, k} = + BulkSchema.new() + |> Program.add_facts([ + BulkSchema.emp(:alice, :eng), + BulkSchema.emp(:bob, :eng), + BulkSchema.emp(:carol, :ops), + BulkSchema.emp(:dave, :eng), + BulkSchema.emp(:eve, :ops) + ]) + |> Program.materialize() + + result = Knowledge.get(k, "dept_count") |> MapSet.to_list() |> MapSet.new() + assert {:eng, 3} in result + assert {:ops, 2} in result + end + end + + describe "Program.materialize/1,2" do + test "pipe-friendly materialization" do + prog = + Program.new() + |> Program.add_relation("edge", [:atom, :atom]) + |> Program.add_fact("edge", [:a, :b]) + + {:ok, k} = Program.materialize(prog) + assert Knowledge.get(k, "edge") |> MapSet.size() == 1 + end + + test "passes options through" do + defmodule MatOptsSchema do + use ExDatalog.Schema + + relation :edge do + field(:from, :atom) + field(:to, :atom) + end + + fact(edge(:a, :b)) + end + + {:ok, k} = Program.materialize(MatOptsSchema.program(), storage: ExDatalog.Storage.Map) + assert Knowledge.get(k, "edge") |> MapSet.size() == 1 + end + end + + describe "end-to-end: Schema.new + runtime facts + aggregates + materialize" do + test "full runtime pipeline" do + defmodule RuntimeDeptCount do + use ExDatalog.Schema + + relation :emp do + field(:name, :atom) + field(:dept, :atom) + end + + relation :dept_count do + field(:dept, :atom) + field(:n, :integer) + end + + rule dept_count(D, N) do + emp(E, D) + count(E, N) + end + end + + {:ok, k} = + RuntimeDeptCount.new() + |> Program.add_facts([ + RuntimeDeptCount.emp(:alice, :eng), + RuntimeDeptCount.emp(:bob, :eng), + RuntimeDeptCount.emp(:carol, :ops) + ]) + |> Program.materialize() + + result = Knowledge.get(k, "dept_count") |> MapSet.to_list() |> Enum.sort() + assert {:eng, 2} in result + assert {:ops, 1} in result + end + + test "mixing compile-time facts with runtime facts" do + defmodule MixedSchema do + use ExDatalog.Schema + + relation :edge do + field(:from, :atom) + field(:to, :atom) + end + + relation :path do + field(:from, :atom) + field(:to, :atom) + end + + fact(edge(:a, :b)) + + rule path(X, Y) do + edge(X, Y) + end + + rule path(X, Z) do + edge(X, Y) + path(Y, Z) + end + end + + # program() includes compile-time fact :a→:b + prog = MixedSchema.program() + # Add runtime facts + prog = Program.add_fact(prog, MixedSchema.edge(:b, :c)) + prog = Program.add_fact(prog, MixedSchema.edge(:c, :d)) + + {:ok, k} = Program.materialize(prog) + result = Knowledge.get(k, "path") |> MapSet.to_list() |> Enum.sort() + assert {:a, :b} in result + assert {:b, :c} in result + assert {:c, :d} in result + assert {:a, :c} in result + assert {:a, :d} in result + assert {:b, :d} in result + end + end +end diff --git a/test/ex_datalog/schema_test.exs b/test/ex_datalog/schema_test.exs index bb38e5f..2aa8086 100644 --- a/test/ex_datalog/schema_test.exs +++ b/test/ex_datalog/schema_test.exs @@ -617,7 +617,7 @@ defmodule ExDatalog.SchemaTest do describe "UnsupportedFeature" do test "aggregate syntax in rule head raises DSL.CompileError" do - assert_raise ExDatalog.DSL.CompileError, ~r/aggregates are not yet supported/, fn -> + assert_raise ExDatalog.DSL.CompileError, ~r/use count\/sum\/min\/max for aggregates/, fn -> Code.compile_string(""" defmodule AggHeadTestErr do use ExDatalog.Schema @@ -643,7 +643,7 @@ defmodule ExDatalog.SchemaTest do end test "aggregate syntax in rule body raises DSL.CompileError" do - assert_raise ExDatalog.DSL.CompileError, ~r/aggregates are not yet supported/, fn -> + assert_raise ExDatalog.DSL.CompileError, ~r/use count\/sum\/min\/max for aggregates/, fn -> Code.compile_string(""" defmodule AggBodyTestErr do use ExDatalog.Schema @@ -1446,6 +1446,92 @@ defmodule ExDatalog.SchemaTest do assert MapSet.size(Knowledge.get(knowledge, "reachable")) == 1 end end + + describe "aggregate DSL" do + test "count aggregate in rule body" do + defmodule CountDSLTest do + use ExDatalog.Schema + + relation :emp do + field(:name, :atom) + field(:dept, :atom) + end + + relation :dept_count do + field(:dept, :atom) + field(:n, :integer) + end + + fact(emp(:alice, :eng)) + fact(emp(:bob, :eng)) + fact(emp(:carol, :ops)) + + rule dept_count(D, N) do + emp(E, D) + count(E, N) + end + end + + {:ok, knowledge} = CountDSLTest.materialize() + result = Knowledge.get(knowledge, "dept_count") + assert {:eng, 2} in result + assert {:ops, 1} in result + end + + test "sum aggregate in rule body" do + defmodule SumDSLTest do + use ExDatalog.Schema + + relation :salary do + field(:name, :atom) + field(:dept, :atom) + field(:amount, :integer) + end + + relation :dept_total do + field(:dept, :atom) + field(:total, :integer) + end + + fact(salary(:alice, :eng, 100)) + fact(salary(:bob, :eng, 80)) + + rule dept_total(D, T) do + salary(E, D, A) + sum(A, T) + end + end + + {:ok, knowledge} = SumDSLTest.materialize() + assert {:eng, 180} in Knowledge.get(knowledge, "dept_total") + end + + test "aggregate in head position raises DSL.CompileError" do + assert_raise ExDatalog.DSL.CompileError, ~r/cannot appear in a rule head/, fn -> + Code.compile_string(""" + defmodule AggHeadAggTest do + use ExDatalog.Schema + + relation :emp do + field(:name, :atom) + field(:dept, :atom) + end + + relation :dept_count do + field(:dept, :atom) + field(:n, :integer) + end + + fact(emp(:alice, :eng)) + + rule dept_count(D, count(E, N)) do + emp(E, D) + end + end + """) + end + end + end end defmodule ExDatalog.SchemaCoverageTest do @@ -1814,7 +1900,7 @@ defmodule ExDatalog.SchemaErrorTest do end test "aggregate in rule head raises DSL.CompileError" do - assert_raise ExDatalog.DSL.CompileError, ~r/aggregates are not yet supported/, fn -> + assert_raise ExDatalog.DSL.CompileError, ~r/use count\/sum\/min\/max for aggregates/, fn -> Code.compile_string(""" defmodule AggHeadErrTest do use ExDatalog.Schema @@ -1840,7 +1926,7 @@ defmodule ExDatalog.SchemaErrorTest do end test "aggregate in rule body raises DSL.CompileError" do - assert_raise ExDatalog.DSL.CompileError, ~r/aggregates are not yet supported/, fn -> + assert_raise ExDatalog.DSL.CompileError, ~r/use count\/sum\/min\/max for aggregates/, fn -> Code.compile_string(""" defmodule AggBodyErrTest do use ExDatalog.Schema diff --git a/test/ex_datalog/v050_coverage_test.exs b/test/ex_datalog/v050_coverage_test.exs new file mode 100644 index 0000000..b6e4134 --- /dev/null +++ b/test/ex_datalog/v050_coverage_test.exs @@ -0,0 +1,103 @@ +defmodule ExDatalog.V050CoverageTest do + use ExUnit.Case, async: true + + alias ExDatalog.{Atom, Callback, Constraint, Constraints, IR, Rule, Term} + alias ExDatalog.Planner.{Join, Plan, Predicate, Stratum} + + describe "planner structs" do + test "structs hold their fields" do + stratum = %Stratum{index: 0, rules: [], relations: ["edge"]} + join = %Join{relation: "edge", position: 0} + pred = %Predicate{kind: :comparison, op: :gt} + + plan = %Plan{strategy: :semi_naive, strata: [stratum], joins: [join], predicates: [pred]} + + assert plan.strategy == :semi_naive + assert hd(plan.strata).relations == ["edge"] + assert hd(plan.joins).relation == "edge" + assert hd(plan.joins).strategy == :nested_loop + assert hd(plan.predicates).kind == :comparison + end + end + + describe "Aggregate.evaluate/3 is not callable per-binding" do + test "raises a clear error" do + c = %IR.Constraint{op: :count, left: {:var, "X"}, right: nil, result: {:var, "N"}} + + assert_raise RuntimeError, ~r/not evaluated per-binding/, fn -> + Constraints.Aggregate.evaluate(c, %{}, %ExDatalog.Constraint.Context{}) + end + end + end + + describe "Rule callback helpers" do + test "has_callbacks?/1 and callbacks/1" do + cb = Callback.new(String, :length, [Term.var("S")]) + + rule = + Rule.new( + Atom.new("r", [Term.var("S")]), + [{:positive, Atom.new("s", [Term.var("S")])}, {:callback, cb}] + ) + + assert Rule.has_callbacks?(rule) + assert Rule.callbacks(rule) == [cb] + assert "S" in Rule.variables(rule) + assert Rule.body_atoms(rule) == [Atom.new("s", [Term.var("S")])] + end + + test "value-returning callback contributes its result variable to variables/1" do + cb = Callback.new(M, :f, [Term.var("X")], Term.var("R")) + + rule = + Rule.new( + Atom.new("r", [Term.var("X"), Term.var("R")]), + [{:positive, Atom.new("s", [Term.var("X")])}, {:callback, cb}] + ) + + assert "R" in Rule.variables(rule) + end + end + + describe "Rule aggregate helpers" do + test "aggregate_constraints/1 and non_aggregate_constraints/1 partition constraints" do + agg = Constraint.count(Term.var("E"), Term.var("N")) + cmp = Constraint.gt(Term.var("N"), Term.const(0)) + + rule = + Rule.new( + Atom.new("r", [Term.var("D"), Term.var("N")]), + [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}], + [agg, cmp] + ) + + assert Rule.has_aggregates?(rule) + assert Rule.aggregate_constraints(rule) == [agg] + assert Rule.non_aggregate_constraints(rule) == [cmp] + end + end + + describe "Callback introspection" do + test "input_variables/1 and result_variable/1" do + cb = Callback.new(M, :f, [Term.var("A"), Term.const(1)], Term.var("R")) + assert Callback.input_variables(cb) == ["A"] + assert Callback.result_variable(cb) == "R" + + bool = Callback.new(M, :g, [Term.var("A")]) + assert Callback.result_variable(bool) == nil + end + end + + describe "Constraint aggregate op guards" do + test "aggregate_op?/1 and aggregate?/1" do + assert Constraint.aggregate_op?(:count) + refute Constraint.aggregate_op?(:gt) + assert Constraint.aggregate?(Constraint.sum(Term.var("X"), Term.var("T"))) + end + + test "min and max constructors" do + assert %Constraint{op: :min} = Constraint.min(Term.var("X"), Term.var("V")) + assert %Constraint{op: :max} = Constraint.max(Term.var("X"), Term.var("V")) + end + end +end