;; lib/datalog/tests/semi_naive.sx — semi-naive correctness vs naive.
;;
;; Strategy: differential — run both saturators on each program and
;; compare the resulting per-relation tuple counts. Counting (not
;; element-wise set equality) keeps the suite fast under the bundled
;; conformance session; correctness on the inhabitants is covered by
;; eval.sx and builtins.sx (which use dl-saturate! by default — the
;; semi-naive saturator).

(define dl-sn-pass 0)
(define dl-sn-fail 0)
(define dl-sn-failures (list))

(define
  dl-sn-test!
  (fn
    (name got expected)
    (if
      (equal? got expected)
      (set! dl-sn-pass (+ dl-sn-pass 1))
      (do
        (set! dl-sn-fail (+ dl-sn-fail 1))
        (append!
          dl-sn-failures
          (str name "\n    expected: " expected "\n    got:      " got))))))

;; Load `source` into both a semi-naive and a naive db and return a
;; list of (rel-name semi-count naive-count) triples. Both sets must
;; have the same union of relation names.
(define
  dl-sn-counts
  (fn
    (source)
    (let
      ((db-s (dl-program source)) (db-n (dl-program source)))
      (do
        (dl-saturate! db-s)
        (dl-saturate-naive! db-n)
        (let
          ((out (list)))
          (do
            (for-each
              (fn
                (k)
                (append!
                  out
                  (list
                    k
                    (len (dl-relation db-s k))
                    (len (dl-relation db-n k)))))
              (keys (get db-s :facts)))
            out))))))

(define
  dl-sn-counts-agree?
  (fn
    (counts)
    (cond
      ((= (len counts) 0) true)
      (else
        (let
          ((row (first counts)))
          (and
            (= (nth row 1) (nth row 2))
            (dl-sn-counts-agree? (rest counts))))))))

(define
  dl-sn-chain-source
  (fn
    (n)
    (let
      ((parts (list "")))
      (do
        (define
          dl-sn-loop
          (fn
            (i)
            (when
              (< i n)
              (do
                (append! parts (str "parent(" i ", " (+ i 1) "). "))
                (dl-sn-loop (+ i 1))))))
        (dl-sn-loop 0)
        (str
          (join "" parts)
          "ancestor(X, Y) :- parent(X, Y). "
          "ancestor(X, Z) :- parent(X, Y), ancestor(Y, Z).")))))

(define
  dl-sn-run-all!
  (fn
    ()
    (do
      (dl-sn-test!
        "ancestor closure counts match"
        (dl-sn-counts-agree?
          (dl-sn-counts
            "parent(a, b). parent(b, c). parent(c, d).\n             ancestor(X, Y) :- parent(X, Y).\n             ancestor(X, Z) :- parent(X, Y), ancestor(Y, Z)."))
        true)
      (dl-sn-test!
        "cyclic reach counts match"
        (dl-sn-counts-agree?
          (dl-sn-counts
            "edge(1, 2). edge(2, 3). edge(3, 1). edge(3, 4).\n             reach(X, Y) :- edge(X, Y).\n             reach(X, Z) :- edge(X, Y), reach(Y, Z)."))
        true)
      (dl-sn-test!
        "same-gen counts match"
        (dl-sn-counts-agree?
          (dl-sn-counts
            "parent(a, b). parent(a, c). parent(b, d). parent(c, e).\n             person(a). person(b). person(c). person(d). person(e).\n             sg(X, X) :- person(X).\n             sg(X, Y) :- parent(P1, X), sg(P1, P2), parent(P2, Y)."))
        true)
      (dl-sn-test!
        "rules with builtins counts match"
        (dl-sn-counts-agree?
          (dl-sn-counts
            "n(1). n(2). n(3). n(4). n(5).\n             small(X) :- n(X), <(X, 5).\n             succ(X, Y) :- n(X), <(X, 5), is(Y, +(X, 1))."))
        true)
      (dl-sn-test!
        "static rule fires under semi-naive"
        (dl-sn-counts-agree?
          (dl-sn-counts "p(a). p(b). q(X) :- p(X), =(X, a)."))
        true)
      ;; Chain length 12 — multiple semi-naive iterations against
      ;; the recursive ancestor rule (differential vs naive).
      (dl-sn-test!
        "chain-12 ancestor counts match"
        (dl-sn-counts-agree? (dl-sn-counts (dl-sn-chain-source 12)))
        true)
      ;; Chain length 25 — semi-naive only — first-arg index makes
      ;; this tractable in conformance budget.
      (dl-sn-test!
        "chain-25 ancestor count value (semi only)"
        (let
          ((db (dl-program (dl-sn-chain-source 25))))
          (do (dl-saturate! db) (len (dl-relation db "ancestor"))))
        325)
      (dl-sn-test!
        "query through semi saturate"
        (let
          ((db (dl-program "parent(a, b). parent(b, c).\n                  ancestor(X, Y) :- parent(X, Y).\n                  ancestor(X, Z) :- parent(X, Y), ancestor(Y, Z).")))
          (len (dl-query db (list (quote ancestor) (quote a) (quote X)))))
        2))))

(define
  dl-semi-naive-tests-run!
  (fn
    ()
    (do
      (set! dl-sn-pass 0)
      (set! dl-sn-fail 0)
      (set! dl-sn-failures (list))
      (dl-sn-run-all!)
      {:failures dl-sn-failures :total (+ dl-sn-pass dl-sn-fail) :passed dl-sn-pass :failed dl-sn-fail})))