statement.c

postgresql-odbc,跨平台应用

	ipdopts = SC_get_IPDF(self);
	has_out_para = FALSE;
	if (self->statement_type == STMT_TYPE_PROCCALL &&
		(SC_get_errornumber(self) == STMT_OK ||
		 SC_get_errornumber(self) == STMT_INFO_ONLY))
	{
		Int2	io, out;
		has_out_para = (CountParameters(self, NULL, &io, &out) > 0);
	}
	if (has_out_para)
	{	/* get the return value of the procedure call */
		RETCODE		ret;
		HSTMT		hstmt = (HSTMT) self;

		self->bind_row = 0;
		ret = SC_fetch(hstmt);
inolog("!!SC_fetch return =%d\n", ret);
		if (SQL_SUCCEEDED(ret))
		{
			APDFields	*apdopts = SC_get_APDF(self);
			SQLULEN		offset = apdopts->param_offset_ptr ? *apdopts->param_offset_ptr : 0;
			ARDFields	*ardopts = SC_get_ARDF(self);
			const ParameterInfoClass	*apara;
			const ParameterImplClass	*ipara;
			int	save_bind_size = ardopts->bind_size, gidx, num_p;

			ardopts->bind_size = apdopts->param_bind_type;
			num_p = self->num_params;
			if (ipdopts->allocated < num_p)
				num_p = ipdopts->allocated;
			for (i = 0, gidx = 0; i < num_p; i++)
			{
				ipara = ipdopts->parameters + i;
				if (ipara->paramType == SQL_PARAM_OUTPUT ||
				    ipara->paramType == SQL_PARAM_INPUT_OUTPUT)
				{
					apara = apdopts->parameters + i;	
					ret = PGAPI_GetData(hstmt, gidx + 1, apara->CType, apara->buffer + offset, apara->buflen, apara->used ? LENADDR_SHIFT(apara->used, offset) : NULL);
					if (!SQL_SUCCEEDED(ret))
					{
						SC_set_error(self, STMT_EXEC_ERROR, "GetData to Procedure return failed.", func);
						break;
					}
					gidx++;
				}
			}
			ardopts->bind_size = save_bind_size; /* restore */
		}
		else
		{
			SC_set_error(self, STMT_EXEC_ERROR, "SC_fetch to get a Procedure return failed.", func);
		}
	}
cleanup:
#undef	return
	SC_SetExecuting(self, FALSE);
	CLEANUP_FUNC_CONN_CS(func_cs_count, conn);
	if (CONN_DOWN != conn->status)
		conn->status = oldstatus;
	/* self->status = STMT_FINISHED; */
	if (SC_get_errornumber(self) == STMT_OK)
		return SQL_SUCCESS;
	else if (SC_get_errornumber(self) < STMT_OK)
		return SQL_SUCCESS_WITH_INFO;
	else
	{
		if (!SC_get_errormsg(self) || !SC_get_errormsg(self)[0])
		{
			SC_set_errormsg(self, "Error while executing the query");
			SC_log_error(func, NULL, self);
		}
		return SQL_ERROR;
	}
}

#define	CALLBACK_ALLOC_ONCE	4
int enqueueNeedDataCallback(StatementClass *stmt, NeedDataCallfunc func, void *data)
{
	if (stmt->num_callbacks >= stmt->allocated_callbacks)
	{
		SC_REALLOC_return_with_error(stmt->callbacks, NeedDataCallback,
			sizeof(NeedDataCallback) * (stmt->allocated_callbacks +
				CALLBACK_ALLOC_ONCE), stmt, "NeedDataCallback enqueue error", 0);
		stmt->allocated_callbacks += CALLBACK_ALLOC_ONCE;
	}
	stmt->callbacks[stmt->num_callbacks].func = func;
	stmt->callbacks[stmt->num_callbacks].data = data;
	stmt->num_callbacks++;

inolog("enqueueNeedDataCallack stmt=%p, func=%p, count=%d\n", stmt, func, stmt->num_callbacks);
	return stmt->num_callbacks;
}

