/** @file  psDB.c
 *
 * -*- mode: C; c-basic-indent: 4; tab-width: 8; indent-tabs-mode: nil -*-
 * vim: set cindent ts=8 sw=4 expandtab:
 *
 *  @brief database functions
 *
 *  This file contains functions that perform basic database operations.  MySQL
 *  4.1.2 or newer is required.
 *
 *  @author Aaron Culliney
 *  @author Joshua Hoblitt
 *
 *  @version $Revision: 1.48 $ $Name: not supported by cvs2svn $
 *  @date $Date: 2006-04-20 02:22:28 $
 *
 *  Copyright 2005 Joshua Hoblitt, University of Hawaii
 */

#ifndef OMIT_PSDB

#include <stdio.h>
#include <stdarg.h>
#include <string.h>
#undef __STRICT_ANSI__
#include <stdlib.h>
#define __STRICT_ANSI__
#include <math.h>
#include <mysql.h>
#include <mysql_com.h> // enum_field_types

#include "psDB.h"
#include "psMemory.h"
#include "psAbort.h"
#include "psError.h"
#include "psString.h"
#include "psErrorText.h"

typedef struct
{
    enum enum_field_types type;
    bool            isUnsigned;
}
mysqlType;

static psHash   *pTypeToSQLlookupTable = NULL;
static psHash   *sqlToPTypeLookupTable = NULL;
static psHash   *mysqlToSqlLookupTable = NULL;
static psHash   *pTypeToMysqlLookupTable = NULL;

// database utility functions
static inline bool psDBPackRow(MYSQL_BIND *bind, const psMetadata *values, psU32 paramCount);

// SQL generation functions
static char    *psDBGenerateCreateTableSQL(const char *tableName, const psMetadata *where);
static char    *psDBGenerateSelectRowSQL(const char *tableName, const char *col,
        const psMetadata *where, psU64 limit);
static char    *psDBGenerateInsertRowSQL(const char *tableName, const psMetadata *row);
static char    *psDBGenerateUpdateRowSQL(const char *tableName, const psMetadata *where,
        const psMetadata *values);
static char    *psDBGenerateDeleteRowSQL(const char *tableName, const psMetadata *where,
        unsigned long long limit);
static char    *psDBGenerateWhereSQL(const psMetadata *where);
static char    *psDBGenerateSetSQL(const psMetadata *set
                                  );
// lookup table functions
static psElemType psDBMySQLToPType(enum enum_field_types type, unsigned int flags);
static char    *psDBPTypeToSQL(psElemType pType);
static mysqlType *psDBPTypeToMySQL(psElemType pType);

static psHash  *psDBGetPTypeToSQLTable(void);
static void     psDBPTypeToSQLTableCleanup(void);

static psHash  *psDBGetSQLToPTypeTable(void);
static void     psDBSQLToPTypeTableCleanup(void);

static psHash  *psDBGetMySQLToSQLTable(void);
static void     psDBMySQLToSQLTableCleanup(void);

static psHash  *psDBGetPTypeToMySQLTable(void);
static void     psDBPTypeToMySQLTableCleanup(void);

static psPtr    psDBMySQLTypeAlloc(enum enum_field_types type, bool isUnsigned);
static void     psDBAddToLookupTable(psHash *lookupTable, psU32 type, const char *string);
static void     psDBAddVoidToLookupTable(psHash *lookupTable, psU32 type, psPtr value);

// pType utility functions
static psPtr    psDBGetPTypeNaN(psElemType pType);
static bool     psDBIsPTypeNaN(psElemType pType, psPtr data);

// string utility functions
static char    *psDBIntToString(psU64 n);


// public functions
/*****************************************************************************/

psDB *psDBInit(const char *host,
               const char *user,
               const char *passwd,
               const char *dbname)
{
    MYSQL           *mysql;
    psDB            *dbh;

    mysql = mysql_init(NULL);
    if (!mysql) {
        psAbort(__func__, "mysql_init(), out of memory.");
    }

    // Connect to host and mySql server with specified database
    if (!mysql_real_connect(mysql, host, user, passwd, dbname, 0, NULL, 0)) {
        psError(PS_ERR_BAD_PARAMETER_TYPE, true,
                PS_ERRORTEXT_psDB_FAILED_TO_CONNECT,mysql_error(mysql));

        mysql_close(mysql);

        return NULL;
    }

    dbh = psAlloc(sizeof(psDB));

    dbh->mysql = mysql;

    return dbh;
}

void psDBCleanup(psDB *dbh)
{
    // Check if argument dbh is NULL
    if(dbh == NULL) {
        psError(PS_ERR_BAD_PARAMETER_TYPE, true,
                PS_ERRORTEXT_psDB_INVALID_PSDB);
    } else {

        // Attempt to close specified database connection
        mysql_close(dbh->mysql);
        dbh->mysql = NULL;
        psFree(dbh);

        // ASC WARNING NOTE: the psDBSQLToPTypeTableCleanup cleanup routine
        // needs to be called first because it refers to
        // psDBGetPTypeToSQLTable ...
        psDBSQLToPTypeTableCleanup();
        psDBMySQLToSQLTableCleanup();
        psDBPTypeToSQLTableCleanup();
        psDBPTypeToMySQLTableCleanup();
    }
}

bool psDBCreate(psDB *dbh,
                const char *dbname)
{
    char            *query = NULL;
    bool            status;

    psStringAppend(&query, "CREATE DATABASE %s", dbname);

    // the MySQL C API notes that mysql_create_db() is deprecated
    status = p_psDBRunQuery(dbh, query);
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, "Failed to create new database.");
    }

    psFree(query);

    return status;
}

bool psDBChange(psDB *dbh,
                const char *dbname)
{
    // Verify database object is valid
    if(dbh == NULL) {
        psError(PS_ERR_BAD_PARAMETER_TYPE, true,
                PS_ERRORTEXT_psDB_INVALID_PSDB);
        return false;
    }

    // Attempt to select new database
    if (mysql_select_db(dbh->mysql, dbname) != 0) {
        psError(PS_ERR_UNKNOWN, true, PS_ERRORTEXT_psDB_FAILED_TO_CHANGE,
                mysql_error(dbh->mysql));

        return false;
    }

    return true;
}

bool psDBDrop(psDB *dbh,
              const char *dbname)
{
    char            *query = NULL;
    bool            status;

    psStringAppend(&query, "DROP DATABASE %s", dbname);

    // the MySQL C API notes that mysql_drop_db() is deprecated
    status = p_psDBRunQuery(dbh, query);
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, "Failed to drop database.");
    }

    psFree(query);

    return status;
}

bool psDBCreateTable(psDB *dbh,
                     const char *tableName,
                     const psMetadata *md)
{
    char            *query;
    bool            status;

    // Verify the database object is not NULL
    if(dbh == NULL) {
        psError(PS_ERR_BAD_PARAMETER_NULL,true,PS_ERRORTEXT_psDB_INVALID_PSDB);
        return false;
    }

    // Generate SQL query string
    query = psDBGenerateCreateTableSQL(tableName, md);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, PS_ERRORTEXT_psDB_QUERY_GEN_FAIL);
        return false;
    }

    // Run SQL query to create table
    status = p_psDBRunQuery(dbh, query);
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, PS_ERRORTEXT_psDB_TABLE_CREATE_FAIL);
    }

    // Free query string
    psFree(query);

    return status;
}

