pgclient.cpp

通用的数据库中间库

	{
		if(*vp == '\'' )
		{
			*rp++ = '\\';
			*rp++ = *vp;
		}
		else if(*vp == 0 )
		{
			*rp++ = '\\';
			*rp++ = '\\';
			*rp++ = '0';
			*rp++ = '0';
			*rp++ = '0';
		}
		else if( *vp == '\\' )
		{
			*rp++ = '\\';
			*rp++ = '\\';
			*rp++ = '\\';
			*rp++ = '\\';
		}
		else if( isprint((unsigned char) *vp) )
			*rp++ = *vp;
		else
		{
			val = *vp;
			rp[0] = '\\';
			rp[3] = (char)DIG(val & 07);
			val >>= 3;
			rp[2] = (char)DIG(val & 07);
			val >>= 3;
			rp[1] = (char)DIG(val & 03);
			rp += 4;
		}
	}

	*rp = '\0';
	return result;
}

/*static*/
void* IpgConnection::string2byte(const char* szInputText, int &nOutBufLen)
{
	char *tp = (char*)szInputText;

	for( nOutBufLen = 0; *tp != '\0'; nOutBufLen++ )
	{
		if( *tp++ == '\\' )
		{ 
			if( *tp == '\\' )
				tp++;
			else if( !isdigit((unsigned char) *tp++)
				|| !isdigit((unsigned char) *tp++)
				|| !isdigit((unsigned char) *tp++) )
				throw SAException(SA_Library_Error, -1, -1, "Bad input string for type bytea");
		}
	}

	tp = (char*)szInputText;
	void *result = malloc(nOutBufLen);
	char *rp = (char*)result;
	int byte;

	while( *tp != '\0' )
	{
		if (*tp != '\\' || *++tp == '\\')
			*rp++ = *tp++;
		else
		{
			byte = VAL(*tp++);
			byte <<= 3;
			byte += VAL(*tp++);
			byte <<= 3;
			*rp++ = (char)(byte + VAL(*tp++));
		}
	}

	return result;
}

/*virtual */
void IpgCursor::Cancel()
{
	pgConnectionHandles *pConH = (pgConnectionHandles *)m_pCommand->Connection()->NativeHandles();

	if( !g_pgAPI.PQrequestCancel(pConH->conn) )
		throw SAException(SA_RDBMS_API_Error, 0, -1, g_pgAPI.PQerrorMessage(pConH->conn));
}

/*virtual */
bool IpgCursor::ResultSetExists()
{
	if(!m_bResultSetCanBe)
		return false;

	return (g_pgAPI.PQresultStatus(m_handles.res) == PGRES_TUPLES_OK);
}

/*virtual */
void IpgCursor::DescribeFields(
	DescribeFields_cb_t fn)
{
	SAString s = m_pCommand->Option("OidTypeInterpretation");
	bool bOidAsBlob = s.CompareNoCase("LargeObject") == 0;

	int cFields = g_pgAPI.PQnfields(m_handles.res);

	for(int iField = 0; iField < cFields; ++iField)
	{
		Oid NativeType = g_pgAPI.PQftype(m_handles.res, iField);
		int Length = g_pgAPI.PQfsize(m_handles.res, iField);
		int Mod = g_pgAPI.PQfmod(m_handles.res, iField);

		(m_pCommand->*fn)(
			g_pgAPI.PQfname(m_handles.res, iField),
			IpgConnection::CnvtNativeToStd(NativeType, Length, Mod, bOidAsBlob),
			NativeType,
			Length,
			0,
			0,
			false);
	}
}

/*virtual */
void IpgCursor::SetSelectBuffers()
{
	// do nothing
}

/*virtual */
void IpgCursor::SetFieldBuffer(
	int/* nCol*/,	// 1-based
	void * /*pInd*/,
	unsigned int/* nIndSize*/,
	void * /*pSize*/,
	unsigned int/* nSizeSize*/,
	void * /*pValue*/,
	unsigned int/* nValueSize*/)
{
	assert(false);	// we do not use standard allocators
}