RETCODE dequeueNeedDataCallback(RETCODE retcode, StatementClass *stmt)
{
	RETCODE			ret;
	NeedDataCallfunc	func;
	void			*data;
	int			i, cnt;

	mylog("dequeueNeedDataCallback ret=%d count=%d\n", retcode, stmt->num_callbacks);
	if (SQL_NEED_DATA == retcode)
		return retcode;
	if (stmt->num_callbacks <= 0)
		return retcode;
	func = stmt->callbacks[0].func;
	data = stmt->callbacks[0].data;
	for (i = 1; i < stmt->num_callbacks; i++)
		stmt->callbacks[i - 1] = stmt->callbacks[i];
	cnt = --stmt->num_callbacks;
	ret = (*func)(retcode, data);
	free(data);
	if (SQL_NEED_DATA != ret && cnt > 0)
		ret = dequeueNeedDataCallback(ret, stmt);
	return ret;
}

void	cancelNeedDataState(StatementClass *stmt)
{
	int	cnt = stmt->num_callbacks, i;

	stmt->num_callbacks = 0;
	for (i = 0; i < cnt; i++)
	{
		if (stmt->callbacks[i].data)
			free(stmt->callbacks[i].data);
	}
	SC_reset_delegate(SQL_ERROR, stmt);
}

void
SC_log_error(const char *func, const char *desc, const StatementClass *self)
{
	const	char *head;
#ifdef PRN_NULLCHECK
#define nullcheck(a) (a ? a : "(NULL)")
#endif
	if (self)
	{
		QResultClass *res = SC_get_Result(self);
		const ARDFields	*opts = SC_get_ARDF(self);
		const APDFields	*apdopts = SC_get_APDF(self);
		SQLLEN	rowsetSize;

#if (ODBCVER >= 0x0300)
		rowsetSize = (7 == self->transition_status ? opts->size_of_rowset_odbc2 : opts->size_of_rowset);
#else
		rowsetSize = opts->size_of_rowset_odbc2;
#endif /* ODBCVER */
		if (SC_get_errornumber(self) <= 0)
			head = "STATEMENT WARNING";
		else
		{
			head = "STATEMENT ERROR";
			qlog("%s: func=%s, desc='%s', errnum=%d, errmsg='%s'\n",head, func, desc, self->__error_number, nullcheck(self->__error_message));
		}
		mylog("%s: func=%s, desc='%s', errnum=%d, errmsg='%s'\n", head, func, desc, self->__error_number, nullcheck(self->__error_message));
		if (SC_get_errornumber(self) > 0)
		{
			qlog("                 ------------------------------------------------------------\n");
			qlog("                 hdbc=%p, stmt=%p, result=%p\n", self->hdbc, self, res);
			qlog("                 prepare=%d, internal=%d\n", self->prepare, self->internal);
			qlog("                 bindings=%p, bindings_allocated=%d\n", opts->bindings, opts->allocated);
			qlog("                 parameters=%p, parameters_allocated=%d\n", apdopts->parameters, apdopts->allocated);
			qlog("                 statement_type=%d, statement='%s'\n", self->statement_type, nullcheck(self->statement));
			qlog("                 stmt_with_params='%s'\n", nullcheck(self->stmt_with_params));
			qlog("                 data_at_exec=%d, current_exec_param=%d, put_data=%d\n", self->data_at_exec, self->current_exec_param, self->put_data);
			qlog("                 currTuple=%d, current_col=%d, lobj_fd=%d\n", self->currTuple, self->current_col, self->lobj_fd);
			qlog("                 maxRows=%d, rowset_size=%d, keyset_size=%d, cursor_type=%d, scroll_concurrency=%d\n", self->options.maxRows, rowsetSize, self->options.keyset_size, self->options.cursor_type, self->options.scroll_concurrency);
			qlog("                 cursor_name='%s'\n", SC_cursor_name(self));

			qlog("                 ----------------QResult Info -------------------------------\n");

			if (res)
			{
				qlog("                 fields=%p, backend_tuples=%p, tupleField=%d, conn=%p\n", res->fields, res->backend_tuples, res->tupleField, res->conn);
				qlog("                 fetch_count=%d, num_total_rows=%d, num_fields=%d, cursor='%s'\n", res->fetch_number, QR_get_num_total_tuples(res), res->num_fields, nullcheck(QR_get_cursor(res)));
				qlog("                 message='%s', command='%s', notice='%s'\n", nullcheck(res->message), nullcheck(res->command), nullcheck(res->notice));
				qlog("                 status=%d, inTuples=%d\n", QR_get_rstatus(res), QR_is_fetching_tuples(res));
			}

			/* Log the connection error if there is one */
			CC_log_error(func, desc, self->hdbc);
		}
	}
	else
	{
		qlog("INVALID STATEMENT HANDLE ERROR: func=%s, desc='%s'\n", func, desc);
		mylog("INVALID STATEMENT HANDLE ERROR: func=%s, desc='%s'\n", func, desc);
	}
#undef PRN_NULLCHECK
}

