AtomStorage.cc

Go to the documentation of this file.

/*
 * FUNCTION:
 * Persistent Atom storage, SQL-backed.
 *
 * Atoms are saved to, and restored from, an SQL DB.
 * Atoms are identified by means of unique ID's, which are taken to
 * be the atom Handles, as maintained by the TLB. In particular, the
 * system here depends on the handles in the TLB and in the SQL DB
 * to be consistent (i.e. kept in sync).
 *
 * Copyright (c) 2008,2009,2013 Linas Vepstas <linas@linas.org>
 *
 * LICENSE:
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License v3 as
 * published by the Free Software Foundation and including the exceptions
 * at http://opencog.org/wiki/Licenses
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program; if not, write to:
 * Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */
#ifdef HAVE_SQL_STORAGE

#include <stdlib.h>
#include <unistd.h>

#include <chrono>
#include <memory>
#include <thread>

#include <opencog/util/oc_assert.h>
#include <opencog/atomspace/Atom.h>
#include <opencog/atomspace/ClassServer.h>
#include <opencog/atomspace/CountTruthValue.h>
#include <opencog/atomspace/IndefiniteTruthValue.h>
#include <opencog/atomspace/Link.h>
#include <opencog/atomspace/Node.h>
#include <opencog/atomspace/ProbabilisticTruthValue.h>
#include <opencog/atomspace/SimpleTruthValue.h>
#include <opencog/atomspace/TLB.h>
#include <opencog/atomspace/TruthValue.h>

#include "AtomStorage.h"
#include "odbcxx.h"

using namespace opencog;

#define USE_INLINE_EDGES

/* ================================================================ */

class AtomStorage::Response
{
    public:
        ODBCRecordSet *rs;

        // Temporary cache of info about atom being assembled.
        Handle handle;
        int itype;
        const char * name;
        int tv_type;
        double mean;
        double confidence;
        double count;
        const char *outlist;
        int height;

        Response()
        {
            tname = "";
            itype = 0;
            intval = 0;
        }

        bool create_atom_column_cb(const char *colname, const char * colvalue)
        {
            // printf ("%s = %s\n", colname, colvalue);
            if (!strcmp(colname, "type"))
            {
                itype = atoi(colvalue);
            }
            else if (!strcmp(colname, "name"))
            {
                name = colvalue;
            }
            else if (!strcmp(colname, "outgoing"))
            {
                outlist = colvalue;
            }
            if (!strcmp(colname, "tv_type"))
            {
                tv_type = atoi(colvalue);
            }
            else if (!strcmp(colname, "stv_mean"))
            {
                mean = atof(colvalue);
            }
            else if (!strcmp(colname, "stv_confidence"))
            {
                confidence = atof(colvalue);
            }
            else if (!strcmp(colname, "stv_count"))
            {
                count = atof(colvalue);
            }
            else if (!strcmp(colname, "uuid"))
            {
                UUID uuid = strtoul(colvalue, NULL, 10);
                handle = Handle(uuid);
            }
            return false;
        }
        bool create_atom_cb(void)
        {
            // printf ("---- New atom found ----\n");
            rs->foreach_column(&Response::create_atom_column_cb, this);

            return false;
        }

        AtomTable *table;
        AtomStorage *store;
        bool load_all_atoms_cb(void)
        {
            // printf ("---- New atom found ----\n");
            rs->foreach_column(&Response::create_atom_column_cb, this);

            AtomPtr atom(store->makeAtom(*this, handle));
            table->add(atom, true);
            return false;
        }

        // Load an atom into the atom table, but only if it's not in
        // it already.  The goal is to avoid clobbering the truth value
        // that is currently in the AtomTable.  Adding an atom to the
        // atom table that already exists causes the two TV's to be
        // merged, which is probably not what was wanted...
        bool load_if_not_exists_cb(void)
        {
            // printf ("---- New atom found ----\n");
            rs->foreach_column(&Response::create_atom_column_cb, this);

            if (not table->holds(handle))
            {
                AtomPtr atom(store->makeAtom(*this, handle));
                load_recursive_if_not_exists(atom);
            }
            return false;
        }

        // Helper function for the above.  The problem is that, when
        // adding links of unknown provenance, it could happen that
        // the outgoing set of the link has not yet been loaded.  In
        // that case, we have to load the outgoing set first.
        void load_recursive_if_not_exists(AtomPtr atom)
        {
            LinkPtr link(LinkCast(atom));
            if (link)
            {
                const HandleSeq& oset = link->getOutgoingSet();
                for (Handle h : oset)
                {
                    if (table->holds(h)) continue;
                    AtomPtr a(store->getAtom(h));
                    load_recursive_if_not_exists(a);
                }
            }
            table->add(atom, true);
        }

        std::vector<Handle> *hvec;
        bool fetch_incoming_set_cb(void)
        {
            // printf ("---- New atom found ----\n");
            rs->foreach_column(&Response::create_atom_column_cb, this);

            // Note, unlike the above 'load' routines, this merely fetches
            // the atoms, and returns a vector of them.  They are loaded
            // into the atomspace later, by the caller.
            Handle h(store->makeAtom(*this, handle));
            hvec->push_back(h);
            return false;
        }

        bool row_exists;
        bool row_exists_cb(void)
        {
            row_exists = true;
            return false;
        }

#ifndef USE_INLINE_EDGES
        // Temporary cache of info about the outgoing set.
        std::vector<Handle> *outvec;
        Handle dst;
        int pos;

        bool create_edge_cb(void)
        {
            // printf ("---- New edge found ----\n");
            rs->foreach_column(&Response::create_edge_column_cb, this);
            int sz = outvec->size();
            if (sz <= pos) outvec->resize(pos+1);
            outvec->at(pos) = dst;
            return false;
        }
        bool create_edge_column_cb(const char *colname, const char * colvalue)
        {
            // printf ("%s = %s\n", colname, colvalue);
            if (!strcmp(colname, "dst_uuid"))
            {
                dst = Handle(strtoul(colvalue, (char **) NULL, 10));
            }
            else if (!strcmp(colname, "pos"))
            {
                pos = atoi(colvalue);
            }
            return false;
        }
#endif /* USE_INLINE_EDGES */

