PqQuery.cpp

Go to the documentation of this file.

#include <dcl/Config.h>
 
#ifdef __APPLE__
#include <libkern/OSByteOrder.h>
#endif
#include <stdlib.h>     // malloc
#include <string.h>     // memset
 
#include <libpq-fe.h>
 
#include <dcl/Object.h>
#if __DCL_HAVE_ALLOC_DEBUG
#undef __DCL_ALLOC_LEVEL
#define __DCL_ALLOC_LEVEL   __DCL_ALLOC_INTERNAL
#endif
 
#include <dcl/SQLCore.h>
 
#include "PqConnection.h"
#include "PqQuery.h"
#include "PqField.h"
#include "PqParam.h"
#include "PqTypes.h"
 
#define __TRACE_THIS    0
#if __TRACE_THIS
#define __DCL_TRACE0_N  __DCL_TRACE0
#define __DCL_TRACE1_N  __DCL_TRACE1
#define __DCL_TRACE2_N  __DCL_TRACE2
#define __DCL_TRACE3_N  __DCL_TRACE3
#define __DCL_TRACE4_N  __DCL_TRACE4
#define __DCL_TRACE5_N  __DCL_TRACE5
#define __DCL_TRACE6_N  __DCL_TRACE6
#else
#define __DCL_TRACE0_N(fmt)
#define __DCL_TRACE1_N(fmt, arg)
#define __DCL_TRACE2_N(fmt, arg1, arg2)
#define __DCL_TRACE3_N(fmt, arg1, arg2, arg3)
#define __DCL_TRACE4_N(fmt, arg1, arg2, arg3, arg4)
#define __DCL_TRACE5_N(fmt, arg1, arg2, arg3, arg4, arg5)
#define __DCL_TRACE6_N(fmt, arg1, arg2, arg3, arg4, arg5, arg6)
#endif
 
#undef __THIS_FILE__
static const char_t __THIS_FILE__[] = __T("dcl/sql/PqQuery.cpp");
 
__DCL_BEGIN_NAMESPACE
 
#define __SET_ERROR(_error) \
    connection()->setErrorStatus(_error, __THIS_FILE__, __LINE__)
#define __SET_ERROR_MSG(_message) \
    conn()->setErrorMessage(_message, __THIS_FILE__, __LINE__)
 
IMPLEMENT_CLASSINFO(PqQuery, SQL::Query)
 
PqQuery::PqQuery(PqConnection* pConnection)
    : Query(pConnection)
{
    Query::__placeholder = L'$';
    __stmtNo = 0;
    __nfields = 0;
 
    __inBINDs = NULL;
    __fields = NULL;
    __params = NULL;
 
    __res = NULL;
    __row = -1;
}
 
PqQuery::~PqQuery()
{
#ifdef __DCL_DEBUG
    if (!reset()) {
        char buf[256];
        size_t buflen = sizeof(buf) - 1;
        bool b = conn()->__getErrorMessage(buf, &buflen);
        buf[b ? buflen : 0] = '\0';
        __DCL_TRACE1(L"Warning! Query reset error! %hs\n", buf);
    }
#else
    (void)reset();
#endif
}
 
void PqQuery::__destroy()
{
    delete this;
}
 
bool PqQuery::reset()
{
    Query::__eof = true;
    Query::__affectedRows = -1;
 
    // clear fields
    if (__fields) {
        __DCL_ASSERT(Query::__fieldCount > 0);
        delete[] __fields;
        __fields = NULL;
        Query::__fieldCount = 0;
        __nfields = 0;
    }
 
    // clear binds
    if (__params) {
        __DCL_ASSERT(Query::__paramCount > 0);
        delete[] __params;
        __params = NULL;
        Query::__paramCount = 0;
    }
 
    if (__inBINDs) {
        free(__inBINDs);
        __inBINDs = NULL;
    }
 
    if (__res) {
        PQclear(__res);
        __res = NULL;
        __row = -1;
    }
 
    bool r = true;
    if (!__cursorID.isEmpty()) {
        __DCL_ASSERT(!__query.isEmpty());
        __query.clear();
 
        ByteString cmd = "CLOSE " + __cursorID;
        PGresult* res = PQexec(conn()->connHandle(), cmd);
        if (PQresultStatus(res) != PGRES_COMMAND_OK) {
            __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
            r = false;
        }
        PQclear(res);
#ifdef LIBPQ_HAS_CLOSE_PREPARED
        res = PQclosePortal(conn()->connHandle(), __cursorID);
        if (r && PQresultStatus(res) != PGRES_COMMAND_OK) {
            __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
            r = false;
        }
        PQclear(res);
#endif
        __cursorID.clear();
    }
 
    if (!__stmtID.isEmpty()) {
#ifdef LIBPQ_HAS_CLOSE_PREPARED
        PGresult* res = PQclosePrepared(conn()->connHandle(), __stmtID);
        if (r && PQresultStatus(res) != PGRES_COMMAND_OK) {
            __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
            r = false;
        }
        PQclear(res);
#endif
        __stmtID.clear();
    }
 
    return r;
}
 
