             [clojure.set :as set]
             [clojure.string :as s]
             [hugsql.core :as hugsql])
-  (:import [java.sql Timestamp]))
+  (:import [akvo.lumen.import.common Geoshape Geopoint]
+           [java.sql Timestamp]))
 
 (hugsql/def-db-fns "akvo/lumen/transformation.sql")
 (hugsql/def-db-fns "akvo/lumen/transformation/engine.sql")

    (add-column conn {:table-name table-name
                      :new-column-name (get column "columnName")
                      :column-type (condp = (get column "type")
                                     "text" "text"
                                     "number" "double precision"
                                     "date" "timestamptz")})))

(defn insert-merged-data
  "Insert the merged values into the target dataset"

          (or aggregationColumn mergeColumn)
          aggregationDirection))

#_(defn to-sql-types [row merge-columns]
    ;; For now we only need to change date types.
    (let [date-columns (filter #(= "date" (get % "type")) merge-columns)]
      (reduce (fn [result-row {:strs [columnName]}]
                (update result-row columnName #(when % (Timestamp. %))))
              row
              date-columns)))

(defn to-sql-types [row merge-columns]
  (let [columns-requiring-cast (filter #(contains? #{"date" "geoshape" "geopoint" }
                                                   (get % "type"))
                                       merge-columns)]
    (reduce (fn [result-row {:strs [columnName type]}]
              (update result-row columnName (fn [v]
                                              (when v (case type
                                                        "date" (Timestamp. v)
                                                        "geoshape" (Geoshape. v)
                                                        "geopoint" (Geopoint. v))))))
            row
            columns-requiring-cast)))

(defn fetch-data
  "Fetch data from the source dataset and returns a map with the shape
