/** @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.92 $ $Name: not supported by cvs2svn $
 *  @date $Date: 2006-09-07 01:43:36 $
 *
 *  Copyright (C) 2005-2006  Joshua Hoblitt, University of Hawaii
 */

#ifndef OMIT_PSDB

#include <stdio.h>
#include <stdarg.h>
#include <string.h>
#include <float.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 "psAssert.h"
#include "psAbort.h"
#include "psError.h"
#include "psString.h"
#include "psTrace.h"


// set the pointer to NULL if we are actually freeing the memory
#define PSDB_NULL_FREE(ptr) \
if (psMemGetRefCounter(ptr) == 1) { \
    psFree(ptr); \
    ptr = NULL; \
} else { \
    psFree(ptr); \
}

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

typedef struct
{
    MYSQL_BIND      *bind;
    psU32           n;
}
psDBMysqlRow;

static pthread_mutex_t lookupTableMutex = PTHREAD_MUTEX_INITIALIZER;
static psHash   *pTypeToSQLlookupTable = NULL;
static psHash   *sqlToPTypeLookupTable = NULL;
static psHash   *mysqlToSqlLookupTable = NULL;
static psHash   *pTypeToMysqlLookupTable = NULL;

// database utility functions
static inline bool psDBPackMySQLRow(psDBMysqlRow *mysqlRow, const psMetadata *values);
static psDBMysqlRow *psDBMysqlRowAlloc(psU32 paramCount);
static void psDBMysqlRowFree(psDBMysqlRow *row);
static void psDBMysqlRowRecycle(psDBMysqlRow *mysqlRow);

// 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    *psDBGenerateSetSQL(const psMetadata *set
                                  );
static char *psDBGenerateConditionalSQL(const psMetadataItem *item, const char *tableName);

// 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  *psDBPTypeToSQLTableSetup(void);
static psHash  *psDBPTypeToSQLTableGet(void);
static void     psDBPTypeToSQLTableCleanup(void);

static psHash  *psDBSQLToPTypeTableSetup(void);
static psHash  *psDBSQLToPTypeTableGet(void);
static void     psDBSQLToPTypeTableCleanup(void);

static psHash  *psDBMySQLToSQLTableSetup(void);
static psHash  *psDBMySQLToSQLTableGet(void);
static void     psDBMySQLToSQLTableCleanup(void);

static psHash  *psDBPTypeToMySQLTableSetup(void);
static psHash  *psDBPTypeToMySQLTableGet(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.");
    }

    // without this call we won't pick up anything from my.cnf
    mysql_options(mysql, MYSQL_READ_DEFAULT_GROUP, "client");

    // 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,
                _("Failed to connect to database.  Error: %s"),mysql_error(mysql));

        mysql_close(mysql);

        return NULL;
    }

    dbh = psAlloc(sizeof(psDB));

    dbh->mysql = mysql;

    // explicit transactions default to false
    if (!psDBExplicitTrans(dbh, false)) {
        psError(PS_ERR_UNKNOWN, true,
                "failed to set transaction type. Error: %s", mysql_error(mysql));

        mysql_close(mysql);
        psFree(dbh);

        return NULL;
    }

    if (psMemGetThreadSafety()) {
        pthread_mutex_lock(&lookupTableMutex);

        // psDBPTypeToSQLTableSetu must be run first
        psDBPTypeToSQLTableSetup();
        psDBSQLToPTypeTableSetup();
        psDBMySQLToSQLTableSetup();
        psDBPTypeToMySQLTableSetup();

        pthread_mutex_unlock(&lookupTableMutex);
    } else {
        // psDBPTypeToSQLTableSetu must be run first
        psDBPTypeToSQLTableSetup();
        psDBSQLToPTypeTableSetup();
        psDBMySQLToSQLTableSetup();
        psDBPTypeToMySQLTableSetup();
    }

    psTrace("psLib.db", PS_LOG_INFO, "connected to database %s", dbname);

    return dbh;
}

