volksdata/src/store_mdb.c

#include "lsup/store_mdb.h"

/**
 * Number of DBs defined. See MAIN_TABLE and LOOKUP_TABLE defines below.
 */
#define N_DB 10

/**
 * 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];
typedef struct mdbstore_iter_t MDBIterator;

/// Store state flags.
typedef enum {
    LSSTORE_OPEN         = 1<<0,                    ///< Env is open.
} StoreFlags;

/// Iterator state flags.
typedef enum {
    ITER_OPEN_TXN       = 1<<0,         /**< A transaction is open.
                                          *
                                          *  The iterator has begun a new
                                          *  transaction on initialization
                                          *  which needs to be closed.  If
                                          *  false, the iterator is using an
                                          *  existing transaction which will
                                          *  not be closed with
                                          *  #mdbiter_free().
                                          */
} IterFlags;

typedef enum {
    OP_ADD,
    OP_REMOVE,
} StoreOp;

typedef struct mdbstore_t {
    MDB_env *           env;            ///< Environment handle.
    MDB_dbi             dbi[N_DB];      ///< DB handles. Refer to DbIdx enum.
    StoreFlags          flags;          ///< Store state flags.
} 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)(MDBIterator *it);


/// Triple iterator.
typedef struct mdbstore_iter_t {
    MDBStore *          store;      ///< MDB store handle.
    IterFlags           flags;      ///< Iterator flags.
    MDB_txn *           txn;        ///< MDB transaction.
    MDB_cursor *        cur;        ///< MDB cursor.
    MDB_cursor *        ctx_cur;    ///< MDB c:spo index cursor.
    MDB_val             key;        ///< Internal data handler.
    MDB_val             data;       ///< Internal data handler.
    LSUP_TripleKey      spok;       ///< Triple to be populated with match.
    LSUP_Key *          ck;         /**< Context array.
                                      *
                                      *  This shall be populated for each
                                      * matching triple if requested. */
    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.
    LSUP_Key            luc;        ///< Ctx key to filter by. May be NULL_KEY.
    size_t              i;          ///< Internal counter for paged lookups.
    size_t              ct;         ///< Current count of records inserted or
                                    ///< results found.
    int                 rc;         ///< MDB_* return code for the next result.
} MDBIterator;


/*
 * Static variables.
 */

#define DUPSORT_MASK        MDB_DUPSORT
#define DUPFIXED_MASK       MDB_DUPSORT | MDB_DUPFIXED

/**
 * Main DBs. These are the master information containers.
 *
 * The number of entries must match the N_DB constant defined above.
 */
#define MAIN_TABLE \
/*          #ID pfx #DB label   #Flags  */ \
    ENTRY(  T_ST,   "t:st",     0               )   /* Key to ser. term */  \
    ENTRY(  SPO_C,  "spo:c",    DUPFIXED_MASK   )   /* Triple to context */ \
    ENTRY(  IDK_ID, "idk:id",   0               )   /* ID key to ID */      \

/**
 * Lookup DBs. These are indices and may be destroyed and rebuilt.
 */
#define LOOKUP_TABLE \
/*          #ID pfx  #DB label #Flags   */ \
    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 */    \

/**
 * 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(
        MDBStore *store, StoreOp op, LSUP_TripleKey spok, LSUP_Key ck,
        MDB_txn *txn);
static LSUP_rc mdbstore_add_term (void *h, const LSUP_Buffer *sterm, void *th);

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


static const char *
mdbstore_path_from_id (const char *id)
{
    // Set environment path.
    if (!id) id = getenv ("LSUP_MDB_STORE_URN");
    if (!id) {
        id = LSUP_MDB_STORE_URN;
        log_info (
            "`LSUP_MDB_STORE_URN' environment variable is not "
            "set. The default URN %s has been set as the store ID.", id
        );
    }
    if (strncmp ("file://", id, 7) != 0) {
        log_error ("MDB store ID must be in the `file://<abs_path>` format.");
        return NULL;
    }
    return id + 7;
}


/**
 * Store interface.
 */

