fe-exec.c

关系型数据库 Postgresql 6.5.2

}

/*
 * Consume any available input from the backend
 * 0 return: some kind of trouble
 * 1 return: no problem
 */

int
PQconsumeInput(PGconn *conn)
{
	if (!conn)
		return 0;

	/*
	 * Load more data, if available. We do this no matter what state we
	 * are in, since we are probably getting called because the
	 * application wants to get rid of a read-select condition. Note that
	 * we will NOT block waiting for more input.
	 */
	if (pqReadData(conn) < 0)
		return 0;
	/* Parsing of the data waits till later. */
	return 1;
}


/*
 * parseInput: if appropriate, parse input data from backend
 * until input is exhausted or a stopping state is reached.
 * Note that this function will NOT attempt to read more data from the backend.
 */

static void
parseInput(PGconn *conn)
{
	char		id;

	/*
	 * Loop to parse successive complete messages available in the buffer.
	 */
	for (;;)
	{

		/*
		 * Quit if in COPY_OUT state: we expect raw data from the server
		 * until PQendcopy is called.  Don't try to parse it according to
		 * the normal protocol.  (This is bogus.  The data lines ought to
		 * be part of the protocol and have identifying leading
		 * characters.)
		 */
		if (conn->asyncStatus == PGASYNC_COPY_OUT)
			return;

		/*
		 * OK to try to read a message type code.
		 */
		conn->inCursor = conn->inStart;
		if (pqGetc(&id, conn))
			return;

		/*
		 * NOTIFY and NOTICE messages can happen in any state besides COPY
		 * OUT; always process them right away.
		 */
		if (id == 'A')
		{
			if (getNotify(conn))
				return;
		}
		else if (id == 'N')
		{
			if (getNotice(conn))
				return;
		}
		else
		{

			/*
			 * Other messages should only be processed while in BUSY
			 * state. (In particular, in READY state we hold off further
			 * parsing until the application collects the current
			 * PGresult.) If the state is IDLE then we got trouble.
			 */
			if (conn->asyncStatus != PGASYNC_BUSY)
			{
				if (conn->asyncStatus == PGASYNC_IDLE)
				{
					sprintf(conn->errorMessage,
					  "Backend message type 0x%02x arrived while idle\n",
							id);
					DONOTICE(conn, conn->errorMessage);
					/* Discard the unexpected message; good idea?? */
					conn->inStart = conn->inEnd;
				}
				return;
			}
			switch (id)
			{
				case 'C':		/* command complete */
					if (conn->result == NULL)
						conn->result = PQmakeEmptyPGresult(conn,
													   PGRES_COMMAND_OK);
					if (pqGets(conn->result->cmdStatus, CMDSTATUS_LEN, conn))
						return;
					conn->asyncStatus = PGASYNC_READY;
					break;
				case 'E':		/* error return */
					if (pqGets(conn->errorMessage, ERROR_MSG_LENGTH, conn))
						return;
					/* delete any partially constructed result */
					pqClearAsyncResult(conn);
					/* and build an error result holding the error message */
					conn->result = PQmakeEmptyPGresult(conn,
													   PGRES_FATAL_ERROR);
					conn->asyncStatus = PGASYNC_READY;
					break;
				case 'Z':		/* backend is ready for new query */
					conn->asyncStatus = PGASYNC_IDLE;
					break;
				case 'I':		/* empty query */
					/* read and throw away the closing '\0' */
					if (pqGetc(&id, conn))
						return;
					if (id != '\0')
					{
						sprintf(conn->errorMessage,
						  "unexpected character %c following 'I'\n", id);
						DONOTICE(conn, conn->errorMessage);
					}
					if (conn->result == NULL)
						conn->result = PQmakeEmptyPGresult(conn,
													  PGRES_EMPTY_QUERY);
					conn->asyncStatus = PGASYNC_READY;
					break;
				case 'K':		/* secret key data from the backend */

					/*
					 * This is expected only during backend startup, but
					 * it's just as easy to handle it as part of the main
					 * loop.  Save the data and continue processing.
					 */
					if (pqGetInt(&(conn->be_pid), 4, conn))
						return;
					if (pqGetInt(&(conn->be_key), 4, conn))
						return;
					break;
				case 'P':		/* synchronous (normal) portal */
					if (pqGets(conn->errorMessage, ERROR_MSG_LENGTH, conn))
						return;
					/* We pretty much ignore this message type... */
					break;
				case 'T':		/* row descriptions (start of query
								 * results) */
					if (conn->result == NULL)
					{
						/* First 'T' in a query sequence */
						if (getRowDescriptions(conn))
							return;
					}
					else
					{

						/*
						 * A new 'T' message is treated as the start of
						 * another PGresult.  (It is not clear that this
						 * is really possible with the current backend.)
						 * We stop parsing until the application accepts
						 * the current result.
						 */
						conn->asyncStatus = PGASYNC_READY;
						return;
					}
					break;
				case 'D':		/* ASCII data tuple */
					if (conn->result != NULL)
					{
						/* Read another tuple of a normal query response */
						if (getAnotherTuple(conn, FALSE))
							return;
					}
					else
					{
						sprintf(conn->errorMessage,
							 "Backend sent D message without prior T\n");
						DONOTICE(conn, conn->errorMessage);
						/* Discard the unexpected message; good idea?? */
						conn->inStart = conn->inEnd;
						return;
					}
					break;
				case 'B':		/* Binary data tuple */
					if (conn->result != NULL)
					{
						/* Read another tuple of a normal query response */
						if (getAnotherTuple(conn, TRUE))
							return;
					}
					else
					{
						sprintf(conn->errorMessage,
							 "Backend sent B message without prior T\n");
						DONOTICE(conn, conn->errorMessage);
						/* Discard the unexpected message; good idea?? */
						conn->inStart = conn->inEnd;
						return;
					}
					break;
				case 'G':		/* Start Copy In */
					conn->asyncStatus = PGASYNC_COPY_IN;
					break;
				case 'H':		/* Start Copy Out */
					conn->asyncStatus = PGASYNC_COPY_OUT;
					break;
				default:
					sprintf(conn->errorMessage,
					"unknown protocol character '%c' read from backend.  "
					"(The protocol character is the first character the "
							"backend sends in response to a query it receives).\n",
							id);
					/* Discard the unexpected message; good idea?? */
					conn->inStart = conn->inEnd;
					/* delete any partially constructed result */
					pqClearAsyncResult(conn);
					/* and build an error result holding the error message */
					conn->result = PQmakeEmptyPGresult(conn,
													   PGRES_FATAL_ERROR);
					conn->asyncStatus = PGASYNC_READY;
					return;
			}					/* switch on protocol character */
		}
		/* Successfully consumed this message */
		conn->inStart = conn->inCursor;
	}
}


