lsup_rdf/src/store_mdb.c

#include <ftw.h>

#include "store_mdb.h"

/**
 * Number of DBs defined.
 */
#define N_DB 12

/**
 * Memory map size.
 */
#if !(defined __LP64__ || defined __LLP64__) || \
        defined _WIN32 && !defined _WIN64
    #define DEFAULT_MAPSIZE 1<<31 // 2Gb (limit for 32-bit systems)
#else
    #define DEFAULT_MAPSIZE 1UL<<40 // 1Tb
#endif

#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_DIRTY_TXN    = 4, // Main txn was opened in a subroutine.
} StoreState;


typedef enum {
    OP_ADD,
    OP_REMOVE,
} StoreOp;


typedef struct MDBStore {
    MDB_env *           env;        // Environment handle.
    MDB_txn *           txn;        // Current transaction.
    MDB_dbi             dbi[N_DB];  // DB handles. Refer to DbIdx enum.
    LSUP_Buffer *       default_ctx;// Default ctx as a serialized URI.
    StoreState          state;      // Store state.
} MDBStore;


/** @brief Iterator operation.
 *
 * Function executed for each iteration of a #MDBIterator. It assumes that a
 * result triple has already been found and is ready to be composed and
 * yielded.
 *
 * Upon call, the rc value of the iterator structure is set to the MDB_* rc
 * value for the next result. It is up to the caller to evaluate this value
 * and decide whether to call the function again.
 */
typedef void (*iter_op_fn_t)(struct MDBIterator *it);


/** @brief Triple iterator.
 */
typedef struct MDBIterator {
    MDBStore *          store;      // MDB store pointer.
    MDB_txn *           txn;        // MDB transaction.
    MDB_cursor *        cur;        // MDB cursor.
    MDB_val             key, data;  // Internal data handlers.
    LSUP_TripleKey      spok;       // Triple to be populated with match.
    LSUP_Key            ck;         // Ctx key to filter by. May be NULL_TRP.
    iter_op_fn_t        iter_op_fn; // Function used to look up next match.
    const uint8_t *     term_order; // Term order used in 1-2bound look-ups.
    LSUP_Key            luk[3];     // 0÷3 lookup keys.
    size_t              i;          // Internal counter for paged lookups.
    int                 rc;         // MDB_* return code for the next result.
    StoreState          state;      // State flags.
} MDBIterator;


/*
 * 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 DUPSORT_MASK        MDB_DUPSORT
#define DUPFIXED_MASK       MDB_DUPSORT | MDB_DUPFIXED
#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

/**
 * Main DBs. These are the master information containers.
 *
 * Data columns are: identifier prefix, DB label, flags.
 */
#define MAIN_TABLE \
    ENTRY(  T_ST,    "t:st",    0               )   /* Key to ser. term */  \
    ENTRY(  SPO_C,   "spo:c",   DUPFIXED_MASK   )   /* Triple to context */ \
    ENTRY(  C_,      "c:",      0               )   /* Track empty ctx */   \
    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",    DUPFIXED_MASK   )   /* 1-bound lookup */    \
    ENTRY(  P_SO,    "p:so",    DUPFIXED_MASK   )   /* 1-bound lookup */    \
    ENTRY(  O_SP,    "o:sp",    DUPFIXED_MASK   )   /* 1-bound lookup */    \
    ENTRY(  PO_S,    "po:s",    DUPFIXED_MASK   )   /* 2-bound lookup */    \
    ENTRY(  SO_P,    "so:p",    DUPFIXED_MASK   )   /* 2-bound lookup */    \
    ENTRY(  SP_O,    "sp:o",    DUPFIXED_MASK   )   /* 2-bound lookup */    \
    ENTRY(  C_SPO,   "c:spo",   DUPFIXED_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.
 *
 * TODO deprecate?
 *
 * 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 index_triple(
        LSUP_MDBStore *store, StoreOp op,
        LSUP_TripleKey spok, LSUP_Key ck);

inline static LSUP_rc lookup_0bound(
        MDBStore *store, MDBIterator *it, size_t *ct);
