lsup_rdf/src/store_mdb.c

#include <ftw.h>

#include "uthash.h"

#include "store_mdb.h"

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

/**
 * Memory map size.
 */
#if (defined DEBUG || defined TESTING)
    #define DEFAULT_MAPSIZE 1<<24 // 16Mb (limit for Valgrind)
#elif !(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 ENV_DIR_MODE 0750
#define ENV_FILE_MODE 0640


/*
 * Data types.
 */

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_t {
    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)(LSUP_MDBIterator *it);


/** @brief Triple iterator.
 */
typedef struct mdbstore_iter_t {
    MDBStore *          store;      // MDB store pointer.
    MDB_txn *           txn;        // MDB transaction.
    MDB_cursor *        cur;        // MDB cursor.
    MDB_cursor *        ctx_cur;    // MDB c:spo index 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.
} MDBIterator;


// Set of single keys.
typedef struct key_set_t {
    LSUP_Key            key;
    UT_hash_handle      hh;
} KeySet;

// Set of triple keys.
typedef struct triple_set_t {
    LSUP_TripleKey      spok;
    UT_hash_handle      hh;
} TripleSet;

// Map of context to triple set.
typedef struct ctx_triple_map_t {
    LSUP_Key            ck;
    TripleSet *         spok;
    UT_hash_handle      hh;
} CtxTripleMap;

/*
 * Static variables.
 */

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

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

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 (MDBIterator *it, size_t *ct);
inline static LSUP_rc lookup_1bound (uint8_t idx0, MDBIterator *it, size_t *ct);
inline static LSUP_rc lookup_2bound (
        uint8_t idx0, uint8_t idx1, MDBIterator *it, size_t *ct);
inline static LSUP_rc lookup_3bound(MDBIterator *it, size_t *ct);


/**
 * API.
 */

