/** @file  psDB.c
 *
 *  @brief database functions
 *
 *  This file contains functions that perform basic database operations.  MySQL
 *  4.1.2 or newer is required.
 *
 *  @author Joshua Hoblitt
 *
 *  @version $Revision: 1.34 $ $Name: not supported by cvs2svn $
 *  @date $Date: 2005-02-17 03:22:06 $
 *
 *  Copyright 2005 Joshua Hoblitt, University of Hawaii
 */

#include <stdio.h>
#include <stdarg.h>
#include <string.h>
#include <math.h>
#include <mysql/mysql.h>
#include <mysql/mysql_com.h> // enum_field_types

#include "psDB.h"
#include "pslib.h"

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

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

// SQL generation functions
static char    *psDBGenerateCreateTableSQL(const char *tableName, psMetadata *where);
static char    *psDBGenerateSelectRowSQL(const char *tableName, const char *col, psMetadata *where, psU64 limit);
static char    *psDBGenerateInsertRowSQL(const char *tableName, psMetadata *row);
static char    *psDBGenerateUpdateRowSQL(const char *tableName, psMetadata *where, psMetadata *values);
static char    *psDBGenerateDeleteRowSQL(const char *tableName, psMetadata *where);
static char    *psDBGenerateWhereSQL(psMetadata *where);
static char    *psDBGenerateSetSQL(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);
static ssize_t  psStringAppend(char **dest, const char *format, ...);
static ssize_t  psStringPrepend(char **dest, const char *format, ...);


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

    if (!mysql_real_connect(mysql, host, user, passwd, dbname, 0, NULL, 0)) {
        psError(PS_ERR_UNKNOWN, true, "Failed to connect to database.  Error: %s", mysql_error(mysql));

        mysql_close(mysql);

        return NULL;
    }

    dbh = psAlloc(sizeof(psDB));

    dbh->mysql = mysql;

    return dbh;
}

void psDBCleanup(psDB *dbh)
{
    mysql_close(dbh->mysql);
    psFree(dbh);

    // these lookup tables should probably be moved into psDB to prevent them
    // from being flushed in the case that a database handle is closed while
    // one or more are still open
    psDBMySQLToSQLTableCleanup();
    psDBPTypeToSQLTableCleanup();
    psDBSQLToPTypeTableCleanup();
    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 = 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)
{
    if (mysql_select_db(dbh->mysql, dbname) != 0) {
        psError(PS_ERR_UNKNOWN, true, "Failed to change database.  Error: %s", 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 = 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, psMetadata *md)
{
    char            *query;
    bool            status;

    query = psDBGenerateCreateTableSQL(tableName, md);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "Query generation failed.");

        return NULL;
    }

    status = psDBRunQuery(dbh, query);
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, "Failed to create table.");
    }

    psFree(query);

    return status;
}

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

    psStringAppend(&query, "DROP TABLE %s", tableName);

    status = psDBRunQuery(dbh, query);
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, "Failed to drop table.");
    }

    psFree(query);

    return status;
}

psArray *psDBSelectColumn(psDB *dbh, const char *tableName, const char *col, psU64 limit)
{
    MYSQL_RES       *result;            // complete db result set
    MYSQL_ROW       row;                // single row of db result set
    char            *query;             // SQL query
    psU64           rowCount;           // number of rows in db result set
    psU64           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

    query = psDBGenerateSelectRowSQL(tableName, col, NULL, limit);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "Query generation failed.");

        return NULL;
    }

    if (!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 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 = (psU64)mysql_num_rows(result);

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

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

        // represent NULL as an empty string
        if (row[0] == NULL) {
            data = psStringCopy("");
        } else {
            data = psStringNCopy(row[0], dataSize);
        }

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

    mysql_free_result(result);

    return column;
}