void psDBCleanup(psDB *dbh)
{
    // quietly handle NULLs
    if (!dbh) {
        return;
    }

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

    // ASC WARNING NOTE: the psDBSQLToPTypeTableCleanup cleanup routine
    // needs to be called first because it refers to
    // psDBPTypeToSQLTableGet ...
    //
    // The *cleanup functions should be thread safe as they just call psFree()
    // but we don't want to be cleaning up & setting up at the same time
    if (psMemGetThreadSafety()) {
        pthread_mutex_lock(&lookupTableMutex);

        psDBSQLToPTypeTableCleanup();
        psDBMySQLToSQLTableCleanup();
        psDBPTypeToSQLTableCleanup();
        psDBPTypeToMySQLTableCleanup();

        pthread_mutex_unlock(&lookupTableMutex);
    } else {
        psDBSQLToPTypeTableCleanup();
        psDBMySQLToSQLTableCleanup();
        psDBPTypeToSQLTableCleanup();
        psDBPTypeToMySQLTableCleanup();
    }

    psTrace("psLib.db", PS_LOG_INFO, "disconnected");
}

bool psDBCreate(psDB *dbh,
                const char *dbname)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(dbname, false);

    char *query = NULL;
    psStringAppend(&query, "CREATE DATABASE %s", dbname);

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

    psFree(query);

    psTrace("psLib.db", PS_LOG_INFO, "created a database named %s", dbname);

    return status;
}

bool psDBChange(psDB *dbh,
                const char *dbname)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(dbname, false);

    // Attempt to select new database
    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;
    }

    psTrace("psLib.db", PS_LOG_INFO, "changed to using database %s", dbname);

    return true;
}

bool psDBDrop(psDB *dbh,
              const char *dbname)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(dbname, false);

    char            *query = NULL;
    psStringAppend(&query, "DROP DATABASE %s", dbname);

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

    psFree(query);

    psTrace("psLib.db", PS_LOG_INFO, "dropped database %s", dbname);

    return status;
}

bool psDBCreateTable(psDB *dbh,
                     const char *tableName,
                     const psMetadata *md)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(tableName, false);
    PS_ASSERT_PTR_NON_NULL(md, false);

    // Generate SQL query string
    char *query = psDBGenerateCreateTableSQL(tableName, md);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, _("Query generation failed."));
        return false;
    }

    // Run SQL query to create table
    bool status = p_psDBRunQuery(dbh, query);
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, _("Failed to create table."));
    }

    psFree(query);

    psTrace("psLib.db", PS_LOG_INFO, "created table %s", tableName);

    return status;
}

bool psDBDropTable(psDB *dbh,
                   const char *tableName)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(tableName, false);

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

    // Execute query
    bool status = p_psDBRunQuery(dbh, query);
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, _("Failed to drop table."));
    }

    psFree(query);

    psTrace("psLib.db", PS_LOG_INFO, "dropped table %s", tableName);

    return status;
}

psArray *psDBSelectColumn(psDB *dbh,
                          const char *tableName,
                          const char *col,
                          unsigned long long limit)
{
    PS_ASSERT_PTR_NON_NULL(dbh, NULL);
    PS_ASSERT_PTR_NON_NULL(tableName, NULL);
    PS_ASSERT_PTR_NON_NULL(col, NULL);

    MYSQL_RES       *result;            // complete db result set
    MYSQL_ROW       row;                // single row of db result set
    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
    char *query = psDBGenerateSelectRowSQL(tableName, col, NULL, limit);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, _("Query generation failed."));
        return NULL;
    }

    // Execut SQL query string
    if (!p_psDBRunQuery(dbh, query)) {
        psError(PS_ERR_UNKNOWN, false, _("Failed to select column."));
        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, _("Query returned no data.  Error: %s"),
                    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)
{
    PS_ASSERT_PTR_NON_NULL(dbh, NULL);
    PS_ASSERT_PTR_NON_NULL(tableName, NULL);
    PS_ASSERT_PTR_NON_NULL(col, NULL);

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

    psVector *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)
{
    PS_ASSERT_PTR_NON_NULL(dbh, NULL);
    PS_ASSERT_PTR_NON_NULL(tableName, NULL);

    // Create select row query
    char *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);

    // return the result
    return p_psDBFetchResult(dbh);
}

bool psDBInsertOneRow(psDB *dbh,
                      const char *tableName,
                      const psMetadata *row)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(tableName, false);
    PS_ASSERT_PTR_NON_NULL(row, false);

    // Create array to store single row
    psArray *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, _("Failed to insert row."));
        psFree(rowSet);
        return false;
    }

    psFree(rowSet);

    return true;
}

