#include #include #include #include "store_mdb.h" /* * TODO At the moment up to 64-bit key / hash values are allowed. Later on, * 128-bit keys should be allowed by compile options, and that will no longer * be compatible with integer keys and data. When 128-bit keys are supported, * integer keys should remain available for code compiled with 64-bit keys. */ #define INT_KEY_MASK MDB_INTEGERKEY #define INT_DUP_KEY_MASK MDB_DUPSORT | MDB_DUPFIXED | MDB_INTEGERKEY #define INT_DUPDATA_MASK MDB_DUPSORT | MDB_DUPFIXED | MDB_INTEGERDUP /** * Number of DBs defined. */ #define N_DB 12 #define DEFAULT_ENV_PATH "./mdb_store" #define ENV_DIR_MODE 0750 #define ENV_FILE_MODE 0640 typedef char DbLabel[8]; // TODO Most of these are no longer used. Clean up. typedef enum { LSSTORE_INIT = 1, // Is the store environment set up on disk? LSSTORE_OPEN = 3, // Is the environment open? Assumes init is set. LSSTORE_DB_CREATED = 5, // Are DBs created? Assumes init is set. } StoreState; typedef enum { OP_ADD, OP_REMOVE, } StoreOp; struct MDBStore { MDB_env * env; // Environment handle. MDB_txn * txn; // Current transaction. If RW, it may have // nested transactions. MDB_dbi dbi[N_DB]; // DB handles. Refer to DbIdx enum. LSUP_Buffer * default_ctx;// Default context as a serialized URI. StoreState state; // Store state (initialized, open etc.) }; /** @brief Common match callback arguments. */ struct MatchArgs { LSUP_Key luks[2]; uint8_t idx0, idx1; size_t *ct; mdb_store_match_fn_t callback_fn; void *ctx; }; /** * Main DBs. These are the master information containers. * * Data columns are: identifier prefix, DB label, flags. */ #define MAIN_TABLE \ ENTRY( T_ST, "t:st", INT_KEY_MASK ) /* Key to ser. term */ \ ENTRY( SPO_C, "spo:c", INT_DUPDATA_MASK) /* Triple to context */ \ ENTRY( C_, "c:", INT_KEY_MASK ) /* Track empty contexts */\ ENTRY( PFX_NS, "pfx:ns", 0 ) /* Prefix to NS */ \ /** * Lookup DBs. These are indices and may be destroyed and rebuilt. */ #define LOOKUP_TABLE \ ENTRY( S_PO, "s:po", INT_DUP_KEY_MASK) /* 1-bound lookup */ \ ENTRY( P_SO, "p:so", INT_DUP_KEY_MASK) /* 1-bound lookup */ \ ENTRY( O_SP, "o:sp", INT_DUP_KEY_MASK) /* 1-bound lookup */ \ ENTRY( PO_S, "po:s", INT_DUPDATA_MASK) /* 2-bound lookup */ \ ENTRY( SO_P, "so:p", INT_DUPDATA_MASK) /* 2-bound lookup */ \ ENTRY( SP_O, "sp:o", INT_DUPDATA_MASK) /* 2-bound lookup */ \ ENTRY( C_SPO, "c:spo", INT_DUP_KEY_MASK) /* Context lookup */ \ ENTRY( NS_PFX, "ns:pfx", 0 ) /* NS to prefix */ \ /** * DB labels. They are prefixed with DB_ */ #define ENTRY(a, b, c) static const DbLabel DB_##a = b; MAIN_TABLE LOOKUP_TABLE #undef ENTRY /** * Numeric index of each DB. Prefixed with IDX_ * * These index numbers are referred to in all the arrays defeined below. They * are independent from the LMDB dbi values which are considered opaque here. */ typedef enum { #define ENTRY(a, b, c) IDX_##a, MAIN_TABLE LOOKUP_TABLE #undef ENTRY } DBIdx; /** * DB labels. */ static const char *db_labels[N_DB] = { #define ENTRY(a, b, c) DB_##a, MAIN_TABLE LOOKUP_TABLE #undef ENTRY }; /** * DB flags. These are aligned with the dbi_labels index. */ static const unsigned int db_flags[N_DB] = { #define ENTRY(a, b, c) c, MAIN_TABLE LOOKUP_TABLE #undef ENTRY }; /** * 1-bound and 2-bound lookup indices. * * N.B. Only the first 6 (1-bound and 2-bound term lookup) are used. * The others are added just because they belong logically to the lookup table. */ static DBIdx lookup_indices[9] = { #define ENTRY(a, b, c) IDX_##a, LOOKUP_TABLE #undef ENTRY }; /** * Order in which keys are looked up if two terms are bound. * The indices with the smallest average number of values per key should be * looked up first. * * 0 = s:po * 1 = p:so * 2 = o:sp */ static const uint8_t lookup_rank[3] = {0, 2, 1}; static const uint8_t lookup_ordering_1bound[3][3] = { {0, 1, 2}, // s:po {1, 0, 2}, // p:so {2, 0, 1}, // o:sp }; static const uint8_t lookup_ordering_2bound[3][3] = { {1, 2, 0}, // po:s {0, 2, 1}, // so:p {0, 1, 2}, // sp:o }; /** * Static prototypes. */ static int dbi_init(LSUP_MDBStore *store); static int index_triple( LSUP_MDBStore *store, StoreOp op, LSUP_TripleKey spok, LSUP_Key ck); inline static LSUP_rc match_callback_0bound( struct MDBStore *store, struct MatchArgs *args); inline static LSUP_rc match_callback_1bound( struct MDBStore *store, struct MatchArgs *args); inline static LSUP_rc match_callback_2bound( struct MDBStore *store, struct MatchArgs *args); inline static LSUP_rc match_callback_3bound( struct MDBStore *store, struct MatchArgs *args); /* TODO inline static int check_txn_open(MDB_txn *txn, bool write); */ /* TODO static int unlink_cb( const char *fpath, const struct stat *sb, int typeflag, struct FTW *ftwbuf); static int rmrf(char *path); */ /** * API. */ LSUP_rc LSUP_store_setup(char **path/*, bool clear*/) // TODO clear { int rc; // Set environment path. if (path == NULL || (*path = getenv("LSUP_STORE_PATH")) == NULL) { printf( "WARNING: `LSUP_STORE_PATH' environment variable is not set. " "A local directory will be used to store the LSUP data."); *path = DEFAULT_ENV_PATH; } // Verify that a writable directory exists or can be created. struct stat path_stat; rc = stat(*path, &path_stat); if (rc == ENOENT) { if (mkdir(*path, ENV_DIR_MODE) != 0) abort(); } else if (!S_ISDIR(path_stat.st_mode)) { printf("%s is not a valid directory.\n", path); abort(); /* TODO clear } else { if (clear) { rmrf(*path); if (mkdir(*path, ENV_DIR_MODE) != 0) abort(); } */ } // Open a temporary environment and txn to create the DBs. MDB_env *env; mdb_env_create(&env); mdb_env_open(env, *path, 0, ENV_FILE_MODE); MDB_txn *txn; mdb_txn_begin(env, NULL, 0, &txn); for (int i = 0; i < N_DB; i++) { MDB_dbi dbi; mdb_dbi_open(txn, db_labels[i], db_flags[i] | MDB_CREATE, &dbi); mdb_dbi_close(env, dbi); } mdb_txn_commit(txn); mdb_env_close(env); return rc; } LSUP_rc LSUP_store_open( LSUP_MDBStore *store, const char *path, LSUP_Buffer *default_ctx) { if(store->state & LSSTORE_OPEN) return LSUP_NOACTION; if (default_ctx == NULL) store->default_ctx = NULL; else { CRITICAL(store->default_ctx = malloc(sizeof(LSUP_Buffer))); LSUP_buffer_copy(store->default_ctx, default_ctx); } // Set map size. size_t mapsize; char *env_mapsize = getenv("LSUP_MDB_MAPSIZE"); if (env_mapsize == NULL) { mapsize = 0x10000000000; // 1Tb } else { sscanf(env_mapsize, "%lu", &mapsize); } mdb_env_set_mapsize(store->env, mapsize); mdb_env_set_maxdbs(store->env, N_DB); int rc = mdb_env_open(store->env, path, 0, ENV_FILE_MODE); // Assign DB handles to store->dbi. MDB_txn *txn; mdb_txn_begin(store->env, NULL, 0, &txn); for (int i = 0; i < N_DB; i++) { mdb_dbi_open( txn, db_labels[i], db_flags[i], store->dbi + i); } mdb_txn_commit(txn); store->state = LSSTORE_OPEN; return rc; } LSUP_rc LSUP_store_stats(LSUP_MDBStore *store) { // TODO MDB_stat env_stat, db_stats[N_DB]; return 0; } size_t LSUP_store_size(LSUP_MDBStore *store) { if(!(store->state & LSSTORE_INIT)) return 0; MDB_stat stat; mdb_stat(store->txn, store->dbi[IDX_SPO_C], &stat); return stat.ms_entries; } LSUP_rc LSUP_store_add( LSUP_MDBStore *store, const LSUP_Buffer *sc, const LSUP_SerTerm **data, const size_t data_size) { MDB_val key_v, data_v; bool txn_pending = false; if (!store->txn) { mdb_txn_begin(store->env, NULL, MDB_RDONLY, &store->txn); txn_pending = true; } // Take care of context first. // Serialize and hash. LSUP_Key ck = NULL_KEY; if (store->default_ctx != NULL) { if (sc == NULL) sc = store->default_ctx; ck = LSUP_sterm_to_key(sc); // Insert t:st for context. TRACE("Adding context: %s", sc); key_v.mv_data = &ck; key_v.mv_size = KLEN; data_v.mv_data = sc->addr; data_v.mv_size = sc->size; mdb_put( store->txn, store->dbi[IDX_T_ST], &key_v, &data_v, MDB_NOOVERWRITE); } int rc = LSUP_NOACTION; for (size_t i = 0; i < data_size && rc == LSUP_OK; i++) { const LSUP_SerTerm *sspo = data[i]; /* LSUP_SerTerm sspo[3]; CHECK(LSUP_term_serialize(spo->s, sspo), _add_free_sterm1); CHECK(LSUP_term_serialize(spo->p, sspo + 1), _add_free_sterm2); CHECK(LSUP_term_serialize(spo->s, sspo + 2), _add_free_sterms); */ LSUP_TripleKey spok = NULL_TRP; // Add triple. TRACE("Inserting spok: {%lx, %lx, %lx}", spok[0], spok[1], spok[2]); int put_rc[2] = {LSUP_OK, MDB_KEYEXIST}; // Insert t:st for s, p, o for (int j = 0; j < 3; j++) { spok[i] = LSUP_sterm_to_key(sspo + i); key_v.mv_data = spok + i; key_v.mv_size = KLEN; data_v.mv_data = (sspo + i)->addr; data_v.mv_size = (sspo + i)->size; MCHECK( mdb_put( store->txn, store->dbi[IDX_T_ST], &key_v, &data_v, MDB_NOOVERWRITE), put_rc, _add_close_txn); } // Insert spo:c key_v.mv_data = spok; key_v.mv_size = TRP_KLEN; // In triple mode, data is empty. data_v.mv_data = ck == NULL_KEY ? NULL : &ck; data_v.mv_size = ck == NULL_KEY ? 0 : KLEN; MCHECK( mdb_put( store->txn, store->dbi[IDX_SPO_C], &key_v, &data_v, MDB_NODUPDATA), put_rc, _add_close_txn); // Index. PCHECK(index_triple(store, OP_ADD, spok, ck), _add_close_txn); // Free serialized terms. /* _add_free_sterms: LSUP_buffer_done(sspo + 2); _add_free_sterm2: LSUP_buffer_done(sspo + 1); _add_free_sterm1: LSUP_buffer_done(sspo); */ } _add_close_txn: // Only return commit rc if it fails. if (txn_pending) { if (rc >= LSUP_OK) { int txn_rc; if((txn_rc = mdb_txn_commit(store->txn)) != MDB_SUCCESS) { mdb_txn_abort(store->txn); rc = txn_rc; } } else mdb_txn_abort(store->txn); store->txn = NULL; } return rc; } LSUP_Key LSUP_store_sterm_to_key( LSUP_MDBStore *store, const LSUP_SerTerm *sterm) { // TODO this will be replaced by a lookup when 128-bit hash is introduced. return LSUP_sterm_to_key(sterm); } /* LSUP_Key LSUP_store_get_key( LSUP_MDBStore *store, const LSUP_SerTerm *sterm) { } */ LSUP_rc LSUP_store_key_to_sterm( LSUP_MDBStore *store, const LSUP_Key key, LSUP_SerTerm *sterm) { LSUP_rc rc = LSUP_NORESULT; MDB_txn *txn; mdb_txn_begin(store->env, NULL, MDB_RDONLY, &txn); MDB_val key_v, data_v; key_v.mv_data = (void*)&key; key_v.mv_size = KLEN; int mdb_rc = mdb_get(txn, store->dbi[IDX_T_ST], &key_v, &data_v); if (mdb_rc == MDB_SUCCESS) { sterm->addr = data_v.mv_data; sterm->size = data_v.mv_size; rc = LSUP_OK; } else if (UNLIKELY(mdb_rc != MDB_NOTFOUND)) rc = LSUP_ERROR; mdb_txn_abort(txn); return rc; } LSUP_rc LSUP_store_match_callback( LSUP_MDBStore *store, LSUP_SerTerm sspoc[], size_t *ct, mdb_store_match_fn_t callback_fn, void *ctx) { LSUP_TripleKey spok = { LSUP_sterm_to_key(sspoc), LSUP_sterm_to_key(sspoc + 1), LSUP_sterm_to_key(sspoc + 2), }; LSUP_Key ck = store->default_ctx ? LSUP_sterm_to_key(sspoc + 3) : NULL_KEY; struct MatchArgs args_s; struct MatchArgs *args = &args_s; args->ct = ct; args->callback_fn = callback_fn; args->ctx = ctx; // s p o (all terms bound) if (spok[0] != NULL_KEY && spok[1] != NULL_KEY && spok[2] != NULL_KEY) { return match_callback_3bound(store, args); } else if (spok[0] != NULL_KEY) { args->luks[0] = spok[0]; args->idx0 = 0; if (spok[1] != NULL_KEY) { // s p ? args->luks[1] = spok[1]; args->idx1 = 1; return match_callback_2bound(store, args); } else if (spok[2] != NULL_KEY) { // s ? o args->luks[1] = spok[2]; args->idx1 = 2; return match_callback_2bound(store, args); } else { // s ? ? return match_callback_1bound(store, args); } } else if (spok[1] != NULL_KEY) { args->luks[0] = spok[1]; if (spok[2] != NULL_KEY) { // ? p o args->luks[1] = spok[2]; args->idx1 = 2; return match_callback_2bound(store, args); } else { // ? p ? args->idx0 = 1; return match_callback_1bound(store, args); } } else if (spok[2] != NULL_KEY) { // ? ? o args->luks[0] = spok[2]; args->idx0 = 2; return match_callback_1bound(store, args); } else { // ? ? ? (all terms unbound) return match_callback_0bound(store, args); } } LSUP_rc LSUP_store_remove( LSUP_MDBStore *store, const LSUP_Buffer *sc, LSUP_TripleKey data[], size_t data_size) { LSUP_rc rc = LSUP_NOACTION; LSUP_Key ck = NULL_KEY; if (store->default_ctx != NULL) { if (sc == NULL) sc = store->default_ctx; ck = LSUP_sterm_to_key(sc); } MDB_txn *txn; mdb_txn_begin(store->env, NULL, 0, &txn); MDB_cursor *dcur, *icur; mdb_cursor_open(txn, store->dbi[IDX_SPO_C], &dcur); mdb_cursor_open(txn, store->dbi[IDX_C_SPO], &icur); MDB_val spok_v, ck_v; LSUP_TripleKey spok_cur; spok_v.mv_size = TRP_KLEN; ck_v.mv_size = KLEN; for(size_t i = 0; i < data_size; i++) { spok_v.mv_data = data + i; rc = mdb_cursor_get(dcur, &spok_v, &ck_v, MDB_GET_BOTH); if (rc == MDB_NOTFOUND) continue; if (UNLIKELY(rc != MDB_SUCCESS)) goto _remove_abort; // Delete spo:c entry. mdb_cursor_del(dcur, 0); // Restore ck address after each delete. ck_v.mv_data = &ck; // Delete c::spo entry. rc = mdb_cursor_get(icur, &ck_v, &spok_v, MDB_GET_BOTH); if (rc == MDB_NOTFOUND) continue; if (UNLIKELY(rc != MDB_SUCCESS)) goto _remove_abort; mdb_cursor_del(icur, 0); spok_v.mv_data = data + i; // If there are no more contexts associated with this triple, // remove from indices. rc = mdb_cursor_get(dcur, &spok_v, NULL, MDB_SET); if (rc == MDB_SUCCESS) continue; if (UNLIKELY(rc != MDB_NOTFOUND)) goto _remove_abort; index_triple(store, OP_REMOVE, data[i], ck); } if(UNLIKELY(mdb_txn_commit(txn) != MDB_SUCCESS)) { rc = LSUP_TXN_ERR; goto _remove_abort; } return rc; _remove_abort: mdb_txn_abort(txn); return rc; } void LSUP_store_done(LSUP_MDBStore *store) { if (store->state & LSSTORE_OPEN) { TRACE(STR, "Closing MDB env.\n"); mdb_env_close(store->env); } if (store->default_ctx != NULL) LSUP_buffer_done(store->default_ctx); } /* * * Static functions. * * */ /* TODO static int unlink_cb( const char *fpath, const struct stat *sb, int typeflag, struct FTW *ftwbuf) { int rv = remove(fpath); if (rv) perror(fpath); return rv; } static int rmrf(char *path) { return nftw(path, unlink_cb, 64, FTW_DEPTH | FTW_PHYS); } */ /** * @brief Open and allocate DB handles in an array. * * @param bool create [in]: If true, the DBs are created. This is only needed * on bootstrap. */ static LSUP_rc dbi_init(LSUP_MDBStore *store) { bool db_created = store->state & LSSTORE_DB_CREATED; MDB_txn *txn; unsigned int txn_flags = db_created ? MDB_RDONLY : 0; unsigned int create_flag = db_created ? 0 : MDB_CREATE; mdb_txn_begin(store->env, NULL, txn_flags, &txn); for (int i = 0; i < N_DB; i++) { mdb_dbi_open( txn, db_labels[i], db_flags[i] | create_flag, store->dbi + i); } store->state |= LSSTORE_DB_CREATED; mdb_txn_commit(txn); return 0; } /* TODO inline static int check_txn_open(MDB_txn *txn, bool write) { if (txn == NULL) { mdb_txn_begin(LSUP_mdbenv, NULL, write ? 0 : MDB_RDONLY, &txn); return LSUP_OK; } return LSUP_NOACTION; } */ static LSUP_rc index_triple( LSUP_MDBStore *store, StoreOp op, LSUP_TripleKey spok, LSUP_Key ck) { int rc = LSUP_NOACTION; MDB_val v1, v2; // Index c:spo. if (op == OP_REMOVE) { if (ck != NULL_KEY) { MDB_cursor *cur; v1.mv_data = &ck; v1.mv_size = KLEN; v2.mv_data = spok; v2.mv_size = TRP_KLEN; mdb_cursor_open(store->txn, store->dbi[IDX_C_SPO], &cur); rc = mdb_cursor_get(cur, &v1, &v2, MDB_GET_BOTH); if(rc == MDB_SUCCESS) mdb_cursor_del(cur, 0); mdb_cursor_close(cur); } } else if (op == OP_ADD) { if (ck != NULL_KEY) { v1.mv_data = &ck; v1.mv_size = KLEN; v2.mv_data = spok; v2.mv_size = TRP_KLEN; mdb_put( store->txn, store->dbi[IDX_C_SPO], &v1, &v2, MDB_NODUPDATA); } } else return LSUP_VALUE_ERR; LSUP_DoubleKey dbl_keys[3] = { {spok[1], spok[2]}, // po {spok[0], spok[2]}, // so {spok[0], spok[1]}, // sp }; // Add terms to index. v1.mv_size = KLEN; v2.mv_size = DBL_KLEN; for (int i = 0; i < 3; i++) { MDB_dbi db1 = lookup_indices[i]; // s:po, p:so, o:sp MDB_dbi db2 = lookup_indices[i + 3]; // po:s, so:p, sp:o v1.mv_data = spok + i; v2.mv_data = dbl_keys[i]; if (op == OP_REMOVE) { MDB_cursor *cur1, *cur2; mdb_cursor_open(store->txn, store->dbi[lookup_indices[i]], &cur1); rc = mdb_cursor_get(cur1, &v1, &v2, MDB_GET_BOTH); if (rc == MDB_SUCCESS) mdb_cursor_del(cur1, 0); mdb_cursor_close(cur1); // Restore pointers invalidated after delete. v1.mv_data = spok + i; v2.mv_data = dbl_keys[i]; mdb_cursor_open( store->txn, store->dbi[lookup_indices[i + 3]], &cur2); rc = mdb_cursor_get(cur2, &v2, &v1, MDB_GET_BOTH); if (rc == MDB_SUCCESS) mdb_cursor_del(cur2, 0); mdb_cursor_close(cur2); } else { // OP_ADD is guaranteed. mdb_put(store->txn, db1, &v1, &v2, MDB_NODUPDATA); mdb_put(store->txn, db2, &v2, &v1, MDB_NODUPDATA); } } return rc; } /* * * Match callbacks. * * */ inline static LSUP_rc match_callback_0bound(struct MDBStore *store, struct MatchArgs *args) { int rc = LSUP_NORESULT; MDB_txn *txn; if(store->txn) txn = store->txn; else mdb_txn_begin(store->env, NULL, MDB_RDONLY, &txn); MDB_val key_v; if(args->ct) { MDB_stat stat; mdb_stat(store->txn, store->dbi[IDX_SPO_C], &stat); *args->ct = stat.ms_entries; } MDB_cursor *cur; mdb_cursor_open(txn, store->dbi[IDX_SPO_C], &cur); if(args->callback_fn) { rc = mdb_cursor_get(cur, &key_v, NULL, MDB_FIRST); while (rc != MDB_NOTFOUND) { LSUP_TripleKey spok; rc = args->callback_fn(spok, args->ctx); if (rc < 0) goto _match0b_abort; } } _match0b_abort: mdb_cursor_close(cur); if (txn != store->txn) mdb_txn_abort(txn); return rc; } inline static LSUP_rc match_callback_1bound(struct MDBStore *store, struct MatchArgs *args) { int rc = LSUP_NORESULT; const uint8_t *term_order = lookup_ordering_1bound[args->idx0]; MDB_txn *txn; if(store->txn) txn = store->txn; else mdb_txn_begin(store->env, NULL, MDB_RDONLY, &txn); MDB_cursor *cur; mdb_cursor_open(txn, store->dbi[args->idx0], &cur); MDB_val key_v, data_v; key_v.mv_data = &args->luks; key_v.mv_size = KLEN; if(args->ct) { mdb_cursor_get(cur, &key_v, &data_v, MDB_SET); mdb_cursor_count(cur, args->ct); } if(args->callback_fn) { rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_SET); if (rc == MDB_SUCCESS) rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_GET_MULTIPLE); while (rc != MDB_NOTFOUND) { LSUP_Key **lu_dset = data_v.mv_data; for (int i = 0; i < data_v.mv_size / DBL_KLEN; i++) { // Build triple key from lookup key and result keys. LSUP_TripleKey spok; spok[term_order[0]] = args->luks[0]; spok[term_order[1]] = lu_dset[i][0]; spok[term_order[2]] = lu_dset[i][1]; rc = args->callback_fn(spok, args->ctx); if (rc < 0) goto _match1b_abort; } rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_NEXT_MULTIPLE); } } _match1b_abort: mdb_cursor_close(cur); if (txn != store->txn) mdb_txn_abort(txn); return rc; } inline static LSUP_rc match_callback_2bound(struct MDBStore *store, struct MatchArgs *args) { int rc = LSUP_NORESULT; uint8_t luk1_offset, luk2_offset; const uint8_t *term_order; MDB_dbi dbi = 0; // Establish lookup ordering with some awkward offset math. for(int i = 0; i < 3; i++) { if ( ( args->idx0 == lookup_ordering_2bound[i][0] && args->idx1 == lookup_ordering_2bound[i][1] ) || ( args->idx0 == lookup_ordering_2bound[i][1] && args->idx1 == lookup_ordering_2bound[i][0] ) ) { term_order = lookup_ordering_2bound[i]; if (term_order[0] == args->idx0) { luk1_offset = 0; luk2_offset = 1; } else { luk1_offset = 1; luk2_offset = 0; } dbi = store->dbi[lookup_indices[i + 3]]; break; } } if (dbi == 0) { TRACE( "Values %d and %d not found in lookup keys.", args->idx0, args->idx1); return LSUP_VALUE_ERR; } // Compose term keys in lookup key. LSUP_DoubleKey luk; luk[luk1_offset] = args->luks[0]; luk[luk2_offset] = args->luks[1]; MDB_txn *txn; if(store->txn) txn = store->txn; else mdb_txn_begin(store->env, NULL, MDB_RDONLY, &txn); MDB_cursor *cur; mdb_cursor_open(txn, store->dbi[dbi], &cur); MDB_val key_v, data_v; key_v.mv_data = luk; key_v.mv_size = DBL_KLEN; if(args->ct) { mdb_cursor_get(cur, &key_v, &data_v, MDB_SET); mdb_cursor_count(cur, args->ct); } if(args->callback_fn) { rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_SET); if (rc == MDB_SUCCESS) rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_GET_MULTIPLE); while (rc != MDB_NOTFOUND) { LSUP_Key *lu_dset = data_v.mv_data; for (int i = 0; i < data_v.mv_size / DBL_KLEN; i++) { // Build triple key from lookup key and result keys. LSUP_TripleKey spok; spok[term_order[0]] = luk[0]; spok[term_order[1]] = luk[1]; spok[term_order[2]] = lu_dset[i]; rc = args->callback_fn(spok, args->ctx); if (rc < 0) goto _match2b_abort; } rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_NEXT_MULTIPLE); } } _match2b_abort: mdb_cursor_close(cur); if (txn != store->txn) mdb_txn_abort(txn); return rc; } inline static LSUP_rc match_callback_3bound(struct MDBStore *store, struct MatchArgs *args) { int rc = LSUP_NORESULT; MDB_txn *txn; if(store->txn) txn = store->txn; else mdb_txn_begin(store->env, NULL, MDB_RDONLY, &txn); MDB_cursor *cur; rc = mdb_cursor_open(txn, store->dbi[IDX_SPO_C], &cur); MDB_val key_v, data_v; key_v.mv_data = args->luks; key_v.mv_size = TRP_KLEN; if(args->ct) { if (mdb_cursor_get(cur, &key_v, NULL, MDB_SET) == MDB_SUCCESS) rc = mdb_cursor_count(cur, args->ct); } if(args->callback_fn) { rc = mdb_cursor_get(cur, &key_v, NULL, MDB_FIRST); while (rc != MDB_NOTFOUND) { LSUP_TripleKey spok; rc = args->callback_fn(spok, args->ctx); if (rc < 0) goto _match3b_abort; } } _match3b_abort: mdb_cursor_close(cur); if (txn != store->txn) mdb_txn_abort(txn); return rc; }