/** @brief Create the MDB environment and databases on disk.
 *
 * This function takes care of creating the environment path if not existing,
 * and checking that it's a writable directory. If the path is not specified
 * in the LSUP_MDB_STORE_URN environment variable, a default directory is used.
 */
static LSUP_rc
mdbstore_setup (const char *id, bool clear)
{
    const char *path = mdbstore_path_from_id (id);
    if (!path) return LSUP_VALUE_ERR;

    // If the directory exists (unless clear == true), do nothing.
    if (clear) rm_r (path);
    LSUP_rc rc = mkdir_p (path, ENV_DIR_MODE);
    log_info ("Create dir rc: %s", LSUP_strerror (rc));
    PRCCK (rc);

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

    RCCK (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;
    RCCK (mdb_txn_begin (env, NULL, 0, &txn));
    MDB_dbi dbi;
    for (int i = 0; i < N_DB; i++) {
        LOG_TRACE("Creating DB %s", db_labels[i]);
        RCCK (
            mdb_dbi_open (txn, db_labels[i], db_flags[i] | MDB_CREATE, &dbi)
        );
    }

    // Bootstrap the permanent store with initial data.
    MDB_stat stat;
    CHECK (mdb_dbi_open (
                txn, db_labels[IDX_T_ST], db_flags[IDX_T_ST], &dbi), fail);
    CHECK (mdb_stat (txn, dbi, &stat), fail);

    if (stat.ms_entries == 0) {
        LOG_DEBUG ("Loading initial data into %s", path);
        // Index default context.
        MDB_cursor *cur;
        CHECK (mdb_cursor_open (txn, dbi, &cur), fail);
        LSUP_Key k = LSUP_buffer_hash (LSUP_default_ctx_buf);
        MDB_val key, data;
        key.mv_data = &k;
        key.mv_size = sizeof (k);

        data.mv_data = LSUP_default_ctx_buf->addr;
        data.mv_size = LSUP_default_ctx_buf->size;

        LSUP_rc db_rc = mdb_cursor_put (cur, &key, &data, 0);
        CHECK (db_rc, fail);
    }

    mdb_txn_commit (txn);
    mdb_env_close (env);

    return clear ? LSUP_OK : rc;

fail:
    if (rc >= 0) rc = LSUP_DB_ERR;
    mdb_txn_abort (txn);
    return rc;
}


/** @brief Open an MDB store.
 *
 * The store must have been set up with #mdbstore_setup.
 *
 * Some environment variables affect various store parameters:
 *
 * - LSUP_MDB_MAPSIZE Long int specifying the size of the memory map. Usually
 *   it is not necessary to modify this, unless one is operating under memory
 *   and disk constraints. The default map size is 1Tb.
 */
static void *
mdbstore_new (const char *id, size_t _unused)
{
    (void) _unused;
    const char *path = mdbstore_path_from_id (id);
    if (!path) return NULL;

    MDBStore *store;
    CALLOC_GUARD (store, NULL);

    RCNL (mdb_env_create (&store->env));
    MDB_txn *txn = 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, "%zu", &mapsize);
    log_info (
            "Setting environment map size at %s to %zu Mb.",
            path, mapsize / 1024 / 1024);
    CHECK (mdb_env_set_mapsize (store->env, mapsize), fail);
    CHECK (mdb_env_set_maxdbs (store->env, N_DB), fail);
    CHECK (mdb_env_open (store->env, path, 0, ENV_FILE_MODE), fail);

    // Assign DB handles to store->dbi.
    CHECK (mdb_txn_begin (store->env, NULL, 0, &txn), fail);
    for (int i = 0; i < N_DB; i++)
        CHECK (mdb_dbi_open (
                txn, db_labels[i], db_flags[i], store->dbi + i), fail);

    store->flags |= LSSTORE_OPEN;
    mdb_txn_commit (txn);
    txn = NULL;

    return store;

fail:
    if (txn) mdb_txn_abort (txn);
    mdb_env_close (store->env);

    return NULL;
}


static void
mdbstore_free (void *h)
{
    MDBStore *store = h;
    if (store->flags & 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);
    }

    free (store);
}


