volksdata/src/store_htable.c

#include "hashmap.h"

#include "volksdata/store_htable.h"


/**
 * Callback type for key comparison.
 */
typedef bool (*VOLK_key_eq_fn_t)(
        const VOLK_Key spok[], const VOLK_Key luk[]);


typedef struct idx_entry_t {
    VOLK_Key            key;        ///< Serialized term key.
    void *              data;       ///< Serialized term data.
    size_t              size;       ///< Serialized term size.
} IndexEntry;

typedef struct ht_store_t {
    struct hashmap *    keys;       ///< Triple keys (set).
    struct hashmap *    idx;        ///< Map of keys to serialized terms.
} HTStore;

typedef struct ht_iterator_t {
    HTStore *           store;      ///< Store being iterated.
    size_t              cur;        ///< Internal hash table cursor.
    VOLK_Key            luk[3];     ///< 0÷3 lookup keys.
    VOLK_key_eq_fn_t    eq_fn;      ///< Equality function to test triples.
    int                 rc;         ///< Return code for *next* result.
                                    ///< When the end of results is reached,
                                    ///< this is set to VOLK_END.
    VOLK_TripleKey *    entry;      ///< Retrieved SPO key.
} HTIterator;


/**
 * Dummy callback for queries with all parameters unbound. Returns true.
*/
static bool lookup_none_eq_fn (
        const VOLK_Key spok[], const VOLK_Key luk[])
{
    (void) spok;
    (void) luk;
    return true;
}

/**
 * Keyset lookup for S key.
 */
static bool lookup_sk_eq_fn (
        const VOLK_Key spok[], const VOLK_Key luk[])
{ return spok[0] == luk[0]; }

/**
 * Keyset lookup for P key.
 */
static bool lookup_pk_eq_fn (
        const VOLK_Key spok[], const VOLK_Key luk[])
{ return spok[1] == luk[1]; }

/**
 * Keyset lookup for O key.
 */
static bool lookup_ok_eq_fn (
        const VOLK_Key spok[], const VOLK_Key luk[])
{ return spok[2] == luk[2]; }

/**
 * Keyset lookup for S and P keys.
 */
static bool lookup_spk_eq_fn (
        const VOLK_Key spok[], const VOLK_Key luk[])
{ return spok[0] == luk[0] && spok[1] == luk[1]; }

/**
 * Keyset lookup for S and O keys.
 */
static bool lookup_sok_eq_fn (
        const VOLK_Key spok[], const VOLK_Key luk[])
{ return spok[0] == luk[0] && spok[2] == luk[2]; }

/**
 * Keyset lookup for P and O keys.
 */
static bool lookup_pok_eq_fn (
        const VOLK_Key spok[], const VOLK_Key luk[])
{ return spok[1] == luk[1] && spok[2] == luk[2]; }

/**
 * Keyset lookup for S, P and O keys.
 */
static bool lookup_spok_eq_fn (
        const VOLK_Key spok[], const VOLK_Key luk[])
{ return spok[0] == luk[0] && spok[1] == luk[1] && spok[2] == luk[2]; }


/* * * CALLBACKS * * */

uint64_t trp_key_hash_fn (
        const void *item, uint64_t seed0, uint64_t seed1)
{
    (void) seed1;

    return (uint64_t) (VOLK_HASH (item, TRP_KLEN, seed0));
}


int trp_key_cmp_fn (const void *a, const void *b, void *udata)
{
    (void) udata;

    return memcmp (a, b, TRP_KLEN);
}


/**
 * Hash callback function for index hashmap.
 */
static uint64_t htstore_idx_hash_fn (
        const void *item, uint64_t seed0, uint64_t seed1)
{
    (void) seed0;
    (void) seed1;

    return ((IndexEntry *) item)->key;
}


/**
 * Compare callback function for index hashmap.
 */
static int htstore_idx_cmp_fn (const void *a, const void *b, void *udata)
{
    (void) udata;

    return ((const IndexEntry *) a)->key - ((const IndexEntry *) b)->key;
}


/**
 * Delete callback function for hashmap.
 */
static void htstore_idx_free_fn (void *item)
{ free (((IndexEntry *) item)->data); }


/* * * Other prototypes. * * */

inline static VOLK_rc
tkey_to_strp (
        const HTStore *store, const VOLK_Key *spok, VOLK_BufferTriple *sspo);


static VOLK_rc
add_key_iter (HTIterator *it, const VOLK_Key *spok);


/** @brief Advance iterator by next key.
 *
 * Result is stored in it->entry.
 */
static VOLK_rc
htiter_next_key (HTIterator *it);


/*
 * Interface members.
 */

/** @brief Create a store for an individual graph.
 *
 * @param[in] id Graph identifier. This may or may not be set. The store does
 *  not use this value internally, and does not check for duplicates.
 *
 * @param[in] size Initial size of the store (in number of triples to
 * preallocate). It may be 0.
 *
 * @return New graph store.
 */