bool psDBDropTable(psDB *dbh,
                   const char *tableName)
{
    char            *query = NULL;
    bool            status;

    // Create SQL command string to drop table
    psStringAppend(&query, "DROP TABLE %s", tableName);

    // Execute query
    status = p_psDBRunQuery(dbh, query);
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, PS_ERRORTEXT_psDB_TABLE_DROP_FAIL);
    }

    psFree(query);

    return status;
}

psArray *psDBSelectColumn(psDB *dbh,
                          const char *tableName,
                          const char *col,
                          unsigned long long limit)
{
    MYSQL_RES       *result;            // complete db result set
    MYSQL_ROW       row;                // single row of db result set
    char            *query;             // SQL query
    my_ulonglong    rowCount;           // number of rows in db result set
    unsigned long   dataSize;           // size of field
    unsigned int    fieldCount;         // number of fields in db result set
    psArray         *column = NULL;     // return array
    psPtr           data;               // copy of result field

    // Generate SQL query string
    query = psDBGenerateSelectRowSQL(tableName, col, NULL, limit);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, PS_ERRORTEXT_psDB_QUERY_GEN_FAIL);
        return NULL;
    }

    // Execut SQL query string
    if (!p_psDBRunQuery(dbh, query)) {
        psError(PS_ERR_UNKNOWN, false, PS_ERRORTEXT_psDB_SEL_COL_FAIL);
        psFree(query);
        return NULL;
    }
    psFree(query);

    // Obtain query result and check for no data condition
    result = mysql_store_result(dbh->mysql);

    if (!result) {
        // no result set
        fieldCount = mysql_field_count(dbh->mysql);

        // if field count is zero the query returned no data.  If it's non-zero
        // then something bad has happened.
        if (fieldCount != 0) {
            psError(PS_ERR_UNEXPECTED_NULL, true, PS_ERRORTEXT_psDB_QUERY_NO_DATA,
                    mysql_error(dbh->mysql));
            return NULL;
        }
    }

    // Get number of rows returned in result
    rowCount = mysql_num_rows(result);
    // XXX has mysql's semantics changed?  If this is zero then why was a
    // result set returned?
    if (!rowCount) {
        mysql_free_result(result);
        return NULL;
    }

    // pre-allocate enough elements to hold the complete result set
    // then reset n to 0 so elements are added from the beginning of
    // the array
    column = psArrayAlloc(rowCount);
    column->n = 0;

    // Fetch each result
    while ((row = mysql_fetch_row(result))) {
        // get the first element of lengths array that is part of the
        // result set
        dataSize = *(mysql_fetch_lengths(result));

        // represent NULL as an empty string
        if (row[0] == NULL) {
            data = psStringCopy("");
        } else {
            data = psAlloc(dataSize+1);
            memcpy(data, row[0], dataSize);
            ((char*)data)[dataSize] = '\0';
        }

        // add field to return array
        psArrayAdd(column, 0, data);
        psFree(data);
    }

    // Clean up mysql memory
    mysql_free_result(result);

    return column;
}

// dest = assign to, source = source string psArray, conv = conversion function,
// type = type to cast to, pType = psElemType
#define PS_STR_ARRAY_TO_PTYPE(dest, source, conv, type, pType) \
{ \
    psPtr           myNaN; \
    int             i; \
    \
    for (i = 0; i < source->n; i++) { \
        if (strlen(source->data[i])) { \
            dest[i] = (type)conv(source->data[i]); \
        } else { \
            myNaN = psDBGetPTypeNaN(pType); \
            dest[i] = *(type *)myNaN; \
            psFree(myNaN); \
        } \
    } \
}

psVector *psDBSelectColumnNum(psDB *dbh,
                              const char *tableName,
                              const char *col,
                              psElemType type,
                              unsigned long long limit)
{
    psArray         *stringColumn;      // source psArray
    psVector        *column;            // dest psVector

    stringColumn = psDBSelectColumn(dbh, tableName, col, limit);
    if (!stringColumn) {
        // could be an error or the result set was just empty
        return NULL;
    }

    column = psVectorAlloc(stringColumn->n, type);
    //    column->n = column->nalloc;

    // conversion functions are a portability issue
    switch (type) {
    case PS_TYPE_S8:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.S8, stringColumn, atoi, psS8, PS_TYPE_S8);
        break;
    case PS_TYPE_S16:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.S16, stringColumn, atoi, psS16, PS_TYPE_S16);
        break;
    case PS_TYPE_S32:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.S32, stringColumn, atoi, psS32, PS_TYPE_S32);
        break;
    case PS_TYPE_S64:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.S64, stringColumn, atoll, psS64, PS_TYPE_S64);
        break;
    case PS_TYPE_U8:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.U8, stringColumn, atoi, psU8, PS_TYPE_U8);
        break;
    case PS_TYPE_U16:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.U16, stringColumn, atoi, psU16, PS_TYPE_U16);
        break;
    case PS_TYPE_U32:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.U32, stringColumn, atoi, psU32, PS_TYPE_U32);
        break;
    case PS_TYPE_U64:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.U64, stringColumn, atoll, psU64, PS_TYPE_U64);
        break;
    case PS_TYPE_F32:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.F32, stringColumn, atof, psF32, PS_TYPE_F32);
        break;
    case PS_TYPE_F64:
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.F64, stringColumn, atof, psF64, PS_TYPE_F64);
        break;
    case PS_TYPE_C32:
        // this is a bogus SQL type
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.C32, stringColumn, atof, psC32, PS_TYPE_C32);
        break;
    case PS_TYPE_C64:
        // this is a bogus SQL type
        PS_STR_ARRAY_TO_PTYPE(column->data.C64, stringColumn, atof, psC64, PS_TYPE_C64);
        break;
    case PS_TYPE_BOOL:
        // valid for psVector?
        column->n = column->nalloc;
        PS_STR_ARRAY_TO_PTYPE(column->data.U8, stringColumn, atoi, psU8, PS_TYPE_U8);
        break;
    default:
        psError(PS_ERR_BAD_PARAMETER_TYPE, true,
                "Invalid type specified in psDBSelectColumnNum.\n");
        psFree(stringColumn);
        psFree(column);
        return NULL;
        break;
    }

    psFree(stringColumn);

    return column;
}