        // deal twith the type-to-id map
        bool type_cb(void)
        {
            rs->foreach_column(&Response::type_column_cb, this);
            store->set_typemap(itype, tname);
            return false;
        }
        const char * tname;
        bool type_column_cb(const char *colname, const char * colvalue)
        {
            if (!strcmp(colname, "type"))
            {
                itype = atoi(colvalue);
            }
            else if (!strcmp(colname, "typename"))
            {
                tname = colvalue;
            }
            return false;
        }
#ifdef OUT_OF_LINE_TVS
        // Callbacks for SimpleTruthValues
        int tvid;
        bool create_tv_cb(void)
        {
            // printf ("---- New SimpleTV found ----\n");
            rs->foreach_column(&Response::create_tv_column_cb, this);
            return false;
        }
        bool create_tv_column_cb(const char *colname, const char * colvalue)
        {
            printf ("%s = %s\n", colname, colvalue);
            if (!strcmp(colname, "mean"))
            {
                mean = atof(colvalue);
            }
            else if (!strcmp(colname, "count"))
            {
                count = atof(colvalue);
            }
            return false;
        }

#endif /* OUT_OF_LINE_TVS */

        // get generic positive integer values
        unsigned long intval;
        bool intval_cb(void)
        {
            rs->foreach_column(&Response::intval_column_cb, this);
            return false;
        }
        bool intval_column_cb(const char *colname, const char * colvalue)
        {
            // we're not going to bother to check the column name ...
            intval = strtoul(colvalue, NULL, 10);
            return false;
        }

        // Get all handles in the database.
        std::set<UUID> *id_set;
        bool note_id_cb(void)
        {
            rs->foreach_column(&Response::note_id_column_cb, this);
            return false;
        }
        bool note_id_column_cb(const char *colname, const char * colvalue)
        {
            // we're not going to bother to check the column name ...
            UUID id = strtoul(colvalue, NULL, 10);
            id_set->insert(id);
            return false;
        }

};

/* ================================================================ */

ODBCConnection* AtomStorage::get_conn()
{
    return conn_pool.pop();
}

void AtomStorage::put_conn(ODBCConnection* db_conn)
{
    conn_pool.push(db_conn);
}

/* ================================================================ */

bool AtomStorage::idExists(const char * buff)
{
    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.row_exists = false;
    rp.rs = db_conn->exec(buff);
    rp.rs->foreach_row(&Response::row_exists_cb, &rp);
    rp.rs->release();
    put_conn(db_conn);
    return rp.row_exists;
}

/* ================================================================ */
#define BUFSZ 250

#ifndef USE_INLINE_EDGES

class AtomStorage::Outgoing
{
    private:
        ODBCConnection *db_conn;
        unsigned int pos;
        Handle src_handle;
    public:
        Outgoing (ODBCConnection *c, Handle h)
        {
            db_conn = c;
            src_handle = h;
            pos = 0;
        }
        bool each_handle (Handle h)
        {
            char buff[BUFSZ];
            UUID src_uuid = src_handle.value();
            UUID dst_uuid = h.value();
            snprintf(buff, BUFSZ, "INSERT  INTO Edges "
                    "(src_uuid, dst_uuid, pos) VALUES (%lu, %lu, %u);",
                    src_uuid, dst_uuid, pos);

            Response rp;
            rp.rs = db_conn->exec(buff);
            rp.rs->release();
            pos ++;
            return false;
        }
};

void AtomStorage::storeOutgoing(AtomPtr atom, Handle h)
{
    Outgoing out(db_conn, h);

    foreach_outgoing_handle(h, &Outgoing::each_handle, &out);
}

#endif /* USE_INLINE_EDGES */

/* ================================================================ */
// Constructors

void AtomStorage::init(const char * dbname,
                       const char * username,
                       const char * authentication)
{
    // Create six, by default ... maybe make more? 
    // There should probably be a few more here, than the number of
    // startWriterThread() calls below.
#define DEFAULT_NUM_CONNS 6
    for (int i=0; i<DEFAULT_NUM_CONNS; i++)
    {
        ODBCConnection* db_conn = new ODBCConnection(dbname, username, authentication);
        conn_pool.push(db_conn);
    }
    type_map_was_loaded = false;
    max_height = 0;

    for (int i=0; i< TYPEMAP_SZ; i++)
    {
        db_typename[i] = NULL;
    }

    local_id_cache_is_inited = false;
    if (!connected()) return;

    reserve();
}

AtomStorage::AtomStorage(const char * dbname,
                         const char * username,
                         const char * authentication)
    : _write_queue(this, &AtomStorage::vdo_store_atom)
{
    init(dbname, username, authentication);
}

AtomStorage::AtomStorage(const std::string& dbname,
                         const std::string& username,
                         const std::string& authentication)
    : _write_queue(this, &AtomStorage::vdo_store_atom)
{
    init(dbname.c_str(), username.c_str(), authentication.c_str());
}

AtomStorage::~AtomStorage()
{
    if (connected())
        setMaxHeight(getMaxObservedHeight());

    while (not conn_pool.is_empty())
    {
        ODBCConnection* db_conn = conn_pool.pop();
        delete db_conn;
    }

    for (int i=0; i<TYPEMAP_SZ; i++)
    {
        if (db_typename[i]) free(db_typename[i]);
    }
}

bool AtomStorage::connected(void)
{
    ODBCConnection* db_conn = get_conn();
    bool have_connection = db_conn->connected();
    put_conn(db_conn);
    return have_connection;
}

/* ================================================================ */

#define STMT(colname,val) { \
    if(update) { \
        if (notfirst) { cols += ", "; } else notfirst = 1; \
        cols += colname; \
        cols += " = "; \
        cols += val; \
    } else { \
        if (notfirst) { cols += ", "; vals += ", "; } else notfirst = 1; \
        cols += colname; \
        vals += val; \
    } \
}

