               :words
               [[:id :int "PRIMARY KEY"]
                [:word "VARCHAR(32)"]])))
-    (doseq [word words]
-      (jdbc/insert! cpool :words word))))
+    (doseq [word all-words]
+      (jdbc/execute! cpool (mysql/upsert :words
+                                         (dissoc word :id)
+                                         (dissoc word :word))))))
 
 (defn transform-word
   [word]

                     :password password})]
    (with-test-env [test-env [4 env-config peer-config]]
      (ensure-database! username password subname db-name)
      (pipe (spool words) in true)
      (onyx.test-helper/validate-enough-peers! test-env job)
      (->> (:job-id (onyx.api/submit-job peer-config job))
           (onyx.api/await-job-completion peer-config))


(deftest sql-update-output-test
  (let [{:keys [env-config peer-config mysql-config]} (read-config
                                                       (io/resource "config.edn")
                                                       {:profile :test})
        {:keys [sql/username sql/password sql/subname sql/db-name]} mysql-config
        job (build-job username password subname db-name 10 1000)
        {:keys [in]} (get-core-async-channels job)