psArray *psDBSelectRows(psDB *dbh,
                        const char *tableName,
                        const psMetadata *where,
                        unsigned long long limit)
{
    MYSQL_RES       *result;            // complete db result set
    MYSQL_ROW       row;                // single row of db result set
    MYSQL_FIELD     *field;             // field type info
    char            *query;             // SQL query
    my_ulonglong    rowCount;           // number of rows in db result set
    unsigned int    fieldCount;         // number of fields in db result set
    unsigned long   *fieldLength;       // field sizes
    long            len;                // field length
    psArray         *resultSet;         // return array
    int             i;                  // field index
    psMetadata      *md;                // a row
    psU32           pType;              // psElemType of a field
    psPtr           data;               // copy of result field

    // Create select row query
    query = psDBGenerateSelectRowSQL(tableName, NULL, where, limit);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "Query generation failed.");
        return NULL;
    }

    // Run SQL query
    if (!p_psDBRunQuery(dbh, query)) {
        psError(PS_ERR_UNKNOWN, false, "Query execution failed.");
        psFree(query);
        return NULL;
    }
    psFree(query);

    result = mysql_store_result(dbh->mysql);
    if (!result) {
        // no result set
        fieldCount = mysql_field_count(dbh->mysql);

        // if field count is zero the query should have returned no data.  If
        // it's non-zero then something bad has happened.
        if (fieldCount != 0) {
            psError(PS_ERR_UNEXPECTED_NULL, true, "Query returned no data.  Error: %s", mysql_error(dbh->mysql));

            return NULL;
        }
    }

    rowCount = mysql_num_rows(result);
    // XXX has mysql's semantics changed?  If this is zero then why was a
    // result set returned?
    if (rowCount == 0) {
        mysql_free_result(result);
        return NULL;
    }

    // pre-allocate enough elements to hold the complete result set
    // then reset n to 0 so elements are added from the beginning of
    // the array
    resultSet = psArrayAlloc(rowCount);
    resultSet->n = 0;

    field = mysql_fetch_fields(result);
    fieldCount = mysql_num_fields(result);

    while ((row = mysql_fetch_row(result))) {
        // allocate new psMetadata to represent a row
        md = psMetadataAlloc();

        fieldLength = mysql_fetch_lengths(result);

        for (i = 0; i < fieldCount; i++) {
            // lookup MySQL column type
            pType = psDBMySQLToPType(field[i].type, field[i].flags);

            len = fieldLength[i];
            if (len) {
                data = psAlloc(len+1);
                memcpy(data, row[i], len);
                ((char*)data)[len] = '\0';
            } else {
                data = psDBGetPTypeNaN(pType);
            }

            // copy field data and convert NULLs to the appropriate NaN value
            if (pType == PS_DATA_STRING) {
                psMetadataAddStr(md, PS_LIST_TAIL, field[i].name, 0, "", data);
            } else if (pType == PS_DATA_S32) {
                psMetadataAddS32(md, PS_LIST_TAIL, field[i].name, 0, "", atoll(data));
            } else if (pType == PS_DATA_F32) {
                psMetadataAddF32(md, PS_LIST_TAIL, field[i].name, 0, "", atof(data));
            } else if (pType == PS_DATA_F64) {
                psMetadataAddF64(md, PS_LIST_TAIL, field[i].name, 0, "", atof(data));
            } else if (pType == PS_DATA_BOOL) {
                psMetadataAdd(md, PS_LIST_TAIL, field[i].name, pType, "", atoi(data));
            } else {
                // XXX: assume binary string ...
                psMetadataAddStr(md, PS_LIST_TAIL, field[i].name, 0, "", data);
            }

            psFree(data);
        }

        // add row to result set
        psArrayAdd(resultSet, 0, md);
        psFree(md);
    }

    mysql_free_result(result);

    return resultSet;
}

bool psDBInsertOneRow(psDB *dbh,
                      const char *tableName,
                      const psMetadata *row)
{
    psArray         *rowSet;           // psArray of row to insert

    // Check for null row
    if(row == NULL) {
        psError(PS_ERR_UNKNOWN,true,PS_ERRORTEXT_psDB_INSERT_ROW_FAIL);
        return false;
    }

    // Create array to store single row
    rowSet = psArrayAlloc(1);
    rowSet->n = 0;
    psArrayAdd(rowSet, 0, (psPtr)row);

    // Execute function to insert rows
    if (!psDBInsertRows(dbh, tableName, rowSet)) {
        psError(PS_ERR_UNKNOWN, false, PS_ERRORTEXT_psDB_INSERT_ROW_FAIL);
        psFree(rowSet);
        return false;
    }

    psFree(rowSet);

    return true;
}

bool psDBInsertRows(psDB *dbh,
                    const char *tableName,
                    const psArray *rowSet)
{
    psMetadata      *row;               // row of data
    char            *query;             // SQL query
    MYSQL_STMT      *stmt;              // prepared db statement
    MYSQL_BIND      *bind;              // field values to insert
    unsigned long   paramCount;         // number of placeholders in query
    psU64           j;                  // row index

    // Verify database connections is set up
    if(dbh == NULL) {
        psError(PS_ERR_UNEXPECTED_NULL,true,PS_ERRORTEXT_psDB_INVALID_PSDB);
        return false;
    }

    // Verify row array is non-NULL
    if(rowSet == NULL) {
        psError(PS_ERR_UNEXPECTED_NULL,true,PS_ERRORTEXT_psDB_INSERT_ROW_FAIL);
        return false;
    }

    // we are assuming that all rows in the set have an identical with reguard
    // to field count and type
    row = rowSet->data[0];

    // Generate SQL query string
    query = psDBGenerateInsertRowSQL(tableName, row);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, PS_ERRORTEXT_psDB_QUERY_GEN_FAIL);
        return false;
    }

    // Prepare SQL statement
    stmt = mysql_stmt_init(dbh->mysql);
    if (!stmt) {
        psAbort(__func__, "mysql_stmt_init(), out of memory.");
    }
    if (mysql_stmt_prepare(stmt, query, (unsigned long)strlen(query))) {
        psError(PS_ERR_UNKNOWN, true, "Failed to prepare query.  Error: %s", mysql_stmt_error(stmt));
        mysql_stmt_close(stmt);
        psFree(query);
        return false;
    }

    psFree(query);

    // how many place holders are in our query
    paramCount = mysql_stmt_param_count(stmt);

    // structure larger enough to hold one field of data per place holder
    bind = psAlloc(sizeof(MYSQL_BIND) * paramCount);

    // loop over rows
    for (j = 0; j < rowSet->n; j++) {
        row = rowSet->data[j];

        // reset bind for each row
        memset(bind, 0, sizeof(MYSQL_BIND) * paramCount);

        if (!psDBPackRow(bind, row, paramCount)) {
            psError(PS_ERR_UNKNOWN, false, "Failed to pack params into bind structure.");

            mysql_rollback(dbh->mysql);

            psFree(bind);
            mysql_stmt_close(stmt);
        }

        if (mysql_stmt_bind_param(stmt, bind)) {
            psError(PS_ERR_UNKNOWN, true, "Failed to bind params.  Error: %s", mysql_stmt_error(stmt));

            mysql_rollback(dbh->mysql);

            psFree(bind);
            mysql_stmt_close(stmt);

            return false;
        }

        if (mysql_stmt_execute(stmt)) {
            psError(PS_ERR_UNKNOWN, true, "Failed to execute prepared statement.  Error: %s",
                    mysql_stmt_error(stmt));

            mysql_rollback(dbh->mysql);

            psFree(bind);
            mysql_stmt_close(stmt);

            return false;
        }
    } // end loop over rows

    // point of no return
    mysql_commit(dbh->mysql);

    psFree(bind);
    mysql_stmt_close(stmt);

    return true;
}

psArray *psDBDumpRows(psDB *dbh,
                      const char *tableName)
{
    return psDBSelectRows(dbh, tableName, NULL, 0);
}