#define STMTI(colname,ival) { \
    char buff[BUFSZ]; \
    snprintf(buff, BUFSZ, "%u", ival); \
    STMT(colname, buff); \
}

#define STMTF(colname,fval) { \
    char buff[BUFSZ]; \
    snprintf(buff, BUFSZ, "%12.8g", fval); \
    STMT(colname, buff); \
}

/* ================================================================ */

#ifdef OUT_OF_LINE_TVS

bool AtomStorage::tvExists(int tvid)
{
    char buff[BUFSZ];
    snprintf(buff, BUFSZ, "SELECT tvid FROM SimpleTVs WHERE tvid = %u;", tvid);
    return idExists(buff);
}

int AtomStorage::storeTruthValue(AtomPtr atom, Handle h)
{
    int notfirst = 0;
    std::string cols;
    std::string vals;
    std::string coda;

    const TruthValue &tv = atom->getTruthValue();

    const SimpleTruthValue *stv = dynamic_cast<const SimpleTruthValue *>(&tv);
    if (NULL == stv)
    {
        fprintf(stderr, "Error: non-simple truth values are not handled\n");
        return 0;
    }

    int tvid = TVID(tv);

    // If its a stock truth value, there is nothing to do.
    if (tvid <= 4) return tvid;

    // Use the TLB Handle as the UUID.
    char tvidbuff[BUFSZ];
    snprintf(tvidbuff, BUFSZ, "%u", tvid);

    bool update = tvExists(tvid);
    if (update)
    {
        cols = "UPDATE SimpleTVs SET ";
        vals = "";
        coda = " WHERE tvid = ";
        coda += tvidbuff;
        coda += ";";
    }
    else
    {
        cols = "INSERT INTO SimpleTVs (";
        vals = ") VALUES (";
        coda = ");";
        STMT("tvid", tvidbuff);
    }

    STMTF("mean", tv.getMean());
    STMTF("count", tv.getCount());

    std::string qry = cols + vals + coda;
    Response rp;
    rp.rs = db_conn->exec(qry.c_str());
    rp.rs->release();

    return tvid;
}

int AtomStorage::TVID(const TruthValue &tv)
{
    if (tv == TruthValue::NULL_TV()) return 0;
    if (tv == TruthValue::TRIVIAL_TV()) return 1;
    if (tv == TruthValue::FALSE_TV()) return 2;
    if (tv == TruthValue::TRUE_TV()) return 3;
    if (tv == TruthValue::DEFAULT_TV()) return 4;

    Response rp;
    rp.rs = db_conn->exec("SELECT NEXTVAL('tvid_seq');");
    rp.rs->foreach_row(&Response::tvid_seq_cb, &rp);
    rp.rs->release();
    return rp.tvid;
}

TruthValue* AtomStorage::getTV(int tvid)
{
    if (0 == tvid) return (TruthValue *) & TruthValue::NULL_TV();
    if (1 == tvid) return (TruthValue *) & TruthValue::DEFAULT_TV();
    if (2 == tvid) return (TruthValue *) & TruthValue::FALSE_TV();
    if (3 == tvid) return (TruthValue *) & TruthValue::TRUE_TV();
    if (4 == tvid) return (TruthValue *) & TruthValue::TRIVIAL_TV();

    char buff[BUFSZ];
    snprintf(buff, BUFSZ, "SELECT * FROM SimpleTVs WHERE tvid = %u;", tvid);

    Response rp;
    rp.rs = db_conn->exec(buff);
    rp.rs->foreach_row(&Response::create_tv_cb, &rp);
    rp.rs->release();

    SimpleTruthValue *stv = new SimpleTruthValue(rp.mean,rp.count);
    return stv;
}

#endif /* OUT_OF_LINE_TVS */

/* ================================================================== */

int AtomStorage::get_height(AtomPtr atom)
{
    LinkPtr l(LinkCast(atom));
    if (NULL == l) return 0;

    int maxd = 0;
    int arity = l->getArity();

    const HandleSeq& out = l->getOutgoingSet();
    for (int i=0; i<arity; i++)
    {
        Handle h = out[i];
        int d = get_height(h);
        if (maxd < d) maxd = d;
    }
    return maxd +1;
}

/* ================================================================ */

std::string AtomStorage::oset_to_string(const std::vector<Handle>& out,
                                        int arity)
{ 
    std::string str;
    str += "\'{";
    for (int i=0; i<arity; i++)
    {
        Handle h = out[i];
        if (i != 0) str += ", ";
        char buff[BUFSZ];
        UUID uuid = h.value();
        snprintf(buff, BUFSZ, "%lu", uuid);
        str += buff;
    }
    str += "}\'";
    return str;
}

/* ================================================================ */

void AtomStorage::flushStoreQueue()
{
    _write_queue.flush_queue();
}

/* ================================================================ */
void AtomStorage::storeAtom(AtomPtr atom, bool synchronous)
{
    get_ids();

    // If a synchronous store, avoid the queues entirely.
    if (synchronous)
    {
        do_store_atom(atom);
        return;
    }
    _write_queue.enqueue(atom);
}

int AtomStorage::do_store_atom(AtomPtr atom)
{
    LinkPtr l(LinkCast(atom));
    if (NULL == l)
    {
        do_store_single_atom(atom, 0);
        return 0;
    }

    int lheight = 0;
    int arity = l->getArity();
    const HandleSeq& out = l->getOutgoingSet();
    for (int i=0; i<arity; i++)
    {
        // Recurse.
        int heig = do_store_atom(out[i]);
        if (lheight < heig) lheight = heig;
    }

    // Height of this link is, by definition, one more than tallest
    // atom in outgoing set.
    lheight ++;
    do_store_single_atom(atom, lheight);
    return lheight;
}

void AtomStorage::vdo_store_atom(AtomPtr& atom)
{
    do_store_atom(atom);
}

/* ================================================================ */
void AtomStorage::storeSingleAtom(AtomPtr atom)
{
    get_ids();
    int height = get_height(atom);
    do_store_single_atom(atom, height);
}