bool PqQuery::__prepare(
    const char* _sql, size_t _sqllen, size_t _paramCount
)
{
    if(!reset())
        return false;
 
    // PostgreSQL 17 이전의 libpq 에는 PQclosePrepared, PQclosePortal 이 없다.
    // stmtID가 동일하면 prepared statement "stmt_0" already exists
    // 연결이 지속되는 동안 이 값을 계속 증가시킨다.
    __stmtNo = conn()->stmtNo();
    ByteString stmtID = ByteString::format("stmt_%d", __stmtNo);
    PGresult* res = PQprepare(conn()->connHandle(), stmtID, _sql, 0, NULL);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
        PQclear(res);
        return false;
    }
    PQclear(res);
 
    res = PQdescribePrepared(conn()->connHandle(), stmtID);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
        PQclear(res);
        return false;
    }
 
    __DCL_ASSERT((__params == NULL)
        && (Query::__paramCount == 0)
        && (__inBINDs == NULL)
    );
 
    int nparams = PQnparams(res);
    if (nparams) {
        __DCL_ASSERT((size_t)nparams == _paramCount);;
 
#if defined(__DCL_DEBUG) && __TRACE_THIS && 0
        for (int i = 0; i < nparams; i++) {
            Oid oid = PQparamtype(res, i);
            __DCL_TRACE3_N(L"[%2d] [%4d %-10ls]\n", i,
                oid, __dataTypeName(oid)
            );
        }
#endif
        size_t alloc = nparams * (sizeof(char*) + sizeof(int) * 3);
        __DCL_TRACE2_N(L"__inBINDs nparams[%2d] * (8+4+4+4=20) alloc[%zd]\n",
            nparams, alloc);
        __inBINDs = malloc(alloc);
        if (__inBINDs == NULL) {
            __SET_ERROR(SQL::eOutOfMemory);
            return false;
        }
        memset(__inBINDs, 0, alloc);
 
#define __BIND_VALUES   ((const char**)((char*)__inBINDs))
#define __BIND_TYPES    ((Oid*)((char*)__inBINDs \
    + (Query::__paramCount * (sizeof(char*) + sizeof(int) * 0))))
#define __BIND_LENGTHS  ((int*)((char*)__inBINDs \
    + (Query::__paramCount * (sizeof(char*) + sizeof(int) * 1))))
#define __BIND_FORMATS  ((int*)((char*)__inBINDs \
    + (Query::__paramCount * (sizeof(char*) + sizeof(int) * 2))))
 
#define __SET_BIND_TYPE(_index, _type)  __BIND_TYPES[_index] = _type
#define __SET_BIND_VALUE(_index, _value)    __BIND_VALUES[_index] = _value
#define __SET_BIND_LENGTH(_index, _length)  __BIND_LENGTHS[_index] = _length
#define __SET_BIND_FORMAT(_index, _format)  __BIND_FORMATS[_index] = _format
 
        Query::__paramCount = nparams;
        __params = new PqParam[Query::__paramCount];
        if (__params == NULL) {
            __SET_ERROR(SQL::eOutOfMemory);
            return false;
        }
 
        for (size_t i = 0; i < Query::__paramCount; i++) {
            if (!__params[i].init(this, (int)i, res))
                return false;
        }
    }
 
    __stmtID = stmtID;
    __nfields = PQnfields(res);
    if (__nfields) {
        // SELECT, EXECUTE ...
        // execute에서 CURSOR를 만든다.
        if (__CURSOR_ENABLED(conn())) {
            __query.assign(_sql, _sqllen);
        }
    }
 
    PQclear(res);
    return true;
}
 