#if 0
static char *
mdbstore_id (const void *h)
{
    const MDBStore *store = h;
    const char *path;
    mdb_env_get_path (store->env, &path);

    return strcat ("file://", path);
}
#endif


static LSUP_rc
mdbstore_stat (const MDBStore *store, MDB_stat *stat)
{
    if (!(store->flags & LSSTORE_OPEN)) 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;
}


static size_t
mdbstore_size (const void *h)
{
    const MDBStore *store = h;
    // Size is calculated outside of any pending write txn.

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

    return stat.ms_entries;
}


static LSUP_rc
mdbstore_txn_begin (void *h, int flags, void **th)
{
    MDBStore *store = h;

    RCCK (mdb_txn_begin (store->env, NULL, flags, (MDB_txn **) th));

    return LSUP_OK;
}


static LSUP_rc
mdbstore_txn_commit (void *th)
{
    RCCK (mdb_txn_commit ((MDB_txn *) th));

    return LSUP_OK;
}


static void
mdbstore_txn_abort (void *th)
{ mdb_txn_abort ((MDB_txn *) th); }


static void *
mdbiter_txn (void *h)
{ return ((MDBIterator *) h)->txn; }


/** @brief Begin an add loop.
 *
 * @sa #store_add_init_fn_t
 *
 * @param[in] th Previously opened MDB_txn handle, if the add loop shall be
 *  run within an enclosing transaction. The transaction must be read-write.
 *  The operation will always open a new transaction that is closed with
 *  #mdbstore_add_done() or #mdbstore_add_abort(). If this parameter is not
 *  NULL, the loop transaction will have the passed txn set as its parent.
 */
