diff --git a/backend/src/app/graph/arrow.clj b/backend/src/app/graph/arrow.clj new file mode 100644 index 0000000000..baee44b48a --- /dev/null +++ b/backend/src/app/graph/arrow.clj @@ -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)))))) diff --git a/backend/src/app/graph/bulk.clj b/backend/src/app/graph/bulk.clj deleted file mode 100644 index df51a87492..0000000000 --- a/backend/src/app/graph/bulk.clj +++ /dev/null @@ -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)))) diff --git a/backend/src/app/graph/ingest.clj b/backend/src/app/graph/ingest.clj index b95def578f..cf13d91559 100644 --- a/backend/src/app/graph/ingest.clj +++ b/backend/src/app/graph/ingest.clj @@ -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?}))))))) diff --git a/backend/src/app/graph/schema/nodes.clj b/backend/src/app/graph/schema/nodes.clj index 2a7ff0c4ae..9f863ee310 100644 --- a/backend/src/app/graph/schema/nodes.clj +++ b/backend/src/app/graph/schema/nodes.clj @@ -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]}] diff --git a/backend/src/app/graph/schema/values.clj b/backend/src/app/graph/schema/values.clj index 20ef3f86e8..2a618d5c02 100644 --- a/backend/src/app/graph/schema/values.clj +++ b/backend/src/app/graph/schema/values.clj @@ -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)))))