volksdata/src/store_mdb.c

#include <ftw.h>

#include "store_mdb.h"

/*
 * TODO At the moment up to 64-bit key / hash values are allowed. Later on,
 * 128-bit keys should be allowed by compile options, and that will no longer
 * be compatible with integer keys and data. When 128-bit keys are supported,
 * integer keys should remain available for code compiled with 64-bit keys.
 */
#define INT_KEY_MASK        MDB_INTEGERKEY
#define INT_DUP_KEY_MASK    MDB_DUPSORT | MDB_DUPFIXED | MDB_INTEGERKEY
#define INT_DUPDATA_MASK    MDB_DUPSORT | MDB_DUPFIXED | MDB_INTEGERDUP

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

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


typedef enum {
    OP_ADD,
    OP_REMOVE,
} StoreOp;


struct MDBStore {
    MDB_env *           env;        // Environment handle.
    MDB_txn *           txn;        // Current transaction. If RW, it may have
                                    //  nested transactions.
    MDB_dbi             dbi[N_DB];  // DB handles. Refer to DbIdx enum.
    LSUP_Buffer *       default_ctx;// Default context as a serialized URI.
    StoreState          state;      // Store state (initialized, open etc.)
};


/** @brief Common match callback arguments.
 */
struct MatchArgs {
    LSUP_Key luks[2];
    uint8_t idx0, idx1;
    size_t *ct;
    mdb_store_match_fn_t callback_fn;
    void *ctx;
};

static LSUP_Buffer *default_ctx = NULL; // Default context URI for quad store.

/**
 * Main DBs. These are the master information containers.
 *
 * Data columns are: identifier prefix, DB label, flags.
 */
#define MAIN_TABLE \
    ENTRY(  T_ST,    "t:st",    INT_KEY_MASK    )   /* Key to ser. term */  \
    ENTRY(  SPO_C,   "spo:c",   INT_DUPDATA_MASK)   /* Triple to context */ \
    ENTRY(  C_,      "c:",      INT_KEY_MASK    )   /* Track empty contexts */\
    ENTRY(  PFX_NS,  "pfx:ns",  0               )   /* Prefix to NS */      \

/**
 * Lookup DBs. These are indices and may be destroyed and rebuilt.
 */
#define LOOKUP_TABLE \
    ENTRY(  S_PO,    "s:po",    INT_DUP_KEY_MASK)   /* 1-bound lookup */    \
    ENTRY(  P_SO,    "p:so",    INT_DUP_KEY_MASK)   /* 1-bound lookup */    \
    ENTRY(  O_SP,    "o:sp",    INT_DUP_KEY_MASK)   /* 1-bound lookup */    \
    ENTRY(  PO_S,    "po:s",    INT_DUPDATA_MASK)   /* 2-bound lookup */    \
    ENTRY(  SO_P,    "so:p",    INT_DUPDATA_MASK)   /* 2-bound lookup */    \
    ENTRY(  SP_O,    "sp:o",    INT_DUPDATA_MASK)   /* 2-bound lookup */    \
    ENTRY(  C_SPO,   "c:spo",   INT_DUP_KEY_MASK)   /* Context lookup */    \
    ENTRY(  NS_PFX,  "ns:pfx",  0               )   /* NS to prefix */      \

/**
 * DB labels. They are prefixed with DB_
 */
#define ENTRY(a, b, c) static const DbLabel DB_##a = b;
MAIN_TABLE
LOOKUP_TABLE
#undef ENTRY

/**
 * Numeric index of each DB. Prefixed with IDX_
 *
 * These index numbers are referred to in all the arrays defeined below. They
 * are independent from the LMDB dbi values which are considered opaque here.
 */
typedef enum {
#define ENTRY(a, b, c) IDX_##a,
    MAIN_TABLE
    LOOKUP_TABLE
#undef ENTRY
} DBIdx;

/**
 * DB labels.
 */
static const char *db_labels[N_DB] = {
#define ENTRY(a, b, c) DB_##a,
    MAIN_TABLE
    LOOKUP_TABLE
#undef ENTRY
};

/**
 * DB flags. These are aligned with the dbi_labels index.
 */
static const unsigned int db_flags[N_DB] = {
#define ENTRY(a, b, c) c,
    MAIN_TABLE
    LOOKUP_TABLE
#undef ENTRY
};