static void *
mdbstore_add_init (void *h, const LSUP_Buffer *sc, void *th)
{
    MDBStore *store = h;
    /* 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;

    CHECK (mdb_txn_begin (store->env, (MDB_txn *) th, 0, &it->txn), fail);

    if (sc) {
        // Store context if it's not the default one.
        it->luc = LSUP_buffer_hash (sc);

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

        int db_rc = mdb_put (
                it->txn, it->store->dbi[IDX_T_ST],
                &it->key, &it->data, MDB_NOOVERWRITE);
        if (db_rc != MDB_SUCCESS && db_rc != MDB_KEYEXIST) {
            log_error (LSUP_strerror (db_rc));
            mdb_txn_abort (it->txn);
            return NULL;
        }
    } else {
        LOG_DEBUG("No context passed to iterator, using default.");
        it->luc = LSUP_buffer_hash (LSUP_default_ctx_buf);
    }

    return it;

fail:
    free (it);

    return NULL;
}


/*
 * NOTE: at the moment #mdbstore_remove() or another
 * #mdbstore_init() cannot be called between #mdbstore_add_init and
 * #mdbstore_add_abort or #mdbstore_add_done. FIXME
 *
 */
static LSUP_rc
mdbstore_add_iter (void *h, const LSUP_BufferTriple *sspo)
{
    if (UNLIKELY (!h)) return LSUP_VALUE_ERR;

    MDBIterator *it = h;
    int db_rc = LSUP_NOACTION;
    LSUP_TripleKey spok = NULL_TRP;

    // Add triple terms.
    for (int i = 0; i < 3; i++) {
        LSUP_Buffer *st = LSUP_btriple_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->txn, it->store->dbi[IDX_T_ST],
                &it->key, &it->data, MDB_NOOVERWRITE);
        if (db_rc != MDB_SUCCESS && db_rc != MDB_KEYEXIST) {
            LOG_RC (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->luc);

    // 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->luc;
    it->data.mv_size = it->luc == NULL_KEY ? 0 : KLEN;

    db_rc = mdb_put(
            it->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->luc, it->txn);
    if (rc == LSUP_OK) it->i++;

    return rc;
}


static LSUP_rc
mdbstore_add_done (void *h)
{
    MDBIterator *it = h;
    LSUP_rc rc = LSUP_OK;

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

    free (it);

    return rc;
}


static void
mdbstore_add_abort (void *h)
{
    MDBIterator *it = h;
    mdb_txn_abort (it->txn);

    free (it);
}


static LSUP_rc
key_to_sterm (
        MDBStore *store, MDB_txn *txn, 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 (txn, store->dbi[IDX_T_ST], &key_v, &data_v);

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

    return rc;
}


static void *
mdbstore_lookup (
        void *h, const LSUP_Buffer *ss, const LSUP_Buffer *sp,
        const LSUP_Buffer *so, const LSUP_Buffer *sc, void *th, size_t *ct)
{
    LSUP_TripleKey spok = {
        LSUP_buffer_hash (ss),
        LSUP_buffer_hash (sp),
        LSUP_buffer_hash (so),
    };

    MDBIterator *it;
    CALLOC_GUARD (it, NULL);

    it->store = h;
    it->luc = LSUP_buffer_hash (sc);
    LOG_DEBUG("Lookup context: %lx", it->luc);

    if (ct) *ct = 0;

    uint8_t idx0, idx1;

    if (th) it->txn = th;
    else if (!it->txn) {
        // Start RO transaction if not in a write txn already.
        it->rc = mdb_txn_begin (it->store->env, NULL, MDB_RDONLY, &it->txn);
        if (it->rc != MDB_SUCCESS) {
            log_error ("Database error in lookup: %s", LSUP_strerror (it->rc));
            return NULL;
        }
        LOG_TRACE ("Opening new MDB transaction @%p", it->txn);
        it->flags |= ITER_OPEN_TXN;
    }

    // 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];
        PRCNL (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;
            PRCNL (lookup_2bound (idx0, idx1, it, ct));

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

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

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

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

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

    return it;
}


/** @brief Get next iterator key.
 *
 * ckset is filled with an array of contexts that the triple appears
 * in, if not NULL.
 */
static LSUP_rc
mdbiter_next_key (MDBIterator *it)
{
    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);
    MDB_val key, data;
    int db_rc;

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

    if (it->luc) {
        rc = LSUP_NORESULT;  // Flow control value, will never be returned.
        do {
            //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).
             */
            LOG_TRACE (
                    "Found spok: {%lx, %lx, %lx}",
                    it->spok[0], it->spok[1], it->spok[2]);

            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 in context: %x", it->luc);

            } else if (db_rc == MDB_NOTFOUND) {
                LOG_TRACE("Triple not found in context: %x", it->luc);
                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;
            }
        } while (rc == LSUP_NORESULT);

    } else {
        LOG_TRACE (
                "Found spok in any context: {%lx, %lx, %lx}",
                it->spok[0], it->spok[1], it->spok[2]);

        rc = LSUP_OK;
    }

    // Get all contexts for a triple.
    key.mv_data = it->spok;
    db_rc = mdb_cursor_get (it->ctx_cur, &key, &data, MDB_SET_KEY);
    if (db_rc != MDB_SUCCESS) {
        log_error ("No context found for triple!");
        return LSUP_DB_ERR;
    }

    size_t ct;
    db_rc = mdb_cursor_count (it->ctx_cur, &ct);
    if (db_rc != MDB_SUCCESS) return LSUP_DB_ERR;
    // 1 spare for sentinel. Always allocated even on zero matches.
    LSUP_Key *tmp_ck = realloc (it->ck, sizeof (*it->ck) * (ct + 1));
    if (!tmp_ck) return LSUP_MEM_ERR;
    it->ck = tmp_ck;

    size_t i = 0;
    do {
        //LOG_TRACE("Copying to slot #%lu @%p", i, it->ck + i);
        memcpy (it->ck + i++, data.mv_data, sizeof (*it->ck));
    } while (
            mdb_cursor_get (it->ctx_cur, &key, &data, MDB_NEXT_DUP)
            == MDB_SUCCESS);
    //LOG_TRACE("setting sentinel @%p", it->ck + i);
    it->ck[i] = NULL_KEY;

    return rc;
}


static LSUP_rc
mdbiter_next (
        void *h, LSUP_BufferTriple *sspo, LSUP_Buffer **ctx_p)
{
    MDBIterator *it = h;
    LSUP_rc rc = mdbiter_next_key (it);

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

            // TODO error handling.
        }

        // Contexts for current triple.
        if (ctx_p) {
            // Preallocate.
            size_t i = 0;
            while (it->ck[i++]); // Include sentinel in count.
            LSUP_Buffer *ctx;
            LOG_TRACE("Allocating %lu context buffers + sentinel.", i - 1);
            ctx = malloc(i * sizeof (*ctx));
            if (!ctx) return LSUP_MEM_ERR;

            for (i = 0; it->ck[i]; i++)
                key_to_sterm (it->store, it->txn, it->ck[i], ctx + i);
            memset (ctx + i, 0, sizeof (*ctx)); // Sentinel

            // TODO error handling.
            *ctx_p = ctx;
        }
    }

    return rc;
}


