Import datalog

projects/datalog/test/python/test_datalog_evaluator.py

@@ -0,0 +1,227 @@
+"""Query evaluation unit tests."""
+
+from datalog.easy import read, select
+from datalog.types import (
+    CachedDataset,
+    Constant,
+    Dataset,
+    LVar,
+    PartlyIndexedDataset,
+    Rule,
+    TableIndexedDataset,
+)
+
+import pytest
+
+
+DBCLS = [Dataset, CachedDataset, TableIndexedDataset, PartlyIndexedDataset]
+
+
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_id_query(db_cls):
+    """Querying for a constant in the dataset."""
+
+    ab = (
+        Constant("a"),
+        Constant("b"),
+    )
+    assert not select(db_cls([], []), ("a", "b",))
+    assert select(db_cls([ab], []), ("a", "b",)) == [((("a", "b"),), {},)]
+
+
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_lvar_query(db_cls):
+    """Querying for a binding in the dataset."""
+
+    d = read("""a(b). a(c).""", db_cls=db_cls)
+
+    assert select(d, ("a", "X")) == [
+        ((("a", "b"),), {"X": "b"}),
+        ((("a", "c"),), {"X": "c"}),
+    ]
+
+
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_lvar_unification(db_cls):
+    """Querying for MATCHING bindings in the dataset."""
+
+    d = read("""edge(b, c). edge(c, c).""", db_cls=db_cls)
+
+    assert select(d, ("edge", "X", "X",)) == [((("edge", "c", "c"),), {"X": "c"})]
+
+
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_rule_join(db_cls):
+    """Test a basic join query - the parent -> grandparent relation."""
+
+    child = Constant("child")
+    gc = Constant("grandchild")
+
+    d = read(
+        """
+child(a, b).
+child(b, c).
+child(b, d).
+child(b, e).
+
+grandchild(A, B) :-
+  child(A, C),
+  child(C, B).
+""",
+        db_cls=db_cls,
+    )
+
+    assert select(d, ("grandchild", "a", "X",)) == [
+        ((("grandchild", "a", "c"),), {"X": "c"}),
+        ((("grandchild", "a", "d"),), {"X": "d"}),
+        ((("grandchild", "a", "e"),), {"X": "e"}),
+    ]
+
+
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_antijoin(db_cls):
+    """Test a query containing an antijoin."""
+
+    d = read(
+        """
+a(foo, bar).
+b(foo, bar).
+a(baz, qux).
+% matching b(baz, qux). is our antijoin test
+
+no-b(X, Y) :-
+  a(X, Y),
+  ~b(X, Z).
+""",
+        db_cls=db_cls,
+    )
+
+    assert select(d, ("no-b", "X", "Y")) == [
+        ((("no-b", "baz", "qux"),), {"X": "baz", "Y": "qux"})
+    ]
+
+
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_nested_antijoin(db_cls):
+    """Test a query which negates a subquery which uses an antijoin.
+
+  Shouldn't exercise anything more than `test_antjoin` does, but it's an interesting case since you
+  actually can't capture the same semantics using a single query. Antijoins can't propagate positive
+  information (create lvar bindings) so I'm not sure you can express this another way without a
+  different evaluation strategy.
+
+  """
+
+    d = read(
+        """
+a(foo, bar).
+b(foo, bar).
+a(baz, qux).
+b(baz, quack).
+
+b-not-quack(X, Y) :-
+  b(X, Y),
+  ~=(Y, quack).
+
+a-no-nonquack(X, Y) :-
+  a(X, Y),
+  ~b-not-quack(X, Y).
+""",
+        db_cls=db_cls,
+    )
+
+    assert select(d, ("a-no-nonquack", "X", "Y")) == [
+        ((("a-no-nonquack", "baz", "qux"),), {"X": "baz", "Y": "qux"})
+    ]
+
+
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_alternate_rule(db_cls):
+    """Testing that both recursion and alternation work."""
+
+    d = read(
+        """
+edge(a, b).
+edge(b, c).
+edge(c, d).
+edge(d, e).
+edge(e, f).
+
+path(A, B) :-
+  edge(A, B).
+
+path(A, B) :-
+  edge(A, C),
+  path(C, B).
+""",
+        db_cls=db_cls,
+    )
+
+    # Should be able to recurse to this one.
+    assert select(d, ("path", "a", "f")) == [((("path", "a", "f"),), {})]
+
+
+# FIXME (arrdem 2019-06-13):
+#
+#   This test is BROKEN for the simple dataset.  In order for left-recursive production rules to
+#   work, they have to ground out somewhere.  Under the naive, cache-less datalog this is an
+#   infinite left recursion.  Under the cached versions, the right-recursion becomes iteration over
+#   an incrementally realized list which ... is weird but does work because the recursion grounds
+#   out in iterating over an empty list on the 2nd round then falls through to the other production
+#   rule which generates ground tuples and feeds everything.
+#
+#   It's not clear how to make this work with the current (lack of) query planner on the naive db as
+#   really fixing this requires some amount of insight into the data dependency structure and may
+#   involve reordering rules.
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_alternate_rule_lrec(db_cls):
+    """Testing that both recursion and alternation work."""
+
+    d = read(
+        """
+edge(a, b).
+edge(b, c).
+edge(c, d).
+edge(d, e).
+edge(e, f).
+
+path(A, B) :-
+  edge(A, B).
+
+path(A, B) :-
+  path(A, C),
+  edge(C, B).
+""",
+        db_cls=db_cls,
+    )
+
+    # Should be able to recurse to this one.
+    assert select(d, ("path", "a", "f")) == [((("path", "a", "f"),), {})]
+
+
+@pytest.mark.parametrize("db_cls,", DBCLS)
+def test_cojoin(db_cls):
+    """Tests that unification occurs correctly."""
+
+    d = read(
+        """
+edge(a, b).
+edge(b, c).
+edge(c, d).
+edge(d, e).
+edge(e, f).
+edge(c, q).
+
+two_path(A, B, C) :- edge(A, B), edge(B, C).
+""",
+        db_cls=db_cls,
+    )
+
+    # Should be able to recurse to this one.
+    assert [t for t, _ in select(d, ("two_path", "A", "B", "C"))] == [
+        (("two_path", "a", "b", "c"),),
+        (("two_path", "b", "c", "d"),),
+        (("two_path", "b", "c", "q"),),
+        (("two_path", "c", "d", "e"),),
+        (("two_path", "d", "e", "f"),),
+    ]

projects/datalog/test/python/test_datalog_reader.py

@@ -0,0 +1,31 @@
+"""
+Reader tests.
+"""
+
+import pytest
+
+from datalog.reader import read
+
+EXS = [
+    "%foo\n",
+    """a(b).""",
+    """edge(a).""",
+    """a(b, c).""",
+    """edge(smfc-ait, smfc).""",
+    """edge("smfc-ait", smfc).""",
+    """path(A, B) :- edge(A, C), path(C, B).""",
+    """path(A, B) :-
+  edge(A, C),
+  path(C, B).""",
+    """path(A, B) :- % one comment
+  edge(A, C), % the next
+  path(C, B).""",
+    """foo(A, B) :-
+  ~bar(A, B),
+  qux(A, B).""",
+]
+
+
+@pytest.mark.parametrize("ex,", EXS)
+def test_reader(ex):
+    assert read(ex)