/**
 * 1-bound and 2-bound lookup indices.
 *
 * N.B. Only the first 6 (1-bound and 2-bound term lookup) are used.
 * The others are added just because they belong logically to the lookup table.
 */
static DBIdx lookup_indices[9] = {
#define ENTRY(a, b, c) IDX_##a,
    LOOKUP_TABLE
#undef ENTRY
};

/**
 * Order in which keys are looked up if two terms are bound.
 * The indices with the smallest average number of values per key should be
 * looked up first.
 *
 * 0 = s:po
 * 1 = p:so
 * 2 = o:sp
 */
static const uint8_t lookup_rank[3] = {0, 2, 1};

static const uint8_t lookup_ordering_1bound[3][3] = {
    {0, 1, 2}, // s:po
    {1, 0, 2}, // p:so
    {2, 0, 1}, // o:sp
};

static const uint8_t lookup_ordering_2bound[3][3] = {
    {1, 2, 0}, // po:s
    {0, 2, 1}, // so:p
    {0, 1, 2}, // sp:o
};


/**
 * Static prototypes.
 */
static int index_triple(
        LSUP_MDBStore *store, StoreOp op,
        LSUP_TripleKey spok, LSUP_Key ck);

inline static LSUP_rc match_callback_0bound(
        struct MDBStore *store, struct MatchArgs *args);
inline static LSUP_rc match_callback_1bound(
        struct MDBStore *store, struct MatchArgs *args);
inline static LSUP_rc match_callback_2bound(
        struct MDBStore *store, struct MatchArgs *args);
inline static LSUP_rc match_callback_3bound(
        struct MDBStore *store, struct MatchArgs *args);
/* TODO
inline static int check_txn_open(MDB_txn *txn, bool write);
*/
/* TODO
static int unlink_cb(
        const char *fpath, const struct stat *sb,
        int typeflag, struct FTW *ftwbuf);
static int rmrf(char *path);
*/

/**
 * API.
 */

LSUP_rc
LSUP_store_setup(char **path/*, bool clear*/) // TODO clear
{
    int rc;

    // Set environment path.
    if (path == NULL && (*path = getenv("LSUP_STORE_PATH")) == NULL) {
        // 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);
    }

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


LSUP_MDBStore *
LSUP_store_new(const char *path, 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);

    if (default_ctx == NULL) store->default_ctx = NULL;
    else {
        CRITICAL(store->default_ctx = malloc(sizeof(LSUP_Buffer)));
        LSUP_buffer_copy(store->default_ctx, default_ctx);
    }

    // Set map size.
    size_t mapsize;
    char *env_mapsize = getenv("LSUP_MDB_MAPSIZE");

    if (env_mapsize == NULL) {
        mapsize = DEFAULT_MAPSIZE;
    } else {
        sscanf(env_mapsize, "%lu", &mapsize);
    }
    TRACE("mapsize rc: %d", rc);
    if(rc != MDB_SUCCESS) return NULL;

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

    rc = mdb_env_open(store->env, path, 0, ENV_FILE_MODE);
    if (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 (rc != MDB_SUCCESS) {
            mdb_txn_abort(txn);
            return NULL;
        }
    }

    mdb_txn_commit(txn);

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

    return store;
}


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


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

    MDB_stat stat;
    mdb_stat(store->txn, store->dbi[IDX_SPO_C], &stat);

    return stat.ms_entries;
}