inline static LSUP_rc lookup_1bound(
        MDBStore *store, uint8_t idx0, MDBIterator *it, size_t *ct);
inline static LSUP_rc lookup_2bound(
        MDBStore *store, uint8_t idx0, uint8_t idx1,
        MDBIterator *it, size_t *ct);
inline static LSUP_rc lookup_3bound(
        MDBStore *store, MDBIterator *it, size_t *ct);
/* 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_mdbstore_setup(char **path/*, bool clear*/) // TODO clear
{
    int rc;

    // Set environment path.
    if (path == NULL && (*path = getenv("LSUP_STORE_PATH")) == NULL) {
        // FIXME This won't work for multiple graphs with different disk
        // back ends. A random path generator needs to be used.
        *path = DEFAULT_ENV_PATH;
        fprintf(
                stderr,
                "WARNING: `LSUP_STORE_PATH' environment variable is not set. "
                "The default location %s will be used as the graph store.\n",
                *path);
    }

    // TODO Verify that a writable directory exists or can be created.
    //struct stat path_stat;

    /*
    // TODO clear
    if (clear) {
        rmrf(*path);
        if (mkdir(*path, ENV_DIR_MODE) != 0) abort();
    }
    */

    if (mkdir_p(*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_set_maxdbs(env, N_DB);
    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++) {
        TRACE("Creating DB %s", db_labels[i]);
        MDB_dbi dbi;
        rc = mdb_dbi_open(txn, db_labels[i], db_flags[i] | MDB_CREATE, &dbi);
        if (rc != MDB_SUCCESS) return rc;
    }

    mdb_txn_commit(txn);
    mdb_env_close(env);

    return rc;
}


MDBStore *
LSUP_mdbstore_new(const char *path, const LSUP_Buffer *default_ctx)
{
    int rc;
    LSUP_MDBStore *store;
    CRITICAL(store = malloc(sizeof(LSUP_MDBStore)));

    rc = mdb_env_create(&store->env);
    TRACE("create rc: %d", rc);

    store->default_ctx = (
            default_ctx ?
            LSUP_buffer_new(default_ctx->size, default_ctx->addr) : NULL);

    // Set map size.
    size_t mapsize;
    char *env_mapsize = getenv("LSUP_MDB_MAPSIZE");
    if (env_mapsize == NULL) mapsize = DEFAULT_MAPSIZE;
    else sscanf(env_mapsize, "%lu", &mapsize);

    rc = mdb_env_set_maxdbs(store->env, N_DB);
    if (UNLIKELY (rc != MDB_SUCCESS)) return NULL;

    rc = mdb_env_open(store->env, path, 0, ENV_FILE_MODE);
    if (UNLIKELY (rc != MDB_SUCCESS)) return NULL;

    // 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++) {
        rc = mdb_dbi_open(txn, db_labels[i], db_flags[i], store->dbi + i);
        if (UNLIKELY (rc != MDB_SUCCESS)) {
            mdb_txn_abort(txn);
            return NULL;
        }
    }

    mdb_txn_commit(txn);

    store->state |= LSSTORE_OPEN;
    store->txn = NULL;

    return store;
}


void
LSUP_mdbstore_free(LSUP_MDBStore *store)
{
    if (store->state & LSSTORE_OPEN) {
        TRACE(STR, "Closing MDB env.\n");
        mdb_env_close(store->env);
    }

    if (store->default_ctx) {
        LSUP_buffer_done(store->default_ctx);
        free(store->default_ctx);
    }

    free(store);
}


LSUP_rc
LSUP_mdbstore_stats(LSUP_MDBStore *store)
{
    // TODO
    // MDB_stat env_stat, db_stats[N_DB];
    return 0;
}


size_t
LSUP_mdbstore_size(LSUP_MDBStore *store)
{
    if(!(store->state & LSSTORE_INIT)) return 0;

    bool txn_pending = false;
    if (!store->txn) {
        mdb_txn_begin(store->env, NULL, 0, &store->txn);
        txn_pending = true;
    }

    MDB_stat stat;
    int rc = mdb_stat(store->txn, store->dbi[IDX_SPO_C], &stat);
    TRACE("Stat rc: %d\n", rc);
    // TODO error handling.

    if(txn_pending) {
        mdb_txn_abort(store->txn);
        store->txn = NULL;
    }

    return stat.ms_entries;
}