bool PqQuery::__execute()
{
    if (__nfields && __CURSOR_ENABLED(conn())) {
        // SELECT, EXECUTE
        ByteString cursorID = ByteString::format("curs_%d", __stmtNo);
        ByteStringBuilder cmd(__query.length() + 20);
        cmd.assign("DECLARE ").append(cursorID).append(" CURSOR FOR ").append(__query);
        PGresult* res = PQexecParams(conn()->connHandle(), cmd,
            (int)Query::__paramCount, __BIND_TYPES,
            __BIND_VALUES, __BIND_LENGTHS, __BIND_FORMATS,
            __FORMAT_DEFAULT
        );
        if (PQresultStatus(res) != PGRES_COMMAND_OK) {
            __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
            PQclear(res);
            return false;
        }
        PQclear(res);
        __cursorID = cursorID;
        Query::__eof = false;
 
        res = PQdescribePortal(conn()->connHandle(), cursorID);
        if (PQresultStatus(res) != PGRES_COMMAND_OK) {
            __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
            PQclear(res);
            return false;
        }
#if defined(__DCL_DEBUG) && __TRACE_THIS && 0
        for (int i = 0; i < PQnfields(res); i++) {
            Oid oid = PQftype(res, i);
            __DCL_TRACE6_N(L"[%2d] size [%6d %7d %4d] [%4d %-16hs]\n", i,
                PQfsize(res, i), PQfmod(res, i), PQfformat(res, i),
                oid, PQfname(res, i)
            );
        }
#endif
        __DCL_ASSERT(
            (__fields == NULL)
            && (Query::__fieldCount == 0)
        );
        Query::__fieldCount = __nfields;
        __fields = new PqField[Query::__fieldCount];
        if (__fields == NULL) {
            __SET_ERROR(SQL::eOutOfMemory);
            return false;
        }
 
        for (size_t i = 0; i < Query::__fieldCount; i++) {
            if (!__fields[i].init(this, (int)i, res))
                return false;
        }
        PQclear(res);
    }
    else {
        // SELECT, EXECUTE, INSERT, UPDATE, DELETE, ...
#if defined(__DCL_DEBUG) && __TRACE_THIS && 0
        for (size_t i = 0; i < Query::__paramCount; i++) {
            __DCL_TRACE4_N(L"length [%8d] value[%p] type [%4d %ls] \n",
                __BIND_LENGTHS[i], __BIND_VALUES[i],
                __BIND_TYPES[i], 
                (__BIND_TYPES[i] ? __dataTypeName(__BIND_TYPES[i]) : L"")
            );
        }
#endif
        PGresult* res = PQexecPrepared(conn()->connHandle(), __stmtID,
            (int)Query::__paramCount,
            __BIND_VALUES, __BIND_LENGTHS, __BIND_FORMATS,
            __FORMAT_DEFAULT
        );
        switch (PQresultStatus(res)) {
            case PGRES_COMMAND_OK: {
                // INSERT, UPDATE, DELETE, MERGE, MOVE, FETCH, COPY
                char* psz = PQcmdTuples(res);
                if (psz) {
                    Query::__affectedRows = strtoll(psz, NULL, 10);
                }
                PQclear(res);
                break;
            }
            case PGRES_TUPLES_OK: {
                if (__nfields != (int)Query::__fieldCount) {
                    __DCL_ASSERT(
                        (__fields == NULL)
                        && (Query::__fieldCount == 0)
                    );
                    Query::__fieldCount = __nfields;
                    __fields = new PqField[Query::__fieldCount];
                    if (__fields == NULL) {
                        __SET_ERROR(SQL::eOutOfMemory);
                        return false;
                    }
 
                    for (size_t i = 0; i < Query::__fieldCount; i++) {
                        if (!__fields[i].init(this, (int)i, res))
                            return false;
                    }
                }
                __DCL_ASSERT(PQnfields(res) > 0);
                __DCL_ASSERT(__row == -1);
                __res = res;
                Query::__eof = false;
                break;
            }
            default: {
                __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
                PQclear(res);
                return false;
            }
        }
    }
 
    for(size_t i = 0; i < Query::__paramCount; i++) {
        __SET_BIND_TYPE(i, 0);
        __SET_BIND_VALUE(i, NULL);
        __SET_BIND_LENGTH(i, 0);
        __SET_BIND_FORMAT(i, __FORMAT_TEXT);
 
        if (!(__params[i].onAfterExecute()))
            return false;
    }
 
    return true;
}
 