bool psDBInsertRows(psDB *dbh,
                    const char *tableName,
                    const psArray *rowSet)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(tableName, false);
    PS_ASSERT_PTR_NON_NULL(rowSet, false);

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

    // Generate SQL query string
    char *query = psDBGenerateInsertRowSQL(tableName, row);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, _("Query generation failed."));
        return false;
    }

    if (p_psDBRunQueryPrepared(dbh, rowSet, query) < 0) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "insert failed");
        psFree(query);
        return false;
    }

    psFree(query);

    return true;
}

psArray *psDBDumpRows(psDB *dbh,
                      const char *tableName)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(tableName, false);

    return psDBSelectRows(dbh, tableName, NULL, 0);
}

psMetadata *psDBDumpCols(psDB *dbh,
                         const char *tableName)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(tableName, false);

    MYSQL_RES       *result;
    MYSQL_FIELD     *field;
    unsigned int    fieldCount;
    psU32           pType;
    unsigned int    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);

    // this metadata is returned
    psMetadata *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 (!pType) {
            psError(PS_ERR_UNKNOWN, false, "Failed to lookup type.");
            psFree(table);
            mysql_free_result(result);
            return NULL;
        }
        //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);
        }
    }

    // Clean up mysql memory
    mysql_free_result(result);

    return table;
}

long psDBUpdateRows(psDB *dbh,
                    const char *tableName,
                    const psMetadata *where,
                    psMetadata *values)
{
    PS_ASSERT_PTR_NON_NULL(dbh, -1);
    PS_ASSERT_PTR_NON_NULL(tableName, -1);
    PS_ASSERT_PTR_NON_NULL(values, -1);

    // Generate SQL query to update row
    char *query = psDBGenerateUpdateRowSQL(tableName, where, values);
    if (!query) {
        psError(PS_ERR_UNEXPECTED_NULL, false, _("Query generation failed."));
        return -1;
    }

    psArray *rowData = psArrayAlloc(1);
    psArrayAdd(rowData, 0, values);
    long rowsAffected = p_psDBRunQueryPrepared(dbh, rowData, query);
    psFree(rowData);
    psFree(query);
    if (rowsAffected < 0) {
        psError(PS_ERR_UNEXPECTED_NULL, false, "update failed");
        return -1;
    }

    return rowsAffected;
}

long psDBDeleteRows(psDB *dbh,
                    const char *tableName,
                    const psMetadata *where,
                    unsigned long long limit)
{
    PS_ASSERT_PTR_NON_NULL(dbh, -1);
    PS_ASSERT_PTR_NON_NULL(tableName, -1);

    // Create SQL statement string
    char *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
    psS64 rows = (psS64)mysql_affected_rows(dbh->mysql);

    return rows;
}

long psDBLastInsertID(psDB *dbh)
{
    PS_ASSERT_PTR_NON_NULL(dbh, -1);

    psTrace("psLib.db", PS_LOG_INFO, "calling myql_insert_id()");

    // XXX return type is actually my_ulonglong - should the return be psU64?
    return (long)mysql_insert_id(dbh->mysql);
}

bool psDBExplicitTrans(psDB *dbh, bool mode)
{
    PS_ASSERT_PTR_NON_NULL(dbh, -1);

    psTrace("psLib.db", PS_LOG_INFO, "calling mysql_autocommit(): %u", !mode);

    // mode needs to be inverted as autocommits are the opposide of explicit
    // transactions.
    // the return value also needs to be inverted for the same reason.
    // is it safe to assume my_bool always safely casts to bool?
    return !(bool)mysql_autocommit(dbh->mysql, !mode);
}

bool psDBTransaction(psDB *dbh)
{
    PS_ASSERT_PTR_NON_NULL(dbh, -1);

    bool status = p_psDBRunQuery(dbh, "START TRANSACTION");
    if (!status) {
        psError(PS_ERR_UNKNOWN, false, "Failed to start a new transaction.");
    }

    return status;
}

bool psDBCommit(psDB *dbh)
{
    PS_ASSERT_PTR_NON_NULL(dbh, -1);

    psTrace("psLib.db", PS_LOG_INFO, "calling myql_commit()");

    // is it safe to assume my_bool always safely casts to bool?
    // mysql_commit - Zero if successful. Non-zero if an error occurred.
    return !(bool)mysql_commit(dbh->mysql);
}

