mirror of
https://github.com/penpot/penpot.git
synced 2026-08-09 14:28:55 +00:00
⚡ Load graph ingest via Ladybug COPY bulk import
This commit is contained in:
Alejandro Alonso
parent
78109b0054
commit
76a0d9ef5e
131
backend/src/app/graph/bulk.clj
Normal file
131
backend/src/app/graph/bulk.clj
Normal file
@ -0,0 +1,131 @@
|
||||
;; This Source Code Form is subject to the terms of the Mozilla Public
|
||||
;; License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
;;
|
||||
;; Copyright (c) KALEIDOS INC Sucursal en España SL
|
||||
|
||||
(ns app.graph.bulk
|
||||
"Bulk Ladybug ingest via COPY FROM CSV.
|
||||
|
||||
Node and relationship rows are written to a temporary staging directory
|
||||
and loaded with one COPY statement per table (or per rel FROM/TO pair)."
|
||||
(:require
|
||||
[app.graph.ladybug :as ladybug]
|
||||
[clojure.java.io :as io]
|
||||
[clojure.string :as str]
|
||||
[datoteka.fs :as fs])
|
||||
(:import
|
||||
java.io.File))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
(def ^:private copy-csv-options
|
||||
"Ladybug COPY CSV options. QUOTE must be set explicitly or commas in
|
||||
string fields are treated as column separators."
|
||||
"HEADER=true, DELIM=',', QUOTE='\"'")
|
||||
|
||||
(defn- csv-escape-string
|
||||
[s]
|
||||
(str "\"" (str/replace (str s) "\"" "\"\"") "\""))
|
||||
|
||||
(defn- csv-cell
|
||||
[v]
|
||||
(cond
|
||||
(uuid? v) (str v)
|
||||
(string? v) (csv-escape-string v)
|
||||
(number? v) (str (long v))
|
||||
:else (csv-escape-string (str v))))
|
||||
|
||||
(defn- cypher-file-path
|
||||
[^File file]
|
||||
(-> (.getAbsolutePath file)
|
||||
(str/replace "\\" "\\\\")
|
||||
(str/replace "'" "\\'")))
|
||||
|
||||
(defn- node-columns
|
||||
[rows]
|
||||
(let [cols (into #{} (mapcat keys rows))
|
||||
preferred [:id :name :version :revision :index]]
|
||||
(into (vec (filter cols preferred))
|
||||
(sort (remove (set preferred) cols)))))
|
||||
|
||||
(defn- write-node-csv!
|
||||
[^File file rows]
|
||||
(let [columns (node-columns rows)]
|
||||
(with-open [w (io/writer file :encoding "UTF-8")]
|
||||
(.write w (str (str/join "," (map name columns)) "\n"))
|
||||
(doseq [row rows]
|
||||
(.write w (str (str/join "," (map #(csv-cell (get row %)) columns))
|
||||
"\n"))))))
|
||||
|
||||
(defn- write-edge-csv!
|
||||
[^File file edges]
|
||||
(with-open [w (io/writer file :encoding "UTF-8")]
|
||||
(.write w "from,to,position\n")
|
||||
(doseq [{:keys [from-id to-id position]} edges]
|
||||
(.write w (str (csv-cell from-id) ","
|
||||
(csv-cell to-id) ","
|
||||
(csv-cell position) "\n")))))
|
||||
|
||||
(defn- delete-tree!
|
||||
[path]
|
||||
(when (fs/exists? path)
|
||||
(doseq [f (reverse (file-seq (io/file path)))]
|
||||
(.delete ^File f))))
|
||||
|
||||
(defn staging-dir
|
||||
"Directory for temporary COPY CSV files."
|
||||
[db-path file-id]
|
||||
(if (= db-path ":memory:")
|
||||
(str (fs/path (System/getProperty "java.io.tmpdir")
|
||||
"penpot-graph-bulk"
|
||||
(str file-id)))
|
||||
(str (fs/path (str db-path ".bulk") (str file-id)))))
|
||||
|
||||
(defn- copy-node-table!
|
||||
[conn table ^File csv-file]
|
||||
(let [statement (str "COPY `" table "` FROM '" (cypher-file-path csv-file)
|
||||
"' (" copy-csv-options ");")]
|
||||
(try
|
||||
(ladybug/exec-on-connection! conn [statement])
|
||||
(catch clojure.lang.ExceptionInfo e
|
||||
(throw (ex-info (str "COPY node table failed: " table)
|
||||
(merge (ex-data e)
|
||||
{:table table
|
||||
:csv-file (.getAbsolutePath csv-file)})
|
||||
e))))))
|
||||
|
||||
(defn- copy-edge-group!
|
||||
[conn from-table to-table ^File csv-file]
|
||||
(let [statement (str "COPY `IsChildOf` FROM '" (cypher-file-path csv-file) "' "
|
||||
"(from='" from-table "', to='" to-table "', "
|
||||
copy-csv-options ");")]
|
||||
(try
|
||||
(ladybug/exec-on-connection! conn [statement])
|
||||
(catch clojure.lang.ExceptionInfo e
|
||||
(throw (ex-info (str "COPY edge group failed: " from-table " -> " to-table)
|
||||
(merge (ex-data e)
|
||||
{:from-table from-table
|
||||
:to-table to-table
|
||||
:csv-file (.getAbsolutePath csv-file)})
|
||||
e))))))
|
||||
|
||||
(defn load-projection!
|
||||
"Load projected nodes and edges into an open Ladybug connection."
|
||||
[conn {:keys [nodes edges]} staging-path]
|
||||
(fs/create-dir staging-path)
|
||||
(try
|
||||
(doseq [[table rows] (sort-by key nodes)
|
||||
:when (seq rows)]
|
||||
(let [csv-file (io/file staging-path (str table ".csv"))]
|
||||
(write-node-csv! csv-file rows)
|
||||
(copy-node-table! conn table csv-file)))
|
||||
(doseq [[[from-table to-table] group]
|
||||
(sort-by identity (group-by (juxt :from-table :to-table) edges))
|
||||
:when (seq group)]
|
||||
(let [csv-file (io/file staging-path
|
||||
(str "IsChildOf_" from-table "_" to-table ".csv"))]
|
||||
(write-edge-csv! csv-file group)
|
||||
(copy-edge-group! conn from-table to-table csv-file)))
|
||||
(finally
|
||||
(delete-tree! staging-path))))
|
||||
@ -12,6 +12,7 @@
|
||||
[app.common.logging :as l]
|
||||
[app.common.types.file :as ctf]
|
||||
[app.db :as db]
|
||||
[app.graph.bulk :as bulk]
|
||||
[app.graph.ladybug :as ladybug]
|
||||
[app.graph.project.document :as project.document]
|
||||
[app.graph.project.transforms :as project.transforms]
|
||||
@ -20,7 +21,7 @@
|
||||
[app.srepl.helpers :as h]))
|
||||
|
||||
(defn ingest-file!
|
||||
[system file-id & {:keys [db-path reset-db?]
|
||||
[system file-id & {:keys [db-path reset-db? skip-stats?]
|
||||
:or {reset-db? true}}]
|
||||
(let [file-id (h/parse-uuid file-id)
|
||||
file (db/run! system #(bfc/get-file % file-id :realize? true))
|
||||
@ -44,12 +45,14 @@
|
||||
:schema schema/schema-version)
|
||||
(let [data (:data file)
|
||||
ddl (schema/ddl-statements)
|
||||
{:keys [statements stats]}
|
||||
(project.document/projection-statements data file)
|
||||
ingest-statements (conj (into ddl statements) "CHECKPOINT;")]
|
||||
{:keys [nodes edges stats]}
|
||||
(project.document/projection-data data file)
|
||||
staging-path (bulk/staging-dir db-path file-id)]
|
||||
(ladybug/with-connection! db-path
|
||||
(fn [conn]
|
||||
(ladybug/exec-on-connection! conn ingest-statements)
|
||||
(ladybug/exec-on-connection! conn ddl)
|
||||
(bulk/load-projection! conn {:nodes nodes :edges edges} staging-path)
|
||||
(ladybug/exec-on-connection! conn ["CHECKPOINT;"])
|
||||
{:file-id file-id
|
||||
:revn (:revn file)
|
||||
:name (or (:name data) (:name file))
|
||||
@ -57,4 +60,5 @@
|
||||
:schema-version schema/schema-version
|
||||
:projection {:stats stats}
|
||||
:transforms (project.transforms/apply-transforms! system db-path data file)
|
||||
:stats (stats/summarize-connection conn)})))))
|
||||
:stats (when-not skip-stats?
|
||||
(stats/summarize-connection conn))})))))
|
||||
|
||||
@ -85,11 +85,12 @@
|
||||
(defn- check-success!
|
||||
[^QueryResult result statement]
|
||||
(when-not (.isSuccess result)
|
||||
(ex/raise :type :internal
|
||||
:code :ladybug-query-failed
|
||||
:hint "Ladybug query failed"
|
||||
:statement statement
|
||||
:err (.getErrorMessage result))))
|
||||
(let [err (.getErrorMessage result)]
|
||||
(ex/raise :type :internal
|
||||
:code :ladybug-query-failed
|
||||
:hint (str "Ladybug query failed: " err)
|
||||
:statement statement
|
||||
:err err))))
|
||||
|
||||
(def ^:private default-query-timeout-seconds 120)
|
||||
|
||||
|
||||
@ -12,10 +12,8 @@
|
||||
(:require
|
||||
[app.common.logging :as l]
|
||||
[app.common.uuid :as uuid]
|
||||
[app.graph.ladybug :as ladybug]
|
||||
[app.graph.project.specs :as specs]
|
||||
[app.graph.schema :as graph.schema]
|
||||
[clojure.string :as str]))
|
||||
[app.graph.schema :as graph.schema]))
|
||||
|
||||
(def root-frame-id
|
||||
uuid/zero)
|
||||
@ -31,24 +29,6 @@
|
||||
:image "Image"
|
||||
:svg-raw "SVGRaw"})
|
||||
|
||||
(defn- create-node-statement
|
||||
[table attrs]
|
||||
(let [props (str/join ", "
|
||||
(map (fn [[k v]]
|
||||
(str "`" (name k) "`: " (ladybug/format-value v)))
|
||||
attrs))]
|
||||
(str "CREATE (n:`" table "` {" props "});")))
|
||||
|
||||
(defn- validated-node-statement
|
||||
[check-fn table attrs]
|
||||
(create-node-statement table (check-fn attrs)))
|
||||
|
||||
(defn- merge-edge-statement
|
||||
[from-table from-id to-table to-id position]
|
||||
(str "MATCH (c:`" from-table "` {`id`: " (ladybug/format-uuid from-id) "}), "
|
||||
"(p:`" to-table "` {`id`: " (ladybug/format-uuid to-id) "}) "
|
||||
"MERGE (c)-[:`IsChildOf` {`position`: " (ladybug/format-int position) "}]->(p);"))
|
||||
|
||||
(defn- document-attrs
|
||||
[file data]
|
||||
{:id (or (:id data) (:id file))
|
||||
@ -68,8 +48,8 @@
|
||||
|
||||
(defn- shape-node-attrs
|
||||
[shape]
|
||||
{:id (:id shape)
|
||||
:name (:name shape)})
|
||||
(specs/check-shape-node {:id (:id shape)
|
||||
:name (:name shape)}))
|
||||
|
||||
(defn- container-table?
|
||||
[table]
|
||||
@ -81,76 +61,87 @@
|
||||
(when-let [shapes (:shapes parent)]
|
||||
(vec (reverse shapes))))
|
||||
|
||||
(defn- initial-acc
|
||||
[]
|
||||
{:nodes {}
|
||||
:edges []
|
||||
:stats {:documents 0 :pages 0 :shapes 0}})
|
||||
|
||||
(declare project-shape-ids)
|
||||
|
||||
(defn- project-shape
|
||||
"Project one shape node and recurse into its children."
|
||||
[objects statements stats table shape parent-table parent-id position]
|
||||
(let [shape-id (:id shape)
|
||||
statements' (conj statements
|
||||
(validated-node-statement specs/check-shape-node table
|
||||
(shape-node-attrs shape))
|
||||
(merge-edge-statement table shape-id
|
||||
parent-table parent-id position))
|
||||
stats' (update stats :shapes inc)]
|
||||
[objects acc table shape parent-table parent-id position]
|
||||
(let [shape-id (:id shape)
|
||||
acc' (-> acc
|
||||
(update-in [:nodes table] (fnil conj []) (shape-node-attrs shape))
|
||||
(update :edges conj {:from-table table
|
||||
:from-id shape-id
|
||||
:to-table parent-table
|
||||
:to-id parent-id
|
||||
:position position})
|
||||
(update-in [:stats :shapes] inc))]
|
||||
(if-let [child-ids (when (container-table? table)
|
||||
(child-shape-ids shape))]
|
||||
(project-shape-ids objects statements' stats' table shape-id child-ids)
|
||||
[statements' stats'])))
|
||||
(project-shape-ids objects acc' table shape-id child-ids)
|
||||
acc')))
|
||||
|
||||
(defn- project-shape-ids
|
||||
[objects statements stats parent-table parent-id child-ids]
|
||||
[objects acc parent-table parent-id child-ids]
|
||||
(reduce
|
||||
(fn [[stmts st] [position shape-id]]
|
||||
(fn [acc [position shape-id]]
|
||||
(if-let [shape (get objects shape-id)]
|
||||
(if-let [table (shape-table shape)]
|
||||
(project-shape objects stmts st table shape parent-table parent-id position)
|
||||
(project-shape objects acc table shape parent-table parent-id position)
|
||||
(do
|
||||
(l/wrn :hint "unsupported shape type for graph slice"
|
||||
:shape-id (str shape-id)
|
||||
:type (:type shape))
|
||||
[stmts st]))
|
||||
acc))
|
||||
(do
|
||||
(l/wrn :hint "missing shape in page objects"
|
||||
:shape-id (str shape-id))
|
||||
[stmts st])))
|
||||
[statements stats]
|
||||
acc)))
|
||||
acc
|
||||
(map-indexed vector child-ids)))
|
||||
|
||||
(defn- project-page
|
||||
[statements stats doc-id page position]
|
||||
[acc doc-id page position]
|
||||
(let [page-id (:id page)
|
||||
objects (:objects page)
|
||||
root (get objects root-frame-id)
|
||||
statements' (conj statements
|
||||
(validated-node-statement specs/check-page "Page"
|
||||
(page-attrs page position))
|
||||
(merge-edge-statement "Page" page-id "Document" doc-id position))
|
||||
stats' (update stats :pages inc)]
|
||||
page-node (specs/check-page (page-attrs page position))
|
||||
acc' (-> acc
|
||||
(update-in [:nodes "Page"] (fnil conj []) page-node)
|
||||
(update :edges conj {:from-table "Page"
|
||||
:from-id page-id
|
||||
:to-table "Document"
|
||||
:to-id doc-id
|
||||
:position position})
|
||||
(update-in [:stats :pages] inc))]
|
||||
(if-let [top-level-ids (child-shape-ids root)]
|
||||
(project-shape-ids objects statements' stats' "Page" page-id top-level-ids)
|
||||
[statements' stats'])))
|
||||
(project-shape-ids objects acc' "Page" page-id top-level-ids)
|
||||
acc')))
|
||||
|
||||
(defn projection-statements
|
||||
"Build Cypher statements for projecting `data` into Ladybug.
|
||||
(defn projection-data
|
||||
"Build node/edge rows for projecting `data` into Ladybug.
|
||||
|
||||
Returns `{:statements [...] :stats {...}}` without executing them."
|
||||
Returns `{:nodes {table [attrs ...]} :edges [...] :stats {...}}`."
|
||||
[data file]
|
||||
(let [doc-id (or (:id data) (:id file))
|
||||
pages (seq (reverse (:pages data)))
|
||||
initial [(validated-node-statement specs/check-document "Document"
|
||||
(document-attrs file data))]
|
||||
[statements stats]
|
||||
(if (empty? pages)
|
||||
[initial {:documents 1 :pages 0 :shapes 0}]
|
||||
(reduce (fn [[stmts st] [position page-id]]
|
||||
(if-let [page (get-in data [:pages-index page-id])]
|
||||
(project-page stmts st doc-id page position)
|
||||
(do
|
||||
(l/wrn :hint "missing page in pages-index"
|
||||
:page-id (str page-id))
|
||||
[stmts st])))
|
||||
[initial {:documents 1 :pages 0 :shapes 0}]
|
||||
(map-indexed vector pages)))]
|
||||
{:statements statements
|
||||
:stats stats}))
|
||||
(let [doc-id (or (:id data) (:id file))
|
||||
doc-node (specs/check-document (document-attrs file data))
|
||||
pages (seq (reverse (:pages data)))
|
||||
acc0 (-> (initial-acc)
|
||||
(update-in [:nodes "Document"] (fnil conj []) doc-node)
|
||||
(assoc-in [:stats :documents] 1))
|
||||
acc (if (empty? pages)
|
||||
acc0
|
||||
(reduce (fn [acc [position page-id]]
|
||||
(if-let [page (get-in data [:pages-index page-id])]
|
||||
(project-page acc doc-id page position)
|
||||
(do
|
||||
(l/wrn :hint "missing page in pages-index"
|
||||
:page-id (str page-id))
|
||||
acc)))
|
||||
acc0
|
||||
(map-indexed vector pages)))]
|
||||
(select-keys acc [:nodes :edges :stats])))
|
||||
|
||||
50
backend/src/app/graph/project/specs.clj
Normal file
50
backend/src/app/graph/project/specs.clj
Normal file
@ -0,0 +1,50 @@
|
||||
;; This Source Code Form is subject to the terms of the Mozilla Public
|
||||
;; License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
;;
|
||||
;; Copyright (c) KALEIDOS INC Sucursal en España SL
|
||||
|
||||
(ns app.graph.project.specs
|
||||
"Malli schemas for graph projection payloads.
|
||||
|
||||
These are intentionally small subsets of the Penpot file model: they
|
||||
validate only the attributes written into Ladybug nodes for the
|
||||
current vertical slice."
|
||||
(:require
|
||||
[app.common.schema :as sm]))
|
||||
|
||||
(def schema:document
|
||||
[:map {:title "GraphDocument"}
|
||||
[:id ::sm/uuid]
|
||||
[:name :string]
|
||||
[:version {:optional true} :int]
|
||||
[:revision {:optional true} :int]])
|
||||
|
||||
(def schema:page
|
||||
[:map {:title "GraphPage"}
|
||||
[:id ::sm/uuid]
|
||||
[:name :string]
|
||||
[:index {:optional true} :int]])
|
||||
|
||||
(def schema:shape-node
|
||||
[:map {:title "GraphShapeNode"}
|
||||
[:id ::sm/uuid]
|
||||
[:name :string]])
|
||||
|
||||
(def check-document
|
||||
(sm/check-fn schema:document
|
||||
:type :validation
|
||||
:code :graph-document-projection
|
||||
:hint "invalid graph document projection"))
|
||||
|
||||
(def check-page
|
||||
(sm/check-fn schema:page
|
||||
:type :validation
|
||||
:code :graph-page-projection
|
||||
:hint "invalid graph page projection"))
|
||||
|
||||
(def check-shape-node
|
||||
(sm/check-fn schema:shape-node
|
||||
:type :validation
|
||||
:code :graph-shape-projection
|
||||
:hint "invalid graph shape projection"))
|
||||
@ -429,7 +429,8 @@
|
||||
|
||||
Options:
|
||||
- `:db-path` path or `:memory:`
|
||||
- `:reset-db?` delete any existing db first (default true)"
|
||||
- `:reset-db?` delete any existing db first (default true)
|
||||
- `:skip-stats?` skip post-ingest MATCH count queries (default false)"
|
||||
[file-id & opts]
|
||||
(let [result (apply graph.ingest/ingest-file! main/system file-id opts)]
|
||||
(graph.report/print-ingest! result)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user