       (let [spec (:spec (data-source-spec-by-job-execution-id conn {:job-execution-id job-execution-id}))]
         (with-open [importer (import/dataset-importer (get spec "source") config)]
           (let [columns (import/columns importer)
-                records (import/records importer)]
+                records (import/records importer)
+                stores (apply assoc {} (interleave columns (repeatedly #(atom {}))))
+                rows-count (atom 0)]
             (import/create-dataset-table conn table-name columns)
             (import/add-key-constraints conn table-name columns)
-            (doseq [record (map (comp import/coerce-to-sql (partial l :raw-row)) records)]
-              (l :coerced record)
+            (doseq [record (map (comp import/coerce-to-sql (partial splitable stores columns)) records)]
+              (swap! rows-count inc)
               (jdbc/insert! conn table-name record))
-            (successful-import conn job-execution-id table-name columns spec claims data-source))))
+            (successful-import conn job-execution-id table-name
+                               (map #(assoc % :splitable
+                                            (map (fn [[k v]] [k (assoc v :rows @rows-count)]) @(last %2)))
+                                    columns stores)
+                               spec claims data-source))))
       (catch Throwable e
         (failed-import conn job-execution-id (.getMessage e) table-name)
         (log/error e)

                                  :table-name table-name
                                  :imported-table-name imported-table-name
                                  :version 1
                                  :columns (mapv (fn [{:keys [title id type key multiple-type multiple-id]}]
                                                   {:columnName (name id)
                                                    :direction nil
                                                    :hidden false

  org.postgis.Point
  (sql-value [v] (val->geometry-pgobj v)))

(defn l [t o]
  (log/error t o)
  o)

(defn fun*fun [c v]
  (let [regex "[^a-zA-Z0-9\\s]"
        freqs (frequencies (re-seq (re-pattern regex) v))]    
    (reset! c
            (reduce (fn [c2 [k v2]]
                      (assoc c2 k
                             {:max-coincidences-in-one-row (max v2 (get-in c2 [k :max-coincidences-in-one-row] 0))
                              :total-row-coincidences (inc (get-in c2 [k :total-row-coincidences] 0))
                              :total-coincidences (+ v2 (get-in c2 [k :total-coincidences] 0))}))
                    @c freqs))))

(defn splitable [stores columns record]
  (log/debug :splitable stores columns)
  (reduce-kv
   (fn [result k v]
     (let [column (first (filter #(= k (:id %)) columns))]
       (when (= :text (:type column))
         (let [store (get stores column)]
           (fun*fun store v)))))
   {}
   record)
  record)

(defn do-import
  "Import runs within a future and since this is not taking part of ring