bool psDBRollback(psDB *dbh)
{
    PS_ASSERT_PTR_NON_NULL(dbh, -1);

    psTrace("psLib.db", PS_LOG_INFO, "calling myql_rollback()");

    // is it safe to assume my_bool always safely casts to bool?
    // mysql_rollback - Zero if successful. Non-zero if an error occurred.
    return !(bool)mysql_rollback(dbh->mysql);
}


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

bool p_psDBRunQuery(psDB *dbh,
                    const char *format,
                    ...)
{
    PS_ASSERT_PTR_NON_NULL(dbh, false);
    PS_ASSERT_PTR_NON_NULL(format, false);

    char *query = NULL;

    // Run query
    va_list ap;
    va_start(ap, format);
    psStringAppendV(&query, format, ap);
    va_end(ap);

    psTrace("psLib.db", PS_LOG_INFO, "Executing SQL: %s", query);

    if (mysql_real_query(dbh->mysql, query, (unsigned long)strlen(query)) !=0) {
        //The following if statement was added to standardize outputs between platforms for testing purposes.
        char mysqlTemp[165];
        strncpy(mysqlTemp,
                "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'null) WHERE' at line 1", 165);
        if ( !strncmp(mysql_error(dbh->mysql), mysqlTemp, 145) ) {
            psError(PS_ERR_UNKNOWN, true, _("Failed to execute SQL query.  Error: %s"), mysqlTemp);
        } else {
            psError(PS_ERR_UNKNOWN, true, _("Failed to execute SQL query.  Error: %s"), mysql_error(dbh->mysql));
        }
        psFree(query);
        return false;
    }

    psFree(query);

    return true;
}

long p_psDBRunQueryPrepared(psDB *dbh,
                            const psArray *rowSet,
                            const char *format,
                            ...)
{
    PS_ASSERT_PTR_NON_NULL(dbh, -1);
    PS_ASSERT_PTR_NON_NULL(rowSet, -1);
    PS_ASSERT_PTR_NON_NULL(format, -1);

    char *query = NULL;

    // generate query string
    va_list ap;
    va_start(ap, format);
    psStringAppendV(&query, format, ap);
    va_end(ap);

    psTrace("psLib.db", PS_LOG_INFO, "Executing SQL: %s", query);

    // Prepare SQL statement
    MYSQL_STMT *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
    psS32 paramCount = mysql_stmt_param_count(stmt);

    // structure large enough to hold one field of data per place holder
    psDBMysqlRow *mysqlRow = psDBMysqlRowAlloc(paramCount);

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

        psDBMysqlRowRecycle(mysqlRow);

        if (!psDBPackMySQLRow(mysqlRow, rowData)) {
            psError(PS_ERR_UNKNOWN, false, "Failed to pack params into bind structure.");

            psFree(mysqlRow);
            mysql_stmt_close(stmt);

            return -1;
        }

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

            psFree(mysqlRow);
            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));

            psFree(mysqlRow);
            mysql_stmt_close(stmt);

            return -1;
        }
    } // end loop over rows

    psFree(mysqlRow);

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

    mysql_stmt_close(stmt);

    return rowsAffected;
}

