github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-08-06 21:08:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

WIP arrow api

Browse Source
This commit is contained in:
Alejandro Alonso 2026-07-16 12:38:32 +02:00
parent 46937e143a
commit 0d5e1c33c0
7 changed files with 189 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
backend/deps.edn
View File

@ -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"}

26
backend/dev/graph_arrow_spike.clj
View File

@ -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")))

2
backend/scripts/_env
View File

@ -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

2
backend/scripts/run.template.sh
View File

@ -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};

110
backend/src/app/graph/bulk.clj
View File

@ -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]

89
backend/src/app/graph/ingest.clj
View File

@ -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))))

6
backend/src/app/srepl/main.clj
View File

@ -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)