psMetadata *psDBDumpCols(psDB *dbh,
                         const char *tableName)
{
    MYSQL_RES       *result;
    MYSQL_FIELD     *field;
    unsigned int    fieldCount;
    psMetadata      *table;
    psU32           pType;
    unsigned int    i;
    psPtr           column;

    // Verify database object is not null
    if(dbh == NULL) {
        psError(PS_ERR_UNEXPECTED_NULL,true,PS_ERRORTEXT_psDB_INVALID_PSDB);
        return NULL;
    }

    // Verify table name is not null
    if(tableName == NULL) {
        psError(PS_ERR_UNEXPECTED_NULL,true,PS_ERRORTEXT_psDB_NULL_TABLE);
        return NULL;
    }

    // find column types
    result = mysql_list_fields(dbh->mysql, tableName, NULL);
    if (!result) {
        psError(PS_ERR_UNEXPECTED_NULL, true, "Failed to retrieve column types.");
    }

    field = mysql_fetch_fields(result);
    fieldCount = mysql_num_fields(result);

    table = psMetadataAlloc();

    // fetch each column and load into psMetadata
    for (i =0; i < fieldCount; i++) {
        // find ptype of column
        pType = psDBMySQLToPType(field[i].type, field[i].flags);
        //psLogMsg( __func__, PS_LOG_INFO, "pType=[%ld]\n", pType );

        // if the ptype is PS_TYPE_PTR assume that it's a string and fetch the
        // column as an psArray of strings; otherwise fetch the column as a
        // psVector.
        if (pType == PS_DATA_STRING) {
            // PS_DATA_UNKNOWN -> PS_DATA_ARRAY ?
            column = psDBSelectColumn(dbh, tableName, field[i].name, 0);
            psMetadataAddArray(table, PS_LIST_TAIL, field[i].name, 0, "", column);
            //            psMetadataAddStr(table, PS_LIST_TAIL, field[i].name, "", column);
            psFree(column);
        } else {
            column = psDBSelectColumnNum(dbh, tableName, field[i].name, pType, 0);
            psMetadataAddVector(table, PS_LIST_TAIL, field[i].name, 0, "", column);
            psFree(column);
        }
    }

    return table;
}

long psDBUpdateRows(psDB *dbh,
                    const char *tableName,
                    const psMetadata *where,
                    const psMetadata *values)
{
    char            *query;
    MYSQL_STMT      *stmt;              // prepared db statement
    unsigned long   paramCount;         // number of placeholders in query
    MYSQL_BIND      *bind;              // field values to insert
    my_ulonglong    rowsAffected;       // number of rows affected by query

    // Verify database object is not NULL
    if(dbh == NULL) {
        psError(PS_ERR_UNEXPECTED_NULL,true,PS_ERRORTEXT_psDB_INVALID_PSDB);
        return -1;
    }

    // Generate SQL query to update row
    query = psDBGenerateUpdateRowSQL(tableName, where, values);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, PS_ERRORTEXT_psDB_QUERY_GEN_FAIL);
        return -1;
    }

    // Initialize SQL statement
    stmt = mysql_stmt_init(dbh->mysql);
    if (!stmt) {
        psAbort(__func__, "mysql_stmt_init(), out of memory.");
    }

    // Prepare SQL statement
    if (mysql_stmt_prepare(stmt, query, (unsigned long)strlen(query))) {
        psError(PS_ERR_UNKNOWN, true, PS_ERRORTEXT_psDB_SQL_PREPARE_FAIL, mysql_stmt_error(stmt));
        mysql_stmt_close(stmt);
        psFree(query);
        return -1;
    }
    psFree(query);

    // how many place holders are in our query
    paramCount = mysql_stmt_param_count(stmt);

    // structure large enough to hold one field of data per place holder
    bind = psAlloc(sizeof(MYSQL_BIND) * paramCount);

    // init bind
    memset(bind, 0, sizeof(MYSQL_BIND) * paramCount);

    if (!psDBPackRow(bind, values, paramCount)) {
        psFree(bind);
        mysql_stmt_close(stmt);
        return -1;
    }

    if (mysql_stmt_bind_param(stmt, bind)) {
        psError(PS_ERR_UNKNOWN, true, "Failed to bind params.  Error: %s", mysql_stmt_error(stmt));
        mysql_rollback(dbh->mysql);
        psFree(bind);
        mysql_stmt_close(stmt);
        return -1;
    }

    if (mysql_stmt_execute(stmt)) {
        psError(PS_ERR_UNKNOWN, true, "Failed to execute prepared statement.  Error: %s",
                mysql_stmt_error(stmt));
        mysql_rollback(dbh->mysql);
        psFree(bind);
        mysql_stmt_close(stmt);
        return -1;
    }

    psFree(bind);

    // mysql_stmt_affected_rows() must be called before a commit
    rowsAffected = mysql_stmt_affected_rows(stmt);

    // point of no return
    mysql_commit(dbh->mysql);

    mysql_stmt_close(stmt);

    return rowsAffected;
}

long psDBDeleteRows(psDB *dbh,
                    const char *tableName,
                    const psMetadata *where,
                    unsigned long long limit)
{
    char            *query;
    psS64           rows = 0;

    // Verify database is not NULL
    if(dbh == NULL) {
        psError(PS_ERR_UNEXPECTED_NULL, true, PS_ERRORTEXT_psDB_INVALID_PSDB);
        return -1;
    }

    // Create SQL statement string
    query = psDBGenerateDeleteRowSQL(tableName, where,limit);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "Query generation failed.");
        return -1;
    }

    // Run SQL query
    if (!p_psDBRunQuery(dbh, query)) {
        psError(PS_ERR_UNKNOWN, false, "Delete failed.");
        mysql_rollback(dbh->mysql);
        psFree(query);
        return -1;
    }
    psFree(query);

    // Get the number of affected row for the delete command
    rows = (psS64)mysql_affected_rows(dbh->mysql);

    // point of no return
    mysql_commit(dbh->mysql);

    return rows;
}

// database utility functions
/*****************************************************************************/

bool p_psDBRunQuery(psDB *dbh,
                    const char *format,
                    ...)
{
    va_list argPtr;
    char *dest = NULL;

    // Verify database object is not NULL
    if(dbh == NULL) {
        psError(PS_ERR_BAD_PARAMETER_NULL,true,PS_ERRORTEXT_psDB_INVALID_PSDB);
        return false;
    }

    // Run query
    va_start(argPtr, format);
    if(argPtr == NULL) {
        dest=psStringCopy(format);
    } else {
        int destSize = vsnprintf(dest, 0, format, argPtr);
        dest = psAlloc(destSize+1);
        vsnprintf(dest, destSize+1, format, argPtr);
    }
    va_end(argPtr);
    if (mysql_real_query(dbh->mysql, dest, (unsigned long)strlen(dest)) !=0) {
        psError(PS_ERR_UNKNOWN, true, PS_ERRORTEXT_psDB_SQL_QUERY_FAIL, mysql_error(dbh->mysql));
        psFree(dest);
        return false;
    }

    psFree(dest);
    return true;
}