static void
mdbiter_free (void *h)
{
    MDBIterator *it = h;
    if (!it) return;

    if (it->cur) mdb_cursor_close (it->cur);
    if (it->ctx_cur) mdb_cursor_close (it->ctx_cur);
    if (it->flags & ITER_OPEN_TXN) mdb_txn_abort (it->txn);
    free (it->ck);

    free (it);
}


static LSUP_rc
mdbstore_update_ctx (
        void *h, const LSUP_Buffer *old_c, const LSUP_Buffer *new_c, void *th)
{
    LSUP_rc rc = LSUP_NOACTION;
    MDBStore *store = h;
    unsigned char *trp_data = NULL;

    LSUP_Key
        old_ck = LSUP_buffer_hash (old_c),
        new_ck = LSUP_buffer_hash (new_c);
    // lu_key, lu_data look up all triples with old context in c:spo, and
    // replace old c with new c.
    MDB_txn
        *p_txn = th,
        *txn;
    CHECK (
            rc = mdb_txn_begin (store->env, p_txn, 0, &txn),
            finally);

    MDB_cursor *i_cur, *d_cur;
    CHECK (
            rc = mdb_cursor_open (txn, store->dbi[IDX_C_SPO], &i_cur),
            close_txn);
    // TODO error handling.

    MDB_val key, data;

    // Return error if the graph URI already exists.
    key.mv_data = &new_ck;
    key.mv_size = KLEN;
    rc = mdb_cursor_get (i_cur, &key, &data, MDB_FIRST_DUP);
    if (rc == MDB_SUCCESS) {
        log_error (
                "Context key %lu already exists. Not replacing old graph.",
                new_ck);
        rc = LSUP_CONFLICT;
        goto close_i;
    }

    // Add new context term.
    CHECK (rc = mdbstore_add_term (store, new_c, txn), close_i);

    key.mv_data = &old_ck;
    // Count triples in cursor.
    rc = mdb_cursor_get (i_cur, &key, &data, MDB_SET);
    if (rc == MDB_NOTFOUND) {
        log_info ("No triples found associated with old context.");
        rc = LSUP_NOACTION;
        goto close_i;
    }
    if (rc != MDB_SUCCESS) {
        rc = LSUP_DB_ERR;
        goto close_i;
    }

    // From here on, it can only be LSUP_OK or error.
    rc = LSUP_OK;
    size_t trp_ct;
    CHECK (rc = mdb_cursor_count (i_cur, &trp_ct), close_i);
    trp_data = malloc (trp_ct * TRP_KLEN);
    if (UNLIKELY (!trp_data)) {
        rc = LSUP_MEM_ERR;
        goto close_i;
    }

    // Copy triple data as one block to temp buffer so that entries can be
    // deleted while cursors are active.
    rc = mdb_cursor_get (i_cur, &key, &data, MDB_GET_MULTIPLE);
    if (rc != MDB_SUCCESS) {
        rc = rc == MDB_NOTFOUND ? LSUP_NOACTION : LSUP_DB_ERR;
        goto close_i;
    }
    size_t loc_cur = 0;
    do {
        memcpy (trp_data + loc_cur, data.mv_data, data.mv_size);
        loc_cur += data.mv_size;
    } while (mdb_cursor_get (
                i_cur, &key, &data, MDB_NEXT_MULTIPLE) == MDB_SUCCESS);

    // Zap c:spo entries in one go.
    key.mv_data = &old_ck;
    key.mv_size = KLEN;
    data.mv_size = TRP_KLEN;
    CHECK (rc = mdb_cursor_get (i_cur, &key, NULL, MDB_SET), close_i);
    CHECK (rc = mdb_cursor_del (i_cur, MDB_NODUPDATA), close_i);

    // Re-ad c:spo data individually.
    key.mv_data = &new_ck;
    for (size_t i = 0; i < trp_ct; i++) {
        data.mv_data = trp_data + i * data.mv_size;
        CHECK (
                rc = mdb_cursor_put (i_cur, &key, &data, MDB_APPENDDUP),
                close_i);
    }
    // Re-add c:spo data in bulk from buffer with new context.
    // FIXME this is not working. Replaced by the for loop above.
    /*
    MDB_val data_block[] = {
        { .mv_data = &new_ck, .mv_size = TRP_KLEN },
        { .mv_data = NULL, .mv_size = trp_ct },
    };
    db_rc = mdb_cursor_put (i_cur, &key, data_block, MDB_MULTIPLE);
    */

    // Main table.
    // Replace spo:c values one by one.
    CHECK (rc = mdb_cursor_open (txn, store->dbi[IDX_SPO_C], &d_cur), close_i);
    key.mv_size = TRP_KLEN;
    data.mv_size = KLEN;
    for (size_t i = 0; i < trp_ct; i++) {
        key.mv_data = trp_data + i * key.mv_size;
        data.mv_data = &old_ck;
        CHECK (
                rc = mdb_cursor_get (d_cur, &key, &data, MDB_GET_BOTH),
                close_d);
        CHECK (rc = mdb_cursor_del (d_cur, 0), close_d);
        data.mv_data = &new_ck;
        CHECK (
                rc = mdb_cursor_put (d_cur, &key, &data, MDB_NOOVERWRITE),
                close_d);
    }

close_d:
    mdb_cursor_close (d_cur);
close_i:
    mdb_cursor_close (i_cur);
close_txn:
    if (rc == LSUP_OK) {
        RCCK (mdb_txn_commit (txn));
    } else mdb_txn_abort (txn);

    if (trp_data) free (trp_data);
finally:

    return rc;
}


