graph.c

Go to the documentation of this file.

#include "volksdata/graph.h"
 
/*
 * Data types.
 */
 
struct graph_t {
    VOLK_Term               *uri;           
    VOLK_Store *            store;          
};
 
struct graph_iter_t {
    const VOLK_Graph *      graph;          
    void *                  data;           
    size_t                  ct;             
    VOLK_BufferTriple *     sspo;           
};
 
 
/*
 * Static prototypes.
 */
 
inline static VOLK_rc
graph_iter_next_buffer (VOLK_GraphIterator *it);
 
 
/*TODO
#define ENTRY(a, b) (be) == (VOLK_STORE_##a) ||
static inline bool
check_backend (VOLK_StoreType be)
{ return (BACKEND_TBL false); }
#undef ENTRY
*/
 
/*
 * Graph API.
 */
 
VOLK_Graph *
VOLK_graph_new (VOLK_Store *store, const char *uri_str)
{
    // Create a HTable graph by default.
    VOLK_Graph *gr;
    MALLOC_GUARD (gr, NULL);
 
    gr->uri =
        uri_str? VOLK_iriref_new (uri_str) :
        VOLK_iriref_new (NULL);
    NLNL (gr->uri);
 
    // An embedded store has the same ID as the graph.
    gr->store = (store) ? store : VOLK_store_new (
            VOLK_STORE_HTABLE, gr->uri->data, 0, false);
    NLNL (gr->store);
 
    log_debug ("Graph created.");
    return gr;
}
 
 
VOLK_Graph *
VOLK_graph_get_txn (
        void *txn, VOLK_Store *store, const VOLK_Term *uri, size_t *ct)
{
    VOLK_Buffer *sc = VOLK_term_serialize (uri);
    void *it = VOLK_store_lookup_txn (
            store, txn, NULL, NULL, NULL, sc, NULL);
 
    void *_txn;
    if (txn) _txn = txn;
    else if (VOLK_store_features (store) & VOLK_STORE_TXN) {
        RCNL (VOLK_store_begin (store, 0, &_txn));
    } else _txn = NULL;
 
    VOLK_Graph *gr = VOLK_graph_new (NULL, uri->data);
    VOLK_BufferTriple *sspo = BTRP_DUMMY;
    void *add_it = VOLK_graph_add_init_txn (_txn, gr);
 
    size_t _ct = 0;
    while (VOLK_store_iter_next (store, it, sspo, NULL) == VOLK_OK) {
        // This is deserializing a buffer triple that will be re-serialized by
        // VOLK_graph_add_iter. But it's necessary to relativize URIs.
        VOLK_Triple *spo = VOLK_triple_new_from_btriple (sspo);
        VOLK_graph_add_iter (add_it, spo);
        VOLK_triple_free (spo);
        _ct++;
    }
    VOLK_graph_iter_free (add_it);
    VOLK_store_iter_free (store, it);
    VOLK_buffer_free (sc);
    VOLK_btriple_free (sspo);
 
    // Do not create a new graph if no results were found.
    if (_ct == 0) {
        VOLK_graph_free (gr);
        gr = NULL;
    }
    if (ct) *ct = _ct;
    if (_txn != txn) VOLK_store_abort (store, _txn);
 
    return gr;
}
 
 
VOLK_rc
VOLK_graph_bool_op_txn (
        void *txn, const VOLK_bool_op op,
        const VOLK_Graph *gr1, const VOLK_Graph *gr2, VOLK_Graph *res)
{
    VOLK_rc rc = VOLK_NOACTION;
    if (UNLIKELY (
            op != VOLK_BOOL_UNION
            && op != VOLK_BOOL_SUBTRACTION
            && op != VOLK_BOOL_INTERSECTION
            && op != VOLK_BOOL_XOR)) {
        log_error ("Invalid boolean operation: %d.", op);
 
        return VOLK_VALUE_ERR;
    }
 
    /*
     * Graphs could possibly be in different storage back ends, and may or
     * may not have independently opened transactions, so they have to be
     * handled individually.
     */
    void
        *lu1_txn = NULL,
        *lu2_txn = NULL,
        *res_txn = NULL;
 
    /* BEGIN union block. */
 
    if (op == VOLK_BOOL_UNION) {
        // No need to use a transaction here: the graph is freed on failure.
        rc = VOLK_graph_copy_contents (gr1, res, NULL, NULL, NULL);
        PCHECK (rc, fail);
        rc = VOLK_graph_copy_contents (gr2, res, NULL, NULL, NULL);
        PCHECK (rc, fail);
 
        return VOLK_OK;
    }
 
    /* END union block. */
 
    VOLK_Buffer
        *res_sc = VOLK_store_features (res->store) & VOLK_STORE_CTX ?
                VOLK_term_serialize (res->uri) : NULL,
        *gr1_sc = VOLK_store_features (gr1->store) & VOLK_STORE_CTX ?
                VOLK_term_serialize (gr1->uri) : NULL,
        *gr2_sc = VOLK_store_features (gr2->store) & VOLK_STORE_CTX ?
                VOLK_term_serialize (gr2->uri) : NULL;
    void *lu1_it, *lu2_it, *add_it;
    VOLK_BufferTriple *sspo = BTRP_DUMMY;
    size_t ct;
 
    // Whether gr1 or gr2 txn will be open independently from res txn.
    bool
        open_txn1 = false,
        open_txn2 = false;
 
    // TODO check for transaction consistency here, and if it needs a separate
    // transaction handle.
    add_it = VOLK_store_add_init_txn (res->store, txn, res_sc);
 
    if (VOLK_store_features (res->store) & VOLK_STORE_TXN)
        res_txn = VOLK_store_iter_txn (res->store, add_it);
 
    /* If either source graph is in the same store as the destination and has
     * an open transaction, reuse that transaction. A new reader cannot be
     * opened in LMDB while a writer is open already.
     */
    // Handle gr1 transaction.
    if (VOLK_store_features (gr1->store) & VOLK_STORE_TXN) {
        if (gr1->store == res->store) lu1_txn = res_txn;
        else {
            CHECK (VOLK_store_begin (
                    gr1->store, VOLK_STORE_TXN_RO, &lu1_txn), fail);
            open_txn1 = true;
        }
    }
 
    // Handle gr2 transaction.
    if (VOLK_store_features (gr2->store) & VOLK_STORE_TXN) {
        if (gr2->store == res->store) lu2_txn = res_txn;
        else if (gr2->store == gr1->store) lu2_txn = lu1_txn;
        // FIXME: see above.
        else {
            CHECK (VOLK_store_begin (
                    gr2->store, VOLK_STORE_TXN_RO, &lu2_txn), fail);
            open_txn2 = true;
        }
    }
    log_trace (
            "lu1_txn: %p ; lu2_txn: %p ; res_txn: %p",
            lu1_txn, lu2_txn, res_txn);
 
    /* BEGIN XOR block. */
 
    if (op == VOLK_BOOL_XOR) {
        // Add triples from gr2 if not found in gr1.
        lu2_it = VOLK_store_lookup_txn (
                gr2->store, lu2_txn, NULL, NULL, NULL, gr2_sc, NULL);
        while (VOLK_store_iter_next (
                    gr2->store, lu2_it, sspo, NULL) == VOLK_OK) {
            lu1_it = VOLK_store_lookup_txn (
                gr1->store, lu1_txn, sspo->s, sspo->p, sspo->o, gr1_sc, &ct
            );
            if (ct == 0) {
                VOLK_store_add_iter (res->store, add_it, sspo);
                rc = VOLK_OK;
            }
            VOLK_store_iter_free (gr1->store, lu1_it);
        }
        VOLK_store_iter_free (gr2->store, lu2_it);
    }
 
    /* BEGIN subtraction and intersection block. */
 
    lu1_it = VOLK_store_lookup_txn (
            gr1->store, lu1_txn, NULL, NULL, NULL, gr1_sc, NULL);
    while (VOLK_store_iter_next  (gr1->store, lu1_it, sspo, NULL) == VOLK_OK) {
        lu2_it = VOLK_store_lookup_txn (
                gr2->store, lu2_txn, sspo->s, sspo->p, sspo->o, gr2_sc, &ct);
        if (UNLIKELY (!lu2_it)) {
            rc = VOLK_DB_ERR;
            VOLK_store_iter_free (gr1->store, lu1_it);
            goto fail;
        }
        // For XOR and subtraction, add if not found.
        // For intersection, add if found.
        if ((ct == 0) ^ (op == VOLK_BOOL_INTERSECTION)) {
            VOLK_store_add_iter (res->store, add_it, sspo);
            rc = VOLK_OK;
        }
        VOLK_store_iter_free (gr2->store, lu2_it);
    }
    VOLK_store_iter_free (gr1->store, lu1_it);
 
    if (open_txn1) VOLK_store_commit (gr1->store, lu1_txn);
    if (open_txn2) VOLK_store_commit (gr2->store, lu2_txn);
 
    VOLK_store_add_done (res->store, add_it);
    VOLK_btriple_free (sspo);
    VOLK_buffer_free (res_sc);
    VOLK_buffer_free (gr1_sc);
    VOLK_buffer_free (gr2_sc);
 
    /* END subtraction, intersection, XOR block. */
 
    return rc;
 
fail:
    if (lu1_txn) VOLK_store_abort (gr1->store, lu1_txn);
    if (lu2_txn) VOLK_store_abort (gr2->store, lu2_txn);
    VOLK_graph_free (res);
    return rc;
}
 
 
void
VOLK_graph_free (VOLK_Graph *gr)
{
    if (UNLIKELY (!gr)) return;
 
    VOLK_term_free (gr->uri);
    // If the store is embedded, it goes away with the associated graph.
    if (VOLK_store_features (gr->store) & VOLK_STORE_EMBED) {
        VOLK_store_free (gr->store);
    }
 
    free (gr);
}
 
 
const VOLK_Term *
VOLK_graph_uri (const VOLK_Graph *gr) { return gr->uri; }
 
 
VOLK_Store *
VOLK_graph_store (const VOLK_Graph *gr)
{ return gr->store; }
 
 
VOLK_rc
VOLK_graph_set_uri_txn (void *txn, VOLK_Graph *gr, const char *uri_str)
{
    VOLK_rc rc = VOLK_OK;
    VOLK_Buffer *old_sc = NULL, *new_sc = NULL;
 
    VOLK_Term *uri = VOLK_iriref_new (uri_str);
    if (UNLIKELY (!uri)) {
        rc = VOLK_MEM_ERR;
        goto finally;
    }
 
    // Update context for triples in the graph.
    if (VOLK_store_features (gr->store) & VOLK_STORE_CTX) {
        old_sc = VOLK_term_serialize (gr->uri);
        new_sc = VOLK_term_serialize (uri);
        if (UNLIKELY (!old_sc || !new_sc)) {
            rc = VOLK_MEM_ERR;
            goto finally;
        }
 
        PCHECK (
            rc = VOLK_store_update_ctx_txn (gr->store, txn, old_sc, new_sc),
            finally
        );
 
        // Overall success even if rc of underlying fn was VOLK_NOACTION.
        if (rc == VOLK_NOACTION) rc = VOLK_OK;
    }
 
    VOLK_term_free (gr->uri);
    gr->uri = uri;
 
finally:
    if (old_sc) VOLK_buffer_free (old_sc);
    if (new_sc) VOLK_buffer_free (new_sc);
 
    return rc;
}
 
 
size_t
VOLK_graph_size_txn (void *txn, const VOLK_Graph *gr)
{
    void *_txn;
    if (txn) _txn = txn;
    else if (VOLK_store_features (gr->store) & VOLK_STORE_TXN)
        RCCK (VOLK_store_begin (gr->store, 0, &_txn));
    else _txn = NULL;
 
    size_t ct = 0;
    VOLK_Buffer *sc =
            VOLK_store_features (gr->store) & VOLK_STORE_TXN ?
            VOLK_term_serialize (gr->uri) : NULL;
    void *it = VOLK_store_lookup_txn (
        gr->store, _txn, NULL, NULL, NULL, sc, &ct);
    if (!it) {
        log_error ("Error initializing lookup.");
        ct = -1;
    }
    VOLK_store_iter_free (gr->store, it);
 
    VOLK_buffer_free (sc);
    if (_txn != txn) VOLK_store_abort (gr->store, _txn);
 
    return ct;
}
 
 
bool
VOLK_graph_equals_txn (void *txn, const VOLK_Graph *gr1, const VOLK_Graph *gr2)
{
    void *_txn1, *_txn2;
    if (txn) {
        if (gr1->store == gr2->store) _txn2 = _txn1 = txn;
        else {
            log_error (
                "Graphs must be from the same store when a transaction handle "
                "is passed."
            );
            return false;
        }
    }
    else {
        if (VOLK_store_features (gr1->store) & VOLK_STORE_TXN)
            RCCK (VOLK_store_begin (gr1->store, VOLK_STORE_TXN_RO, &_txn1));
        else _txn1 = NULL;
 
        if (gr2->store == gr1->store) _txn2 = _txn1;
        else if (VOLK_store_features (gr2->store) & VOLK_STORE_TXN)
            RCCK (VOLK_store_begin (gr2->store, VOLK_STORE_TXN_RO, &_txn2));
        else _txn2 = NULL;
    }
 
    VOLK_Triple *spo = NULL;
    bool match = true;
 
    VOLK_GraphIterator *it = VOLK_graph_lookup_txn (
            _txn1, gr1, NULL, NULL, NULL, NULL);
    NLNL (it);
    while (VOLK_graph_iter_next (it, &spo) != VOLK_END) {
        match = VOLK_graph_contains_txn (_txn2, gr2, spo);
        VOLK_triple_free (spo);
        if (!match) goto finally;
    }
    VOLK_graph_iter_free (it);
 
    it = VOLK_graph_lookup_txn (
            _txn2, gr2, NULL, NULL, NULL, NULL);
    NLNL (it);
    while (VOLK_graph_iter_next (it, &spo) != VOLK_END) {
        match = VOLK_graph_contains_txn (_txn1, gr1, spo);
        VOLK_triple_free (spo);
        if (!match) goto finally;
    }
 
finally:
    VOLK_graph_iter_free (it);
    if (_txn1 != txn) VOLK_store_abort (gr1->store, _txn1);
    if (_txn2 != txn && _txn2 != _txn1) VOLK_store_abort (gr2->store, _txn2);
 
    return match;
}
 
 
VOLK_GraphIterator *
VOLK_graph_add_init_txn (void *txn, VOLK_Graph *gr)
{
    VOLK_GraphIterator *it;
    CALLOC_GUARD (it, NULL);
 
    VOLK_Buffer *sc = VOLK_term_serialize (gr->uri);
 
    it->data = VOLK_store_add_init_txn (gr->store, txn, sc);
    VOLK_buffer_free (sc);
 
    it->graph = gr;
 
    return it;
}
 
 
VOLK_rc
VOLK_graph_add_iter (VOLK_GraphIterator *it, const VOLK_Triple *spo)
{
    log_trace (
            "Adding triple {%s, %s, %s} to %s",
            spo->s->data, spo->p->data, spo->o->data,
            VOLK_graph_uri(it->graph)->data);
 
    // Make relative s and o.
    VOLK_Term *rel_s, *rel_o;
    if (spo->s->type == VOLK_TERM_IRIREF)
        rel_s = VOLK_iriref_new_rel (it->graph->uri, spo->s);
    else rel_s = spo->s;
    if (spo->o->type == VOLK_TERM_IRIREF)
        rel_o = VOLK_iriref_new_rel (it->graph->uri, spo->o);
    else rel_o = spo->o;
 
    VOLK_Triple *rel_spo = VOLK_triple_new (rel_s, spo->p, rel_o);
    log_trace (
            "Adding relative triple: {%s, %s, %s}",
            rel_s->data, spo->p->data, rel_o->data);
 
    // Serialize relative triple.
    VOLK_BufferTriple *sspo = VOLK_triple_serialize (rel_spo);
    NLRCCK (sspo, VOLK_MEM_ERR);
 
    // Selectively free triple members and structure.
    if (rel_s != spo->s) VOLK_term_free (rel_s);
    if (rel_o != spo->o) VOLK_term_free (rel_o);
    free (rel_spo);
 
    VOLK_rc rc = VOLK_store_add_iter (it->graph->store, it->data, sspo);
    PCHECK (rc, finally);
 
    // Store datatype term permanently.
    if (rc == VOLK_OK) {
        for (int i = 0; i < 3; i++) {
            VOLK_Term *term = VOLK_triple_pos (spo, i);
            if (term->type == VOLK_TERM_LITERAL) {
                VOLK_Buffer *ser_dtype = VOLK_term_serialize (term->datatype);
                void *txn = VOLK_store_iter_txn (it->graph->store, it->data);
                VOLK_rc term_rc = VOLK_store_add_term_txn (
                        it->graph->store, txn, ser_dtype);
                PCHECK (term_rc, finally);
                VOLK_buffer_free (ser_dtype);
            }
        }
    }
 
 
finally:
    VOLK_btriple_free (sspo);
 
    return rc;
}
 
 
VOLK_rc
VOLK_graph_add_txn (
        void *txn, VOLK_Graph *gr, VOLK_Triple *const *trp, size_t *ct)
{
    VOLK_rc rc = VOLK_NOACTION;
 
    // Initialize iterator.
    VOLK_GraphIterator *it = VOLK_graph_add_init_txn (txn, gr);
 
    if (ct) *ct = 0;
    // Serialize and insert RDF triples.
    for (size_t i = 0; trp[i] != NULL; i++) {
        log_trace ("Inserting triple #%lu", i);
 
        VOLK_rc db_rc = VOLK_graph_add_iter (it, trp[i]);
 
        if (db_rc == VOLK_OK) {
            rc = VOLK_OK;
            if (ct) (*ct)++;
            // A duplicate will return VOLK_NOACTION and not increment ct.
        }
        if (UNLIKELY (db_rc < 0)) {
            rc = db_rc;
            goto finally;
        }
    }
 
finally:
    VOLK_graph_iter_free (it);
 
    return rc;
}
 
 
VOLK_rc
VOLK_graph_remove_txn (
        void *txn, VOLK_Graph *gr,
        const VOLK_Term *s, const VOLK_Term *p, const VOLK_Term *o,
        size_t *ct)
{
    VOLK_Buffer
        *ss = VOLK_term_serialize (s),
        *sp = VOLK_term_serialize (p),
        *so = VOLK_term_serialize (o),
        *sc = VOLK_term_serialize (gr->uri);
 
    log_debug (
        "Removing triples by terms: %s %s %s",
        s ? s->data : "<nil>",
        p ? p->data : "<nil>",
        o ? o->data : "<nil>"
    );
 
    VOLK_rc rc = VOLK_store_remove_txn (
            gr->store, txn, ss, sp, so, sc, ct);
 
    VOLK_buffer_free (ss);
    VOLK_buffer_free (sp);
    VOLK_buffer_free (so);
    VOLK_buffer_free (sc);
 
    return rc;
}
 
 
VOLK_rc
VOLK_graph_copy_contents_txn (
        void *txn, const VOLK_Graph *src, VOLK_Graph *dest,
        const VOLK_Term *s, const VOLK_Term *p, const VOLK_Term *o)
{
    void *_txn1, *_txn2;
    bool
        src_is_txn = VOLK_store_features (src->store) & VOLK_STORE_TXN,
        dest_is_txn = VOLK_store_features (dest->store) & VOLK_STORE_TXN;
 
    if (txn) {
        if (
                src_is_txn && dest_is_txn &&
                strcmp (src->store->id, dest->store->id)) {
            log_error (
                "Graphs must be from the same store when a transaction handle "
                "is passed."
            );
            return VOLK_VALUE_ERR;
        }
        _txn1 = src_is_txn ? txn : NULL;
        _txn2 = dest_is_txn ? txn : NULL;
    } else {
        if (src_is_txn)
            RCCK (VOLK_store_begin (src->store, VOLK_STORE_TXN_RO, &_txn1));
        else _txn1 = NULL;
        if (dest_is_txn)
            RCCK (VOLK_store_begin (dest->store, 0, &_txn2));
        else _txn2 = NULL;
    }
 
    VOLK_rc rc = VOLK_NOACTION;
 
    VOLK_GraphIterator *it = VOLK_graph_lookup_txn (_txn1, src, s, p, o, NULL);
 
    VOLK_Triple *spo = NULL;
    VOLK_GraphIterator *add_it = VOLK_graph_add_init_txn (_txn2, dest);
    while (VOLK_graph_iter_next (it, &spo) != VOLK_END) {
        VOLK_rc add_rc = VOLK_graph_add_iter (add_it, spo);
        VOLK_triple_free (spo);
        if (LIKELY (add_rc == VOLK_OK)) rc = VOLK_OK;
        else if (add_rc < 0) {
            rc = add_rc;
            break;
        }
    }
 
    VOLK_graph_iter_free (add_it);
    VOLK_graph_iter_free (it);
    if (_txn1 != txn) VOLK_store_abort (src->store, _txn1);
    if (_txn2 != txn) VOLK_store_commit (dest->store, _txn2);
 
    return rc;
}
 
 
VOLK_GraphIterator *
VOLK_graph_lookup_txn (
        void *txn, const VOLK_Graph *gr,
        const VOLK_Term *s, const VOLK_Term *p, const VOLK_Term *o,
        size_t *ct)
{
    VOLK_GraphIterator *it = malloc (sizeof (*it));
    NLNL (it);
 
    // Make relative s and o.
    VOLK_Term *rel_s, *rel_o;
    if (s && s->type == VOLK_TERM_IRIREF) {
        rel_s = VOLK_iriref_new_rel (gr->uri, s);
        log_debug ("Relative S lookup: %s", rel_s->data);
    } else rel_s = (VOLK_Term *)s;
    if (o && o->type == VOLK_TERM_IRIREF) {
        rel_o = VOLK_iriref_new_rel (gr->uri, o);
        log_debug ("Relative O lookup: %s", rel_o->data);
    } else rel_o = (VOLK_Term *)o;
 
    VOLK_Buffer
        *ss = VOLK_term_serialize (rel_s),
        *sp = VOLK_term_serialize (p),
        *so = VOLK_term_serialize (rel_o),
        *sc = VOLK_store_features (gr->store) & VOLK_STORE_CTX ?
                VOLK_term_serialize (gr->uri) : NULL;
 
    // Selectively free triple members and structure.
    if (rel_s != s) VOLK_term_free (rel_s);
    if (rel_o != o) VOLK_term_free (rel_o);
 
    it->data = VOLK_store_lookup_txn (gr->store, txn, ss, sp, so, sc, ct);
 
    VOLK_buffer_free (ss);
    VOLK_buffer_free (sp);
    VOLK_buffer_free (so);
    VOLK_buffer_free (sc);
 
    if (UNLIKELY (!it->data)) {
        free (it);
        return NULL;
    }
 
    it->graph = gr;
 
    if (VOLK_store_features (it->graph->store) & VOLK_STORE_COW) {
        // Copy-on-write store.
        it->sspo = BTRP_DUMMY;
        if (UNLIKELY (it->sspo == NULL)) return NULL;
        it->sspo->s->flags |= VOLK_BUF_BORROWED;
        it->sspo->p->flags |= VOLK_BUF_BORROWED;
        it->sspo->o->flags |= VOLK_BUF_BORROWED;
    } else {
        // TODO copy-on-retrieval store. No implementations yet.
    }
 
    return it;
}
 
 
VOLK_rc
VOLK_graph_iter_next (VOLK_GraphIterator *it, VOLK_Triple **spo_p)
{
    VOLK_rc rc;
    PRCCK (rc = graph_iter_next_buffer (it));
    if (rc != VOLK_OK) return rc;
 
    VOLK_Triple *spo = VOLK_triple_new (
        VOLK_term_new_from_buffer (it->sspo->s),
        VOLK_term_new_from_buffer (it->sspo->p),
        VOLK_term_new_from_buffer (it->sspo->o)
    );
    if (UNLIKELY (!spo)) return VOLK_MEM_ERR;
 
    *spo_p = spo;
 
    return VOLK_OK;
}
 
 
const VOLK_Graph *
VOLK_graph_iter_graph (VOLK_GraphIterator *it)
{ return it->graph; }
 
 
bool
VOLK_graph_iter_is_add (VOLK_GraphIterator *it)
{ return it->sspo == NULL; }
 
 
void
VOLK_graph_iter_free (VOLK_GraphIterator *it)
{
    if (UNLIKELY (!it)) return;
 
    if (it->sspo) {
        // Free lookup iterator.
        VOLK_store_iter_free (it->graph->store, it->data);
 
        /*
         * This deallocates resources properly by preserving borrowed pointers
         * from the store in case of VOLK_STORE_COW stores.
         */
        if (VOLK_store_features (it->graph->store) & VOLK_STORE_COW) {
            VOLK_btriple_free (it->sspo);
            log_debug ("Freeing dummy triple @ %p", it->sspo);
        } else {
            // TODO copy-on-retrieval stores. None yet.
        }
 
    } else {
        // Add iterator has sspo == NULL. Use add_done_fn for that.
        VOLK_store_add_done (it->graph->store, it->data);
    }
 
    free (it);
}
 
 
bool
VOLK_graph_contains_txn (
        void *txn, const VOLK_Graph *gr, const VOLK_Triple *spo)
{
    VOLK_GraphIterator *it = VOLK_graph_lookup_txn (
            txn, gr, spo->s, spo->p, spo->o, NULL);
    VOLK_Triple *tmp_spo = NULL;
    bool rc = VOLK_graph_iter_next (it, &tmp_spo) != VOLK_END;
 
    VOLK_triple_free (tmp_spo);
    VOLK_graph_iter_free (it);
 
    return rc;
}
 
 
void
VOLK_graph_print (const VOLK_Graph *gr)
{
    size_t ct;
    VOLK_GraphIterator *it = VOLK_graph_lookup (gr, NULL, NULL, NULL, &ct);
    if (UNLIKELY (!it)) {
        log_error ("Could not inspect graph for printing.");
        return;
    }
 
    printf ("\n*** Graph %s (%zu triples):\n", gr->uri->data, ct);
 
    VOLK_Triple *spo = NULL;
    ct = 0;
    VOLK_rc rc = VOLK_NORESULT;
    while ((rc = VOLK_graph_iter_next (it, &spo)) == VOLK_OK) {
        printf (
                "#%-6zu {%s  %s  %s}\n",
                ct, spo->s->data, spo->p->data, spo->o->data);
        VOLK_triple_free (spo);
        ct++;
    }
    VOLK_graph_iter_free (it);
    if (rc != VOLK_END)
        log_error (
                "Output truncated due to abnormal return: %s",
                VOLK_strerror (rc));
}
 
 
VOLK_LinkMap *
VOLK_graph_connections_txn (
    void *txn, const VOLK_Graph *gr, const VOLK_Term *t,
    const VOLK_LinkType type
)
{
    const VOLK_Term
        *s = NULL,
        *p = NULL,
        *o = NULL;
 
    // Position of passed term and link terms, respectively.
    VOLK_TriplePos pos1, pos2;
 
    if (type == VOLK_LINK_INBOUND) {
        o = t;
        pos1 = TRP_POS_O;
        pos2 = TRP_POS_P;
    } else if (type == VOLK_LINK_OUTBOUND) {
        s = t;
        pos1 = TRP_POS_S;
        pos2 = TRP_POS_P;
    } else if (type == VOLK_LINK_EDGE) {
        p = t;
        pos1 = TRP_POS_P;
        pos2 = TRP_POS_S;
    } else {
        log_error ("Invalid connection type: %d", type);
        return NULL;
    }
 
    void *_txn;
    if (txn) _txn = txn;
    else if (VOLK_store_features (gr->store) & VOLK_STORE_TXN) {
        RCNL (VOLK_store_begin (gr->store, VOLK_STORE_TXN_RO, &_txn));
    } else _txn = NULL;
 
    // Gather all linking terms in a set first.
    VOLK_GraphIterator *it = VOLK_graph_lookup_txn (_txn, gr, s, p, o, NULL);
    NLNL (it);
 
    VOLK_TermSet *lts = VOLK_term_set_new();
    while (graph_iter_next_buffer (it) != VOLK_END) {
        VOLK_Term *ex = NULL;
        VOLK_Term *ins = VOLK_term_new_from_buffer (
                    VOLK_btriple_pos (it->sspo, pos2));
        VOLK_term_set_add (lts, ins, &ex);
    }
    VOLK_graph_iter_free(it);
 
    VOLK_LinkMap *ret = VOLK_link_map_new (t, type);
    if (!ret) return NULL;
    size_t i = 0;
    VOLK_Term *lt;
    while (VOLK_term_set_next (lts, &i, &lt) != VOLK_END) {
        VOLK_link_map_add (
                ret, VOLK_term_copy (lt),
                VOLK_graph_term_set_txn (_txn, gr, t, pos1, lt, pos2));
    }
    VOLK_term_set_free (lts);
    if (_txn != txn) VOLK_store_abort (gr->store, _txn);
 
    return ret;
}
 
 
VOLK_TermSet *
VOLK_graph_term_set_txn (
        void *txn, const VOLK_Graph *gr,
        const VOLK_Term *t1, const VOLK_TriplePos t1_pos,
        const VOLK_Term *t2, const VOLK_TriplePos t2_pos)
{
    if (t1_pos == t2_pos) {
        log_error ("Term 1 and 2 positions cannot be the same!");
        return NULL;
    }
 
    const VOLK_Term *spo_l[3] = {NULL};
    spo_l[t1_pos] = t1;
    spo_l[t2_pos] = t2;
    VOLK_TriplePos rpos = 0;  // Position of term to be added to results.
    for (unsigned i = 0; i < 3; i++)
        if (t1_pos != i && t2_pos != i) rpos = i;
 
    VOLK_GraphIterator *it = VOLK_graph_lookup_txn (
            txn, gr, spo_l[0], spo_l[1], spo_l[2], NULL);
 
    VOLK_TermSet *ts = VOLK_term_set_new();
    while (graph_iter_next_buffer (it) != VOLK_END) {
        // There cannot be duplicates in a 2-bound lookup.
        VOLK_term_set_add (
                ts,
                VOLK_term_new_from_buffer (VOLK_btriple_pos (it->sspo, rpos)),
                NULL);
    }
    VOLK_graph_iter_free (it);
 
    return ts;
}
 
 
VOLK_TermSet *
VOLK_graph_unique_terms_txn (
        void *txn, const VOLK_Graph *gr, VOLK_TriplePos pos)
{
    // TODO We should use spo indices for stores that have them...
    VOLK_GraphIterator *it = VOLK_graph_lookup_txn (
            txn, gr, NULL, NULL, NULL, NULL);
 
    VOLK_TermSet *ts = VOLK_term_set_new();
    while (graph_iter_next_buffer (it) != VOLK_END) {
        VOLK_Term
            *ex = NULL,
            *ins = VOLK_term_new_from_buffer (VOLK_btriple_pos (it->sspo, pos));
        VOLK_term_set_add (ts, ins, &ex);
    }
    VOLK_graph_iter_free(it);
 
    return ts;
}
 
 
size_t
VOLK_graph_add_link_map (VOLK_GraphIterator *it, VOLK_LinkMap *lm)
{
    VOLK_Triple *spo = TRP_DUMMY;
    size_t ct = 0;
    VOLK_LinkMapIterator *lmit = VOLK_link_map_iter_new (lm);
 
    while (VOLK_link_map_triples (lmit, spo) != VOLK_END) {
        VOLK_rc rc = VOLK_graph_add_iter (it, spo);
        if (rc >= 0) ct++;
        PRCCK (rc);
    }
    VOLK_link_map_iter_free (lmit);
    free (spo);
 
    return ct;
}
 
 
VOLK_Term *
VOLK_bnode_add_collection (VOLK_GraphIterator *it, VOLK_TermSet *ts)
{
    VOLK_Term
        *s = VOLK_term_new (VOLK_TERM_BNODE, NULL, NULL),
        *rdf_first = VOLK_iriref_new ("rdf:first"),
        *rdf_rest = VOLK_iriref_new ("rdf:rest"),
        *rdf_nil = VOLK_iriref_new ("rdf:nil"),
        *link;
 
    VOLK_Triple *spo = TRP_DUMMY;
    link = s;
    size_t i = 0;
    VOLK_Term *t;
    while (VOLK_term_set_next (ts, &i, &t) != VOLK_END) {
        spo->s = link;
        spo->p = rdf_first;
        spo->o = t;
        PRCNL (VOLK_graph_add_iter (it, spo));
 
        spo->p = rdf_rest;
        size_t save_i = i; // Save iterator position to restore it after peek.
        spo->o = (
                // Peek into the next result.
                VOLK_term_set_next (ts, &i, NULL) != VOLK_END ?
                VOLK_term_new (VOLK_TERM_BNODE, NULL, NULL)
                : rdf_nil);
        i = save_i; // Restore the iterator that advanced when peeking.
 
        PRCNL (VOLK_graph_add_iter (it, spo));
 
        if (link != s) VOLK_term_free (link);
        // Current object becomes next subject. Irrelevant for last item.
        link = spo->o;
    }
 
    VOLK_term_free (rdf_first);
    VOLK_term_free (rdf_rest);
    VOLK_term_free (rdf_nil);
    free (spo);
 
    return s;
}
 
 
/*
 * Static functions.
 */

inline static VOLK_rc
graph_iter_next_buffer (VOLK_GraphIterator *it)
{ return VOLK_store_iter_next (it->graph->store, it->data, it->sspo, NULL); }
 

 
VOLK_Graph * VOLK_graph_new_ns (VOLK_Store *store, const char *ns_str);