// dest = assign too, 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 pType, const psU64 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, pType);

    // conversion functions are a portability issue
    switch (pType) {
        case PS_TYPE_S8:
            PS_STR_ARRAY_TO_PTYPE(column->data.S8, stringColumn, atoi, psS8, PS_TYPE_S8);
            break;
        case PS_TYPE_S16:
            PS_STR_ARRAY_TO_PTYPE(column->data.S16, stringColumn, atoi, psS16, PS_TYPE_S16);
            break;
        case PS_TYPE_S32:
            PS_STR_ARRAY_TO_PTYPE(column->data.S32, stringColumn, atoi, psS32, PS_TYPE_S32);
            break;
        case PS_TYPE_S64:
            PS_STR_ARRAY_TO_PTYPE(column->data.S64, stringColumn, atoll, psS64, PS_TYPE_S64);
            break;
        case PS_TYPE_U8:
            PS_STR_ARRAY_TO_PTYPE(column->data.U8, stringColumn, atoi, psU8, PS_TYPE_U8);
            break;
        case PS_TYPE_U16:
            PS_STR_ARRAY_TO_PTYPE(column->data.U16, stringColumn, atoi, psU16, PS_TYPE_U16);
            break;
        case PS_TYPE_U32:
            PS_STR_ARRAY_TO_PTYPE(column->data.U32, stringColumn, atoi, psU32, PS_TYPE_U32);
            break;
        case PS_TYPE_U64:
            PS_STR_ARRAY_TO_PTYPE(column->data.U64, stringColumn, atoll, psU64, PS_TYPE_U64);
            break;
        case PS_TYPE_F32:
            PS_STR_ARRAY_TO_PTYPE(column->data.F32, stringColumn, atof, psF32, PS_TYPE_F32);
            break;
        case PS_TYPE_F64:
            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
            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?
            break;
        case PS_TYPE_PTR:
            // valid for psVector?
            break;
    }

    psFree(stringColumn);

    return column;
}

psArray *psDBSelectRows(psDB *dbh, const char *tableName, psMetadata *where, const psU64 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
    psU64           rowCount;           // number of rows in db result set
    psU64           fieldCount;         // number of fields in db result set
    psU64           *fieldLength;       // field sizes
    psArray         *resultSet;         // return array
    int             i;                  // field index
    psMetadata      *md;                // a row
    psU32           pType;              // psElemType of a field
    psU32           type;               // psMetadataType of a a field
    psPtr           data;               // copy of result field

    query = psDBGenerateSelectRowSQL(tableName, NULL, where, limit);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "Query generation failed.");

        return NULL;
    }

    if (!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 = (psU64)mysql_num_rows(result);

    // 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 = (psU64 *)mysql_fetch_lengths(result);

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

            // copy field data and convert NULLs to the appropriate NaN value
            if (pType == PS_TYPE_PTR) {
                type = PS_META_STR;
                data = fieldLength[i] ? psStringNCopy(row[i], fieldLength[i])
                                      : psDBGetPTypeNaN(pType);
            } else {
                type = PS_META_PRIMITIVE;
                if (fieldLength[i]) {
                    data = psAlloc(fieldLength[i]);
                    memcpy(data, row[i], fieldLength[i]);
                } else {
                    data = psDBGetPTypeNaN(pType);
                }
            }

            // add new field to row
            psMetadataAdd(md, i, field[i].name, pType, type, NULL, data);

            psFree(data);
        }

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

    mysql_free_result(result);

    return resultSet;
}

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

    // this is to prevent row from being free'd when rowSet is free'd. this
    // will have to be removed when psArrayAdd is modified to do this for you.
    psMemIncrRefCounter(row);

    rowSet = psArrayAlloc(1);
    rowSet->n = 0;
    psArrayAdd(rowSet, 0, row);

    if (!psDBInsertRows(dbh, tableName, rowSet)) {
        psError(PS_ERR_UNKNOWN, false, "Insert failed.");

        psFree(rowSet);

        return false;
    }

    psFree(rowSet);

    return true;
}

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

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

    query = psDBGenerateInsertRowSQL(tableName, row);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "Query generation failed.");

        return NULL;
    }

    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 = (psU32)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;
    psU32           fieldCount;
    psMetadata      *table;
    psU32           pType;
    psU32           i;
    psPtr           column;

    // 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);
        
        // 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_TYPE_PTR) {
            // PS_META_UNKNOWN -> PS_META_ARRAY ?
            column = psDBSelectColumn(dbh, tableName, field[i].name, 0);
            psMetadataAdd(table, 0, field[i].name, PS_TYPE_PTR, PS_META_UNKNOWN, "psArray", column);
            // FIXME
            //psFree(column);
        } else {
            column = psDBSelectColumnNum(dbh, tableName, field[i].name, pType, 0);
            psMetadataAdd(table, 0, field[i].name, PS_TYPE_PTR, PS_META_VEC, "psVector", column);
            // FIXME
            //psFree(column);
        }
    }
    
    return table;
}

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

    query = psDBGenerateUpdateRowSQL(tableName, where, values);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "Query generation failed.");

        return -1;
    }

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

    psFree(query);

    // how many place holders are in our query
    paramCount = (psU32)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);

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

    rowsAffected = (psU64)mysql_stmt_affected_rows(stmt);

    mysql_stmt_close(stmt);

    return rowsAffected;
}