void AtomStorage::do_store_single_atom(AtomPtr atom, int aheight)
{
    setup_typemap();

    int notfirst = 0;
    std::string cols;
    std::string vals;
    std::string coda;

    // Use the TLB Handle as the UUID.
    char uuidbuff[BUFSZ];
    Handle h(atom->getHandle());
    if (TLB::isInvalidHandle(h))
        throw RuntimeException(TRACE_INFO, "Trying to save atom with an invalid handle!");

    UUID uuid = h.value();
    snprintf(uuidbuff, BUFSZ, "%lu", uuid);

    std::unique_lock<std::mutex> lck = maybe_create_id(uuid);
    bool update = not lck.owns_lock();
    if (update)
    {
        cols = "UPDATE Atoms SET ";
        vals = "";
        coda = " WHERE uuid = ";
        coda += uuidbuff;
        coda += ";";
    }
    else
    {
        cols = "INSERT INTO Atoms (";
        vals = ") VALUES (";
        coda = ");";

        STMT("uuid", uuidbuff);
    }

    // Store the atom type and node name only if storing for the
    // first time ever. Once an atom is in an atom table, it's
    // name can type cannot be changed. Only its truth value can
    // change.
    if (false == update)
    {
        // Store the atomspace UUID
        UUID asuid = 0;
        AtomTable * at = atom->getAtomTable();
        // We allow storage of atoms that don't belong to an atomspace.
        if (at) asuid = at->get_uuid();
        snprintf(uuidbuff, BUFSZ, "%lu", asuid);
        STMT("space", uuidbuff);

        // Store the atom UUID
        Type t = atom->getType();
        int dbtype = storing_typemap[t];
        STMTI("type", dbtype);
    
        // Store the node name, if its a node
        NodePtr n(NodeCast(atom));
        if (n)
        {
#if 0
            std::string qname = n->getName();
            escape_single_quotes(qname);
            qname.insert(0U,1U,'\'');
            qname += "'";
#else
            // Use postgres $-quoting to make unicode strings
            // easier to deal with. 
            std::string qname = " $ocp$";
            qname += n->getName();
            qname += "$ocp$ ";
#endif
            STMT("name", qname);

            // Nodes have a height of zero by definition.
            STMTI("height", 0);
        }
        else
        {
            if (max_height < aheight) max_height = aheight;
            STMTI("height", aheight);

#ifdef USE_INLINE_EDGES
            LinkPtr l(LinkCast(atom));
            if (l)
            {
                int arity = l->getArity();
                if (arity)
                {
                    cols += ", outgoing";
                    vals += ", ";
                    vals += oset_to_string(l->getOutgoingSet(), arity);
                }
            }
#endif /* USE_INLINE_EDGES */
        }
    }

    // Store the truth value
    TruthValuePtr tv(atom->getTruthValue());
    TruthValueType tvt = NULL_TRUTH_VALUE;
    if (tv) tvt = tv->getType();
    STMTI("tv_type", tvt);

    switch (tvt)
    {
        case NULL_TRUTH_VALUE:
            break;
        case SIMPLE_TRUTH_VALUE:
        case COUNT_TRUTH_VALUE:
        case PROBABILISTIC_TRUTH_VALUE:
            STMTF("stv_mean", tv->getMean());
            STMTF("stv_confidence", tv->getConfidence());
            STMTF("stv_count", tv->getCount());
            break;
        case INDEFINITE_TRUTH_VALUE:
        {
            IndefiniteTruthValuePtr itv = std::static_pointer_cast<IndefiniteTruthValue>(tv);
            STMTF("stv_mean", itv->getL());
            STMTF("stv_count", itv->getU());
            STMTF("stv_confidence", itv->getConfidenceLevel());
            break;
        }
        default:
            throw RuntimeException(TRACE_INFO,
                "Error: store_single: Unknown truth value type\n");
    }

    std::string qry = cols + vals + coda;
    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.rs = db_conn->exec(qry.c_str());
    rp.rs->release();
    put_conn(db_conn);

#ifndef USE_INLINE_EDGES
    // Store the outgoing handles only if we are storing for the first
    // time, otherwise do nothing. The semantics is that, once the
    // outgoing set has been determined, it cannot be changed.
    if (false == update)
    {
        storeOutgoing(atom);
    }
#endif /* USE_INLINE_EDGES */

    // Make note of the fact that this atom has been stored.
    add_id_to_cache(uuid);
}

/* ================================================================ */
void AtomStorage::setup_typemap(void)
{
    /* Only need to set up the typemap once. */
    if (type_map_was_loaded) return;
    type_map_was_loaded = true;

    // If we are here, we need to reconcile the types currently in
    // use, with a possibly pre-existing typemap. New types must be
    // stored.  So we start by loading a map from SQL (if its there).
    //
    // Be careful to initialize the typemap with invalid types,
    // in case there are unexpected holes in the map!
    for (int i=0; i< TYPEMAP_SZ; i++)
    {
        loading_typemap[i] = NOTYPE;
        storing_typemap[i] = -1;
        db_typename[i] = NULL;
    }

    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.rs = db_conn->exec("SELECT * FROM TypeCodes;");
    rp.store = this;
    rp.rs->foreach_row(&Response::type_cb, &rp);
    rp.rs->release();

    unsigned int numberOfTypes = classserver().getNumberOfClasses();
    for (Type t=0; t<numberOfTypes; t++)
    {
        int sqid = storing_typemap[t];
        /* If this typename is not yet known, record it */
        if (-1 == sqid)
        {
            const char * tname = classserver().getTypeName(t).c_str();

            // Let the sql id be the same as the current type number,
            // unless this sql number is already in use, in which case
            // we need to find another, unused one.  Its in use if we
            // have a string name associated to it.
            sqid = t;

            if ((db_typename[sqid] != NULL) &&
                (loading_typemap[sqid] != t))
            {
                // Find some (any) unused type index to use in the
                // sql table. Use the lowest unused value that we
                // can find.
                for (sqid = 0; sqid<TYPEMAP_SZ; sqid++)
                {
                    if (NULL == db_typename[sqid]) break;
                }

                if (TYPEMAP_SZ <= sqid)
                {
                    put_conn(db_conn);
                    fprintf(stderr, "Fatal Error: type table overflow!\n");
                    abort();
                }
            }

            char buff[BUFSZ];
            snprintf(buff, BUFSZ,
                     "INSERT INTO TypeCodes (type, typename) "
                     "VALUES (%d, \'%s\');",
                     sqid, tname);
            rp.rs = db_conn->exec(buff);
            rp.rs->release();
            set_typemap(sqid, tname);
        }
    }
    put_conn(db_conn);
}