/*
 *	Extended Query 
 */

static BOOL
RequestStart(StatementClass *stmt, ConnectionClass *conn, const char *func)
{
	BOOL	ret = TRUE;

	if (SC_accessed_db(stmt))
		return TRUE;
	if (SQL_ERROR == SetStatementSvp(stmt))
	{
		char	emsg[128];

		snprintf(emsg, sizeof(emsg), "internal savepoint error in %s", func);
		SC_set_error(stmt, STMT_INTERNAL_ERROR, emsg, func);
		return FALSE;
	}
	if (!CC_is_in_trans(conn) && !CC_is_in_autocommit(conn))
	{
		if (ret = CC_begin(conn), !ret)
			return ret;
	}
	return ret;
}

BOOL
SendBindRequest(StatementClass *stmt, const char *plan_name)
{
	CSTR	func = "SendBindRequest";
	ConnectionClass	*conn = SC_get_conn(stmt);

	mylog("%s: plan_name=%s\n", func, plan_name);
	if (!RequestStart(stmt, conn, func))
		return FALSE;
	if (!BuildBindRequest(stmt, plan_name))
		return FALSE;

	return TRUE;
}

QResultClass *SendSyncAndReceive(StatementClass *stmt, QResultClass *res, const char *comment)
{
	CSTR func = "SendSyncAndReceive";
	ConnectionClass	*conn = SC_get_conn(stmt);
	SocketClass	*sock = conn->sock;
	char		id;
	Int4		response_length;
	UInt4		oid;
	int		num_p, num_io_params;
	int		i, pidx;
	Int2		num_discard_params, paramType;
	BOOL		rcvend = FALSE, msg_truncated;
	char		msgbuffer[ERROR_MSG_LENGTH + 1];
	IPDFields	*ipdopts;
	QResultClass	*newres = NULL;

	if (!RequestStart(stmt, conn, func))
		return NULL;

	SOCK_put_char(sock, 'S');	/* Sync command */
	SOCK_put_int(sock, 4, 4);	/* length */
	SOCK_flush_output(sock);

	if (!res)
		newres = res = QR_Constructor();
	for (;!rcvend;)
	{
		id = SOCK_get_id(sock);
		if ((SOCK_get_errcode(sock) != 0) || (id == EOF))
		{
			SC_set_error(stmt, CONNECTION_NO_RESPONSE, "No response rom the backend", func);

			mylog("%s: 'id' - %s\n", func, SC_get_errormsg(stmt));
			CC_on_abort(conn, CONN_DEAD);
			QR_Destructor(newres);
			return NULL;
		}
inolog("desc id=%c", id);
		response_length = SOCK_get_response_length(sock);
inolog(" response_length=%d\n", response_length);
		switch (id)
		{
			case 'C':
				SOCK_get_string(sock, msgbuffer, sizeof(msgbuffer));
				mylog("command response=%s\n", msgbuffer);
				QR_set_command(res, msgbuffer);
				if (QR_is_fetching_tuples(res))
				{
					QR_set_no_fetching_tuples(res);
					/* in case of FETCH, Portal Suspend never arrives */
					if (strnicmp(msgbuffer, "SELECT", 6) == 0)
					{
						mylog("%s: reached eof now\n", func);
						QR_set_reached_eof(res);
					}
					else
					{
						int	ret1, ret2;

						ret1 = ret2 = 0;
						if (sscanf(msgbuffer, "%*s %d %d", &ret1, &ret2) > 1)
							res->recent_processed_row_count = ret2;
						else
							res->recent_processed_row_count = ret1;
					}
				}
				break;
			case 'E': /* ErrorMessage */
				msg_truncated = handle_error_message(conn, msgbuffer, sizeof(msgbuffer), res->sqlstate, comment, res);

				rcvend = TRUE;
				break;
			case 'N': /* Notice */
				msg_truncated = handle_notice_message(conn, msgbuffer, sizeof(msgbuffer), res->sqlstate, comment, res);
				break;
			case '1': /* ParseComplete */
				if (stmt->plan_name)
					SC_set_prepared(stmt, PREPARED_PERMANENTLY);
				else
					SC_set_prepared(stmt, PREPARED_TEMPORARILY);
				break;
			case '2': /* BindComplete */
				QR_set_fetching_tuples(res);
				break;
			case '3': /* CloseComplete */
				QR_set_no_fetching_tuples(res);
				break;
			case 'Z': /* ReadyForQuery */
				rcvend = TRUE;
				EatReadyForQuery(conn);
				break;
			case 't': /* ParameterDesription */
				num_p = SOCK_get_int(sock, 2);
inolog("num_params=%d info=%d\n", stmt->num_params, num_p);
				num_discard_params = 0;
				if (stmt->discard_output_params)
					CountParameters(stmt, NULL, NULL, &num_discard_params);
				if (num_discard_params < stmt->proc_return)
					num_discard_params = stmt->proc_return;
				if (num_p + num_discard_params != (int) stmt->num_params)
				{
					mylog("ParamInfo unmatch num_params(=%d) != info(=%d)+discard(=%d)\n", stmt->num_params, num_p, num_discard_params);
					/* stmt->num_params = (Int2) num_p + num_discard_params; it's possible in case of multi command queries */
				}
				ipdopts = SC_get_IPDF(stmt);
				extend_iparameter_bindings(ipdopts, stmt->num_params);
#ifdef	NOT_USED
				if (stmt->discard_output_params)
				{
					for (i = 0, pidx = stmt->proc_return; i < num_p && pidx < stmt->num_params; pidx++)
					{
						paramType = ipdopts->parameters[pidx].paramType;
						if (SQL_PARAM_OUTPUT == paramType)
						{
							i++;
							continue;
						}
						oid = SOCK_get_int(sock, 4);
						ipdopts->parameters[pidx].PGType = oid;
					}
				}
				else
				{
					for (i = 0, pidx = stmt->proc_return; i < num_p; i++, pidx++)
					{
						paramType = ipdopts->parameters[pidx].paramType;	
						oid = SOCK_get_int(sock, 4);
						if (SQL_PARAM_OUTPUT != paramType ||
						    PG_TYPE_VOID != oid)
							ipdopts->parameters[pidx].PGType = oid;
					}
				}
#endif /* NOT_USED */
				pidx = stmt->current_exec_param;
				if (pidx >= 0)
					pidx--;
				for (i = 0; i < num_p; i++)
				{
					SC_param_next(stmt, &pidx, NULL, NULL);
					if (pidx >= stmt->num_params)
					{
						mylog("%dth parameter's position(%d) is out of bound[%d]\n", i, pidx, stmt->num_params);
						break;
					}
					oid = SOCK_get_int(sock, 4);
					paramType = ipdopts->parameters[pidx].paramType;	
					if (SQL_PARAM_OUTPUT != paramType ||
					    PG_TYPE_VOID != oid)
						ipdopts->parameters[pidx].PGType = oid;
				}
				break;
			case 'T': /* RowDesription */
				QR_set_conn(res, conn);
				if (CI_read_fields(res->fields, conn))
				{
					Int2	dummy1, dummy2;
					int	cidx;

					QR_set_rstatus(res, PORES_FIELDS_OK);
					res->num_fields = CI_get_num_fields(res->fields);
					if (QR_haskeyset(res))
						res->num_fields -= res->num_key_fields;
					num_io_params = CountParameters(stmt, NULL, &dummy1, &dummy2);
					if (stmt->proc_return > 0 ||
					    num_io_params > 0)
					{
						ipdopts = SC_get_IPDF(stmt);
						extend_iparameter_bindings(ipdopts, stmt->num_params);
						for (i = 0, cidx = 0; i < stmt->num_params; i++)
						{
							if (i < stmt->proc_return)
								ipdopts->parameters[i].paramType = SQL_PARAM_OUTPUT;
							paramType =ipdopts->parameters[i].paramType;
 							if (SQL_PARAM_OUTPUT == paramType ||
							    SQL_PARAM_INPUT_OUTPUT == paramType)
							{
inolog("!![%d].PGType %u->%u\n", i, ipdopts->parameters[i].PGType, CI_get_oid(res->fields, cidx));
								ipdopts->parameters[i].PGType = CI_get_oid(res->fields, cidx);
								cidx++;
							}
						}
					}
				}
				else
				{
					QR_set_rstatus(res, PORES_BAD_RESPONSE);
					QR_set_message(res, "Error reading field information");
					rcvend = TRUE;
				}
				break;
			case 'B': /* Binary data */
			case 'D': /* ASCII data */
				QR_get_tupledata(res, id == 'B');
				break;
			case 'S': /* parameter status */
				getParameterValues(conn);
				break;
			case 's':	/* portal suspend */
				QR_set_no_fetching_tuples(res);
				break;
			default:
				break;
		}
	}
	return res;
}

