From 23e55f07421a48f332d7d2ff9a2181e45143a85e Mon Sep 17 00:00:00 2001 From: Alejandro Alonso Date: Tue, 14 Jul 2026 08:00:24 +0200 Subject: [PATCH] :zap: Load graph ingest via Ladybug COPY bulk import --- backend/src/app/graph/bulk.clj | 131 +++++++++++++++++++++ backend/src/app/graph/ingest.clj | 16 ++- backend/src/app/graph/ladybug.clj | 11 +- backend/src/app/graph/project/document.clj | 129 ++++++++++---------- backend/src/app/graph/project/specs.clj | 50 ++++++++ backend/src/app/srepl/main.clj | 3 +- 6 files changed, 259 insertions(+), 81 deletions(-) create mode 100644 backend/src/app/graph/bulk.clj create mode 100644 backend/src/app/graph/project/specs.clj diff --git a/backend/src/app/graph/bulk.clj b/backend/src/app/graph/bulk.clj new file mode 100644 index 0000000000..2cb4518dff --- /dev/null +++ b/backend/src/app/graph/bulk.clj @@ -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)))) diff --git a/backend/src/app/graph/ingest.clj b/backend/src/app/graph/ingest.clj index 180f0cc342..6fedba50c5 100644 --- a/backend/src/app/graph/ingest.clj +++ b/backend/src/app/graph/ingest.clj @@ -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))}))))) diff --git a/backend/src/app/graph/ladybug.clj b/backend/src/app/graph/ladybug.clj index 9c8998fcb2..af5f45f9a4 100644 --- a/backend/src/app/graph/ladybug.clj +++ b/backend/src/app/graph/ladybug.clj @@ -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) diff --git a/backend/src/app/graph/project/document.clj b/backend/src/app/graph/project/document.clj index ea71b2b3a2..ec00e74806 100644 --- a/backend/src/app/graph/project/document.clj +++ b/backend/src/app/graph/project/document.clj @@ -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]))) diff --git a/backend/src/app/graph/project/specs.clj b/backend/src/app/graph/project/specs.clj new file mode 100644 index 0000000000..3050b02e04 --- /dev/null +++ b/backend/src/app/graph/project/specs.clj @@ -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")) diff --git a/backend/src/app/srepl/main.clj b/backend/src/app/srepl/main.clj index 99a0384f4f..6c71c23b6a 100644 --- a/backend/src/app/srepl/main.clj +++ b/backend/src/app/srepl/main.clj @@ -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)