MDBIterator *
LSUP_mdbstore_add_init(LSUP_MDBStore *store, const LSUP_Buffer *sc)
{
    /* An iterator is used here. Some members are a bit misused but it does
     * its job without having to define a very similar struct.
     */
    MDBIterator *it;
    CRITICAL (it = malloc (sizeof (*it)));
    it->i = 0;

    if (!store->txn) {
        mdb_txn_begin(store->env, NULL, 0, &store->txn);
        // We are starting the main DB txn and we need to close it afterwards.
        it->state = LSSTORE_DIRTY_TXN;
    }

    // Take care of context first.
    // Serialize and hash.
    it->ck = NULL_KEY;

    if (store->default_ctx != NULL) {
        if (sc == NULL) sc = store->default_ctx;

        it->ck = LSUP_sterm_to_key(sc);

        // Insert t:st for context.
        //TRACE("Adding context: %s", sc);
        it->key.mv_data = &it->ck;
        it->key.mv_size = KLEN;
        it->data.mv_data = sc->addr;
        it->data.mv_size = sc->size;

        if (mdb_put(
                store->txn, store->dbi[IDX_T_ST],
                &it->key, &it->data, MDB_NOOVERWRITE) != MDB_SUCCESS)
            it->rc = LSUP_DB_ERR;
    }

    return it;
}


LSUP_rc
LSUP_mdbstore_add_iter(MDBIterator *it, const LSUP_SerTriple *sspo)
{
    int db_rc;
    LSUP_TripleKey spok = NULL_TRP;

    // Add triple.
    for (int j = 0; j < 3; j++) {
        LSUP_Buffer *st = LSUP_striple_pos(sspo, j);

        printf("Inserting term: ");
        LSUP_buffer_print(st);
        printf("\n");

        spok[j] = LSUP_sterm_to_key(st);

        it->key.mv_data = spok + j;
        it->key.mv_size = KLEN;
        it->data.mv_data = st->addr;
        it->data.mv_size = st->size;

        db_rc = mdb_put(
                it->store->txn, it->store->dbi[IDX_T_ST],
                &it->key, &it->data, MDB_NOOVERWRITE);
        if (db_rc == MDB_SUCCESS) it->rc = LSUP_OK;
        else if (db_rc != MDB_KEYEXIST) {
            it->rc = LSUP_DB_ERR;
            return it->rc;
        }
    }

    TRACE("Inserting spok: {%lx, %lx, %lx}", spok[0], spok[1], spok[2]);

    // Insert spo:c.
    it->key.mv_data = spok;
    it->key.mv_size = TRP_KLEN;

    // In triple mode, data is empty (= NULL_KEY).
    it->data.mv_data = &it->ck;
    it->data.mv_size = it->ck == NULL_KEY ? 0 : KLEN;

    db_rc = mdb_put(
            it->store->txn, it->store->dbi[IDX_SPO_C],
            &it->key, &it->data, MDB_NODUPDATA);
    if (db_rc == MDB_SUCCESS) it->rc = LSUP_OK;
    else if (db_rc != MDB_KEYEXIST) {
        it->rc = LSUP_DB_ERR;
        return it->rc;
    }

    // Index.
    it->rc = index_triple(it->store, OP_ADD, spok, it->ck);

    if(it->rc == LSUP_OK) it->i++;

    return it->rc;
}


LSUP_rc
LSUP_mdbstore_add_done(MDBIterator *it, size_t *inserted)
{
    // Only return commit rc if it fails.
    if (it->state & LSSTORE_DIRTY_TXN) {
        if (it->rc == LSUP_OK) {
            if(mdb_txn_commit(it->store->txn) != MDB_SUCCESS) {
                mdb_txn_abort(it->store->txn);
                it->rc = LSUP_DB_ERR;
            }
        } else mdb_txn_abort(it->store->txn);

        it->store->txn = NULL;
    }

    return it->rc;
}