BOOL
SendParseRequest(StatementClass *stmt, const char *plan_name, const char *query, Int4 qlen, Int2 num_params)
{
	CSTR	func = "SendParseRequest";
	ConnectionClass	*conn = SC_get_conn(stmt);
	SocketClass	*sock = conn->sock;
	Int4		sta_pidx, end_pidx;
	size_t		pileng, leng;

	mylog("%s: plan_name=%s query=%s\n", func, plan_name, query);
	qlog("%s: plan_name=%s query=%s\n", func, plan_name, query);
	if (!RequestStart(stmt, conn, func))
		return FALSE;

	SOCK_put_char(sock, 'P'); /* Parse command */
	if (SOCK_get_errcode(sock) != 0)
	{
		CC_set_error(conn, CONNECTION_COULD_NOT_SEND, "Could not send P request to backend", func);
		CC_on_abort(conn, CONN_DEAD);
		return FALSE;
	}

	pileng = sizeof(Int2);
	if (stmt->discard_output_params)
		num_params = 0;
	else if (num_params != 0)
	{
#ifdef	NOT_USED
		sta_pidx += stmt->proc_return;
#endif /* NOT_USED */
		int	pidx;

		sta_pidx = stmt->current_exec_param;
		if (num_params < 0)
			end_pidx = stmt->num_params - 1;
		else
			end_pidx = sta_pidx + num_params - 1;
#ifdef	NOT_USED
		num_params = end_pidx - sta_pidx + 1;
#endif /* NOT_USED */
		for (num_params = 0, pidx = sta_pidx - 1;;)
		{
			SC_param_next(stmt, &pidx, NULL, NULL);
			if (pidx > end_pidx)
				break;
			else if (pidx < end_pidx)
				num_params++;
			else
			{
				num_params++;
				break;
			}
		}
mylog("sta_pidx=%d end_pidx=%d num_p=%d\n", sta_pidx, end_pidx, num_params);
		pileng += (sizeof(UInt4) * num_params);
	}
	qlen = (SQL_NTS == qlen) ? strlen(query) : qlen; 
	leng = st