static void *
htstore_new (const char *id, size_t size)
{
    (void) id;

    HTStore *ht;
    CALLOC_GUARD (ht, NULL);

    ht->idx = hashmap_new (
            sizeof (IndexEntry), 0, VOLK_HASH_SEED, 0,
            htstore_idx_hash_fn, htstore_idx_cmp_fn, htstore_idx_free_fn,
            NULL);

    ht->keys = hashmap_new (
            sizeof (VOLK_TripleKey), size, VOLK_HASH_SEED, 0,
            trp_key_hash_fn, trp_key_cmp_fn, NULL,
            NULL);

    return ht;
}


#if 0
static VOLK_rc
htstore_copy_contents (HTStore *dest, const HTStore *src)
{
    size_t i = 0;
    VOLK_TripleKey *spok;
    IndexEntry *entry;
    while (hashmap_iter (src->keys, &i, (void **) &spok)) {
        hashmap_set (dest->keys, spok);
        if (UNLIKELY (hashmap_oom (dest->keys))) return VOLK_MEM_ERR;
    }
    i = 0;
    while (hashmap_iter (src->idx, &i, (void **) &entry)) {
        hashmap_set (dest->idx, entry);
        if (UNLIKELY (hashmap_oom (dest->idx))) return VOLK_MEM_ERR;
    }

    return VOLK_OK;
}
#endif


static void
htstore_free (void *h)
{
    HTStore *store = h;
    hashmap_free (store->idx);
    hashmap_free (store->keys);
    free (store);
}


static size_t
htstore_size (const void *h)
{
    const HTStore *store = h;
    return hashmap_count (store->keys);
}


static VOLK_rc
htstore_add_term (void *h, const VOLK_Buffer *sterm, void *_unused)
{
    (void) _unused;
    HTStore *store = h;
    IndexEntry entry_s = {
        .key = VOLK_buffer_hash (sterm),
    };
    if (hashmap_get (store->idx, &entry_s)) return VOLK_NOACTION;

    entry_s.data = malloc (sterm->size);
    memcpy (entry_s.data, sterm->addr, sterm->size);
    entry_s.size = sterm->size;

    LOG_TRACE("Adding term key: %lx", entry_s.key);
    hashmap_set (store->idx, &entry_s);
    //LOG_TRACE("Term index size: %lu", hashmap_count (store->idx));

    return VOLK_OK;
}


static void *
htstore_add_init (void *h, const VOLK_Buffer *_unused, void *_unused2)
{
    (void) _unused;
    (void) _unused2;
    HTIterator *it;
    MALLOC_GUARD (it, NULL);

    it->store = h;

    return it;
}


static VOLK_rc
htstore_add_iter (void *h, const VOLK_BufferTriple *sspo)
{
    HTIterator *it = h;
    VOLK_TripleKey spok = {
        VOLK_buffer_hash (sspo->s),
        VOLK_buffer_hash (sspo->p),
        VOLK_buffer_hash (sspo->o),
    };

    VOLK_rc rc = add_key_iter (it, spok);

    if (rc != VOLK_OK) return rc;

    for (int i = 0; i < 3; i++)
        htstore_add_term (it->store, VOLK_btriple_pos (sspo, i), NULL);

    return rc;
}


static VOLK_rc
htstore_add_done (void *h)
{
    free (h);
    return VOLK_OK;
}


static void *
htstore_lookup (
        void *h,
        const VOLK_Buffer *ss, const VOLK_Buffer *sp, const VOLK_Buffer *so,
        const VOLK_Buffer *_unused, void *_unused2, size_t *ct)
{
    (void) _unused;
    (void) _unused2;

    HTStore *store = h;
    HTIterator *it;
    CALLOC_GUARD (it, NULL);

    it->store = store;
    it->rc = VOLK_END;

    if (ct) *ct = 0;
    if (hashmap_count (store->keys) == 0) return it;

    it->luk[0] = VOLK_buffer_hash (ss);
    it->luk[1] = VOLK_buffer_hash (sp);
    it->luk[2] = VOLK_buffer_hash (so);

    // s p o
    if (ss && sp && so) {
        it->eq_fn = lookup_spok_eq_fn;

    } else if (ss) {
        // s p ?
        if (sp) {
            it->eq_fn = lookup_spk_eq_fn;

        // s ? o
        } else if (so) {
            it->eq_fn = lookup_sok_eq_fn;

        // s ? ?
        } else {
            it->eq_fn = lookup_sk_eq_fn;
        }

    } else if (sp) {
        // ? p o
        if (so) {
            it->eq_fn = lookup_pok_eq_fn;

        // ? p ?
        } else it->eq_fn = lookup_pk_eq_fn;

    // ? ? o
    } else if (so) {
        it->eq_fn = lookup_ok_eq_fn;

    // ? ? ?
    } else it->eq_fn = lookup_none_eq_fn;

    LOG_TRACE("LUK: {%lx, %lx, %lx}", it->luk[0], it->luk[1], it->luk[2]);

    if (ct) {
        // Loop over results to determine count.
        while (htiter_next_key (it) == VOLK_OK) (*ct)++;

        // Reposition cursor to the hashtable beginning.
        it->cur = 0;
        it->rc = VOLK_OK;
    }

    return it;
}