void AtomStorage::set_typemap(int dbval, const char * tname)
{
    Type realtype = classserver().getType(tname);
    loading_typemap[dbval] = realtype;
    storing_typemap[realtype] = dbval;
    if (db_typename[dbval] != NULL) free (db_typename[dbval]);
    db_typename[dbval] = strdup(tname);
}

/* ================================================================ */
bool AtomStorage::atomExists(Handle h)
{
#ifdef ASK_SQL_SERVER
    char buff[BUFSZ];
    UUID uuid = h.value();
    snprintf(buff, BUFSZ, "SELECT uuid FROM Atoms WHERE uuid = %lu;", uuid);
    return idExists(buff);
#else
    std::unique_lock<std::mutex> lock(id_cache_mutex);
    // look at the local cache of id's to see if the atom is in storage or not.
    return local_id_cache.count(h.value());
#endif
}

void AtomStorage::add_id_to_cache(UUID uuid)
{
    std::unique_lock<std::mutex> lock(id_cache_mutex);
    local_id_cache.insert(uuid);

    // If we were previously making this ID, then we are done.
    // The other half of this is in maybe_create_id() below.
    if (0 < id_create_cache.count(uuid))
    {
        id_create_cache.erase(uuid);
    }
}

std::unique_lock<std::mutex> AtomStorage::maybe_create_id(UUID uuid)
{
    std::unique_lock<std::mutex> create_lock(id_create_mutex);
    std::unique_lock<std::mutex> cache_lock(id_cache_mutex);
    // Look at the local cache of id's to see if the atom is in storage or not.
    if (0 < local_id_cache.count(uuid))
        return std::unique_lock<std::mutex>();

    // Is some other thread in the process of adding this ID?
    if (0 < id_create_cache.count(uuid))
    {
        cache_lock.unlock();
        while (true)
        {
            // If we are here, some other thread is making this UUID, 
            // and so we need to wait till they're done. Wait by stalling
            // on the creation lock.
            std::unique_lock<std::mutex> local_create_lock(id_create_mutex);
            // If we are here, then someone finished creating some UUID.
            // Was it our ID? If so, we are done; if not, wait some more.
            cache_lock.lock();
            if (0 == id_create_cache.count(uuid))
            {
                OC_ASSERT(0 < local_id_cache.count(uuid),
                    "Atom for UUID was not created!");
                return std::unique_lock<std::mutex>();
            }
            cache_lock.unlock();
        } 
    }

    // If we are here, then no one has attempted to make this UUID before.
    // Grab the maker lock, and make the damned thing already.
    id_create_cache.insert(uuid);
    return create_lock;
}

void AtomStorage::get_ids(void)
{
    std::unique_lock<std::mutex> lock(id_cache_mutex);

    if (local_id_cache_is_inited) return;
    local_id_cache_is_inited = true;

    local_id_cache.clear();
    ODBCConnection* db_conn = get_conn();

    // It appears that, when the select statment returns more than
    // about a 100K to a million atoms or so, some sort of heap
    // corruption occurs in the odbc code, causing future mallocs
    // to fail. So limit the number of records processed in one go.
    // It also appears that asking for lots of records increases
    // the memory fragmentation (and/or there's a memory leak in odbc??)
#define USTEP 12003
    unsigned long rec;
    unsigned long max_nrec = getMaxObservedUUID();
    for (rec = 0; rec <= max_nrec; rec += USTEP)
    {
        char buff[BUFSZ];
        snprintf(buff, BUFSZ, "SELECT uuid FROM Atoms WHERE "
                "uuid > %lu AND uuid <= %lu;",
                 rec, rec+USTEP);

        Response rp;
        rp.id_set = &local_id_cache;
        rp.rs = db_conn->exec(buff);
        rp.rs->foreach_row(&Response::note_id_cb, &rp);
        rp.rs->release();
    }
    put_conn(db_conn);
}

/* ================================================================ */

#ifndef USE_INLINE_EDGES
void AtomStorage::getOutgoing(std::vector<Handle> &outv, Handle h)
{
    char buff[BUFSZ];
    UUID uuid = h.value();
    snprintf(buff, BUFSZ, "SELECT * FROM Edges WHERE src_uuid = %lu;", uuid);

    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.rs = db_conn->exec(buff);
    rp.outvec = &outv;
    rp.rs->foreach_row(&Response::create_edge_cb, &rp);
    rp.rs->release();
    put_conn(db_conn);
}
#endif /* USE_INLINE_EDGES */

/* ================================================================ */

/* One-size-fits-all atom fetcher */
AtomPtr  AtomStorage::getAtom(const char * query, int height)
{
    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.handle = Handle::UNDEFINED;
    rp.rs = db_conn->exec(query);
    rp.rs->foreach_row(&Response::create_atom_cb, &rp);

    // Did we actually find anything?
    // DO NOT USE TLB::IsInvalidHandle() HERE! It won't work, duhh!
    if (rp.handle.value() == Handle::UNDEFINED.value())
    {
        rp.rs->release();
        put_conn(db_conn);
        return NULL;
    }

    rp.height = height;
    AtomPtr atom(makeAtom(rp, rp.handle));
    rp.rs->release();
    put_conn(db_conn);
    return atom;
}

