From 0d5e1c33c0712d9ad95d20f3804ec3629959243e Mon Sep 17 00:00:00 2001 From: Alejandro Alonso Date: Thu, 16 Jul 2026 12:38:32 +0200 Subject: [PATCH] WIP arrow api --- backend/deps.edn | 3 +- backend/dev/graph_arrow_spike.clj | 26 +++---- backend/scripts/_env | 2 +- backend/scripts/run.template.sh | 2 +- backend/src/app/graph/bulk.clj | 110 ++++++++++++++++++++++++++---- backend/src/app/graph/ingest.clj | 89 ++++++++++++++++++------ backend/src/app/srepl/main.clj | 6 +- 7 files changed, 189 insertions(+), 49 deletions(-) diff --git a/backend/deps.edn b/backend/deps.edn index d43c81c954..dcafa97973 100644 --- a/backend/deps.edn +++ b/backend/deps.edn @@ -75,7 +75,8 @@ {:dev {:jvm-opts ["--sun-misc-unsafe-memory-access=allow" "--enable-native-access=ALL-UNNAMED" - "--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED"] + ;; Arrow jars are on the classpath (unnamed module), not module-path. + "--add-opens=java.base/java.nio=ALL-UNNAMED"] :extra-deps {com.bhauman/rebel-readline {:mvn/version "0.1.11"} clojure-humanize/clojure-humanize {:mvn/version "0.2.2"} diff --git a/backend/dev/graph_arrow_spike.clj b/backend/dev/graph_arrow_spike.clj index e374c0b04b..ee3b19f589 100644 --- a/backend/dev/graph_arrow_spike.clj +++ b/backend/dev/graph_arrow_spike.clj @@ -2,8 +2,12 @@ ;; Run inside devenv from backend/: ;; clojure -M:dev -m graph-arrow-spike ;; -;; Requires --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED +;; Requires --add-opens=java.base/java.nio=ALL-UNNAMED ;; (configured in deps.edn :dev :jvm-opts). +;; +;; FINDING: Arrow tables created with createArrowTable are NOT available as direct +;; identifiers in COPY statements, but ARE available as nodes in MATCH queries. +;; SOLUTION: Use "COPY table FROM (MATCH (n:arrow_table) RETURN ...)" pattern. (ns graph-arrow-spike (:gen-class) @@ -65,25 +69,23 @@ (println "=== 2) createArrowTable staging ===") (let [root2 (page-root alloc) batches (doto (ArrayList.) (.add root2))] - (with-open [r (.createArrowTable conn "_stg_Page" ^List batches alloc)] + (with-open [r (.createArrowTable conn "stg_Page" ^List batches alloc)] (check! r "createArrowTable")) (println "=== 3) query staging ===") - (try-query! conn "MATCH (n:_stg_Page) RETURN n.id, n.name;" "stg") + (try-query! conn "MATCH (n:stg_Page) RETURN n.id, n.name;" "stg") - (println "=== 4) COPY Page FROM _stg_Page ===") - (when-not (try-query! conn "COPY Page FROM _stg_Page;" "copy-ident") - (println "=== 4b) COPY via MATCH subquery ===") - (try-query! conn - "COPY Page FROM (MATCH (n:_stg_Page) RETURN n.id AS id, n.name AS name);" - "copy-subq")) + (println "=== 4) COPY Page FROM Arrow table via MATCH subquery ===") + (try-query! conn + "COPY Page FROM (MATCH (n:stg_Page) RETURN n.id AS id, n.name AS name);" + "copy-via-match") (println "=== 5) query native Page ===") (try-query! conn "MATCH (n:Page) RETURN n.id, n.name;" "page") (println "=== 6) dropArrowTable ===") - (with-open [r (.dropArrowTable conn "_stg_Page")] + (with-open [r (.dropArrowTable conn "stg_Page")] (println "drop success?" (.isSuccess r) "err" (.getErrorMessage r))) - (.close root2) - (println "DONE")))) + (.close root2)) + (println "DONE"))) \ No newline at end of file diff --git a/backend/scripts/_env b/backend/scripts/_env index 33ab6f0f15..d04fb6db15 100644 --- a/backend/scripts/_env +++ b/backend/scripts/_env @@ -85,7 +85,7 @@ export JAVA_OPTS="\ --sun-misc-unsafe-memory-access=allow \ --enable-preview \ --enable-native-access=ALL-UNNAMED \ - --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED"; + --add-opens=java.base/java.nio=ALL-UNNAMED"; function setup_minio() { if [ "${PENPOT_OBJECTS_STORAGE_BACKEND}" != "s3" ]; then diff --git a/backend/scripts/run.template.sh b/backend/scripts/run.template.sh index 143f14146d..19f47e6c0a 100644 --- a/backend/scripts/run.template.sh +++ b/backend/scripts/run.template.sh @@ -18,7 +18,7 @@ if [ -f ./environ ]; then source ./environ fi -export JAVA_OPTS="-Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager -Dlog4j2.configurationFile=log4j2.xml -XX:-OmitStackTraceInFastThrow --sun-misc-unsafe-memory-access=allow --enable-native-access=ALL-UNNAMED --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED --enable-preview $JVM_OPTS $JAVA_OPTS" +export JAVA_OPTS="-Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager -Dlog4j2.configurationFile=log4j2.xml -XX:-OmitStackTraceInFastThrow --sun-misc-unsafe-memory-access=allow --enable-native-access=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --enable-preview $JVM_OPTS $JAVA_OPTS" ENTRYPOINT=${1:-app.main}; diff --git a/backend/src/app/graph/bulk.clj b/backend/src/app/graph/bulk.clj index 2038c527b0..33b2571b8e 100644 --- a/backend/src/app/graph/bulk.clj +++ b/backend/src/app/graph/bulk.clj @@ -38,18 +38,102 @@ [s] (str "\"" (str/replace (csv-normalize-string s) "\"" "\"\"") "\"")) +(defn- number-string + [v] + (if (== v (long v)) (str (long v)) (str (double v)))) + +(defn- list-base-type + [lbug-type] + (when (and lbug-type (str/ends-with? lbug-type "[]")) + (subs lbug-type 0 (- (count lbug-type) 2)))) + +(defn- list-item-string + [base-type item] + (case base-type + "UUID" (str item) + "STRING" (str "\"" (str/replace (csv-normalize-string (str item)) "\"" "\"\"") "\"") + (#{"DOUBLE" "INT64"} base-type) + (number-string item) + + (if (string? item) + (str "\"" (csv-normalize-string item) "\"") + (str item)))) + +(defn- list-cell + [lbug-type v] + (when (some? v) + (let [base-type (list-base-type lbug-type)] + (if (empty? v) + "[]" + (str "[" (str/join ", " (map #(list-item-string base-type %) v)) "]"))))) + +(defn- csv-cell* + "Raw COPY cell value before CSV quoting." + [table col v] + (let [lbug-type (nodes/column-ladybug-type table col)] + (cond + (#{"DOUBLE" "INT64"} lbug-type) + (when (number? v) (number-string v)) + + (= lbug-type "BOOLEAN") + (when (boolean? v) (str v)) + + (= lbug-type "UUID") + (when (some? v) (str v)) + + (= lbug-type "TIMESTAMP") + (when (some? v) (str v)) + + (list-base-type lbug-type) + (list-cell lbug-type v) + + (nil? v) + "" + + (= lbug-type "JSON") + (json/encode v) + + (string? v) + (csv-normalize-string v) + + (keyword? v) + (name v) + + (uuid? v) + (str v) + + (number? v) + (number-string v) + + (boolean? v) + (str v) + + (map? v) + (json/encode v) + + (coll? v) + (json/encode v) + + :else + (str v)))) + (defn- csv-cell + [table col v] + (let [cell (csv-cell* table col v)] + (cond + (nil? cell) "" + (string? cell) (csv-escape-string cell) + :else (csv-escape-string (str cell))))) + +(defn- csv-scalar-cell [v] (cond - (nil? v) "" - (uuid? v) (str v) - (string? v) (csv-escape-string v) - (number? v) (if (== v (long v)) (str (long v)) (str (double v))) - (boolean? v) (str v) - (keyword? v) (csv-escape-string (name v)) - (map? v) (csv-escape-string (json/encode v)) - (coll? v) (csv-escape-string (json/encode v)) - :else (csv-escape-string (str v)))) + (nil? v) "" + (uuid? v) (csv-escape-string (str v)) + (string? v) (csv-escape-string v) + (number? v) (csv-escape-string (number-string v)) + (boolean? v) (csv-escape-string (str v)) + :else (csv-escape-string (str v)))) (defn- cypher-file-path [^File file] @@ -63,7 +147,7 @@ (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)) + (.write w (str (str/join "," (map #(csv-cell table % (get row %)) columns)) "\n")))))) (defn- write-edge-csv! @@ -71,9 +155,9 @@ (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"))))) + (.write w (str (csv-scalar-cell from-id) "," + (csv-scalar-cell to-id) "," + (csv-scalar-cell position) "\n"))))) (defn- delete-tree! [path] diff --git a/backend/src/app/graph/ingest.clj b/backend/src/app/graph/ingest.clj index b5bfb9dac7..6306f7364b 100644 --- a/backend/src/app/graph/ingest.clj +++ b/backend/src/app/graph/ingest.clj @@ -12,6 +12,8 @@ [app.common.logging :as l] [app.common.types.file :as ctf] [app.db :as db] + [app.graph.arrow :as graph.arrow] + [app.graph.arrow-simple :as arrow-simple] [app.graph.bulk :as bulk] [app.graph.ladybug :as ladybug] [app.graph.project.document :as project.document] @@ -20,7 +22,8 @@ [app.graph.stats :as stats] [app.srepl.helpers :as h]) (:import - com.ladybugdb.Connection)) + com.ladybugdb.Connection + org.apache.arrow.memory.BufferAllocator)) (defn- fetch-file! [system file-id] @@ -37,10 +40,10 @@ :file-id (str file-id))) [file-id file])) -(defn ingest-on-connection! - "Project `file-id` into an already open Ladybug `conn`." - [system ^Connection conn file-id & {:keys [db-path skip-stats? skip-validation?] - :or {skip-stats? true}}] +(defn- ingest-on-connection*! + [system ^Connection conn file-id + {:keys [db-path skip-stats? skip-validation? use-arrow? arrow-alloc] + :or {skip-stats? true use-arrow? true}}] (let [[file-id file] (fetch-file! system file-id) db-path (or db-path (ladybug/db-path-for-file file-id)) data (:data file)] @@ -50,13 +53,26 @@ :file-id (str file-id) :revn (:revn file) :db-path db-path - :schema schema/schema-version) - (let [ddl (schema/ddl-statements) + :schema schema/schema-version + :use-arrow? use-arrow?) + (let [ddl (schema/ddl-statements) {:keys [nodes edges stats]} - (project.document/projection-data data file) - staging-path (bulk/staging-dir db-path file-id)] + (project.document/projection-data data file)] (ladybug/exec-on-connection! conn ddl) - (bulk/load-projection! conn {:nodes nodes :edges edges} staging-path) + + (if use-arrow? + (do + (when-not arrow-alloc + (ex/raise :type :internal + :code :arrow-allocator-unavailable + :hint "Arrow ingest requires an Arrow allocator")) + (l/inf :hint "Using simple Arrow-based projection loading") + (arrow-simple/load-projection-with-arrow-simple! + conn {:nodes nodes :edges edges} ^BufferAllocator arrow-alloc)) + (let [staging-path (bulk/staging-dir db-path file-id)] + (l/inf :hint "Using CSV-based projection loading") + (bulk/load-projection! conn {:nodes nodes :edges edges} staging-path))) + (ladybug/exec-on-connection! conn ["CHECKPOINT;"]) {:file-id file-id :revn (:revn file) @@ -70,15 +86,50 @@ :stats (when-not skip-stats? (stats/summarize-connection conn))}))) +(defn ingest-on-connection! + "Project `file-id` into an already open Ladybug `conn`. + + When `:use-arrow?` is true and no `:arrow-alloc` is supplied, a temporary + RootAllocator is created for this call and closed afterwards." + [system ^Connection conn file-id & {:keys [use-arrow? arrow-alloc] :as opts + :or {use-arrow? true}}] + (let [opts (cond-> opts + (nil? (:use-arrow? opts)) + (assoc :use-arrow? true))] + (if (and use-arrow? (nil? arrow-alloc)) + (graph.arrow/with-allocator! + (fn [alloc] + (ingest-on-connection*! system conn file-id + (assoc opts :arrow-alloc alloc)))) + (ingest-on-connection*! system conn file-id opts)))) + (defn ingest-file! - [system file-id & {:keys [db-path reset-db? skip-stats? skip-validation?] - :or {reset-db? true}}] - (let [db-path (or db-path (ladybug/db-path-for-file (h/parse-uuid file-id)))] + "Ingest a file into a Ladybug database. + + By default the returned `:projection` only keeps `:stats` (not the full + `:nodes`/`:edges` maps) so REPL/`*1*` does not retain huge projections. + Pass `:keep-projection? true` when callers need the raw projection. + + Arrow allocators are closed after the Ladybug connection/database so + staging buffers are not retained for the process lifetime." + [system file-id & {:keys [db-path reset-db? skip-stats? skip-validation? use-arrow? keep-projection?] + :or {reset-db? true use-arrow? true}}] + (let [db-path (or db-path (ladybug/db-path-for-file (h/parse-uuid file-id))) + run (fn [arrow-alloc] + (ladybug/with-connection! db-path + (fn [conn] + (cond-> (ingest-on-connection*! + system conn file-id + {:db-path db-path + :skip-stats? skip-stats? + :skip-validation? skip-validation? + :use-arrow? use-arrow? + :arrow-alloc arrow-alloc}) + (not keep-projection?) + (update :projection select-keys [:stats])))))] (when reset-db? (ladybug/reset-db-path! db-path)) - (ladybug/with-connection! db-path - (fn [conn] - (ingest-on-connection! system conn file-id - :db-path db-path - :skip-stats? skip-stats? - :skip-validation? skip-validation?))))) + (if use-arrow? + ;; Allocator outside connection: close after Ladybug drops Arrow tables. + (graph.arrow/with-allocator! run) + (run nil)))) diff --git a/backend/src/app/srepl/main.clj b/backend/src/app/srepl/main.clj index 64331b85d9..f1cb7fbc9c 100644 --- a/backend/src/app/srepl/main.clj +++ b/backend/src/app/srepl/main.clj @@ -425,12 +425,14 @@ "Project a Penpot file into a per-file Ladybug database. Loads and realizes the file from the database, ensures the slice schema, - projects Document/Page/shape nodes, and returns graph stats. + projects Document/Page/shape nodes, and prints a short summary. Options: - `:db-path` path or `:memory:` - `:reset-db?` delete any existing db first (default true) - - `:skip-stats?` skip post-ingest MATCH count queries (default false)" + - `:skip-stats?` skip post-ingest MATCH count queries (default true) + - `:use-arrow?` Arrow bulk load (default true) + - `:keep-projection?` keep full `:nodes`/`:edges` in the return value (default false)" [file-id & opts] (let [result (apply graph.ingest/ingest-file! main/system file-id opts)] (graph.report/print-ingest! result)