 
 (defn- filter-clause->predicate [{:keys [compound-type subclauses], :as clause}]
   (case compound-type
-    :and `(~'and ~@(map filter-clause->predicate subclauses))
-    :or  `(~'or  ~@(map filter-clause->predicate subclauses))
+    :and (apply kfns/pred-and (map filter-clause->predicate subclauses))
+    :or  (apply kfns/pred-or  (map filter-clause->predicate subclauses))
     nil  (filter-subclause->predicate clause)))
 
-(defmethod apply-form :filter [[_ clause]]
-  `(where ~(filter-clause->predicate clause)))
-
-
-(defmethod apply-form :join-tables [[_ join-tables]]
-  (vec (for [{:keys [table-name pk-field source-field]} join-tables]
-         `(join ~table-name
-                (~'= ~(keyword (format "%s.%s" (:name (:source-table (:query *query*))) (:field-name source-field)))
-                     ~(keyword (format "%s.%s" table-name                               (:field-name pk-field))))))))
-
-
-(defmethod apply-form :limit [[_ value]]
-  `(limit ~value))
-
-
-(defmethod apply-form :order-by [[_ subclauses]]
-  (vec (for [{:keys [field direction]} subclauses]
-         `(order ~(formatted field)
-                 ~(case direction
-                    :ascending  :ASC
-                    :descending :DESC)))))
-
-;; TODO - page can be preprocessed away -- converted to a :limit clause and an :offset clause
-;; implement this at some point.
-(defmethod apply-form :page [[_ {:keys [items page]}]]
-  {:pre [(integer? items)
-         (> items 0)
-         (integer? page)
-         (> page 0)]}
-  `[(limit ~items)
-    (offset ~(* items (- page 1)))])
-
-
-;; ## Debugging Functions (Internal)
-
+(defn- apply-filter [korma-query {clause :filter}]
+  (k/where korma-query (filter-clause->predicate clause)))
+
+(defn- apply-join-tables [korma-query {join-tables :join-tables, {source-table-name :name} :source-table}]
+  (loop [korma-query korma-query, [{:keys [table-name pk-field source-field]} & more] join-tables]
+    (let [korma-query (k/join korma-query table-name
+                              (= (keyword (format "%s.%s" source-table-name (:field-name source-field)))
+                                 (keyword (format "%s.%s" table-name        (:field-name pk-field)))))]
+      (if (seq more)
+        (recur korma-query more)
+        korma-query))))
+
+(defn- apply-limit [korma-query {value :limit}]
+  (k/limit korma-query value))
+
+(defn- apply-order-by [korma-query {subclauses :order-by}]
+  (loop [korma-query korma-query, [{:keys [field direction]} & more] subclauses]
+    (let [korma-query (k/order korma-query (formatted field) (case direction
+                                                               :ascending  :ASC
+                                                               :descending :DESC))]
+      (if (seq more)
+        (recur korma-query more)
+        korma-query))))
+
+(defn- apply-page [korma-query {{:keys [items page]} :page}]
+  (-> korma-query
+      (k/limit items)
+      (k/offset (* items (dec page)))))
+
+
+;;
 (defn- log-korma-form
   [korma-form]
   (when-not qp/*disable-qp-logging*
     (log/debug
-     (u/format-color 'green "\n\nKORMA FORM: 😏\n%s" (->> (nth korma-form 2)                                    ; korma form is wrapped in a let clause. Discard it
-                                                         (walk/prewalk (fn [form]                               ; strip korma.core/ qualifications from symbols in the form
-                                                                         (cond                                  ; to remove some of the clutter
-                                                                           (symbol? form)  (symbol (name form))
-                                                                           (keyword? form) (keyword (name form))
-                                                                           :else           form)))
-                                                         (u/pprint-to-str)))
-     (u/format-color 'blue  "\nSQL: 😈\n%s\n"        (-> (eval (let [[let-form binding-form & body] korma-form] ; wrap the (select ...) form in a sql-only clause
-                                                                `(~let-form ~binding-form                       ; has to go there to work correctly
-                                                                            (sql-only ~@body))))
-                                                        (s/replace #"\sFROM" "\nFROM")                          ; add newlines to the SQL to make it more readable
-                                                        (s/replace #"\sLEFT JOIN" "\nLEFT JOIN")
-                                                        (s/replace #"\sWHERE" "\nWHERE")
-                                                        (s/replace #"\sGROUP BY" "\nGROUP BY")
-                                                        (s/replace #"\sORDER BY" "\nORDER BY")
-                                                        (s/replace #"\sLIMIT" "\nLIMIT"))))))
+     (u/format-color 'blue "\nSQL: 😈\n%s\n" (-> (k/as-sql korma-form)
+                                                (s/replace #"\sFROM" "\nFROM")           ; add newlines to the SQL to make it more readable
+                                                (s/replace #"\sLEFT JOIN" "\nLEFT JOIN")
+                                                (s/replace #"\sWHERE" "\nWHERE")
+                                                (s/replace #"\sGROUP BY" "\nGROUP BY")
+                                                (s/replace #"\sORDER BY" "\nORDER BY")
+                                                (s/replace #"\sLIMIT" "\nLIMIT"))))))
+
+(defn process-structured
+  "Convert QUERY into a korma `select` form, execute it, and annotate the results."
+  [{{:keys [source-table] :as query} :query, driver :driver, database :database, :as outer-query}]
+  (binding [*query* outer-query]
+    (try
+      (let [entity      (korma-entity database source-table)
+            timezone    (driver/report-timezone)
+            korma-query (cond-> (k/select* entity)
+                          (:aggregation query) (apply-aggregation query)
+                          (:breakout    query) (apply-breakout    query)
+                          (:fields      query) (apply-fields      query)
+                          (:filter      query) (apply-filter      query)
+                          (:join-tables query) (apply-join-tables query)
+                          (:limit       query) (apply-limit       query)
+                          (:order-by    query) (apply-order-by    query)
+                          (:page        query) (apply-page        query))]
+
+        ;; log query
+        (when (config/config-bool :mb-db-logging)
+          (log-korma-form korma-query))
+
+        ;; execute query
+        (kdb/with-db (:db entity)
+          (if (and (seq timezone)
+                   (contains? (:features driver) :set-timezone))
+            (kdb/transaction
+              (k/exec-raw ((:timezone->set-timezone-sql driver) timezone))
+              (k/exec korma-query))
+            (k/exec korma-query))))
+
+      (catch java.sql.SQLException e
+        (let [^String message (or (->> (.getMessage e) ; error message comes back like "Error message ... [status-code]" sometimes
+                                       (re-find  #"(?s)(^.*)\s+\[[\d-]+\]$") ; status code isn't useful and makes unit tests hard to write so strip it off
+                                       second) ; (?s) = Pattern.DOTALL - tell regex `.` to match newline characters as well
+                                  (.getMessage e))]
+          (throw (Exception. message)))))))
+
+(defn process-and-run
+  "Process and run a query and return results."
+  [{:keys [type] :as query}]
+  (case (keyword type)
+    :native (native/process-and-run query)
+    :query  (process-structured query)))

     (formatted this false))
    ([{value :value, {unit :unit} :field} _]
     ;; prevent Clojure from converting this to #inst literal, which is a util.date
     ((:date (:driver *query*)) unit `(Timestamp/valueOf ~(.toString value)))))

  RelativeDateTimeValue
  (formatted
    ([this]
     (formatted this false))
    ([{:keys [amount unit], {field-unit :unit} :field} _]
     (let [driver (:driver *query*)]
       ((:date driver) field-unit (if (zero? amount)
                                     (sqlfn :NOW)
                                     ((:date-interval driver) unit amount)))))))


(defmethod apply-form :aggregation [[_ {:keys [aggregation-type field]}]]
  (if-not field
    ;; aggregation clauses w/o a Field
    (case aggregation-type
      :rows  nil                               ; don't need to do anything special for `rows` - `select` selects all rows by default
      :count `(aggregate (~'count :*) :count))
    ;; aggregation clauses with a Field
    (let [field (formatted field)]
      (case aggregation-type
        :avg      `(aggregate (~'avg ~field) :avg)
        :count    `(aggregate (~'count ~field) :count)
        :distinct `(aggregate (~'count (sqlfn :DISTINCT ~field)) :count)
        :stddev   `(fields [(sqlfn :stddev ~field) :stddev])
        :sum      `(aggregate (~'sum ~field) :sum)))))


(defmethod apply-form :breakout [[_ fields]]
  `[ ;; Group by all the breakout fields
    (group  ~@(map formatted fields))

    ;; Add fields form only for fields that weren't specified in :fields clause -- we don't want to include it twice, or korma will barf
    (fields ~@(->> fields
                   (filter (partial (complement contains?) (set (:fields (:query *query*)))))
                   (map (u/rpartial formatted :include-as))))])


(defmethod apply-form :fields [[_ fields]]
  `(fields ~@(map (u/rpartial formatted :include-as) fields)))


(defn- filter-subclause->predicate
  "Given a filter SUBCLAUSE, return a Korma filter predicate form for use in korma `where`."

  (:require [clojure.core.match :refer [match]]
            [clojure.tools.logging :as log]
            [clojure.string :as s]
            (korma [core :as k]
                   [db :as kdb])
            (korma.sql [fns :as kfns]
                       [utils :as utils])
            [metabase.config :as config]
            [metabase.driver :as driver]
            [metabase.driver.query-processor :as qp]