psS64 psDBDeleteRows(psDB *dbh, const char *tableName, psMetadata *where)
{
    char            *query;

    query = psDBGenerateDeleteRowSQL(tableName, where);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "Query generation failed.");

        return -1;
    }

    if (!psDBRunQuery(dbh, query)) {
        psError(PS_ERR_UNKNOWN, false, "Delete failed.");

        mysql_rollback(dbh->mysql);

        psFree(query);

        return -1;
    }

    psFree(query);

    if (!where) {
        // we truncated the table so mysql_affected_rows() won't work
        return 1;
    }

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

    return (psS64)mysql_affected_rows(dbh->mysql);
}

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

static bool psDBRunQuery(psDB *dbh, const char *query)
{
    if (mysql_real_query(dbh->mysql, query, (unsigned long)strlen(query)) !=0) {
        psError(PS_ERR_UNKNOWN, true, "Failed to execute query.  Error: %s\n", mysql_error(dbh->mysql));
           
        return false;
    }

    return true;
}

static inline bool psDBPackRow(MYSQL_BIND *bind, 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

    // check size of values == paramCount ?

    cursor = psListIteratorAlloc(values->list, 0);

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

        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_META_PRIMITIVE) {
            switch (item->pType) {
                case PS_TYPE_S32:
                    bind[i].length  = 0;
                    bind[i].buffer  = &item->data.S32;
                    bind[i].is_null = psDBIsPTypeNaN(item->pType, &item->data.S32) 
                                    ? (my_bool *)&isNull
                                    : NULL;
                    break;
                case PS_TYPE_F32:
                    bind[i].length  = 0;
                    bind[i].buffer  = &item->data.F32;
                    bind[i].is_null = psDBIsPTypeNaN(item->pType, &item->data.F32) 
                                    ? (my_bool *)&isNull
                                    : NULL;
                    break;
                case PS_TYPE_F64:
                    bind[i].length  = 0;
                    bind[i].buffer  = &item->data.F64;
                    bind[i].is_null = psDBIsPTypeNaN(item->pType, &item->data.F64) 
                                    ? (my_bool *)&isNull
                                    : NULL;
                    break;
                default:
                    psError(PS_ERR_BAD_PARAMETER_TYPE , true,
            "FIXME: Only PS_META_PRIMITIVE values of PS_TYPE_S32, PS_TYPE_F32, and PS_TYPE_F64 supported.");
                    break;
            }
        // convert NaNs to NULL and set the buffer_length for strings
        } else if ((item->type == PS_META_STR) && (item->pType == PS_TYPE_PTR)) {
            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,
                "Only types of PS_META_PRIMITIVE and PS_META_STR are supported.");

            psFree(cursor);

            return false;
        }

        // increment field index
        i++;
    }

    psFree(cursor);

    return true;
}


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

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

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

        return NULL;
    }

    psStringAppend(&query, "CREATE TABLE %s (", tableName);

    cursor = psListIteratorAlloc(table->list, 0);

    // find column name and type
    while ((item = psListGetNext(cursor))) {
        if (item->type == PS_META_PRIMITIVE) {
            // + column name + _ + column type
            colType = psDBPTypeToSQL(item->pType);
            psStringAppend(&query, "%s %s", item->name, colType);
            psFree(colType);
        } else if ((item->type == PS_META_STR) && (item->pType == PS_TYPE_PTR)) {
            // + column name + _ + varchar( + length + )
            psStringAppend(&query, "%s VARCHAR(%s)", item->name, item->data.V);
        } else {
            psError(PS_ERR_BAD_PARAMETER_TYPE, true,
                "Only types of PS_META_PRIMITIVE and PS_META_STR are supported.");

            psFree(query);
            psFree(cursor);

            return NULL;
        }

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

    psListIteratorSet(cursor, 0);

    // find database indexes
    while ((item = psListGetNext(cursor))) {
        if ((strncmp(item->comment, "Primary Key", strlen("Primary Key"))) == 0) {
            psStringAppend(&query, ", PRIMARY KEY(%s)", item->name);
        } else if ((strncmp(item->comment, "Key", strlen("Key"))) == 0) {
            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, psMetadata *where, const 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, psMetadata *row)
{
    char            *query = NULL;
    psListIterator  *cursor;
    psMetadataItem  *item;

    psStringAppend(&query, "INSERT INTO %s (", tableName);

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

    // get field names
    while ((item = psListGetNext(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 = psListGetNext(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, psMetadata *where, psMetadata *values)
{
    char            *query = NULL;
    char            *setSQL;
    char            *whereSQL;

    if ((!values) || (!where)) {
        psError(PS_ERR_BAD_PARAMETER_NULL, true, "values and where params may not be NULL.");

        return NULL;
    }

    setSQL = psDBGenerateSetSQL(values);
    if (!setSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "SQL substring generation failed.");

        return NULL;
    }

    whereSQL = psDBGenerateWhereSQL(where);
    if (!whereSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "SQL substring generation failed.");

        return NULL;
    }

    psStringAppend(&query, "UPDATE %s %s %s", tableName, setSQL, whereSQL);
    psFree(setSQL);
    psFree(whereSQL);

    return query;
}

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

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

        return query;
    }
    
    whereSQL = psDBGenerateWhereSQL(where);
    if (!whereSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "SQL substring generation failed.");

        return NULL;
    }

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

    return query;
}

static char *psDBGenerateWhereSQL(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);

    // find column name and match pattern
    while ((item = psListGetNext(cursor))) {
        // item->data must be a string
        if (item->type == PS_META_STR) {
            // + column name + _ + like + _ + ' + value + '
            if (*(char *)item->data.V == '\0') {
                psStringAppend(&query, "%s IS NULL", item->name);
            } else {
                psStringAppend(&query, "%s LIKE '%s'", item->name, item->data.V);
            }
        } else {
            psError(PS_ERR_BAD_PARAMETER_TYPE, true, "Only a type of PS_META_STR is 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(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);

    // find column name 
    while ((item = psListGetNext(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();
    sqlToPSTable    = psDBGetSQLToPTypeTable();

    // 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.");

        psFree(sqlToPSTable);

        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 ");
    }

    // convert MySQL type to PS type
    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)
{
    static psHash   *lookupTable = NULL;

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

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

    // simulate true ref counting
    psMemIncrRefCounter(lookupTable);

    return lookupTable;
}

static void psDBPTypeToSQLTableCleanup(void) 
{
    psHash          *lookupTable;

    lookupTable = psDBGetPTypeToSQLTable();

    psMemDecrRefCounter(lookupTable);
    psFree(lookupTable);
}

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

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

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

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

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

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

    // simulate true ref counting
    psMemIncrRefCounter(lookupTable);

    return lookupTable;
}

static void psDBSQLToPTypeTableCleanup(void) 
{
    psHash          *lookupTable;

    lookupTable = psDBGetSQLToPTypeTable();

    psMemDecrRefCounter(lookupTable);
    psFree(lookupTable);
}

static psHash *psDBGetMySQLToSQLTable(void)
{
    static psHash   *lookupTable = NULL;

    if (!lookupTable) {
        lookupTable = psHashAlloc(20);

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

    // simulate true ref counting
    psMemIncrRefCounter(lookupTable);

    return lookupTable;
}

static void psDBMySQLToSQLTableCleanup(void)
{
    psHash          *lookupTable;

    lookupTable = psDBGetMySQLToSQLTable();

    psMemDecrRefCounter(lookupTable);
    psFree(lookupTable);
}

static psHash *psDBGetPTypeToMySQLTable(void)
{
    static psHash   *lookupTable = NULL;

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

        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_S8,     psDBMySQLTypeAlloc(MYSQL_TYPE_TINY,       false));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_S16,    psDBMySQLTypeAlloc(MYSQL_TYPE_SHORT,      false));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_S32,    psDBMySQLTypeAlloc(MYSQL_TYPE_LONG,       false));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_S64,    psDBMySQLTypeAlloc(MYSQL_TYPE_LONGLONG,   false));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_U8,     psDBMySQLTypeAlloc(MYSQL_TYPE_TINY,       true));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_U16,    psDBMySQLTypeAlloc(MYSQL_TYPE_SHORT,      true));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_U32,    psDBMySQLTypeAlloc(MYSQL_TYPE_LONG,       true));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_U64,    psDBMySQLTypeAlloc(MYSQL_TYPE_LONGLONG,   true));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_F32,    psDBMySQLTypeAlloc(MYSQL_TYPE_FLOAT,      false));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_F64,    psDBMySQLTypeAlloc(MYSQL_TYPE_DOUBLE,     false));
        // bogus type
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_C32,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        // bogus type
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_C64,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_BOOL,   psDBMySQLTypeAlloc(MYSQL_TYPE_TINY,       true));
        psDBAddVoidToLookupTable(lookupTable, PS_TYPE_PTR,    psDBMySQLTypeAlloc(MYSQL_TYPE_VAR_STRING, false));
    }

    // simulate true ref counting
    psMemIncrRefCounter(lookupTable);

    return lookupTable;
}

