 
               (when-not exists?
                 (inc! (:new-catalog metrics))
-                (add-catalog-metadata! hash api-version version transaction-uuid)
-                (let [refs-to-hashes (zipmap (keys resources) resource-hashes)]
+                (let [catalog-id     (:id (add-catalog-metadata! hash api-version version transaction-uuid))
+                      refs-to-hashes (zipmap (keys resources) resource-hashes)]
                   (time! (:add-resources metrics)
-                         (add-resources! hash resources refs-to-hashes))
+                         (add-resources! catalog-id resources refs-to-hashes))
                   (time! (:add-edges metrics)
-                         (add-edges! hash edges refs-to-hashes))))))
+                         (add-edges! catalog-id edges refs-to-hashes))))))
 
            hash)))
 

  1. Each key corresponds to a table, and each value is a list of rows
  2. The mapping of keys and values to table names and columns is done
     by `add-resources!`"
  [catalog-hash {:keys [type title exported parameters tags file line] :as resource} resource-hash persisted?]
  {:pre  [(every? string? #{catalog-hash type title})]
   :post [(= (set (keys %)) #{:resource :parameters :parameters_cache})]}
  (let [values {:resource         [[resource-hash type title (sutils/to-jdbc-varchar-array tags) exported file line catalog-hash]]
                :parameters       []
                :parameters_cache []}]



(defn add-resources!
  "Persist the given resource and associate it with the given catalog."
  [catalog-id refs-to-resources refs-to-hashes]
  (let [persisted?      (resources-exist? (utils/valset refs-to-hashes))
        resource-values (for [[ref resource] refs-to-resources
                              :let [hash (refs-to-hashes ref)]]
                          (resource->values catalog-id resource hash (persisted? hash)))
        lookup-table    [[:parameters_cache "INSERT INTO resource_params_cache (resource, parameters) VALUES (?, ?)"]
                         [:resource "INSERT INTO catalog_resources (catalog_id,resource,type,title,tags,exported,file,line) VALUES (?,?,?,?,?,?,?,?)"]
                         [:parameters "INSERT INTO resource_params (resource,name,value) VALUES (?,?,?)"]]]
    (sql/transaction
     (doseq [[lookup the-sql] lookup-table