static LSUP_rc
mdbstore_remove (
        void *h, const LSUP_Buffer *ss, const LSUP_Buffer *sp,
        const LSUP_Buffer *so, const LSUP_Buffer *sc, void *th, size_t *ct)
{
    MDBStore *store = h;
    LSUP_rc
        rc = LSUP_NOACTION,
        db_rc = 0;

    LSUP_Key ck = NULL_KEY;

    if (sc == NULL) sc = LSUP_default_ctx_buf;
    ck = LSUP_buffer_hash (sc);

    MDB_txn *txn;
    mdb_txn_begin (store->env, (MDB_txn *) th, 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;

    // The lookup operates within the current (bottom) write transaction.
    MDBIterator *it = mdbstore_lookup (store, ss, sp, so, sc, txn, ct);
    if (UNLIKELY (!it)) return LSUP_DB_ERR;
    if (ct) LOG_DEBUG("Found %lu triples to remove.", *ct);

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

        db_rc = mdb_cursor_get (dcur, &spok_v, &ck_v, MDB_GET_BOTH);
        if (db_rc == MDB_NOTFOUND) continue;
        if (UNLIKELY (db_rc != MDB_SUCCESS)) goto fail;

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

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

        // Restore ck address after each delete.
        spok_v.mv_data = it->spok;
        ck_v.mv_data = &ck;

        // Delete c:spo entry.
        db_rc = mdb_cursor_get (icur, &ck_v, &spok_v, MDB_GET_BOTH);
        if (db_rc == MDB_NOTFOUND) continue;
        if (UNLIKELY (db_rc != MDB_SUCCESS)) goto fail;

        db_rc = mdb_cursor_del (icur, 0);
        if (UNLIKELY (db_rc != MDB_SUCCESS)) goto fail;

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

        // If there are no more contexts associated with this triple,
        // remove from indices.
        db_rc = mdb_cursor_get (dcur, &spok_v, NULL, MDB_SET);
        if (db_rc == MDB_SUCCESS) continue;
        if (UNLIKELY (db_rc != MDB_NOTFOUND)) goto fail;

        rc = index_triple (store, OP_REMOVE, it->spok, ck, txn);
    }

    mdbiter_free (it);

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

    return rc;

fail:
    mdb_txn_abort (txn);

    log_error ("Database error: %s", LSUP_strerror (db_rc));

    return rc == LSUP_TXN_ERR ? rc : LSUP_DB_ERR;
}


