From d695554123ce6ad45454d76a3d295d1b97e3e88a Mon Sep 17 00:00:00 2001
From: Richard Newman <rnewman@twinql.com>
Date: Fri, 15 Jul 2016 09:00:49 -0700
Subject: [PATCH] Initial work on executing queries. r=nalexander

Signed-off-by: Richard Newman <rnewman@twinql.com>
---
 src/datomish/exec.cljc      | 39 +++++++++++++++++++++
 src/datomish/exec_repl.cljc | 69 +++++++++++++++++++++++++++++++++++++
 src/datomish/pair_chan.cljc | 27 +++++++++++++--
 src/datomish/query.cljc     | 30 +++++++++-------
 src/datomish/sqlite.cljc    | 30 +++++++++++-----
 5 files changed, 172 insertions(+), 23 deletions(-)
 create mode 100644 src/datomish/exec.cljc
 create mode 100644 src/datomish/exec_repl.cljc

diff --git a/src/datomish/exec.cljc b/src/datomish/exec.cljc
new file mode 100644
index 00000000..0b7b095b
--- /dev/null
+++ b/src/datomish/exec.cljc
@@ -0,0 +1,39 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+(ns datomish.exec
+  #?(:cljs
+     (:require-macros
+      [datomish.util :refer [while-let]]
+      [datomish.pair-chan :refer [go-pair <?]]
+      [cljs.core.async.macros :refer [go]]))
+  (:require
+     [datomish.sqlite :as s]
+     [datomish.sqlite-schema :as ss]
+     [datomish.query :as dq]
+     #?@(:clj
+           [[datomish.jdbc-sqlite]
+            [datomish.pair-chan :refer [go-pair <?]]
+            [datomish.util :refer [while-let]]
+            [clojure.core.async :refer
+             [go                   ; macro in cljs.
+              <! >! chan close! take!]]])
+     #?@(:cljs
+           [[datomish.promise-sqlite]
+            [datomish.pair-chan]
+            [datomish.util]
+            [cljs.core.async :as a :refer
+             [<! >! chan close! take!]]])))
+
+(defn <?run
+  "Execute the provided query on the provided DB.
+   Returns a transduced channel of [result err] pairs.
+   Closes the channel when fully consumed."
+  [db find]
+  (let [context (dq/find->prepared-context (dq/parse find))
+        row-pair-transducer (dq/row-pair-transducer context (dq/sql-projection context))
+        chan (chan 50 row-pair-transducer)]
+
+    (s/<?all-rows db (dq/context->sql-string context) chan)
+    chan))
diff --git a/src/datomish/exec_repl.cljc b/src/datomish/exec_repl.cljc
new file mode 100644
index 00000000..2a6ccdee
--- /dev/null
+++ b/src/datomish/exec_repl.cljc
@@ -0,0 +1,69 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+(ns datomish.exec-repl
+  #?(:cljs
+     (:require-macros
+      [datomish.util :refer [while-let]]
+      [datomish.pair-chan :refer [go-pair <?]]
+      [cljs.core.async.macros :refer [go]]))
+  (:require
+     [datomish.exec :as exec]
+     [datomish.sqlite :as s]
+     [datomish.sqlite-schema :as ss]
+     [datomish.query :as dq]
+     #?@(:clj
+           [[datomish.jdbc-sqlite]
+            [datomish.pair-chan :refer [go-pair <?]]
+            [datomish.util :refer [while-let]]
+            [clojure.core.async]])
+     #?@(:cljs
+           [[datomish.promise-sqlite]
+            [datomish.pair-chan]
+            [datomish.util]])))
+
+#?(:clj
+(defn pair-channel->lazy-seq
+  "Returns a blocking lazy sequence of items taken from the provided channel."
+  [channel]
+  (lazy-seq
+   (when-let [v (clojure.core.async/<!! channel)]
+     (if (second v)
+       (cons v nil)
+       (cons v (channel->lazy-seq channel)))))))
+
+#?(:clj
+(defn run-to-pair-seq
+  "Given an open database, returns a lazy sequence of results.
+   When fully consumed, underlying resources will be released."
+  [db find]
+  (pair-channel->lazy-seq (exec/<?run db find))))
+
+#_(defn xxopen []
+    (datomish.pair-chan/go-pair
+      (let [d (datomish.pair-chan/<? (s/<sqlite-connection "/tmp/foo.sqlite"))]
+        (clojure.core.async/<!! (ss/<ensure-current-version d))
+        (def db d))))
+
+;; With an open DB…
+#_(datomish.exec/run-to-pair-seq
+    db
+    '[:find ?page :in $ :where [?page :page/starred true ?t]])
+
+;; In a Clojure REPL with no open DB…
+#_(clojure.core.async/<!!
+    (datomish.exec-repl/<open-and-run-to-seq-promise
+      "/tmp/foo.sqlite"
+      '[:find ?page :in $ :where [?page :page/starred true ?t]]))
+
+#_(defn test-cljs []
+    (datomish.pair-chan/go-pair
+      (let [d (datomish.pair-chan/<? (s/<sqlite-connection "/tmp/foo.sqlite"))]
+        (cljs.core.async/<! (ss/<ensure-current-version d))
+        (let [chan (exec/<?run d
+                               '[:find ?page :in $ :where [?page :page/starred true ?t]])]
+          (println (datomish.pair-chan/<? chan))
+          (println (datomish.pair-chan/<? chan))
+          (println (datomish.pair-chan/<? chan))))))
+
diff --git a/src/datomish/pair_chan.cljc b/src/datomish/pair_chan.cljc
index fd3a2efc..326477ff 100644
--- a/src/datomish/pair_chan.cljc
+++ b/src/datomish/pair_chan.cljc
@@ -16,11 +16,34 @@
   [then else]
   (if (cljs-env? &env) then else))
 
+(defmacro go-safely [[chan chan-form] & body]
+  "Evaluate `body` forms in a `go` block. Binds `chan-form` to `chan`.
+   `chan-form` must evaluate to an error-channel.
+   If `body` throws, the exception is propagated into `chan` and `chan` is closed.
+   Returns `chan`."
+  `(if-cljs
+     (let [~chan ~chan-form]
+       (cljs.core.async.macros/go
+         (try
+           (do ~@body)
+           (catch js/Error ex#
+             (cljs.core.async/>! ~chan [nil ex#]))))
+       ~chan)
+     (let [~chan ~chan-form]
+       (clojure.core.async/go
+         (try
+           (do ~@body)
+           (catch Exception ex#
+             (clojure.core.async/>! ~chan [nil ex#]))))
+       ~chan)))
+
 ;; It's a huge pain to declare cross-environment macros.  This is awful, but making the namespace a
 ;; parameter appears to be *even worse*.  Note also that `go` is not in a consistent namespace...
 (defmacro go-pair [& body]
-  "Evaluate `body` forms in a `go` block.  Catch errors and return a
-  pair chan (a promise channel resolving to `[result error]`)."
+  "Evaluate `body` forms in a `go` block to yield a result.
+   Catch errors during evaluation.
+   Return a promise channel that yields a pair: the result (or nil), and any
+   error thrown (or nil)."
   `(if-cljs
      (let [pc-chan# (cljs.core.async/promise-chan)]
        (cljs.core.async.macros/go
diff --git a/src/datomish/query.cljc b/src/datomish/query.cljc
index d2b403cb..e66db737 100644
--- a/src/datomish/query.cljc
+++ b/src/datomish/query.cljc
@@ -160,16 +160,14 @@
              [(lookup-variable context var) (var->sql-var var)]))
          elements)))
 
-(defn row-transducer [context projection rf]
+(defn row-pair-transducer [context projection]
   ;; For now, we only support straight var lists, so
   ;; our transducer is trivial.
-  (let [columns-in-order (map second projection)
-        row-mapper (fn [row] (map columns-in-order row))]
-    (fn
-      ([] (rf))
-      ([result] (rf result))
-      ([result input]
-       (rf result (row-mapper input))))))
+  (let [columns-in-order (map second projection)]
+    (map (fn [[row err]]
+           (if err
+             [row err]
+             [(map row columns-in-order) nil])))))
 
 (defn context->sql-clause [context]
   {:select (sql-projection context)
@@ -178,6 +176,12 @@
             nil
             (cons :and (:wheres context)))})
 