LSUP_rc
LSUP_store_add(
        LSUP_MDBStore *store, const LSUP_Buffer *sc,
        const LSUP_SerTriple *data, const size_t data_size)
{
    MDB_val key_v, data_v;

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

    // Take care of context first.
    // Serialize and hash.
    LSUP_Key ck = NULL_KEY;

    // Assign default store context if existing and ctx is NULL.
    if (store->default_ctx && !sc) sc = store->default_ctx;

    if (sc) {
        ck = LSUP_sterm_to_key(sc);

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

        mdb_put(
                store->txn, store->dbi[IDX_T_ST],
                &key_v, &data_v, MDB_NOOVERWRITE);
    }

    LSUP_rc rc = LSUP_NOACTION;
    int db_rc;
    for (size_t i = 0; i < data_size; i++) {
        const LSUP_SerTriple *sspo = data + i;
        LSUP_TripleKey spok = NULL_TRP;

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

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

            spok[j] = LSUP_sterm_to_key(st);

            key_v.mv_data = spok + j;
            key_v.mv_size = KLEN;
            data_v.mv_data = st->addr;
            data_v.mv_size = st->size;

            db_rc = mdb_put(
                    store->txn, store->dbi[IDX_T_ST],
                    &key_v, &data_v, MDB_NOOVERWRITE);
            if (db_rc == MDB_SUCCESS) rc = LSUP_OK;
            else if (db_rc != MDB_KEYEXIST) goto _add_close_txn;
        }

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

        // Insert spo:c.
        key_v.mv_data = spok;
        key_v.mv_size = TRP_KLEN;

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


        db_rc = mdb_put(
                store->txn, store->dbi[IDX_SPO_C],
                &key_v, &data_v, MDB_NODUPDATA);
        if (db_rc == MDB_SUCCESS) rc = LSUP_OK;
        else if (db_rc != MDB_KEYEXIST) goto _add_close_txn;

        // Index.
        PCHECK(index_triple(store, OP_ADD, spok, ck), db_rc, _add_close_txn);
    }

_add_close_txn:
    // Only return commit rc if it fails.
    if (txn_pending) {
        if (rc == LSUP_OK) {
            if((db_rc = mdb_txn_commit(store->txn)) != MDB_SUCCESS) {
                mdb_txn_abort(store->txn);
                rc = db_rc;
            }
        } else mdb_txn_abort(store->txn);

        store->txn = NULL;
    }

    return rc;
}


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


/*
LSUP_Key
LSUP_store_get_key(
        LSUP_MDBStore *store, const LSUP_SerTerm *sterm)
{

}
*/


LSUP_rc
LSUP_store_key_to_sterm(
        LSUP_MDBStore *store, const LSUP_Key key, LSUP_SerTerm *sterm)
{
    LSUP_rc rc = LSUP_NORESULT;

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

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

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

    if (mdb_rc == MDB_SUCCESS) {
        sterm->addr = data_v.mv_data;
        sterm->size = data_v.mv_size;
        rc = LSUP_OK;
    }
    else if (UNLIKELY(mdb_rc != MDB_NOTFOUND)) rc = LSUP_ERROR;

    mdb_txn_abort(txn);

    return rc;
}


LSUP_rc
LSUP_store_match_callback(
        LSUP_MDBStore *store, LSUP_SerTerm sspoc[], size_t *ct,
        mdb_store_match_fn_t callback_fn, void *ctx)
{
    LSUP_TripleKey spok = {
        LSUP_sterm_to_key(sspoc),
        LSUP_sterm_to_key(sspoc + 1),
        LSUP_sterm_to_key(sspoc + 2),
    };

    LSUP_Key ck = store->default_ctx ? LSUP_sterm_to_key(sspoc + 3) : NULL_KEY;

    struct MatchArgs args_s;
    struct MatchArgs *args = &args_s;
    args->ct = ct;
    args->callback_fn = callback_fn;
    args->ctx = ctx;

    // s p o (all terms bound)
    if (spok[0] != NULL_KEY && spok[1] != NULL_KEY && spok[2] != NULL_KEY) {
        return match_callback_3bound(store, args);
    }

    else if (spok[0] != NULL_KEY) {
        args->luks[0] = spok[0];
        args->idx0 = 0;

        if (spok[1] != NULL_KEY) { // s p ?
            args->luks[1] = spok[1];
            args->idx1 = 1;
            return match_callback_2bound(store, args);

        } else if (spok[2] != NULL_KEY) { // s ? o
            args->luks[1] = spok[2];
            args->idx1 = 2;
            return match_callback_2bound(store, args);

        } else { // s ? ?
            return match_callback_1bound(store, args);
        }

    } else if (spok[1] != NULL_KEY) {
        args->luks[0] = spok[1];

        if (spok[2] != NULL_KEY) { // ? p o
            args->luks[1] = spok[2];
            args->idx1 = 2;
            return match_callback_2bound(store, args);

        } else { // ? p ?
            args->idx0 = 1;
            return match_callback_1bound(store, args);
        }

    } else if (spok[2] != NULL_KEY) { // ? ? o
        args->luks[0] = spok[2];
        args->idx0 = 2;
        return match_callback_1bound(store, args);

    } else { // ? ? ? (all terms unbound)
        return match_callback_0bound(store, args);
    }
}