LSUP_rc
LSUP_mdbstore_add(
        LSUP_MDBStore *store, const LSUP_Buffer *sc,
        const LSUP_SerTriple strp[], const size_t ct, size_t *inserted)
{
    MDBIterator *it = LSUP_mdbstore_add_init(store, sc);
    if (it->rc < 0) {
        for (size_t i = 0; i < ct; i++)
            if (LSUP_mdbstore_add_iter(it, strp + i) < 0) break;
    }

    return LSUP_mdbstore_add_done(it, inserted);
}


LSUP_Key
LSUP_mdbstore_sterm_to_key(
        LSUP_MDBStore *store, const LSUP_Buffer *sterm)
{
    // TODO this will be replaced by a lookup when 128-bit hash is introduced.
    return LSUP_sterm_to_key(sterm);
}


LSUP_rc
LSUP_mdbstore_key_to_sterm(
        LSUP_MDBStore *store, const LSUP_Key key, LSUP_Buffer *sterm)
{
    LSUP_rc 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, 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;

    if (txn != store->txn) mdb_txn_abort(txn);

    return rc;
}


MDBIterator *
LSUP_mdbstore_lookup(
        LSUP_MDBStore *store, const LSUP_SerTriple *sspo,
        const LSUP_Buffer *sc, size_t *ct)
{
    LSUP_TripleKey spok = {
        LSUP_sterm_to_key(sspo->s),
        LSUP_sterm_to_key(sspo->p),
        LSUP_sterm_to_key(sspo->o),
    };

    LSUP_MDBIterator *it;
    CRITICAL(it = malloc (sizeof (*it)));

    it->store = store;
    it->ck = store->default_ctx ? LSUP_sterm_to_key(sc) : NULL_KEY;

    if(ct) *ct = 0;

    uint8_t idx0, idx1;

    // s p o (all terms bound)
    if (spok[0] != NULL_KEY && spok[1] != NULL_KEY && spok[2] != NULL_KEY) {
        it->luk[0] = spok[0];
        it->luk[1] = spok[1];
        it->luk[2] = spok[2];
        RCNL (lookup_3bound (store, it, ct));

    } else if (spok[0] != NULL_KEY) {
        it->luk[0] = spok[0];
        idx0 = 0;

        // s p ?
        if (spok[1] != NULL_KEY) {
            it->luk[1] = spok[1];
            idx1 = 1;
            RCNL (lookup_2bound (store, idx0, idx1, it, ct));

        // s ? o
        } else if (spok[2] != NULL_KEY) {
            it->luk[1] = spok[2];
            idx1 = 2;
            RCNL (lookup_2bound (store, idx0, idx1, it, ct));

        // s ? ?
        } else RCNL (lookup_1bound (store, idx0, it, ct));

    } else if (spok[1] != NULL_KEY) {
        it->luk[0] = spok[1];
        idx0 = 1;

        // ? p o
        if (spok[2] != NULL_KEY) {
            it->luk[1] = spok[2];
            idx1 = 2;
            RCNL (lookup_2bound (store, idx0, idx1, it, ct));

        // ? p ?
        } else RCNL (lookup_1bound (store, idx0, it, ct));

    // ? ? o
    } else if (spok[2] != NULL_KEY) {
        it->luk[0] = spok[2];
        idx0 = 2;
        RCNL (lookup_1bound(store, idx0, it, ct));

    // ? ? ? (all terms unbound)
    } else RCNL (lookup_0bound(store, it, ct));

    return it;
}