LSUP_rc
LSUP_mdbstore_setup (const char *path, bool clear)
{
    int rc;

    // Set environment path.
    if (!path) return LSUP_ERROR;

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

    if (clear) rm_r (path);
    if (mkdir_p (path, ENV_DIR_MODE) != 0) return LSUP_IO_ERR;

    // Open a temporary environment and txn to create the DBs.
    MDB_env *env;
    mdb_env_create (&env);

    mdb_env_set_maxdbs (env, N_DB);
    RCCK (mdb_env_open (env, path, 0, ENV_FILE_MODE));
    log_debug ("Environment opened at %s.", path);

    MDB_txn *txn;
    mdb_txn_begin (env, NULL, 0, &txn);
    for (int i = 0; i < N_DB; i++) {
        log_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 db_rc;
    LSUP_MDBStore *store;
    MALLOC_GUARD (store, NULL);

    db_rc = mdb_env_create (&store->env);
    log_trace ("create rc: %d", db_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);
    log_info (
            "Setting environment map size at %s to %lu bytes.",
            path, mapsize);
    db_rc = mdb_env_set_mapsize (store->env, mapsize);

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

    db_rc = mdb_env_open (store->env, path, 0, ENV_FILE_MODE);
    if (UNLIKELY (db_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++) {
        db_rc = mdb_dbi_open (txn, db_labels[i], db_flags[i], store->dbi + i);
        if (UNLIKELY (db_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) {
        const char *path;
        mdb_env_get_path (store->env, &path);
        log_info ("Closing MDB env at %s.", path);
        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_stat (LSUP_MDBStore *store, MDB_stat *stat)
{
    if (!(store->state & LSSTORE_INIT)) return 0;

    MDB_txn *txn;
    mdb_txn_begin (store->env, NULL, MDB_RDONLY, &txn);

    if (mdb_stat (txn, store->dbi[IDX_SPO_C], stat) != MDB_SUCCESS)
        return LSUP_DB_ERR;
    mdb_txn_abort (txn);

    return LSUP_OK;
}


size_t
LSUP_mdbstore_size (LSUP_MDBStore *store)
{
    // Size is calculated outside of any pending write txn.

    MDB_stat stat;
    if (LSUP_mdbstore_stat (store, &stat) != LSUP_OK) return 0;

    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;
    MALLOC_GUARD (it, NULL);

    it->store = store;
    it->i = 0;

    // No other write transaction may be open.
    mdb_txn_begin (store->env, NULL, 0, &it->store->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_buffer_hash (sc);

        // Insert t:st for context.
        //log_debug ("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(
                it->store->txn, it->store->dbi[IDX_T_ST],
                &it->key, &it->data, MDB_NOOVERWRITE) != MDB_SUCCESS)
            return NULL;
    }

    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 i = 0; i < 3; i++) {
        LSUP_Buffer *st = LSUP_striple_pos (sspo, i);

        spok[i] = LSUP_buffer_hash (st);

        it->key.mv_data = spok + i;
        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 && db_rc != MDB_KEYEXIST) {
            log_error (
                    "MDB error while inserting term: %s", LSUP_strerror(db_rc));
            return LSUP_DB_ERR;
        }
    }

    log_trace ("Inserting spok: {%lx, %lx, %lx}", spok[0], spok[1], spok[2]);
    log_trace ("Into context: %lx", it->ck);

    // 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_KEYEXIST) return LSUP_NOACTION;
    if (db_rc != MDB_SUCCESS) {
        log_error (
                "MDB error while inserting triple: %s", LSUP_strerror(db_rc));
        return LSUP_DB_ERR;
    }

    // Index.
    LSUP_rc rc = index_triple (it->store, OP_ADD, spok, it->ck);
    if (rc == LSUP_OK) it->i++;

    return rc;
}


LSUP_rc
LSUP_mdbstore_add_done (MDBIterator *it)
{
    LSUP_rc rc = LSUP_OK;

    if (mdb_txn_commit (it->store->txn) != MDB_SUCCESS) {
        mdb_txn_abort (it->store->txn);
        rc = LSUP_DB_ERR;
    }

    it->store->txn = NULL;

    free (it);

    return rc;
}


void
LSUP_mdbstore_add_abort (MDBIterator *it)
{
    mdb_txn_abort (it->store->txn);

    it->store->txn = NULL;

    free (it);
}


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 (UNLIKELY (!it)) return LSUP_DB_ERR;

    for (size_t i = 0; i < ct; i++) {
        LSUP_rc rc = LSUP_mdbstore_add_iter (it, strp + i);
        if (UNLIKELY (rc < 0)) {
            LSUP_mdbstore_add_abort (it);
            return rc;
        }
    }
    *inserted = it->i;

    return LSUP_mdbstore_add_done (it);
}


/*
static LSUP_Key
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_buffer_hash (sterm);
}
*/


static LSUP_rc
key_to_sterm (LSUP_MDBIterator *it, const LSUP_Key key, LSUP_Buffer *sterm)
{
    LSUP_rc rc = LSUP_NORESULT;
    int db_rc;

    MDB_val key_v, data_v;
    key_v.mv_data = (void*)&key;
    key_v.mv_size = KLEN;

    db_rc = mdb_get (it->txn, it->store->dbi[IDX_T_ST], &key_v, &data_v);

    if (db_rc == MDB_SUCCESS) {
        free (sterm->addr);
        sterm->addr = data_v.mv_data;
        sterm->size = data_v.mv_size;
        rc = LSUP_OK;
    } else if (db_rc == MDB_NOTFOUND) {
        free (sterm->addr);
        sterm->addr = NULL;
        sterm->size = 0;
    } else rc = LSUP_ERROR;

    return rc;
}


MDBIterator *
LSUP_mdbstore_lookup(
        LSUP_MDBStore *store, const LSUP_Buffer *ss, const LSUP_Buffer *sp,
        const LSUP_Buffer *so, const LSUP_Buffer *sc, size_t *ct)
{
    LSUP_TripleKey spok = {
        LSUP_buffer_hash (ss),
        LSUP_buffer_hash (sp),
        LSUP_buffer_hash (so),
    };

    LSUP_MDBIterator *it;
    CALLOC_GUARD (it, NULL);

    it->store = store;
    it->ck = store->default_ctx ? LSUP_buffer_hash (sc) : NULL_KEY;
    log_debug ("Lookup context: %lx", it->ck);

    if (ct) *ct = 0;

    uint8_t idx0, idx1;

    // Start RO transaction if not in a write txn already.
    if (it->store->txn) it->txn = it->store->txn;
    else {
        it->rc = mdb_txn_begin (it->store->env, NULL, MDB_RDONLY, &it->txn);
        if (it->rc != MDB_SUCCESS) {
            log_error ("Database error: %s", LSUP_strerror (it->rc));
            return NULL;
        }
    }

    // Context index loop.
    if (UNLIKELY (mdb_cursor_open (
            it->txn, it->store->dbi[IDX_SPO_C], &it->ctx_cur) != MDB_SUCCESS))
        return NULL;

    /*
     * Lookup decision tree.
     */
    // 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 (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 (idx0, idx1, it, ct));

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

        // s ? ?
        } else RCNL (lookup_1bound (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 (idx0, idx1, it, ct));

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

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

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

    return it;
}


/** @brief Get next iterator key.
 *
 * The ck pointer is filled with an array of contexts that the triple appears
 * in, if not NULL.
 */
inline static LSUP_rc
mdbiter_next_key (LSUP_MDBIterator *it, KeySet **ck_p)
{
    if (UNLIKELY (!it)) return LSUP_VALUE_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)) {
        log_error ("Database error: %s", LSUP_strerror (it->rc));
        return LSUP_DB_ERR;
    }

    LSUP_rc rc;

    /* Retrieve current value and advance cursor to the next result.
     * it->rc is set to the result of the next iteration.
     */
    it->iter_op_fn (it);
    log_trace (
            "Found  spok: {%lx, %lx, %lx}",
            it->spok[0], it->spok[1], it->spok[2]);

    MDB_val key, data;
    int db_rc;

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

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

        while (rc == LSUP_NORESULT) {
            //log_debug ("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 (LSUP_DB_ERR).
            key.mv_data = it->spok;

            db_rc = mdb_cursor_get (it->ctx_cur, &key, &data, MDB_GET_BOTH);

            if (db_rc == MDB_SUCCESS) {
                rc = LSUP_OK;
                log_trace ("Triple found for context.");

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

            } else {
                log_error ("Database error: %s", LSUP_strerror (db_rc));
                rc = LSUP_DB_ERR;
            }

        }

    } else rc = LSUP_OK;

    // Get all contexts for a triple if requested. Add up to previous
    // iterations if the same pointer is passed.
    if (ck_p) {
        key.mv_data = it->spok;
        db_rc = mdb_cursor_get (it->ctx_cur, &key, &data, MDB_SET_KEY);
        if (db_rc == MDB_SUCCESS) {
            do {
                KeySet *entry;
                HASH_FIND (hh, *ck_p, data.mv_data, KLEN, entry);
                if (!entry) {
                    MALLOC_GUARD (entry, LSUP_MEM_ERR);
                    entry->key = *(LSUP_Key *) data.mv_data;
                    HASH_ADD (hh, *ck_p, key, KLEN, entry);
                }
            } while (
                    mdb_cursor_get (it->ctx_cur, &key, &data, MDB_NEXT_DUP)
                    == MDB_SUCCESS);
        }
    }

    return rc;
}