LSUP_rc
LSUP_store_remove(
        LSUP_MDBStore *store, const LSUP_Buffer *sc,
        LSUP_TripleKey data[], size_t data_size)
{
    LSUP_rc rc = LSUP_NOACTION;

    LSUP_Key ck = NULL_KEY;

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

    MDB_txn *txn;

    mdb_txn_begin(store->env, NULL, 0, &txn);

    MDB_cursor *dcur, *icur;
    mdb_cursor_open(txn, store->dbi[IDX_SPO_C], &dcur);
    mdb_cursor_open(txn, store->dbi[IDX_C_SPO], &icur);

    MDB_val spok_v, ck_v;
    LSUP_TripleKey spok_cur;

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

    for(size_t i = 0; i < data_size; i++) {
        spok_v.mv_data = data + i;

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

        // Delete spo:c entry.
        mdb_cursor_del(dcur, 0);

        // Restore ck address after each delete.
        ck_v.mv_data = &ck;

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

        mdb_cursor_del(icur, 0);
        spok_v.mv_data = data + i;

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

        index_triple(store, OP_REMOVE, data[i], ck);
    }

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

    return rc;

_remove_abort:
    mdb_txn_abort(txn);

    return rc;
}


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

    if (store->default_ctx != NULL) LSUP_buffer_done(store->default_ctx);

    free(store);
}


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


/* TODO
inline static int
check_txn_open(MDB_txn *txn, bool write)
{
    if (txn == NULL) {
        mdb_txn_begin(LSUP_mdbenv, NULL, write ? 0 : MDB_RDONLY, &txn);

        return LSUP_OK;
    }

    return LSUP_NOACTION;
}
*/

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

    // Index c:spo.
    if (op == OP_REMOVE) {
        if (ck != NULL_KEY) {
            MDB_cursor *cur;

            v1.mv_data = &ck;
            v1.mv_size = KLEN;
            v2.mv_data = spok;
            v2.mv_size = TRP_KLEN;

            mdb_cursor_open(store->txn, store->dbi[IDX_C_SPO], &cur);
            rc = mdb_cursor_get(cur, &v1, &v2, MDB_GET_BOTH);
            if(rc == MDB_SUCCESS) mdb_cursor_del(cur, 0);

            mdb_cursor_close(cur);
        }

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

            mdb_put(
                    store->txn, store->dbi[IDX_C_SPO],
                    &v1, &v2, MDB_NODUPDATA);
        }

    } else return LSUP_VALUE_ERR;

    LSUP_DoubleKey dbl_keys[3] = {
        {spok[1], spok[2]},  // po
        {spok[0], spok[2]},  // so
        {spok[0], spok[1]},  // sp
    };

    // Add terms to index.
    v1.mv_size = KLEN;
    v2.mv_size = DBL_KLEN;

    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.
            CHECK(
                    mdb_put(store->txn, db1, &v1, &v2, MDB_NODUPDATA),
                    db_rc, _index_triple_exit);
            CHECK(
                    mdb_put(store->txn, db2, &v2, &v1, MDB_NODUPDATA),
                    db_rc, _index_triple_exit);
        }
    }

_index_triple_exit:
    return rc;
}


/* * * Match callbacks. * * */

inline static LSUP_rc
match_callback_0bound(struct MDBStore *store, struct MatchArgs *args)
{
    int rc = LSUP_NORESULT;

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

    MDB_val key_v;

    if(args->ct) {
        MDB_stat stat;
        mdb_stat(store->txn, store->dbi[IDX_SPO_C], &stat);

        *args->ct = stat.ms_entries;
    }

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

    if(args->callback_fn) {
        rc = mdb_cursor_get(cur, &key_v, NULL, MDB_FIRST);
        while (rc != MDB_NOTFOUND) {
            LSUP_TripleKey spok;

            rc = args->callback_fn(spok, args->ctx);
            if (rc < 0) goto _match0b_abort;
        }
    }

_match0b_abort:
    mdb_cursor_close(cur);
    if (txn != store->txn) mdb_txn_abort(txn);

    return rc;
}