LSUP_rc
mdbiter_next_key(LSUP_MDBIterator *it)
{
    if (UNLIKELY (!it)) return LSUP_DB_ERR;

    // Only advance if the previous it->rc wasn't already at the end.
    if(it->rc == MDB_NOTFOUND) return LSUP_END;

    if (UNLIKELY (it->rc != MDB_SUCCESS)) {
        fprintf(stderr, mdb_strerror(it->rc));
        return LSUP_DB_ERR;
    }

    LSUP_rc rc;

    it->iter_op_fn(it);

    if (it->ck) {
        rc = LSUP_NORESULT;  // Intermediary value, will never be returned.

        MDB_cursor *cur;
        MDB_val key, data;

        mdb_cursor_open
            (mdb_cursor_txn(it->cur), it->store->dbi[IDX_SPO_C], &cur);

        key.mv_size = TRP_KLEN;
        data.mv_data = &it->ck;
        data.mv_size = KLEN;

        while (rc == LSUP_NORESULT) {
            TRACE(STR, "begin ctx loop.");
            // If ctx is specified, look if the matching triple is associated
            // with it. If not, move on to the next triple.
            // The loop normally exits when a triple with matching ctx is found
            // (LSUP_OK), if there are no more triples (LSUP_END), or if there
            // is an error (LSUPP_DB_ERR).
            key.mv_data = it->spok;

            int db_rc = mdb_cursor_get(cur, &key, &data, MDB_GET_BOTH);

            if (db_rc == MDB_SUCCESS) {
                rc = LSUP_OK;
                TRACE(STR, "Triple found for context.");
            }

            else if (db_rc == MDB_NOTFOUND) {
                TRACE(STR, "No triples found for context.");
                if (it->rc == MDB_NOTFOUND) rc = LSUP_END;
                else it->iter_op_fn(it);

            } else {
                fprintf(stderr, mdb_strerror(it->rc));
                rc = LSUP_DB_ERR;
            }

        }

        mdb_cursor_close(cur);

    } else rc = LSUP_OK;

    return rc;
}


LSUP_rc
LSUP_mdbiter_next(LSUP_MDBIterator *it, LSUP_SerTriple *sspo)
{
    LSUP_rc rc = mdbiter_next_key(it);

    if (sspo && rc == LSUP_OK) {
        LSUP_mdbstore_key_to_sterm(it->store, it->spok[0], sspo->s);
        LSUP_mdbstore_key_to_sterm(it->store, it->spok[1], sspo->p);
        LSUP_mdbstore_key_to_sterm(it->store, it->spok[2], sspo->o);

        // TODO error handling.
    }

    return rc;
}


void
LSUP_mdbiter_free(MDBIterator *it)
{
    if (it) {
        mdb_cursor_close(it->cur);
        if(it->store->txn != it->txn) mdb_txn_abort(it->txn);

        free(it);
        it = NULL;
    }
}


