--- sql/pgsql/parser3pgsql.C 2003/01/15 10:55:38 1.11 +++ sql/pgsql/parser3pgsql.C 2008/06/24 17:43:48 1.29 @@ -1,13 +1,13 @@ /** @file Parser PgSQL driver. - Copyright(c) 2001, 2002 ArtLebedev Group (http://www.artlebedev.com) + Copyright(c) 2001, 2003 ArtLebedev Group (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.11 2003/01/15 10:55:38 paf Exp $"; +static const char *RCSId="$Id: parser3pgsql.C,v 1.29 2008/06/24 17:43:48 misha Exp $"; #include "config_includes.h" @@ -22,7 +22,7 @@ static const char *RCSId="$Id: parser3pg // actually writing chunks of that size failed, reduced it twice #define LO_BUFSIZE 0x1000 // from postgres_ext.h -#define InvalidOid ((Oid) 0) +//#define InvalidOid ((Oid) 0) #include "ltdl.h" @@ -41,23 +41,47 @@ 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=xyz& // transcode by parser + charset=xyz& // transcode by server with 'set CLIENT_ENCODING=xyz' + datestyle=xyz& // set DATESTYLE=xyz + autocommit=1& + WithoutDefaultTransaction=1 // == autocommit=0 */ void connect( - char *used_only_in_connect_url, + char *url, SQL_Driver_services& services, - void **connection ///< output: PGconn * + void **connection_ref ///< output: Connection* ) { - char *user=used_only_in_connect_url; - char *host=lsplit(user, '@'); + 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 *cstrBackwardCompAskServerToTranscode=0; + char* datestyle=0; + + Connection& connection=*(Connection *)services.malloc(sizeof(Connection)); + *connection_ref=&connection; + connection.services=&services; + connection.cstrClientCharset=0; + connection.autocommit=true; + 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) { - charset=value; + while(options){ + if(char *key=lsplit(&options, '&')){ + if(*key){ + if(char *value=lsplit(key, '=')){ + if(strcmp(key, "ClientCharset")==0) { // transcoding with parser + toupper_str(value, value, strlen(value)); + connection.cstrClientCharset=value; + } else if(strcasecmp(key, "charset")==0){ // transcoding with server + cstrBackwardCompAskServerToTranscode=value; } else if(strcasecmp(key, "datestyle")==0) { datestyle=value; + } else if(strcasecmp(key, "autocommit")==0){ + if(atoi(value)==0) + connection.autocommit=false; + } else if(strcmp(key, "WithoutDefaultTransaction")==0){ // backward, use autocommit=0 + if(atoi(value)==1) + connection.autocommit=false; } else services._throw("unknown connect option" /*key*/); } else @@ -126,87 +172,94 @@ public: } } - if(charset) { - // set CLIENT_ENCODING - char statement[MAX_STRING]="set CLIENT_ENCODING="; // win - 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] + if(cstrBackwardCompAskServerToTranscode){ + char statement[MAX_STRING]="set CLIENT_ENCODING="; + strncat(statement, cstrBackwardCompAskServerToTranscode, MAX_STRING); + + execute_resultless(connection, statement); } - if(datestyle) { - // set DATESTYLE + if(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] + strncat(statement, datestyle, MAX_STRING); + + execute_resultless(connection, statement); } - *(PGconn **)connection=conn; - begin_transaction(services, conn); + begin_transaction(connection); } - void disconnect(void *connection) { - PQfinish((PGconn *)connection); + + void disconnect(void *aconnection) { + Connection& connection=*static_cast(aconnection); + + PQfinish(connection.conn); + connection.conn=0; } - 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); + + void commit(void *aconnection) { + execute_transaction_cmd(aconnection, "COMMIT"); } - 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 rollback(void *aconnection) { + execute_transaction_cmd(aconnection, "ROLLBACK"); } - bool ping(SQL_Driver_services&, void *connection) { - return PQstatus((PGconn *)connection)==CONNECTION_OK; + bool ping(void *aconnection) { + Connection& connection=*static_cast(aconnection); + + return PQstatus(connection.conn)==CONNECTION_OK; } - unsigned int quote( - SQL_Driver_services&, void *connection, - char *to, const char *from, unsigned int length) { - if(to) { // store mode - unsigned int result=length; - while(length--) { - switch(*from) { - case '\'': // "'" -> "''" - *to++='\''; result++; - break; - case '\\': // "\" -> "\\" - *to++='\\'; result++; - break; - } - *to++=*from++; - } - return result; - } else // estimate mode - return length*2; - } - void query( - SQL_Driver_services& services, void *connection, - const char *astatement, unsigned long offset, unsigned long limit, + const char* quote( + void *aconnection, + const char *from, unsigned int length) { + Connection& connection=*static_cast(aconnection); + + char *result=(char*)connection.services->malloc_atomic(length*2+1); + int err=0; + PQescapeStringConn (connection.conn, + result, from, length, + &err); + return result; + } + + 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) { // _asm int 3; + Connection& connection=*static_cast(aconnection); + const char* cstrClientCharset=connection.cstrClientCharset; + SQL_Driver_services& services=*connection.services; + PGconn *conn=connection.conn; + + const char** paramValues; + if(placeholders_count>0){ + //services._throw("bind variables not supported (yet)"); + int binds_size=sizeof(char) * placeholders_count; + paramValues = static_cast(services.malloc_atomic(binds_size)); + bind_parameters(placeholders_count, placeholders, paramValues, connection); + } - PGconn *conn=(PGconn *)connection; + // transcode from $request:charset to ?ClientCharset + if(cstrClientCharset) { + size_t transcoded_statement_size; + services.transcode(astatement, strlen(astatement), + astatement, transcoded_statement_size, + services.request_charset(), + cstrClientCharset); + } - const char *statement=preprocess_statement(services, conn, + const char *statement=preprocess_statement(connection, astatement, offset, limit); - PGresult *res=PQexec(conn, statement); + 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; @@ -239,10 +292,19 @@ public: for(int i=0; itranscode(ph.name, strlen(ph.name), + ph.name, name_length, + connection.services->request_charset(), + connection.cstrClientCharset); + + if(ph.value) { + connection.services->transcode(ph.value, strlen(ph.value), + ph.value, value_length, + connection.services->request_charset(), + connection.cstrClientCharset); + } + } + if( atoi(ph.name) <= 0 || atoi(ph.name) > placeholders_count) { + connection.services->_throw("bad bind parameter key"); + } + paramValues[atoi(ph.name)-1] = ph.value; + } + } + + + void execute_transaction_cmd(void *aconnection, const char *query) { + Connection& connection=*static_cast(aconnection); + + if(connection.autocommit){ + Connection& connection=*static_cast(aconnection); + execute_resultless(connection, query); + } + + begin_transaction(connection); + } + + /** + Executes a query and throws the result. + */ + void execute_resultless(const Connection& connection, const char *query) { + if(PGresult *res=PQexec(connection.conn, query)) + PQclear(res); // throw out the result [don't need but must call] else throwPQerror; } - const char *preprocess_statement(SQL_Driver_services& services, PGconn *conn, + void begin_transaction(Connection& connection) { + if(connection.autocommit){ + execute_resultless(connection, "BEGIN"); + } + } + + const char *preprocess_statement(Connection& connection, const char *astatement, unsigned long offset, unsigned long limit) { + PGconn *conn=connection.conn; + size_t statement_size=strlen(astatement); - char *result=(char *)services.malloc(statement_size + char *result=(char *)connection.services->malloc(statement_size +MAX_NUMBER*2+15 // limit # offset # +MAX_STRING // in case of short 'strings' +1); @@ -338,6 +466,7 @@ private: // private funcs o[0]=='/' && o[1]=='*' && o[2]=='*') { // name start + const char* saved_o=o; o+=3; while(*o) if( @@ -345,6 +474,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 +488,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 +506,10 @@ private: // private funcs break; } else o++; // /**skip**/'xxx' + if(saved_o) { + o=saved_o; + *n++=*o++; + } } else *n++=*o++; } @@ -387,19 +521,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,14 +551,24 @@ 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; + 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; @@ -434,6 +577,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; @@ -450,8 +597,8 @@ private: // conn client library funcs li const char *dlink(const char *dlopen_file_spec) { if(lt_dlinit()) return lt_dlerror(); - lt_dlhandle handle=lt_dlopen(dlopen_file_spec); - if(!handle) + lt_dlhandle handle=lt_dlopen(dlopen_file_spec); + if(!handle) return "can not open the dynamic link module"; #define DSLINK(name, action) \ @@ -473,7 +620,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 +631,6 @@ private: // conn client library funcs li return 0; } - }; extern "C" SQL_Driver *SQL_DRIVER_CREATE() {