#if 0
static int
mdbstore_tkey_exists (MDBStore *store, LSUP_Key tkey)
{
    int db_rc, rc;
    MDB_val key, data;
    key.mv_data = &tkey;
    key.mv_size = KLEN;

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

    MDB_cursor *cur = NULL;
    mdb_cursor_open (txn, store->dbi[IDX_T_ST], &cur);

    db_rc = mdb_cursor_get (cur, &key, &data, MDB_SET);

    if (db_rc == MDB_SUCCESS) rc = 1;
    else if (db_rc == MDB_NOTFOUND) rc = 0;
    else {
        log_error ("DB error: %s", LSUP_strerror (db_rc));
        rc = LSUP_DB_ERR;
    }

    if (cur) mdb_cursor_close (cur);
    if (txn) mdb_txn_abort (txn);

    return rc;
}
#endif


/** @brief Add a term to the store.
 *
 * @param[in] h #MDBStore handle.
 *
 * @param[in] sterm Serialized term to store.
 *
 * @param[in] th transaction handle. It may be NULL, in which case a new
 * transaction is opened and closed for the operation.
 *
 * @return LSUP_OK on success; <0 on error.
 */
static LSUP_rc
mdbstore_add_term (void *h, const LSUP_Buffer *sterm, void *th)
{
    //LOG_TRACE("Adding term to MDB store: %s", sterm->addr);
    MDBStore *store = h;
    int db_rc;
    MDB_val key, data;

    MDB_txn *txn;
    // If an active transaction was passed, use it, otherwise open and
    // close a new one.
    bool borrowed_txn = (th != NULL);
    if (borrowed_txn) txn = th;
    else RCCK (mdb_txn_begin (store->env, NULL, 0, &txn));

    MDB_cursor *cur;
    CHECK (mdb_cursor_open (txn, store->dbi[IDX_T_ST], &cur), fail);

    LSUP_Key k = LSUP_buffer_hash (sterm);
    key.mv_data = &k;
    key.mv_size = sizeof (k);

    data.mv_data = sterm->addr;
    data.mv_size = sterm->size;

    db_rc = mdb_cursor_put (cur, &key, &data, MDB_NOOVERWRITE);
    if (db_rc != MDB_KEYEXIST) CHECK (db_rc, fail);

    if (!borrowed_txn) CHECK (db_rc = mdb_txn_commit (txn), fail);

    return LSUP_OK;

fail:
    if (!borrowed_txn) mdb_txn_abort (txn);
    LOG_TRACE("Aborted txn for adding term.");
    return LSUP_DB_ERR;
}


LSUP_Buffer **
mdbstore_ctx_list (void *h, void *th)
{
    MDBStore *store = h;
    LSUP_rc db_rc;
    MDB_txn *txn;
    if (th) txn = th;
    else CHECK (mdb_txn_begin (store->env, NULL, MDB_RDONLY, &txn), fail);


    MDB_cursor *cur;
    CHECK (mdb_cursor_open (txn, store->dbi[IDX_C_SPO], &cur), fail);
    MDB_val key, data;
    db_rc = mdb_cursor_get (cur, &key, &data, MDB_FIRST);

    LSUP_Buffer **tdata = NULL;
    size_t i = 0;
    while (db_rc == MDB_SUCCESS) {
        tdata = realloc (tdata, (i + 1) * sizeof (*tdata));
        if (UNLIKELY (!tdata)) goto fail;
        tdata[i] = BUF_DUMMY;
        LSUP_Key tkey = *(LSUP_Key*)key.mv_data;
        CHECK (key_to_sterm (store, txn, tkey, tdata[i]), fail);
        db_rc = mdb_cursor_get (cur, &key, &data, MDB_NEXT_NODUP);
        i++;
    }

    tdata = realloc (tdata, i * sizeof (data));
    tdata[i] = NULL;  // Sentinel
    mdb_cursor_close (cur);
    if (txn != th && txn != NULL) mdb_txn_abort (txn);

    return tdata;

fail:
    if (txn != th && txn != NULL) mdb_txn_abort (txn);
    if (tdata) free (tdata);
    return NULL;
}