psArray *p_psDBFetchResult(psDB *dbh)
{
    PS_ASSERT_PTR_NON_NULL(dbh, NULL);

    MYSQL_RES       *result;            // complete db result set
    MYSQL_ROW       row;                // single row of db result set
    MYSQL_FIELD     *field;             // field type info
    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

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

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

    psTrace("psLib.db", PS_LOG_INFO, "query returned %lld rows with %d fields",             rowCount, fieldCount);

    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);
            if (!pType) {
                psError(PS_ERR_UNKNOWN, false, "Failed to lookup type.");
                psFree(md);
                mysql_free_result(result);
                return NULL;
            }

            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 if (pType == PS_DATA_TIME) {
                psTime *time = psTimeStrptime((char *)data, "%EY-%m-%d%t%T");
                if (!time) {
                    psError(PS_ERR_UNKNOWN, true, "error parsing MySQL DateTime string");
                    return NULL;
                }
                psMetadataAddTime(md, PS_LIST_TAIL, field[i].name, 0, "", time);
                psFree(time);
                #if 0
                // this is the procedure needed for stmt
                MYSQL_TIME *myTime = (MYSQL_TIME *)&data;
                // convert MYSQL_TIME to struct tm
                struct tm tmTime;
                tmTime.tm_year  = (int)myTime->year - 1900;
                tmTime.tm_mon   = (int)myTime->month - 1;
                tmTime.tm_mday  = (int)myTime->day;
                tmTime.tm_hour  = (int)myTime->hour;
                tmTime.tm_min   = (int)myTime->minute;
                tmTime.tm_sec   = (int)myTime->second;
                // assume for the time being that we don't have negative time
                //(bool)myTime->neg
                // currently unused by mysql nor does struct tm support it
                // (unsigned long)myTime->second_part;
                psTime *time = psTimeFromTM(&tmTime);
                psMetadataAddTime(md, PS_LIST_TAIL, field[i].name, 0, "",
                                  time);
                psFree(time);
                #endif

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

static inline bool psDBPackMySQLRow(psDBMysqlRow *mysqlRow,
                                    const psMetadata *values)
{
    PS_ASSERT_PTR_NON_NULL(mysqlRow, false);
    PS_ASSERT_PTR_NON_NULL(values, false);

    mysqlType       *mType;             // type tmp variable
    static bool     isNull = true;      // used in a MYSQL_BIND to indicate NULL

    MYSQL_BIND *bind = mysqlRow->bind;

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

    // loop over fields
    psMetadataItem *item = NULL;              // field in row
    for (psU32 i = 0; (item = psListGetAndIncrement(cursor)); i++) {
        // 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

            if (item->data.str) {
                // will handle the case of "" as a NULL database value
                bind[i].buffer_length = (unsigned long)strlen(item->data.str);
                bind[i].length  = &bind[i].buffer_length;
                bind[i].buffer  = psStringCopy(item->data.V);
                bind[i].is_null = *item->data.str == '\0'
                                  ? (my_bool *)&isNull
                                  : NULL;
            } else {
                // handles the case of NULL as a NULL database value
                bind[i].buffer_length = 0;
                bind[i].length  = &bind[i].buffer_length;
                bind[i].buffer  = NULL;
                bind[i].is_null = (my_bool *)&isNull;
            }
        } else if (item->type == PS_DATA_TIME) {
            // XXX we're abusing the comment field of the metadata item here as
            // we need to have a buffer that exists outside this functions
            // scope without leaking memory
            // make a copy of the psTime so we don't modify user data when we
            // try to do the conversion
            if (item->data.str) {
                psTime *time = (psTime *)item->data.V;
                struct tm *tmTime = psTimeToTM(time);

                // XXX it wouldn't hurt to make this conversion it's own
                // function
                // myTime is used as the 'buffer' so it doesn't have to be
                // free'd
                MYSQL_TIME *myTime = psAlloc(sizeof(MYSQL_TIME));
                myTime->year    = (unsigned int)tmTime->tm_year + 1900;
                myTime->month   = (unsigned int)tmTime->tm_mon + 1;
                myTime->day     = (unsigned int)tmTime->tm_mday;
                myTime->hour    = (unsigned int)tmTime->tm_hour;
                myTime->minute  = (unsigned int)tmTime->tm_min;
                myTime->second  = (unsigned int)tmTime->tm_sec;
                // assume for the time being that we don't have negative time
                // as ISO8601 doesn't support dates prior to 0
                myTime->neg     = (my_bool)false;
                // currently unused by mysql
                myTime->second_part  = (unsigned long)time->nsec;
                psFree(tmTime);

                bind[i].buffer  = myTime;
                bind[i].buffer_length = 0;
                bind[i].length  = &bind[i].buffer_length;
                bind[i].is_null = NULL;
            } else {
                // handles the case of NULL as a NULL database value
                bind[i].buffer_length = 0;
                bind[i].length  = &bind[i].buffer_length;
                bind[i].buffer  = NULL;
                bind[i].is_null = (my_bool *)&isNull;
            }
        } 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), PS_DATA_STRING"
                    "and PS_DATA_TIME are supported.");

            psFree(cursor);

            return false;
        }
    }

    psFree(cursor);

    return true;
}