void IpgCursor::ConvertPGTupleToFields(int nTuple)
{
	int cFields = m_pCommand->FieldCount();
	for(int iField = 0; iField < cFields; ++iField)
	{
		SAField &Field = m_pCommand->Field(iField+1);
		SADataType_t eFieldType = Field.FieldType();


		if( g_pgAPI.PQgetisnull(m_handles.res, nTuple, iField) )
		{
			*Field.m_pbNull = true;
			continue;
		}
		*Field.m_pbNull = false;

		char *sValue = g_pgAPI.PQgetvalue(m_handles.res, nTuple, iField);
		int nRealSize = g_pgAPI.PQgetlength(m_handles.res, nTuple, iField);

		if( g_pgAPI.PQbinaryTuples(m_handles.res) )
		{
			Field.m_eDataType = SA_dtBytes;
			*Field.m_pString = SAString((void*)sValue, nRealSize);
			continue;
		}

		switch(eFieldType)
		{
		case SA_dtUnknown:
			throw SAException(SA_Library_Error, -1, -1, IDS_UNKNOWN_DATA_TYPE);
		case SA_dtBool: // Output as 't' or 'f' char
			Field.m_eDataType = SA_dtBool;
			if( sValue[0] == 't' )
				*(bool*)Field.m_pScalar = true;
			else if( sValue[0] == 'f' )
				*(bool*)Field.m_pScalar = false;
			else
				assert(false);
			break;
		case SA_dtShort:
			Field.m_eDataType = SA_dtShort;
			*(short*)Field.m_pScalar = (short)atol(sValue);
			break;
		case SA_dtLong:
			Field.m_eDataType = SA_dtLong;
			*(long*)Field.m_pScalar = atol(sValue);
			break;
		case SA_dtDouble:
			if( Field.FieldNativeType() == CASHOID ) // $d,ddd.cc, money
			{
				size_t i, out = 0;
				for( i = 0; i < strlen(sValue); i++ )
				{
					if(sValue[i] == '$' || sValue[i] == ',' || sValue[i] == ')' )
						;					/* skip these characters */
					else if( sValue[i] == '(' )
						sValue[out++] = '-';
					else
						sValue[out++] = sValue[i];
				}
				sValue[out] = '\0';				
			}
			Field.m_eDataType = SA_dtDouble;
			*(double*)Field.m_pScalar = atof(sValue);
			break;
		case SA_dtDateTime:
			Field.m_eDataType = SA_dtDateTime;
			IpgConnection::CnvtInternalToDateTime(
				*Field.m_pDateTime,
				sValue);
			break;
		case SA_dtString:
			Field.m_eDataType = SA_dtString;
			*Field.m_pString =
				SAString(sValue, nRealSize);
			break;
		case SA_dtBytes:
			{
				Field.m_eDataType = SA_dtBytes;
				int nByteSize = 0;
				void* pBytes = IpgConnection::string2byte(sValue, nByteSize);
				*Field.m_pString = SAString(pBytes, nByteSize);
				free(pBytes);
			}
			break;
		case SA_dtLongBinary:
			Field.m_eDataType = SA_dtLongBinary;
			*(long*)Field.m_pScalar = nTuple; // Pass tuple id
			break;
		case SA_dtLongChar:
			Field.m_eDataType = SA_dtLongChar;
			*(long*)Field.m_pScalar = nTuple; // Pass tuple id
			break;
		case SA_dtBLob:
			Field.m_eDataType = SA_dtBLob;
			*(long*)Field.m_pScalar = atol(sValue); // Pass Oid of the large object (should be LOB!!!) as long
			break;
		case SA_dtCLob:
			Field.m_eDataType = SA_dtCLob;
			*(long*)Field.m_pScalar = atol(sValue); // Pass Oid of the large object (should be LOB!!!) as long
			break;
		default:
			assert(false);	// unknown type
		}

		if(isLongOrLob(eFieldType) )
			ConvertLongOrLOB(ISA_FieldValue, Field, NULL, 0);
	}
}

/*virtual */
bool IpgCursor::FetchNext()
{
	if(m_nCurrentTuple < m_nTuplesCount)
		ConvertPGTupleToFields(m_nCurrentTuple++);
	else
		m_bResultSetCanBe = false;

	return m_bResultSetCanBe;
}

/*virtual */
long IpgCursor::GetRowsAffected()
{
	return (long)atoi(g_pgAPI.PQcmdTuples(m_handles.res));
}