bool PqQuery::__fetch()
{
    if (__res) {
        __row++;
    }
 
    if (!__cursorID.isEmpty() && (__res == NULL || PQntuples(__res) == __row)) {
        // cursor를 사용하도록 되어 있는데,
        // 처음 FETCH이거나, 이전에 FETCH한 행들의 마지막에 도달했다.
        ByteStringBuilder cmd;
        cmd.assign("FETCH ").append(conn()->cursor() == 0 ?
            "NEXT" : ByteString::valueOf(conn()->cursor()))
            .append(" IN ").append(__cursorID);
        __DCL_TRACE1_N(L"__execute [%hs]\n", cmd.data());
        PGresult* res = PQexecParams(conn()->connHandle(), cmd,
            0, NULL, NULL, NULL, NULL, __FORMAT_DEFAULT
        );
        if (PQresultStatus(res) != PGRES_TUPLES_OK) {
            __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
            PQclear(res);
            return false;
        }
 
        if (__res) {
            PQclear(__res);
        }
        __res = res;
        __row = 0;
    }
 
    bool r = true;
    if (__row == PQntuples(__res)) {
        Query::__eof = true;
        if (!__cursorID.isEmpty()) {
            // FETCH한 행이 없으므로, cursor를 닫는다.
            __DCL_ASSERT(!__query.isEmpty());
            __query.clear();
 
            ByteString cmd = "CLOSE " + __cursorID;
            PGresult* res = PQexec(conn()->connHandle(), cmd);
            if (PQresultStatus(res) != PGRES_COMMAND_OK) {
                __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
                r = false;
            }
            PQclear(res);
#ifdef LIBPQ_HAS_CLOSE_PREPARED
            res = PQclosePortal(conn()->connHandle(), __cursorID);
            if (r && PQresultStatus(res) != PGRES_COMMAND_OK) {
                __SET_ERROR_MSG(PQerrorMessage(conn()->connHandle()));
                r = false;
            }
            PQclear(res);
#endif
            __cursorID.clear();
        }
    }
 
    return r;
}
 
bool PqQuery::getValue(int _fieldIndex, const char*& _value, int& _length) const
{
    __DCL_ASSERT(__res != NULL);
    if (PQgetisnull(__res, __row, _fieldIndex))
        return false;
 
    _value = PQgetvalue(__res, __row, _fieldIndex);
    _length = PQgetlength(__res, __row, _fieldIndex);
    return true;
}
 
bool PqQuery::setValue(int _paramIndex, Oid _type, const void* _value, unsigned _length, int _format)
{
    __SET_BIND_TYPE(_paramIndex, _type);
    __SET_BIND_VALUE(_paramIndex, (const char*)_value);
    __SET_BIND_LENGTH(_paramIndex, (int)_length);
    __SET_BIND_FORMAT(_paramIndex, _format);
 
    return true;
}
 
bool PqQuery::__getField(size_t _index, SQL::Field** _fieldHandleOut)
{
    __DCL_ASSERT(Query::__fieldCount > 0);
    __DCL_ASSERT((0 <= _index) && (_index < Query::__fieldCount));
    *_fieldHandleOut = &__fields[_index];
    return true;
}
 
bool PqQuery::__getParam(size_t _index, SQL::Param** _paramHandleOut)
{
    __DCL_ASSERT(Query::__paramCount > 0);
    __DCL_ASSERT((0 <= _index) && (_index < Query::__paramCount));
    *_paramHandleOut = &__params[_index];
    return true;
}
 
__DCL_END_NAMESPACE