static psDBMysqlRow *psDBMysqlRowAlloc(psU32 paramCount)
{
    psDBMysqlRow *row = psAlloc(sizeof(psDBMysqlRow));
    psMemSetDeallocator(row, (psFreeFunc)psDBMysqlRowFree);

    row->bind = psAlloc(sizeof(MYSQL_BIND) * paramCount);
    memset(row->bind, 0, sizeof(MYSQL_BIND) * paramCount);
    row->n = paramCount;

    return row;
}

static void psDBMysqlRowFree(psDBMysqlRow *mysqlRow)
{
    PS_ASSERT_PTR_NON_NULL(mysqlRow, );

    MYSQL_BIND *bind = mysqlRow->bind;

    psDBMysqlRowRecycle(mysqlRow);

    psFree(bind);
}

static void psDBMysqlRowRecycle(psDBMysqlRow *mysqlRow)
{
    PS_ASSERT_PTR_NON_NULL(mysqlRow, );

    MYSQL_BIND *bind = mysqlRow->bind;

    for (psU32 i = 0; i < mysqlRow->n; i++) {
        // buffer_length is only defined for pointers to character buffers
        // primitive types will have this value set to zero.
        if (bind[i].buffer_length) {
            psFree(bind[i].buffer);
        }
    }
}


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

static char *psDBGenerateCreateTableSQL(const char *tableName,
                                        const psMetadata *table)
{
    PS_ASSERT_PTR_NON_NULL(tableName, NULL);
    PS_ASSERT_PTR_NON_NULL(table, NULL);

    char            *query = NULL;      // complete query
    psMetadataItem  *item;              // column description
    psListIterator  *cursor;            // column iterator
    char            *colType;           // type lookup table

    // 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)
                || (item->type == PS_DATA_TIME)
           ) {
            // + 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.str);
        } 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, ", ");
        }
    }


    // find database indexes
    // check for Primary Keys first as they have to be part of the same Primary
    // Key stmt
    // Reset iterator to head of list
    psListIteratorSet(cursor, 0);
    psArray *pKeys = psArrayAlloc(1);
    while ((item = psListGetAndIncrement(cursor))) {
        if (strstr(item->comment, "Primary Key")) {
            psArrayAdd(pKeys, 0, item->name);
        }
    }
    if (psArrayLength(pKeys)) {
        psStringAppend(&query, ", PRIMARY KEY(");
        for (int i = 0; i < psArrayLength(pKeys); i++) {
            if (i < 1) {
                psStringAppend(&query, "%s", (char *)pKeys->data[i]);
            } else {
                psStringAppend(&query, ", %s", (char *)pKeys->data[i]);
            }
        }
        psStringAppend(&query, ")");
    }
    psFree(pKeys);

    // Reset iterator to head of list
    psListIteratorSet(cursor, 0);
    // look for regular keys
    while ((item = psListGetAndIncrement(cursor))) {
        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)
{
    PS_ASSERT_PTR_NON_NULL(tableName, NULL);

    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, tableName);
        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)
{
    PS_ASSERT_PTR_NON_NULL(tableName, NULL);
    PS_ASSERT_PTR_NON_NULL(row, NULL);

    char            *query = NULL;
    psListIterator  *cursor;
    psMetadataItem  *item;

    // 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)
{
    PS_ASSERT_PTR_NON_NULL(tableName, NULL);
    PS_ASSERT_PTR_NON_NULL(values, NULL);

    char            *query = NULL;
    char            *setSQL;
    char            *whereSQL;

    // Create set SQL substring
    setSQL = psDBGenerateSetSQL(values);
    if (!setSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, _("SQL substring generation failed."));
        return NULL;
    }

    // Create where SQL substring
    whereSQL = psDBGenerateWhereSQL(where, tableName);
    if (!whereSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, _("SQL substring generation failed."));
        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)
{
    PS_ASSERT_PTR_NON_NULL(tableName, NULL);

    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, tableName);
    if (!whereSQL) {
        psError(PS_ERR_UNEXPECTED_NULL, false, _("SQL substring generation failed."));
        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;
}

char *psDBGenerateWhereSQL(const psMetadata *where, const char *tableName)
{
    PS_ASSERT_PTR_NON_NULL(where, NULL);

    psString search = psDBGenerateWhereConditionSQL(where, tableName);
    if (search) {
        psStringPrepend(&search, "WHERE ");
    }

    return search;
}