AtomPtr  AtomStorage::getAtom(Handle h)
{
    setup_typemap();
    char buff[BUFSZ];
    UUID uuid = h.value();
    snprintf(buff, BUFSZ, "SELECT * FROM Atoms WHERE uuid = %lu;", uuid);

    return getAtom(buff, -1);
}

std::vector<Handle> AtomStorage::getIncomingSet(Handle h)
{
    std::vector<Handle> iset;

    setup_typemap();
    char buff[BUFSZ];
    UUID uuid = h.value();
    snprintf(buff, BUFSZ,
        "SELECT * FROM Atoms WHERE outgoing @> ARRAY[CAST(%lu AS BIGINT)];", uuid);

    // Note: "select * from atoms where outgoing@>array[556];" will return
    // all links with atom 556 in the outgoing set -- i.e. the incoming set of 556.
    // Could also use && here instead of @> Don't know if one is faster or not.
    // The cast to BIGINT is needed, as otherwise on gets
    // ERROR:  operator does not exist: bigint[] @> integer[]

    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.store = this;
    rp.height = -1;
    rp.hvec = &iset;
    rp.rs = db_conn->exec(buff);
    rp.rs->foreach_row(&Response::fetch_incoming_set_cb, &rp);
    rp.rs->release();
    put_conn(db_conn);

    return iset;
}

NodePtr AtomStorage::getNode(Type t, const char * str)
{
    setup_typemap();
    char buff[40*BUFSZ];

    // Use postgres $-quoting to make unicode strings easier to deal with. 
    int nc = snprintf(buff, 4*BUFSZ, "SELECT * FROM Atoms WHERE "
        "type = %hu AND name = $ocp$%s$ocp$ ;", storing_typemap[t], str);

    if (40*BUFSZ-1 <= nc)
    {
        fprintf(stderr, "Error: AtomStorage::getNode: buffer overflow!\n");
        buff[40*BUFSZ-1] = 0x0;
        fprintf(stderr, "\tnc=%d buffer=>>%s<<\n", nc, buff);
        return NULL;
    }

    return NodeCast(getAtom(buff, 0));
}

LinkPtr AtomStorage::getLink(Type t, const std::vector<Handle>&oset)
{
    setup_typemap();

    char buff[BUFSZ];
    snprintf(buff, BUFSZ, 
        "SELECT * FROM Atoms WHERE type = %hu AND outgoing = ",
        storing_typemap[t]);

    std::string ostr = buff;
    ostr += oset_to_string(oset, oset.size());
    ostr += ";";

    AtomPtr atom = getAtom(ostr.c_str(), 1);
    return LinkCast(atom);
}

AtomPtr AtomStorage::makeAtom(Response &rp, Handle h)
{
    // Now that we know everything about an atom, actually construct one.
    AtomPtr atom(h);
    Type realtype = loading_typemap[rp.itype];

    if (NOTYPE == realtype)
    {
        throw RuntimeException(TRACE_INFO,
            "Fatal Error: OpenCog does not have a type called %s\n",
            db_typename[rp.itype]);
        return NULL;
    }

    if (NULL == atom)
    {
        // All height zero atoms are nodes,
        // All positive height atoms are links.
        // A negative height is "unknown" and must be checked.
        if ((0 == rp.height) || 
            ((-1 == rp.height) &&
              classserver().isA(realtype, NODE)))
        {
            atom = createNode(realtype, rp.name);
        }
        else
        {
            std::vector<Handle> outvec;
#ifndef USE_INLINE_EDGES
            getOutgoing(outvec, h);
#else
            char *p = (char *) rp.outlist;
            while (p)
            {
                // Break if there is no more atom in the outgoing set
                // or the outgoing set is empty in the first place
                if (*p == '}' or *p == '\0') break;
                Handle hout = (Handle) strtoul(p+1, &p, 10);
                outvec.push_back(hout);
            }
#endif /* USE_INLINE_EDGES */
            atom = createLink(realtype, outvec);
        }

        // Create via the factory for specific types of atom
        // Otherwise at the later stage of the sql-load process when the
        // same atom is being added to the AtomTable (which will also call
        // the same factory function), the dynamic-casting of any atom of
        // one of these types will fail, resulting a new atom being created.
        // But the new atom is having a different UUID, if there exist another
        // link connecting to this atom, the system will fail to find the
        // correct handle of this atom (because it is using the original
        // UUID, as retrieve from the SQL database). Since each of the atoms
        // in the outgoing set of a link needs to be valid, we will get
        // an "Atom in outgoing set isn't known!" error as a result
        atom = AtomTable::factory(realtype, atom);
    }
    else
    {
        // Perform at least some basic sanity checking ...
        if (realtype != atom->getType())
        {
            UUID uuid = h.value();
            throw RuntimeException(TRACE_INFO,
                "Fatal Error: mismatched atom type for existing atom! "
                "uuid=%lu real=%d atom=%d\n",
                uuid, realtype, atom->getType());
        }
        // If we are here, and the atom uuid is set, then it should match.
        if (Handle::UNDEFINED.value() != atom->_uuid and 
            atom->_uuid != h.value())
        {
            throw RuntimeException(TRACE_INFO,
                "Fatal Error: mismatched handle and atom UUID's, atom=%lu handle=%lu",
                atom->_uuid, h.value());
        }
    }

    // Give the atom the correct UUID. The AtomTable will need this.
    atom->_uuid = h.value();

    // Now get the truth value
    switch (rp.tv_type)
    {
        case NULL_TRUTH_VALUE:
            break;

        case SIMPLE_TRUTH_VALUE:
        {
            TruthValuePtr stv(SimpleTruthValue::createTV(rp.mean, rp.count));
            atom->setTruthValue(stv);
            break;
        }
        case COUNT_TRUTH_VALUE:
        {
            TruthValuePtr ctv(CountTruthValue::createTV(rp.mean, rp.confidence, rp.count));
            atom->setTruthValue(ctv);
            break;
        }
        case INDEFINITE_TRUTH_VALUE:
        {
            TruthValuePtr itv(IndefiniteTruthValue::createTV(rp.mean, rp.count, rp.confidence));
            atom->setTruthValue(itv);
            break;
        }
        case PROBABILISTIC_TRUTH_VALUE:
        {
            TruthValuePtr ptv(ProbabilisticTruthValue::createTV(rp.mean, rp.confidence, rp.count));
            atom->setTruthValue(ptv);
            break;
        }
        default:
            throw RuntimeException(TRACE_INFO,
                "Error: makeAtom: Unknown truth value type\n");
    }

    load_count ++;
    if (load_count%10000 == 0)
    {
        fprintf(stderr, "\tLoaded %lu atoms.\n", (unsigned long) load_count);
    }

    add_id_to_cache(h.value());
    return atom;
}

