--- sql/pgsql/parser3pgsql.C 2003/07/24 10:09:40 1.13 +++ 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.13 2003/07/24 10:09: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,28 +50,58 @@ 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_throw(PQerrorMessage(connection.conn)) #define PQclear_throw(msg) { \ PQclear(res); \ - services._throw(msg); \ + connection.services->_throw(msg); \ } - #define PQclear_throwPQerror PQclear_throw(PQerrorMessage(conn)) + #define PQclear_throwPQerror PQclear_throw(PQerrorMessage(connection.conn)) /** connect - @param used_only_in_connect_url - format: @b user:pass@host[:port]|[local]/database + @param url + format: @b user:pass@host[:port]|[local]/database? + ClientCharset=charset& // transcode by parser + charset=value& // transcode by server with 'SET CLIENT_ENCODING=value' + datestyle=value& // 'SET DATESTYLE=value' available values are: ISO|SQL|Postgres|European|US|German [default=ISO] + autocommit=1& // each transaction is commited automatically (default) + WithoutDefaultTransaction=0 // 1 -- disable any BEGIN TRAN/COMMIT/ROLLBACK [can NOT be used with autocommit option] */ void connect( - char *used_only_in_connect_url, - SQL_Driver_services& services, - void **connection ///< output: PGconn * - ) { - char *user=used_only_in_connect_url; - char *host=lsplit(user, '@'); - char *db=lsplit(host, '/'); - char *pwd=lsplit(user, ':'); - char *port=lsplit(host, ':'); + char* url, + SQL_Driver_services& services, + void** connection_ref ///< output: Connection* + ){ + char* user=url; + char* host=rsplit(user, '@'); + char* db=lsplit(host, '/'); + char* pwd=lsplit(user, ':'); + char* port=lsplit(host, ':'); char *options=lsplit(db, '?'); - PGconn *conn=PQsetdbLogin( + char* charset=0; + char* datestyle=0; + + Connection& connection=*(Connection *)services.malloc(sizeof(Connection)); + *connection_ref=&connection; + connection.services=&services; + connection.client_charset=0; + connection.autocommit=true; + connection.without_default_transactions=false; + + connection.conn=PQsetdbLogin( (host&&strcasecmp(host, "local")==0)?NULL/* local Unix domain socket */:host, port, NULL, NULL, db, user, pwd); - if(!conn) + + if(!connection.conn) services._throw("PQsetdbLogin failed"); - if(PQstatus(conn)!=CONNECTION_OK) - throwPQerror; - char *charset=0; - char *datestyle=0; + if(PQstatus(connection.conn)!=CONNECTION_OK) + throwPQerror; - while(options) { - if(char *key=lsplit(&options, '&')) { - if(*key) { - if(char *value=lsplit(key, '=')) { - if(strcasecmp(key, "charset")==0) { + while(options){ + if(char *key=lsplit(&options, '&')){ + if(*key){ + if(char *value=lsplit(key, '=')){ + if(strcmp(key, "ClientCharset")==0){ + toupper_str(value, value, strlen(value)); + connection.client_charset=value; + } else if(strcasecmp(key, "charset")==0){ charset=value; - } else if(strcasecmp(key, "datestyle")==0) { + } else if(strcasecmp(key, "datestyle")==0){ datestyle=value; + } else if(strcasecmp(key, "autocommit")==0){ + if(connection.without_default_transactions) + services._throw("options WithoutDefaultTransaction and autocommit can't be used together"); + if(atoi(value)==0) + connection.autocommit=false; + } else if(strcmp(key, "WithoutDefaultTransaction")==0){ + if(!connection.autocommit) + services._throw("options WithoutDefaultTransaction and autocommit can't be used together"); + if(atoi(value)==1){ + connection.without_default_transactions=true; + connection.autocommit=false; + } } else services._throw("unknown connect option" /*key*/); } else @@ -126,122 +200,178 @@ public: } } - if(charset) { - // set CLIENT_ENCODING - char statement[MAX_STRING]="set CLIENT_ENCODING="; // win + if(charset){ + char statement[MAX_STRING+1]="SET CLIENT_ENCODING="; strncat(statement, charset, MAX_STRING); - - PGresult *res=PQexec(conn, statement); - if(!res) - throwPQerror; - PQclear(res); // throw out the result [don't need but must call] + + _execute_cmd(connection, statement); } - if(datestyle) { - // set DATESTYLE - char statement[MAX_STRING]="set DATESTYLE="; // ISO,SQL,Postgres,European,NonEuropean=US,German,DEFAULT=ISO - strncat(statement, charset, MAX_STRING); - - PGresult *res=PQexec(conn, statement); - if(!res) - throwPQerror; - PQclear(res); // throw out the result [don't need but must call] - } - - *(PGconn **)connection=conn; - begin_transaction(services, conn); - } - void disconnect(void *connection) { - PQfinish((PGconn *)connection); - } - void commit(SQL_Driver_services& services, void *connection) { - PGconn *conn=(PGconn *)connection; - if(PGresult *res=PQexec(conn, "COMMIT")) - PQclear(res); - else - throwPQerror; - begin_transaction(services, conn); + if(datestyle){ + char statement[MAX_STRING+1]="SET DATESTYLE="; + strncat(statement, datestyle, MAX_STRING); + + _execute_cmd(connection, statement); + } + + _transaction_begin(connection); } - void rollback(SQL_Driver_services& services, void *connection) { - PGconn *conn=(PGconn *)connection; - if(PGresult *res=PQexec(conn, "ROLLBACK")) - PQclear(res); - else - throwPQerror; - begin_transaction(services, 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); + _transaction_commit(connection); + _transaction_begin(connection); } - bool ping(SQL_Driver_services&, void *connection) { - return PQstatus((PGconn *)connection)==CONNECTION_OK; + void rollback(void *aconnection){ + Connection& connection=*static_cast(aconnection); + _transaction_rollback(connection); + _transaction_begin(connection); } - const char* quote( - SQL_Driver_services& services, void *connection, - const char *from, unsigned int length) { - char *result=(char*)services.malloc_atomic(length*2+1); - char *to=result; - while(length--) { - switch(*from) { + bool ping(void *aconnection) { + Connection& connection=*static_cast(aconnection); + return PQstatus(connection.conn)==CONNECTION_OK; + } + + // 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; + + for(from=str; from "''" - *to++='\''; result++; + *to++='\''; break; case '\\': // "\" -> "\\" - *to++='\\'; result++; + *to++='\\'; break; } - *to++=*from++; + *to++=*from; } + *to=0; return result; - } - void query( - SQL_Driver_services& services, void *connection, - const char *astatement, unsigned long offset, unsigned long limit, - SQL_Driver_query_event_handlers& handlers) { -// _asm int 3; - - PGconn *conn=(PGconn *)connection; - - const char *statement=preprocess_statement(services, conn, - astatement, offset, limit); + } - PGresult *res=PQexec(conn, statement); + void query(void *aconnection, + 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); + SQL_Driver_services& services=*connection.services; + PGconn *conn=connection.conn; + + 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); + } 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 size - 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; - size=(size_t)size_tell; - if(size) { - // read - str=(char*)services.malloc(size+1); - if(!lo_read_ex(conn, fd, str, size_tell)) - PQclear_throw("lo_read can not read all bytes of object"); - str[size]=0; - } else - str=0; - if(lo_close(conn, fd)<0) - PQclear_throwPQerror; - } else - PQclear_throwPQerror; - } else { - // normal column, read it normally - size=(size_t)PQgetlength(res, r, i); - if(size) { - str=(char*)services.malloc(size+1); - memcpy(str, cell, size+1); - } else - str=0; + 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; + 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, size)); + CHECK(handlers.add_row_cell(sql_error, str, length)); } } 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(SQL_Driver_services& services, PGconn *conn) { - if(PGresult *res=PQexec(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(SQL_Driver_services& services, PGconn *conn, - const char *astatement, unsigned long offset, unsigned long limit) { - size_t statement_size=strlen(astatement); + const char *_preprocess_statement( + Connection& connection, + const char *astatement, + size_t statement_size, + unsigned long offset, + unsigned long limit + ){ + PGconn *conn=connection.conn; - char *result=(char *)services.malloc(statement_size - +MAX_NUMBER*2+15 // limit # offset # + if(!statement_size) + statement_size=strlen(astatement); + + char *result=(char *)connection.services->malloc(statement_size + +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; @@ -336,8 +558,9 @@ private: // private funcs while(*o) { if( o[0]=='/' && - o[1]=='*' && + o[1]=='*' && o[2]=='*') { // name start + const char* saved_o=o; o+=3; while(*o) if( @@ -345,6 +568,7 @@ private: // private funcs o[1]=='*' && o[2]=='/' && o[3]=='\'') { // name end + saved_o=0; // found, marking that o+=4; Oid oid=lo_creat(conn, INV_READ|INV_WRITE); if(oid==InvalidOid) @@ -358,13 +582,13 @@ private: // private funcs if(escaped) { // write pending, skip "\" or "'" if(!lo_write_ex(conn, fd, start, o-start)) - services._throw("lo_write could not write all bytes of object (1)"); + connection.services->_throw("lo_write could not write all bytes of object (1)"); start=++o; } else o++; } if(!lo_write_ex(conn, fd, start, o-start)) - services._throw("lo_write can not write all bytes of object (2)"); + connection.services->_throw("lo_write can not write all bytes of object (2)"); if(lo_close(conn, fd)<0) throwPQerror; } else @@ -376,6 +600,10 @@ private: // private funcs break; } else o++; // /**skip**/'xxx' + if(saved_o) { + o=saved_o; + *n++=*o++; + } } else *n++=*o++; } @@ -387,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; } @@ -418,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; @@ -434,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; @@ -448,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); \ @@ -473,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); @@ -482,7 +733,6 @@ private: // conn client library funcs li return 0; } - }; extern "C" SQL_Driver *SQL_DRIVER_CREATE() {