LSUP_rc
LSUP_mdbstore_remove(
        MDBStore *store, const LSUP_SerTriple *sspo,
        const LSUP_Buffer *sc, size_t *ct)
{
    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;

    spok_v.mv_size = TRP_KLEN;
    ck_v.mv_size = KLEN;

    LSUP_MDBIterator *it = LSUP_mdbstore_lookup(store, sspo, sc, ct);
    if (UNLIKELY (!it)) return LSUP_DB_ERR;

    while (mdbiter_next_key(it)) {
        spok_v.mv_data = it->spok;

        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 = it->spok;

        // 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, it->spok, 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;
}


/* * * 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); }
*/


static LSUP_rc
index_triple(
        LSUP_MDBStore *store, StoreOp op,
        LSUP_TripleKey spok, LSUP_Key ck)
{
    int rc = LSUP_NOACTION;
    MDB_val v1, v2;

    printf("Indexing triple: %lx %lx %lx\n", spok[0], spok[1], spok[2]);

    // 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;

    int db_rc;
    for (int i = 0; i < 3; i++) {
        MDB_dbi db1 = store->dbi[lookup_indices[i]];      // s:po, p:so, o:sp
        MDB_dbi db2 = store->dbi[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.
            printf("Indexing in %s: ", db_labels[lookup_indices[i]]);
            printf(
                    "%lx: %lx %lx\n", *(size_t*)(v1.mv_data),
                    *(size_t*)(v2.mv_data), *(size_t*)(v2.mv_data) + 1);

            db_rc = mdb_put(store->txn, db1, &v1, &v2, MDB_NODUPDATA);
            if (db_rc != MDB_SUCCESS && db_rc != MDB_KEYEXIST)
                return LSUP_DB_ERR;

            printf("Indexing in %s: ", db_labels[lookup_indices[i + 3]]);
            printf(
                    "%lx %lx: %lx\n", *(size_t*)(v2.mv_data),
                    *(size_t*)(v2.mv_data) + 1, *(size_t*)(v1.mv_data));

            db_rc = mdb_put(store->txn, db2, &v2, &v1, MDB_NODUPDATA);
            if (db_rc != MDB_SUCCESS && db_rc != MDB_KEYEXIST)
                return LSUP_DB_ERR;
        }
    }

    return rc;
}


/* * * Term-specific iterators. * * */

/** @brief Advance 0-bound iterator.
 *
 * Cursor: spo:c
 */
inline static void
it_next_0bound(MDBIterator *it)
{
    memcpy(it->spok, it->data.mv_data, sizeof(LSUP_TripleKey));

    it->rc = mdb_cursor_get(it->cur, &it->key, NULL, MDB_NEXT);
}


/** @brief Advance 1-bound iterator.
 *
 * Uses paged data in a nested loop.
 *
 * Cursor: s:po, p:so, or o:sp.
 */
inline static void
it_next_1bound(MDBIterator *it)
{
    LSUP_DoubleKey *lu_dset = it->data.mv_data;

    it->spok[it->term_order[0]] = it->luk[0];
    it->spok[it->term_order[1]] = lu_dset[it->i][0];
    it->spok[it->term_order[2]] = lu_dset[it->i][1];

    TRACE(
            "Composed triple: {%lu %lu %lu}",
            it->spok[0], it->spok[1], it->spok[2]);

    // Ensure next block within the same page is not beyond the last.
    if(it->i < it->data.mv_size / DBL_KLEN - 1) {
        it->i ++;
        TRACE("Increasing page cursor to %lu.", it->i);
        TRACE("it->rc: %d", it->rc);

    } else {
        // If the last block in the page is being yielded,
        // move cursor to beginning of next page.
        it->i = 0;
        TRACE("Reset page cursor to %lu.", it->i);
        it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_NEXT_MULTIPLE);
        TRACE("it->rc: %d", it->rc);
    }
}


/** @brief Advance 2-bound iterator.
 *
 * Uses paged data in a nested loop.
 *
 * Cursor: po:s, so:p, or sp:o.
 */
inline static void
it_next_2bound(MDBIterator *it)
{
    LSUP_Key *lu_dset = it->data.mv_data;

    it->spok[it->term_order[0]] = it->luk[0];
    it->spok[it->term_order[1]] = it->luk[1];
    it->spok[it->term_order[2]] = lu_dset[it->i];

    // Ensure next block within the same page is not beyond the last.
    if(it->i < it->data.mv_size / KLEN - 1)
        it->i ++;
    else {
        // If the last block in the page is being yielded,
        // move cursor to beginning of next page.
        it->i = 0;
        it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_NEXT_MULTIPLE);
    }
}


/** @brief Advance 3-bound iterator.
 *
 * This is a special case of 0÷1 results; either there was one matching triple,
 * which was already set in the first result, or there was none, i.e. it->rc is
 * already MDB_NOTFOUND and this function will not be called.
 */
inline static void
it_next_3bound(MDBIterator *it)
{
    it->rc = MDB_NOTFOUND;
}


/* * * Term-specific lookups. * * */