char *psDBGenerateWhereConditionSQL(const psMetadata *where, const char *tableName)
{
    PS_ASSERT_PTR_NON_NULL(where, NULL);

    char *query = NULL;

    // we need to know if an item is 'MULTI' so we have to march through the
    // list by key name and not by items
    psList *keys = psHashKeyList(where->hash);
    psListIterator *cursor = psListIteratorAlloc(keys, 0, false);

    // find column name and match pattern
    psString itemName = NULL;
    while ((itemName = psListGetAndIncrement(cursor))) {
        psMetadataItem *item = psMetadataLookup(where, itemName);

        if (item->type == PS_DATA_METADATA_MULTI) {
            char *logicalOp = "AND";

            // scan through the list and change the logicalOp joining
            // conditionals together to "OR" if a comment of "==" is found
            psListIterator *mCursor = psListIteratorAlloc(item->data.list, 0,
                                      false);
            psMetadataItem *mItem = NULL;
            while ((mItem = psListGetAndIncrement(mCursor))) {
                if (mItem->comment && strstr(mItem->comment, "==")) {
                    logicalOp = "OR";
                    break;
                }
            }

            // "(" + conditional [ + " AND/OR " + conditional ] + ")"
            // reset the iterator
            psListIteratorSet(mCursor, 0);
            psStringAppend(&query, "(");
            while ((mItem = psListGetAndIncrement(mCursor))) {
                char *conditional = psDBGenerateConditionalSQL(mItem, tableName);
                if (!conditional) {
                    psError(PS_ERR_UNKNOWN, false,
                            "SQL conditional generation failed.");
                    psFree(mCursor);
                    psFree(cursor);
                    psFree(keys);
                    psFree(query);
                    return NULL;
                }

                psStringAppend(&query, "%s", conditional);
                psFree(conditional);

                // + " AND/OR "
                if (!mCursor->offEnd) {
                    psStringAppend(&query, " %s ", logicalOp);
                }

            }
            psFree(mCursor);

            psStringAppend(&query, ")");
        } else {
            char *conditional = psDBGenerateConditionalSQL(item, tableName);
            if (!conditional) {
                psError(PS_ERR_UNKNOWN, false,
                        "SQL conditional generation failed.");
                psFree(cursor);
                psFree(keys);
                psFree(query);
                return NULL;
            }

            psStringAppend(&query, "%s", conditional);
            psFree(conditional);
        }

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

    //    psFree(cursor);
    psFree(keys);

    return query;
}

static char *psDBGenerateSetSQL(const psMetadata *set
                               )
{
    PS_ASSERT_PTR_NON_NULL(set
                           , NULL);

    char *query = psStringCopy("SET ");

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

    // find column name
    psMetadataItem *item = NULL;
    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;
}

static char *psDBGenerateConditionalSQL(const psMetadataItem *item, const char *tableName)
{
    PS_ASSERT_PTR_NON_NULL(item, NULL);

    char *query = NULL;

    // stringify the psMetadataItem into a SQL search specification
    // XXX we're making a big assumption here that the MySQL server handles
    // floating point in the exact same way as the client.  This is a bit scary
    // as MySQL uses native types on the server end.  In theory all IEEE754
    // math is the same but know that isn't always the case.  At least forcing
    // the comparison to be done on the server provides some consistency
    // between clients on different archs.

    // if tableName is specified prepend it to the item name
    if (tableName) {
        psStringAppend(&query, "%s.", tableName);
    }

    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, "(ABS(%s - %.8f) < %.8f)", item->name, (float)(item->data.F32), FLT_EPSILON * 10);
    } else if ((item->type == PS_DATA_F64) || (item->type == PS_TYPE_F64)) {
        psStringAppend(&query, "(ABS(%s - %.17f) < %.17f)", item->name, (double)(item->data.F64), DBL_EPSILON * 10);
    } 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 || *item->data.str == '\0') {
            psStringAppend(&query, "%s IS NULL", item->name);
        } else {
            if (item->comment && strstr(item->comment, "LIKE")) {
                // 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.str);
            } else {
                psStringAppend(&query, "%s = '%s'", item->name, item->data.str);
            }
        }
    } 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");

        return NULL;
    }

    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 = psDBMySQLToSQLTableGet();

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

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

    // 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 = psDBSQLToPTypeTableGet();
    value = psHashLookup(sqlToPSTable, sqlType);

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

        return 0;
    }

    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 = psDBPTypeToSQLTableGet();

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

    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 = psDBPTypeToMySQLTableGet();

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

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

        return NULL;
    }

    psMemIncrRefCounter(mType);

    return mType;
}