/*
 * parseInput subroutine to read a 'T' (row descriptions) message.
 * We build a PGresult structure containing the attribute data.
 * Returns: 0 if completed message, EOF if not enough data yet.
 *
 * Note that if we run out of data, we have to release the partially
 * constructed PGresult, and rebuild it again next time.  Fortunately,
 * that shouldn't happen often, since 'T' messages usually fit in a packet.
 */

static int
getRowDescriptions(PGconn *conn)
{
	PGresult   *result;
	int			nfields;
	int			i;

	result = PQmakeEmptyPGresult(conn, PGRES_TUPLES_OK);

	/* parseInput already read the 'T' label. */
	/* the next two bytes are the number of fields	*/
	if (pqGetInt(&(result->numAttributes), 2, conn))
	{
		PQclear(result);
		return EOF;
	}
	nfields = result->numAttributes;

	/* allocate space for the attribute descriptors */
	if (nfields > 0)
	{
		result->attDescs = (PGresAttDesc *)
			pqResultAlloc(result, nfields * sizeof(PGresAttDesc), TRUE);
		MemSet((char *) result->attDescs, 0, nfields * sizeof(PGresAttDesc));
	}

	/* get type info */
	for (i = 0; i < nfields; i++)
	{
		char		typName[MAX_MESSAGE_LEN];
		int			typid;
		int			typlen;
		int			atttypmod;

		if (pqGets(typName, MAX_MESSAGE_LEN, conn) ||
			pqGetInt(&typid, 4, conn) ||
			pqGetInt(&typlen, 2, conn) ||
			pqGetInt(&atttypmod, 4, conn))
		{
			PQclear(result);
			return EOF;
		}

		/*
		 * Since pqGetInt treats 2-byte integers as unsigned, we need to
		 * coerce the special value "-1" to signed form.  (-1 is sent for
		 * variable-length fields.)  Formerly, libpq effectively did a
		 * sign-extension on the 2-byte value by storing it in a signed
		 * short. Now we only coerce the single value 65535 == -1; values
		 * 32768..65534 are taken as valid field lengths.
		 */
		if (typlen == 0xFFFF)
			typlen = -1;
		result->attDescs[i].name = pqResultStrdup(result, typName);
		result->attDescs[i].typid = typid;
		result->attDescs[i].typlen = typlen;
		result->attDescs[i].atttypmod = atttypmod;
	}

	/* Success! */
	conn->result = result;
	return 0;
}

/*
 * parseInput subroutine to read a 'B' or 'D' (row data) message.
 * We add another tuple to the existing PGresult structure.
 * Returns: 0 if completed message, EOF if error or not enough data yet.
 *
 * Note that if we run out of data, we have to suspend and reprocess
 * the message after more data is received.  We keep a partially constructed
 * tuple in conn->curTuple, and avoid reallocating already-allocated storage.
 */

