init

db/keywords.c

@@ -0,0 +1,152 @@
+/*-------------------------------------------------------------------------
+ *
+ * keywords.c
+ *	  lexical token lookup for reserved words in PostgreSQL
+ *
+ * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  $PostgreSQL: pgsql/src/backend/parser/keywords.c,v 1.177 2006/10/07 21:51:02 petere Exp $
+ *
+ *-------------------------------------------------------------------------
+ */
+
+///////////////////////////////////////////////////////////////////////////
+//
+// pgAdmin note: This file is based on src/backend/parser/keywords.c and
+//               src/backend/parser/kwlookup.c from PostgreSQL, but extended
+//               to support EntepriseDB and Greenplum.
+//
+//               This file is under the BSD licence, per PostgreSQL.
+///////////////////////////////////////////////////////////////////////////
+
+#include "postgres.h"
+#include "parser/keywords.h"
+
+/*
+ * List of (keyword-name, keyword-token-value) pairs.
+ */
+#define PG_KEYWORD(a,b,c) {a,c},
+const ScanKeyword ScanKeywords[] = {
+#include <parser/kwlist.h>
+};
+const int NumScanKeywords = lengthof(ScanKeywords);
+
+/*
+ * Additional pairs here. They need to live in a separate array since
+ * the ScanKeywords array needs to be sorted!
+ *
+ * !!WARNING!!: This list must be sorted, because binary
+ *              search is used to locate entries.
+ */
+#define PG_KEYWORD2(a,b) {a,b},
+const ScanKeyword ScanKeywordsExtra[] = {
+	PG_KEYWORD2("connect", RESERVED_KEYWORD)
+	PG_KEYWORD2("convert", RESERVED_KEYWORD)
+	PG_KEYWORD2("distributed", UNRESERVED_KEYWORD)
+	PG_KEYWORD2("exec", RESERVED_KEYWORD)
+	PG_KEYWORD2("log", UNRESERVED_KEYWORD)
+	PG_KEYWORD2("long", RESERVED_KEYWORD)
+	PG_KEYWORD2("minus", RESERVED_KEYWORD)
+	PG_KEYWORD2("nocache", RESERVED_KEYWORD)
+	PG_KEYWORD2("number", RESERVED_KEYWORD)
+	PG_KEYWORD2("package", RESERVED_KEYWORD)
+	PG_KEYWORD2("pls_integer", RESERVED_KEYWORD)
+	PG_KEYWORD2("raw", RESERVED_KEYWORD)
+	PG_KEYWORD2("return", RESERVED_KEYWORD)
+	PG_KEYWORD2("smalldatetime", RESERVED_KEYWORD)
+	PG_KEYWORD2("smallfloat", RESERVED_KEYWORD)
+	PG_KEYWORD2("smallmoney", RESERVED_KEYWORD)
+	PG_KEYWORD2("sysdate", RESERVED_KEYWORD)
+	PG_KEYWORD2("systimestap", RESERVED_KEYWORD)
+	PG_KEYWORD2("tinyint", RESERVED_KEYWORD)
+	PG_KEYWORD2("tinytext", RESERVED_KEYWORD)
+	PG_KEYWORD2("varchar2", RESERVED_KEYWORD)
+};
+const int NumScanKeywordsExtra = lengthof(ScanKeywordsExtra);
+
+/*
+ * ScanKeywordLookup - see if a given word is a keyword
+ *
+ * Returns a pointer to the ScanKeyword table entry, or NULL if no match.
+ *
+ * The match is done case-insensitively.  Note that we deliberately use a
+ * dumbed-down case conversion that will only translate 'A'-'Z' into 'a'-'z',
+ * even if we are in a locale where tolower() would produce more or different
+ * translations.  This is to conform to the SQL99 spec, which says that
+ * keywords are to be matched in this way even though non-keyword identifiers
+ * receive a different case-normalization mapping.
+ */
+const ScanKeyword *
+ScanKeywordLookup(const char *text)
+{
+    int			len,
+                i;
+    char		word[NAMEDATALEN];
+    const ScanKeyword *low;
+    const ScanKeyword *high;
+
+    len = strlen(text);
+    /* We assume all keywords are shorter than NAMEDATALEN. */
+    if (len >= NAMEDATALEN)
+        return NULL;
+
+    /*
+     * Apply an ASCII-only downcasing.	We must not use tolower() since it may
+     * produce the wrong translation in some locales (eg, Turkish).
+     */
+    for (i = 0; i < len; i++)
+    {
+        char		ch = text[i];
+
+        if (ch >= 'A' && ch <= 'Z')
+            ch += 'a' - 'A';
+        word[i] = ch;
+    }
+    word[len] = '\0';
+
+    /*
+     * Now do a binary search using plain strcmp() comparison.
+     */
+    low = &ScanKeywords[0];
+    high = endof(ScanKeywords) - 1;
+    while (low <= high)
+    {
+        const ScanKeyword *middle;
+        int			difference;
+
+        middle = low + (high - low) / 2;
+        difference = strcmp(middle->name, word);
+        if (difference == 0)
+            return middle;
+        else if (difference < 0)
+            low = middle + 1;
+        else
+            high = middle - 1;
+    }
+
+	/*
+	 * If not found, also do a binary search in the list of extra
+	 * keywords.
+	 */
+    low = &ScanKeywordsExtra[0];
+    high = endof(ScanKeywordsExtra) - 1;
+    while (low <= high)
+    {
+        const ScanKeyword *middle;
+        int			difference;
+
+        middle = low + (high - low) / 2;
+        difference = strcmp(middle->name, word);
+        if (difference == 0)
+            return middle;
+        else if (difference < 0)
+            low = middle + 1;
+        else
+            high = middle - 1;
+    }
+
+    return NULL;
+}

db/module.mk

@@ -0,0 +1,20 @@
+#######################################################################
+#
+# pgAdmin III - PostgreSQL Tools
+# 
+# Copyright (C) 2002 - 2016, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+# module.mk - pgadmin/db/ Makefile fragment
+#
+#######################################################################
+
+pgadmin3_SOURCES += \
+	db/keywords.c \
+	db/pgConn.cpp \
+	db/pgSet.cpp \
+	db/pgQueryThread.cpp
+
+EXTRA_DIST += \
+        db/module.mk
+

db/pgQueryThread.cpp

@@ -0,0 +1,901 @@
+//////////////////////////////////////////////////////////////////////////
+//
+// pgAdmin III - PostgreSQL Tools
+//
+// Copyright (C) 2002 - 2016, The pgAdmin Development Team
+// This software is released under the PostgreSQL Licence
+//
+// pgQueryThread.cpp - PostgreSQL threaded query class
+//
+//////////////////////////////////////////////////////////////////////////
+
+#include "pgAdmin3.h"
+
+// wxWindows headers
+#include <wx/wx.h>
+
+// PostgreSQL headers
+#include <libpq-fe.h>
+
+// App headers
+#include "db/pgSet.h"
+#include "db/pgConn.h"
+#include "db/pgQueryThread.h"
+#include "db/pgQueryResultEvent.h"
+#include "utils/pgDefs.h"
+#include "utils/sysLogger.h"
+
+const wxEventType PGQueryResultEvent = wxNewEventType();
+
+// default notice processor for the pgQueryThread
+// we do assume that the argument passed will be always the
+// object of pgQueryThread
+void pgNoticeProcessor(void *_arg, const char *_message)
+{
+	wxString str(_message, wxConvUTF8);
+
+	wxLogNotice(wxT("%s"), str.Trim().c_str());
+	if (_arg)
+	{
+		((pgQueryThread *)_arg)->AppendMessage(str);
+	}
+}
+
+
+// support for multiple queries support
+pgQueryThread::pgQueryThread(pgConn *_conn, wxEvtHandler *_caller,
+                             PQnoticeProcessor _processor, void *_noticeHandler) :
+	wxThread(wxTHREAD_JOINABLE), m_currIndex(-1), m_conn(_conn),
+	m_cancelled(false), m_multiQueries(true), m_useCallable(false),
+	m_caller(_caller), m_processor(pgNoticeProcessor), m_noticeHandler(NULL),
+	m_eventOnCancellation(true)
+{
+	// check if we can really use the enterprisedb callable statement and
+	// required
+#ifdef __WXMSW__
+	if (PQiGetOutResult && PQiPrepareOut && PQiSendQueryPreparedOut)
+	{
+		// we do not need all of pqi stuff as90 onwards
+		if (m_conn && m_conn->conn && m_conn->GetIsEdb()
+		        && !m_conn->EdbMinimumVersion(9, 0))
+		{
+			m_useCallable = true;
+		}
+	}
+#else // __WXMSW__
+#ifdef EDB_LIBPQ
+	// use callable statement only with enterprisedb advanced servers
+	// we do not need all of pqi stuff as90 onwards
+	if (m_conn && m_conn->conn && m_conn->GetIsEdb()
+	        && !m_conn->EdbMinimumVersion(9, 0))
+	{
+		m_useCallable = true;
+	}
+#endif // EDB_LIBPQ
+#endif // __WXMSW__
+
+	if (m_conn && m_conn->conn)
+	{
+		PQsetnonblocking(m_conn->conn, 1);
+	}
+
+	if (_processor != NULL)
+	{
+		m_processor = _processor;
+		if (_noticeHandler)
+			m_noticeHandler = _noticeHandler;
+		else
+			m_noticeHandler = (void *)this;
+	}
+	else
+	{
+		m_noticeHandler = (void *)this;
+	}
+}
+
+pgQueryThread::pgQueryThread(pgConn *_conn, const wxString &_qry,
+                             int _resultToRetrieve, wxWindow *_caller, long _eventId, void *_data)
+	: wxThread(wxTHREAD_JOINABLE), m_currIndex(-1), m_conn(_conn),
+	  m_cancelled(false), m_multiQueries(false), m_useCallable(false),
+	  m_caller(NULL), m_processor(pgNoticeProcessor), m_noticeHandler(NULL),
+	  m_eventOnCancellation(true)
+{
+	if (m_conn && m_conn->conn)
+	{
+		PQsetnonblocking(m_conn->conn, 1);
+	}
+	m_queries.Add(
+	    new pgBatchQuery(_qry, (pgParamsArray *)NULL, _eventId, _data, false,
+	                     _resultToRetrieve));
+
+	wxLogInfo(wxT("queueing : %s"), _qry.c_str());
+	m_noticeHandler = (void *)this;
+
+	if (_caller)
+	{
+		m_caller = _caller->GetEventHandler();
+	}
+}
+
+void pgQueryThread::SetEventOnCancellation(bool eventOnCancelled)
+{
+	m_eventOnCancellation = eventOnCancelled;
+}
+
+void pgQueryThread::AddQuery(const wxString &_qry, pgParamsArray *_params,
+                             long _eventId, void *_data, bool _useCallable, int _resultToRetrieve)
+{
+	m_queries.Add(
+	    new pgBatchQuery(_qry, _params, _eventId, _data,
+	                     // use callable statement only if supported
+	                     m_useCallable && _useCallable, _resultToRetrieve));
+
+	wxLogInfo(wxT("queueing (%ld): %s"), GetId(), _qry.c_str());
+}
+
+
+pgQueryThread::~pgQueryThread()
+{
+	m_conn->RegisterNoticeProcessor(0, 0);
+	WX_CLEAR_ARRAY(m_queries);
+}
+
+
+wxString pgQueryThread::GetMessagesAndClear(int idx)
+{
+	wxString msg;
+
+	if (idx == -1)
+		idx = m_currIndex;
+
+	if (idx >= 0 && idx <= m_currIndex)
+	{
+		wxCriticalSectionLocker lock(m_criticalSection);
+		msg = m_queries[idx]->m_message;
+		m_queries[idx]->m_message = wxT("");
+	}
+
+	return msg;
+}
+
+
+void pgQueryThread::AppendMessage(const wxString &str)
+{
+	if (m_queries[m_currIndex]->m_message.IsEmpty())
+	{
+		wxCriticalSectionLocker lock(m_criticalSection);
+		if (str != wxT("\n"))
+			m_queries[m_currIndex]->m_message.Append(str);
+	}
+	else
+	{
+		wxCriticalSectionLocker lock(m_criticalSection);
+		m_queries[m_currIndex]->m_message.Append(wxT("\n")).Append(str);
+	}
+}
+
+
+int pgQueryThread::Execute()
+{
+	wxMutexLocker lock(m_queriesLock);
+
+	PGresult       *result           = NULL;
+	wxMBConv       &conv             = *(m_conn->conv);
+
+	wxString       &query            = m_queries[m_currIndex]->m_query;
+	int            &resultToRetrieve = m_queries[m_currIndex]->m_resToRetrieve;
+	long           &rowsInserted     = m_queries[m_currIndex]->m_rowsInserted;
+	Oid            &insertedOid      = m_queries[m_currIndex]->m_insertedOid;
+	// using the alias for the pointer here, in order to save the result back
+	// in the pgBatchQuery object
+	pgSet         *&dataSet          = m_queries[m_currIndex]->m_resultSet;
+	int            &rc               = m_queries[m_currIndex]->m_returnCode;
+	pgParamsArray  *params           = m_queries[m_currIndex]->m_params;
+	bool            useCallable      = m_queries[m_currIndex]->m_useCallable;
+	pgError        &err              = m_queries[m_currIndex]->m_err;
+
+	wxCharBuffer queryBuf = query.mb_str(conv);
+
+	if (PQstatus(m_conn->conn) != CONNECTION_OK)
+	{
+		rc = pgQueryResultEvent::PGQ_CONN_LOST;
+		err.msg_primary = _("Connection to the database server lost");
+
+		return(RaiseEvent(rc));
+	}
+
+	if (!queryBuf && !query.IsEmpty())
+	{
+		rc = pgQueryResultEvent::PGQ_STRING_INVALID;
+		m_conn->SetLastResultError(NULL, _("the query could not be converted to the required encoding."));
+		err.msg_primary = _("Query string is empty");
+
+		return(RaiseEvent(rc));
+	}
+
+	// Honour the parameters (if any)
+	if (params && params->GetCount() > 0)
+	{
+		int    pCount = params->GetCount();
+		int    ret    = 0,
+		       idx    = 0;
+
+		Oid         *pOids    = (Oid *)malloc(pCount * sizeof(Oid));
+		const char **pParams  = (const char **)malloc(pCount * sizeof(const char *));
+		int         *pLens    = (int *)malloc(pCount * sizeof(int));
+		int         *pFormats = (int *)malloc(pCount * sizeof(int));
+		// modes are used only by enterprisedb callable statement
+#if defined (__WXMSW__) || (EDB_LIBPQ)
+		int         *pModes   = (int *)malloc(pCount * sizeof(int));
+#endif
+
+		for (; idx < pCount; idx++)
+		{
+			pgParam *param = (*params)[idx];
+
+			pOids[idx] = param->m_type;
+			pParams[idx] = (const char *)param->m_val;
+			pLens[idx] = param->m_len;
+			pFormats[idx] = param->GetFormat();
+#if defined (__WXMSW__) || (EDB_LIBPQ)
+			pModes[idx] = param->m_mode;
+#endif
+		}
+
+		if (useCallable)
+		{
+#if defined (__WXMSW__) || (EDB_LIBPQ)
+			wxLogInfo(wxString::Format(
+			              _("using an enterprisedb callable statement (queryid:%ld, threadid:%ld)"),
+			              (long)m_currIndex, (long)GetId()));
+			wxString stmt = wxString::Format(wxT("pgQueryThread-%ld-%ld"), this->GetId(), m_currIndex);
+			PGresult *res = PQiPrepareOut(m_conn->conn, stmt.mb_str(wxConvUTF8),
+			                              queryBuf, pCount, pOids, pModes);
+
+			if( PQresultStatus(res) != PGRES_COMMAND_OK)
+			{
+				rc = pgQueryResultEvent::PGQ_ERROR_PREPARE_CALLABLE;
+				err.SetError(res, &conv);
+
+				PQclear(res);
+
+				goto return_with_error;
+			}
+
+			ret = PQiSendQueryPreparedOut(m_conn->conn, stmt.mb_str(wxConvUTF8),
+			                              pCount, pParams, pLens, pFormats, 1);
+
+			if (ret != 1)
+			{
+				rc = pgQueryResultEvent::PGQ_ERROR_EXECUTE_CALLABLE;
+
+				m_conn->SetLastResultError(NULL, _("Failed to run PQsendQuery in pgQueryThread"));
+				err.msg_primary = wxString(PQerrorMessage(m_conn->conn), conv);
+
+				PQclear(res);
+				res = NULL;
+
+				goto return_with_error;
+			}
+
+			PQclear(res);
+			res = NULL;
+#else
+			rc = -1;
+			wxASSERT_MSG(false,
+			             _("the program execution flow must not reach to this point in pgQueryThread"));
+
+			goto return_with_error;
+#endif
+		}
+		else
+		{
+			// assumptions: we will need the results in text format only
+			ret = PQsendQueryParams(m_conn->conn, queryBuf, pCount, pOids, pParams, pLens, pFormats, 0);
+
+			if (ret != 1)
+			{
+				rc = pgQueryResultEvent::PGQ_ERROR_SEND_QUERY;
+
+				m_conn->SetLastResultError(NULL,
+				                           _("Failed to run PQsendQueryParams in pgQueryThread"));
+
+				err.msg_primary = _("Failed to run PQsendQueryParams in pgQueryThread.\n") +
+				                  wxString(PQerrorMessage(m_conn->conn), conv);
+
+				goto return_with_error;
+			}
+		}
+		goto continue_without_error;
+
+return_with_error:
+		{
+			free(pOids);
+			free(pParams);
+			free(pLens);
+			free(pFormats);
+#if defined (__WXMSW__) || (EDB_LIBPQ)
+			free(pModes);
+#endif
+			return (RaiseEvent(rc));
+		}
+	}
+	else
+	{
+		// use the PQsendQuery api in case, we don't have any parameters to
+		// pass to the server
+		if (!PQsendQuery(m_conn->conn, queryBuf))
+		{
+			rc = pgQueryResultEvent::PGQ_ERROR_SEND_QUERY;
+
+			err.msg_primary = _("Failed to run PQsendQueryParams in pgQueryThread.\n") +
+			                  wxString(PQerrorMessage(m_conn->conn), conv);
+
+			return(RaiseEvent(rc));
+		}
+	}
+
+continue_without_error:
+	int resultsRetrieved = 0;
+	PGresult *lastResult = 0;
+	bool connExecutionCancelled = false;
+
+	while (true)
+	{
+		// This is a 'joinable' thread, it is not advisable to call 'delete'
+		// function on this.
+		// Hence - it does not make sense to use the function 'testdestroy' here.
+		// We introduced the 'CancelExecution' function for the same purpose.
+		//
+		// We will have to call the CancelExecution function of pgConn to
+		// inform the backend that the user has cancelled the execution.
+		//
+		// We will need to consume all the results before quiting from the thread.
+		// Otherwise - the connection object will become unusable, and throw
+		// error - because libpq connections expects application to consume all
+		// the result, before executing another query
+		//
+		if (m_cancelled && rc != pgQueryResultEvent::PGQ_EXECUTION_CANCELLED)
+		{
+			// We shouldn't be calling cancellation multiple time
+			if (!connExecutionCancelled)
+			{
+				m_conn->CancelExecution();
+				connExecutionCancelled = true;
+			}
+			rc = pgQueryResultEvent::PGQ_EXECUTION_CANCELLED;
+
+			err.msg_primary = _("Execution Cancelled");
+
+			if (lastResult)
+			{
+				PQclear(lastResult);
+				lastResult = NULL;
+			}
+		}
+
+		if ((rc = PQconsumeInput(m_conn->conn)) != 1)
+		{
+			if (m_cancelled)
+			{
+				rc = pgQueryResultEvent::PGQ_EXECUTION_CANCELLED;
+				err.msg_primary = _("Execution Cancelled");
+
+				if (lastResult)
+				{
+					PQclear(lastResult);
+					lastResult = NULL;
+				}
+				// There is nothing more to consume.
+				// We can quit the thread now.
+				//
+				// Raise the event - if the component asked for it on
+				// execution cancellation.
+				if (m_eventOnCancellation)
+					RaiseEvent(rc);
+
+				return rc;
+			}
+
+			if (rc == 0)
+			{
+				err.msg_primary = wxString(PQerrorMessage(m_conn->conn), conv);
+			}
+			if (PQstatus(m_conn->conn) == CONNECTION_BAD)
+			{
+				err.msg_primary = _("Connection to the database server lost");
+				rc = pgQueryResultEvent::PGQ_CONN_LOST;
+			}
+			else
+			{
+				rc = pgQueryResultEvent::PGQ_ERROR_CONSUME_INPUT;
+			}
+
+			return(RaiseEvent(rc));
+		}
+
+		if (PQisBusy(m_conn->conn))
+		{
+			Yield();
+			this->Sleep(10);
+
+			continue;
+		}
+
+		// if resultToRetrieve is given, the nth result will be returned,
+		// otherwise the last result set will be returned.
+		// all others are discarded
+		PGresult *res = PQgetResult(m_conn->conn);
+
+		if (!res)
+			break;
+
+		if((PQresultStatus(res) == PGRES_NONFATAL_ERROR) ||
+		        (PQresultStatus(res) == PGRES_FATAL_ERROR) ||
+		        (PQresultStatus(res) == PGRES_BAD_RESPONSE))
+		{
+			result = res;
+			err.SetError(res, &conv);
+
+			// Wait for the execution to be finished
+			// We need to fetch all the results, before sending the error
+			// message
+			do
+			{
+				if (PQconsumeInput(m_conn->conn) != 1)
+				{
+					if (m_cancelled)
+					{
+						rc = pgQueryResultEvent::PGQ_EXECUTION_CANCELLED;
+
+						// Release the result as the query execution has been cancelled by the
+						// user
+						if (result)
+							PQclear(result);
+
+						result = NULL;
+
+						if (m_eventOnCancellation)
+							RaiseEvent(rc);
+
+						return rc;
+					}
+					goto out_of_consume_input_loop;
+				}
+
+				if ((res = PQgetResult(m_conn->conn)) == NULL)
+				{
+					goto out_of_consume_input_loop;
+				}
+
+				// Release the temporary results
+				PQclear(res);
+				res = NULL;
+
+				if (PQisBusy(m_conn->conn))
+				{
+					Yield();
+					this->Sleep(10);
+				}
+			}
+			while (true);
+
+			break;
+		}
+
+#if defined (__WXMSW__) || (EDB_LIBPQ)
+		// there should be 2 results in the callable statement - the first is the
+		// dummy, the second contains our out params.
+		if (useCallable)
+		{
+			PQclear(res);
+			result = PQiGetOutResult(m_conn->conn);
+		}
+#endif
+		if (PQresultStatus(res) == PGRES_COPY_IN)
+		{
+			rc = PGRES_COPY_IN;
+			PQputCopyEnd(m_conn->conn, "not supported by pgadmin");
+		}
+
+		if (PQresultStatus(res) == PGRES_COPY_OUT)
+		{
+			int copyRc;
+			char *buf;
+			int copyRows = 0;
+			int lastCopyRc = 0;
+
+			rc = PGRES_COPY_OUT;
+
+			AppendMessage(_("query returned copy data:\n"));
+
+			while((copyRc = PQgetCopyData(m_conn->conn, &buf, 1)) >= 0)
+			{
+				// Ignore processing the query result, when it has already been
+				// cancelled by the user
+				if (m_cancelled)
+				{
+					if (!connExecutionCancelled)
+					{
+						m_conn->CancelExecution();
+						connExecutionCancelled = true;
+					}
+					continue;
+				}
+
+				if (buf != NULL)
+				{
+					if (copyRows < 100)
+					{
+						wxString str(buf, conv);
+						wxCriticalSectionLocker cs(m_criticalSection);
+						m_queries[m_currIndex]->m_message.Append(str);
+
+					}
+					else if (copyRows == 100)
+						AppendMessage(_("Query returned more than 100 copy rows, discarding the rest...\n"));
+
+					PQfreemem(buf);
+				}
+				if (copyRc > 0)
+					copyRows++;
+
+				if (lastCopyRc == 0 && copyRc == 0)
+				{
+					Yield();
+					this->Sleep(10);
+				}
+				if (copyRc == 0)
+				{
+					if (!PQconsumeInput(m_conn->conn))
+					{
+						// It might be the case - it is a result of the
+						// execution cancellation.
+						if (m_cancelled)
+						{
+							rc = pgQueryResultEvent::PGQ_EXECUTION_CANCELLED;
+
+							// Release the result as the query execution has been cancelled by the
+							// user
+							if (result)
+								PQclear(result);
+
+							result = NULL;
+
+							if (m_eventOnCancellation)
+								RaiseEvent(rc);
+
+							return rc;
+						}
+						if (PQstatus(m_conn->conn) == CONNECTION_BAD)
+						{
+							err.msg_primary = _("Connection to the database server lost");
+							rc = pgQueryResultEvent::PGQ_CONN_LOST;
+						}
+						else
+						{
+							rc = pgQueryResultEvent::PGQ_ERROR_CONSUME_INPUT;
+
+							err.msg_primary = wxString(PQerrorMessage(m_conn->conn), conv);
+						}
+						return(RaiseEvent(rc));
+					}
+				}
+				lastCopyRc = copyRc;
+			}
+
+			res = PQgetResult(m_conn->conn);
+
+			if (!res)
+				break;
+		}
+
+		resultsRetrieved++;
+
+		// Save the current result, as asked by the component
+		// But - only if the execution is not cancelled
+		if (!m_cancelled && resultsRetrieved == resultToRetrieve)
+		{
+			result = res;
+			insertedOid = PQoidValue(res);
+			if (insertedOid && insertedOid != (Oid) - 1)
+				AppendMessage(wxString::Format(_("query inserted one row with oid %d.\n"), insertedOid));
+			else
+				AppendMessage(wxString::Format(wxPLURAL("query result with %d row will be returned.\n", "query result with %d rows will be returned.\n",
+				                                        PQntuples(result)), PQntuples(result)));
+			continue;
+		}
+
+		if (lastResult)
+		{
+			if (!m_cancelled && PQntuples(lastResult))
+				AppendMessage(wxString::Format(wxPLURAL("query result with %d row discarded.\n", "query result with %d rows discarded.\n",
+				                                        PQntuples(lastResult)), PQntuples(lastResult)));
+			PQclear(lastResult);
+		}
+		lastResult = res;
+	}
+
+out_of_consume_input_loop:
+	if (m_cancelled)
+	{
+		rc = pgQueryResultEvent::PGQ_EXECUTION_CANCELLED;
+
+		// Release the result as the query execution has been cancelled by the
+		// user
+		if (result)
+			PQclear(result);
+
+		result = NULL;
+
+		if (m_eventOnCancellation)
+			RaiseEvent(rc);
+
+		return rc;
+	}
+
+	if (!result)
+		result = lastResult;
+
+	err.SetError(result, &conv);
+
+	AppendMessage(wxT("\n"));
+
+	rc = PQresultStatus(result);
+	if (rc == PGRES_TUPLES_OK)
+	{
+		dataSet = new pgSet(result, m_conn, conv, m_conn->needColQuoting);
+		dataSet->MoveFirst();
+	}
+	else if (rc == PGRES_COMMAND_OK)
+	{
+		char *s = PQcmdTuples(result);
+		if (*s)
+			rowsInserted = atol(s);
+	}
+	else if (rc == PGRES_FATAL_ERROR ||
+	         rc == PGRES_NONFATAL_ERROR ||
+	         rc == PGRES_BAD_RESPONSE)
+	{
+		if (result)
+		{
+			AppendMessage(wxString(PQresultErrorMessage(result), conv));
+			PQclear(result);
+			result = NULL;
+		}
+		else
+		{
+			AppendMessage(wxString(PQerrorMessage(m_conn->conn), conv));
+		}
+
+		return(RaiseEvent(rc));
+	}
+
+	insertedOid = PQoidValue(result);
+	if (insertedOid == (Oid) - 1)
+		insertedOid = 0;
+
+	return(RaiseEvent(1));
+}
+
+int pgQueryThread::RaiseEvent(int _retval)
+{
+#if !defined(PGSCLI)
+	if (m_caller)
+	{
+		pgQueryResultEvent resultEvent(GetId(), m_queries[m_currIndex], m_queries[m_currIndex]->m_eventID);
+
+		// client data
+		resultEvent.SetClientData(m_queries[m_currIndex]->m_data);
+		resultEvent.SetInt(_retval);
+
+		m_caller->AddPendingEvent(resultEvent);
+	}
+#endif
+	return _retval;
+}
+
+
+void *pgQueryThread::Entry()
+{
+	do
+	{
+		if (m_currIndex < (((int)m_queries.GetCount()) - 1))
+		{
+			// Create the PGcancel object to enable cancelling the running
+			// query
+			m_conn->SetConnCancel();
+			m_currIndex++;
+
+			m_queries[m_currIndex]->m_returnCode = -2;
+			m_queries[m_currIndex]->m_rowsInserted = -1l;
+
+			wxLogSql(wxT("Thread executing query (%d:%s:%d): %s"),
+			         m_currIndex + 1, m_conn->GetHost().c_str(), m_conn->GetPort(),
+			         m_queries[m_currIndex]->m_query.c_str());
+
+			// register the notice processor for the current query
+			m_conn->RegisterNoticeProcessor(m_processor, m_noticeHandler);
+
+			// execute the current query now
+			Execute();
+
+			// remove the notice processor now
+			m_conn->RegisterNoticeProcessor(0, 0);
+
+			// reset the PGcancel object
+			m_conn->ResetConnCancel();
+		}
+
+		if (!m_multiQueries || m_cancelled)
+			break;
+
+		wxThread::Sleep(10);
+	}
+	while (true);
+
+	return(NULL);
+}
+
+int pgQueryThread::DeleteReleasedQueries()
+{
+	int res = 0,
+	    idx = 0;
+
+	if (m_queriesLock.TryLock() == wxMUTEX_BUSY)
+		return res;
+
+	for (; idx <= m_currIndex; idx++)
+	{
+		if (m_queries[idx]->m_resultSet != NULL)
+		{
+			pgSet *set = m_queries[idx]->m_resultSet;
+			m_queries[idx]->m_resultSet = NULL;
+			delete set;
+			set = NULL;
+
+			res++;
+		}
+	}
+	m_queriesLock.Unlock();
+
+	return res;
+}
+
+
+pgError pgQueryThread::GetResultError(int idx)
+{
+	wxMutexLocker lock(m_queriesLock);
+
+	if (idx == -1)
+		idx = m_currIndex;
+
+	return m_queries[idx]->m_err;
+}
+
+
+const wxString &pgBatchQuery::GetErrorMessage()
+{
+	return m_err.msg_primary;
+}
+
+pgBatchQuery::~pgBatchQuery()
+{
+	if (m_resultSet)
+	{
+		delete m_resultSet;
+		m_resultSet = NULL;
+	}
+
+	if (m_params)
+	{
+		WX_CLEAR_ARRAY((*m_params));
+		delete m_params;
+		m_params = NULL;
+	}
+}
+
+bool pgBatchQuery::Release()
+{
+	bool res = false;
+
+	if (m_resultSet != NULL)
+	{
+		res = true;
+
+		pgSet *set = m_resultSet;
+		m_resultSet = NULL;
+
+		if (set)
+			delete set;
+		set = NULL;
+	}
+
+	if (m_params)
+	{
+		res = true;
+
+		WX_CLEAR_ARRAY((*m_params));
+		delete m_params;
+		m_params = NULL;
+	}
+
+	return res;
+}
+
+pgQueryResultEvent::pgQueryResultEvent(
+    unsigned long _thrdId, pgBatchQuery *_qry, int _id) :
+	wxCommandEvent(PGQueryResultEvent, _id), m_thrdId(_thrdId),
+	m_query(_qry) { }
+
+pgQueryResultEvent::pgQueryResultEvent(const pgQueryResultEvent &_ev)
+	: wxCommandEvent(_ev), m_thrdId(_ev.m_thrdId), m_query(_ev.m_query) { }
+
+
+pgParam::pgParam(Oid _type, void *_val, int _len, short _mode)
+	: m_type(_type), m_val(_val), m_len(_len), m_mode(_mode)
+{
+	switch(m_type)
+	{
+		case PGOID_TYPE_CHAR:
+		case PGOID_TYPE_NAME:
+		case PGOID_TYPE_TEXT:
+		case PGOID_TYPE_VARCHAR:
+		case PGOID_TYPE_CSTRING:
+			m_format = 0;
+		default:
+			m_format = 1;
+	}
+}
+
+// wxString data
+pgParam::pgParam(Oid _oid, wxString *_val, wxMBConv *_conv, short _mode)
+	: m_mode(_mode)
+{
+	if (m_mode == PG_PARAM_OUT || !_val)
+	{
+		m_len = 0;
+	}
+	else
+	{
+		m_len = _val->Len();
+	}
+	if (_val)
+	{
+		char *str = (char *)malloc(m_len + 1);
+		if (!_val->IsEmpty() && _mode != PG_PARAM_OUT)
+		{
+			strncpy(str,
+			        (const char *)_val->mb_str(
+			            *(_conv != NULL ? _conv : &wxConvLocal)), m_len);
+			str[m_len] = '\0';
+		}
+		else
+		{
+			str[0] = '\0';
+		}
+		m_val = (void *)(str);
+	}
+	else
+	{
+		m_val = NULL;
+	}
+	m_type = _oid;
+
+	// text format
+	m_format = 0;
+}
+
+
+pgParam::~pgParam()
+{
+	if (m_val)
+		free(m_val);
+	m_val = NULL;
+}
+
+
+int pgParam::GetFormat()
+{
+	return m_format;
+}

db/pgSet.cpp

@@ -0,0 +1,443 @@
+//////////////////////////////////////////////////////////////////////////
+//
+// pgAdmin III - PostgreSQL Tools
+//
+// Copyright (C) 2002 - 2016, The pgAdmin Development Team
+// This software is released under the PostgreSQL Licence
+//
+// pgSet.cpp - PostgreSQL ResultSet class
+//
+//////////////////////////////////////////////////////////////////////////
+
+#include "pgAdmin3.h"
+
+// wxWindows headers
+#include <wx/wx.h>
+
+// PostgreSQL headers
+#include <libpq-fe.h>
+
+// App headers
+#include "db/pgSet.h"
+#include "db/pgConn.h"
+#include "utils/sysLogger.h"
+#include "utils/pgDefs.h"
+
+pgSet::pgSet()
+	: conv(wxConvLibc)
+{
+	conn = 0;
+	res = 0;
+	nCols = 0;
+	nRows = 0;
+	pos = 0;
+}
+
+pgSet::pgSet(PGresult *newRes, pgConn *newConn, wxMBConv &cnv, bool needColQt)
+	: conv(cnv)
+{
+	needColQuoting = needColQt;
+
+	conn = newConn;
+	res = newRes;
+
+	// Make sure we have tuples
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		nCols = 0;
+		nRows = 0;
+		pos = 0;
+	}
+	else
+	{
+		nCols = PQnfields(res);
+		for (int x = 0; x < nCols + 1; x++)
+		{
+			colTypes.Add(wxT(""));
+			colFullTypes.Add(wxT(""));
+			colClasses.Add(0);
+		}
+
+		nRows = PQntuples(res);
+		MoveFirst();
+	}
+}
+
+
+pgSet::~pgSet()
+{
+	PQclear(res);
+}
+
+
+
+OID pgSet::ColTypeOid(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	return PQftype(res, col);
+}
+
+long pgSet::ColTypeMod(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	return PQfmod(res, col);
+}
+
+
+long pgSet::GetInsertedCount() const
+{
+	char *cnt = PQcmdTuples(res);
+	if (!*cnt)
+		return -1;
+	else
+		return atol(cnt);
+}
+
+
+pgTypClass pgSet::ColTypClass(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	if (colClasses[col] != 0)
+		return (pgTypClass)colClasses[col];
+
+	wxString typoid = ExecuteScalar(
+	                      wxT("SELECT CASE WHEN typbasetype=0 THEN oid else typbasetype END AS basetype\n")
+	                      wxT("  FROM pg_type WHERE oid=") + NumToStr(ColTypeOid(col)));
+
+	switch (StrToLong(typoid))
+	{
+		case PGOID_TYPE_BOOL:
+			colClasses[col] = PGTYPCLASS_BOOL;
+			break;
+		case PGOID_TYPE_INT8:
+		case PGOID_TYPE_INT2:
+		case PGOID_TYPE_INT4:
+		case PGOID_TYPE_OID:
+		case PGOID_TYPE_XID:
+		case PGOID_TYPE_TID:
+		case PGOID_TYPE_CID:
+		case PGOID_TYPE_FLOAT4:
+		case PGOID_TYPE_FLOAT8:
+		case PGOID_TYPE_MONEY:
+		case PGOID_TYPE_BIT:
+		case PGOID_TYPE_NUMERIC:
+			colClasses[col] = PGTYPCLASS_NUMERIC;
+			break;
+		case PGOID_TYPE_BYTEA:
+		case PGOID_TYPE_CHAR:
+		case PGOID_TYPE_NAME:
+		case PGOID_TYPE_TEXT:
+		case PGOID_TYPE_VARCHAR:
+			colClasses[col] = PGTYPCLASS_STRING;
+			break;
+		case PGOID_TYPE_TIMESTAMP:
+		case PGOID_TYPE_TIMESTAMPTZ:
+		case PGOID_TYPE_TIME:
+		case PGOID_TYPE_TIMETZ:
+		case PGOID_TYPE_INTERVAL:
+			colClasses[col] = PGTYPCLASS_DATE;
+			break;
+		default:
+			colClasses[col] = PGTYPCLASS_OTHER;
+			break;
+	}
+
+	return (pgTypClass)colClasses[col];
+}
+
+
+wxString pgSet::ColType(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	if (!colTypes[col].IsEmpty())
+		return colTypes[col];
+
+	wxString szSQL, szResult;
+	szSQL.Printf(wxT("SELECT format_type(oid,NULL) as typname FROM pg_type WHERE oid = %d"), (int)ColTypeOid(col));
+	szResult = ExecuteScalar(szSQL);
+	colTypes[col] = szResult;
+
+	return szResult;
+}
+
+wxString pgSet::ColFullType(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	if (!colFullTypes[col].IsEmpty())
+		return colFullTypes[col];
+
+	wxString szSQL, szResult;
+	szSQL.Printf(wxT("SELECT format_type(oid,%d) as typname FROM pg_type WHERE oid = %d"), (int)ColTypeMod(col), (int)ColTypeOid(col));
+	szResult = ExecuteScalar(szSQL);
+	colFullTypes[col] = szResult;
+
+	return szResult;
+}
+
+int pgSet::ColScale(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	// TODO
+	return 0;
+}
+wxString pgSet::ColName(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	return wxString(PQfname(res, col), conv);
+}
+
+
+int pgSet::ColNumber(const wxString &colname) const
+{
+	int col;
+
+	if (needColQuoting)
+	{
+		wxString quotedColName = colname;
+		quotedColName.Replace(wxT("\""), wxT("\"\""));
+		col = PQfnumber(res, (wxT("\"") + quotedColName + wxT("\"")).mb_str(conv));
+	}
+	else
+		col = PQfnumber(res, colname.mb_str(conv));
+
+	if (col < 0)
+	{
+		wxLogError(__("Column not found in pgSet: %s"), colname.c_str());
+	}
+	return col;
+}
+
+bool pgSet::HasColumn(const wxString &colname) const
+{
+	if (needColQuoting)
+	{
+		wxString quotedColName = colname;
+		quotedColName.Replace(wxT("\""), wxT("\"\""));
+		return (PQfnumber(res, (wxT("\"") + quotedColName + wxT("\"")).mb_str(conv)) < 0 ? false : true);
+	}
+	else
+		return (PQfnumber(res, colname.mb_str(conv)) < 0 ? false : true);
+
+}
+
+
+
+char *pgSet::GetCharPtr(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	return PQgetvalue(res, pos - 1, col);
+}
+
+
+char *pgSet::GetCharPtr(const wxString &col) const
+{
+	return PQgetvalue(res, pos - 1, ColNumber(col));
+}
+
+
+wxString pgSet::GetVal(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	return wxString(GetCharPtr(col), conv);
+}
+
+
+wxString pgSet::GetVal(const wxString &colname) const
+{
+	return GetVal(ColNumber(colname));
+}
+
+
+long pgSet::GetLong(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	char *c = PQgetvalue(res, pos - 1, col);
+	if (c)
+		return atol(c);
+	else
+		return 0;
+}
+
+
+long pgSet::GetLong(const wxString &col) const
+{
+	char *c = PQgetvalue(res, pos - 1, ColNumber(col));
+	if (c)
+		return atol(c);
+	else
+		return 0;
+}
+
+
+bool pgSet::GetBool(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	char *c = PQgetvalue(res, pos - 1, col);
+	if (c)
+	{
+		if (*c == 't' || *c == '1' || !strcmp(c, "on"))
+			return true;
+	}
+	return false;
+}
+
+
+bool pgSet::GetBool(const wxString &col) const
+{
+	return GetBool(ColNumber(col));
+}
+
+
+wxDateTime pgSet::GetDateTime(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	wxDateTime dt;
+	wxString str = GetVal(col);
+	/* This hasn't just been used. ( Is not infinity ) */
+	if (!str.IsEmpty())
+		dt.ParseDateTime(str);
+	return dt;
+}
+
+
+wxDateTime pgSet::GetDateTime(const wxString &col) const
+{
+	return GetDateTime(ColNumber(col));
+}
+
+
+wxDateTime pgSet::GetDate(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	wxDateTime dt;
+	wxString str = GetVal(col);
+	/* This hasn't just been used. ( Is not infinity ) */
+	if (!str.IsEmpty())
+		dt.ParseDate(str);
+	return dt;
+}
+
+
+wxDateTime pgSet::GetDate(const wxString &col) const
+{
+	return GetDate(ColNumber(col));
+}
+
+
+double pgSet::GetDouble(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	return StrToDouble(GetVal(col));
+}
+
+
+double pgSet::GetDouble(const wxString &col) const
+{
+	return GetDouble(ColNumber(col));
+}
+
+
+wxULongLong pgSet::GetLongLong(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	char *c = PQgetvalue(res, pos - 1, col);
+	if (c)
+		return atolonglong(c);
+	else
+		return 0;
+}
+
+wxULongLong pgSet::GetLongLong(const wxString &col) const
+{
+	return GetLongLong(ColNumber(col));
+}
+
+
+OID pgSet::GetOid(const int col) const
+{
+	wxASSERT(col < nCols && col >= 0);
+
+	char *c = PQgetvalue(res, pos - 1, col);
+	if (c)
+		return (OID)strtoul(c, 0, 10);
+	else
+		return 0;
+}
+
+
+OID pgSet::GetOid(const wxString &col) const
+{
+	return GetOid(ColNumber(col));
+}
+
+
+wxString pgSet::ExecuteScalar(const wxString &sql) const
+{
+	return conn->ExecuteScalar(sql);
+}
+
+
+//////////////////////////////////////////////////////////////////
+
+pgSetIterator::pgSetIterator(pgConn *conn, const wxString &qry)
+{
+	set = conn->ExecuteSet(qry);
+	first = true;
+}
+
+
+pgSetIterator::pgSetIterator(pgSet *s)
+{
+	set = s;
+	first = true;
+}
+
+
+pgSetIterator::~pgSetIterator()
+{
+	if (set)
+		delete set;
+}
+
+
+bool pgSetIterator::RowsLeft()
+{
+	if (!set)
+		return false;
+
+	if (first)
+	{
+		if (!set->NumRows())
+			return false;
+		first = false;
+	}
+	else
+		set->MoveNext();
+
+	return !set->Eof();
+}
+
+
+bool pgSetIterator::MovePrev()
+{
+	if (!set)
+		return false;
+
+	set->MovePrevious();
+	return true;
+}