objective categorical abstract machine language personal data server
Bulk insert records/MST blocks on importRepo
+134
-29
+26
-29
pegasus/lib/repository.ml
+26
-29
pegasus/lib/repository.ml
···
607
607
|> List.filter (fun cid ->
608
608
(not (Cid.equal cid root)) && not (Cid.Set.mem cid leaf_cids) )
609
609
in
610
+
(* collect mst node blocks for insert *)
611
+
let mst_blocks =
612
+
List.filter_map
613
+
(fun cid ->
614
+
match Block_map.get cid all_blocks with
615
+
| Some block ->
616
+
Some (cid, block)
617
+
| None ->
618
+
None )
619
+
mst_node_cids
620
+
in
621
+
(* collect record data for insert *)
622
+
let since = Tid.now () in
623
+
let record_data =
624
+
List.filter_map
625
+
(fun (path, cid) ->
626
+
match Block_map.get cid all_blocks with
627
+
| Some data ->
628
+
Some (path, cid, data, since)
629
+
| None ->
630
+
failwith ("missing record block: " ^ Cid.to_string cid) )
631
+
leaves
632
+
in
610
633
let%lwt _ =
611
634
Util.use_pool t.db.db (fun conn ->
612
635
Util.transact conn (fun () ->
613
-
(* store commit *)
614
636
let$! _ = User_store.Queries.put_commit root commit_bytes conn in
615
-
(* store mst nodes *)
616
-
let%lwt () =
617
-
Lwt_list.iter_s
618
-
(fun cid ->
619
-
match Block_map.get cid all_blocks with
620
-
| Some block ->
621
-
let$! _ = User_store.Queries.put_block cid block conn in
622
-
Lwt.return_unit
623
-
| None ->
624
-
Lwt.return_unit )
625
-
mst_node_cids
626
-
in
627
-
(* delete existing records *)
637
+
let$! () = User_store.Queries.clear_mst conn in
638
+
let$! () = User_store.Bulk.put_blocks mst_blocks conn in
628
639
let$! () =
629
640
[%rapper execute {sql| DELETE FROM records |sql}] () conn
630
641
in
631
-
(* store records *)
632
-
let%lwt () =
633
-
Lwt_list.iter_s
634
-
(fun (path, cid) ->
635
-
match Block_map.get cid all_blocks with
636
-
| Some data ->
637
-
let$! _ =
638
-
User_store.Queries.put_record ~path ~cid ~data
639
-
~since:(Tid.now ()) conn
640
-
in
641
-
Lwt.return_unit
642
-
| None ->
643
-
failwith ("missing record block: " ^ Cid.to_string cid) )
644
-
leaves
645
-
in
642
+
let$! () = User_store.Bulk.put_records record_data conn in
646
643
Lwt.return_ok () ) )
647
644
in
648
645
(* clear cached block_map so it's rebuilt on next access *)
+108
pegasus/lib/user_store.ml
+108
pegasus/lib/user_store.ml
···
528
528
let storage_str = Blob_store.storage_to_string storage in
529
529
Util.use_pool t.db
530
530
@@ Queries.list_blobs_by_storage ~storage:storage_str ~limit ~cursor
531
+
532
+
module Bulk = struct
533
+
open struct
534
+
let escape_sql_string s = Str.global_replace (Str.regexp "'") "''" s
535
+
536
+
let bytes_to_hex data =
537
+
let buf = Buffer.create (Bytes.length data * 2) in
538
+
Bytes.iter
539
+
(fun c -> Buffer.add_string buf (Printf.sprintf "%02x" (Char.code c)))
540
+
data ;
541
+
Buffer.contents buf
542
+
543
+
let chunk_list n lst =
544
+
if n <= 0 then invalid_arg "negative n passed to chunk_list" ;
545
+
let rec take_n acc remaining xs =
546
+
match (remaining, xs) with
547
+
| _, [] ->
548
+
(List.rev acc, [])
549
+
| 0, rest ->
550
+
(List.rev acc, rest)
551
+
| _, x :: xs' ->
552
+
take_n (x :: acc) (remaining - 1) xs'
553
+
in
554
+
let rec go xs =
555
+
match xs with
556
+
| [] ->
557
+
[]
558
+
| _ ->
559
+
let chunk, rest = take_n [] n xs in
560
+
chunk :: go rest
561
+
in
562
+
go lst
563
+
end
564
+
565
+
let put_blocks (blocks : (Cid.t * bytes) list) conn =
566
+
if List.is_empty blocks then Lwt.return_ok ()
567
+
else
568
+
let module C = (val conn : Caqti_lwt.CONNECTION) in
569
+
let chunks = chunk_list 200 blocks in
570
+
let rec process_chunks = function
571
+
| [] ->
572
+
Lwt.return_ok ()
573
+
| chunk :: rest -> (
574
+
let values =
575
+
List.map
576
+
(fun (cid, data) ->
577
+
let cid_str = escape_sql_string (Cid.to_string cid) in
578
+
let hex_data = bytes_to_hex data in
579
+
Printf.sprintf "('%s', X'%s')" cid_str hex_data )
580
+
chunk
581
+
|> String.concat ", "
582
+
in
583
+
let sql =
584
+
Printf.sprintf
585
+
"INSERT INTO mst (cid, data) VALUES %s ON CONFLICT DO NOTHING"
586
+
values
587
+
in
588
+
let query =
589
+
Caqti_request.Infix.( ->. ) Caqti_type.unit Caqti_type.unit sql
590
+
in
591
+
let%lwt result = C.exec query () in
592
+
match result with
593
+
| Ok () ->
594
+
process_chunks rest
595
+
| Error e ->
596
+
Lwt.return_error e )
597
+
in
598
+
process_chunks chunks
599
+
600
+
let put_records (records : (string * Cid.t * bytes * string) list) conn =
601
+
if List.is_empty records then Lwt.return_ok ()
602
+
else
603
+
let module C = (val conn : Caqti_lwt.CONNECTION) in
604
+
let chunks = chunk_list 100 records in
605
+
let rec process_chunks = function
606
+
| [] ->
607
+
Lwt.return_ok ()
608
+
| chunk :: rest -> (
609
+
let values =
610
+
List.map
611
+
(fun (path, cid, data, since) ->
612
+
let hex_data = bytes_to_hex data in
613
+
Printf.sprintf "('%s', '%s', X'%s', '%s')"
614
+
(escape_sql_string path)
615
+
(escape_sql_string (Cid.to_string cid))
616
+
hex_data (escape_sql_string since) )
617
+
chunk
618
+
|> String.concat ", "
619
+
in
620
+
let sql =
621
+
Printf.sprintf
622
+
"INSERT INTO records (path, cid, data, since) VALUES %s ON \
623
+
CONFLICT (path) DO UPDATE SET cid = excluded.cid, data = \
624
+
excluded.data, since = excluded.since"
625
+
values
626
+
in
627
+
let query =
628
+
Caqti_request.Infix.( ->. ) Caqti_type.unit Caqti_type.unit sql
629
+
in
630
+
let%lwt result = C.exec query () in
631
+
match result with
632
+
| Ok () ->
633
+
process_chunks rest
634
+
| Error e ->
635
+
Lwt.return_error e )
636
+
in
637
+
process_chunks chunks
638
+
end