/*virtual */
void IpgCursor::ReadLongOrLOB(
	ValueType_t eValueType,
	SAValueRead &vr,
	void *pValue,
	unsigned int nBufSize,
	saLongOrLobReader_t fnReader,
	unsigned int nReaderWantedPieceSize,
	void *pAddlData)
{
	SADataType_t eDataType;

	switch( eValueType )
	{
	case ISA_FieldValue:
		eDataType = ((SAField&)vr).FieldType();
		break;
	default:
		assert(eValueType == ISA_ParamValue);
		eDataType = ((SAParam&)vr).ParamType();
	}

	switch( eDataType )
	{
	case SA_dtLongBinary:
		ReadLongBinary(eValueType, vr, pValue, nBufSize, fnReader, nReaderWantedPieceSize, pAddlData);
		break;
	case SA_dtLongChar:
		ReadLongChar(eValueType, vr, pValue, nBufSize, fnReader, nReaderWantedPieceSize, pAddlData);
		break;
	case SA_dtBLob:
	case SA_dtCLob:
		ReadBLOB(eValueType, vr, pValue, nBufSize, fnReader, nReaderWantedPieceSize, pAddlData);
		break;
	default:
		assert(false);
	}
}

void IpgCursor::ReadLongBinary(
	ValueType_t/* eValueType*/,
	SAValueRead &vr,
	void * /*pValue*/,
	unsigned int/* nFieldBufSize*/,
	saLongOrLobReader_t fnReader,
	unsigned int nReaderWantedPieceSize,
	void *pAddlData)
{
	int iField = ((SAField &)vr).Pos() - 1;
	int nTuple = *((int*)vr.m_pScalar);

	// get long data and size
	int nRealSize;
	char *pData = (char*)IpgConnection::string2byte(
		g_pgAPI.PQgetvalue(m_handles.res, nTuple, iField), nRealSize);
	unsigned long nLongSize = nRealSize;

	unsigned char* pBuf;
	unsigned int nPieceSize = vr.PrepareReader(
		nLongSize,
		IpgConnection::MaxLongPiece,
		pBuf,
		fnReader,
		nReaderWantedPieceSize,
		pAddlData);
	assert(nPieceSize <= IpgConnection::MaxLongPiece);

	SAPieceType_t ePieceType = SA_FirstPiece;
	unsigned long nTotalRead = 0;
	do
	{
		nPieceSize =
			sa_min(nPieceSize, nLongSize - nTotalRead);

		memcpy(pBuf, pData + nTotalRead, nPieceSize);
		unsigned int actual_read = nPieceSize;

		nTotalRead += actual_read;
		assert(nTotalRead <= nLongSize);

        if(nTotalRead == nLongSize)
        {
            if(ePieceType == SA_NextPiece)
                ePieceType = SA_LastPiece;
            else    // the whole Long was read in one piece
            {
                assert(ePieceType == SA_FirstPiece);
                ePieceType = SA_OnePiece;
            }
        }
		vr.InvokeReader(ePieceType, pBuf, actual_read);

		if(ePieceType == SA_FirstPiece)
			ePieceType = SA_NextPiece;
	}
	while(nTotalRead < nLongSize);

	free(pData);
}

void IpgCursor::ReadLongChar(
	ValueType_t/* eValueType*/,
	SAValueRead &vr,
	void * /*pValue*/,
	unsigned int/* nFieldBufSize*/,
	saLongOrLobReader_t fnReader,
	unsigned int nReaderWantedPieceSize,
	void *pAddlData)
{
	int iField = ((SAField &)vr).Pos() - 1;
	int nTuple = *((int*)vr.m_pScalar);

	// get long data and size
	const char *pData = g_pgAPI.PQgetvalue(m_handles.res, nTuple, iField);
	unsigned long nLongSize = g_pgAPI.PQgetlength(m_handles.res, nTuple, iField);

	unsigned char* pBuf;
	unsigned int nPieceSize = vr.PrepareReader(
		nLongSize,
		IpgConnection::MaxLongPiece,
		pBuf,
		fnReader,
		nReaderWantedPieceSize,
		pAddlData);
	assert(nPieceSize <= IpgConnection::MaxLongPiece);

	SAPieceType_t ePieceType = SA_FirstPiece;
	unsigned long nTotalRead = 0;
	do
	{
		nPieceSize =
			sa_min(nPieceSize, nLongSize - nTotalRead);

		memcpy(pBuf, pData + nTotalRead, nPieceSize);
		unsigned int actual_read = nPieceSize;

		nTotalRead += actual_read;
		assert(nTotalRead <= nLongSize);

        if(nTotalRead == nLongSize)
        {
            if(ePieceType == SA_NextPiece)
                ePieceType = SA_LastPiece;
            else    // the whole Long was read in one piece
            {
                assert(ePieceType == SA_FirstPiece);
                ePieceType = SA_OnePiece;
            }
        }
		vr.InvokeReader(ePieceType, pBuf, actual_read);

		if(ePieceType == SA_FirstPiece)
			ePieceType = SA_NextPiece;
	}
	while(nTotalRead < nLongSize);
}

