Volksdata 1.0b12
RDF library and triple store
Loading...
Searching...
No Matches
store_mdb.c
Go to the documentation of this file.
2
6#define N_DB 9
7
11#if (defined DEBUG || defined TESTING)
12#define DEFAULT_MAPSIZE 1<<24 // 16Mb (limit for Valgrind)
13#elif !(defined __LP64__ || defined __LLP64__) || \
14 defined _WIN32 && !defined _WIN64
15#define DEFAULT_MAPSIZE 1<<31 // 2Gb (limit for 32-bit systems)
16#else
17#define DEFAULT_MAPSIZE 1UL<<40 // 1Tb
18#endif
19
20#define ENV_DIR_MODE 0750
21#define ENV_FILE_MODE 0640
22
23/*
24 * Data types.
25 */
26
27typedef char DbLabel[8];
28typedef struct mdbstore_iter_t MDBIterator;
29
31typedef enum {
32 LSSTORE_OPEN = 1<<0,
34
36typedef enum {
38 //<
39 //< The iterator has begun a new
40 //< transaction on initialization
41 //< which needs to be closed. If
42 //< false, the iterator is using an
43 //< existing transaction which will
44 //< not be closed with
45 //< #mdbiter_free().
46} IterFlags;
47
48typedef enum {
51} StoreOp;
52
53typedef struct mdbstore_t {
54 MDB_env * env;
55 MDB_dbi dbi[N_DB];
57} MDBStore;
58
69typedef void (*iter_op_fn_t)(MDBIterator *it);
70
71
73typedef struct mdbstore_iter_t {
76 MDB_txn * txn;
77 MDB_cursor * cur;
78 MDB_cursor * ctx_cur;
79 MDB_val key;
80 MDB_val data;
87 const uint8_t * term_order;
90 size_t i;
91 size_t ct;
93 int rc;
95
96
97/*
98 * Static variables.
99 */
100
101#define DUPFIXED_MASK MDB_DUPSORT | MDB_DUPFIXED
102
108#define MAIN_TABLE \
109/* #ID pfx #DB label #Flags */ \
110 ENTRY( T_ST, "t:st", 0 ) /* Key to ser. term */ \
111 ENTRY( SPO_C, "spo:c", DUPFIXED_MASK ) /* Triple to context */ \
112
113
116#define LOOKUP_TABLE \
117/* #ID pfx #DB label #Flags */ \
118 ENTRY( S_PO, "s:po", DUPFIXED_MASK ) /* 1-bound lookup */ \
119 ENTRY( P_SO, "p:so", DUPFIXED_MASK ) /* 1-bound lookup */ \
120 ENTRY( O_SP, "o:sp", DUPFIXED_MASK ) /* 1-bound lookup */ \
121 ENTRY( PO_S, "po:s", DUPFIXED_MASK ) /* 2-bound lookup */ \
122 ENTRY( SO_P, "so:p", DUPFIXED_MASK ) /* 2-bound lookup */ \
123 ENTRY( SP_O, "sp:o", DUPFIXED_MASK ) /* 2-bound lookup */ \
124 ENTRY( C_SPO, "c:spo", DUPFIXED_MASK ) /* Context lookup */ \
125
126
129#define ENTRY(a, b, c) static const DbLabel DB_##a = b;
132#undef ENTRY
133
134/*
135 * Numeric index of each DB. Prefixed with IDX_
136 *
137 * These index numbers are referred to in all the arrays defeined below. They
138 * are independent from the LMDB dbi values which are considered opaque here.
139 */
140typedef enum {
141#define ENTRY(a, b, c) IDX_##a,
144#undef ENTRY
145} DBIdx;
146
150static const char *db_labels[N_DB] = {
151#define ENTRY(a, b, c) DB_##a,
154#undef ENTRY
155};
156
157/*
158 * DB flags. These are aligned with the dbi_labels index.
159 */
160static const unsigned int db_flags[N_DB] = {
161#define ENTRY(a, b, c) c,
164#undef ENTRY
165};
166
167/*
168 * 1-bound and 2-bound lookup indices.
169 *
170 * N.B. Only the first 6 (1-bound and 2-bound term lookup) are used.
171 * The others are added just because they belong logically to the lookup table.
172 */
173static DBIdx lookup_indices[9] = {
174#define ENTRY(a, b, c) IDX_##a,
176#undef ENTRY
177};
178
179static const uint8_t lookup_ordering_1bound[3][3] = {
180 {0, 1, 2}, // s:po
181 {1, 0, 2}, // p:so
182 {2, 0, 1}, // o:sp
183};
184
185static const uint8_t lookup_ordering_2bound[3][3] = {
186 {1, 2, 0}, // po:s
187 {0, 2, 1}, // so:p
188 {0, 1, 2}, // sp:o
189};
190
191
192/*
193 * Static prototypes.
194 */
195static int index_triple(
196 MDBStore *store, StoreOp op, VOLK_TripleKey spok, VOLK_Key ck,
197 MDB_txn *txn);
198static VOLK_rc mdbstore_add_term (void *h, const VOLK_Buffer *sterm, void *th);
199
200inline static VOLK_rc lookup_0bound (MDBIterator *it, size_t *ct);
201inline static VOLK_rc lookup_1bound (
202 uint8_t idx0, MDBIterator *it, size_t *ct);
203inline static VOLK_rc lookup_2bound (
204 uint8_t idx0, uint8_t idx1, MDBIterator *it, size_t *ct);
205inline static VOLK_rc lookup_3bound(MDBIterator *it, size_t *ct);
206
207
208static const char *
209mdbstore_path_from_id (const char *id)
210{
211 // Set environment path.
212 if (!id) id = getenv ("VOLK_MDB_STORE_URN");
213 if (!id) {
215 log_info (
216 "`VOLK_MDB_STORE_URN' environment variable is not "
217 "set. The default URN %s has been set as the store ID.", id
218 );
219 } else if (strncmp ("file://", id, 7) != 0) {
220 log_error ("MDB store ID must be in the `file://<abs_path>` format.");
221
222 return NULL;
223 }
224
225 return id + 7;
226}
227
228
229/*
230 * Inliners.
231 */
232
233static inline VOLK_rc
234txn_begin (MDB_env *env, MDB_txn *p, unsigned int f, MDB_txn **tp) {
235 RCCK (mdb_txn_begin (env, p, f, tp));
236 const char *path;
237 RCCK (mdb_env_get_path (env, &path));
238 STRACE (
239 "BEGIN %s transaction %p child of %p in env %s",
240 f == 0 ? "RW" : "RO", *tp, p, path);
241
242 return VOLK_OK;
243}
244
245
246static inline VOLK_rc
247txn_commit (MDB_txn *t) {
248 log_debug ("COMMIT transaction %p", t);
249 return mdb_txn_commit (t);
250}
251
252
256
263static VOLK_rc
264mdbstore_setup (const char *id, bool clear)
265{
266 if (!VOLK_env_is_init) return VOLK_ENV_ERR;
267
268 const char *path = mdbstore_path_from_id (id);
269 if (!path) return VOLK_VALUE_ERR;
270
271 // If the directory exists (unless clear == true), do nothing.
272 if (clear) rm_r (path);
273 VOLK_rc rc = mkdir_p (path, ENV_DIR_MODE);
274 log_debug ("Create dir rc: %s", VOLK_strerror (rc));
275 PRCCK (rc);
276
277 // Open a temporary environment and txn to create the DBs.
278 MDB_env *env;
279 RCCK (mdb_env_create (&env));
280
281 RCCK (mdb_env_set_maxdbs (env, N_DB));
282 RCCK (mdb_env_open (env, path, 0, ENV_FILE_MODE));
283 log_debug ("Environment opened at %s.", path);
284
285 MDB_txn *txn;
286 RCCK (txn_begin (env, NULL, 0, &txn));
287 MDB_dbi dbi;
288 for (int i = 0; i < N_DB; i++) {
289 log_trace ("Creating DB %s", db_labels[i]);
290 RCCK (
291 mdb_dbi_open (txn, db_labels[i], db_flags[i] | MDB_CREATE, &dbi)
292 );
293 }
294
295 // Bootstrap the permanent store with initial data.
296 MDB_stat stat;
297 CHECK (mdb_dbi_open (
298 txn, db_labels[IDX_T_ST], db_flags[IDX_T_ST], &dbi), fail);
299 CHECK (mdb_stat (txn, dbi, &stat), fail);
300
301 if (stat.ms_entries == 0) {
302 log_debug ("Loading initial data into %s", path);
303 // Index default context.
304 MDB_cursor *cur;
305 CHECK (mdb_cursor_open (txn, dbi, &cur), fail);
307 MDB_val key, data;
308 key.mv_data = &k;
309 key.mv_size = sizeof (k);
310
311 data.mv_data = VOLK_default_ctx_buf->addr;
312 data.mv_size = VOLK_default_ctx_buf->size;
313
314 VOLK_rc db_rc = mdb_cursor_put (cur, &key, &data, 0);
315 CHECK (db_rc, fail);
316 }
317
318 CHECK (txn_commit (txn), fail);
319 mdb_env_close (env);
320
321 return clear ? VOLK_OK : rc;
322
323fail:
324 if (rc >= 0) rc = VOLK_DB_ERR;
325 mdb_txn_abort (txn);
326 return rc;
327}
328
329
340static void *
341mdbstore_new (const char *id, size_t _unused)
342{
343 if (!VOLK_env_is_init) {
344 log_error (VOLK_strerror (VOLK_ENV_ERR));
345 return NULL;
346 }
347
348 (void) _unused;
349 const char *path = mdbstore_path_from_id (id);
350 if (!path) return NULL;
351
352 MDBStore *store;
353 CALLOC_GUARD (store, NULL);
354
355 RCNL (mdb_env_create (&store->env));
356 MDB_txn *txn = NULL;
357
358 // Set map size.
359 size_t mapsize;
360 char *env_mapsize = getenv ("VOLK_MDB_MAPSIZE");
361 if (env_mapsize == NULL) mapsize = DEFAULT_MAPSIZE;
362 else sscanf (env_mapsize, "%zu", &mapsize);
363 log_debug (
364 "Setting environment map size at %s to %zu Mb.",
365 path, mapsize / 1024 / 1024);
366 CHECK (mdb_env_set_mapsize (store->env, mapsize), fail);
367 CHECK (mdb_env_set_maxdbs (store->env, N_DB), fail);
368 CHECK (mdb_env_open (store->env, path, 0, ENV_FILE_MODE), fail);
369
370 // Assign DB handles to store->dbi.
371 CHECK (txn_begin (store->env, NULL, 0, &txn), fail);
372 for (int i = 0; i < N_DB; i++)
373 CHECK (mdb_dbi_open (
374 txn, db_labels[i], db_flags[i], store->dbi + i), fail);
375
376 store->flags |= LSSTORE_OPEN;
377 CHECK (txn_commit(txn), fail);
378 txn = NULL;
379
380 log_info ("Created MDB environment at %s", path);
381
382 return store;
383
384fail:
385 if (txn) mdb_txn_abort (txn);
386 mdb_env_close (store->env);
387
388 return NULL;
389}
390
391
392static void
393mdbstore_free (void *h)
394{
395 MDBStore *store = h;
396 if (store->flags & LSSTORE_OPEN) {
397 const char *path;
398 mdb_env_get_path (store->env, &path);
399 log_info ("Closing MDB environment at %s.", path);
400 mdb_env_close (store->env);
401 }
402
403 free (store);
404}
405
406
407static VOLK_rc
408mdbstore_stat (const MDBStore *store, MDB_stat *stat)
409{
410 if (!(store->flags & LSSTORE_OPEN)) return 0;
411
412 MDB_txn *txn;
413 RCCK (txn_begin (store->env, NULL, MDB_RDONLY, &txn));
414
415 if (mdb_stat (txn, store->dbi[IDX_SPO_C], stat) != MDB_SUCCESS)
416 return VOLK_DB_ERR;
417 mdb_txn_abort (txn);
418
419 return VOLK_OK;
420}
421
422
423static size_t
424mdbstore_size (const void *h)
425{
426 const MDBStore *store = h;
427 // Size is calculated outside of any pending write txn.
428
429 MDB_stat stat;
430 if (mdbstore_stat (store, &stat) != VOLK_OK) return 0;
431
432 return stat.ms_entries;
433}
434
435
436char *
437mdbstore_id (const void *h)
438{
439 const MDBStore *store = h;
440 const char *path;
441
442 RCNL (mdb_env_get_path (store->env, &path));
443
444 char *id = malloc (strlen (path) + 8);
445 sprintf (id, "file://%s", path);
446
447 return id;
448}
449
450
451static VOLK_rc
452mdbstore_txn_begin (void *h, VOLK_StoreFlags flags, void **th)
453{
454 MDBStore *store = h;
455 flags = flags & VOLK_STORE_TXN_RO ? MDB_RDONLY : 0;
456
457 RCCK (txn_begin (store->env, NULL, flags, (MDB_txn **) th));
458
459 return VOLK_OK;
460}
461
462
463static VOLK_rc
464mdbstore_txn_commit (void *th)
465{
466 RCCK (txn_commit (th));
467
468 return VOLK_OK;
469}
470
471
472static void
473mdbstore_txn_abort (void *th)
474{ mdb_txn_abort (th); }
475
476
477static void *
478mdbiter_txn (void *h)
479{ return ((MDBIterator *) h)->txn; }
480
481
491static void *
492mdbstore_add_init (void *h, const VOLK_Buffer *sc, void *th)
493{
494 MDBStore *store = h;
495 /* An iterator is used here. Some members are a bit misused but it does
496 * its job without having to define a very similar struct.
497 */
498 MDBIterator *it = calloc (1, sizeof (*it));
499 NLNL (it);
500
501 it->store = store;
502 if (th) it->txn = th;
503 else {
504 CHECK (txn_begin (store->env, th, 0, &it->txn), fail);
505 it->flags |= ITER_OPEN_TXN;
506 }
507
508 if (sc) {
509 // Store context if it's not the default one.
510 it->luc = VOLK_buffer_hash (sc);
511
512 // Insert t:st for context.
513 //log_debug ("Adding context: %s", sc);
514 it->key.mv_data = &it->luc;
515 it->key.mv_size = KLEN;
516 it->data.mv_data = sc->addr;
517 it->data.mv_size = sc->size;
518
519 int db_rc = mdb_put (
520 it->txn, it->store->dbi[IDX_T_ST],
521 &it->key, &it->data, MDB_NOOVERWRITE);
522 if (db_rc != MDB_SUCCESS && db_rc != MDB_KEYEXIST) {
523 log_error (VOLK_strerror (db_rc));
524 mdb_txn_abort (it->txn);
525 return NULL;
526 }
527 } else {
528 log_debug ("No context passed to iterator, using default.");
530 }
531
532 return it;
533
534fail:
535 free (it);
536
537 return NULL;
538}
539
540
541/*
542 * NOTE: at the moment #mdbstore_remove() or another
543 * #mdbstore_init() cannot be called between #mdbstore_add_init and
544 * #mdbstore_add_abort or #mdbstore_add_done. FIXME
545 *
546 */
547static VOLK_rc
548mdbstore_add_iter (void *h, const VOLK_BufferTriple *sspo)
549{
550 if (UNLIKELY (!h)) return VOLK_VALUE_ERR;
551
552 MDBIterator *it = h;
553 int db_rc = VOLK_NOACTION;
555
556 // Add triple terms.
557 for (int i = 0; i < 3; i++) {
558 VOLK_Buffer *st = VOLK_btriple_pos (sspo, i);
559
560 spok[i] = VOLK_buffer_hash (st);
561
562 it->key.mv_data = spok + i;
563 it->key.mv_size = KLEN;
564 it->data.mv_data = st->addr;
565 it->data.mv_size = st->size;
566
567 db_rc = mdb_put(
568 it->txn, it->store->dbi[IDX_T_ST],
569 &it->key, &it->data, MDB_NOOVERWRITE);
570 if (db_rc != MDB_SUCCESS && db_rc != MDB_KEYEXIST) {
571 LOG_RC (db_rc);
572 return VOLK_DB_ERR;
573 }
574 }
575
576 log_trace ("Inserting spok: {%x, %x, %x}", spok[0], spok[1], spok[2]);
577 log_trace ("Into context: %x", it->luc);
578
579 // Insert spo:c.
580 it->key.mv_data = spok;
581 it->key.mv_size = TRP_KLEN;
582
583 // In triple mode, data is empty (= NULL_KEY).
584 it->data.mv_data = &it->luc;
585 it->data.mv_size = it->luc == NULL_KEY ? 0 : KLEN;
586
587 db_rc = mdb_put(
588 it->txn, it->store->dbi[IDX_SPO_C],
589 &it->key, &it->data, MDB_NODUPDATA);
590
591 if (db_rc == MDB_KEYEXIST) return VOLK_NOACTION;
592 if (db_rc != MDB_SUCCESS) {
593 log_error (
594 "MDB error while inserting triple: %s", VOLK_strerror(db_rc));
595 return VOLK_DB_ERR;
596 }
597
598 // Index.
599 VOLK_rc rc = index_triple (it->store, OP_ADD, spok, it->luc, it->txn);
600 if (rc == VOLK_OK) it->i++;
601
602 return rc;
603}
604
605
606static VOLK_rc
607mdbstore_add_done (void *h)
608{
609 MDBIterator *it = h;
610
611 VOLK_rc rc = VOLK_OK;
612 log_debug ("Committing add transaction.");
613
614 if (it->flags & ITER_OPEN_TXN) {
615 if (txn_commit (it->txn) != MDB_SUCCESS) {
616 log_error ("Error committing transaction. Aborting instead.");
617 mdb_txn_abort (it->txn);
618 rc = VOLK_TXN_ERR;
619 }
620 }
621
622 free (it);
623
624 RCCK (rc);
625 return rc;
626}
627
628
629static void
630mdbstore_add_abort (void *h)
631{
632 MDBIterator *it = h;
633 if (it->flags & ITER_OPEN_TXN) mdb_txn_abort (it->txn);
634
635 free (it);
636}
637
638
639static VOLK_rc
640key_to_sterm (
641 MDBStore *store, MDB_txn *txn, const VOLK_Key key, VOLK_Buffer *sterm)
642{
644 int db_rc;
645
646 MDB_val key_v, data_v;
647 key_v.mv_data = (void*)&key;
648 key_v.mv_size = KLEN;
649
650 db_rc = mdb_get (txn, store->dbi[IDX_T_ST], &key_v, &data_v);
651
652 sterm->flags |= VOLK_BUF_BORROWED;
653 if (db_rc == MDB_SUCCESS) {
654 sterm->addr = data_v.mv_data;
655 sterm->size = data_v.mv_size;
656 rc = VOLK_OK;
657 } else if (db_rc == MDB_NOTFOUND) {
658 sterm->addr = NULL;
659 sterm->size = 0;
660 } else {
661 LOG_RC (db_rc);
662 rc = VOLK_DB_ERR;
663 }
664
665 return rc;
666}
667
668
669static void *
670mdbstore_lookup (
671 void *h, const VOLK_Buffer *ss, const VOLK_Buffer *sp,
672 const VOLK_Buffer *so, const VOLK_Buffer *sc, void *th, size_t *ct)
673{
674 VOLK_TripleKey spok = {
675 VOLK_buffer_hash (ss),
676 VOLK_buffer_hash (sp),
677 VOLK_buffer_hash (so),
678 };
679
680 MDBIterator *it;
681 CALLOC_GUARD (it, NULL);
682
683 it->store = h;
684 it->luc = VOLK_buffer_hash (sc);
685 log_debug ("Lookup context: %x", it->luc);
686
687 if (ct) *ct = 0;
688
689 uint8_t idx0, idx1;
690
691 if (th) it->txn = th;
692 else {
693 RCNL (txn_begin (it->store->env, NULL, MDB_RDONLY, &it->txn));
694 log_trace ("Opening new lookup transaction @%p", it->txn);
695 it->flags |= ITER_OPEN_TXN;
696 }
697
698 // Context index loop.
699 RCNL (mdb_cursor_open (
700 it->txn, it->store->dbi[IDX_SPO_C], &it->ctx_cur));
701
702 /*
703 * Lookup decision tree.
704 */
705 // s p o (all terms bound)
706 if (spok[0] != NULL_KEY && spok[1] != NULL_KEY && spok[2] != NULL_KEY) {
707 it->luk[0] = spok[0];
708 it->luk[1] = spok[1];
709 it->luk[2] = spok[2];
710 PRCNL (lookup_3bound (it, ct));
711
712 } else if (spok[0] != NULL_KEY) {
713 it->luk[0] = spok[0];
714 idx0 = 0;
715
716 // s p ?
717 if (spok[1] != NULL_KEY) {
718 it->luk[1] = spok[1];
719 idx1 = 1;
720 PRCNL (lookup_2bound (idx0, idx1, it, ct));
721
722 // s ? o
723 } else if (spok[2] != NULL_KEY) {
724 it->luk[1] = spok[2];
725 idx1 = 2;
726 PRCNL (lookup_2bound (idx0, idx1, it, ct));
727
728 // s ? ?
729 } else PRCNL (lookup_1bound (idx0, it, ct));
730
731 } else if (spok[1] != NULL_KEY) {
732 it->luk[0] = spok[1];
733 idx0 = 1;
734
735 // ? p o
736 if (spok[2] != NULL_KEY) {
737 it->luk[1] = spok[2];
738 idx1 = 2;
739 PRCNL (lookup_2bound (idx0, idx1, it, ct));
740
741 // ? p ?
742 } else PRCNL (lookup_1bound (idx0, it, ct));
743
744 // ? ? o
745 } else if (spok[2] != NULL_KEY) {
746 it->luk[0] = spok[2];
747 idx0 = 2;
748 PRCNL (lookup_1bound (idx0, it, ct));
749
750 // ? ? ? (all terms unbound)
751 } else PRCNL (lookup_0bound (it, ct));
752
753 return it;
754}
755
756
762static VOLK_rc
763mdbiter_next_key (MDBIterator *it)
764{
766 // Only advance if the previous it->rc wasn't already at the end.
767 if (it->rc == MDB_NOTFOUND) return VOLK_END;
768 RCCK (it->rc);
769
770 /* Retrieve current value and advance cursor to the next result.
771 * it->rc is set to the result of the next iteration.
772 */
773 it->iter_op_fn (it);
774 MDB_val key, data;
775 int db_rc;
776
777 key.mv_size = TRP_KLEN;
778 data.mv_data = &it->luc;
779 data.mv_size = KLEN;
780
781 VOLK_rc rc;
782 if (it->luc) {
783 rc = VOLK_NORESULT; // Flow control value, will never be returned.
784 do {
785 //log_debug ("begin ctx loop.");
786 /* If ctx is specified, look if the matching triple is associated
787 * with it. If not, move on to the next triple.
788 * The loop normally exits when a triple with matching ctx is found
789 * (VOLK_OK), if there are no more triples (VOLK_END), or if there
790 * is an error (VOLK_DB_ERR).
791 */
792 log_trace (
793 "Found spok: {%x, %x, %x}",
794 it->spok[0], it->spok[1], it->spok[2]);
795
796 key.mv_data = it->spok;
797 db_rc = mdb_cursor_get (it->ctx_cur, &key, &data, MDB_GET_BOTH);
798
799 if (db_rc == MDB_NOTFOUND) {
800 log_trace ("Triple not found in context: %x", it->luc);
801 // if the iterator is at the end, stop here.
802 if (it->rc == MDB_NOTFOUND) rc = VOLK_END;
803 // Otherwise, look up the next key.
804 else it->iter_op_fn (it);
805
806 } else {
807 RCCK (db_rc);
808 rc = VOLK_OK;
809 log_trace ("Triple found in context: %x", it->luc);
810 }
811 } while (rc == VOLK_NORESULT);
812
813 } else {
814 log_trace (
815 "Found spok in any context: {%x, %x, %x}",
816 it->spok[0], it->spok[1], it->spok[2]);
817 rc = VOLK_OK;
818 }
819
820 // Get all contexts for a triple.
821 key.mv_data = it->spok;
822 db_rc = mdb_cursor_get (it->ctx_cur, &key, &data, MDB_SET_KEY);
823 if (db_rc != MDB_SUCCESS) {
824 log_error ("No context found for triple!");
825 return VOLK_DB_ERR;
826 }
827
828 size_t ct;
829 db_rc = mdb_cursor_count (it->ctx_cur, &ct);
830 if (db_rc != MDB_SUCCESS) return VOLK_DB_ERR;
831 // 1 spare for sentinel. Always allocated even on zero matches.
832 VOLK_Key *tmp_ck = realloc (it->ck, sizeof (*it->ck) * (ct + 1));
833 NLRCCK (tmp_ck, VOLK_MEM_ERR);
834 it->ck = tmp_ck;
835
836 size_t i = 0;
837 do {
838 //log_trace ("Copying to slot #%lu @%p", i, it->ck + i);
839 memcpy (it->ck + i++, data.mv_data, sizeof (*it->ck));
840 } while (
841 mdb_cursor_get (it->ctx_cur, &key, &data, MDB_NEXT_DUP)
842 == MDB_SUCCESS);
843 //log_trace ("setting sentinel @%p", it->ck + i);
844 it->ck[i] = NULL_KEY;
845
846 return rc;
847}
848
849
850static VOLK_rc
851mdbiter_next (
852 void *h, VOLK_BufferTriple *sspo, VOLK_Buffer **ctx_p)
853{
854 MDBIterator *it = h;
855 VOLK_rc rc = mdbiter_next_key (it);
856
857 if (rc == VOLK_OK) {
858 if (sspo) {
859 RCCK (key_to_sterm (it->store, it->txn, it->spok[0], sspo->s));
860 RCCK (key_to_sterm (it->store, it->txn, it->spok[1], sspo->p));
861 RCCK (key_to_sterm (it->store, it->txn, it->spok[2], sspo->o));
862 }
863
864 // Contexts for current triple.
865 if (ctx_p) {
866 // Preallocate.
867 size_t i = 0;
868 while (it->ck[i++]); // Include sentinel in count.
869 VOLK_Buffer *ctx;
870 log_trace ("Allocating %lu context buffers + sentinel.", i - 1);
871 ctx = malloc(i * sizeof (*ctx));
872 if (!ctx) return VOLK_MEM_ERR;
873
874 for (i = 0; it->ck[i]; i++)
875 RCCK (key_to_sterm (it->store, it->txn, it->ck[i], ctx + i));
876 memset (ctx + i, 0, sizeof (*ctx)); // Sentinel
877 NLRCCK (ctx + i, VOLK_MEM_ERR);
878
879 // TODO error handling.
880 *ctx_p = ctx;
881 }
882 }
883
884 return rc;
885}
886
887
888static void
889mdbiter_free (void *h)
890{
891 if (!h) return;
892 MDBIterator *it = h;
893
894 if (it->cur) mdb_cursor_close (it->cur);
895 if (it->ctx_cur) mdb_cursor_close (it->ctx_cur);
896 if (it->flags & ITER_OPEN_TXN) mdb_txn_abort (it->txn);
897 free (it->ck);
898
899 free (it);
900}
901
902
903static VOLK_rc
904mdbstore_update_ctx (
905 void *h, const VOLK_Buffer *old_c, const VOLK_Buffer *new_c, void *th)
906{
908 MDBStore *store = h;
909 unsigned char *trp_data = NULL;
910
912 old_ck = VOLK_buffer_hash (old_c),
913 new_ck = VOLK_buffer_hash (new_c);
914 // lu_key, lu_data look up all triples with old context in c:spo, and
915 // replace old c with new c.
916 MDB_txn
917 *p_txn = th,
918 *txn;
919 CHECK (rc = txn_begin (store->env, p_txn, 0, &txn), finally);
920
921 MDB_cursor *i_cur, *d_cur;
922 CHECK (
923 rc = mdb_cursor_open (txn, store->dbi[IDX_C_SPO], &i_cur),
924 close_txn);
925
926 MDB_val key, data;
927
928 // Return error if the graph URI already exists.
929 key.mv_data = &new_ck;
930 key.mv_size = KLEN;
931 rc = mdb_cursor_get (i_cur, &key, &data, MDB_SET);
932 if (rc == MDB_SUCCESS) {
933 log_error (
934 "Context key %lu already exists. Not replacing old graph.",
935 new_ck);
936 rc = VOLK_CONFLICT;
937 goto close_i;
938 } else if (rc != MDB_NOTFOUND) CHECK (rc, close_i);
939
940 // Add new context term.
941 CHECK (rc = mdbstore_add_term (store, new_c, txn), close_i);
942
943 key.mv_data = &old_ck;
944 // Count triples in cursor.
945 rc = mdb_cursor_get (i_cur, &key, &data, MDB_SET);
946 if (rc == MDB_NOTFOUND) {
947 log_info ("No triples found associated with old context.");
948 rc = VOLK_NOACTION;
949 goto close_i;
950 } else CHECK (rc, close_i);
951
952 // From here on, it can only be VOLK_OK or error.
953 rc = VOLK_OK;
954 size_t trp_ct;
955 CHECK (rc = mdb_cursor_count (i_cur, &trp_ct), close_i);
956 trp_data = malloc (trp_ct * TRP_KLEN);
957 if (UNLIKELY (!trp_data)) {
958 rc = VOLK_MEM_ERR;
959 goto close_i;
960 }
961
962 // Copy triple data as one block to temp buffer so that entries can be
963 // deleted while cursors are active.
964 rc = mdb_cursor_get (i_cur, &key, &data, MDB_GET_MULTIPLE);
965 if (rc != MDB_SUCCESS) {
966 rc = rc == MDB_NOTFOUND ? VOLK_NOACTION : VOLK_DB_ERR;
967 goto close_i;
968 }
969 size_t loc_cur = 0;
970 do {
971 memcpy (trp_data + loc_cur, data.mv_data, data.mv_size);
972 loc_cur += data.mv_size;
973 } while (mdb_cursor_get (
974 i_cur, &key, &data, MDB_NEXT_MULTIPLE) == MDB_SUCCESS);
975
976 // Zap c:spo entries in one go.
977 key.mv_data = &old_ck;
978 key.mv_size = KLEN;
979 data.mv_size = TRP_KLEN;
980 CHECK (rc = mdb_cursor_get (i_cur, &key, NULL, MDB_SET), close_i);
981 CHECK (rc = mdb_cursor_del (i_cur, MDB_NODUPDATA), close_i);
982
983 // Re-ad c:spo data individually.
984 key.mv_data = &new_ck;
985 for (size_t i = 0; i < trp_ct; i++) {
986 data.mv_data = trp_data + i * data.mv_size;
987 CHECK (
988 rc = mdb_cursor_put (i_cur, &key, &data, MDB_APPENDDUP),
989 close_i);
990 }
991 // Re-add c:spo data in bulk from buffer with new context.
992 // FIXME this is not working. Replaced by the for loop above.
993 /*
994 MDB_val data_block[] = {
995 { .mv_data = &new_ck, .mv_size = TRP_KLEN },
996 { .mv_data = NULL, .mv_size = trp_ct },
997 };
998 db_rc = mdb_cursor_put (i_cur, &key, data_block, MDB_MULTIPLE);
999 */
1000
1001 // Main table.
1002 // Replace spo:c values one by one.
1003 CHECK (rc = mdb_cursor_open (txn, store->dbi[IDX_SPO_C], &d_cur), close_i);
1004 key.mv_size = TRP_KLEN;
1005 data.mv_size = KLEN;
1006 for (size_t i = 0; i < trp_ct; i++) {
1007 key.mv_data = trp_data + i * key.mv_size;
1008 data.mv_data = &old_ck;
1009 CHECK (
1010 rc = mdb_cursor_get (d_cur, &key, &data, MDB_GET_BOTH),
1011 close_d);
1012 CHECK (rc = mdb_cursor_del (d_cur, 0), close_d);
1013 data.mv_data = &new_ck;
1014 CHECK (
1015 rc = mdb_cursor_put (d_cur, &key, &data, MDB_NOOVERWRITE),
1016 close_d);
1017 }
1018
1019close_d:
1020 mdb_cursor_close (d_cur);
1021close_i:
1022 mdb_cursor_close (i_cur);
1023close_txn:
1024 if (rc == VOLK_OK) RCCK (txn_commit (txn));
1025 else mdb_txn_abort (txn);
1026
1027 if (trp_data) free (trp_data);
1028finally:
1029
1030 return rc;
1031}
1032
1033
1034static VOLK_rc
1035mdbstore_remove (
1036 void *h, const VOLK_Buffer *ss, const VOLK_Buffer *sp,
1037 const VOLK_Buffer *so, const VOLK_Buffer *sc, void *th, size_t *ct_p)
1038{
1039 MDBStore *store = h;
1041
1042 MDB_txn *txn;
1043 RCCK (txn_begin (store->env, th, 0, &txn));
1044 MDB_cursor *cur;
1045 mdb_cursor_open (txn, store->dbi[IDX_SPO_C], &cur);
1046
1047 if (sc == NULL) sc = VOLK_default_ctx_buf;
1048 VOLK_Key ck = VOLK_buffer_hash (sc);
1049 MDB_val spok_v, ck_v;
1050 spok_v.mv_size = TRP_KLEN;
1051 ck_v.mv_size = KLEN;
1052 ck_v.mv_data = &ck;
1053
1054 // Gather all the matching triples in a first pass.
1055 size_t *ct = ct_p ? ct_p : malloc (sizeof (*ct));
1056 NLRCCK (ct, VOLK_MEM_ERR);
1057 MDBIterator *it = mdbstore_lookup (store, ss, sp, so, sc, txn, ct);
1058 NLRCCK (it, VOLK_DB_ERR);
1059 log_debug ("Found %lu triples to remove.", *ct);
1060 VOLK_Key *keys = malloc (*ct * sizeof (VOLK_Key) * 3);
1061 NLRCCK (it, VOLK_MEM_ERR);
1062 size_t i = 0;
1063 while (mdbiter_next_key (it) == VOLK_OK) {
1064 log_trace ("Adding triple #%zu to remove list.", i);
1065 memcpy (keys + (3 * i++), &it->spok, TRP_KLEN);
1066 }
1067 mdbiter_free (it);
1068
1069 // Iterate over the gathered keys and delete them.
1070 for (i = 0; i < *ct; i++) {
1071 spok_v.mv_data = keys + i * 3;
1072 ck_v.mv_data = &ck;
1073 log_trace (
1074 "Removing triple #%zu: %x {%x %x %x}",
1075 i,
1076 ((VOLK_Key *)ck_v.mv_data)[0],
1077 ((VOLK_Key *)spok_v.mv_data)[0],
1078 ((VOLK_Key *)spok_v.mv_data)[1],
1079 ((VOLK_Key *)spok_v.mv_data)[2]);
1080
1081 rc = mdb_cursor_get (cur, &spok_v, &ck_v, MDB_GET_BOTH);
1082 if (rc == MDB_NOTFOUND) {
1083 log_warn ("No key found in spo:c DB.");
1084 continue; // TODO This could be a data problem.
1085 } else CHECK (rc, fail);
1086
1087 // Delete spo:c entry.
1088 CHECK (rc = mdb_cursor_del (cur, 0), fail);
1089 CHECK (rc = index_triple (
1090 store, OP_REMOVE, keys + i * 3, ck, txn
1091 ), fail);
1092 }
1093 free(keys);
1094 CHECK (txn_commit (txn), fail);
1095
1096 return rc;
1097
1098fail:
1099 mdb_txn_abort (txn);
1100 RCCK (rc);
1101
1102 return VOLK_DB_ERR;
1103}
1104
1105
1117static VOLK_rc
1118mdbstore_add_term (void *h, const VOLK_Buffer *sterm, void *th)
1119{
1120 //log_trace ("Adding term to MDB store: %s", sterm->addr);
1121 MDBStore *store = h;
1122 int db_rc;
1123 MDB_val key, data;
1124
1125 MDB_txn *txn;
1126 // If an active transaction was passed, use it, otherwise open and
1127 // close a new one.
1128 if (th) txn = th;
1129 else RCCK (txn_begin (store->env, th, 0, &txn));
1130
1131 MDB_cursor *cur;
1132 CHECK (mdb_cursor_open (txn, store->dbi[IDX_T_ST], &cur), fail);
1133
1134 VOLK_Key k = VOLK_buffer_hash (sterm);
1135 key.mv_data = &k;
1136 key.mv_size = sizeof (k);
1137
1138 data.mv_data = sterm->addr;
1139 data.mv_size = sterm->size;
1140
1141 db_rc = mdb_cursor_put (cur, &key, &data, MDB_NOOVERWRITE);
1142 if (db_rc != MDB_KEYEXIST) CHECK (db_rc, fail);
1143
1144 if (txn != th) CHECK (db_rc = txn_commit (txn), fail);
1145
1146 return VOLK_OK;
1147
1148fail:
1149 if (txn != th) mdb_txn_abort (txn);
1150 log_trace ("Aborted txn for adding term.");
1151 return VOLK_DB_ERR;
1152}
1153
1154
1155VOLK_Buffer **
1156mdbstore_ctx_list (void *h, void *th)
1157{
1158 MDBStore *store = h;
1159 VOLK_rc db_rc;
1160 MDB_txn *txn;
1161 VOLK_Buffer **tdata = NULL;
1162 if (th) txn = th;
1163 else CHECK (txn_begin (store->env, NULL, MDB_RDONLY, &txn), fail_txn);
1164
1165
1166 MDB_cursor *cur;
1167 CHECK (mdb_cursor_open (txn, store->dbi[IDX_C_SPO], &cur), fail);
1168 MDB_val key, data;
1169
1170 db_rc = mdb_cursor_get (cur, &key, &data, MDB_FIRST);
1171 size_t i = 0;
1172 while (db_rc == MDB_SUCCESS) {
1173 i++;
1174 db_rc = mdb_cursor_get (cur, &key, &data, MDB_NEXT_NODUP);
1175 }
1176 tdata = malloc ((i + 1) * sizeof (*tdata));
1177 NLCHECK (tdata, fail);
1178
1179 db_rc = mdb_cursor_get (cur, &key, &data, MDB_FIRST);
1180 i = 0;
1181 while (db_rc == MDB_SUCCESS) {
1182 tdata[i] = BUF_DUMMY;
1183 VOLK_Key tkey = *(VOLK_Key*)key.mv_data;
1184 CHECK (key_to_sterm (store, txn, tkey, tdata[i++]), fail);
1185 db_rc = mdb_cursor_get (cur, &key, &data, MDB_NEXT_NODUP);
1186 }
1187 tdata[i] = NULL; // Sentinel
1188 mdb_cursor_close (cur);
1189 if (txn != th) mdb_txn_abort (txn);
1190
1191 return tdata;
1192
1193fail:
1194 if (txn != th) mdb_txn_abort (txn);
1195 if (tdata) free (tdata);
1196fail_txn:
1197 return NULL;
1198}
1199
1200
1202 .name = "MDB Store",
1205
1206 .setup_fn = mdbstore_setup,
1207 .new_fn = mdbstore_new,
1208 .free_fn = mdbstore_free,
1209
1210 .size_fn = mdbstore_size,
1211 .id_fn = mdbstore_id,
1212
1213 .txn_begin_fn = mdbstore_txn_begin,
1214 .txn_commit_fn = mdbstore_txn_commit,
1215 .txn_abort_fn = mdbstore_txn_abort,
1216 .iter_txn_fn = mdbiter_txn,
1217
1218 .add_init_fn = mdbstore_add_init,
1219 .add_iter_fn = mdbstore_add_iter,
1220 .add_abort_fn = mdbstore_add_abort,
1221 .add_done_fn = mdbstore_add_done,
1222 .add_term_fn = mdbstore_add_term,
1223
1224 .update_ctx_fn = mdbstore_update_ctx,
1225
1226 .lookup_fn = mdbstore_lookup,
1227 .lu_next_fn = mdbiter_next,
1228 .lu_free_fn = mdbiter_free,
1229
1230 .remove_fn = mdbstore_remove,
1231
1232 .ctx_list_fn = mdbstore_ctx_list,
1233};
1234
1235
1236/* * * Static functions. * * */
1237
1247static VOLK_rc
1248index_triple(
1249 MDBStore *store, StoreOp op, VOLK_TripleKey spok, VOLK_Key ck,
1250 MDB_txn *txn)
1251{
1252 int db_rc;
1254 MDB_val v1, v2;
1255
1256 log_trace ("Indexing triple: {%x %x %x}", spok[0], spok[1], spok[2]);
1257
1258 // Index c:spo.
1259 if (op == OP_REMOVE) {
1260 log_trace ("Indexing op: REMOVE");
1261 if (ck == NULL_KEY) goto skip_remove;
1262
1263 MDB_cursor *cur;
1264 v1.mv_data = &ck;
1265 v1.mv_size = KLEN;
1266 v2.mv_data = spok;
1267 v2.mv_size = TRP_KLEN;
1268
1269 RCCK (mdb_cursor_open (txn, store->dbi[IDX_C_SPO], &cur));
1270 db_rc = mdb_cursor_get (cur, &v1, &v2, MDB_GET_BOTH);
1271 if (db_rc != MDB_NOTFOUND) {
1272 RCCK (db_rc);
1273 RCCK (mdb_cursor_del (cur, 0));
1274 v1.mv_data = &ck;
1275 v2.mv_data = spok;
1276 rc = VOLK_OK;
1277 }
1278 mdb_cursor_close (cur);
1279skip_remove:
1280
1281 } else if (op == OP_ADD) {
1282 log_trace ("Indexing op: ADD");
1283 if (ck == NULL_KEY) goto skip_add;
1284
1285 v1.mv_data = &ck;
1286 v1.mv_size = KLEN;
1287 v2.mv_data = spok;
1288 v2.mv_size = TRP_KLEN;
1289
1290 db_rc = mdb_put(
1291 txn, store->dbi[IDX_C_SPO],
1292 &v1, &v2, MDB_NODUPDATA);
1293 if (db_rc != MDB_SUCCESS) return VOLK_DB_ERR;
1294 if (db_rc != MDB_KEYEXIST) rc = VOLK_OK;
1295skip_add:
1296
1297 } else return VOLK_VALUE_ERR;
1298
1299 VOLK_DoubleKey dbl_keys[3] = {
1300 {spok[1], spok[2]}, // po
1301 {spok[0], spok[2]}, // so
1302 {spok[0], spok[1]}, // sp
1303 };
1304
1305 // Add or remove index terms.
1306 v1.mv_size = KLEN;
1307 v2.mv_size = DBL_KLEN;
1308
1309 for (int i = 0; i < 3; i++) {
1310 MDB_dbi
1311 db1 = store->dbi[lookup_indices[i]], // s:po, p:so, o:sp
1312 db2 = store->dbi[lookup_indices[i + 3]]; // po:s, so:p, sp:o
1313
1314 v1.mv_data = spok + i;
1315 v2.mv_data = dbl_keys[i];
1316
1317 if (op == OP_REMOVE) {
1318 // Remove from 1-bound index.
1319 MDB_cursor *cur1;
1320 mdb_cursor_open(txn, store->dbi[lookup_indices[i]], &cur1);
1321 RCCK (db_rc = mdb_cursor_get (cur1, &v1, &v2, MDB_GET_BOTH));
1322 mdb_cursor_del (cur1, 0);
1323 mdb_cursor_close (cur1);
1324
1325 // Restore pointers invalidated after delete.
1326 v1.mv_data = spok + i;
1327 v2.mv_data = dbl_keys[i];
1328
1329 // Remove from 2-bound index.
1330 MDB_cursor *cur2;
1331 mdb_cursor_open(txn, store->dbi[lookup_indices[i + 3]], &cur2);
1332 RCCK (db_rc = mdb_cursor_get (cur2, &v2, &v1, MDB_GET_BOTH));
1333 mdb_cursor_del (cur2, 0);
1334 mdb_cursor_close (cur2);
1335
1336 rc = VOLK_OK;
1337
1338 } else { // OP_ADD is guaranteed.
1339 // Add to 1-bound index.
1340 log_trace ("Indexing in %s: ", db_labels[lookup_indices[i]]);
1341 log_trace (
1342 "%x: %x %x", *(size_t*)(v1.mv_data),
1343 *(size_t*)(v2.mv_data), *(size_t*)(v2.mv_data) + 1);
1344
1345 db_rc = mdb_put (txn, db1, &v1, &v2, MDB_NODUPDATA);
1346
1347 if (db_rc == MDB_SUCCESS) rc = VOLK_OK;
1348 else if (db_rc != MDB_KEYEXIST) return VOLK_DB_ERR;
1349
1350 // Add to 2-bound index.
1351 log_trace ("Indexing in %s: ", db_labels[lookup_indices[i + 3]]);
1352 log_trace (
1353 "%x %x: %x", *(size_t*)(v2.mv_data),
1354 *(size_t*)(v2.mv_data) + 1, *(size_t*)(v1.mv_data));
1355
1356 db_rc = mdb_put (txn, db2, &v2, &v1, MDB_NODUPDATA);
1357
1358 if (db_rc == MDB_SUCCESS) rc = VOLK_OK;
1359 else if (db_rc != MDB_KEYEXIST) return VOLK_DB_ERR;
1360 }
1361 }
1362
1363 return rc;
1364}
1365
1366
1367/* * * Term-specific iterators. * * */
1368
1373inline static void
1374it_next_0bound (MDBIterator *it)
1375{
1376 memcpy (it->spok, it->key.mv_data, sizeof (VOLK_TripleKey));
1377 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_NEXT_NODUP);
1378}
1379
1380
1385inline static void
1386it_next_0bound_ctx (MDBIterator *it)
1387{
1388 memcpy (it->spok, it->data.mv_data, sizeof (VOLK_TripleKey));
1389 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_NEXT_DUP);
1390}
1391
1392
1399inline static void
1400it_next_1bound (MDBIterator *it)
1401{
1402 VOLK_DoubleKey *lu_dset = it->data.mv_data;
1403
1404 it->spok[it->term_order[0]] = it->luk[0];
1405 it->spok[it->term_order[1]] = lu_dset[it->i][0];
1406 it->spok[it->term_order[2]] = lu_dset[it->i][1];
1407
1408 log_trace (
1409 "Composed triple: {%x %x %x}",
1410 it->spok[0], it->spok[1], it->spok[2]);
1411
1412 // Ensure next block within the same page is not beyond the last.
1413 if (it->i < it->data.mv_size / DBL_KLEN - 1) {
1414 it->i ++;
1415 //log_debug ("Increasing page cursor to %lu.", it->i);
1416 //log_debug ("it->rc: %d", it->rc);
1417
1418 } else {
1419 // If the last block in the page is being yielded,
1420 // move cursor to beginning of next page.
1421 it->i = 0;
1422 //log_debug ("Reset page cursor to %lu.", it->i);
1423 it->rc = mdb_cursor_get (
1424 it->cur, &it->key, &it->data, MDB_NEXT_MULTIPLE);
1425 }
1426}
1427
1428
1435inline static void
1436it_next_2bound (MDBIterator *it)
1437{
1438 VOLK_Key *lu_dset = it->data.mv_data;
1439
1440 it->spok[it->term_order[0]] = it->luk[0];
1441 it->spok[it->term_order[1]] = it->luk[1];
1442 it->spok[it->term_order[2]] = lu_dset[it->i];
1443
1444 // Ensure next block within the same page is not beyond the last.
1445 if (it->i < it->data.mv_size / KLEN - 1)
1446 it->i ++;
1447 else {
1448 // If the last block in the page is being yielded,
1449 // move cursor to beginning of next page.
1450 it->i = 0;
1451 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_NEXT_MULTIPLE);
1452 }
1453}
1454
1455
1462inline static void
1463it_next_3bound (MDBIterator *it)
1464{ it->rc = MDB_NOTFOUND; }
1465
1466
1467/* * * Term-specific lookups. * * */
1468
1469inline static VOLK_rc
1470lookup_0bound (MDBIterator *it, size_t *ct)
1471{
1472 log_debug ("Looking up 0 bound terms.");
1473
1474 // Context search looks for all values in c:spo.
1475 if (it->luc != NULL_KEY) {
1476 // Look up by given context.
1477 RCCK (it->rc = mdb_cursor_open (
1478 it->txn, it->store->dbi[IDX_C_SPO], &it->cur));
1479
1480 it->key.mv_data = &it->luc;
1481 it->key.mv_size = KLEN;
1482
1483 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET);
1484 if (ct) {
1485 if (it->rc == MDB_NOTFOUND) *ct = 0;
1486 else {
1487 RCCK (it->rc);
1488 mdb_cursor_count (it->cur, ct);
1489 }
1490 log_debug ("Found %lu triples.", *ct);
1491 }
1492 //it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_FIRST_DUP);
1493 it->iter_op_fn = it_next_0bound_ctx;
1494
1495 // No-context search looks for all keys in spo:c.
1496 } else {
1497 RCCK (it->rc = mdb_cursor_open (
1498 it->txn, it->store->dbi[IDX_SPO_C], &it->cur));
1499 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_FIRST);
1500 if (ct) {
1501 if (it->rc == MDB_NOTFOUND) *ct = 0;
1502 else {
1503 MDB_stat stat;
1504 // s:po and 1- and 2-bound indices have 1 entry per triple.
1505 mdb_stat (it->txn, it->store->dbi[IDX_S_PO], &stat);
1506
1507 *ct = stat.ms_entries;
1508 }
1509 log_debug ("Found %lu triples.", *ct);
1510 }
1511 it->iter_op_fn = it_next_0bound;
1512 }
1513 if (it->rc != MDB_NOTFOUND) RCCK (it->rc);
1514
1515 return VOLK_OK;
1516}
1517
1518
1519inline static VOLK_rc
1520lookup_1bound (uint8_t idx0, MDBIterator *it, size_t *ct)
1521{
1522 it->term_order = (const uint8_t*)lookup_ordering_1bound[idx0];
1523
1524 log_debug ("Looking up 1 bound term: %x", it->luk[0]);
1525
1526 RCCK (mdb_cursor_open (
1527 it->txn, it->store->dbi[lookup_indices[idx0]], &it->cur));
1528
1529 it->key.mv_data = it->luk;
1530 it->key.mv_size = KLEN;
1531
1532 if (ct) {
1533 // If a context is specified, the only way to count triples matching
1534 // the context is to loop over them.
1535 if (it->luc != NULL_KEY) {
1536 log_debug ("Counting in context: %x", it->luc);
1537 MDBIterator *ct_it;
1538 MALLOC_GUARD (ct_it, VOLK_MEM_ERR);
1539 /*
1540 memcpy (ct_it, it, sizeof (*ct_it));
1541 */
1542
1543 ct_it->store = it->store;
1544 ct_it->txn = it->txn;
1545 ct_it->ctx_cur = it->ctx_cur;
1546 ct_it->key = it->key;
1547 ct_it->data = it->data;
1548 ct_it->ck = NULL;
1549 ct_it->luk[0] = it->luk[0];
1550 ct_it->luc = it->luc;
1551 ct_it->i = 0;
1552
1553 PRCCK (lookup_1bound (idx0, ct_it, NULL));
1554
1555 VOLK_rc db_rc;
1556 while (VOLK_END != (db_rc = mdbiter_next_key (ct_it))) {
1557 PRCCK (db_rc);
1558 (*ct)++;
1559 }
1560
1561 // Free the counter iterator without freeing the shared txn.
1562 if (ct_it->cur) mdb_cursor_close (ct_it->cur);
1563 free (ct_it->ck);
1564 free (ct_it);
1565
1566 } else {
1567 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET);
1568 if (it->rc == MDB_SUCCESS) mdb_cursor_count (it->cur, ct);
1569 }
1570 log_debug ("Found %lu triples.", *ct);
1571 }
1572
1573 it->i = 0;
1574 it->iter_op_fn = it_next_1bound;
1575
1576 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET);
1577 if (it->rc == MDB_SUCCESS)
1578 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_GET_MULTIPLE);
1579 if (it->rc != MDB_NOTFOUND) RCCK (it->rc);
1580
1581 return VOLK_OK;
1582}
1583
1584
1585inline static VOLK_rc
1586lookup_2bound(uint8_t idx0, uint8_t idx1, MDBIterator *it, size_t *ct)
1587{
1588 uint8_t luk1_offset, luk2_offset;
1589 MDB_dbi dbi = 0;
1590
1591 // Establish lookup ordering with some awkward offset math.
1592 for (int i = 0; i < 3; i++) {
1593 if (
1594 (
1595 idx0 == lookup_ordering_2bound[i][0] &&
1596 idx1 == lookup_ordering_2bound[i][1]
1597 ) || (
1598 idx0 == lookup_ordering_2bound[i][1] &&
1599 idx1 == lookup_ordering_2bound[i][0]
1600 )
1601 ) {
1602 it->term_order = (const uint8_t*)lookup_ordering_2bound[i];
1603 if (it->term_order[0] == idx0) {
1604 luk1_offset = 0;
1605 luk2_offset = 1;
1606 } else {
1607 luk1_offset = 1;
1608 luk2_offset = 0;
1609 }
1610 dbi = it->store->dbi[lookup_indices[i + 3]];
1611 log_debug (
1612 "Looking up 2 bound in %s",
1613 db_labels[lookup_indices[i + 3]]);
1614
1615 break;
1616 }
1617 }
1618
1619 if (dbi == 0) {
1620 log_error (
1621 "Values %d and %d not found in lookup keys.",
1622 idx0, idx1);
1623 return VOLK_VALUE_ERR;
1624 }
1625
1626 // Compose term keys in lookup key.
1627 VOLK_DoubleKey luk;
1628 luk[luk1_offset] = it->luk[0];
1629 luk[luk2_offset] = it->luk[1];
1630
1631 it->key.mv_data = luk;
1632 it->key.mv_size = DBL_KLEN;
1633
1634 mdb_cursor_open (it->txn, dbi, &it->cur);
1635 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET);
1636
1637 if (ct) {
1638 // If a context is specified, the only way to count triples matching
1639 // the context is to loop over them.
1640 if (it->luc != NULL_KEY) {
1641 MDBIterator *ct_it;
1642 MALLOC_GUARD (ct_it, VOLK_MEM_ERR);
1643
1644 ct_it->store = it->store;
1645 ct_it->txn = it->txn;
1646 ct_it->ctx_cur = it->ctx_cur;
1647 ct_it->ck = NULL;
1648 ct_it->luk[0] = it->luk[0];
1649 ct_it->luk[1] = it->luk[1];
1650 ct_it->luc = it->luc;
1651 ct_it->i = 0;
1652
1653 lookup_2bound (idx0, idx1, ct_it, NULL);
1654
1655 while (mdbiter_next_key (ct_it) != VOLK_END) (*ct) ++;
1656
1657 // Free the counter iterator without freeing the shared txn.
1658 if (ct_it->cur) mdb_cursor_close (ct_it->cur);
1659 free (ct_it->ck);
1660 free (ct_it);
1661
1662 } else {
1663 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET);
1664 if (it->rc == MDB_SUCCESS) mdb_cursor_count (it->cur, ct);
1665 }
1666 log_debug ("Found %lu triples.", *ct);
1667 }
1668
1669 it->i = 0;
1670 it->iter_op_fn = it_next_2bound;
1671
1672 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_SET);
1673 if (it->rc == MDB_SUCCESS)
1674 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_GET_MULTIPLE);
1675 if (it->rc != MDB_NOTFOUND) RCCK (it->rc);
1676
1677 return VOLK_OK;
1678}
1679
1680
1681inline static VOLK_rc
1682lookup_3bound (MDBIterator *it, size_t *ct)
1683{
1684 log_debug (
1685 "Looking up 3 bound: {%x, %x, %x}",
1686 it->luk[0], it->luk[1], it->luk[2]);
1687
1688 it->key.mv_data = it->luk;
1689
1690 if (it->luc != NULL_KEY) {
1691 it->rc = mdb_cursor_open (
1692 it->txn, it->store->dbi[IDX_SPO_C], &it->cur);
1693
1694 it->key.mv_size = TRP_KLEN;
1695 it->data.mv_data = &it->luc;
1696 it->data.mv_size = KLEN;
1697
1698 } else {
1699 it->rc = mdb_cursor_open (it->txn, it->store->dbi[IDX_S_PO], &it->cur);
1700
1701 it->key.mv_size = KLEN;
1702 it->data.mv_data = it->luk + 1;
1703 it->data.mv_size = DBL_KLEN;
1704 }
1705
1706 it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_GET_BOTH);
1707 if (it->rc != MDB_NOTFOUND) RCCK (it->rc);
1708
1709 mdb_cursor_close (it->cur);
1710 it->cur = NULL;
1711
1712 if (ct && it->rc == MDB_SUCCESS) *ct = 1;
1713
1714 it->iter_op_fn = it_next_3bound;
1715 memcpy (it->spok, it->luk, sizeof (VOLK_TripleKey));
1716
1717 return VOLK_OK;
1718}
1719
1720
#define STRACE
Definition core.h:59
#define UNLIKELY(x)
Definition core.h:65
#define NULL_TRP
"NULL" triple, a value that is never user-provided.
Definition core.h:92
#define DBL_KLEN
Definition core.h:84
#define KLEN
Definition core.h:83
#define TRP_KLEN
Definition core.h:85
bool VOLK_env_is_init
Whether the environment is initialized.
Definition core.c:11
#define MALLOC_GUARD(var, rc)
Allocate one pointer with malloc and return rc if it fails.
Definition core.h:435
#define RCNL(exp)
Return NULL if exp returns a nonzero value.
Definition core.h:405
#define NLCHECK(exp, marker)
Log error and jump to marker if exp is NULL.
Definition core.h:363
#define PRCNL(exp)
Return NULL if exp returns a negative value (=error).
Definition core.h:416
#define CHECK(exp, marker)
Jump to marker if exp does not return VOLK_OK.
Definition core.h:337
#define NLNL(exp)
Log error and return NULL if exp is NULL.
Definition core.h:427
#define CALLOC_GUARD(var, rc)
Allocate one pointer with calloc and return rc if it fails.
Definition core.h:441
#define log_debug(...)
Definition core.h:320
#define RCCK(exp)
Return exp return value if it is of VOLK_rc type and nonzero.
Definition core.h:372
#define NLRCCK(exp, _rc)
Return rc return code if exp is NULL.
Definition core.h:396
#define LOG_RC(rc)
Log an error or warning for return codes that are not VOLK_OK.
Definition core.h:331
#define PRCCK(exp)
Return exp return value if it is of VOLK_rc type and negative (=error).
Definition core.h:384
#define log_trace(...)
Definition core.h:322
VOLK_rc rm_r(const char *path)
Remove a directory recursively (POSIX compatible).
Definition core.c:124
VOLK_rc mkdir_p(const char *_path, mode_t mode)
Make recursive directories.
Definition core.c:50
#define VOLK_VALUE_ERR
An invalid input value was provided.
Definition core.h:167
#define VOLK_MEM_ERR
Memory allocation error.
Definition core.h:182
#define VOLK_NORESULT
No result yielded.
Definition core.h:138
#define VOLK_DB_ERR
Low-level database error.
Definition core.h:173
#define VOLK_CONFLICT
Conflict warning.
Definition core.h:158
#define VOLK_END
Loop end.
Definition core.h:145
#define VOLK_OK
Generic success return code.
Definition core.h:121
#define VOLK_NOACTION
No action taken.
Definition core.h:131
int VOLK_rc
Return code.
Definition core.h:117
#define VOLK_ENV_ERR
Error while handling environment setup; or environment not initialized.
Definition core.h:196
#define VOLK_TXN_ERR
Error handling a store transaction.
Definition core.h:170
const char * VOLK_strerror(VOLK_rc rc)
Return an error message for a return code.
Definition core.c:196
#define NULL_KEY
"NULL" key, a value that is never user-provided.
Definition buffer.h:15
VOLK_Key VOLK_buffer_hash(const VOLK_Buffer *buf)
Hash a buffer.
Definition buffer.h:175
VOLK_Buffer * VOLK_btriple_pos(const VOLK_BufferTriple *trp, VOLK_TriplePos n)
Get serialized triple by term position.
Definition buffer.h:297
VOLK_Buffer * VOLK_default_ctx_buf
Serialized default context.
Definition buffer.c:5
#define BUF_DUMMY
Dummy buffer to be used with VOLK_buffer_init.
Definition buffer.h:154
@ VOLK_BUF_BORROWED
Definition buffer.h:28
size_t VOLK_Key
Term key, i.e., hash of a serialized term.
Definition core.h:270
VOLK_Key VOLK_TripleKey[3]
Array of three VOLK_Key values, representing a triple.
Definition core.h:274
VOLK_Key VOLK_DoubleKey[2]
Array of two VOLK_Key values.
Definition core.h:272
VOLK_StoreFlags
Store flags passed to various operations.
@ VOLK_STORE_TXN_RO
Start a read-only transaction.
@ VOLK_STORE_TXN
Supports transaction handling.
@ VOLK_STORE_COW
Copy on write.
@ VOLK_STORE_IDX
Store is fully SPO(C)-indexed.
@ VOLK_STORE_PERM
@ VOLK_STORE_CTX
VOLK_Buffer ** mdbstore_ctx_list(void *h, void *th)
Definition store_mdb.c:1156
StoreFlags
Store state flags.
Definition store_mdb.c:31
@ LSSTORE_OPEN
Env is open.
Definition store_mdb.c:32
#define MAIN_TABLE
Definition store_mdb.c:108
char DbLabel[8]
Definition store_mdb.c:27
#define ENV_FILE_MODE
Definition store_mdb.c:21
#define DEFAULT_MAPSIZE
Definition store_mdb.c:15
#define ENV_DIR_MODE
Definition store_mdb.c:20
const VOLK_StoreInt mdbstore_int
MDB store interface.
Definition store_mdb.c:1201
char * mdbstore_id(const void *h)
Definition store_mdb.c:437
#define LOOKUP_TABLE
Definition store_mdb.c:116
void(* iter_op_fn_t)(MDBIterator *it)
Iterator operation.
Definition store_mdb.c:69
#define N_DB
Definition store_mdb.c:6
IterFlags
Iterator state flags.
Definition store_mdb.c:36
@ ITER_OPEN_TXN
A transaction is open.
Definition store_mdb.c:37
StoreOp
Definition store_mdb.c:48
@ OP_ADD
Definition store_mdb.c:49
@ OP_REMOVE
Definition store_mdb.c:50
DBIdx
Definition store_mdb.c:140
@ IDX_SPO_C
Definition store_mdb.c:142
@ IDX_S_PO
Definition store_mdb.c:143
@ IDX_T_ST
Definition store_mdb.c:142
@ IDX_C_SPO
Definition store_mdb.c:143
LMDB graph store backend.
#define VOLK_MDB_STORE_URN
Default MDB store identifier and location.
Definition store_mdb.h:38
Triple iterator.
Definition store_mdb.c:73
MDB_cursor * ctx_cur
MDB c:spo index cursor.
Definition store_mdb.c:78
VOLK_TripleKey spok
Triple to be populated with match.
Definition store_mdb.c:81
VOLK_Key luk[3]
0รท3 lookup keys.
Definition store_mdb.c:88
MDB_cursor * cur
MDB cursor.
Definition store_mdb.c:77
const uint8_t * term_order
Term order used in 1-2bound look-ups.
Definition store_mdb.c:87
MDBStore * store
MDB store handle.
Definition store_mdb.c:74
IterFlags flags
Iterator flags.
Definition store_mdb.c:75
MDB_txn * txn
MDB transaction.
Definition store_mdb.c:76
VOLK_Key luc
Ctx key to filter by. May be NULL_KEY.
Definition store_mdb.c:89
size_t i
Internal counter for paged lookups.
Definition store_mdb.c:90
iter_op_fn_t iter_op_fn
Function used to look up next match.
Definition store_mdb.c:86
MDB_val key
Internal data handler.
Definition store_mdb.c:79
size_t ct
Definition store_mdb.c:91
int rc
MDB_* return code for the next result.
Definition store_mdb.c:93
MDB_val data
Internal data handler.
Definition store_mdb.c:80
VOLK_Key * ck
Definition store_mdb.c:82
StoreFlags flags
Store state flags.
Definition store_mdb.c:56
MDB_dbi dbi[9]
DB handles. Refer to DbIdx enum.
Definition store_mdb.c:55
MDB_env * env
Environment handle.
Definition store_mdb.c:54
Triple of byte buffers.
Definition buffer.h:60
VOLK_Buffer * o
Definition buffer.h:63
VOLK_Buffer * s
Definition buffer.h:61
VOLK_Buffer * p
Definition buffer.h:62
General-purpose data buffer.
Definition buffer.h:47
VOLK_BufferFlag flags
Definition buffer.h:50
unsigned char * addr
Definition buffer.h:48
size_t size
Definition buffer.h:49
Store interface.