/* ================================================================ */

void AtomStorage::load(AtomTable &table)
{
    unsigned long max_nrec = getMaxObservedUUID();
    TLB::reserve_upto(max_nrec);
    fprintf(stderr, "Max observed UUID is %lu\n", max_nrec);
    load_count = 0;
    max_height = getMaxObservedHeight();
    fprintf(stderr, "Max Height is %d\n", max_height);

    setup_typemap();

    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.table = &table;
    rp.store = this;

    for (int hei=0; hei<=max_height; hei++)
    {
        unsigned long cur = load_count;

#if GET_ONE_BIG_BLOB
        char buff[BUFSZ];
        snprintf(buff, BUFSZ, "SELECT * FROM Atoms WHERE height = %d;", hei);
        rp.height = hei;
        rp.rs = db_conn->exec(buff);
        rp.rs->foreach_row(&Response::load_all_atoms_cb, &rp);
        rp.rs->release();
#else
        // It appears that, when the select statment returns more than
        // about a 100K to a million atoms or so, some sort of heap
        // corruption occurs in the iodbc code, causing future mallocs
        // to fail. So limit the number of records processed in one go.
        // It also appears that asking for lots of records increases
        // the memory fragmentation (and/or there's a memory leak in iodbc??)
        // XXX Not clear is UnixODBC suffers from this same problem.
#define STEP 12003
        unsigned long rec;
        for (rec = 0; rec <= max_nrec; rec += STEP)
        {
            char buff[BUFSZ];
            snprintf(buff, BUFSZ, "SELECT * FROM Atoms WHERE "
                    "height = %d AND uuid > %lu AND uuid <= %lu;",
                     hei, rec, rec+STEP);
            rp.height = hei;
            rp.rs = db_conn->exec(buff);
            rp.rs->foreach_row(&Response::load_all_atoms_cb, &rp);
            rp.rs->release();
        }
#endif
        fprintf(stderr, "Loaded %lu atoms at height %d\n", load_count - cur, hei);
    }
    put_conn(db_conn);
    fprintf(stderr, "Finished loading %lu atoms in total\n",
        (unsigned long) load_count);

    // synchrnonize!
    table.barrier();
}

void AtomStorage::loadType(AtomTable &table, Type atom_type)
{
    unsigned long max_nrec = getMaxObservedUUID();
    TLB::reserve_upto(max_nrec);
    logger().debug("AtomStorage::loadType: Max observed UUID is %lu\n", max_nrec);
    load_count = 0;

    // For links, assume a worst-case height.
    // For nodes, its easy ... max_height is zero.
    if (classserver().isNode(atom_type))
        max_height = 0;
    else
        max_height = getMaxObservedHeight();
    logger().debug("AtomStorage::loadType: Max Height is %d\n", max_height);

    setup_typemap();
    int db_atom_type = storing_typemap[atom_type];

    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.table = &table;
    rp.store = this;

    for (int hei=0; hei<=max_height; hei++)
    {
        unsigned long cur = load_count;

#if GET_ONE_BIG_BLOB
        char buff[BUFSZ];
        snprintf(buff, BUFSZ,
            "SELECT * FROM Atoms WHERE height = %d AND type = %d;",
             hei, db_atom_type);
        rp.height = hei;
        rp.rs = db_conn->exec(buff);
        rp.rs->foreach_row(&Response::load_if_not_exists_cb, &rp);
        rp.rs->release();
#else
        // It appears that, when the select statment returns more than
        // about a 100K to a million atoms or so, some sort of heap
        // corruption occurs in the iodbc code, causing future mallocs
        // to fail. So limit the number of records processed in one go.
        // It also appears that asking for lots of records increases
        // the memory fragmentation (and/or there's a memory leak in iodbc??)
        // XXX Not clear is UnixODBC suffers from this same problem.
#define STEP 12003
        unsigned long rec;
        for (rec = 0; rec <= max_nrec; rec += STEP)
        {
            char buff[BUFSZ];
            snprintf(buff, BUFSZ, "SELECT * FROM Atoms WHERE type = %d "
                    "AND height = %d AND uuid > %lu AND uuid <= %lu;",
                     db_atom_type, hei, rec, rec+STEP);
            rp.height = hei;
            rp.rs = db_conn->exec(buff);
            rp.rs->foreach_row(&Response::load_if_not_exists_cb, &rp);
            rp.rs->release();
        }
#endif
        logger().debug("AtomStorage::loadType: Loaded %lu atoms of type %d at height %d\n",
            load_count - cur, db_atom_type, hei);
    }
    put_conn(db_conn);
    logger().debug("AtomStorage::loadType: Finished loading %lu atoms in total\n",
        (unsigned long) load_count);

    // Synchronize!
    table.barrier();
}

bool AtomStorage::store_cb(AtomPtr atom)
{
    storeSingleAtom(atom);
    store_count ++;
    if (store_count%1000 == 0)
    {
        fprintf(stderr, "\tStored %lu atoms.\n", (unsigned long) store_count);
    }
    return false;
}