static inline bool psDBPackRow(MYSQL_BIND *bind,
                               const psMetadata *values,
                               psU32 paramCount)
{
    psListIterator  *cursor;            // row iterator
    psMetadataItem  *item;              // field in row
    mysqlType       *mType;             // type tmp variable
    static bool     isNull = true;      // used in a MYSQL_BIND to indicate NULL,
    // this will be used outside of this func
    psU32           i;                  // field index

    // Verify values or bind are not null
    if((values == NULL) || (bind == NULL)) {
        return false;
    }

    // check size of values == paramCount ?
    cursor = psListIteratorAlloc(values->list, 0, false);

    // loop over fields
    i = 0;
    while ((item = psListGetAndIncrement(cursor))) {
        // lookup pType -> mysql type
        mType = psDBPTypeToMySQL(item->type);

        bind[i].buffer_type = mType->type;
        bind[i].is_unsigned = mType->isUnsigned;

        psFree(mType);

        // input data length is determined by the MYSQL_TYPE_* unless it's a string
        if (item->type == PS_TYPE_S32) {
            bind[i].length  = 0;
            bind[i].buffer  = &item->data.S32;
            bind[i].is_null = psDBIsPTypeNaN(item->type, &item->data.S32)
                              ? (my_bool *)&isNull
                              : NULL;
        } else if (item->type == PS_TYPE_F32) {
            bind[i].length  = 0;
            bind[i].buffer  = &item->data.F32;
            bind[i].is_null = psDBIsPTypeNaN(item->type, &item->data.F32)
                              ? (my_bool *)&isNull
                              : NULL;
        } else if (item->type == PS_TYPE_F64) {
            bind[i].length  = 0;
            bind[i].buffer  = &item->data.F64;
            bind[i].is_null = psDBIsPTypeNaN(item->type, &item->data.F64)
                              ? (my_bool *)&isNull
                              : NULL;
        } else if (item->type == PS_TYPE_BOOL) {
            // XXX: ASC HACK NOTE (2005/06/03): set extreme bytes to the
            // boolean character value.  sizeof(psBool)==4 which triggers an
            // endianess issue in the MySQL conversion (reading only 1 byte),
            // on Macintosh hardware (and maybe others?)
            unsigned int c  = (unsigned int)item->data.B;
            item->data.S32  = (unsigned int)((c<<24) | c);
            bind[i].length  = 0;
            bind[i].buffer  = &item->data.B;
            bind[i].is_null = psDBIsPTypeNaN(item->type, &item->data.B)
                              ? (my_bool *)&isNull
                              : NULL;
        } else if (item->type == PS_DATA_STRING) {
            // convert NaNs to NULL and set the buffer_length for strings

            bind[i].buffer_length = (unsigned long)strlen((char *)item->data.V);
            bind[i].length  = &bind[i].buffer_length;
            bind[i].buffer  = item->data.V;
            bind[i].is_null = *(char *)item->data.V == '\0'
                              ? (my_bool *)&isNull
                              : NULL;
        } else {
            psError(PS_ERR_BAD_PARAMETER_TYPE , true,
                    "FIXME: Only type of PS_TYPE_S32 (PS_DATA_S32), "
                    "PS_TYPE_F32 (PS_DATA_F32), PS_TYPE_F64 (PS_DATA_F64), "
                    "PS_TYPE_BOOL (PS_DATA_BOOL), "
                    "and PS_DATA_STRING are supported.");

            psFree(cursor);

            return false;
        }

        // increment field index
        i++;
    }

    psFree(cursor);

    return true;
}


// SQL generation functions
/*****************************************************************************/

static char *psDBGenerateCreateTableSQL(const char *tableName,
                                        const psMetadata *table)
{
    char            *query = NULL;      // complete query
    psMetadataItem  *item;              // column description
    psListIterator  *cursor;            // column iterator
    char            *colType;           // type lookup table

    // Verify input parameters are not null
    if ( (tableName==NULL) || (table==NULL) ) {
        psError(PS_ERR_BAD_PARAMETER_NULL, true, PS_ERRORTEXT_psDB_TABLE_PARAM_NULL);
        return NULL;
    }

    // Begin to create SQL string to create table
    psStringAppend(&query, "CREATE TABLE %s (", tableName);

    // Set list iterator at head of list
    cursor = psListIteratorAlloc(table->list, 0, false);

    // find column name and type
    while ((item = psListGetAndIncrement(cursor))) {
        if ((item->type == PS_DATA_S32)  || (item->type == PS_DATA_F32) || (item->type == PS_DATA_F64) ||
                (item->type == PS_TYPE_S32)  || (item->type == PS_TYPE_F32) || (item->type == PS_TYPE_F64) ||
                (item->type == PS_TYPE_BOOL) || (item->type == PS_DATA_BOOL)) {
            // + column name + _ + column type
            colType = psDBPTypeToSQL(item->type);
            psStringAppend(&query, "%s %s", item->name, colType);
            psFree(colType);
        } else if (item->type == PS_DATA_STRING) {
            // + column name + _ + varchar( + length + )
            psStringAppend(&query, "%s VARCHAR(%s)", item->name, item->data.V);
        } else {
            psError(PS_ERR_BAD_PARAMETER_TYPE, true,
                    "FIXME: Only type of PS_DATA_S32, PS_DATA_F32, PS_DATA_F64, PS_DATA_BOOL, "
                    "and PS_DATA_STRING are supported, (not %d).", item->type);

            psFree(query);
            psFree(cursor);

            return NULL;
        }
        if (strstr(item->comment, "AUTO_INCREMENT")) {
            psStringAppend(&query, "%s", " AUTO_INCREMENT");
        }

        // add a , after every column declaration except the last one
        if (!cursor->offEnd) {
            psStringAppend(&query, ", ");
        }
    }

    // Reset iterator to head of list
    psListIteratorSet(cursor, 0);

    // find database indexes
    while ((item = psListGetAndIncrement(cursor))) {
        // XXX NOTE: check for "Primary Key" first before "Key".  Yes we're
        // playing fast and loose here, but I think it's fine...
        if (strstr(item->comment, "Primary Key")) {
            psStringAppend(&query, ", PRIMARY KEY(%s)", item->name);
        } else if (strstr(item->comment, "Key")) {
            psStringAppend(&query, ", KEY(%s)", item->name);
        } else if (strstr(item->comment, "AUTO_INCREMENT")) {
            // this needs to be recognized as a key if it wasn't already
            psStringAppend(&query, ", KEY(%s)", item->name);
        }
    }

    psFree(cursor);

    // end column types + table type
    psStringAppend(&query, ") ENGINE=innodb");

    return query;
}

static char *psDBGenerateSelectRowSQL(const char *tableName,
                                      const char *col,
                                      const psMetadata *where,
                                      psU64 limit)
{
    char            *query = NULL;
    char            *whereSQL;
    char            *limitString;

    // select all columns if col is NULL
    if (col) {
        psStringAppend(&query, "SELECT %s FROM %s", col, tableName);
    } else {
        psStringAppend(&query, "SELECT * FROM %s", tableName);
    }

    // select all rows if where is NULL
    if (where) {
        whereSQL = psDBGenerateWhereSQL(where);
        if (!whereSQL) {
            psError(PS_ERR_UNEXPECTED_NULL, false, "SQL substring generation failed.");

            psFree(query);

            return NULL;
        }
        psStringAppend(&query, " %s", whereSQL);
        psFree(whereSQL);
    }

    // treat limit == 0 as "no limit"
    if (limit) {
        limitString = psDBIntToString(limit);
        psStringAppend(&query, " LIMIT %s", limitString);
        psFree(limitString);
    }

    return query;
}