static VOLK_rc
htstore_remove(
        void *h, const VOLK_Buffer *ss, const VOLK_Buffer *sp,
        const VOLK_Buffer *so,  const VOLK_Buffer *_unused,
        void *_unused2, size_t *ct_p)
{
    (void) _unused;
    (void) _unused2;
    HTStore *store = h;
    size_t ct;

    HTIterator *it = htstore_lookup (store, ss, sp, so, NULL, NULL, &ct);
    if (UNLIKELY (!it)) return VOLK_DB_ERR;

    VOLK_rc rc;
    if (ct == 0) {
        rc = VOLK_NOACTION;
        goto finally;
    }

    while (htiter_next_key (it) == VOLK_OK) {
        LOG_TRACE(
                "Deleting {%lx, %lx, %lx}.",
                it->entry[0][0], it->entry[0][1], it->entry[0][2]);
        hashmap_delete (store->keys, it->entry);
        rc = VOLK_OK;
        it->cur = 0; // Reset cursor, buckets are rearranged after deletion.
    }

finally:
    free (it);
    if (ct_p) *ct_p = ct;

    return rc;
}


static VOLK_rc
htiter_next_key (HTIterator *it)
{
    if (UNLIKELY (!it)) return VOLK_VALUE_ERR;

    // This value is for internal looping only. It shall never be returned.
    it->rc = VOLK_NORESULT;

    do {
        // Loop through all triples until a match is found, or end is reached.
        if (hashmap_iter (it->store->keys, &it->cur, (void **) &it->entry)) {
            //LOG_TRACE("it->cur: %lu", it->cur);
            if (it->eq_fn (*it->entry, it->luk)) {
                LOG_TRACE(
                    "Found spok: {%lx, %lx, %lx}",
                    it->entry[0][0], it->entry[0][1], it->entry[0][2]
                );

                it->rc = VOLK_OK;
            }
        }
        else it->rc = VOLK_END;

    } while (it->rc == VOLK_NORESULT);

    return it->rc;
}


static VOLK_rc
htiter_next (void *h, VOLK_BufferTriple *sspo, VOLK_Buffer **_unused)
{
    (void) _unused;
    HTIterator *it = h;
    VOLK_rc rc = htiter_next_key (it);
    if (rc != VOLK_OK) return rc;

    return tkey_to_strp (it->store, *it->entry, sspo);
}


const VOLK_StoreInt htstore_int = {
    .name           = "Hashtable Store",
    .features       = VOLK_STORE_COW,

    .setup_fn       = NULL,
    .new_fn         = htstore_new,
    .free_fn        = htstore_free,

    .size_fn        = htstore_size,

    .add_init_fn    = htstore_add_init,
    .add_iter_fn    = htstore_add_iter,
    .add_done_fn    = htstore_add_done,
    .add_term_fn    = htstore_add_term,

    .lookup_fn      = htstore_lookup,
    .lu_next_fn     = htiter_next,
    .lu_free_fn     = free,

    .remove_fn      = htstore_remove,
};


/*
 * Other statics.
 */

inline static VOLK_rc
tkey_to_strp (
        const HTStore *store, const VOLK_Key *spok, VOLK_BufferTriple *sspo)
{
    // Data owned by the store.
    const IndexEntry *tmp;

    for (int i = 0; i < 3; i++) {
        tmp = hashmap_get (store->idx, spok + i);
        if (UNLIKELY (!tmp)) return VOLK_DB_ERR;
        VOLK_btriple_pos(sspo, i)->addr = tmp->data;
        VOLK_btriple_pos(sspo, i)->size = tmp->size;
        VOLK_btriple_pos(sspo, i)->flags |= VOLK_BUF_BORROWED;
    }

    return VOLK_OK;
}


static VOLK_rc
add_key_iter (HTIterator *it, const VOLK_Key *spok)
{
    // Add triple.
    LOG_TRACE("Inserting spok: {%lx, %lx, %lx}", spok[0], spok[1], spok[2]);

    if (hashmap_get (it->store->keys, spok)) {
        LOG_TRACE("Triple found. Not adding.");
        return VOLK_NOACTION;
    }

    LOG_TRACE("Triple not found, inserting.");

    hashmap_set (it->store->keys, (void *)spok);
    if (hashmap_oom(it->store->keys)) return VOLK_MEM_ERR;

    return VOLK_OK;
}