inline static LSUP_rc
match_callback_1bound(struct MDBStore *store, struct MatchArgs *args)
{
    int rc = LSUP_NORESULT;

    const uint8_t *term_order = lookup_ordering_1bound[args->idx0];

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

    MDB_cursor *cur;
    mdb_cursor_open(txn, store->dbi[args->idx0], &cur);

    MDB_val key_v, data_v;
    key_v.mv_data = &args->luks;
    key_v.mv_size = KLEN;

    if(args->ct) {
        mdb_cursor_get(cur, &key_v, &data_v, MDB_SET);
        mdb_cursor_count(cur, args->ct);
    }

    if(args->callback_fn) {
        rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_SET);
        if (rc == MDB_SUCCESS)
            rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_GET_MULTIPLE);

        while (rc != MDB_NOTFOUND) {
            LSUP_Key **lu_dset = data_v.mv_data;
            for (int i = 0; i < data_v.mv_size / DBL_KLEN; i++) {
                // Build triple key from lookup key and result keys.
                LSUP_TripleKey spok;

                spok[term_order[0]] = args->luks[0];
                spok[term_order[1]] = lu_dset[i][0];
                spok[term_order[2]] = lu_dset[i][1];

                rc = args->callback_fn(spok, args->ctx);
                if (rc < 0) goto _match1b_abort;
            }
            rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_NEXT_MULTIPLE);
        }
    }

_match1b_abort:
    mdb_cursor_close(cur);
    if (txn != store->txn) mdb_txn_abort(txn);

    return rc;
}


inline static LSUP_rc
match_callback_2bound(struct MDBStore *store, struct MatchArgs *args)
{
    int rc = LSUP_NORESULT;

    uint8_t luk1_offset, luk2_offset;
    const uint8_t *term_order;
    MDB_dbi dbi = 0;

    // Establish lookup ordering with some awkward offset math.
    for(int i = 0; i < 3; i++) {
        if (
            (
                args->idx0 == lookup_ordering_2bound[i][0] &&
                args->idx1 == lookup_ordering_2bound[i][1]
            ) || (
                args->idx0 == lookup_ordering_2bound[i][1] &&
                args->idx1 == lookup_ordering_2bound[i][0]
            )
        ) {
            term_order = lookup_ordering_2bound[i];
            if (term_order[0] == args->idx0) {
                luk1_offset = 0;
                luk2_offset = 1;
            } else {
                luk1_offset = 1;
                luk2_offset = 0;
            }
            dbi = store->dbi[lookup_indices[i + 3]];

            break;
        }
    }

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

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

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

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

    MDB_val key_v, data_v;
    key_v.mv_data = luk;
    key_v.mv_size = DBL_KLEN;

    if(args->ct) {
        mdb_cursor_get(cur, &key_v, &data_v, MDB_SET);
        mdb_cursor_count(cur, args->ct);
    }

    if(args->callback_fn) {
        rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_SET);
        if (rc == MDB_SUCCESS)
            rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_GET_MULTIPLE);

        while (rc != MDB_NOTFOUND) {
            LSUP_Key *lu_dset = data_v.mv_data;
            for (int i = 0; i < data_v.mv_size / DBL_KLEN; i++) {
                // Build triple key from lookup key and result keys.
                LSUP_TripleKey spok;

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

                rc = args->callback_fn(spok, args->ctx);
                if (rc < 0) goto _match2b_abort;
            }
            rc = mdb_cursor_get(cur, &key_v, &data_v, MDB_NEXT_MULTIPLE);
        }
    }

_match2b_abort:
    mdb_cursor_close(cur);
    if (txn != store->txn) mdb_txn_abort(txn);

    return rc;
}


inline static LSUP_rc
match_callback_3bound(struct MDBStore *store, struct MatchArgs *args)
{
    int rc = LSUP_NORESULT;

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

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

    MDB_val key_v, data_v;
    key_v.mv_data = args->luks;
    key_v.mv_size = TRP_KLEN;

    if(args->ct) {
        if (mdb_cursor_get(cur, &key_v, NULL, MDB_SET) == MDB_SUCCESS)
            rc = mdb_cursor_count(cur, args->ct);
    }

    if(args->callback_fn) {
        rc = mdb_cursor_get(cur, &key_v, NULL, MDB_FIRST);
        while (rc != MDB_NOTFOUND) {
            LSUP_TripleKey spok;

            rc = args->callback_fn(spok, args->ctx);
            if (rc < 0) goto _match3b_abort;
        }
    }

_match3b_abort:
    mdb_cursor_close(cur);
    if (txn != store->txn) mdb_txn_abort(txn);

    return rc;
}