void IpgCursor::ReadBLOB(
	ValueType_t/* eValueType*/,
	SAValueRead &vr,
	void * /*pValue*/,
	unsigned int/* nFieldBufSize*/,
	saLongOrLobReader_t fnReader,
	unsigned int nReaderWantedPieceSize,
	void *pAddlData)
{
	pgConnectionHandles *pConH = (pgConnectionHandles *)m_pCommand->Connection()->NativeHandles();
	Oid blobOid = *((long*)vr.m_pScalar);
	SAConnection *pSAConnection = m_pCommand->Connection();

	// Start transaction if not started
	if( pSAConnection->AutoCommit() != SA_AutoCommitOff )
		Check(g_pgAPI.PQexec(pConH->conn, "BEGIN"));

	int fd = g_pgAPI.lo_open(pConH->conn, blobOid, INV_READ);
    if( fd < 0 )
		throw SAException(SA_Library_Error, -1, -1, "lo_open -> negative number");

	unsigned int nLongSize = 0;	// no way to determine size before fetch
	unsigned char* pBuf;
	unsigned int nPieceSize = vr.PrepareReader(
		nLongSize,
		IpgConnection::MaxBlobPiece,
		pBuf,
		fnReader,
		nReaderWantedPieceSize,
		pAddlData);

	assert(nPieceSize <= IpgConnection::MaxBlobPiece);

	SAPieceType_t ePieceType = SA_FirstPiece;
	int nReaded;
	
	while( (nReaded = g_pgAPI.lo_read(pConH->conn, fd, (char*)pBuf, nPieceSize)) != 0 )
	{
	    if( nReaded < 0 )
		{
			g_pgAPI.lo_close(pConH->conn, fd);
			
			// Rollback transaction, if previos start it
			if( pSAConnection->AutoCommit() != SA_AutoCommitOff )
				g_pgAPI.PQexec(pConH->conn, "ROLLBACK");

			throw SAException(SA_Library_Error, -1, -1, "lo_read -> negative number");
		}

		if( ePieceType != SA_FirstPiece )
		{
			if( nReaded < (int)nPieceSize )
				ePieceType = SA_LastPiece;
			else
				ePieceType = SA_NextPiece;
		}
		else if( nReaded < (int)nPieceSize )
			ePieceType = SA_OnePiece;

		vr.InvokeReader(ePieceType, pBuf, nReaded);

		if(ePieceType == SA_FirstPiece)
			ePieceType = SA_NextPiece;
	}

	g_pgAPI.lo_close(pConH->conn, fd);

	// End transaction, if previos start it
	if( pSAConnection->AutoCommit() != SA_AutoCommitOff )
		Check(g_pgAPI.PQexec(pConH->conn, "END"));
}

/*virtual */
void IpgCursor::DescribeParamSP()
{
	assert(false);	// no SPs in PostgreSQL
}

/*virtual */
bool IpgCursor::FetchParamsNext()
{
	assert(false);	// no SPs in PostgreSQL
	return false;
}

/*virtual */
saCommandHandles *IpgCursor::NativeHandles()
{
	return &m_handles;
}

//////////////////////////////////////////////////////////////////////
// PostgreSQL globals
//////////////////////////////////////////////////////////////////////

ISAConnection *newIpgConnection(SAConnection *pSAConnection)
{
	return new IpgConnection(pSAConnection);
}