+(defn context->sql-string [context]
+  (->
+    context
+    context->sql-clause
+    (sql/format :quoting sql-quoting-style)))
+
 (defn- validate-with [with]
   (when-not (nil? with)
     (raise "`with` not supported.")))
@@ -206,13 +210,15 @@
   ;; There's some confusing use of 'where' and friends here. That's because
   ;; the parsed Datalog includes :where, and it's also input to honeysql's
   ;; SQL formatter.
-  (context->sql-clause
-    (find->prepared-context find)))
+  (-> find find->prepared-context context->sql-clause))
 
 (defn find->sql-string
   "Take a parsed `find` expression and turn it into SQL."
   [find]
-  (-> find find->sql-clause (sql/format :quoting sql-quoting-style)))
+  (->
+    find
+    find->sql-clause
+    (sql/format :quoting sql-quoting-style)))
 
 (defn parse
   "Parse a Datalog query array into a structured `find` expression."
@@ -225,7 +231,7 @@
       '[:find ?page :in $ :where [?page :page/starred true ?t] ])))
 
 (comment
-  (datomish.query/find->sql-string
+  (datomish.query/find->prepared-context
     (datomish.query/parse
       '[:find ?timestampMicros ?page
         :in $
diff --git a/src/datomish/sqlite.cljc b/src/datomish/sqlite.cljc
index c7f966f6..db4e804a 100644
--- a/src/datomish/sqlite.cljc
+++ b/src/datomish/sqlite.cljc
@@ -5,16 +5,16 @@
 (ns datomish.sqlite
   #?(:cljs
      (:require-macros
-      [datomish.pair-chan :refer [go-pair <?]]
+      [datomish.pair-chan :refer [go-pair go-safely <?]]
       [cljs.core.async.macros :refer [go]]))
   #?(:clj
      (:require
-      [datomish.pair-chan :refer [go-pair <?]]
-      [clojure.core.async :refer [go <! >!]])
+      [datomish.pair-chan :refer [go-pair go-safely <?]]
+      [clojure.core.async :refer [go <! >! chan put! take! close!]])
      :cljs
      (:require
       [datomish.pair-chan]
-      [cljs.core.async :as a :refer [<! >!]])))
+      [cljs.core.async :as a :refer [<! >! chan put! take! close!]])))
 
 (defprotocol ISQLiteConnection
   (-execute!
@@ -49,11 +49,23 @@
 (defn reduce-rows
   [db [sql & bindings] initial f]
   (let [acc (atom initial)]
-    (go
-      (let [[_ err] (<! (-each db sql bindings #(swap! acc f %)))]
-        (if err
-          [nil err]
-          [@acc nil])))))
+    (go-pair
+      (<? (-each db sql bindings #(swap! acc f %)))
+      @acc)))
+
+(defn <?all-rows
+  "Takes a new channel, put!ing rows as [row err] pairs
+   into it as they arrive from storage. Closes the channel
+   when no more results exist. Consume with <?."
+  [db [sql & bindings :as rest] chan]
+  (go-safely [c chan]
+    (let [result (<! (-each db sql bindings (fn [row] (put! c [row nil]))))]
+      ;; We assume that a failure will result in the promise
+      ;; channel being rejected and no further row callbacks
+      ;; being called.
+      (when (second result)
+        (put! result c))
+      (close! c))))
 
 (defn all-rows
   [db [sql & bindings :as rest]]