LSUP_rc
LSUP_mdbiter_next (
        LSUP_MDBIterator *it, LSUP_SerTriple *sspo, LSUP_Buffer **ctx_p)
{
    LSUP_rc rc;
    KeySet *ck = NULL;

    rc = (ctx_p) ? mdbiter_next_key (it, &ck) : mdbiter_next_key (it, NULL);

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

            // TODO error handling.
        }

        // One-shot contexts for current triple.
        if (ctx_p) {
            KeySet *ccur;
            LSUP_Buffer *ctx = malloc (HASH_COUNT (ck) * sizeof (*ctx));
            size_t i = 0;
            for (ccur = ck; ccur != NULL; ccur = ccur->hh.next)
                key_to_sterm (it, ccur->key, ctx + (i++));

            // TODO error handling.
        }
    }

    return rc;
}


size_t
LSUP_mdbiter_cur (LSUP_MDBIterator *it)
{ return it->i; }


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

        free (it);
    }
}


LSUP_rc
LSUP_mdbstore_remove(
        LSUP_MDBStore *store, const LSUP_Buffer *ss, const LSUP_Buffer *sp,
        const LSUP_Buffer *so, 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_buffer_hash (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;
    ck_v.mv_data = &ck;

    LSUP_MDBIterator *it = LSUP_mdbstore_lookup (store, ss, sp, so, sc, ct);
    if (UNLIKELY (!it)) return LSUP_DB_ERR;
    if (ct) log_debug ("Found %lu triples to remove.", *ct);

    while (mdbiter_next_key (it, NULL) == LSUP_OK) {
        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;

        log_trace (
                "Removing {%lx, %lx, %lx}",
                it->spok[0], it->spok[1], it->spok[2]);

        // Delete spo:c entry.
        rc = mdb_cursor_del (dcur, 0);
        if (UNLIKELY (rc != MDB_SUCCESS)) goto _remove_abort;

        // Restore ck address after each delete.
        spok_v.mv_data = it->spok;
        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;

        rc = mdb_cursor_del (icur, 0);
        if (UNLIKELY (rc != MDB_SUCCESS)) goto _remove_abort;

        spok_v.mv_data = it->spok;
        ck_v.mv_data = &ck;

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

    LSUP_mdbiter_free (it);

    if (UNLIKELY (mdb_txn_commit (txn) != MDB_SUCCESS)) {
        rc = LSUP_TXN_ERR;
        goto _remove_abort;
    }

    return rc;

_remove_abort:
    mdb_txn_abort (txn);
    log_error ("Database error: %s", LSUP_strerror (rc));

    return rc;
}


LSUP_Buffer **
LSUP_mdbstore_lookup_contexts (
        LSUP_MDBStore *store, const LSUP_Buffer *ss, const LSUP_Buffer *sp,
        const LSUP_Buffer *so)
{
    LSUP_MDBIterator *it = LSUP_mdbstore_lookup (
            store, ss, sp, so, NULL, NULL);

    LSUP_rc rc = LSUP_NORESULT;
    KeySet *ckey, *ckeys = NULL, *tmp;
    while (rc != LSUP_END)
        rc = mdbiter_next_key (it, &ckeys);

    size_t i = 0;
    LSUP_Buffer **ctx_a = calloc (HASH_COUNT (ckeys) + 1, sizeof (*ctx_a));
    if (UNLIKELY (!ctx_a)) return NULL;

    HASH_ITER (hh, ckeys, ckey, tmp) {
        ctx_a[i] = BUF_DUMMY;
        key_to_sterm (it, ckey->key, ctx_a[i++]);
        HASH_DEL (ckeys, ckey);
        free (ckey);
    }
    LSUP_mdbiter_free (it);

    return ctx_a;
}


LSUP_rc
LSUP_mdbstore_nsm_get (LSUP_MDBStore *store, LSUP_NSMap **nsm_p)
{
    LSUP_rc rc = LSUP_NORESULT;
    LSUP_NSMap *nsm = LSUP_nsmap_new();
    if (UNLIKELY (!nsm)) return LSUP_MEM_ERR;

    *nsm_p = nsm;

    MDB_txn *txn;
    mdb_txn_begin (store->env, NULL, MDB_RDONLY, &txn);

    MDB_cursor *cur;
    if (mdb_cursor_open (txn, store->dbi[IDX_PFX_NS], &cur) != MDB_SUCCESS) {
        mdb_txn_abort (txn);
        return LSUP_DB_ERR;
    }

    MDB_val ns_v, pfx_v;
    if (mdb_cursor_get (cur, &ns_v, &pfx_v, MDB_FIRST) != MDB_SUCCESS)
        goto finally;

    do {
        ns_pfx pfx;
        char *ns = malloc (ns_v.mv_size);

        strncpy (pfx, pfx_v.mv_data, pfx_v.mv_size);
        strncpy (ns, ns_v.mv_data, ns_v.mv_size);
        LSUP_nsmap_add (nsm, pfx, ns);

        free (ns);
    } while (mdb_cursor_get (
                cur, &ns_v, &pfx_v, MDB_NEXT_NODUP) == MDB_SUCCESS);

finally:
    mdb_cursor_close (cur);
    mdb_txn_abort (txn);

    return rc;
}


LSUP_rc
LSUP_mdbstore_nsm_store (LSUP_MDBStore *store, const LSUP_NSMap *nsm)
{
    MDB_txn *txn;
    if (!store->txn) {
        RCCK (mdb_txn_begin (store->env, NULL, 0, &txn));
    }
    else txn = store->txn;

    LSUP_rc rc = LSUP_NOACTION;
    int db_rc;

    MDB_cursor *dcur = NULL, *icur = NULL;
    if (
        mdb_cursor_open (txn, store->dbi[IDX_PFX_NS], &dcur) != MDB_SUCCESS
        ||
        mdb_cursor_open (txn, store->dbi[IDX_NS_PFX], &icur) != MDB_SUCCESS
    ) {
        rc = LSUP_DB_ERR;
        goto finally;
    }

    MDB_val pfx_v, ns_v;
    const char ***nsm_data = LSUP_nsmap_dump (nsm);

    for (size_t i = 0; nsm_data[i] != NULL; i++) {
        // At least 1 action. If not OK, it will change during the iteration.
        if (i == 0) rc = LSUP_OK;

        pfx_v.mv_data = (void *)nsm_data[i][0];
        pfx_v.mv_size = strlen (nsm_data[i][0]) + 1;
        ns_v.mv_data = (void *)nsm_data[i][1];
        ns_v.mv_size = strlen (nsm_data[i][1]) + 1;

        // If either ns or pfx exist, quit.
        if (
            mdb_cursor_get (dcur, &pfx_v, &ns_v, MDB_SET) != MDB_NOTFOUND
            ||
            mdb_cursor_get (icur, &ns_v, &pfx_v, MDB_SET) != MDB_NOTFOUND
        ) {
            rc = LSUP_CONFLICT;
            goto finally;
        }

        db_rc = mdb_cursor_put (dcur, &pfx_v, &ns_v, 0);
        if (db_rc != MDB_SUCCESS) {
            log_error ("DB error: %s", LSUP_strerror (db_rc));
            rc = LSUP_DB_ERR;
            goto finally;
        }
    }

finally:
    if (icur) mdb_cursor_close (icur);
    if (dcur) mdb_cursor_close (dcur);
    free (nsm_data);
    if (txn != store->txn) mdb_txn_commit (txn);

    return rc;
}


/* * * Static functions. * * */


/** @brief Index an added or removed triple.
 *
 * @param store[in] MDB store to index.
 * @param op[in] Store operation. One of OP_ADD or OP_REMOVE.
 * @param spok[in] Triple key to index.
 * @param ck[in] Context to index, may be NULL.
 */
static LSUP_rc
index_triple(
        LSUP_MDBStore *store, StoreOp op, LSUP_TripleKey spok, LSUP_Key ck)
{
    int db_rc;
    LSUP_rc rc = LSUP_NOACTION;
    MDB_val v1, v2;

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

    // Index c:spo.
    if (op == OP_REMOVE) {
        log_trace ("Indexing 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);
            if (mdb_cursor_get (cur, &v1, &v2, MDB_GET_BOTH) == MDB_SUCCESS) {
                db_rc = mdb_cursor_del (cur, 0);
                if (db_rc != MDB_SUCCESS) return LSUP_DB_ERR;

                rc = LSUP_OK;
            }

            mdb_cursor_close (cur);
        }

    } else if (op == OP_ADD) {
        log_trace ("Indexing op: ADD");
        if (ck != NULL_KEY) {
            v1.mv_data = &ck;
            v1.mv_size = KLEN;
            v2.mv_data = spok;
            v2.mv_size = TRP_KLEN;

            db_rc = mdb_put(
                    store->txn, store->dbi[IDX_C_SPO],
                    &v1, &v2, MDB_NODUPDATA);
            if (db_rc != MDB_SUCCESS) return LSUP_DB_ERR;
            if (db_rc != MDB_KEYEXIST) rc = LSUP_OK;
        }

    } 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 = 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);

            db_rc = mdb_cursor_get (cur1, &v1, &v2, MDB_GET_BOTH);
            if (db_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);

            db_rc = mdb_cursor_get (cur2, &v2, &v1, MDB_GET_BOTH);
            if (db_rc == MDB_SUCCESS) mdb_cursor_del (cur2, 0);
            // TODO error handling.
            rc = LSUP_OK;

            mdb_cursor_close (cur2);

        } else { // OP_ADD is guaranteed.
            // 1-bound index.
            log_trace ("Indexing in %s: ", db_labels[lookup_indices[i]]);
            log_trace (
                    "%lx: %lx %lx", *(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) rc = LSUP_OK;
            else if (db_rc != MDB_KEYEXIST) return LSUP_DB_ERR;

            // 2-bound index.
            log_trace ("Indexing in %s: ", db_labels[lookup_indices[i + 3]]);
            log_trace (
                    "%lx %lx: %lx", *(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) rc = LSUP_OK;
            else if (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->key.mv_data, sizeof (LSUP_TripleKey));

    it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, 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];

    log_trace (
            "Composed triple: {%lx %lx %lx}",
            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 ++;
        //log_debug ("Increasing page cursor to %lu.", it->i);
        //log_debug ("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;
        //log_debug ("Reset page cursor to %lu.", it->i);
        it->rc = mdb_cursor_get (
                it->cur, &it->key, &it->data, MDB_NEXT_MULTIPLE);
    }
}


/** @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 (MDBIterator *it, size_t *ct)
{
    log_debug ("Looking up 0 bound terms.");

    if (ct) {
        if (it->ck != NULL_KEY) {
            // Look up by given context.
            it->rc = mdb_cursor_open (
                    it->txn, it->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, &it->data, MDB_SET);
            if (it->rc == MDB_SUCCESS) mdb_cursor_count (it->cur, ct);

            mdb_cursor_close (it->cur);
            it->cur = NULL;

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

            *ct = stat.ms_entries;
        }
        log_debug ("Found %lu keys.", *ct);
    }

    it->rc = mdb_cursor_open (it->txn, it->store->dbi[IDX_SPO_C], &it->cur);
    if (it->rc != MDB_SUCCESS) {
        log_error ("Database error: %s", LSUP_strerror (it->rc));
        return LSUP_DB_ERR;
    }

    it->rc = mdb_cursor_get (it->cur, &it->key, &it->data, MDB_FIRST);
    /*
    mdb_cursor_close (it->cur);
    it->cur = NULL;
    */

    it->iter_op_fn = it_next_0bound;

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

    return LSUP_OK;
}


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

    log_debug ("Looking up 1 bound term: %lx", it->luk[0]);

    mdb_cursor_open (it->txn, it->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) {
            log_debug ("Counting in context: %lx", it->ck);
            MDBIterator *ct_it = NULL;
            MALLOC_GUARD (ct_it, LSUP_MEM_ERR);

            ct_it->luk[0] = it->luk[0];
            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;
            ct_it->ctx_cur = it->ctx_cur;

            LSUP_rc rc = lookup_1bound (idx0, ct_it, NULL);
            if (rc < 0) return rc;

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

            // Free the counter iterator without freeing the shared txn.
            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_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) {
        log_error ("Database error: %s", LSUP_strerror (it->rc));
        return LSUP_DB_ERR;
    }

    return LSUP_OK;
}