static char *psDBGenerateInsertRowSQL(const char *tableName,
                                      const psMetadata *row)
{
    char            *query = NULL;
    psListIterator  *cursor;
    psMetadataItem  *item;

    // Check if table is NULL
    if(tableName == NULL) {
        psError(PS_ERR_UNEXPECTED_NULL,true,PS_ERRORTEXT_psDB_QUERY_GEN_FAIL);
        return NULL;
    }

    // Check if row is NULL
    if(row == NULL) {
        psError(PS_ERR_UNEXPECTED_NULL,true,PS_ERRORTEXT_psDB_QUERY_GEN_FAIL);
        return NULL;
    }

    // Start building query string
    psStringAppend(&query, "INSERT INTO %s (", tableName);

    cursor = psListIteratorAlloc(row->list, 0, false);

    // get field names
    while ((item = psListGetAndIncrement(cursor))) {
        psStringAppend(&query, item->name);

        // + , + _ between every field name
        if (!cursor->offEnd) {
            psStringAppend(&query, ", ");
        }
    }

    // end of field names
    psStringAppend(&query, ") VALUES (");

    psListIteratorSet(cursor, 0);

    // create value place holders
    while ((item = psListGetAndIncrement(cursor))) {
        psStringAppend(&query, "?");

        // + ", " between every place holder
        if (!cursor->offEnd) {
            psStringAppend(&query, ", ");
        }
    }

    psFree(cursor);

    // end of values
    psStringAppend(&query, ")");

    return query;
}

static char *psDBGenerateUpdateRowSQL(const char *tableName,
                                      const psMetadata *where,
                                      const psMetadata *values)
{
    char            *query = NULL;
    char            *setSQL;
    char            *whereSQL;

    // Verify where and values pointer are not NULL
    if ((!values) || (!where)) {
        psError(PS_ERR_BAD_PARAMETER_NULL, true, PS_ERRORTEXT_psDB_UPDATE_ROW_FAIL);
        return NULL;
    }

    // Create set SQL substring
    setSQL = psDBGenerateSetSQL(values);
    if (!setSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, PS_ERRORTEXT_psDB_SQL_SUBSTR_FAIL);
        return NULL;
    }

    // Create where SQL substring
    whereSQL = psDBGenerateWhereSQL(where);
    if (!whereSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, PS_ERRORTEXT_psDB_SQL_SUBSTR_FAIL);
        return NULL;
    }

    // Append substring to SQL update string
    psStringAppend(&query, "UPDATE %s %s %s", tableName, setSQL, whereSQL);
    psFree(setSQL);
    psFree(whereSQL);

    return query;
}

static char *psDBGenerateDeleteRowSQL(const char *tableName,
                                      const psMetadata *where,
                                      unsigned long long limit)
{
    char            *query = NULL;
    char            *whereSQL;
    char            *limitString;

    // delete all rows if where is NULL
    if (!where) {
        psStringAppend(&query, "TRUNCATE TABLE %s", tableName);
        return query;
    }

    // Generate where SQL substring
    whereSQL = psDBGenerateWhereSQL(where);
    if (!whereSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, PS_ERRORTEXT_psDB_SQL_SUBSTR_FAIL);
        return NULL;
    }

    psStringAppend(&query, "DELETE FROM %s %s", tableName, whereSQL);

    // Complete delete SQL command string
    // treat limit == 0 as "no limit"
    if (limit) {
        limitString = psDBIntToString(limit);
        psStringAppend(&query, " LIMIT %s", limitString);
        psFree(limitString);
    }
    psFree(whereSQL);

    return query;
}

static char *psDBGenerateWhereSQL(const psMetadata *where)
{
    char            *query = NULL;
    psListIterator  *cursor;
    psMetadataItem  *item;

    if (!where) {
        psError(PS_ERR_BAD_PARAMETER_NULL, true, "where param may not be NULL.");
        return NULL;
    }

    query = psStringCopy("WHERE ");

    cursor = psListIteratorAlloc(where->list, 0, false);

    // find column name and match pattern
    while ((item = psListGetAndIncrement(cursor))) {
        // item->data must be a string
        if ((item->type == PS_DATA_S32) || (item->type == PS_TYPE_S32)) {
            psStringAppend(&query, "%s=%d", item->name, (int)(item->data.S32));
        } else if ((item->type == PS_DATA_F32) || (item->type == PS_TYPE_F32)) {
            psStringAppend(&query, "%s=%g", item->name, (double)(item->data.F32));
        } else if ((item->type == PS_DATA_F64) || (item->type == PS_TYPE_F64)) {
            psStringAppend(&query, "%s=%g", item->name, (double)(item->data.F64));
        } else if ((item->type == PS_DATA_BOOL) || (item->type == PS_TYPE_BOOL)) {
            psStringAppend(&query, "%s=%d", item->name, (int)(item->data.B));
        } else if (item->type == PS_DATA_STRING) {
            // + column name + _ + like + _ + ' + value + '
            // check for NULL and empty ("") strings
            if (item->data.V == NULL || *(char *)item->data.V == '\0') {
                psStringAppend(&query, "%s IS NULL", item->name);
            } else {
                // XXX ASC NOTE: we should have a better match for
                // char & varchar columns than this.  LIKE is OK for
                // very large TEXT columns that really shouldn't be
                // used in a where clause...
                psStringAppend(&query, "%s LIKE '%s'", item->name, item->data.V);
            }
        } else {
            psError(PS_ERR_BAD_PARAMETER_TYPE, true,
                    "Only types PS_DATA_S32, PS_DATA_F32, PS_DATA_F64, PS_DATA_BOOL, PS_DATA_STRING are supported");

            psFree(cursor);
            psFree(query);

            return NULL;
        }

        // + " and " after every column declaration except the last one
        if (!cursor->offEnd) {
            psStringAppend(&query, " AND ");
        }
    }

    psFree(cursor);

    return query;
}

static char *psDBGenerateSetSQL(const psMetadata *set
                               )
{
    char            *query = NULL;
    psListIterator  *cursor;
    psMetadataItem  *item;

    if (!set
       ) {
        psError(PS_ERR_BAD_PARAMETER_NULL, true, "set param may not be NULL.");

        return NULL;
    }

    query = psStringCopy("SET ");

    cursor = psListIteratorAlloc(set
                                 ->list, 0, false);

    // find column name
    while ((item = psListGetAndIncrement(cursor))) {
        // + column name + _ + = + _ + ?
        psStringAppend(&query, "%s = ?", item->name);

        // + ", " after every column declaration except the last one
        if (!cursor->offEnd) {
            psStringAppend(&query, ",  ");
        }
    }

    psFree(cursor);

    return query;
}


// lookup table functions
/*****************************************************************************/

