State Monad

api.clj

(ns revelytix.spyder.query.interface
  (:require [revelytix.sparql.parse          :as sparqlp]
            [revelytix.spyder.query.plan     :as plan]
            [revelytix.spyder.query.optimize :as opt]
            [revelytix.spyder.query.process  :as process]
            [revelytix.spyder.log            :as log])
  (:import [revelytix.spyder.query.api QueryException]
           [revelytix.spyder.sql.plan_tree QueryContext]))

(defn always [x]
  (fn [st] [x st]))

(defn bind [m f]
  (fn [st]
    (let [[value st2] (m st)]
      ((f value) st2))))

(defn nxt [m n]
  (bind m (fn [_] n)))

(defmacro let->> [[& bindings] & body]
  (let [[bind-form m] (take 2 bindings)]
    (if (= 2 (count bindings))
      `(bind ~m (fn [~bind-form] ~@body))
      `(bind ~m (fn [~bind-form] (let->> ~(drop 2 bindings) ~@body))))))

(defmacro >>
  ([m] m)
  ([m n] `(nxt ~m ~n))
  ([m n & ms] `(nxt ~m (>> ~n ~@ms))))

(defn update [f]
  (fn [st]
    [st (f st)]))

(defn set [st]
  (update (fn [_] st)))

(defn fetch []
  (update identity))

(defn env-assoc [key value]
  (update (fn [st] (assoc st key value))))

(defn env-assoc-in [ks value]
  (update (fn [st] (assoc-in st ks value))))

(defn env-fetch [key]
  (fn [st]
    [(get st key) st]))

(defn runS [m init]
  (m init))

(defmacro defstateful [name args & body]
  `(defn ~name ~args
     (>> ~@body)))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(def *debug* false)

(defmacro log [& exprs]
  `(always (log/log ~@exprs)))

(defn to-sparql-string [x]
  (if (instance? QueryContext x)
    (:sparql x)
    x))

(defn try+catch [action catchf]
  (fn [st]
    (try
      (runS action st)
      (catch Throwable t
        (catchf t)))))

(defstateful query-context [action]
  (let->> [sparql (env-fetch :sparql)
           sql    (env-fetch :sql)]
    (try+catch
     action
     (fn [t]
       (throw (QueryException. "Error processing query" (to-sparql-string sparql) sql t))))))

(defstateful parse-query [sparql]
  (env-assoc :sparql sparql)
  (query-context (always (sparqlp/parse sparql))))

(defstateful record-timing [key action]
  (let->> [start (always (System/currentTimeMillis))
           result action
           end (always (System/currentTimeMillis))]
    (env-assoc-in [:timings key :start] start)
    (env-assoc-in [:timings key :end]   end)
    (always result)))

(defstateful plan-query [sparql]
  (let->> [parse-tree (parse-query sparql)]
    (record-timing :start-plan (query-context (plan/plan parse-tree)))))

(defstateful optimize-query [sparql]
  (let->> [plan-tree (plan-query sparql)]
    (record-timing :start-optimize (query-context (opt/optimize plan-tree)))))

(defstateful map-query* [throw? sparql]
  (let->> [optimized-plan (optimize-query sparql)]
    (record-timing :start-sql-gen (query-context (if throw?
                                                   (process/translate optimized-plan)
                                                   (process/translate-no-throw optimized-plan))))))

(defstateful map-query [sparql]
  (map-query* true sparql))

(defstateful map-query-no-throw [sparql]
  (map-query* false sparql))

(defstateful process-query* [throw? timeout sparql]
  (let->> [mapped (map-query* throw? sparql)]
    (log "process-query sparql:" sparql)
    (record-timing :start-processing (query-context (if throw?
                                                      (process/process mapped timeout)
                                                      (process/process-no-throw mapped timeout))))))

(defstateful process-query [timeout sparql]
  (process-query* true timeout sparql))

(defstateful process-query-no-throw [timeout sparql]
  (process-query* false timeout sparql))

st-m.clj

;; the State Monad in Clojure a State Monad is a fn that takes a state
;; and returns a 2-tuple of a computed value, plus the (possibly new) state

;; start with the 2 functions that define a monad
(defn always [x]
  (fn [st] [x st]))

(defn bind [m f]
  (fn [st]
    (let [[value st2] (m st)]
      ((f value) st2))))

(defn nxt [m n]
  (bind m (fn [_] n)))

(defmacro let->> [[& bindings] & body]
  (let [[bind-form m] (take 2 bindings)]
    (if (= 2 (count bindings))
      `(bind ~m (fn [~bind-form] ~@body))
      `(bind ~m (fn [~bind-form] (let->> ~(drop 2 bindings) ~@body))))))

(defmacro >>
  ([m] m)
  ([m n] `(nxt ~m ~n))
  ([m n & ms] `(nxt ~m (>> ~n ~@ms))))

;; these are "special" in the sense that they operate on the internals
;; of the monad, together they give you access to everything you could
;; want to manipulate the state being carried around
(defn update [f]
  (fn [st]
    [st (f st)]))

(defn set [st]
  (update (fn [_] st)))

(def fetch []
  (update identity))

;; to run the monad, you must provide it's initial state
(defn runS [m init]
  (m init))

This is my rewritten api.clj. It compiles, but it's not in a completely working state yet, since it relies on some underlying functions to be rewritten in the State Monad style, plan/plan, opt/optimize, process/translate, and process/process. All of those functions lost the parameters that the engine provided, sql-parser, jdbc-metadata, mapping, config, taskpool, pipe-factory, which are now on the global state. They are individually fetchable via env-fetch from any state monad. Plus, before calling into one of these api functions, you need to be sure that you've properly initialized the state of the monad. Currently we're accomplishing this with partial functions wrapped up in engine_core. process-sparql-direct, partial-process-query, process-with-taskpool, and their ilk. Those functions would need to be rewritten to perform the equivalent initialization of the monad's state.