void AtomStorage::store(const AtomTable &table)
{
    max_height = 0;
    store_count = 0;

#ifdef ALTER
    rename_tables();
    create_tables();
#endif

    get_ids();
    UUID max_uuid = TLB::getMaxUUID();
    fprintf(stderr, "Max UUID is %lu\n", max_uuid);

    setup_typemap();

    ODBCConnection* db_conn = get_conn();
    Response rp;

#ifndef USE_INLINE_EDGES
    // Drop indexes, for faster loading.
    // But this only matters for the non-inline eges...
    rp.rs = db_conn->exec("DROP INDEX uuid_idx;");
    rp.rs->release();
    rp.rs = db_conn->exec("DROP INDEX src_idx;");
    rp.rs->release();
#endif

    table.foreachHandleByType(
        [&](Handle h)->void { store_cb(h); }, ATOM, true);

#ifndef USE_INLINE_EDGES
    // Create indexes
    rp.rs = db_conn->exec("CREATE INDEX uuid_idx ON Atoms (uuid);");
    rp.rs->release();
    rp.rs = db_conn->exec("CREATE INDEX src_idx ON Edges (src_uuid);");
    rp.rs->release();
#endif /* USE_INLINE_EDGES */

    rp.rs = db_conn->exec("VACUUM ANALYZE;");
    rp.rs->release();
    put_conn(db_conn);

    setMaxHeight(getMaxObservedHeight());
    fprintf(stderr, "\tFinished storing %lu atoms total.\n",
        (unsigned long) store_count);
}

/* ================================================================ */

void AtomStorage::rename_tables(void)
{
    ODBCConnection* db_conn = get_conn();
    Response rp;

    rp.rs = db_conn->exec("ALTER TABLE Atoms RENAME TO Atoms_Backup;");
    rp.rs->release();
#ifndef USE_INLINE_EDGES
    rp.rs = db_conn->exec("ALTER TABLE Edges RENAME TO Edges_Backup;");
    rp.rs->release();
#endif /* USE_INLINE_EDGES */
    rp.rs = db_conn->exec("ALTER TABLE Global RENAME TO Global_Backup;");
    rp.rs->release();
    rp.rs = db_conn->exec("ALTER TABLE TypeCodes RENAME TO TypeCodes_Backup;");
    rp.rs->release();
    put_conn(db_conn);
}

void AtomStorage::create_tables(void)
{
    ODBCConnection* db_conn = get_conn();
    Response rp;

    // See the file "atom.sql" for detailed documentation as to the 
    // structure of the SQL tables.
    rp.rs = db_conn->exec("CREATE TABLE Atoms ("
                          "uuid     BIGINT PRIMARY KEY,"
                          "space    BIGINT,"
                          "type     SMALLINT,"
                          "type_tv  SMALLINT,"
                          "stv_mean FLOAT,"
                          "stv_confidence FLOAT,"
                          "stv_count FLOAT,"
                          "height   SMALLINT,"
                          "name     TEXT,"
                          "outgoing BIGINT[]);");
    rp.rs->release();

#ifndef USE_INLINE_EDGES
    rp.rs = db_conn->exec("CREATE TABLE Edges ("
                          "src_uuid  INT,"
                          "dst_uuid  INT,"
                          "pos INT);");
    rp.rs->release();
#endif /* USE_INLINE_EDGES */

    rp.rs = db_conn->exec("CREATE TABLE TypeCodes ("
                          "type SMALLINT UNIQUE,"
                          "typename TEXT UNIQUE);");
    rp.rs->release();
    type_map_was_loaded = false;

    rp.rs = db_conn->exec("CREATE TABLE Spaces ("
                          "space     BIGINT,"
                          "parent    BIGINT);");
    rp.rs->release();

    rp.rs = db_conn->exec("CREATE TABLE Global ("
                          "max_height INT);");
    rp.rs->release();
    rp.rs = db_conn->exec("INSERT INTO Global (max_height) VALUES (0);");
    rp.rs->release();

    put_conn(db_conn);
}

void AtomStorage::kill_data(void)
{
    ODBCConnection* db_conn = get_conn();
    Response rp;

    // See the file "atom.sql" for detailed documentation as to the 
    // structure of the SQL tables.
    rp.rs = db_conn->exec("DELETE from Atoms;");
    rp.rs->release();

    rp.rs = db_conn->exec("UPDATE Global SET max_height = 0;");
    rp.rs->release();
    put_conn(db_conn);
}

/* ================================================================ */

void AtomStorage::setMaxHeight(int sqmax)
{
    // Max height of db contents can only get larger! 
    if (max_height < sqmax) max_height = sqmax;

    char buff[BUFSZ];
    snprintf(buff, BUFSZ, "UPDATE Global SET max_height = %d;", max_height);

    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.rs = db_conn->exec(buff);
    rp.rs->release();
    put_conn(db_conn);
}

int AtomStorage::getMaxHeight(void)
{
    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.rs = db_conn->exec("SELECT max_height FROM Global;");
    rp.rs->foreach_row(&Response::intval_cb, &rp);
    rp.rs->release();
    put_conn(db_conn);
    return rp.intval;
}

UUID AtomStorage::getMaxObservedUUID(void)
{
    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.intval = 0;
    rp.rs = db_conn->exec("SELECT uuid FROM Atoms ORDER BY uuid DESC LIMIT 1;");
    rp.rs->foreach_row(&Response::intval_cb, &rp);
    rp.rs->release();
    put_conn(db_conn);
    return rp.intval;
}

int AtomStorage::getMaxObservedHeight(void)
{
    ODBCConnection* db_conn = get_conn();
    Response rp;
    rp.intval = 0;
    rp.rs = db_conn->exec("SELECT height FROM Atoms ORDER BY height DESC LIMIT 1;");
    rp.rs->foreach_row(&Response::intval_cb, &rp);
    rp.rs->release();
    put_conn(db_conn);
    return rp.intval;
}

void AtomStorage::reserve(void)
{
    UUID max_observed_id = getMaxObservedUUID();
    fprintf(stderr, "Reserving UUID up to %lu\n", max_observed_id);
    TLB::reserve_upto(max_observed_id);
}

#endif /* HAVE_SQL_STORAGE */
/* ============================= END OF FILE ================= */