const LSUP_StoreInt mdbstore_int = {
    .name           = "MDB Store",
    .features       = LSUP_STORE_PERM | LSUP_STORE_CTX | LSUP_STORE_IDX
                      | LSUP_STORE_TXN | LSUP_STORE_COW,

    .setup_fn       = mdbstore_setup,
    .new_fn         = mdbstore_new,
    .free_fn        = mdbstore_free,

    .size_fn        = mdbstore_size,

    .txn_begin_fn   = mdbstore_txn_begin,
    .txn_commit_fn  = mdbstore_txn_commit,
    .txn_abort_fn   = mdbstore_txn_abort,
    .iter_txn_fn    = mdbiter_txn,

    .add_init_fn    = mdbstore_add_init,
    .add_iter_fn    = mdbstore_add_iter,
    .add_abort_fn   = mdbstore_add_abort,
    .add_done_fn    = mdbstore_add_done,
    .add_term_fn    = mdbstore_add_term,

    .update_ctx_fn  = mdbstore_update_ctx,

    .lookup_fn      = mdbstore_lookup,
    .lu_next_fn     = mdbiter_next,
    .lu_free_fn     = mdbiter_free,

    .remove_fn      = mdbstore_remove,

    .ctx_list_fn    = mdbstore_ctx_list,
};


/* * * 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.
 * @param[in] th Transaction handle. This MUST be a valid pointer to an open
 *  RW transaction.
 */
static LSUP_rc
index_triple(
        MDBStore *store, StoreOp op, LSUP_TripleKey spok, LSUP_Key ck,
        MDB_txn *txn)
{
    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 (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(
                    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(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(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 (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 (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->luc != 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->luc;
            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 triples.", *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->luc != NULL_KEY) {
            LOG_DEBUG("Counting in context: %lx", it->luc);
            MDBIterator *ct_it;
            MALLOC_GUARD (ct_it, LSUP_MEM_ERR);
            /*
            memcpy (ct_it, it, sizeof (*ct_it));
            */

            ct_it->store = it->store;
            ct_it->txn = it->txn;
            ct_it->ctx_cur = it->ctx_cur;
            ct_it->key = it->key;
            ct_it->data = it->data;
            ct_it->ck = NULL;
            ct_it->luk[0] = it->luk[0];
            ct_it->luc = it->luc;
            ct_it->i = 0;

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

            LSUP_rc db_rc;
            while (LSUP_END != (db_rc = mdbiter_next_key (ct_it))) {
                if (UNLIKELY (db_rc < 0)) return db_rc;
                (*ct)++;
            }

            // Free the counter iterator without freeing the shared txn.
            if (ct_it->cur) mdb_cursor_close (ct_it->cur);
            free (ct_it->ck);
            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);
        }
        LOG_DEBUG("Found %lu triples.", *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->luc != NULL_KEY) {
            MDBIterator *ct_it;
            MALLOC_GUARD (ct_it, LSUP_MEM_ERR);

            ct_it->store = it->store;
            ct_it->txn = it->txn;
            ct_it->ctx_cur = it->ctx_cur;
            ct_it->ck = NULL;
            ct_it->luk[0] = it->luk[0];
            ct_it->luk[1] = it->luk[1];
            ct_it->luc = it->luc;
            ct_it->i = 0;

            lookup_2bound (idx0, idx1, ct_it, NULL);

            while (mdbiter_next_key (ct_it) != LSUP_END) (*ct) ++;

            // Free the counter iterator without freeing the shared txn.
            if (ct_it->cur) mdb_cursor_close (ct_it->cur);
            free (ct_it->ck);
            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);
        }
        LOG_DEBUG("Found %lu triples.", *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->luc != 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->luc;
        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;
}