inline static LSUP_rc
lookup_0bound(MDBStore *store, MDBIterator *it, size_t *ct)
{
    if(store->txn) it->txn = store->txn;
    else {
        it->rc = mdb_txn_begin(store->env, NULL, MDB_RDONLY, &it->txn);
        if (it->rc != MDB_SUCCESS) abort();
    }

    if(ct) {
        if(it->ck != NULL_KEY) {
            // Look up by given context.
            it->rc = mdb_cursor_open(it->txn, store->dbi[IDX_C_SPO], &it->cur);

            it->key.mv_data = &it->ck;
            it->key.mv_size = KLEN;

            it->rc = mdb_cursor_get(it->cur, &it->key, NULL, MDB_SET);
            if (it->rc == MDB_SUCCESS) mdb_cursor_count(it->cur, ct);

            mdb_cursor_close(it->cur);

        } else {
            // Look up all contexts.
            MDB_stat stat;
            mdb_stat(it->txn, store->dbi[IDX_S_PO], &stat);

            *ct = stat.ms_entries;
        }
    }

    mdb_cursor_open(it->txn, store->dbi[IDX_SPO_C], &it->cur);

    it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_FIRST);
    it->iter_op_fn = it_next_0bound;

    if (it->rc != MDB_SUCCESS && it->rc != MDB_NOTFOUND) {
        fprintf(stderr, "Database error: %s", mdb_strerror(it->rc));
        return LSUP_DB_ERR;
    }

    return LSUP_OK;
}


inline static LSUP_rc
lookup_1bound(MDBStore *store, uint8_t idx0, MDBIterator *it, size_t *ct)
{
    it->term_order = (const uint8_t*)lookup_ordering_1bound[idx0];

    TRACE("Looking up 1 bound term: %lu\n", it->luk[0]);

    if(!it->txn) {
        if(store->txn) it->txn = store->txn;
        else {
            it->rc = mdb_txn_begin(store->env, NULL, MDB_RDONLY, &it->txn);
            if (it->rc != MDB_SUCCESS) abort();
        }
    }

    mdb_cursor_open(it->txn, store->dbi[lookup_indices[idx0]], &it->cur);

    it->key.mv_data = it->luk;
    it->key.mv_size = KLEN;

    if(ct) {
        // If a context is specified, the only way to count triples matching
        // the context is to loop over them.
        if (it->ck != NULL_KEY) {
            MDBIterator *ct_it;
            CRITICAL(ct_it = malloc(sizeof(MDBIterator)));

            ct_it->luk[0] = it->luk[0];
            LSUP_TripleKey ct_spok;
            memcpy(ct_it->spok, ct_spok, sizeof(LSUP_TripleKey));
            ct_it->ck = it->ck;
            ct_it->store = it->store;
            ct_it->txn = it->txn;
            ct_it->key = it->key;
            ct_it->data = it->data;
            ct_it->i = 0;
            lookup_1bound(store, idx0, ct_it, NULL);

            while (LSUP_mdbiter_next(ct_it, NULL) != LSUP_END) {
                ct[0] ++;
                TRACE("Counter increased to %lu.", *ct);
            }

            mdb_cursor_close(ct_it->cur);
            free(ct_it);

        } else {
            it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_SET);
            if (it->rc == MDB_SUCCESS) mdb_cursor_count(it->cur, ct);
        }
    }

    it->i = 0;
    it->iter_op_fn = it_next_1bound;

    it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_SET);
    if (it->rc == MDB_SUCCESS)
        it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_GET_MULTIPLE);

    if (it->rc != MDB_SUCCESS && it->rc != MDB_NOTFOUND) {
        fprintf(stderr, "Database error: %s", mdb_strerror(it->rc));
        return LSUP_DB_ERR;
    }

    return LSUP_OK;
}