// PTypeToSQLTable

static psHash *psDBPTypeToSQLTableSetup(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");
        psDBAddToLookupTable(pTypeToSQLlookupTable, PS_DATA_TIME, "DATETIME");
    } else {
        // increment the ref count by one for every psDB
        psMemIncrRefCounter(pTypeToSQLlookupTable);
    }

    return pTypeToSQLlookupTable;
}

static psHash *psDBPTypeToSQLTableGet(void)
{
    return pTypeToSQLlookupTable;
}

static void psDBPTypeToSQLTableCleanup(void)
{
    PSDB_NULL_FREE(pTypeToSQLlookupTable);
}


// SQLToPTypeTable

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

    if (!sqlToPTypeLookupTable) {
        // invert the PSToSQL table
        psToSQLTable = psDBPTypeToSQLTableGet();
        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);
    } else {
        // increment the ref count by one for every psDB
        psMemIncrRefCounter(sqlToPTypeLookupTable);
    }

    return sqlToPTypeLookupTable;
}

static psHash *psDBSQLToPTypeTableGet(void)
{
    return sqlToPTypeLookupTable;
}

static void psDBSQLToPTypeTableCleanup(void)
{
    PSDB_NULL_FREE(sqlToPTypeLookupTable);
}


// MySQLToSQLTable

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

        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_TINY,      "TINYINT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_SHORT,     "SMALLINT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_LONG,      "INT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_INT24,     "MEDIUMINT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_LONGLONG,  "BIGINT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_DECIMAL,   "DECIMAL");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_FLOAT,     "FLOAT");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_DOUBLE,    "DOUBLE");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_TIMESTAMP, "TIMESTAMP");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_DATE,      "DATE");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_TIME,      "TIME");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_DATETIME,  "DATETIME");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_YEAR,      "YEAR");
        // strictly speaking this should CHAR but there is no equivilent ps type
        // for char as psString is already mapped to varchar
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_STRING,    "VARCHAR");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_VAR_STRING,"VARCHAR");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_BLOB,      "BLOB");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_SET,       "SET");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_ENUM,      "ENUM");
        psDBAddToLookupTable(mysqlToSqlLookupTable, MYSQL_TYPE_NULL,      "NULL-type");
    } else {
        // increment the ref count by one for every psDB
        psMemIncrRefCounter(mysqlToSqlLookupTable);
    }

    return mysqlToSqlLookupTable;
}

static psHash *psDBMySQLToSQLTableGet(void)
{
    return mysqlToSqlLookupTable;
}

static void psDBMySQLToSQLTableCleanup(void)
{
    PSDB_NULL_FREE(mysqlToSqlLookupTable);
}


// PTypeToMySQLTable

static psHash *psDBPTypeToMySQLTableSetup(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_TIME,      psDBMySQLTypeAlloc(MYSQL_TYPE_DATETIME, 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));
    } else {
        // increment the ref count by one for every psDB
        psMemIncrRefCounter(pTypeToMysqlLookupTable);
    }

    return pTypeToMysqlLookupTable;
}

static psHash *psDBPTypeToMySQLTableGet(void)
{
    return pTypeToMysqlLookupTable;
}

static void psDBPTypeToMySQLTableCleanup(void)
{
    PSDB_NULL_FREE(pTypeToMysqlLookupTable);
}

static psPtr psDBMySQLTypeAlloc(enum enum_field_types type,
                                bool isUnsigned)
{
    mysqlType *mType = psAlloc(sizeof(mysqlType));
    mType->type       = type;
    mType->isUnsigned = isUnsigned;

    return mType;
}

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

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

    psHashAdd(lookupTable, key, value);

    psFree(key);
    psFree(value);
}

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

    char *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)
{
    // length of string (log10 + 1) + \0
    // if n is 0, length is 1 char + \0
    size_t length = n ? (size_t)log10((double)n) + 1 + 1
                    : 2;
    char *string = psAlloc(length);
    snprintf(string, length, "%li", (long int)n);

    return string;
}

#endif // OMIT_PSDB