inline static LSUP_rc
lookup_2bound(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 = it->store->dbi[lookup_indices[i + 3]];
            log_debug (
                    "Looking up 2 bound in %s",
                    db_labels[lookup_indices[i + 3]]);

            break;
        }
    }

    if (dbi == 0) {
        log_error (
                "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];

    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;
            MALLOC_GUARD (ct_it, LSUP_MEM_ERR);

            ct_it->luk[0] = it->luk[0];
            ct_it->luk[1] = it->luk[1];
            ct_it->ck = it->ck;
            ct_it->store = it->store;
            ct_it->txn = it->txn;
            ct_it->ctx_cur = it->ctx_cur;
            lookup_2bound (idx0, idx1, ct_it, NULL);

            while (LSUP_mdbiter_next (ct_it, NULL, NULL) != LSUP_END) {
                ct[0] ++;
            }

            // Free the counter iterator without freeing the shared txn.
            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) {
        log_error ("Database error: %s", LSUP_strerror (it->rc));
        return LSUP_DB_ERR;
    }

    return LSUP_OK;
}


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

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

    if (it->ck != NULL_KEY) {
        it->rc = mdb_cursor_open (
                it->txn, it->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, it->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);

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

    mdb_cursor_close (it->cur);
    it->cur = NULL;

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

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

    return LSUP_OK;
}