inline static LSUP_rc
lookup_2bound(
        MDBStore *store, uint8_t idx0, uint8_t idx1,
        MDBIterator *it, size_t *ct)
{
    uint8_t luk1_offset, luk2_offset;
    MDB_dbi dbi = 0;

    // Establish lookup ordering with some awkward offset math.
    for(int i = 0; i < 3; i++) {
        if (
            (
                idx0 == lookup_ordering_2bound[i][0] &&
                idx1 == lookup_ordering_2bound[i][1]
            ) || (
                idx0 == lookup_ordering_2bound[i][1] &&
                idx1 == lookup_ordering_2bound[i][0]
            )
        ) {
            it->term_order = (const uint8_t*)lookup_ordering_2bound[i];
            if (it->term_order[0] == idx0) {
                luk1_offset = 0;
                luk2_offset = 1;
            } else {
                luk1_offset = 1;
                luk2_offset = 0;
            }
            dbi = store->dbi[lookup_indices[i + 3]];
            TRACE(
                    "Looking up 2 bound in %s\n",
                    db_labels[lookup_indices[i + 3]]);

            break;
        }
    }

    if (dbi == 0) {
        TRACE(
                "Values %d and %d not found in lookup keys.",
                idx0, idx1);
        return LSUP_VALUE_ERR;
    }

    // Compose term keys in lookup key.
    LSUP_DoubleKey luk;
    luk[luk1_offset] = it->luk[0];
    luk[luk2_offset] = it->luk[1];

    if(!it->txn) {
        if(store->txn) it->txn = store->txn;
        else {
            it->rc = mdb_txn_begin(store->env, NULL, MDB_RDONLY, &it->txn);
            if (it->rc != MDB_SUCCESS) abort();
        }
    }

    it->key.mv_data = luk;
    it->key.mv_size = DBL_KLEN;

    mdb_cursor_open(it->txn, dbi, &it->cur);
    it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_SET);

    if(ct) {
        // If a context is specified, the only way to count triples matching
        // the context is to loop over them.
        if (it->ck != NULL_KEY) {
            MDBIterator *ct_it;
            CRITICAL(ct_it = malloc(sizeof(MDBIterator)));

            ct_it->luk[0] = it->luk[0];
            ct_it->luk[1] = it->luk[1];
            LSUP_TripleKey ct_spok;
            memcpy(ct_it->spok, ct_spok, sizeof(LSUP_TripleKey));
            ct_it->ck = it->ck;
            ct_it->store = it->store;
            ct_it->txn = it->txn;
            lookup_2bound(store, idx0, idx1, ct_it, NULL);

            while (LSUP_mdbiter_next(ct_it, NULL) != LSUP_END) {
                ct[0] ++;
            }
            if (ct_it->cur) mdb_cursor_close(ct_it->cur);
            free(ct_it);

        } else {
            it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_SET);
            if (it->rc == MDB_SUCCESS) mdb_cursor_count(it->cur, ct);
        }
    }

    it->i = 0;
    it->iter_op_fn = it_next_2bound;

    it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_SET);
    if (it->rc == MDB_SUCCESS)
        it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_GET_MULTIPLE);

    if (it->rc != MDB_SUCCESS && it->rc != MDB_NOTFOUND) {
        fprintf(stderr, "Database error: %s", mdb_strerror(it->rc));
        return LSUP_DB_ERR;
    }

    return LSUP_OK;
}


inline static LSUP_rc
lookup_3bound(MDBStore *store, MDBIterator *it, size_t *ct)
{
    TRACE(
            "Looking up 3 bound: {%lx, %lx, %lx}",
            it->luk[0], it->luk[1], it->luk[2]);

    if(store->txn) it->txn = store->txn;
    else mdb_txn_begin(store->env, NULL, MDB_RDONLY, &it->txn);

    it->key.mv_data = it->luk;

    if(it->ck != NULL_KEY) {
        it->rc = mdb_cursor_open(it->txn, store->dbi[IDX_SPO_C], &it->cur);

        it->key.mv_size = TRP_KLEN;
        it->data.mv_data = &it->ck;
        it->data.mv_size = KLEN;

    } else {
        it->rc = mdb_cursor_open(it->txn, store->dbi[IDX_S_PO], &it->cur);

        it->key.mv_size = KLEN;
        it->data.mv_data = it->luk + 1;
        it->data.mv_size = DBL_KLEN;
    }

    it->rc = mdb_cursor_get(it->cur, &it->key, &it->data, MDB_GET_BOTH);

    mdb_cursor_close(it->cur);

    if(ct && it->rc == MDB_SUCCESS) *ct = 1;

    it->iter_op_fn = it_next_3bound;
    memcpy(it->spok, it->luk, sizeof(LSUP_TripleKey));

    if (it->rc != MDB_SUCCESS && it->rc != MDB_NOTFOUND) {
        fprintf(stderr, "Database error: %s", mdb_strerror(it->rc));
        return LSUP_DB_ERR;
    }

    return LSUP_OK;
}