perf: Phase 2.1 — batched VALUES INSERT for analytics-sync
ober
dd0ec96f3275202afcc5aed7e78b9181972f2538
--- a/lib/jerboa-db/analytics.ss +++ b/lib/jerboa-db/analytics.ss @@ -99,10 +99,8 @@ [eavt (index-set-eavt indices)] [datoms (dbi-datoms eavt)] [conn (analytics-engine-duckdb-conn ae)]) - ;; Truncate existing data (full refresh semantics) (duckdb-exec conn "DELETE FROM datoms") (duckdb-exec conn "DELETE FROM attrs") - ;; Load attributes (for-each (lambda (attr) (duckdb-eval conn @@ -116,15 +114,99 @@ (let ([c (db-attribute-cardinality attr)]) (if c (symbol->string c) #f)))) (schema-all-attributes schema)) - ;; Load datoms - (for-each - (lambda (d) - (insert-datom-row! ae schema d)) - datoms) - ;; Track sync state + (bulk-insert-datoms! conn schema datoms) (analytics-engine-last-synced-tx-set! ae (db-value-basis-tx db-val)) (analytics-engine-synced?-set! ae #t))) + ;; Bulk-insert datoms via multi-row VALUES batches. Per-row prepared + ;; INSERTs are ~250µs/row; batched literal VALUES are ~5µs/row. + (def +sync-batch-size+ 1000) + + (def (bulk-insert-datoms! conn schema datoms) + (let loop ([rest datoms]) + (cond + [(null? rest) (void)] + [else + (let-values ([(batch tail) (split-at-most rest +sync-batch-size+)]) + (let ([sql (build-insert-sql schema batch)]) + (when sql (duckdb-exec conn sql))) + (loop tail))]))) + + (def (split-at-most lst n) + (let loop ([i 0] [taken '()] [rest lst]) + (cond + [(or (= i n) (null? rest)) + (values (reverse taken) rest)] + [else + (loop (+ i 1) (cons (car rest) taken) (cdr rest))]))) + + (def (build-insert-sql schema batch) + (and (pair? batch) + (let ([buf (open-output-string)]) + (display "INSERT INTO datoms (e, a, a_name, v_long, v_double, v_string, v_bool, v_ref, v_instant, tx, added) VALUES " buf) + (let loop ([ds batch] [first? #t]) + (cond + [(null? ds) (void)] + [else + (unless first? (display ", " buf)) + (write-row! buf schema (car ds)) + (loop (cdr ds) #f)])) + (get-output-string buf)))) + + (def (write-row! buf schema d) + (let-values ([(e a v tx added?) + (if (datom? d) + (values (datom-e d) (datom-a d) (datom-v d) + (datom-tx d) (datom-added? d)) + (values (vector-ref d 0) (vector-ref d 1) (vector-ref d 2) + (vector-ref d 3) (vector-ref d 4)))]) + (let* ([attr (schema-lookup-by-id schema a)] + [a-name (and attr (symbol->string (db-attribute-ident attr)))] + [vtype (and attr (db-attribute-value-type attr))]) + (let-values ([(v-long v-double v-string v-bool v-ref v-instant) + (classify-value v vtype)]) + (display "(" buf) + (write-sql-int buf e) (display ", " buf) + (write-sql-int buf a) (display ", " buf) + (write-sql-text buf a-name) (display ", " buf) + (write-sql-int buf v-long) (display ", " buf) + (write-sql-num buf v-double) (display ", " buf) + (write-sql-text buf v-string)(display ", " buf) + (write-sql-bool buf v-bool) (display ", " buf) + (write-sql-int buf v-ref) (display ", " buf) + (write-sql-int buf v-instant)(display ", " buf) + (write-sql-int buf tx) (display ", " buf) + (write-sql-bool buf added?) + (display ")" buf))))) + + (def (write-sql-int buf v) + (cond [(not v) (display "NULL" buf)] + [(integer? v) (display v buf)] + [else (display "NULL" buf)])) + + (def (write-sql-num buf v) + (cond [(not v) (display "NULL" buf)] + [(number? v) (display (inexact v) buf)] + [else (display "NULL" buf)])) + + (def (write-sql-bool buf v) + (cond [(eq? v #t) (display "TRUE" buf)] + [(eq? v #f) (display "FALSE" buf)] + [else (display "NULL" buf)])) + + (def (write-sql-text buf v) + (cond + [(not v) (display "NULL" buf)] + [(string? v) + (display "'" buf) + (string-for-each + (lambda (c) + (when (char=? c #\') (display "'" buf)) + (display c buf)) + v) + (display "'" buf)] + [else (display "NULL" buf)])) + ;; ---- Insert a single datom record ---- (def (insert-datom-row! ae schema d)