Closes #2 - initial implementation of direct cypher queries

Author: bbenzikry

.gitignore

@@ -13,3 +13,4 @@ pom.xml.asc
 .nrepl-port
 .cpcache/
 .calva
+.metals

deps.edn

@@ -0,0 +1,3 @@
+{:deps {bbenzikry/neo4clj {:git/url "https://github.com/bbenzikry/neo4clj" :sha "f7fa0a27a56797e646458de6a4efeffb3381294d"}
+        neo4j/neo4j-bi-jdbc {:mvn/version "1.0.0"}
+        org.neo4j/neo4j-cypher {:mvn/version "4.1.1"}}}

project.clj

@@ -1,19 +1,16 @@
-(defproject metabase/neo4j-driver "0.0.2-SNAPSHOT-neo4j-connector-1.0.0"
+(defproject metabase/neo4j-driver "0.0.3-SNAPSHOT-neo4j-connector-1.0.0"
   :min-lein-version "2.5.0"
-
-  ; :repositories [["bintray" "https://dl.bintray.com/meetr/thirdparty"]]
-
-  :dependencies
-  [[neo4j/neo4j-bi-jdbc "1.0.0"]]
-
+  ; git repo support
+  :plugins [[lein-tools-deps "0.4.5"]]
+  :middleware [lein-tools-deps.plugin/resolve-dependencies-with-deps-edn]
+  :lein-tools-deps/config {:config-files [:install :user :project]}
   :aliases
   {"test"       ["with-profile" "+unit_tests" "test"]}
 
   :profiles
   {:provided
    {:dependencies
     [[metabase-core "1.0.0-SNAPSHOT"]]}
-
    :uberjar
    {:auto-clean    true
     :aot           :all

src/metabase/driver/neo4j.clj

@@ -1,7 +1,10 @@
 (ns metabase.driver.neo4j
   "Metabase Neo4j Driver."
+  (:import  [org.neo4j.cypher.internal.parser CypherParser]
+            [org.neo4j.cypher.internal.util OpenCypherExceptionFactory])
   (:require [clojure
              [set :as set]]
+            [clojure.tools.logging :as log]
             [honeysql.core :as hsql]
             [metabase.util
              [honeysql-extensions :as hx]]
@@ -20,6 +23,7 @@
 (defn- make-subname [host port db jdbc-flags]
   (str "//" host ":" port "/" db jdbc-flags))
 
+; Support multiple :classname
 (defn neo4j
   "Create a Clojure JDBC database specification for Neo4j."
   [{:keys [host port db jdbc-flags]
@@ -137,6 +141,21 @@
   [driver database table]
   (sql-jdbc.sync/describe-table driver database table))
 
+
+
+;  |                                          Execution + Cypher support                                            |
+;  | TODO: Connection pooling                                                                                       |
+;  +----------------------------------------------------------------------------------------------------------------+
+
+
+(def ^:private cypher-parser (CypherParser.))
+(def ^:private cypher-exception-factory (OpenCypherExceptionFactory. nil))
+(defn- cypher? [{{query :query} :native}]
+  (try
+    (log/info "Received neo4j query. Checking if cypher ❓")
+    (.parse cypher-parser query cypher-exception-factory nil)
+    true (catch Throwable ex (log/info ex) false)))
+
 (defmethod driver/execute-reducible-query :neo4j
   [driver query chans respond]
-  (neo4j.execute/execute-reducible-query driver query chans respond))
+  (if (cypher? query) (neo4j.execute/execute-reducible-query->cypher driver query chans respond) (neo4j.execute/execute-reducible-query driver query chans respond)))

src/metabase/driver/neo4j/execute.clj

@@ -1,16 +1,26 @@
 (ns metabase.driver.neo4j.execute
+  (:import [org.neo4j.driver
+            Driver])
   (:require
+   [metabase.driver.neo4j.util
+    :refer [with-neo-connection]]
+   [neo4clj.client :as neo4j]
+   [clojure.core.async :as a]
+   [clojure.tools.logging :as log]
    [metabase.driver.sql-jdbc
     [execute :as sql-jdbc.execute]]
    [metabase.mbql.util :as mbql.u]
    [metabase.query-processor
     [context :as context]
     [interface :as qp.i]
     [store :as qp.store]
+    [reducible :as qp.reducible]
     [timezone :as qp.timezone]]))
 
 
 ; We want to do this in order to avoid remarks and commenting which affect the simba JDBC driver
+
+
 (defn execute-reducible-query
   "Implementation of `execute-reducible-query` for neo4j bi connector driver
    Copied as is from sql-jdbc/execute"
@@ -25,4 +35,34 @@
                 rs   (sql-jdbc.execute/execute-query! driver stmt)]
       (let [rsmeta           (.getMetaData rs)
             results-metadata {:cols (sql-jdbc.execute/column-metadata driver rsmeta)}]
-        (respond results-metadata (sql-jdbc.execute/reducible-rows driver rs rsmeta (context/canceled-chan context)))))))
+        (respond results-metadata (sql-jdbc.execute/reducible-rows driver rs rsmeta (context/canceled-chan context)))))))
+
+; TODO add with-neo4j-connection macro
+(defn get-neo-connection
+  [{host :host port :port user :user password :password dbname :dbname}]
+  (let [base (str "bolt://" host ":" port)
+        url (if dbname (str base "/" dbname) base)]
+    (if password (neo4j/connect url user password) (neo4j/connect url user))))
+
+(defn get-cypher-columns [result]
+  (if (seq? result)
+    {:cols (into [] (map #(assoc {} :name %) (keys (first (take 1 result)))))}
+    {:cols [{:name "result"}]}))
+
+(defn execute-reducible-query->cypher
+  "Process and run a native cypher query."
+  [_ {{query :query} :native} context respond]
+  (log/info "Executing reducible query for cypher")
+  (with-neo-connection [^Driver connection (:details (qp.store/database))]
+    (let [results (volatile! (neo4j/execute! connection query))
+          nonseq-val (volatile! false)
+          columns (get-cypher-columns @results)
+          row-thunk #(if-not (seq? @results) ((if-not @nonseq-val (vreset! nonseq-val true) @results) nil)
+                             (let [old @results]
+                               (vswap! results (fn [state] (drop 1 state)))
+                               (vals (first (take 1 old)))))]
+    ; handle cancellation
+      (a/go
+        (when (a/<! (context/canceled-chan context))
+          (neo4j/disconnect connection)))
+      (respond columns (qp.reducible/reducible-rows row-thunk (context/canceled-chan context))))))

src/metabase/driver/neo4j/util.clj

@@ -0,0 +1,45 @@
+(ns metabase.driver.neo4j.util
+  (:import [org.neo4j.driver
+            Driver])
+  (:require
+   [neo4clj.client :as neo4j]
+   [clojure.tools.logging :as log]
+   [metabase.util
+    [i18n :refer [trs]]]
+   [metabase
+    [util :as u]]))
+
+(def ^:dynamic ^Driver *neo-connection*
+  "Connection to a Neo4j database. Bound by top-level `with-neo-connection` so it may be reused within its body."
+  nil)
+
+(defn get-neo-connection
+  [{host :host port :port user :user password :password dbname :dbname}]
+  (let [base (str "bolt://" host ":" port)
+        url (if dbname (str base "/" dbname) base)]
+    (if password (neo4j/connect url user password) (neo4j/connect url user))))
+
+(defn -with-neo-connection
+  "Run `f` with a new connection (bound to `*neo-connection*`) with `details`. Don't use this directly; use
+  `with-neo-connection`."
+  [f details]
+  (let [connection (get-neo-connection details)]
+    (log/debug (u/format-color 'cyan (trs "Opened new Neo4j connection.")))
+    (try
+      (binding [*neo-connection* connection]
+        (f *neo-connection*))
+      (finally
+        (neo4j/disconnect connection)
+        (log/debug (u/format-color 'cyan (trs "Closed Neo4j connection.")))))))
+
+; TODO: use steady connection pool with idle timeout instead of one-off use
+(defmacro with-neo-connection
+  "Open a new neo4j connection based on DB ``details`, bind connection to `binding`, execute `body`, and
+  close the connection. The DB connection is re-used by subsequent calls to `with-neo-connection` within
+  `body`."
+  [[binding details] & body]
+  `(let [f# (fn [~binding]
+              ~@body)]
+     (if *neo-connection*
+       (f# *neo-connection*)
+       (-with-neo-connection f# ~details))))