static psElemType psDBMySQLToPType(enum enum_field_types type,
                                   unsigned int flags)
{
    psHash          *mysqlToSQLTable;   // type lookup table
    psHash          *sqlToPSTable;      // type lookup table
    char            *key;               // hash tmp value
    char            *value;             // hash tmp value
    char            *sqlType;           // copy of lookup table result
    psU32           pType;              // psElemType of a field

    mysqlToSQLTable = psDBGetMySQLToSQLTable();

    // lookup MySQL column type
    key     = psDBIntToString((psU64)type);
    sqlType = psHashLookup(mysqlToSQLTable, key);
    psFree(key);
    psFree(mysqlToSQLTable);

    if (!sqlType) {
        psError(PS_ERR_UNEXPECTED_NULL, true, "type lookup failed.");
        return -1;
    }

    // MySQL column types can not be directly translated to PS
    // types as the the mysql types do not tell you if the value is
    // signed or unsigned.  The result is this ugly conversion from
    // mysql -> ascii -> ptype
    if (flags & UNSIGNED_FLAG) {
        psStringPrepend(&sqlType, "UNSIGNED ");
    }

    //psLogMsg( __func__, PS_LOG_INFO, "sqlType=[%s]\n", sqlType );

    // convert MySQL type to PS type
    sqlToPSTable = psDBGetSQLToPTypeTable();
    value = psHashLookup(sqlToPSTable, sqlType);
    psFree(sqlToPSTable);

    if (!value) {
        psError(PS_ERR_UNEXPECTED_NULL, true, "type lookup failed.");

        return -1;
    }

    pType = (psU32)atol(value);

    return pType;
}

static char *psDBPTypeToSQL(psElemType pType)
{
    psHash          *pTypeToSQLTable;   // type lookup table
    char            *key;               // hash tmp value
    char            *sqlType;             // hash tmp value

    pTypeToSQLTable = psDBGetPTypeToSQLTable();

    key = psDBIntToString((psU64)pType);
    sqlType = psHashLookup(pTypeToSQLTable, key);
    psFree(key);
    psFree(pTypeToSQLTable);

    if (!sqlType) {
        psError(PS_ERR_UNEXPECTED_NULL, true, "type lookup failed.");

        return NULL;
    }

    psMemIncrRefCounter(sqlType);

    return sqlType;
}

static mysqlType *psDBPTypeToMySQL(psElemType pType)
{
    psHash          *pTypeToMySQLTable; // type lookup table
    char            *key;               // hash tmp value
    mysqlType       *mType;             // mysqlType struct to return

    pTypeToMySQLTable = psDBGetPTypeToMySQLTable();

    key = psDBIntToString((psU64)pType);
    mType = psHashLookup(pTypeToMySQLTable, key);
    psFree(key);
    psFree(pTypeToMySQLTable);

    if (!mType) {
        psError(PS_ERR_UNEXPECTED_NULL, true, "type lookup failed.");

        return NULL;
    }

    psMemIncrRefCounter(mType);

    return mType;
}

static psHash *psDBGetPTypeToSQLTable(void)
{

    if (!pTypeToSQLlookupTable) {
        pTypeToSQLlookupTable = psHashAlloc(14);

        // no support for CHAR, TEXT or GLOB
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_S8,  "TINYINT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_S16, "SMALLINT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_S32, "INT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_S64, "BIGINT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_U8,  "UNSIGNED TINYINT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_U16, "UNSIGNED SMALLINT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_U32, "UNSIGNED INT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_U64, "UNSIGNED BIGINT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_F32, "FLOAT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_F64, "DOUBLE");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_C32, "PS_TYPE_C32 is not supported");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_C64, "PS_TYPE_C64 is not supported");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_TYPE_BOOL,"TINYINT");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_DATA_STRING, "VARCHAR");
    }

    // simulate true ref counting
    psMemIncrRefCounter(pTypeToSQLlookupTable);

    return pTypeToSQLlookupTable;
}

static void psDBPTypeToSQLTableCleanup(void)
{
    psFree(pTypeToSQLlookupTable);
    pTypeToSQLlookupTable = NULL;
}

static psHash *psDBGetSQLToPTypeTable(void)
{
    psHash          *psToSQLTable;
    psList          *list;
    psListIterator  *cursor;
    char            *key;
    char            *value;

    if (!sqlToPTypeLookupTable) {
        // invert the PSToSQL table
        psToSQLTable = psDBGetPTypeToSQLTable();
        sqlToPTypeLookupTable = psHashAlloc(psToSQLTable->n);

        list = psHashKeyList(psToSQLTable);
        cursor = psListIteratorAlloc(list, 0, false);

        while ((key = psListGetAndIncrement(cursor))) {
            value = psHashLookup(psToSQLTable, key);
            // switch key and value
            psHashAdd(sqlToPTypeLookupTable, value, key);
        }

        // Add BLOB & TEXT reverse mappings
        value = psDBIntToString((psU64)PS_DATA_STRING);
        psHashAdd(sqlToPTypeLookupTable, "BLOB",    value);
        psHashAdd(sqlToPTypeLookupTable, "TEXT",    value);
        psFree(value);

        // DECIMAL does not exist in the pType to SQL table
        value = psDBIntToString(0);
        psHashAdd(sqlToPTypeLookupTable, "DECIMAL", value);
        psFree(value);

        psFree(cursor);
        psFree(list);
        psFree(psToSQLTable);
    }

    // simulate true ref counting
    psMemIncrRefCounter(sqlToPTypeLookupTable);

    return sqlToPTypeLookupTable;
}

static void psDBSQLToPTypeTableCleanup(void)
{
    psFree(sqlToPTypeLookupTable);
    sqlToPTypeLookupTable = NULL;
}

static psHash *psDBGetMySQLToSQLTable(void)
{
    if (!mysqlToSqlLookupTable) {
        mysqlToSqlLookupTable = psHashAlloc(20);

        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_TINY,      "TINYINT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_SHORT,     "SMALLINT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_LONG,      "INT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_INT24,     "MEDIUMINT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_LONGLONG,  "BIGINT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_DECIMAL,   "DECIMAL");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_FLOAT,     "FLOAT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_DOUBLE,    "DOUBLE");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_TIMESTAMP, "TIMESTAMP");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_DATE,      "DATE");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_TIME,      "TIME");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_DATETIME,  "DATETIME");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_YEAR,      "YEAR");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_STRING,    "CHAR");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_VAR_STRING,"VARCHAR");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_BLOB,      "BLOB");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_SET,       "SET");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_ENUM,      "ENUM");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_NULL,      "NULL-type");
        psDBAddToLookupTable(mysqlToSqlLookupTable, FIELD_TYPE_CHAR,      "TINYINT");
    }

    // simulate true ref counting
    psMemIncrRefCounter(mysqlToSqlLookupTable);

    return mysqlToSqlLookupTable;
}

static void psDBMySQLToSQLTableCleanup(void)
{
    psFree(mysqlToSqlLookupTable);
    mysqlToSqlLookupTable = NULL;
}