static int
getAnotherTuple(PGconn *conn, int binary)
{
	PGresult   *result = conn->result;
	int			nfields = result->numAttributes;
	PGresAttValue *tup;
	char		bitmap[MAX_FIELDS];		/* the backend sends us a bitmap
										 * of which attributes are null */
	int			i;
	int			nbytes;			/* the number of bytes in bitmap  */
	char		bmap;			/* One byte of the bitmap */
	int			bitmap_index;	/* Its index */
	int			bitcnt;			/* number of bits examined in current byte */
	int			vlen;			/* length of the current field value */

	result->binary = binary;

	/* Allocate tuple space if first time for this data message */
	if (conn->curTuple == NULL)
	{
		conn->curTuple = (PGresAttValue *)
			pqResultAlloc(result, nfields * sizeof(PGresAttValue), TRUE);
		if (conn->curTuple == NULL)
			goto outOfMemory;
		MemSet((char *) conn->curTuple, 0, nfields * sizeof(PGresAttValue));
	}
	tup = conn->curTuple;

	/* Get the null-value bitmap */
	nbytes = (nfields + BYTELEN - 1) / BYTELEN;
	if (nbytes >= MAX_FIELDS)
	{
		/* Replace partially constructed result with an error result */
		pqClearAsyncResult(conn);
		sprintf(conn->errorMessage,
				"getAnotherTuple() -- null-values bitmap is too large\n");
		conn->result = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR);
		conn->asyncStatus = PGASYNC_READY;
		/* Discard the broken message */
		conn->inStart = conn->inEnd;
		return EOF;
	}

	if (pqGetnchar(bitmap, nbytes, conn))
		return EOF;

	/* Scan the fields */
	bitmap_index = 0;
	bmap = bitmap[bitmap_index];
	bitcnt = 0;

	for (i = 0; i < nfields; i++)
	{
		if (!(bmap & 0200))
		{
			/* if the field value is absent, make it a null string */
			tup[i].value = result->null_field;
			tup[i].len = NULL_LEN;
		}
		else
		{
			/* get the value length (the first four bytes are for length) */
			if (pqGetInt(&vlen, 4, conn))
				return EOF;
			if (binary == 0)
				vlen = vlen - 4;
			if (vlen < 0)
				vlen = 0;
			if (tup[i].value == NULL)
			{
				tup[i].value = (char *) pqResultAlloc(result, vlen + 1, binary);
				if (tup[i].value == NULL)
					goto outOfMemory;
			}
			tup[i].len = vlen;
			/* read in the value */
			if (vlen > 0)
				if (pqGetnchar((char *) (tup[i].value), vlen, conn))
					return EOF;
			/* we have to terminate this ourselves */
			tup[i].value[vlen] = '\0';
		}
		/* advance the bitmap stuff */
		bitcnt++;
		if (bitcnt == BYTELEN)
		{
			bitmap_index++;
			bmap = bitmap[bitmap_index];
			bitcnt = 0;
		}
		else
			bmap <<= 1;
	}

	/* Success!  Store the completed tuple in the result */
	if (!addTuple(result, tup))
		goto outOfMemory;
	/* and reset for a new message */
	conn->curTuple = NULL;
	return 0;

outOfMemory:
	/* Replace partially constructed result with an error result */
	pqClearAsyncResult(conn);
	sprintf(conn->errorMessage,
			"getAnotherTuple() -- out of memory for result\n");
	conn->result = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR);
	conn->asyncStatus = PGASYNC_READY;
	/* Discard the failed message --- good idea? */
	conn->inStart = conn->inEnd;
	return EOF;
}


/*
 * PQisBusy
 *	 Return TRUE if PQgetResult would block waiting for input.
 */

int
PQisBusy(PGconn *conn)
{
	if (!conn)
		return FALSE;

	/* Parse any available data, if our state permits. */
	parseInput(conn);

	/* PQgetResult will return immediately in all states except BUSY. */
	return conn->asyncStatus == PGASYNC_BUSY;
}


/*
 * PQgetResult
 *	  Get the next PGresult produced by a query.
 *	  Returns NULL if and only if no query work remains.
 */

PGresult   *
PQgetResult(PGconn *conn)
{
	PGresult   *res;

	if (!conn)
		return NULL;

	/* Parse any available data, if our state permits. */
	parseInput(conn);

	/* If not ready to return something, block until we are. */
	while (conn->asyncStatus == PGASYNC_BUSY)
	{
		/* Wait for some more data, and load it. */
		if (pqWait(TRUE, FALSE, conn) ||
			pqReadData(conn) < 0)
		{
			pqClearAsyncResult(conn);
			conn->asyncStatus = PGASYNC_IDLE;
			/* conn->errorMessage has been set by pqWait or pqReadData. */
			return PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR);