* psql - the openGauss interactive terminal
*
* Copyright (c) 2000-2012, PostgreSQL Global Development Group
*
* src/bin/psql/common.c
*/
#include "settings.h"
#include "postgres_fe.h"
#include "common.h"
#include "libpq/pqexpbuffer.h"
#include <ctype.h>
#include <signal.h>
#ifndef WIN32
#include <unistd.h>
#include <sys/mman.h>
#else
#include <io.h>
#include <win32.h>
#endif
#include "portability/instr_time.h"
#include "pqsignal.h"
#include "command.h"
#include "copy.h"
#include "mbprint.h"
#include <sys/wait.h>
#include <poll.h>
#ifndef WIN32
#include "libpq/libpq-int.h"
#endif
#ifdef ENABLE_UT
#define static
#endif
bool canAddHist = true;
char* g_sqlMode = NULL;
static bool ExecQueryUsingCursor(const char* query, double* elapsed_msec);
static bool command_no_begin(const char* query);
static bool is_select_command(const char* query);
static void RecordGucStmt(PGresult* results, const char* query);
static void set_proc_title();
static bool is_explain_command(const char* query);
static int file_lock(int fd, unsigned int operation);
static bool CheckPoolerConnectionStatus();
static void set_searchpath_for_tmptbl(PGconn* conn);
static bool AcceptResult(const PGresult* result, bool print_error = true);
bool GetPrintResult(PGresult** results, bool is_explain, bool is_print, const char* query, bool print_error = true);
extern void resetCheckAfterReconn();
* When a child gets a result successfully, it try to print the result to "pset.queryFout".
* All the results should be printed by sequence, so a traffic light is needed.
* We used a file lock(of pset.queryFout) before, but in some unknown cases, the stdout/stderr
* may be locked by other processes. Even though the owner has exited, the lock is not released
* correctly, which blocks all the child processes.
* So here we use a mutex instead.
*/
struct parallelMutex_t {
pthread_mutex_t mut;
pthread_mutexattr_t mutAttr;
};
static int CreateMutexForParallel();
static int LockMutexForParallel();
static int UnlockMutexForParallel();
static int DestroyMutexForParallel();
#define IsInteractiveMode() ((stdout == pset.queryFout) || (stderr == pset.queryFout))
* "Safe" wrapper around strdup()
*/
char* pg_strdup(const char* string)
{
char* tmp = NULL;
if (NULL == string) {
fprintf(stderr, _("%s: pg_strdup: cannot duplicate null pointer (internal error)\n"), pset.progname);
exit(EXIT_FAILURE);
}
tmp = strdup(string);
if (NULL == tmp) {
psql_error("out of memory\n");
exit(EXIT_FAILURE);
}
return tmp;
}
void* pg_malloc(size_t size)
{
void* tmp = NULL;
if (size == 0)
size = 1;
tmp = malloc(size);
if (NULL == tmp) {
psql_error("out of memory\n");
exit(EXIT_FAILURE);
}
return tmp;
}
void* pg_malloc_zero(size_t size)
{
void* tmp = NULL;
errno_t rc = 0;
tmp = pg_malloc(size);
rc = memset_s(tmp, size, 0, size);
check_memset_s(rc);
return tmp;
}
void* pg_calloc(size_t nmemb, size_t size)
{
void* tmp = NULL;
if (nmemb == 0 || size == 0) {
psql_error("out of memory\n");
exit(EXIT_FAILURE);
}
tmp = calloc(nmemb, size);
if (NULL == tmp) {
psql_error("out of memory\n");
exit(EXIT_FAILURE);
}
return tmp;
}
void* psql_realloc(void* ptr, size_t oldSize, size_t newSize)
{
void* tmp = NULL;
errno_t rc;
if (oldSize > newSize) {
return NULL;
}
tmp = pg_malloc(newSize);
if (tmp == NULL) {
psql_error("out of memory\n");
exit(EXIT_FAILURE);
}
rc = memcpy_s(tmp, newSize, ptr, oldSize);
securec_check_c(rc, "\0", "\0");
free(ptr);
ptr = NULL;
return tmp;
}
* setQFout
* -- handler for -o command line option and \o command
*
* Tries to open file fname (or pipe if fname starts with '|')
* and stores the file handle in pset)
* Upon failure, sets stdout and returns false.
*/
bool setQFout(const char* fname)
{
bool status = true;
char fnametmp[MAXPGPATH] = {'\0'};
if ((pset.queryFout != NULL) && pset.queryFout != stdout && pset.queryFout != stderr) {
if (pset.queryFoutPipe)
pclose(pset.queryFout);
else
fclose(pset.queryFout);
}
if (NULL != fname) {
errno_t err = EOK;
err = strcpy_s(fnametmp, sizeof(fnametmp), fname);
check_strcpy_s(err);
}
if (NULL == fname || fname[0] == '\0') {
pset.queryFout = stdout;
pset.queryFoutPipe = false;
} else if (*fname == '|') {
canonicalize_path(fnametmp);
pset.queryFout = popen(fnametmp + 1, "w");
pset.queryFoutPipe = true;
} else {
canonicalize_path(fnametmp);
pset.queryFout = fopen(fnametmp, "w");
pset.queryFoutPipe = false;
}
if (NULL == (pset.queryFout)) {
psql_error("%s: %s\n", fname, strerror(errno));
pset.queryFout = stdout;
pset.queryFoutPipe = false;
status = false;
}
#ifndef WIN32
pqsignal(SIGPIPE, pset.queryFoutPipe ? SIG_IGN : SIG_DFL);
#endif
return status;
}
* Error reporting for scripts. Errors should look like
* psql:filename:lineno: message
*
*/
void psql_error(const char* fmt, ...)
{
va_list ap;
fflush(stdout);
if (pset.queryFout != stdout)
fflush(pset.queryFout);
if (NULL != pset.inputfile)
fprintf(stderr, "%s:%s:" UINT64_FORMAT ": ", pset.progname, pset.inputfile, pset.lineno);
va_start(ap, fmt);
vfprintf(stderr, _(fmt), ap);
va_end(ap);
}
* for backend Notice messages (INFO, WARNING, etc)
*/
void NoticeProcessor(void* arg, const char* message)
{
(void)arg;
psql_error("%s", message);
}
* Code to support query cancellation
*
* Before we start a query, we enable the SIGINT signal catcher to send a
* cancel request to the backend. Note that sending the cancel directly from
* the signal handler is safe because PQcancel() is written to make it
* so. We use write() to report to stderr because it's better to use simple
* facilities in a signal handler.
*
* On win32, the signal canceling happens on a separate thread, because
* that's how SetConsoleCtrlHandler works. The PQcancel function is safe
* for this (unlike PQrequestCancel). However, a CRITICAL_SECTION is required
* to protect the PGcancel structure against being changed while the signal
* thread is using it.
*
* SIGINT is supposed to abort all long-running psql operations, not only
* database queries. In most places, this is accomplished by checking
* cancel_pressed during long-running loops. However, that won't work when
* blocked on user input (in readline() or fgets()). In those places, we
* set sigint_interrupt_enabled TRUE while blocked, instructing the signal
* catcher to longjmp through sigint_interrupt_jmp. We assume readline and
* fgets are coded to handle possible interruption. (XXX currently this does
* not work on win32, so control-C is less useful there)
*/
volatile bool sigint_interrupt_enabled = false;
sigjmp_buf sigint_interrupt_jmp;
static PGcancel* volatile cancelConn = NULL;
volatile ReadlineStatus readline_status = WAIT_INPUT;
#ifdef WIN32
static CRITICAL_SECTION cancelConnLock;
#endif
#define write_stderr(str) write(fileno(stderr), str, strlen(str))
#ifndef WIN32
static void handle_sigint(SIGNAL_ARGS)
{
int save_errno = errno;
int rc;
char errbuf[256];
ResetQueryRetryController();
if (sigint_interrupt_enabled && readline_status == WAIT_INPUT) {
sigint_interrupt_enabled = false;
siglongjmp(sigint_interrupt_jmp, 1);
}
cancel_pressed = true;
if (cancelConn != NULL) {
if (PQcancel(cancelConn, errbuf, sizeof(errbuf))) {
rc = write_stderr("Cancel request sent\n");
(void)rc;
} else {
rc = write_stderr("Could not send cancel request: ");
(void)rc;
rc = write_stderr(errbuf);
(void)rc;
}
}
if (readline_status == COMPLETE_QUERY) {
readline_status = COMPLETE_CANCELLED;
}
errno = save_errno;
}
void setup_cancel_handler(void)
{
pqsignal(SIGINT, handle_sigint);
}
void ignore_quit_signal(void)
{
pqsignal(SIGQUIT, SIG_IGN);
}
#else
static BOOL WINAPI consoleHandler(DWORD dwCtrlType)
{
char errbuf[256];
if (dwCtrlType == CTRL_C_EVENT || dwCtrlType == CTRL_BREAK_EVENT) {
* Can't longjmp here, because we are in wrong thread :-(
*/
cancel_pressed = true;
EnterCriticalSection(&cancelConnLock);
if (cancelConn != NULL) {
if (PQcancel(cancelConn, errbuf, sizeof(errbuf)))
write_stderr("Cancel request sent\n");
else {
write_stderr("Could not send cancel request: ");
write_stderr(errbuf);
}
}
LeaveCriticalSection(&cancelConnLock);
return TRUE;
} else
return FALSE;
}
void setup_cancel_handler(void)
{
InitializeCriticalSection(&cancelConnLock);
SetConsoleCtrlHandler(consoleHandler, TRUE);
}
#endif
*
* Returns whether our backend connection is still there.
*/
static bool ConnectionUp(void)
{
return PQstatus(pset.db) != CONNECTION_BAD;
}
*
* Verify that we still have a good connection to the backend, and if not,
* see if it can be restored.
*
* Returns true if either the connection was still there, or it could be
* restored successfully; false otherwise. If, however, there was no
* connection and the session is non-interactive, this will exit the program
* with a code of EXIT_BADCONN.
*/
static bool CheckConnection(void)
{
bool OK = false;
OK = ConnectionUp();
if (!OK) {
if (!pset.cur_cmd_interactive) {
psql_error("connection to server was lost\n");
exit(EXIT_BADCONN);
}
fputs(_("The connection to the server was lost. Attempting reset: "), stderr);
PQreset(pset.db);
OK = ConnectionUp();
if (!OK) {
fputs(_("Failed.\n"), stderr);
PQfinish(pset.db);
pset.db = NULL;
ResetCancelConn();
UnsyncVariables();
} else {
fputs(_("Succeeded.\n"), stderr);
resetCheckAfterReconn();
}
}
return OK;
}
* SetCancelConn
*
* Set cancelConn to point to the current database connection.
*/
void SetCancelConn(void)
{
PGcancel* oldCancelConn = NULL;
#ifdef WIN32
EnterCriticalSection(&cancelConnLock);
#endif
oldCancelConn = cancelConn;
cancelConn = NULL;
if (oldCancelConn != NULL)
PQfreeCancel(oldCancelConn);
cancelConn = PQgetCancel(pset.db);
#ifdef WIN32
LeaveCriticalSection(&cancelConnLock);
#endif
}
* ResetCancelConn
*
* Free the current cancel connection, if any, and set to NULL.
*/
void ResetCancelConn(void)
{
PGcancel* oldCancelConn = NULL;
#ifdef WIN32
EnterCriticalSection(&cancelConnLock);
#endif
oldCancelConn = cancelConn;
cancelConn = NULL;
if (oldCancelConn != NULL)
PQfreeCancel(oldCancelConn);
#ifdef WIN32
LeaveCriticalSection(&cancelConnLock);
#endif
}
static bool ISPGresultValid(const PGresult* result)
{
bool ret = false;
if (!result) {
return false;
}
switch (PQresultStatus(result)) {
case PGRES_COMMAND_OK:
case PGRES_TUPLES_OK:
case PGRES_EMPTY_QUERY:
case PGRES_COPY_IN:
case PGRES_COPY_OUT:
ret = true;
break;
case PGRES_BAD_RESPONSE:
case PGRES_NONFATAL_ERROR:
case PGRES_FATAL_ERROR:
ret = false;
break;
default:
ret = false;
psql_error("unexpected PQresultStatus: %d\n", PQresultStatus(result));
break;
}
return ret;
}
* AcceptResult
*
* Checks whether a result is valid, giving an error message if necessary;
* and ensures that the connection to the backend is still up.
*
* Returns true for valid result, false for error state.
*/
static bool AcceptResult(const PGresult* result, bool print_error)
{
bool valid = ISPGresultValid(result);
if (valid) {
return true;
}
const char* error = PQerrorMessage(pset.db);
if (strlen(error)) {
if (pset.max_retry_times > 0 &&
IsQueryNeedRetry((const char*)pset.db->last_sqlstate) && pset.retry_times < pset.max_retry_times) {
errno_t ss_rc = strcpy_s(pset.retry_sqlstate, sizeof(pset.retry_sqlstate), pset.db->last_sqlstate);
securec_check_c(ss_rc, "\0", "\0");
pset.retry_on = true;
} else if (print_error) {
psql_error("%s", error);
}
}
(void)CheckConnection();
return valid;
}
* AcceptResultWithErrMsg
*
* Checks whether a result is valid, giving an error to errMsg if necessary;
* and ensures that the connection to the backend is still up.
*
* Returns true for valid result, false for error state.
*/
static bool AcceptResultWithErrMsg(const PGresult* result,
const char** errMsg, PGconn* conn)
{
bool valid = ISPGresultValid(result);
if (valid) {
return true;
}
const char* error = PQerrorMessage(conn);
if (strlen(error)) {
if (pset.max_retry_times > 0 && PQTRANS_IDLE == PQtransactionStatus(conn) &&
IsQueryNeedRetry((const char*)conn->last_sqlstate) && pset.retry_times < pset.max_retry_times) {
errno_t ssRc = strcpy_s(pset.retry_sqlstate, sizeof(pset.retry_sqlstate), conn->last_sqlstate);
securec_check_c(ssRc, "\0", "\0");
pset.retry_on = true;
}
*errMsg = error;
}
(void)CheckConnection();
return valid;
}
* PSQLexec
*
* This is the way to send special queries (those not directly entered
* by the user). It is subject to -E but not -e.
*
* In autocommit-off mode, a new transaction block is started if start_xact
* is true; nothing special is done when start_xact is false. Typically,
* start_xact = false is used for SELECTs and explicit BEGIN/COMMIT commands.
*
* Caller is responsible for handling the ensuing processing if a COPY
* command is sent.
*
* Note: we don't bother to check PQclientEncoding; it is assumed that no
* caller uses this path to issue "SET CLIENT_ENCODING".
*/
PGresult* PSQLexec(const char* query, bool start_xact)
{
PGresult* res = NULL;
errno_t rc = 0;
if (NULL == pset.db) {
psql_error("You are currently not connected to a database.\n");
return NULL;
}
if (pset.echo_hidden != PSQL_ECHO_HIDDEN_OFF) {
printf(_("********* QUERY **********\n"
"%s\n"
"**************************\n\n"),
query);
fflush(stdout);
if (pset.logfile != NULL) {
fprintf(pset.logfile,
_("********* QUERY **********\n"
"%s\n"
"**************************\n\n"),
query);
fflush(pset.logfile);
}
if (pset.echo_hidden == PSQL_ECHO_HIDDEN_NOEXEC)
return NULL;
}
SetCancelConn();
if (start_xact && !pset.autocommit && PQtransactionStatus(pset.db) == PQTRANS_IDLE) {
res = PQexec(pset.db, "START TRANSACTION ");
if (PQresultStatus(res) != PGRES_COMMAND_OK) {
psql_error("%s", PQerrorMessage(pset.db));
PQclear(res);
ResetCancelConn();
return NULL;
}
PQclear(res);
}
res = PQexec(pset.db, query);
#ifndef WIN32
if (pset.cur_cmd_interactive) {
if ((pset.db != NULL) && (pset.db->last_query != NULL)) {
rc = memset_s(pset.db->last_query, strlen(pset.db->last_query), 0, strlen(pset.db->last_query));
securec_check_c(rc, "\0", "\0");
}
}
#endif
ResetCancelConn();
if (!AcceptResult(res)) {
PQclear(res);
res = NULL;
}
return res;
}
* PrintNotifications: check for asynchronous notifications, and print them out
*/
static void PrintNotifications(void)
{
PGnotify* notify = NULL;
while ((notify = PQnotifies(pset.db)) != NULL) {
if (notify->extra[0])
fprintf(pset.queryFout,
_("Asynchronous notification \"%s\" with payload \"%s\" received from server process with PID %lu.\n"),
notify->relname,
notify->extra,
notify->be_pid);
else
fprintf(pset.queryFout,
_("Asynchronous notification \"%s\" received from server process with PID %lu.\n"),
notify->relname,
notify->be_pid);
fflush(pset.queryFout);
PQfreemem(notify);
}
}
* PrintQueryTuples: assuming query result is OK, print its tuples
*
* Returns true if successful, false otherwise.
*/
static bool PrintQueryTuples(const PGresult* results)
{
printQueryOpt my_popt = pset.popt;
if (NULL != pset.gfname) {
FILE* queryFout_copy = pset.queryFout;
bool queryFoutPipe_copy = pset.queryFoutPipe;
pset.queryFout = stdout;
if (!setQFout(pset.gfname)) {
pset.queryFout = queryFout_copy;
pset.queryFoutPipe = queryFoutPipe_copy;
return false;
}
printQuery(results, &my_popt, pset.queryFout, pset.logfile);
setQFout(NULL);
pset.queryFout = queryFout_copy;
pset.queryFoutPipe = queryFoutPipe_copy;
free(pset.gfname);
pset.gfname = NULL;
} else
printQuery(results, &my_popt, pset.queryFout, pset.logfile);
return true;
}
* ProcessResult: utility function for use by SendQuery() only
*
* When our command string contained a COPY FROM STDIN or COPY TO STDOUT,
* PQexec() has stopped at the PGresult associated with the first such
* command. In that event, we'll marshal data for the COPY and then cycle
* through any subsequent PGresult objects.
*
* When the command string contained no affected COPY command, this function
* degenerates to an AcceptResult() call.
*
* Changes its argument to point to the last PGresult of the command string,
* or NULL if that result was for a COPY FROM STDIN or COPY TO STDOUT.
*
* Returns true on complete success, false otherwise. Possible failure modes
* include purely client-side problems; check the transaction status for the
* server-side opinion.
*
* print_error: Should we print error message to stderr in gsql?
*/
static bool ProcessResult(PGresult** results, bool is_explain, bool print_error)
{
PGresult* next_result = NULL;
bool success = true;
bool first_cycle = true;
if (is_explain) {
*results = PQgetResult(pset.db);
if (*results == NULL && ConnectionUp())
return success;
}
do {
ExecStatusType result_status;
bool is_copy = false;
if (!AcceptResult(*results, print_error)) {
* Failure at this point is always a server-side failure or a
* failure to submit the command string. Either way, we're
* finished with this command string.
*/
success = false;
break;
}
result_status = PQresultStatus(*results);
switch (result_status) {
case PGRES_EMPTY_QUERY:
case PGRES_COMMAND_OK:
case PGRES_TUPLES_OK:
is_copy = false;
break;
case PGRES_COPY_OUT:
case PGRES_COPY_IN:
is_copy = true;
break;
default:
is_copy = false;
psql_error("unexpected PQresultStatus: %d\n", result_status);
break;
}
if (is_copy) {
* Marshal the COPY data. Either subroutine will get the
* connection out of its COPY state, then call PQresultStatus()
* once and report any error.
*
* For COPY OUT, direct the output to pset.copyStream if it's set,
* otherwise to queryFout.
* For COPY IN, use pset.copyStream as data source if it's set,
* otherwise cur_cmd_source.
*/
FILE *copystream;
SetCancelConn();
if (result_status == PGRES_COPY_OUT) {
* pset.copyStream: invoked by \copy
* pset.queryFout: fall back to the generic query output stream
*/
copystream = pset.copyStream ? pset.copyStream : pset.queryFout;
success = handleCopyOut(pset.db, copystream) && success;
} else {
copystream = pset.copyStream ? pset.copyStream : pset.cur_cmd_source;
success = handleCopyIn(pset.db, copystream, PQbinaryTuples(*results)) && success;
}
ResetCancelConn();
* Call PQgetResult() once more. In the typical case of a
* single-command string, it will return NULL. Otherwise, we'll
* have other results to process that may include other COPYs.
*/
PQclear(*results);
*results = next_result = PQgetResult(pset.db);
} else if (is_explain || first_cycle)
break;
else if (!is_explain) {
next_result = PQgetResult(pset.db);
if (next_result != NULL) {
PQclear(*results);
*results = next_result;
}
}
first_cycle = false;
} while (NULL != next_result);
if (!first_cycle && !CheckConnection())
return false;
return success;
}
* PrintQueryStatus: report command status as required
*
* Note: Utility function for use by PrintQueryResults() only.
*/
static void PrintQueryStatus(PGresult* results)
{
char buf[16];
errno_t rc = EOK;
if (!pset.quiet) {
if (pset.popt.topt.format == PRINT_HTML) {
fputs("<p>", pset.queryFout);
html_escaped_print(PQcmdStatus(results), pset.queryFout);
fputs("</p>\n", pset.queryFout);
} else
fprintf(pset.queryFout, "%s\n", PQcmdStatus(results));
}
if (NULL != pset.logfile)
fprintf(pset.logfile, "%s\n", PQcmdStatus(results));
rc = sprintf_s(buf, sizeof(buf), "%u", (unsigned int)PQoidValue(results));
check_sprintf_s(rc);
if (!SetVariable(pset.vars, "LASTOID", buf)) {
psql_error("set variable %s failed.\n", "LASTOID");
}
}
* StoreQueryTuple: assuming query result is OK, save data into variables
*
* Returns true if successful, false otherwise.
*/
static bool StoreQueryTuple(const PGresult* result)
{
bool success = true;
if (PQntuples(result) < 1) {
psql_error("no rows returned for \\gset.\n");
success = false;
} else if (PQntuples(result) > 1) {
psql_error("more than one row returned for \\gset.\n");
success = false;
} else {
int i;
for (i = 0; i < PQnfields(result); i++) {
char* colname = PQfname(result, i);
char* varname = NULL;
char* value = NULL;
size_t len = strlen(pset.gsetPrefix) + strlen(colname) + 1;
varname = (char*)pg_malloc(len);
if (!varname) {
psql_error("out of memory!\n");
return false;
}
errno_t rc = snprintf_s(varname, len, len - 1, "%s%s", pset.gsetPrefix, colname);
securec_check_ss_c(rc, "\0", "\0");
if (VariableHasHook(pset.vars, varname)) {
psql_error("attempt to \\gset into specially treated variable \"%s\" ignored.\n", varname);
free(varname);
continue;
}
if (!PQgetisnull(result, 0, i)) {
value = PQgetvalue(result, 0, i);
} else {
value = NULL;
}
if (!SetVariable(pset.vars, varname, value)) {
psql_error("attempt to \\gset set variable \"%s\" to %s failed.\n", varname, value);
free(varname);
success = false;
break;
}
free(varname);
}
}
return success;
}
* PrintQueryResults: print out (or store) query results as required
*
* Note: Utility function for use by SendQuery() only.
*
* Returns true if the query executed successfully, false otherwise.
*/
static bool PrintQueryResults(PGresult* results)
{
bool success = false;
const char* cmdstatus = NULL;
int ret = 0;
if (NULL == results)
return false;
if (pset.parallel) {
if (IsInteractiveMode()) {
ret = LockMutexForParallel();
} else {
ret = file_lock(fileno(pset.queryFout), LOCK_EX);
}
if (ret == -1) {
psql_error("acquiring lock on output file failed.\n");
exit(EXIT_FAILURE);
}
}
switch (PQresultStatus(results)) {
case PGRES_TUPLES_OK:
if (pset.gsetPrefix != NULL) {
success = StoreQueryTuple(results);
} else {
success = PrintQueryTuples(results);
}
cmdstatus = PQcmdStatus(results);
if (strncmp(cmdstatus, "INSERT", 6) == 0 || strncmp(cmdstatus, "UPDATE", 6) == 0 ||
strncmp(cmdstatus, "DELETE", 6) == 0)
PrintQueryStatus(results);
break;
case PGRES_COMMAND_OK:
PrintQueryStatus(results);
success = true;
break;
case PGRES_EMPTY_QUERY:
success = true;
break;
case PGRES_COPY_OUT:
case PGRES_COPY_IN:
success = true;
break;
case PGRES_BAD_RESPONSE:
case PGRES_NONFATAL_ERROR:
case PGRES_FATAL_ERROR:
success = false;
break;
default:
success = false;
psql_error("unexpected PQresultStatus: %d\n", PQresultStatus(results));
break;
}
fflush(pset.queryFout);
if (pset.parallel) {
if (IsInteractiveMode()) {
(void)UnlockMutexForParallel();
} else {
(void)file_lock(fileno(pset.queryFout), LOCK_UN);
}
}
return success;
}
bool IsQueryNeedRetry(const char* sqlstate)
{
int i = 0;
if (CheckPoolerConnectionStatus()) {
pset.retry_sleep = false;
} else {
pset.retry_sleep = true;
return true;
}
for (i = 0; i < (int)pset.errcodes_list.size(); i++) {
if (pg_strncasecmp(sqlstate, pset.errcodes_list[i], strlen(sqlstate)) == 0)
return true;
}
if (pset.retry_times > 0 && pg_strncasecmp(sqlstate, pset.retry_sqlstate, strlen(sqlstate)) != 0)
ResetQueryRetryController();
return false;
}
static bool CheckPoolerConnectionStatus()
{
PGconn* conn = NULL;
PGresult* res = NULL;
bool status = true;
char* decode_pwd = NULL;
GS_UINT32 pwd_len = 0;
errno_t rc = EOK;
char* old_conninfo_values = NULL;
if (pset.connInfo.values[3] != NULL) {
decode_pwd = SEC_decodeBase64(pset.connInfo.values[3], &pwd_len);
if (decode_pwd == NULL) {
fprintf(stderr, "%s: decode retry connect messages failed.", pset.progname);
exit(EXIT_BADCONN);
}
old_conninfo_values = pset.connInfo.values[3];
pset.connInfo.values[3] = decode_pwd;
}
conn = PQconnectdbParams(pset.connInfo.keywords, pset.connInfo.values, true);
if (decode_pwd != NULL) {
rc = memset_s(decode_pwd, strlen(decode_pwd), 0, strlen(decode_pwd));
securec_check_c(rc, "\0", "\0");
OPENSSL_free(decode_pwd);
decode_pwd = NULL;
pset.connInfo.values[3] = old_conninfo_values;
}
if (CONNECTION_BAD == PQstatus(conn)) {
fprintf(stderr, "%s: %s", pset.progname, (char*)PQerrorMessage(conn));
PQfinish(conn);
return true;
}
res = PQexec(conn, "SELECT * FROM PGXC_POOL_CONNECTION_STATUS();");
if ((res != NULL) && PGRES_TUPLES_OK == PQresultStatus(res)) {
if ((res->tuples != NULL) && pg_strcasecmp(res->tuples[0]->value, "f") == 0)
status = false;
}
PQclear(res);
PQfinish(conn);
return status;
}
void ResetQueryRetryController()
{
pset.retry_times = 0;
errno_t rc = memset_s(pset.retry_sqlstate, sizeof(pset.retry_sqlstate), 0, sizeof(pset.retry_sqlstate));
securec_check_c(rc, "\0", "\0");
pset.retry_on = false;
pset.retry_sleep = false;
}
bool QueryRetryController(const char* query)
{
bool success = false;
pset.retry_times = 0;
for (;;) {
if (pset.retry_on)
printf(_("INFO: query retry %d time(s).\n"), ++pset.retry_times);
if ((pset.db != NULL) && pset.retry_sleep)
pg_usleep(60000000);
if (pset.retry_on) {
success = SendQuery(query);
} else {
break;
}
if (success || pset.retry_times == pset.max_retry_times) {
ResetQueryRetryController();
break;
}
}
return success;
}
bool GetPrintResult(PGresult** results, bool is_explain, bool is_print, const char* query, bool print_error)
{
bool OK = false;
bool return_value = true;
do {
OK = ProcessResult(results, is_explain, print_error);
if (*results == NULL)
break;
if (OK && is_print) {
OK = PrintQueryResults(*results);
RecordGucStmt(*results, query);
}
if (is_explain) {
PQclear(*results);
*results = NULL;
}
if (!OK && is_explain) {
return_value = false;
}
} while (is_explain);
return OK && return_value;
}
static void SendqueryCleanup()
{
ResetCancelConn();
if (pset.gsetPrefix) {
free(pset.gsetPrefix);
pset.gsetPrefix = NULL;
}
}
* SendQuery: send the query string to the backend
* (and print out results)
*
* Note: This is the "front door" way to send a query. That is, use it to
* send queries actually entered by the user. These queries will be subject to
* single step mode.
* To send "the door in back" queries (generated by slash commands, etc.) in a
* controlled way, use PSQLexec().
*
* print_error: Should gsql print error message to stderr with this query ?
*
* Returns true if the query executed successfully, false otherwise.
*/
bool SendQuery(const char* query, bool is_print, bool print_error)
{
PGresult* results = NULL;
PGTransactionStatusType transaction_status;
double elapsed_msec = 0;
bool OK = false;
bool on_error_rollback_savepoint = false;
static bool on_error_rollback_warning = false;
errno_t rc = 0;
#if defined(USE_ASSERT_CHECKING) || defined(FASTCHECK)
if (pset.parseonly)
return true;
#endif
if (NULL == pset.db) {
psql_error("You are currently not connected to a database.\n");
SendqueryCleanup();
return false;
}
if (pset.singlestep) {
char buf[3];
printf(_("***(Single step mode: verify command)*******************************************\n"
"%s\n"
"***(press return to proceed or enter x and return to cancel)********************\n"),
query);
fflush(stdout);
if (fgets(buf, sizeof(buf), stdin) != NULL)
if (buf[0] == 'x') {
SendqueryCleanup();
return false;
}
} else if (pset.echo == PSQL_ECHO_QUERIES) {
puts(query);
fflush(stdout);
}
if (pset.logfile != NULL) {
fprintf(pset.logfile,
_("********* QUERY **********\n"
"%s\n"
"**************************\n\n"),
query);
fflush(pset.logfile);
}
SetCancelConn();
transaction_status = PQtransactionStatus(pset.db);
if (transaction_status == PQTRANS_IDLE && !pset.autocommit && !command_no_begin(query)) {
results = PQexec(pset.db, "START TRANSACTION");
if (PQresultStatus(results) != PGRES_COMMAND_OK) {
psql_error("%s", PQerrorMessage(pset.db));
PQclear(results);
SendqueryCleanup();
return false;
}
PQclear(results);
transaction_status = PQtransactionStatus(pset.db);
}
if (transaction_status == PQTRANS_INTRANS && pset.on_error_rollback != PSQL_ERROR_ROLLBACK_OFF &&
(pset.cur_cmd_interactive || pset.on_error_rollback == PSQL_ERROR_ROLLBACK_ON)) {
if (on_error_rollback_warning == false && pset.sversion < 80000) {
fprintf(stderr,
_("The server (version %d.%d) does not support savepoints for ON_ERROR_ROLLBACK.\n"),
pset.sversion / 10000,
(pset.sversion / 100) % 100);
on_error_rollback_warning = true;
} else {
results = PQexec(pset.db, "SAVEPOINT pg_psql_temporary_savepoint");
if (PQresultStatus(results) != PGRES_COMMAND_OK) {
psql_error("%s", PQerrorMessage(pset.db));
PQclear(results);
SendqueryCleanup();
return false;
}
PQclear(results);
on_error_rollback_savepoint = true;
}
}
if (pset.fetch_count <= 0 || !is_select_command(query)) {
instr_time before, after;
bool is_explain = false;
if (pset.timing && is_print)
INSTR_TIME_SET_CURRENT(before);
is_explain = is_explain_command(query);
if (!is_explain)
results = PQexec(pset.db, query);
else if (!PQsendQuery(pset.db, query))
results = NULL;
if (is_explain) {
OK = GetPrintResult(&results, is_explain, is_print, query, print_error);
#ifdef HAVE_CE
if ((pset.db != NULL) && (pset.db->client_logic != NULL)) {
pset.db->client_logic->isInvalidOperationOnColumn = false;
}
#endif
if (pset.timing && is_print) {
INSTR_TIME_SET_CURRENT(after);
INSTR_TIME_SUBTRACT(after, before);
elapsed_msec = INSTR_TIME_GET_MILLISEC(after);
}
} else {
OK = ProcessResult(&results, is_explain, print_error);
if (pset.timing && is_print) {
INSTR_TIME_SET_CURRENT(after);
INSTR_TIME_SUBTRACT(after, before);
elapsed_msec = INSTR_TIME_GET_MILLISEC(after);
}
if (OK && is_print && results) {
OK = PrintQueryResults(results);
RecordGucStmt(results, query);
}
}
#ifndef WIN32
if (pset.cur_cmd_interactive) {
if ((pset.db != NULL) && (pset.db->last_query != NULL)) {
rc = memset_s(pset.db->last_query, strlen(pset.db->last_query), 0, strlen(pset.db->last_query));
securec_check_c(rc, "\0", "\0");
}
}
#endif
ResetCancelConn();
} else {
OK = ExecQueryUsingCursor(query, &elapsed_msec);
ResetCancelConn();
results = NULL;
}
if (on_error_rollback_savepoint) {
const char* svptcmd = NULL;
transaction_status = PQtransactionStatus(pset.db);
switch (transaction_status) {
case PQTRANS_INERROR:
svptcmd = "ROLLBACK TO pg_psql_temporary_savepoint";
break;
case PQTRANS_IDLE:
break;
case PQTRANS_INTRANS:
* Do nothing if they are messing with savepoints themselves:
* If the user did RELEASE or ROLLBACK, our savepoint is gone.
* If they issued a SAVEPOINT, releasing ours would remove
* theirs.
*/
if ((results != NULL) &&
(strcmp(PQcmdStatus(results), "SAVEPOINT") == 0 || strcmp(PQcmdStatus(results), "RELEASE") == 0 ||
strcmp(PQcmdStatus(results), "ROLLBACK") == 0))
svptcmd = NULL;
else
svptcmd = "RELEASE pg_psql_temporary_savepoint";
break;
case PQTRANS_ACTIVE:
case PQTRANS_UNKNOWN:
default:
OK = false;
if (transaction_status != PQTRANS_UNKNOWN || ConnectionUp())
psql_error("unexpected transaction status (%d)\n", transaction_status);
break;
}
if (NULL != svptcmd) {
PGresult* svptres = NULL;
svptres = PQexec(pset.db, svptcmd);
if (PQresultStatus(svptres) != PGRES_COMMAND_OK) {
psql_error("%s", PQerrorMessage(pset.db));
PQclear(svptres);
PQclear(results);
SendqueryCleanup();
return false;
}
PQclear(svptres);
}
}
PQclear(results);
if (pset.timing && is_print)
printf(_("Time: %.3f ms\n"), elapsed_msec);
if (pset.encoding != PQclientEncoding(pset.db) && PQclientEncoding(pset.db) >= 0) {
pset.encoding = PQclientEncoding(pset.db);
pset.popt.topt.encoding = pset.encoding;
if (!SetVariable(pset.vars, "ENCODING", pg_encoding_to_char(pset.encoding))) {
psql_error("set variable %s failed.\n", "ENCODING");
}
}
PrintNotifications();
SendqueryCleanup();
return OK;
}
* SendQuery: send the query string to the backend
* (and print out results)
*
* Note: This is the "front door" way to send a query. That is, use it to
* send queries actually entered by the user. These queries will be subject to
* single step mode.
* To send "the door in back" queries (generated by slash commands, etc.) in a
* controlled way, use PSQLexec().
*
* print_error: Should gsql print error message to stderr with this query ?
*
* Returns true if the query executed successfully, false otherwise.
*/
static void* StartCopyFrom(void *arg)
{
auto *copyarg = (CopyInArgs *) arg;
PGresult* results = NULL;
bool success = true;
PGresult* next_result = NULL;
bool first_cycle = true;
PQExpBufferData errMsgBuff;
const char* errMsg = NULL;
char* retMsg = nullptr;
initPQExpBuffer(&errMsgBuff);
pthread_mutex_lock(copyarg->stream_mutex);
results = PQexec(copyarg->conn, "START TRANSACTION");
if (PQresultStatus(results) != PGRES_COMMAND_OK) {
printfPQExpBuffer(&errMsgBuff, "%s", PQerrorMessage(copyarg->conn));
PQclear(results);
copyarg->result = false;
pthread_mutex_unlock(copyarg->stream_mutex);
retMsg = pg_strdup(errMsgBuff.data);
termPQExpBuffer(&errMsgBuff);
return retMsg;
}
PQclear(results);
results = PQexec(copyarg->conn, copyarg->query);
pthread_mutex_unlock(copyarg->stream_mutex);
do {
ExecStatusType result_status;
bool is_copy = false;
if (!AcceptResultWithErrMsg(results, &errMsg, copyarg->conn)) {
* Failure at this point is always a server-side failure or a
* failure to submit the command string. Either way, we're
* finished with this command string.
*/
success = false;
if (errMsg) {
appendPQExpBufferStr(&errMsgBuff, errMsg);
}
break;
}
result_status = PQresultStatus(results);
switch (result_status) {
case PGRES_EMPTY_QUERY:
case PGRES_COMMAND_OK:
case PGRES_TUPLES_OK:
is_copy = false;
break;
case PGRES_COPY_IN:
is_copy = true;
break;
default:
is_copy = false;
printfPQExpBuffer(&errMsgBuff, "unexpected PQresultStatus: %d\n", result_status);
break;
}
if (is_copy) {
* Marshal the COPY data. Either subroutine will get the
* connection out of its COPY state, then call PQresultStatus()
* once and report any error.
*/
success = ParallelCopyIn(copyarg, &errMsg) && success;
if (errMsg) {
appendPQExpBufferStr(&errMsgBuff, errMsg);
}
* Call PQgetResult() once more. In the typical case of a
* single-command string, it will return NULL. Otherwise, we'll
* have other results to process that may include other COPYs.
*/
PQclear(results);
results = next_result = PQgetResult(copyarg->conn);
} else if (first_cycle) {
break;
} else if ((next_result = PQgetResult(copyarg->conn)) != NULL) {
PQclear(results);
results = next_result;
}
first_cycle = false;
} while (NULL != next_result);
if (!first_cycle && !CheckConnection())
success = false;
PQclear(results);
copyarg->result = success;
if (errMsgBuff.len) {
retMsg = pg_strdup(errMsgBuff.data);
termPQExpBuffer(&errMsgBuff);
return retMsg;
}
termPQExpBuffer(&errMsgBuff);
return NULL;
}
static void ProcessCopyInResult(const CopyInArgs *copyargs, int nclients)
{
int index;
bool allThreadsSucc = true;
const char* endMsg = NULL;
PGresult* result = NULL;
for (index = 0; index < nclients; index++) {
if (!copyargs[index].result) {
allThreadsSucc = false;
break;
}
}
endMsg = allThreadsSucc ? "commit" : "rollback";
for (index = 0; index < nclients; index++) {
result = PQexec(copyargs[index].conn, endMsg);
if (PQresultStatus(result) != PGRES_COMMAND_OK) {
psql_error("%s", PQerrorMessage(copyargs[index].conn));
}
PQclear(result);
}
}
bool MakeCopyWorker(const char* query, int nclients)
{
CopyInArgs *copyargs = nullptr;
int index;
pthread_mutex_t mutexLock;
char* decode_pwd = nullptr;
GS_UINT32 pwd_len = 0;
errno_t rc = EOK;
char* old_conninfo_values = nullptr;
void* retVal = nullptr;
bool errPrinted = false;
bool success = true;
PGconn* oldConn = NULL;
* We clamp manually-set values to at least 1 client & at most 8 clients,
* if parallel parameter out of range.
*/
if (nclients < 1) {
nclients = 1;
}
if (nclients > 8) {
nclients = 8;
}
if (pset.connInfo.values[3] != NULL) {
decode_pwd = SEC_decodeBase64(pset.connInfo.values[3], &pwd_len);
if (decode_pwd == NULL) {
psql_error("%s: decode the parallel connect value failed.", pset.progname);
return false;
}
old_conninfo_values = pset.connInfo.values[3];
pset.connInfo.values[3] = decode_pwd;
}
copyargs = (CopyInArgs *) malloc(sizeof(CopyInArgs) * nclients);
if (copyargs == NULL) {
psql_error("out of memory\n");
exit(EXIT_FAILURE);
}
for (index = 0; index < nclients; index++) {
PGconn* conn = PQconnectdbParams(pset.connInfo.keywords, pset.connInfo.values, true);
if (PQstatus(conn) == CONNECTION_BAD) {
free(copyargs);
psql_error("%s: %s", pset.progname, (char*)PQerrorMessage(conn));
PQfinish(conn);
return false;
}
copyargs[index].conn = conn;
}
if (decode_pwd != NULL) {
rc = memset_s(decode_pwd, strlen(decode_pwd), 0, strlen(decode_pwd));
securec_check_c(rc, "\0", "\0");
OPENSSL_free(decode_pwd);
decode_pwd = NULL;
pset.connInfo.values[3] = old_conninfo_values;
}
oldConn = pset.db;
for (index = 0; index < nclients; index++) {
pset.db = copyargs[index].conn;
* Here we ignore the set statements errors, as something maybe change
* after user did so.
*/
for (int j = 0; j < pset.num_guc_stmt; j++) {
(void)SendQuery(pset.guc_stmt[j], false, false);
}
}
pset.db = oldConn;
pthread_mutex_init(&mutexLock, NULL);
pset.parallelCopyDone = false;
pset.parallelCopyOk = true;
for (index = 0; index < nclients; index++) {
CopyInArgs *arg = ©args[index];
arg->result = 0;
arg->query = query;
arg->stream_mutex = &mutexLock;
rc = pthread_create(&arg->thread, NULL, StartCopyFrom, arg);
}
* Establish longjmp destination for long wait, the main thread set flag to end subthreads.
* (This is only effective while sigint_interrupt_enabled is TRUE.)
*/
if (sigsetjmp(sigint_interrupt_jmp, 1) != 0) {
pthread_mutex_lock(&mutexLock);
pset.parallelCopyDone = true;
pset.parallelCopyOk = false;
pthread_mutex_unlock(&mutexLock);
}
sigint_interrupt_enabled = true;
for (index = 0; index < nclients; index++) {
pthread_join(copyargs[index].thread, &retVal);
if (retVal) {
if (!errPrinted) {
errPrinted = true;
success = false;
psql_error("%s", (char*)retVal);
}
free(retVal);
retVal = nullptr;
}
}
sigint_interrupt_enabled = false;
ProcessCopyInResult(copyargs, nclients);
for (index = 0; index < nclients; index++) {
PQfinish(copyargs[index].conn);
}
pthread_mutex_destroy(&mutexLock);
free(copyargs);
return success;
}
* ExecQueryUsingCursor: run a SELECT-like query using a cursor
*
* This feature allows result sets larger than RAM to be dealt with.
*
* Returns true if the query executed successfully, false otherwise.
*
* If pset.timing is on, total query time (exclusive of result-printing) is
* stored into *elapsed_msec.
*/
static bool ExecQueryUsingCursor(const char* query, double* elapsed_msec)
{
bool OK = true;
PGresult* results = NULL;
PQExpBufferData buf;
printQueryOpt my_popt = pset.popt;
FILE* queryFout_copy = pset.queryFout;
bool queryFoutPipe_copy = pset.queryFoutPipe;
bool started_txn = false;
bool did_pager = false;
int ntuples;
int fetchCount;
char fetch_cmd[64];
instr_time before, after;
int flush_error;
errno_t rc;
*elapsed_msec = 0;
my_popt.topt.start_table = true;
my_popt.topt.stop_table = false;
my_popt.topt.prior_records = 0;
if (pset.timing)
INSTR_TIME_SET_CURRENT(before);
if (PQtransactionStatus(pset.db) == PQTRANS_IDLE) {
results = PQexec(pset.db, "START TRANSACTION");
OK = AcceptResult(results) && (PQresultStatus(results) == PGRES_COMMAND_OK);
PQclear(results);
if (!OK) {
return false;
}
started_txn = true;
}
initPQExpBuffer(&buf);
appendPQExpBuffer(&buf, "CURSOR _psql_cursor NO SCROLL FOR\n%s", query);
results = PQexec(pset.db, buf.data);
OK = AcceptResult(results) && (PQresultStatus(results) == PGRES_COMMAND_OK);
PQclear(results);
termPQExpBuffer(&buf);
if (!OK) {
goto cleanup;
}
if (pset.timing) {
INSTR_TIME_SET_CURRENT(after);
INSTR_TIME_SUBTRACT(after, before);
*elapsed_msec += INSTR_TIME_GET_MILLISEC(after);
}
* In \gset mode, we force the fetch count to be 2, so that we will throw
* the appropriate error if the query returns more than one row.
*/
if (pset.gsetPrefix) {
fetchCount = GSET_MAX_FETCH_COUNT;
} else {
fetchCount = pset.fetch_count;
}
rc = sprintf_s(fetch_cmd, sizeof(fetch_cmd), "FETCH FORWARD %d FROM _psql_cursor", fetchCount);
check_sprintf_s(rc);
if (NULL != pset.gfname) {
pset.queryFout = stdout;
if (!setQFout(pset.gfname)) {
pset.queryFout = queryFout_copy;
pset.queryFoutPipe = queryFoutPipe_copy;
OK = false;
goto cleanup;
}
}
clearerr(pset.queryFout);
for (;;) {
if (pset.timing)
INSTR_TIME_SET_CURRENT(before);
results = PQexec(pset.db, fetch_cmd);
if (pset.timing) {
INSTR_TIME_SET_CURRENT(after);
INSTR_TIME_SUBTRACT(after, before);
*elapsed_msec += INSTR_TIME_GET_MILLISEC(after);
}
if (PQresultStatus(results) != PGRES_TUPLES_OK) {
if (did_pager) {
ClosePager(pset.queryFout);
pset.queryFout = queryFout_copy;
pset.queryFoutPipe = queryFoutPipe_copy;
did_pager = false;
}
OK = AcceptResult(results);
psql_assert(!OK);
PQclear(results);
break;
}
if (pset.gsetPrefix) {
OK = StoreQueryTuple(results);
PQclear(results);
break;
}
ntuples = PQntuples(results);
if (ntuples < fetchCount) {
my_popt.topt.stop_table = true;
} else if (pset.queryFout == stdout && !did_pager) {
* If query requires multiple result sets, hack to ensure that
* only one pager instance is used for the whole mess
*/
pset.queryFout = PageOutput(100000, my_popt.topt.pager);
did_pager = true;
}
printQuery(results, &my_popt, pset.queryFout, pset.logfile);
PQclear(results);
my_popt.topt.start_table = false;
my_popt.topt.prior_records += ntuples;
* Make sure to flush the output stream, so intermediate results are
* visible to the client immediately. We check the results because if
* the pager dies/exits/etc, there's no sense throwing more data at
* it.
*/
flush_error = fflush(pset.queryFout);
* Check if we are at the end, if a cancel was pressed, or if there
* were any errors either trying to flush out the results, or more
* generally on the output stream at all. If we hit any errors
* writing things to the stream, we presume $PAGER has disappeared and
* stop bothering to pull down more data.
*/
if (ntuples < fetchCount || cancel_pressed || flush_error || ferror(pset.queryFout)) {
break;
}
}
if (pset.gfname != NULL) {
setQFout(NULL);
pset.queryFout = queryFout_copy;
pset.queryFoutPipe = queryFoutPipe_copy;
free(pset.gfname);
pset.gfname = NULL;
} else if (did_pager) {
ClosePager(pset.queryFout);
pset.queryFout = queryFout_copy;
pset.queryFoutPipe = queryFoutPipe_copy;
}
cleanup:
if (pset.timing)
INSTR_TIME_SET_CURRENT(before);
* We try to close the cursor on either success or failure, but on failure
* ignore the result (it's probably just a bleat about being in an aborted
* transaction)
*/
results = PQexec(pset.db, "CLOSE _psql_cursor");
if (OK) {
OK = AcceptResult(results) && (PQresultStatus(results) == PGRES_COMMAND_OK);
}
PQclear(results);
if (started_txn) {
results = PQexec(pset.db, OK ? "COMMIT" : "ROLLBACK");
OK = OK && AcceptResult(results) && (PQresultStatus(results) == PGRES_COMMAND_OK);
PQclear(results);
}
if (pset.timing) {
INSTR_TIME_SET_CURRENT(after);
INSTR_TIME_SUBTRACT(after, before);
*elapsed_msec += INSTR_TIME_GET_MILLISEC(after);
}
return OK;
}
* Advance the given char pointer over white space and SQL comments.
*/
static const char* skip_white_space(const char* query)
{
int cnestlevel = 0;
while (*query) {
int mblen = PQmblen(query, pset.encoding);
* Note: we assume the encoding is a superset of ASCII, so that for
* example "query[0] == '/'" is meaningful. However, we do NOT assume
* that the second and subsequent bytes of a multibyte character
* couldn't look like ASCII characters; so it is critical to advance
* by mblen, not 1, whenever we haven't exactly identified the
* character we are skipping over.
*/
if (isspace((unsigned char)*query))
query += mblen;
else if (query[0] == '/' && query[1] == '*') {
cnestlevel++;
query += 2;
} else if (cnestlevel > 0 && query[0] == '*' && query[1] == '/') {
cnestlevel--;
query += 2;
} else if (cnestlevel == 0 && query[0] == '-' && query[1] == '-') {
query += 2;
* We have to skip to end of line since any slash-star inside the
* -- comment does NOT start a slash-star comment.
*/
while (*query) {
if (*query == '\n') {
query++;
break;
}
query += PQmblen(query, pset.encoding);
}
} else if (cnestlevel > 0) {
query += mblen;
} else {
break;
}
}
return query;
}
* judge begin is belong to anonymous block or transaction,if it belong to
* anonymous block,return false,otherwise return true.
*
*/
static bool is_begin_transaction(const char* query) {
if (pg_strncasecmp(query, "begin", 5) == 0) {
query = skip_white_space(query + 5);
if (query[0] == ';')
return true;
else
return false;
}
return false;
}
* Check whether a command is one of those for which we should NOT start
* a new transaction block (ie, send a preceding BEGIN).
*
* These include the transaction control statements themselves, plus
* certain statements that the backend disallows inside transaction blocks.
*/
static bool command_no_begin(const char* query)
{
int wordlen;
* First we must advance over any whitespace and comments.
*/
query = skip_white_space(query);
* Check word length (since "beginx" is not "begin").
*/
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
* Transaction control commands. These should include every keyword that
* gives rise to a TransactionStmt in the backend grammar, except for the
* savepoint-related commands.
*
* (We assume that START must be START TRANSACTION, since there is
* presently no other "START foo" command.)
*/
if (is_begin_transaction(query))
return true;
if (wordlen == 5 && pg_strncasecmp(query, "abort", 5) == 0)
return true;
if (wordlen == 5 && pg_strncasecmp(query, "start", 5) == 0)
return true;
if (wordlen == 6 && pg_strncasecmp(query, "commit", 6) == 0)
return true;
if (wordlen == 3 && pg_strncasecmp(query, "end", 3) == 0)
return true;
if (wordlen == 8 && pg_strncasecmp(query, "rollback", 8) == 0)
return true;
if (wordlen == 7 && pg_strncasecmp(query, "prepare", 7) == 0) {
query += wordlen;
query = skip_white_space(query);
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
if (wordlen == 11 && pg_strncasecmp(query, "transaction", 11) == 0)
return true;
return false;
}
* Commands not allowed within transactions. The statements checked for
* here should be exactly those that call PreventTransactionChain() in the
* backend.
*/
if (wordlen == 6 && pg_strncasecmp(query, "vacuum", 6) == 0)
return true;
if (wordlen == 7 && pg_strncasecmp(query, "cluster", 7) == 0) {
query += wordlen;
query = skip_white_space(query);
if (isalpha((unsigned char)query[0]))
return false;
return true;
}
if (wordlen == 6 && pg_strncasecmp(query, "create", 6) == 0) {
query += wordlen;
query = skip_white_space(query);
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
if (wordlen == 8 && pg_strncasecmp(query, "database", 8) == 0)
return true;
if (wordlen == 10 && pg_strncasecmp(query, "tablespace", 10) == 0)
return true;
if (wordlen == 6 && pg_strncasecmp(query, "unique", 6) == 0) {
query += wordlen;
query = skip_white_space(query);
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
}
if (wordlen == 5 && pg_strncasecmp(query, "index", 5) == 0) {
query += wordlen;
query = skip_white_space(query);
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
if (wordlen == 12 && pg_strncasecmp(query, "concurrently", 12) == 0)
return true;
}
return false;
}
* Note: these tests will match DROP SYSTEM and REINDEX TABLESPACE, which
* aren't really valid commands so we don't care much. The other four
* possible matches are correct.
*/
if ((wordlen == 4 && pg_strncasecmp(query, "drop", 4) == 0) ||
(wordlen == 7 && pg_strncasecmp(query, "reindex", 7) == 0)) {
query += wordlen;
query = skip_white_space(query);
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
if (wordlen == 8 && pg_strncasecmp(query, "database", 8) == 0)
return true;
if (wordlen == 6 && pg_strncasecmp(query, "system", 6) == 0)
return true;
if (wordlen == 10 && pg_strncasecmp(query, "tablespace", 10) == 0)
return true;
if (wordlen == 5 && (pg_strncasecmp(query, "index", 5) == 0 || pg_strncasecmp(query, "table", 5) == 0)) {
query += wordlen;
query = skip_white_space(query);
wordlen = 0;
while (isalpha((unsigned char) query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
* REINDEX [ TABLE | INDEX ] CONCURRENTLY are not allowed
* in xacts.
*/
if(wordlen == 12 && pg_strncasecmp(query, "concurrently", 12) == 0)
return true;
}
return false;
}
if (wordlen == 7 && pg_strncasecmp(query, "discard", 7) == 0) {
query += wordlen;
query = skip_white_space(query);
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
if (wordlen == 3 && pg_strncasecmp(query, "all", 3) == 0)
return true;
return false;
}
return false;
}
* Check whether the specified command is a SELECT (or VALUES).
*/
static bool is_select_command(const char* query)
{
int wordlen;
* First advance over any whitespace, comments and left parentheses.
*/
for (;;) {
query = skip_white_space(query);
if (query[0] == '(') {
query++;
} else {
break;
}
}
* Check word length (since "selectx" is not "select").
*/
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
if (wordlen == 6 && pg_strncasecmp(query, "select", 6) == 0)
return true;
if (wordlen == 6 && pg_strncasecmp(query, "values", 6) == 0)
return true;
return false;
}
* Check whether the specified command is a EXPLAIN.
*/
static bool is_explain_command(const char* query)
{
int wordlen;
bool result = false;
* First advance over any whitespace, comments and left parentheses.
*/
for (;;) {
query = skip_white_space(query);
if (query[0] == '(') {
query++;
} else {
break;
}
}
wordlen = 0;
while (isalpha((unsigned char)query[wordlen]))
wordlen += PQmblen(&query[wordlen], pset.encoding);
if ((wordlen == 7 && pg_strncasecmp(query, "explain", 7) == 0) ||
(wordlen == 4 && pg_strncasecmp(query, "call", 4) == 0))
result = true;
return result;
}
* Test if the current user is a database superuser.
*
* Note: this will correctly detect superuserness only with a protocol-3.0
* or newer backend; otherwise it will always say "false".
*/
bool is_superuser(void)
{
const char* val = NULL;
if (NULL == pset.db)
return false;
val = PQparameterStatus(pset.db, "is_sysadmin");
if ((val != NULL) && strcmp(val, "on") == 0)
return true;
return false;
}
* Test if the current session uses standard string literals.
*
* Note: With a pre-protocol-3.0 connection this will always say "false",
* which should be the right answer.
*/
bool standard_strings(void)
{
const char* val = NULL;
if (NULL == pset.db)
return false;
val = PQparameterStatus(pset.db, "standard_conforming_strings");
if ((val != NULL) && strcmp(val, "on") == 0)
return true;
return false;
}
* Return the session user of the current connection.
*
* Note: this will correctly detect the session user only with a
* protocol-3.0 or newer backend; otherwise it will return the
* connection user.
*/
const char* session_username(void)
{
const char* val = NULL;
if (NULL == pset.db)
return NULL;
val = PQparameterStatus(pset.db, "session_authorization");
if (NULL != val)
return val;
else
return PQuser(pset.db);
}
*
* substitute '~' with HOME or '~username' with username's home dir
*
*/
void expand_tilde(char** filename)
{
if ((filename == NULL) || ((*filename) == NULL))
return;
* WIN32 doesn't use tilde expansion for file names. Also, it uses tilde
* for short versions of long file names, though the tilde is usually
* toward the end, not at the beginning.
*/
#ifndef WIN32
if (**filename == '~') {
char* fn = NULL;
char oldp;
char *p = NULL;
struct passwd* pw = NULL;
char home[MAXPGPATH];
fn = *filename;
*home = '\0';
p = fn + 1;
while (*p != '/' && *p != '\0') {
p++;
}
oldp = *p;
*p = '\0';
if (*(fn + 1) == '\0') {
(void)get_home_path(home, sizeof(home));
} else if ((pw = getpwnam(fn + 1)) != NULL) {
errno_t err = EOK;
err = strcpy_s(home, sizeof(home), pw->pw_dir);
check_strcpy_s(err);
}
*p = oldp;
if (strlen(home) != 0) {
char* newfn = NULL;
errno_t rc;
size_t len = strlen(home) + strlen(p) + 1;
newfn = (char*)pg_malloc(len);
rc = sprintf_s(newfn, len, "%s%s", home, p);
check_sprintf_s(rc);
free(fn);
fn = NULL;
*filename = newfn;
}
}
#endif
return;
}
* Execute one query in a forked child process.
*/
static bool do_one_parallel(char* query, int fd)
{
bool success = true;
pid_t pid;
if ((pid = fork()) < 0) {
psql_error("Can't set up parallel execution for stmt: %s", query);
success = false;
} else if (pid == 0) {
int j;
char* decode_pwd = NULL;
GS_UINT32 pwd_len = 0;
errno_t rc = EOK;
char* old_conninfo_values = NULL;
if (pset.connInfo.values[3] != NULL) {
decode_pwd = SEC_decodeBase64(pset.connInfo.values[3], &pwd_len);
if (decode_pwd == NULL) {
fprintf(stderr, "%s: decode the parallel connect value failed.", pset.progname);
_exit(EXIT_BADCONN);
}
old_conninfo_values = pset.connInfo.values[3];
pset.connInfo.values[3] = decode_pwd;
}
pset.db = PQconnectdbParams(pset.connInfo.keywords, pset.connInfo.values, true);
if (PQstatus(pset.db) == CONNECTION_BAD) {
fprintf(stderr, "%s: %s", pset.progname, (char*)PQerrorMessage(pset.db));
(void)write(fd, "0", 1);
PQfinish(pset.db);
_exit(EXIT_BADCONN);
}
if (decode_pwd != NULL) {
rc = memset_s(decode_pwd, strlen(decode_pwd), 0, strlen(decode_pwd));
securec_check_c(rc, "\0", "\0");
OPENSSL_free(decode_pwd);
decode_pwd = NULL;
pset.connInfo.values[3] = old_conninfo_values;
}
(void)PQsetErrorVerbosity(pset.db, pset.verbosity);
if (0 == pset.max_retry_times) {
set_proc_title();
}
for (j = 0; j < pset.num_guc_stmt; j++) {
* after user did so.
*/
(void)SendQuery(pset.guc_stmt[j], false, false);
}
if (success) {
success = SendQuery(query);
if (!success && pset.retry_on) {
success = QueryRetryController(query);
}
}
(void)write(fd, success ? "1" : "0", 1);
PQfinish(pset.db);
EmptyRetryErrcodesList(pset.errcodes_list);
* descriptors belongs to their parent process, when subprocess is being
* "exit". This difference makes gsql hang while processing parallel commands.
* The hang issue is caused by rereading the contents processed before.
* "_exit" aborts the current process without cleaning the resources which will
* be cleaned by operating systems. So, it's safe here.
* But if some new operations need to be executed when subprocess of gsql is
* exiting, another solution should be found(such as close the file descriptors
* opened by parents.
*/
_exit(success ? EXIT_SUCCESS : EXIT_FAILURE);
}
return success;
}
* Execute parallel querys in many child processes and check their status.
*/
bool do_parallel_execution(int count, char** stmts)
{
int n, fd[2];
int i, num_parallel;
bool success = true;
struct pollfd ufds;
pset.parallel = true;
if (pipe(fd) < 0) {
psql_error("Can't set up communication for parallel execution\n");
return false;
}
if (IsInteractiveMode()) {
if (0 != CreateMutexForParallel()) {
close(fd[0]);
close(fd[1]);
return false;
}
}
(void)set_searchpath_for_tmptbl(pset.db);
for (i = 0; i < count && (pset.parallel_num == 0 || i < pset.parallel_num); i++) {
success = do_one_parallel(stmts[i], fd[1]);
if (!success) {
break;
}
}
num_parallel = i;
ufds.fd = fd[0];
ufds.events = POLLIN | POLLPRI;
ufds.revents = 0;
do {
int status, retval;
char x;
pid_t pid;
pid = waitpid(-1, &status, 0);
if (pid == -1) {
psql_error("Wait child processes failed and status is: %d\n", status);
success = false;
break;
}
retval = poll(&ufds, 1, 0);
if (retval > 0) {
n = read(fd[0], &x, 1);
success = success && ((x == '1') && (n == 1));
} else {
psql_error("Child processes exit and exit status is: %d\n", status);
success = false;
}
num_parallel--;
if (i < count && !(!success && pset.on_error_stop)) {
success = do_one_parallel(stmts[i++], fd[1]);
if (success) {
num_parallel++;
}
}
} while (num_parallel > 0);
if (IsInteractiveMode()) {
(void)DestroyMutexForParallel();
}
pset.parallel = false;
close(fd[0]);
close(fd[1]);
return success;
}
* Record the set guc statements for child process in parallel execute.
*/
static void RecordGucStmt(PGresult* results, const char* query)
{
char* cmdstatus = PQcmdStatus(results);
errno_t rc = 0;
if ((cmdstatus == NULL) || (strncmp(cmdstatus, "SET", 3) != 0 && strncmp(cmdstatus, "RESET", 5) != 0))
return;
* If SET/RESET statements is more than MAX_STMTS -1, realloc more memory.
* As normal user can't set role/session, there are no sensitive information leak
* risk for the use of realloc.
*/
if (pset.num_guc_stmt % MAX_STMTS == 0) {
if(NULL != pset.guc_stmt) {
char** temp = (char**)pg_calloc(1, sizeof(char*) * (pset.num_guc_stmt + MAX_STMTS));
rc = memcpy_s(temp, sizeof(char*) * pset.num_guc_stmt, pset.guc_stmt, sizeof(char*) * pset.num_guc_stmt);
securec_check_c(rc, "\0", "\0");
free(pset.guc_stmt);
pset.guc_stmt = temp;
}
else {
pset.guc_stmt = (char**)pg_calloc(1, sizeof(char*) * (pset.num_guc_stmt + MAX_STMTS));
}
}
pset.guc_stmt[pset.num_guc_stmt] = (char*)pg_malloc(sizeof(char) * (strlen(query) + 1));
rc = strncpy_s(pset.guc_stmt[pset.num_guc_stmt], strlen(query) + 1, query, strlen(query));
securec_check_c(rc, "\0", "\0");
pset.num_guc_stmt++;
}
* Set the child process proc name for distinction with father process.
*/
static void set_proc_title()
{
int i = 0;
int rc = 0;
int len = 0;
len = strlen(argv_para) + 1;
argv_para += len;
for (i = 0; i < argv_num - 1; i++) {
len = strlen(argv_para) + 1;
rc = memset_s(argv_para, len, 0, len);
securec_check_c(rc, "\0", "\0");
argv_para += len;
}
}
static int file_lock(int fd, unsigned int operation)
{
struct flock lck;
int cmd;
errno_t rc;
rc = memset_s(&lck, sizeof(lck), 0, sizeof(lck));
check_memset_s(rc);
lck.l_whence = SEEK_SET;
lck.l_start = 0;
lck.l_len = 0;
lck.l_pid = getpid();
if (operation & LOCK_UN)
lck.l_type = F_UNLCK;
else if (operation & LOCK_EX)
lck.l_type = F_WRLCK;
else
lck.l_type = F_RDLCK;
if (operation & LOCK_NB)
cmd = F_SETLK;
else
cmd = F_SETLKW;
return fcntl(fd, cmd, &lck);
}
static char* one_query(PGconn *conn, const char* stmt)
{
PGresult* result = PQexec(conn, stmt);
if (PQresultStatus(result) != PGRES_TUPLES_OK) {
PQclear(result);
return NULL;
}
if (PQntuples(result)) {
char* value1 = pg_strdup(PQgetvalue(result, 0, 0));
PQclear(result);
return value1;
}
PQclear(result);
return NULL;
}
static void set_searchpath_for_tmptbl(PGconn* conn)
{
static const char* stmtGetTempSchema = "select pg_catalog.quote_ident(nspname) "
"from pg_namespace where oid = pg_catalog.pg_my_temp_schema();";
char* nsp = one_query(conn, stmtGetTempSchema);
if (nsp == NULL) {
return;
}
static const char* stmtGetSearchPath = "select setting from pg_settings where name = 'search_path';";
char* searchPath = one_query(conn, stmtGetSearchPath);
bool alreadyInPath = (searchPath != NULL && strstr(searchPath, nsp) != NULL);
free(nsp);
free(searchPath);
if (alreadyInPath) {
return;
}
static const char* stmtBuildNewPath = "select "
"pg_catalog.concat( pg_catalog.quote_ident(n.nspname), ',', s.setting )"
"from pg_namespace n, pg_settings s "
"where n.oid = pg_catalog.pg_my_temp_schema() and s.name='search_path';";
char *value1 = one_query(conn, stmtBuildNewPath);
if (value1 == NULL) {
psql_error("get temp schema failed. \n");
PQfinish(conn);
exit(1);
}
char stmtSetSearchPath[128] = {0};
errno_t rc = sprintf_s(stmtSetSearchPath, sizeof(stmtSetSearchPath), "set search_path to %s;", value1);
check_sprintf_s(rc);
bool success = SendQuery(stmtSetSearchPath);
if (!success) {
psql_error("set temp schema failed. \n");
}
free(value1);
}
static int CreateMutexForParallel()
{
pset.parallelMutex = (struct parallelMutex_t*)mmap(
NULL, sizeof(*pset.parallelMutex), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0);
if (NULL == pset.parallelMutex) {
psql_error("Failed to create mutex for parallel execution.\n");
return -1;
}
if (pset.parallelMutex == MAP_FAILED) {
psql_error("Failed to create mutex for parallel execution.\n");
return -1;
}
check_memset_s(memset_s(pset.parallelMutex, sizeof(*pset.parallelMutex), 0, sizeof(*pset.parallelMutex)));
if (0 != pthread_mutexattr_init(&pset.parallelMutex->mutAttr)) {
psql_error("Failed to create mutex attribute for parallel execution.\n");
munmap(pset.parallelMutex, sizeof(*pset.parallelMutex));
return -1;
}
if (0 != pthread_mutexattr_setpshared(&pset.parallelMutex->mutAttr, PTHREAD_PROCESS_SHARED)) {
psql_error("Failed to set mutex attribute to share mode for parallel execution.\n");
pthread_mutexattr_destroy(&pset.parallelMutex->mutAttr);
munmap(pset.parallelMutex, sizeof(*pset.parallelMutex));
return -1;
}
if (0 != pthread_mutex_init(&pset.parallelMutex->mut, &pset.parallelMutex->mutAttr)) {
psql_error("Failed to create mutex for parallel execution.\n");
pthread_mutexattr_destroy(&pset.parallelMutex->mutAttr);
munmap(pset.parallelMutex, sizeof(*pset.parallelMutex));
return -1;
}
return 0;
}
static int LockMutexForParallel()
{
return pthread_mutex_lock(&pset.parallelMutex->mut);
}
static int UnlockMutexForParallel()
{
return pthread_mutex_unlock(&pset.parallelMutex->mut);
}
static int DestroyMutexForParallel()
{
int ret = 0;
if (0 != pthread_mutexattr_destroy(&pset.parallelMutex->mutAttr))
ret = -1;
if (0 != pthread_mutex_destroy(&pset.parallelMutex->mut))
ret = -1;
if (0 != munmap(pset.parallelMutex, sizeof(*pset.parallelMutex)))
ret = -1;
pset.parallelMutex = NULL;
return ret;
}
* GetEnvStr
*
* Note: malloc space for get the return of getenv() function, then return the malloc space.
* so, this space need be free.
*/
char* GetEnvStr(const char* env)
{
char* tmpvar = NULL;
const char* temp = getenv(env);
errno_t rc = 0;
if (temp != NULL) {
size_t len = strlen(temp);
if (len == 0) {
return NULL;
}
tmpvar = (char*)malloc(len + 1);
if (tmpvar != NULL) {
rc = strcpy_s(tmpvar, len + 1, temp);
securec_check_c(rc, "\0", "\0");
return tmpvar;
}
}
return NULL;
}
bool CheckDBCompatibility(PGconn *connection, char *dbCompatibility)
{
bool isCompatibility = false;
PGresult *res = PQexec(connection, "select setting from pg_catalog.pg_settings where name = 'sql_compatibility'");
isCompatibility = (res != NULL && PQresultStatus(res) == PGRES_TUPLES_OK &&
PQntuples(res) == 1 && strcmp(PQgetvalue(res, 0, 0), dbCompatibility) == 0);
PQclear(res);
return isCompatibility;
}
bool CheckSpecificExtension(PGconn *connection, char *extension)
{
bool isHasExtension = false;
char sql[100] = {0};
int rc = 0;
rc = sprintf_s(sql, sizeof(sql), "SELECT extname from pg_catalog.pg_extension where extname = '%s'", extension);
securec_check_ss_c(rc, "\0", "\0");
PGresult *res = PQexec(connection, sql);
isHasExtension = (res != NULL && PQresultStatus(res) == PGRES_TUPLES_OK &&
PQntuples(res) == 1);
PQclear(res);
return isHasExtension;
}
DBFormatType GetDatabaseType()
{
char* compatibilityStr;
DBFormatType dbType = A_FORMAT;
PGresult* res = PQexec(pset.db, "show sql_compatibility");
if (res != NULL && PQresultStatus(res) == PGRES_TUPLES_OK) {
compatibilityStr = PQgetvalue(res, 0, 0);
if (strcmp(compatibilityStr, "A") == 0) {
dbType = A_FORMAT;
} else if (strcmp(compatibilityStr, "B") == 0) {
dbType = B_FORMAT;
} else if (strcmp(compatibilityStr, "C") == 0) {
dbType = C_FORMAT;
} else if (strcmp(compatibilityStr, "D") == 0) {
dbType = D_FORMAT;
} else if (strcmp(compatibilityStr, "PG") == 0) {
dbType = PG_FORMAT;
}
}
PQclear(res);
res = NULL;
return dbType;
}
bool SaveSqlModePipesAsConcat()
{
if (!IS_CMPT(pset.dbType, B_FORMAT)) {
return false;
}
bool needResetSqlMode = false;
PGresult* res = NULL;
PQExpBufferData buf;
initPQExpBuffer(&buf);
printfPQExpBuffer(&buf, "select setting from pg_settings where name = 'dolphin.sql_mode';");
res = PSQLexec(buf.data, false);
if (res != NULL && PQntuples(res) == 1) {
char* dolphinSqlMode = PQgetvalue(res, 0, 0);
if (strstr(dolphinSqlMode, "pipes_as_concat") != NULL) {
needResetSqlMode = false;
} else {
if (g_sqlMode != NULL) {
free(g_sqlMode);
}
g_sqlMode = pg_strdup(dolphinSqlMode);
needResetSqlMode = true;
}
}
if (res != NULL) {
PQclear(res);
res = NULL;
}
termPQExpBuffer(&buf);
return needResetSqlMode;
}
void ResetSqlMode()
{
if (!IS_CMPT(pset.dbType, B_FORMAT)) {
return;
}
if (g_sqlMode == NULL) {
return;
}
PGresult* res = NULL;
PQExpBufferData buf;
initPQExpBuffer(&buf);
printfPQExpBuffer(&buf, "set dolphin.sql_mode = '%s';", g_sqlMode);
res = PSQLexec(buf.data, false);
if (res != NULL) {
PQclear(res);
res = NULL;
}
termPQExpBuffer(&buf);
}
* Need to call the ResetSqlMode to reset dolphin.sql_mode value adter call EnableSqlModePipesAsConcat
*/
void EnableSqlModePipesAsConcat()
{
if (!IS_CMPT(pset.dbType, B_FORMAT)) {
return;
}
if (!SaveSqlModePipesAsConcat()) {
return;
}
PGresult* res = NULL;
PQExpBufferData buf;
initPQExpBuffer(&buf);
if (g_sqlMode == NULL || (strlen(g_sqlMode) == 0)) {
printfPQExpBuffer(&buf, "set dolphin.sql_mode = 'pipes_as_concat';");
} else {
printfPQExpBuffer(&buf, "set dolphin.sql_mode = '%s,pipes_as_concat';", g_sqlMode);
}
res = PSQLexec(buf.data, false);
if (res != NULL) {
PQclear(res);
res = NULL;
}
termPQExpBuffer(&buf);
}
