--- sql/pgsql/parser3pgsql.C 2004/03/26 13:29:40 1.19 +++ sql/pgsql/parser3pgsql.C 2012/06/15 09:09:33 1.41 @@ -1,13 +1,12 @@ /** @file Parser PgSQL driver. - Copyright(c) 2001, 2003 ArtLebedev Group (http://www.artlebedev.com) + Copyright (c) 2001-2012 Art. Lebedev Studio (http://www.artlebedev.com) Author: Alexandr Petrosian (http://paf.design.ru) - 2001.07.30 using PgSQL 7.1.2 + 2007.10.25 using PgSQL 8.1.5 */ -static const char *RCSId="$Id: parser3pgsql.C,v 1.19 2004/03/26 13:29:40 paf Exp $"; #include "config_includes.h" @@ -16,17 +15,32 @@ static const char *RCSId="$Id: parser3pg #include #include -// OIDOID from catalog/pg_type.h +volatile const char * IDENT_PARSER3PGSQL_C="$Id: parser3pgsql.C,v 1.41 2012/06/15 09:09:33 moko Exp $" IDENT_PA_SQL_DRIVER_H; + +// from catalog/pg_type.h +#define BOOLOID 16 +#define INT8OID 20 +#define INT2OID 21 +#define INT4OID 23 #define OIDOID 26 +#define FLOAT4OID 700 +#define FLOAT8OID 701 +#define DATEOID 1082 +#define TIMEOID 1083 +#define TIMESTAMPOID 1114 +#define TIMESTAMPTZOID 1184 +#define TIMETZOID 1266 +#define NUMERICOID 1700 + // LO_BUFSIZE from interfaces\libpq\fe-lobj.c = 8192 (0x2000) // actually writing chunks of that size failed, reduced it twice #define LO_BUFSIZE 0x1000 -// from postgres_ext.h -#define InvalidOid ((Oid) 0) #include "ltdl.h" +#define MAX_COLS 500 + #define MAX_STRING 0x400 #define MAX_NUMBER 20 @@ -36,38 +50,56 @@ static const char *RCSId="$Id: parser3pg #endif #ifndef max -inline int max(int a,int b) { return a>b?a:b; } +inline int max(int a,int b){ return a>b?a:b; } inline int min(int a,int b){ return a(aconnection); - PQfinish(connection.conn); + void disconnect(void *aconnection){ + Connection& connection=*static_cast(aconnection); + PQfinish(connection.conn); connection.conn=0; } - void commit(void *aconnection) { - Connection& connection=*static_cast(aconnection); - if(PGresult *res=PQexec(connection.conn, "COMMIT")) - PQclear(res); - else - throwPQerror; - begin_transaction(connection); - } - void rollback(void *aconnection) { + void commit(void *aconnection){ Connection& connection=*static_cast(aconnection); + _transaction_commit(connection); + _transaction_begin(connection); + } - if(PGresult *res=PQexec(connection.conn, "ROLLBACK")) - PQclear(res); - else - throwPQerror; - begin_transaction(connection); + void rollback(void *aconnection){ + Connection& connection=*static_cast(aconnection); + _transaction_rollback(connection); + _transaction_begin(connection); } bool ping(void *aconnection) { Connection& connection=*static_cast(aconnection); - return PQstatus(connection.conn)==CONNECTION_OK; } - const char* quote( - void *aconnection, - const char *from, unsigned int length) { + // charset here is services.request_charset(), not connection.client_charset + // thus we can't use the sql server quoting support + const char* quote(void *aconnection, const char *str, unsigned int length) + { + const char* from; + const char* from_end=str+length; + + size_t quoted=0; + + for(from=str; from(aconnection); + char *result=(char*)connection.services->malloc_atomic(length + quoted + 1); + char *to = result; - char *result=(char*)connection.services->malloc_atomic(length*2+1); - char *to=result; - while(length--) { - switch(*from) { + for(from=str; from "''" *to++='\''; break; @@ -222,76 +273,103 @@ public: *to++='\\'; break; } - *to++=*from++; + *to++=*from; } + *to=0; return result; - } + } + void query(void *aconnection, - const char *astatement, unsigned long offset, unsigned long limit, - SQL_Driver_query_event_handlers& handlers) { -// _asm int 3; + const char *astatement, + size_t placeholders_count, Placeholder* placeholders, + unsigned long offset, unsigned long limit, + SQL_Driver_query_event_handlers& handlers + ){ Connection& connection=*static_cast(aconnection); - const char* cstrClientCharset=connection.cstrClientCharset; SQL_Driver_services& services=*connection.services; PGconn *conn=connection.conn; - // transcode from $request:charset to connect-string?client_charset - if(cstrClientCharset) { - size_t transcoded_statement_size; - services.transcode(astatement, strlen(astatement), - astatement, transcoded_statement_size, - services.request_charset(), - cstrClientCharset); + const char* client_charset=connection.client_charset; + const char* request_charset=services.request_charset(); + bool transcode_needed=client_charset && strcmp(client_charset, request_charset)!=0; + + const char** paramValues; + if(placeholders_count>0){ + int binds_size=sizeof(char)*placeholders_count; + paramValues = static_cast(services.malloc_atomic(binds_size)); + _bind_parameters(placeholders_count, placeholders, paramValues, connection, transcode_needed); + } + + size_t statement_size=0; + if(transcode_needed){ + // transcode query from $request:charset to ?ClientCharset + statement_size=strlen(astatement); + services.transcode(astatement, statement_size, + astatement, statement_size, + request_charset, + client_charset); + } + + const char *statement=_preprocess_statement(connection, astatement, statement_size, offset, limit); + // error after prepare? + + PGresult *res; + if(placeholders_count>0){ + res=PQexecParams(conn, statement, placeholders_count, NULL, paramValues, NULL, NULL, 0); + } else { + res=PQexec(conn, statement); } - - const char *statement=preprocess_statement(connection, - astatement, offset, limit); - - PGresult *res=PQexec(conn, statement); if(!res) throwPQerror; + bool failed=false; + SQL_Error sql_error; + switch(PQresultStatus(res)) { - case PGRES_EMPTY_QUERY: - PQclear_throw("no query"); - break; - case PGRES_COMMAND_OK: - // empty result: insert|delete|update|... - PQclear(res); - return; - case PGRES_TUPLES_OK: - break; - default: - PQclear_throwPQerror; - break; + case PGRES_EMPTY_QUERY: + PQclear_throw("no query"); + break; + case PGRES_COMMAND_OK: // empty result: insert|delete|update|... + PQclear(res); + if(connection.autocommit) + commit(aconnection); + return; + case PGRES_TUPLES_OK: + break; + default: + PQclear_throwPQerror; + break; } - int column_count=PQnfields(res); - if(!column_count) - PQclear_throw("result contains no columns"); - - bool failed=false; - SQL_Error sql_error; #define CHECK(afailed) \ if(afailed) { \ failed=true; \ goto cleanup; \ } - for(int i=0; iMAX_COLS) + column_count=MAX_COLS; + + unsigned int column_types[MAX_COLS]; + + for(size_t i=0; i=0) { - // seek to end - if(lo_lseek(conn, fd, 0, SEEK_END)<0) - PQclear_throwPQerror; - // get length - int size_tell=lo_tell(conn, fd); - if(size_tell<0) - PQclear_throwPQerror; - // seek to begin - if(lo_lseek(conn, fd, 0, SEEK_SET)<0) - PQclear_throwPQerror; - length=(size_t)size_tell; - if(length) { - // read - char* strm=(char*)services.malloc(length+1); - if(!lo_read_ex(conn, fd, strm, size_tell)) - PQclear_throw("lo_read can not read all bytes of object"); - strm[length]=0; - str=strm; - } else - str=0; - if(lo_close(conn, fd)<0) - PQclear_throwPQerror; - } else - PQclear_throwPQerror; - } else { - // normal column, read it normally - length=(size_t)PQgetlength(res, r, i); - if(length) { - char* strm=(char*)services.malloc(length+1); - memcpy(strm, cell, length+1); - str=strm; - } else - str=0; - } + size_t length=0; + const char* str; - if(str && length) { - // transcode to $request:charset from connect-string?client_charset - if(cstrClientCharset) - services.transcode(str, length, - str, length, - cstrClientCharset, - services.request_charset()); + switch(column_types[i]){ + case BOOLOID: + case INT8OID: + case INT2OID: + case INT4OID: + case FLOAT4OID: + case FLOAT8OID: + case DATEOID: + case TIMEOID: + case TIMESTAMPOID: + case TIMESTAMPTZOID: + case TIMETZOID: + case NUMERICOID: + length=(size_t)PQgetlength(res, r, i); + str=length ? strdup(services, cell, length) : 0; + // transcode is never required for these types + break; + case OIDOID: + { + Oid oid=cell?atoi(cell):0; + int fd=lo_open(conn, oid, INV_READ); + if(fd>=0){ + // seek to end + if(lo_lseek(conn, fd, 0, SEEK_END)<0) + PQclear_throwPQerror; + // get length + int size_tell=lo_tell(conn, fd); + if(size_tell<0) + PQclear_throwPQerror; + // seek to begin + if(lo_lseek(conn, fd, 0, SEEK_SET)<0) + PQclear_throwPQerror; + length=(size_t)size_tell; + if(length){ + // read + char* strm=(char*)services.malloc(length+1); + if(!lo_read_ex(conn, fd, strm, size_tell)) + PQclear_throw("lo_read can not read all bytes of object"); + strm[length]=0; + str=strm; + if(transcode_needed) { + // transcode cell value from ?ClientCharset to $request:charset + services.transcode(str, length, + str, length, + client_charset, + request_charset); + } + } else + str=0; + if(lo_close(conn, fd)<0) + PQclear_throwPQerror; + } else + PQclear_throwPQerror; + break; + } + default: + // normal column, read it normally + length=(size_t)PQgetlength(res, r, i); + str=length ? strdup(services, cell, length) : 0; + if(transcode_needed) { + // transcode cell value from ?ClientCharset to $request:charset + services.transcode(str, length, + str, length, + client_charset, + request_charset); + } + break; } - CHECK(handlers.add_row_cell(sql_error, str, length)); } } @@ -363,36 +460,95 @@ cleanup: PQclear(res); if(failed) services._throw(sql_error); + + if(connection.autocommit) + commit(aconnection); } -private: // private funcs +private: + void _bind_parameters( + size_t placeholders_count, + Placeholder* placeholders, + const char** paramValues, + Connection& connection, + bool transcode_needed + ){ + for(size_t i=0; itranscode(ph.name, strlen(ph.name), + ph.name, name_length, + connection.services->request_charset(), + connection.client_charset); + + if(ph.value) { + size_t value_length; + connection.services->transcode(ph.value, strlen(ph.value), + ph.value, value_length, + connection.services->request_charset(), + connection.client_charset); + } + } + int name_number=atoi(ph.name); + if(name_number <= 0 || (size_t)name_number > placeholders_count) + connection.services->_throw("bad bind parameter key"); + + paramValues[name_number-1]=ph.value; + } + } + + + void _transaction_begin(Connection& connection){ + _execute_transactions_cmd(connection, "BEGIN"); + } + + void _transaction_commit(Connection& connection){ + _execute_transactions_cmd(connection, "COMMIT"); + } + + void _transaction_rollback(Connection& connection){ + _execute_transactions_cmd(connection, "ROLLBACK"); + } + + void _execute_transactions_cmd(const Connection& connection, const char *query){ + if(!connection.without_default_transactions) // with option ?WithoutDefaultTransaction=1 user must execute BEGIN/COMMIT/ROLLBACK by himself + _execute_cmd(connection, query); + } - void begin_transaction(Connection& connection) { - if(PGresult *res=PQexec(connection.conn, "BEGIN")) - PQclear(res); + // executes a query and throw away the result. + void _execute_cmd(const Connection& connection, const char *query){ + if(PGresult *res=PQexec(connection.conn, query)) + PQclear(res); // throw away the result [don't need but must call] else throwPQerror; } - const char *preprocess_statement(Connection& connection, - const char *astatement, unsigned long offset, unsigned long limit) { + const char *_preprocess_statement( + Connection& connection, + const char *astatement, + size_t statement_size, + unsigned long offset, + unsigned long limit + ){ PGconn *conn=connection.conn; - size_t statement_size=strlen(astatement); + if(!statement_size) + statement_size=strlen(astatement); char *result=(char *)connection.services->malloc(statement_size - +MAX_NUMBER*2+15 // limit # offset # + +MAX_NUMBER*2+15 // " limit # offset #" +MAX_STRING // in case of short 'strings' +1); // offset & limit -> suffixes const char *o; - if(offset || limit) { + if(offset || limit!=SQL_NO_LIMIT){ char *cur=result; memcpy(cur, astatement, statement_size); cur+=statement_size; - if(limit) - cur+=snprintf(cur, 7+MAX_NUMBER, " limit %u", limit); + if(limit!=SQL_NO_LIMIT) + cur+=snprintf(cur, 7+MAX_NUMBER, " limit %lu", limit); if(offset) - cur+=snprintf(cur, 8+MAX_NUMBER, " offset %u", offset); + cur+=snprintf(cur, 8+MAX_NUMBER, " offset %lu", offset); o=result; } else o=astatement; @@ -402,7 +558,7 @@ private: // private funcs while(*o) { if( o[0]=='/' && - o[1]=='*' && + o[1]=='*' && o[2]=='*') { // name start const char* saved_o=o; o+=3; @@ -459,19 +615,18 @@ private: // private funcs private: // lo_read/write exchancements bool lo_read_ex(PGconn *conn, int fd, const/*paf*/ char *buf, size_t len) { - int size_read; - while(len && (size_read=lo_read(conn, fd, buf, min(LO_BUFSIZE, len)))>0) { - buf+=size_read; - len-=size_read; - } - return len==0; + return lo_rw_method (conn, fd, buf, len, lo_read); } bool lo_write_ex(PGconn *conn, int fd, const/*paf*/ char *buf, size_t len) { - int size_written; - while(len && (size_written=lo_write(conn, fd, buf, min(LO_BUFSIZE, len)))>0) { - buf+=size_written; - len-=size_written; + return lo_rw_method (conn, fd, buf, len, lo_write); + } + + bool lo_rw_method(PGconn *conn, int fd, const/*paf*/ char *buf, size_t len, int (*lo_func)(PGconn *conn, int fd, const/*paf*/ char *buf, size_t len)) { + int size_op; + while(len && (size_op=lo_func(conn, fd, buf, min(LO_BUFSIZE, len)))>0) { + buf+=size_op; + len-=size_op; } return len==0; } @@ -490,15 +645,25 @@ private: // conn client library funcs typedef char *(*t_PQerrorMessage)(const PGconn* conn); t_PQerrorMessage PQerrorMessage; typedef ConnStatusType (*t_PQstatus)(const PGconn *conn); t_PQstatus PQstatus; typedef PGresult *(*t_PQexec)(PGconn *conn, - const char *query); t_PQexec PQexec; + const char *query); t_PQexec PQexec; + typedef PGresult *(*t_PQexecParams)( + PGconn *conn, + const char *query, + int nParams, + const Oid *paramTypes, + const char * const *paramValues, + const int *paramLengths, + const int *paramFormats, + int resultFormat); t_PQexecParams PQexecParams; + typedef ExecStatusType (*t_PQresultStatus)(const PGresult *res); t_PQresultStatus PQresultStatus; typedef int (*t_PQgetlength)(const PGresult *res, - int tup_num, - int field_num); t_PQgetlength PQgetlength; + int tup_num, + int field_num); t_PQgetlength PQgetlength; typedef char* (*t_PQgetvalue)(const PGresult *res, - int tup_num, - int field_num); t_PQgetvalue PQgetvalue; - typedef int (*t_PQntuples)(const PGresult *res); t_PQntuples PQntuples; + int tup_num, + int field_num); t_PQgetvalue PQgetvalue; + typedef int (*t_PQntuples)(const PGresult *res); t_PQntuples PQntuples; typedef char *(*t_PQfname)(const PGresult *res, int field_index); t_PQfname PQfname; typedef int (*t_PQnfields)(const PGresult *res); t_PQnfields PQnfields; @@ -506,6 +671,10 @@ private: // conn client library funcs typedef Oid (*t_PQftype)(const PGresult *res, int field_num); t_PQftype PQftype; + typedef size_t (*t_PQescapeStringConn)(PGconn *conn, + char *to, const char *from, size_t length, + int *error); t_PQescapeStringConn PQescapeStringConn; + typedef int (*t_lo_open)(PGconn *conn, Oid lobjId, int mode); t_lo_open lo_open; typedef int (*t_lo_close)(PGconn *conn, int fd); t_lo_close lo_close; typedef int (*t_lo_read)(PGconn *conn, int fd, const/*paf*/ char *buf, size_t len); t_lo_read lo_read; @@ -520,11 +689,19 @@ private: // conn client library funcs private: // conn client library funcs linking const char *dlink(const char *dlopen_file_spec) { - if(lt_dlinit()) - return lt_dlerror(); - lt_dlhandle handle=lt_dlopen(dlopen_file_spec); - if(!handle) + if(lt_dlinit()){ + if(const char* result=lt_dlerror()) + return result; + return "can not prepare to dynamic loading"; + } + + lt_dlhandle handle=lt_dlopen(dlopen_file_spec); + + if(!handle){ + if(const char* result=lt_dlerror()) + return result; return "can not open the dynamic link module"; + } #define DSLINK(name, action) \ name=(t_##name)lt_dlsym(handle, #name); \ @@ -545,7 +722,9 @@ private: // conn client library funcs li DLINK(PQclear); DLINK(PQresultStatus); DLINK(PQexec); + DLINK(PQexecParams); DLINK(PQftype); + DLINK(PQescapeStringConn); DLINK(lo_open); DLINK(lo_close); DLINK(lo_read); DLINK(lo_write); DLINK(lo_lseek); DLINK(lo_creat); @@ -554,7 +733,6 @@ private: // conn client library funcs li return 0; } - }; extern "C" SQL_Driver *SQL_DRIVER_CREATE() {