mirror of
https://github.com/penpot/penpot.git
synced 2026-08-09 06:18:47 +00:00
✨ Bulk load through in-memory Arrow; delete the CSV loader
app.graph.arrow stages rows as Arrow VectorSchemaRoots and COPYs from them. No file is written at any point and no value is rendered as text for the engine to re-parse, so the defect class that produced three of this branch's four backend defects cannot recur. app.graph.bulk is deleted whole. csv-representable?, defer-to-cypher?, multiline?, fixup-statements, ladybug-literal, ladybug-list-element, ladybug-list-cell and staging-dir go with it, along with the post-COPY Cypher pass that emitted one SET per row. Measured before deciding: the fixup pass was ~77% execution, 16-22% parse and 6-7% round-trip, and prepared statements could not have recovered any of it — every fixup row carries a MAP column and Ladybug binds scalars only. So this replaces rather than optimizes. Marginal ingest 4.0 -> 1.21 ms/shape; ~25 s extrapolated at 20k shapes against the ~2 min the CSV path projected. Size unchanged. Four engine facts the implementation rests on, each verified against 0.18.2 with a standalone probe: - An Arrow table is not a COPY source identifier but is a MATCH-able node label. - A MAP vector's entries child must be a non-nullable struct, and MapVector.getWriter promotes it to a sparse union, so map vectors are built from an explicit Field and filled child-first. - Ladybug names a staged table's columns and struct fields from the Arrow field names and quotes none of them, so anything needing quotes must arrive quoted — hence cypher-property-key, not column-name, names the Arrow fields. - createArrowRelTable cannot resolve endpoints against a UUID-keyed node table under any encoding, so edges stage as a node table and the COPY subquery joins them. values/coerce is reused unchanged, so the Arrow and Cypher writers cannot disagree about a value's shape; nodes/column-map-key-fn is extracted so they cannot disagree about a MAP's key spelling either. Verified with pytest --graph-origin=penpot-only unchanged at 225/38/1 and --graph-origin=penpot unchanged at 258 passed / 2 pre-existing failures, both baselines re-established against a reverted backend rather than assumed; with bp graph diff between a CSV-built and an Arrow-built graph reporting "Graphs agree"; and with an adversarial round-trip carrying a quote, a backslash, a newline, a CRLF and a tab through STRING, STRING[] elements and MAP values. The diff was necessary, not belt-and-braces: both parity suites passed an earlier revision of this change that was writing EDN into every JSON column, because beadpot's assertions never parse those columns. It also showed Arrow correcting a CSV defect — an empty Component.path was being stored as NULL, because Ladybug's CSV reader cannot distinguish an empty field from an absent one.
This commit is contained in:
Álvaro Tejero Cantero
parent
c8ec4cc159
commit
47be384bef
372
backend/src/app/graph/arrow.clj
Normal file
372
backend/src/app/graph/arrow.clj
Normal file
@ -0,0 +1,372 @@
|
||||
;; 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.arrow
|
||||
"Bulk Ladybug ingest through in-memory Arrow.
|
||||
|
||||
Rows are built as Arrow `VectorSchemaRoot`s in the JVM's off-heap memory,
|
||||
handed to Ladybug as a virtual table, and `COPY`d into the real one. No file
|
||||
is written and no text is parsed, which is the whole point: the CSV path had
|
||||
to serialize every value into a literal Ladybug re-parsed, and its inner
|
||||
parser has no escaping at all, so a third of the old loader existed to detect
|
||||
and route around that (`app.graph.bulk`, since deleted). Arrow carries MAP,
|
||||
STRUCT, fixed-size arrays and multi-line strings natively.
|
||||
|
||||
The type language is Ladybug's, read recursively by `app.graph.schema.values`;
|
||||
this namespace adds the matching Arrow `Field` and a writer for each shape.
|
||||
`values/coerce` shapes a value first — a matrix into six doubles, a colour
|
||||
into a packed integer — exactly as it does for the Cypher path, so the two
|
||||
writers cannot disagree.
|
||||
|
||||
Engine facts this file depends on, each verified against lbug 0.18.2:
|
||||
|
||||
- An Arrow table is **not** a `COPY` source identifier, but it *is* a
|
||||
MATCH-able node label: `COPY T FROM (MATCH (n:stg) RETURN n.a AS a, …)`.
|
||||
- A MAP vector's `entries` child struct must be non-nullable, and
|
||||
`MapVector/getWriter` silently promotes it to a sparse union — so map
|
||||
vectors are built from an explicit `Field` and filled child-first.
|
||||
- Ladybug names a staged struct's fields from the Arrow child names and
|
||||
quotes none of them, so a reserved field name (`column`) must arrive
|
||||
already backticked.
|
||||
- `createArrowRelTable` cannot resolve endpoints against a UUID-keyed node
|
||||
table under any endpoint encoding, so edges are staged as a node table and
|
||||
joined by the `COPY` subquery instead."
|
||||
(:require
|
||||
[app.common.json :as json]
|
||||
[app.graph.ladybug :as ladybug]
|
||||
[app.graph.schema.nodes :as nodes]
|
||||
[app.graph.schema.values :as values]
|
||||
[clojure.string :as str])
|
||||
(:import
|
||||
com.ladybugdb.Connection
|
||||
com.ladybugdb.QueryResult
|
||||
java.nio.charset.StandardCharsets
|
||||
java.util.ArrayList
|
||||
java.util.List
|
||||
org.apache.arrow.memory.BufferAllocator
|
||||
org.apache.arrow.memory.RootAllocator
|
||||
org.apache.arrow.vector.BigIntVector
|
||||
org.apache.arrow.vector.BitVector
|
||||
org.apache.arrow.vector.FieldVector
|
||||
org.apache.arrow.vector.Float8Vector
|
||||
org.apache.arrow.vector.TimeStampMicroVector
|
||||
org.apache.arrow.vector.UInt4Vector
|
||||
org.apache.arrow.vector.VarCharVector
|
||||
org.apache.arrow.vector.VectorSchemaRoot
|
||||
org.apache.arrow.vector.complex.ListVector
|
||||
org.apache.arrow.vector.complex.MapVector
|
||||
org.apache.arrow.vector.complex.StructVector
|
||||
org.apache.arrow.vector.types.FloatingPointPrecision
|
||||
org.apache.arrow.vector.types.TimeUnit
|
||||
org.apache.arrow.vector.types.pojo.ArrowType$Bool
|
||||
org.apache.arrow.vector.types.pojo.ArrowType$FloatingPoint
|
||||
org.apache.arrow.vector.types.pojo.ArrowType$Int
|
||||
org.apache.arrow.vector.types.pojo.ArrowType$List
|
||||
org.apache.arrow.vector.types.pojo.ArrowType$Map
|
||||
org.apache.arrow.vector.types.pojo.ArrowType$Struct
|
||||
org.apache.arrow.vector.types.pojo.ArrowType$Timestamp
|
||||
org.apache.arrow.vector.types.pojo.ArrowType$Utf8
|
||||
org.apache.arrow.vector.types.pojo.Field
|
||||
org.apache.arrow.vector.types.pojo.FieldType
|
||||
org.apache.arrow.vector.types.pojo.Schema))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
;; --------------------------------------------------------------- allocator
|
||||
|
||||
(defn with-allocator!
|
||||
"Invoke `(f allocator)` with a fresh Arrow `RootAllocator`.
|
||||
|
||||
The allocator must outlive the Ladybug connection, because Ladybug releases
|
||||
its references to the staged buffers only when the Arrow tables are dropped —
|
||||
which happens on connection close at the latest. Closing it first surfaces as
|
||||
`IllegalStateException: Memory was leaked`, *thrown while unwinding*, which
|
||||
hides whatever actually failed. Any diagnostic here must catch inside this
|
||||
scope."
|
||||
[f]
|
||||
(with-open [allocator (RootAllocator.)]
|
||||
(f allocator)))
|
||||
|
||||
;; ------------------------------------------------------ Ladybug type → Field
|
||||
|
||||
(def ^:private scalar-arrow-type
|
||||
"Ladybug scalar → Arrow type. `UUID` and `JSON` ride as UTF-8: Ladybug
|
||||
accepts a string into either column and does the conversion itself, which is
|
||||
cheaper than teaching this side two more binary layouts."
|
||||
{"STRING" #(ArrowType$Utf8.)
|
||||
"UUID" #(ArrowType$Utf8.)
|
||||
"JSON" #(ArrowType$Utf8.)
|
||||
"INT64" #(ArrowType$Int. 64 true)
|
||||
"UINT32" #(ArrowType$Int. 32 false)
|
||||
"DOUBLE" #(ArrowType$FloatingPoint. FloatingPointPrecision/DOUBLE)
|
||||
"BOOLEAN" #(ArrowType$Bool.)
|
||||
"TIMESTAMP" #(ArrowType$Timestamp. TimeUnit/MICROSECOND nil)})
|
||||
|
||||
(defn column-field
|
||||
"Arrow `Field` for a column of `ladybug-type`, recursively.
|
||||
|
||||
`nullable?` is false only where Arrow's own invariants demand it — a MAP's
|
||||
`entries` struct and its key."
|
||||
(^Field [^String field-name ladybug-type]
|
||||
(column-field field-name ladybug-type true))
|
||||
(^Field [^String field-name ladybug-type nullable?]
|
||||
(cond
|
||||
;; A list first: `STRUCT(…)[]` starts with `STRUCT(` but is a list of them.
|
||||
(ladybug/list-type? ladybug-type)
|
||||
(Field. field-name (FieldType. nullable? (ArrowType$List.) nil)
|
||||
[(column-field "item" (values/list-element ladybug-type))])
|
||||
|
||||
(ladybug/map-type? ladybug-type)
|
||||
(let [[key-type value-type] (values/map-types ladybug-type)]
|
||||
(Field. field-name (FieldType. nullable? (ArrowType$Map. false) nil)
|
||||
[(Field. "entries" (FieldType. false (ArrowType$Struct.) nil)
|
||||
[(column-field "key" key-type false)
|
||||
(column-field "value" value-type)])]))
|
||||
|
||||
(ladybug/struct-type? ladybug-type)
|
||||
(Field. field-name (FieldType. nullable? (ArrowType$Struct.) nil)
|
||||
;; Backticks kept: Ladybug quotes none of these when it names the
|
||||
;; staged struct's fields, so `column` has to arrive quoted.
|
||||
(mapv (fn [[field field-type]] (column-field field field-type))
|
||||
(values/struct-fields-quoted ladybug-type)))
|
||||
|
||||
:else
|
||||
(if-let [mk (get scalar-arrow-type ladybug-type)]
|
||||
(Field. field-name (FieldType. nullable? (mk) nil) nil)
|
||||
(throw (ex-info (str "no Arrow mapping for Ladybug type: " ladybug-type)
|
||||
{:ladybug-type ladybug-type}))))))
|
||||
|
||||
;; ------------------------------------------------------------------- writer
|
||||
|
||||
(defn- utf8
|
||||
^bytes [v]
|
||||
(.getBytes (if (keyword? v) (name v) (str v)) StandardCharsets/UTF_8))
|
||||
|
||||
(defn- epoch-micros
|
||||
^long [v]
|
||||
(let [^java.time.Instant inst
|
||||
(cond
|
||||
(instance? java.time.Instant v) v
|
||||
(instance? java.util.Date v) (.toInstant ^java.util.Date v)
|
||||
:else (java.time.Instant/parse (str v)))]
|
||||
(+ (* (.getEpochSecond inst) 1000000) (long (quot (.getNano inst) 1000)))))
|
||||
|
||||
(defn- write-scalar!
|
||||
[^FieldVector fv ladybug-type ^long idx v]
|
||||
(case ladybug-type
|
||||
("STRING" "UUID") (.setSafe ^VarCharVector fv idx (utf8 v))
|
||||
;; A JSON column holds JSON, not a Clojure value's print form: `str` on a
|
||||
;; map yields `{:fill-color "#000000"}`, which is EDN and which every
|
||||
;; consumer of `fills`, `content` or `position_data` would fail to parse.
|
||||
;; Same encoder the Cypher path uses (`app.graph.ladybug/format-json`).
|
||||
"JSON" (.setSafe ^VarCharVector fv idx
|
||||
(.getBytes ^String (json/encode v)
|
||||
StandardCharsets/UTF_8))
|
||||
"INT64" (.setSafe ^BigIntVector fv idx (long v))
|
||||
"UINT32" (.setSafe ^UInt4Vector fv idx (unchecked-int (long v)))
|
||||
"DOUBLE" (.setSafe ^Float8Vector fv idx (double v))
|
||||
"BOOLEAN" (.setSafe ^BitVector fv idx (if v 1 0))
|
||||
"TIMESTAMP" (.setSafe ^TimeStampMicroVector fv idx (epoch-micros v))
|
||||
(throw (ex-info (str "no Arrow writer for Ladybug type: " ladybug-type)
|
||||
{:ladybug-type ladybug-type}))))
|
||||
|
||||
(defn write-value!
|
||||
"Write already-coerced `v` into `fv` at `idx`, per `ladybug-type`.
|
||||
|
||||
`map-key-fn` renders the keys of a `MAP(STRING, …)`, for the same reason
|
||||
`app.graph.ladybug/format-typed-value` takes one: the right spelling is a
|
||||
property of the column, not of the writer."
|
||||
;; `idx` is deliberately unhinted: Clojure only accepts primitive args on fns
|
||||
;; of four or fewer, and the map-key renderer has to travel with the value.
|
||||
[^FieldVector fv ladybug-type idx v map-key-fn]
|
||||
(if (nil? v)
|
||||
(.setNull fv (int idx))
|
||||
(cond
|
||||
(ladybug/list-type? ladybug-type)
|
||||
(let [^ListVector lv fv
|
||||
child (.getDataVector lv)
|
||||
element-type (values/list-element ladybug-type)
|
||||
elements (vec (if (or (sequential? v) (set? v)) v [v]))
|
||||
start (.startNewValue lv (int idx))]
|
||||
(dotimes [i (count elements)]
|
||||
(write-value! child element-type (+ start i) (nth elements i) map-key-fn))
|
||||
(.endValue lv (int idx) (count elements)))
|
||||
|
||||
(ladybug/map-type? ladybug-type)
|
||||
(let [^MapVector mv fv
|
||||
^StructVector entries (.getDataVector mv)
|
||||
[key-type value-type] (values/map-types ladybug-type)
|
||||
key-vec (.getChild entries "key")
|
||||
value-vec (.getChild entries "value")
|
||||
render-key (if (and map-key-fn (= "STRING" key-type)) map-key-fn identity)
|
||||
pairs (vec (seq v))
|
||||
start (.startNewValue mv (int idx))]
|
||||
(dotimes [i (count pairs)]
|
||||
(let [[k mv'] (nth pairs i)
|
||||
at (+ start i)]
|
||||
;; The entries struct is non-nullable: every slot must be defined.
|
||||
(.setIndexDefined entries (int at))
|
||||
(write-value! key-vec key-type at (render-key k) nil)
|
||||
(write-value! value-vec value-type at mv' map-key-fn)))
|
||||
(.endValue mv (int idx) (count pairs)))
|
||||
|
||||
(ladybug/struct-type? ladybug-type)
|
||||
(let [^StructVector sv fv]
|
||||
(.setIndexDefined sv (int idx))
|
||||
(doseq [[quoted-field field-type] (values/struct-fields-quoted ladybug-type)]
|
||||
;; The child is named with its backticks; the coerced value is keyed
|
||||
;; without them.
|
||||
(write-value! (.getChild sv quoted-field) field-type idx
|
||||
(get v (str/replace quoted-field "`" "")) map-key-fn)))
|
||||
|
||||
:else
|
||||
(write-scalar! fv ladybug-type (long idx) v))))
|
||||
|
||||
;; ------------------------------------------------------------------ batches
|
||||
|
||||
(defn- fill-vector!
|
||||
[^VectorSchemaRoot root ^String field-name ladybug-type rows value-fn map-key-fn]
|
||||
(let [^FieldVector fv (.getVector root field-name)]
|
||||
(.allocateNew fv)
|
||||
(dotimes [i (count rows)]
|
||||
(write-value! fv ladybug-type i
|
||||
(values/coerce ladybug-type (value-fn (nth rows i)))
|
||||
map-key-fn))
|
||||
(.setValueCount fv (count rows))))
|
||||
|
||||
(defn- node-batch
|
||||
"One `VectorSchemaRoot` holding every projected row of `table`.
|
||||
|
||||
Fields are named with their **backticks** (`cypher-property-key`, not
|
||||
`column-name`): Ladybug derives the staged table's DDL from the Arrow field
|
||||
names and quotes none of them, so a column whose name is a reserved word —
|
||||
`Page.index`, `Document.options` — fails `createArrowTable` outright unless it
|
||||
arrives already quoted. Same rule as for struct field names, and the `COPY`
|
||||
projection below spells them the same way."
|
||||
^VectorSchemaRoot [^BufferAllocator allocator table rows]
|
||||
(let [columns (nodes/column-keys table)
|
||||
fields (mapv (fn [k] (column-field (nodes/cypher-property-key table k)
|
||||
(nodes/column-ladybug-type table k)))
|
||||
columns)
|
||||
root (VectorSchemaRoot/create (Schema. ^List fields) allocator)]
|
||||
(doseq [k columns]
|
||||
(fill-vector! root (nodes/cypher-property-key table k)
|
||||
(nodes/column-ladybug-type table k)
|
||||
rows #(get % k) (nodes/column-map-key-fn table k)))
|
||||
(.setRowCount root (count rows))
|
||||
root))
|
||||
|
||||
(def ^:private edge-fields
|
||||
"Edge staging columns. `id` is the staging table's own key — Ladybug wants a
|
||||
first column to key the virtual table on — and `from`/`to` land as STRING,
|
||||
hence the cast in the join."
|
||||
[(Field. "id" (FieldType. true (ArrowType$Utf8.) nil) nil)
|
||||
(Field. "from" (FieldType. true (ArrowType$Utf8.) nil) nil)
|
||||
(Field. "to" (FieldType. true (ArrowType$Utf8.) nil) nil)
|
||||
(Field. "position" (FieldType. true (ArrowType$Int. 64 true) nil) nil)])
|
||||
|
||||
(defn- edge-batch
|
||||
^VectorSchemaRoot [^BufferAllocator allocator edges]
|
||||
(let [root (VectorSchemaRoot/create (Schema. ^List edge-fields) allocator)
|
||||
^VarCharVector iv (.getVector root "id")
|
||||
^VarCharVector fv (.getVector root "from")
|
||||
^VarCharVector tv (.getVector root "to")
|
||||
^BigIntVector pv (.getVector root "position")
|
||||
n (count edges)]
|
||||
(doseq [^FieldVector v [iv fv tv pv]] (.allocateNew v))
|
||||
(dotimes [i n]
|
||||
(let [{:keys [from-id to-id position]} (nth edges i)]
|
||||
(.setSafe iv i (utf8 i))
|
||||
(.setSafe fv i (utf8 from-id))
|
||||
(.setSafe tv i (utf8 to-id))
|
||||
(if (nil? position) (.setNull pv i) (.setSafe pv i (long position)))))
|
||||
(doseq [^FieldVector v [iv fv tv pv]] (.setValueCount v n))
|
||||
(.setRowCount root n)
|
||||
root))
|
||||
|
||||
;; ------------------------------------------------------------------ staging
|
||||
|
||||
(defn- batches
|
||||
^List [^VectorSchemaRoot root]
|
||||
(doto (ArrayList.) (.add root)))
|
||||
|
||||
(defn- check!
|
||||
[^QueryResult result hint data]
|
||||
(when-not (.isSuccess result)
|
||||
(throw (ex-info (str hint ": " (.getErrorMessage result))
|
||||
(assoc data :err (.getErrorMessage result))))))
|
||||
|
||||
(defn- with-staged-table!
|
||||
"Create Arrow table `staging-name` from `root`, run `(f)`, always drop it."
|
||||
[^Connection conn ^BufferAllocator allocator ^String staging-name
|
||||
^VectorSchemaRoot root data f]
|
||||
(try
|
||||
(with-open [^QueryResult r (.createArrowTable conn staging-name (batches root) allocator)]
|
||||
(check! r "createArrowTable failed" data))
|
||||
(f)
|
||||
(finally
|
||||
;; Dropped even on failure: the staged buffers stay referenced by Ladybug
|
||||
;; until it is, and the allocator's leak check fires on close otherwise.
|
||||
(try (.close ^QueryResult (.dropArrowTable conn staging-name))
|
||||
(catch Throwable _ nil)))))
|
||||
|
||||
(defn- copy-node-table!
|
||||
[^Connection conn table ^String staging-name]
|
||||
(let [projection (str/join ", " (for [k (nodes/column-keys table)
|
||||
:let [c (nodes/cypher-property-key table k)]]
|
||||
(str "n." c " AS " c)))
|
||||
statement (str "COPY `" table "` FROM (MATCH (n:" staging-name ") "
|
||||
"RETURN " projection ");")]
|
||||
(with-open [^QueryResult r (.query conn statement)]
|
||||
(check! r (str "COPY node table failed: " table)
|
||||
{:table table :statement statement}))))
|
||||
|
||||
(defn- copy-edge-group!
|
||||
"Load one FROM/TO pair of `IsChildOf`.
|
||||
|
||||
`createArrowRelTable` is unusable here — it cannot resolve endpoints against a
|
||||
UUID-keyed node table — so the edge list is staged as a node table and the
|
||||
endpoints are resolved by the subquery. The `WHERE` is clause-level because
|
||||
this dialect prohibits an inline pattern `WHERE`, and both sides are pinned by
|
||||
label so the join cannot reach outside the pair."
|
||||
[^Connection conn from-table to-table ^String staging-name]
|
||||
(let [statement (str "COPY `IsChildOf` FROM ("
|
||||
"MATCH (e:" staging-name "), "
|
||||
"(a:" (nodes/match-label from-table) "), "
|
||||
"(b:" (nodes/match-label to-table) ") "
|
||||
"WHERE a.id = cast(e.from AS UUID) "
|
||||
"AND b.id = cast(e.to AS UUID) "
|
||||
"RETURN a.id, b.id, e.position) "
|
||||
"(from='" from-table "', to='" to-table "');")]
|
||||
(with-open [^QueryResult r (.query conn statement)]
|
||||
(check! r (str "COPY edge group failed: " from-table " -> " to-table)
|
||||
{:from-table from-table :to-table to-table :statement statement}))))
|
||||
|
||||
(defn- staging-name
|
||||
[prefix & parts]
|
||||
(str/replace (str/join "_" (cons (str "stg_" prefix) parts)) #"[^A-Za-z0-9_]" "_"))
|
||||
|
||||
;; --------------------------------------------------------------------- load
|
||||
|
||||
(defn load-projection!
|
||||
"Load projected nodes and edges into an open Ladybug connection.
|
||||
|
||||
`allocator` must outlive `conn` — see `with-allocator!`."
|
||||
[^Connection conn {:keys [nodes edges]} ^BufferAllocator allocator]
|
||||
(doseq [[table rows] (sort-by key nodes)
|
||||
:when (seq rows)]
|
||||
(let [name (staging-name "node" table)]
|
||||
(with-open [root (node-batch allocator table rows)]
|
||||
(with-staged-table! conn allocator name root {:table table}
|
||||
#(copy-node-table! conn table name)))))
|
||||
(doseq [[[from-table to-table] group]
|
||||
(sort-by key (group-by (juxt :from-table :to-table) edges))
|
||||
:when (seq group)]
|
||||
(let [name (staging-name "edge" from-table to-table)]
|
||||
(with-open [root (edge-batch allocator group)]
|
||||
(with-staged-table! conn allocator name root
|
||||
{:from-table from-table :to-table to-table}
|
||||
#(copy-edge-group! conn from-table to-table name))))))
|
||||
@ -1,282 +0,0 @@
|
||||
;; 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.common.json :as json]
|
||||
[app.graph.ladybug :as ladybug]
|
||||
[app.graph.schema.nodes :as nodes]
|
||||
[app.graph.schema.values :as values]
|
||||
[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- multiline?
|
||||
"Does this value contain a newline?
|
||||
|
||||
Ladybug's parallel CSV reader rejects quoted newlines outright, and a shape
|
||||
name or text body may well contain one. Rather than flatten them — beadpot
|
||||
keeps them, and a graph is not a place to lose characters — such values are
|
||||
written through Cypher afterwards (`fixup-statements`)."
|
||||
[v]
|
||||
(and (string? v)
|
||||
(or (str/includes? v "\n") (str/includes? v "\r"))))
|
||||
|
||||
(defn- csv-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))))
|
||||
|
||||
(defn- ladybug-list-element
|
||||
"Format one element of a Ladybug LIST column for CSV COPY. Ladybug parses
|
||||
the (CSV-unquoted) field as a list literal: bare values for UUID/number
|
||||
elements, single-quoted strings (backslash-escaped) for STRING/JSON."
|
||||
[elem-type v]
|
||||
(cond
|
||||
(nil? v)
|
||||
"NULL"
|
||||
|
||||
(contains? #{"STRING" "JSON"} elem-type)
|
||||
;; Unreachable in practice: string-bearing lists go through Cypher
|
||||
;; (`csv-representable?`), because Ladybug's CSV list-literal parser has no
|
||||
;; escaping at all. Kept so the function stays total.
|
||||
(let [s (cond
|
||||
(coll? v) (json/encode v)
|
||||
(keyword? v) (name v)
|
||||
:else (str v))]
|
||||
(str "'" (-> s
|
||||
(str/replace "\\" "\\\\")
|
||||
(str/replace "'" "\\'"))
|
||||
"'"))
|
||||
|
||||
:else
|
||||
(str v)))
|
||||
|
||||
(defn- ladybug-literal
|
||||
"A coerced value as the list literal Ladybug's CSV reader expects.
|
||||
|
||||
Recursive, because the types are: `points` is a `DOUBLE[2][]`, and rendering
|
||||
its elements with `str` would emit Clojure's space-separated `[1 2]`, which
|
||||
Ladybug reads as a one-element array."
|
||||
[ladybug-type v]
|
||||
(if-let [[_ element] (and (string? ladybug-type)
|
||||
(re-matches #"(.+?)\[\d*\]$" ladybug-type))]
|
||||
(str "[" (str/join "," (map #(ladybug-literal element %)
|
||||
(if (coll? v) (seq v) [v])))
|
||||
"]")
|
||||
(ladybug-list-element ladybug-type v)))
|
||||
|
||||
(defn- ladybug-list-cell
|
||||
"CSV cell for a LIST-typed column (`UUID[]`, `DOUBLE[4]`, `DOUBLE[2][]`, …).
|
||||
JSON-encoding the collection (as `csv-cell` does) is wrong here: Ladybug
|
||||
expects its own list literal, e.g. `[id1,id2]` with bare elements."
|
||||
[ladybug-type v]
|
||||
(csv-escape-string (ladybug-literal ladybug-type v)))
|
||||
|
||||
(defn- compound-type?
|
||||
"A MAP or STRUCT, but not a *list* of them — a list is handled elementwise."
|
||||
[ladybug-type]
|
||||
(and (string? ladybug-type)
|
||||
(not (ladybug/list-type? ladybug-type))
|
||||
(or (str/starts-with? ladybug-type "MAP(")
|
||||
(str/starts-with? ladybug-type "STRUCT("))))
|
||||
|
||||
(defn- csv-representable?
|
||||
"Can a value of `ladybug-type` survive a CSV round-trip?
|
||||
|
||||
Ladybug parses the *contents* of a CSV field as a Cypher-ish literal for
|
||||
compound types, and that parser has no escape mechanism whatsoever: a comma
|
||||
inside a string element ends the element, and quotes are kept as part of the
|
||||
value rather than delimiting it (verified against 0.18). So only compound
|
||||
types whose elements cannot contain a delimiter — UUID, numbers, booleans —
|
||||
are safe; anything carrying a string or JSON is not, and neither is a MAP or
|
||||
a STRUCT. Those go through Cypher instead (`fixup-statements`), where
|
||||
`app.graph.ladybug` escapes properly.
|
||||
|
||||
Parquet would remove the distinction entirely (masterplan P0 T1 chose it,
|
||||
with CSV as the fallback of last resort); until the JVM side grows a Parquet
|
||||
writer, this is where the line falls."
|
||||
[ladybug-type]
|
||||
(cond
|
||||
(not (string? ladybug-type)) true
|
||||
(compound-type? ladybug-type) false
|
||||
:else
|
||||
(if-let [[_ element] (re-matches #"(.+?)\[\d*\]$" ladybug-type)]
|
||||
;; A list is safe only if its elements are: no delimiter can occur inside
|
||||
;; a UUID or a number, and a nested list or struct reintroduces the
|
||||
;; problem one level down.
|
||||
(and (not (contains? #{"STRING" "JSON"} element))
|
||||
(csv-representable? element))
|
||||
true)))
|
||||
|
||||
(defn- defer-to-cypher?
|
||||
"Must this value be written after the COPY rather than in the CSV?
|
||||
|
||||
Two reasons, both limitations of Ladybug's CSV reader rather than choices:
|
||||
a type its literal parser cannot escape, or a string containing a newline."
|
||||
[ladybug-type v]
|
||||
(or (not (csv-representable? ladybug-type))
|
||||
(multiline? v)))
|
||||
|
||||
(defn- csv-typed-cell
|
||||
[ladybug-type v]
|
||||
;; Shape the value for its column first (a matrix into six doubles, a hex
|
||||
;; colour into a packed integer); the CSV cell then only has to escape plain
|
||||
;; data. Same first step as the Cypher path, so the two agree.
|
||||
(let [v (values/coerce ladybug-type v)]
|
||||
(cond
|
||||
;; Written after the COPY, through Cypher — see `fixup-statements`.
|
||||
(defer-to-cypher? ladybug-type v) ""
|
||||
|
||||
(and (some? v)
|
||||
(string? ladybug-type)
|
||||
(re-matches #".+\[\d*\]$" ladybug-type))
|
||||
(ladybug-list-cell ladybug-type v)
|
||||
|
||||
:else (csv-cell v))))
|
||||
|
||||
(defn- fixup-statements
|
||||
"Cypher to set the values the CSV had to leave empty.
|
||||
|
||||
One statement per row that has any — not per column — so the cost is one
|
||||
round-trip per shape rather than per attribute, and a row with none costs
|
||||
nothing at all."
|
||||
[table rows]
|
||||
(let [columns (nodes/column-keys table)]
|
||||
(for [row rows
|
||||
;; An empty collection is a value, not an absence: a shape with no
|
||||
;; fills has `[]`, and writing NULL instead would tell a consumer the
|
||||
;; attribute is unknown rather than empty.
|
||||
:let [sets (for [k columns
|
||||
:let [v (get row k)
|
||||
t (nodes/column-ladybug-type table k)]
|
||||
:when (some? v)
|
||||
:when (defer-to-cypher? t v)]
|
||||
(str "n." (nodes/cypher-property-key table k) " = "
|
||||
(nodes/format-column-value table k v)))]
|
||||
:when (seq sets)]
|
||||
(str "MATCH (n:" (nodes/match-label table) " {id: "
|
||||
(ladybug/format-uuid (:id row)) "}) "
|
||||
"SET " (str/join ", " sets) ";"))))
|
||||
|
||||
(defn- cypher-file-path
|
||||
[^File file]
|
||||
(-> (.getAbsolutePath file)
|
||||
(str/replace "\\" "\\\\")
|
||||
(str/replace "'" "\\'")))
|
||||
|
||||
(defn- write-node-csv!
|
||||
[^File file table rows]
|
||||
(let [columns (nodes/column-keys table)
|
||||
types (mapv #(nodes/column-ladybug-type table %) columns)]
|
||||
(with-open [w (io/writer file :encoding "UTF-8")]
|
||||
;; COPY binds columns positionally (HEADER=true only skips the row), so
|
||||
;; this header is documentation — but it carries the beadpot column
|
||||
;; names, so a staged CSV reads the same as the table it loads into.
|
||||
(.write w (str (str/join "," (map #(nodes/column-name table %) columns)) "\n"))
|
||||
(doseq [row rows]
|
||||
(.write w (str (str/join "," (map (fn [k t] (csv-typed-cell t (get row k)))
|
||||
columns types))
|
||||
"\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 table rows)
|
||||
(copy-node-table! conn table csv-file)
|
||||
(when-let [stmts (seq (fixup-statements table rows))]
|
||||
(ladybug/exec-on-connection! conn stmts))))
|
||||
(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,7 +12,7 @@
|
||||
[app.common.logging :as l]
|
||||
[app.common.types.file :as ctf]
|
||||
[app.db :as db]
|
||||
[app.graph.bulk :as bulk]
|
||||
[app.graph.arrow :as graph.arrow]
|
||||
[app.graph.ladybug :as ladybug]
|
||||
[app.graph.meta :as graph.meta]
|
||||
[app.graph.project.document :as project.document]
|
||||
@ -21,7 +21,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]
|
||||
@ -38,10 +39,9 @@
|
||||
: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 ^BufferAllocator allocator
|
||||
{:keys [db-path skip-stats? skip-validation?] :or {skip-stats? 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)]
|
||||
@ -52,12 +52,11 @@
|
||||
:revn (:revn file)
|
||||
:db-path db-path
|
||||
:schema schema/schema-version)
|
||||
(let [ddl (schema/ddl-statements)
|
||||
(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)
|
||||
(graph.arrow/load-projection! conn {:nodes nodes :edges edges} allocator)
|
||||
(ladybug/exec-on-connection! conn ["CHECKPOINT;"])
|
||||
(let [transforms (project.transforms/apply-transforms! system conn data file)]
|
||||
;; Written last: its presence doubles as the build-complete marker.
|
||||
@ -75,15 +74,33 @@
|
||||
:stats (when-not skip-stats?
|
||||
(stats/summarize-connection conn))}))))
|
||||
|
||||
(defn ingest-on-connection!
|
||||
"Project `file-id` into an already open Ladybug `conn`.
|
||||
|
||||
Takes an `:arrow-alloc` when the caller already owns one; otherwise it makes
|
||||
a short-lived allocator around this call. A caller that opened the connection
|
||||
itself should pass its own, because the allocator has to be closed *after*
|
||||
the connection — see `app.graph.arrow/with-allocator!`."
|
||||
[system ^Connection conn file-id & {:keys [arrow-alloc] :as opts}]
|
||||
(if arrow-alloc
|
||||
(ingest-on-connection*! system conn file-id arrow-alloc opts)
|
||||
(graph.arrow/with-allocator!
|
||||
(fn [allocator] (ingest-on-connection*! system conn file-id allocator 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)))]
|
||||
(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?)))))
|
||||
;; Allocator outermost: Ladybug holds the staged Arrow buffers until its
|
||||
;; tables are dropped, which is no later than connection close, so the
|
||||
;; allocator must be closed after the connection and the database.
|
||||
(graph.arrow/with-allocator!
|
||||
(fn [allocator]
|
||||
(ladybug/with-connection! db-path
|
||||
(fn [conn]
|
||||
(ingest-on-connection*! system conn file-id allocator
|
||||
{:db-path db-path
|
||||
:skip-stats? skip-stats?
|
||||
:skip-validation? skip-validation?})))))))
|
||||
|
||||
@ -260,6 +260,15 @@
|
||||
[table k]
|
||||
(str "`" (column-name table k) "`"))
|
||||
|
||||
(defn column-map-key-fn
|
||||
"How a MAP column of `table` renders its keys.
|
||||
|
||||
A MAP's keys are values, not schema, so they keep the spelling their consumer
|
||||
parsed — `applied_tokens` is keyed in camelCase. Both writers need this, so it
|
||||
lives next to the column's type rather than in either of them."
|
||||
[table k]
|
||||
(contract/map-key-fn (column-name table k)))
|
||||
|
||||
(defn format-column-value
|
||||
"Cypher literal for `v` in column `k` of `table`.
|
||||
|
||||
@ -270,7 +279,7 @@
|
||||
[table k v]
|
||||
(ladybug/format-typed-value (column-ladybug-type table k)
|
||||
v
|
||||
(contract/map-key-fn (column-name table k))))
|
||||
(column-map-key-fn table k)))
|
||||
|
||||
(defn- create-node-table-ddl
|
||||
[{:keys [table pk]}]
|
||||
|
||||
@ -14,9 +14,9 @@
|
||||
what this namespace does: it turns records and maps into the numbers, vectors
|
||||
and plain maps the type names.
|
||||
|
||||
It deliberately stops there. Escaping — Cypher literals, CSV cells — belongs
|
||||
to the writer (`app.graph.ladybug`, `app.graph.bulk`), so that shaping a
|
||||
value and serializing it are separate concerns and each has one home.
|
||||
It deliberately stops there. Serialization belongs to the writer — Cypher
|
||||
literals in `app.graph.ladybug`, Arrow vectors in `app.graph.arrow` — so that
|
||||
shaping a value and writing it are separate concerns and each has one home.
|
||||
|
||||
The type language is the Ladybug one, read recursively: `T[]`, `T[n]`,
|
||||
`MAP(k, v)`, `STRUCT(name t, …)`. Anything else is passed through."
|
||||
@ -67,14 +67,15 @@
|
||||
(defn- parse-struct
|
||||
"`[[field-name field-type] …]` when `t` is a STRUCT type, else nil.
|
||||
|
||||
Field names arrive backtick-quoted (see `app.graph.schema.types`); the
|
||||
quoting is syntax, so it is stripped here and re-applied by the writer."
|
||||
[t]
|
||||
Field names arrive backtick-quoted (see `app.graph.schema.types`). The
|
||||
quoting is syntax, so it is stripped by default and re-applied by the writer —
|
||||
except for the Arrow writer, which needs it kept (`keep-quotes?`)."
|
||||
[t keep-quotes?]
|
||||
(when-let [[_ args] (re-matches #"STRUCT\((.*)\)$" t)]
|
||||
(for [arg (split-args args)
|
||||
:let [idx (str/index-of arg " ")]
|
||||
:when idx]
|
||||
[(str/replace (subs arg 0 idx) "`" "")
|
||||
[(cond-> (subs arg 0 idx) (not keep-quotes?) (str/replace "`" ""))
|
||||
(str/trim (subs arg (inc idx)))])))
|
||||
|
||||
(def ^:private struct-field-keys
|
||||
@ -135,12 +136,25 @@
|
||||
"`[[field-name field-type] …]` for a STRUCT type, memoized.
|
||||
|
||||
Public because the writers need the same field list to emit a literal."
|
||||
(memoize (fn [ladybug-type] (vec (parse-struct ladybug-type)))))
|
||||
(memoize (fn [ladybug-type] (vec (parse-struct ladybug-type false)))))
|
||||
|
||||
(def struct-fields-quoted
|
||||
"`struct-fields` with the DDL's backticks intact.
|
||||
|
||||
Only the Arrow writer wants this: Ladybug names a staged struct's fields from
|
||||
the Arrow child names and quotes none of them, so a field whose name is a
|
||||
reserved word — a layout grid cell's `column` — has to arrive already quoted
|
||||
or `createArrowTable` fails outright."
|
||||
(memoize (fn [ladybug-type] (vec (parse-struct ladybug-type true)))))
|
||||
|
||||
(def map-types
|
||||
"`[key-type value-type]` for a MAP type, memoized."
|
||||
(memoize (fn [ladybug-type] (parse-map ladybug-type))))
|
||||
|
||||
(def list-element
|
||||
"Element type of a `T[]` / `T[n]` column, memoized; nil when not a list."
|
||||
(memoize (fn [ladybug-type] (first (parse-list ladybug-type)))))
|
||||
|
||||
(declare coerce)
|
||||
|
||||
(defn- coerce-struct
|
||||
@ -183,6 +197,6 @@
|
||||
(into {}
|
||||
(map (fn [[k mv]] [(coerce key-type k) (coerce value-type mv)]))
|
||||
v))
|
||||
(if-let [fields (seq (parse-struct ladybug-type))]
|
||||
(if-let [fields (seq (parse-struct ladybug-type false))]
|
||||
(coerce-struct fields v)
|
||||
v)))))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user