static void psDBPTypeToMySQLTableCleanup(void)
{
    psHash          *lookupTable;

    lookupTable = psDBGetPTypeToMySQLTable();

    psMemDecrRefCounter(lookupTable);
    psFree(lookupTable);
}

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;

    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:
            // what is NaN for a bool?
            break;
        case PS_TYPE_PTR:
            PS_NAN_ALLOC(myNaN, char, '\0');
            break;
    }

    return myNaN;
}

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

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

    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:
            // what is NaN for a bool?
            break;
        case PS_TYPE_PTR:
            isNaN = PS_IS_NAN(char, data, '\0');
            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", n);

    return string;
}

static ssize_t psStringAppend(char **dest, const char *format, ...)
{
    va_list         args;
    size_t          length;             // complete string length (sans \0)
    size_t          oldLength;          // original string length (sans \0)
    ssize_t         tailLength;         // length of string to append

    if (!*dest) {
        *dest = psStringCopy("");
        oldLength = 0;
    } else {
        // size of existing string
        oldLength = strlen(*dest);
    }

    // find the size of the string to append
    va_start(args, format);
    // C99 guarentees vsnprintf() to work as expected with size = 0
    tailLength = vsnprintf(*dest, 0, format, args);
    va_end(args);

    // if the new tail is zero length, return the length of the old string.  if
    // it's a format error, return the error code.
    if (tailLength < 1) {
        return tailLength == 0 ? oldLength : tailLength;
    }

    // new string length (sans \0)
    length = oldLength + tailLength;

    // realloc string to string + tail + \0
    *dest = psRealloc(*dest, length + 1);

    // append tail + \0
    va_start(args, format);
    vsnprintf(*dest + oldLength, tailLength + 1, format, args);
    va_end(args);

    return length;
}

static ssize_t psStringPrepend(char **dest, const char *format, ...)
{
    va_list         args;
    size_t          length;             // complete string length (sans \0)
    ssize_t         headLength;         // length of string to prepend
    char            *oldDest;           // copy of original string

    if (!*dest) {
        // makes the string backup and concatination pointless 
        *dest = psStringCopy("");
        length = 0;
    } else {
        // size of existing string
        length = strlen(*dest);
    }

    // find the size of the string to prepend
    va_start(args, format);
    // C99 guarentees vsnprintf() to work as expected with size = 0
    headLength = vsnprintf(*dest, 0, format, args);
    va_end(args);

    // if the new head is zero length, return the length of the old string.  if
    // it's a format error, return the error code.
    if (headLength < 1) {
        return headLength == 0 ? length : headLength;
    }

    // backup original string
    oldDest = psStringCopy(*dest);

    // new string length (sans \0)
    length += headLength;

    // realloc string to head + string + \0
    *dest = psRealloc(*dest, length + 1);

    // copy the new head to the beginning of string
    va_start(args, format);
    vsnprintf(*dest, length + 1, format, args);
    va_end(args);

    // append the original string
    strncat(*dest, oldDest, length + 1);

    psFree(oldDest);

    return length;
}