static psHash *psDBGetPTypeToMySQLTable(void)
{
    if (!pTypeToMysqlLookupTable) {
        pTypeToMysqlLookupTable = psHashAlloc(14);

        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_S8,     psDBMySQLTypeAlloc(MYSQL_TYPE_TINY,       false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_S16,    psDBMySQLTypeAlloc(MYSQL_TYPE_SHORT,      false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_S32,    psDBMySQLTypeAlloc(MYSQL_TYPE_LONG,       false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_S64,    psDBMySQLTypeAlloc(MYSQL_TYPE_LONGLONG,   false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_U8,     psDBMySQLTypeAlloc(MYSQL_TYPE_TINY,       true));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_U16,    psDBMySQLTypeAlloc(MYSQL_TYPE_SHORT,      true));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_U32,    psDBMySQLTypeAlloc(MYSQL_TYPE_LONG,       true));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_U64,    psDBMySQLTypeAlloc(MYSQL_TYPE_LONGLONG,   true));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_F32,    psDBMySQLTypeAlloc(MYSQL_TYPE_FLOAT,      false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_F64,    psDBMySQLTypeAlloc(MYSQL_TYPE_DOUBLE,     false));
        // bogus type
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_C32,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        // bogus type
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_C64,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_BOOL,   psDBMySQLTypeAlloc(MYSQL_TYPE_TINY,       true));
        // XXX: removed PS_TYPE_PTR, can this be removed too?
        // psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_TYPE_PTR,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));

        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_STRING,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_VECTOR,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_IMAGE,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_HASH,   psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_LOOKUPTABLE,
                                 psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        //        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_JPEG,   psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        //        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_PNG,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        //        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_ASTROM, psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        psDBAddVoidToLookupTable(pTypeToMysqlLookupTable, PS_DATA_UNKNOWN,psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
    }

    // simulate true ref counting
    psMemIncrRefCounter(pTypeToMysqlLookupTable);

    return pTypeToMysqlLookupTable;
}

static void psDBPTypeToMySQLTableCleanup(void)
{
    psFree(pTypeToMysqlLookupTable);
    pTypeToMysqlLookupTable = NULL;
}

static psPtr psDBMySQLTypeAlloc(enum enum_field_types type,
                                bool isUnsigned)
{
    mysqlType       *mType;

    mType = psAlloc(sizeof(mysqlType));
    mType->type       = type;
    mType->isUnsigned = isUnsigned;

    return mType;
}

static void psDBAddToLookupTable(psHash *lookupTable,
                                 psU32 type,
                                 const char *string)
{
    char            *key;
    char            *value;

    key = psDBIntToString((psU64)type);
    value = psStringCopy(string);

    psHashAdd(lookupTable, key, value);

    psFree(key);
    psFree(value);
}

static void psDBAddVoidToLookupTable(psHash *lookupTable,
                                     psU32 type,
                                     psPtr value)
{
    char            *key;

    key = psDBIntToString((psU64)type);

    psHashAdd(lookupTable, key, value);

    // destructive of value parameter
    psFree(value);
    psFree(key);
}


// pType utility functions
/*****************************************************************************/

#define PS_NAN_ALLOC(dest, type, nan) \
dest = psAlloc(sizeof(type)); \
*(type *)dest = nan;

static psPtr psDBGetPTypeNaN(psElemType pType)
{
    psPtr           myNaN = NULL;

    switch (pType) {
    case PS_TYPE_S8:
        PS_NAN_ALLOC(myNaN, psS8, PS_MAX_S8);
        break;
    case PS_TYPE_S16:
        PS_NAN_ALLOC(myNaN, psS16, PS_MAX_S16);
        break;
    case PS_TYPE_S32:
        PS_NAN_ALLOC(myNaN, psS32, PS_MAX_S32);
        break;
    case PS_TYPE_S64:
        PS_NAN_ALLOC(myNaN, psS64, PS_MAX_S64);
        break;
    case PS_TYPE_U8:
        PS_NAN_ALLOC(myNaN, psU8, PS_MAX_U8);
        break;
    case PS_TYPE_U16:
        PS_NAN_ALLOC(myNaN, psU16, PS_MAX_U16);
        break;
    case PS_TYPE_U32:
        PS_NAN_ALLOC(myNaN, psU32, PS_MAX_U32);
        break;
    case PS_TYPE_U64:
        PS_NAN_ALLOC(myNaN, psU64, PS_MAX_U64);
        break;
    case PS_TYPE_F32:
        PS_NAN_ALLOC(myNaN, psF32, NAN);
        break;
    case PS_TYPE_F64:
        PS_NAN_ALLOC(myNaN, psF64, NAN);
        break;
    case PS_TYPE_C32:
        // this is a bogus SQL type
        PS_NAN_ALLOC(myNaN, psC32, NAN);
        break;
    case PS_TYPE_C64:
        // this is a bogus SQL type
        PS_NAN_ALLOC(myNaN, psC64, NAN);
        break;
    case PS_TYPE_BOOL:
        // XXX: what is NaN for a bool?
        PS_NAN_ALLOC(myNaN, psU8, PS_MAX_U8);
        break;
    }

    return myNaN;
}

#define PS_IS_NAN(type, data, nan) *(type *)data == nan

static bool psDBIsPTypeNaN(psElemType pType,
                           psPtr data)
{
    bool    isNaN = NULL;

    switch (pType) {
    case PS_TYPE_S8:
        isNaN = PS_IS_NAN(psS8, data, PS_MAX_S8);
        break;
    case PS_TYPE_S16:
        isNaN = PS_IS_NAN(psS16, data, PS_MAX_S16);
        break;
    case PS_TYPE_S32:
        isNaN = PS_IS_NAN(psS32, data, PS_MAX_S32);
        break;
    case PS_TYPE_S64:
        isNaN = PS_IS_NAN(psS64, data, PS_MAX_S64);
        break;
    case PS_TYPE_U8:
        isNaN = PS_IS_NAN(psU8, data, PS_MAX_U8);
        break;
    case PS_TYPE_U16:
        isNaN = PS_IS_NAN(psU16, data, PS_MAX_U16);
        break;
    case PS_TYPE_U32:
        isNaN = PS_IS_NAN(psU32, data, PS_MAX_U32);
        break;
    case PS_TYPE_U64:
        isNaN = PS_IS_NAN(psU64, data, PS_MAX_U64);
        break;
    case PS_TYPE_F32:
        isNaN = PS_IS_NAN(psF32, data, NAN);
        break;
    case PS_TYPE_F64:
        isNaN = PS_IS_NAN(psF64, data, NAN);
        break;
    case PS_TYPE_C32:
        // this is a bogus SQL type
        isNaN = PS_IS_NAN(psC32, data, NAN);
        break;
    case PS_TYPE_C64:
        // this is a bogus SQL type
        isNaN = PS_IS_NAN(psC64, data, NAN);
        break;
    case PS_TYPE_BOOL:
        // XXX: what is NaN for a bool?
        isNaN = PS_IS_NAN(psU8, data, PS_MAX_U8);
        break;
    }

    return isNaN;
}


// string utility functions
/*****************************************************************************/

static char *psDBIntToString(psU64 n)
{
    char            *string;
    size_t          length;

    // length of string + \0
    // if n is 0, length is 1 char + \0
    length = n ? (size_t)log10((double)n) + 1
             : 2;
    string = psAlloc(length);
    sprintf(string, "%li", (long int)n);

    return string;
}

#endif // OMIT_PSDB
