Adapt Batch and Cursor, clean unused now routines out

Author: aafemt

src/dsql/DsqlBatch.cpp

@@ -224,7 +224,7 @@ DsqlBatch* DsqlBatch::open(thread_db* tdbb, DsqlDmlRequest* req, IMessageMetadat
 	}
 
 	const dsql_msg* message = statement->getSendMsg();
-	if (! (inMetadata && message && req->parseMetadata(inMetadata, message->msg_parameters)))
+	if (! (inMetadata && message && message->msg_parameter > 0))
 	{
 		ERRD_post(Arg::Gds(isc_sqlerr) << Arg::Num(-901) <<
 				  Arg::Gds(isc_batch_param));
@@ -654,18 +654,22 @@ Firebird::IBatchCompletionState* DsqlBatch::execute(thread_db* tdbb)
 	// execute request
 	m_dsqlRequest->req_transaction = transaction;
 	Request* req = m_dsqlRequest->getRequest();
+	DsqlStatement* dStmt = m_dsqlRequest->getDsqlStatement();
 	fb_assert(req);
 
 	// prepare completion interface
 	AutoPtr<BatchCompletionState, SimpleDispose> completionState
 		(FB_NEW BatchCompletionState(m_flags & (1 << IBatch::TAG_RECORD_COUNTS), m_detailed));
 	AutoSetRestore<bool> batchFlag(&req->req_batch_mode, true);
-	const dsql_msg* message = m_dsqlRequest->getDsqlStatement()->getSendMsg();
+	const dsql_msg* sendMessage = dStmt->getSendMsg();
+	MessageNode* message = dStmt->getStatement()->messages[sendMessage->msg_number];
+	// map message to internal engine format
+	// Do it one time only to avoid parsing its metadata for every message
+	m_dsqlRequest->metadataToFormat(m_meta, message);
 	bool startRequest = true;
 
-	bool isExecBlock = m_dsqlRequest->getDsqlStatement()->getType() == DsqlStatement::TYPE_EXEC_BLOCK;
-	const auto receiveMessage = isExecBlock ? m_dsqlRequest->getDsqlStatement()->getReceiveMsg() : nullptr;
-	auto receiveMsgBuffer = isExecBlock ? m_dsqlRequest->req_msg_buffers[receiveMessage->msg_buffer_number] : nullptr;
+	bool isExecBlock = dStmt->getType() == DsqlStatement::TYPE_EXEC_BLOCK;
+	const dsql_msg* receiveMessage = isExecBlock ? dStmt->getReceiveMsg() : nullptr;
 
 	// process messages
 	ULONG remains;
@@ -721,25 +725,18 @@ Firebird::IBatchCompletionState* DsqlBatch::execute(thread_db* tdbb)
 				*id = newId;
 			}
 
-			// map message to internal engine format
-			// pass m_meta one time only to avoid parsing its metadata for every message
-			m_dsqlRequest->mapInOut(tdbb, false, message, start ? m_meta : nullptr, nullptr, data);
-			data += m_messageSize;
-			remains -= m_messageSize;
-
-			UCHAR* msgBuffer = m_dsqlRequest->req_msg_buffers[message->msg_buffer_number];
 			try
 			{
 				// runsend data to request and collect stats
 				ULONG before = req->req_records_inserted + req->req_records_updated +
 					req->req_records_deleted;
-				EXE_send(tdbb, req, message->msg_number, message->msg_length, msgBuffer);
+				EXE_send(tdbb, req, sendMessage->msg_number, m_messageSize, data);
 				ULONG after = req->req_records_inserted + req->req_records_updated +
 					req->req_records_deleted;
 				completionState->regUpdate(after - before);
 
 				if (isExecBlock)
-					EXE_receive(tdbb, req, receiveMessage->msg_number, receiveMessage->msg_length, receiveMsgBuffer);
+					EXE_receive(tdbb, req, receiveMessage->msg_number, receiveMessage->msg_length, nullptr); // We don't care about returned record
 			}
 			catch (const Exception& ex)
 			{
@@ -761,6 +758,9 @@ Firebird::IBatchCompletionState* DsqlBatch::execute(thread_db* tdbb)
 			}
 		}
 
+		data += m_messageSize;
+		remains -= m_messageSize;
+
 		UCHAR* alignedData = FB_ALIGN(data, m_alignment);
 		m_messages.remained(remains, alignedData - data);
 	}

src/dsql/DsqlCursor.cpp

@@ -28,6 +28,7 @@
 
 #include "../dsql/dsql_proto.h"
 #include "../dsql/DsqlCursor.h"
+#include "../dsql/StmtNodes.h"
 
 using namespace Firebird;
 using namespace Jrd;
@@ -36,7 +37,8 @@ static const char* const SCRATCH = "fb_cursor_";
 static const ULONG PREFETCH_SIZE = 65536; // 64 KB
 
 DsqlCursor::DsqlCursor(DsqlDmlRequest* req, ULONG flags)
-	: m_dsqlRequest(req), m_message(req->getDsqlStatement()->getReceiveMsg()),
+	: m_dsqlRequest(req), m_message(req->getDsqlStatement()->getReceiveMsg()->msg_number),
+	  m_messageLength(0),
 	  m_resultSet(NULL), m_flags(flags),
 	  m_space(req->getPool(), SCRATCH),
 	  m_state(BOS), m_eof(false), m_position(0), m_cachedCount(0)
@@ -163,7 +165,7 @@ int DsqlCursor::fetchAbsolute(thread_db* tdbb, UCHAR* buffer, SLONG position)
 	{
 		if (!m_eof)
 		{
-			cacheInput(tdbb);
+			cacheInput(tdbb, buffer);
 			fb_assert(m_eof);
 		}
 
@@ -248,7 +250,7 @@ void DsqlCursor::getInfo(thread_db* tdbb,
 			case IResultSet::INF_RECORD_COUNT:
 				if (isScrollable && !m_eof)
 				{
-					cacheInput(tdbb);
+					cacheInput(tdbb, nullptr);
 					fb_assert(m_eof);
 				}
 				response.insertInt(tag, isScrollable ? m_cachedCount : -1);
@@ -291,48 +293,62 @@ int DsqlCursor::fetchFromCache(thread_db* tdbb, UCHAR* buffer, FB_UINT64 positio
 {
 	if (position >= m_cachedCount)
 	{
-		if (m_eof || !cacheInput(tdbb, position))
+		if (m_eof || !cacheInput(tdbb, buffer, position))
 		{
 			m_state = EOS;
 			return 1;
 		}
 	}
 
 	fb_assert(position < m_cachedCount);
+	fb_assert(m_messageLength > 0); // At this point m_messageLength must be set by cacheInput
 
-	UCHAR* const msgBuffer = m_dsqlRequest->req_msg_buffers[m_message->msg_buffer_number];
-
-	const FB_UINT64 offset = position * m_message->msg_length;
-	const FB_UINT64 readBytes = m_space.read(offset, msgBuffer, m_message->msg_length);
-	fb_assert(readBytes == m_message->msg_length);
-
-	m_dsqlRequest->mapInOut(tdbb, true, m_message, NULL, buffer);
+	const FB_UINT64 offset = position * m_messageLength;
+	const FB_UINT64 readBytes = m_space.read(offset, buffer, m_messageLength);
+	fb_assert(readBytes == m_messageLength);
 
 	m_position = position;
 	m_state = POSITIONED;
 	return 0;
 }
 
-bool DsqlCursor::cacheInput(thread_db* tdbb, FB_UINT64 position)
+bool DsqlCursor::cacheInput(thread_db* tdbb, UCHAR* buffer, FB_UINT64 position)
 {
 	fb_assert(!m_eof);
 
-	const ULONG prefetchCount = MAX(PREFETCH_SIZE / m_message->msg_length, 1);
-	const UCHAR* const msgBuffer = m_dsqlRequest->req_msg_buffers[m_message->msg_buffer_number];
+	// It could not be done before: user buffer length may be unknown until call setDelayedOutputMetadata()
+	if (m_messageLength == 0)
+	{
+		Request* req = m_dsqlRequest->getRequest();
+		const MessageNode* msg = req->getStatement()->messages[m_message];
+		m_messageLength = msg->getFormat(req)->fmt_length;
+	}
+
+	std::unique_ptr<UCHAR[]> ownBuffer;
+	if (buffer == nullptr)
+	{
+		// We are called from getInfo() and there is no user-provided buffer for data.
+		// Create a temporary one.
+		// This code cannot be moved into getInfo() itself because it is most likely called before fetch()
+		// so m_messageLength is still unknown there.
+		ownBuffer.reset(buffer = new UCHAR[m_messageLength]);
+	}
+
+	const ULONG prefetchCount = MAX(PREFETCH_SIZE / m_messageLength, 1);
 
 	while (position >= m_cachedCount)
 	{
 		for (ULONG count = 0; count < prefetchCount; count++)
 		{
-			if (!m_dsqlRequest->fetch(tdbb, NULL))
+			if (!m_dsqlRequest->fetch(tdbb, buffer))
 			{
 				m_eof = true;
 				break;
 			}
 
-			const FB_UINT64 offset = m_cachedCount * m_message->msg_length;
-			const FB_UINT64 writtenBytes = m_space.write(offset, msgBuffer, m_message->msg_length);
-			fb_assert(writtenBytes == m_message->msg_length);
+			const FB_UINT64 offset = m_cachedCount * m_messageLength;
+			const FB_UINT64 writtenBytes = m_space.write(offset, buffer, m_messageLength);
+			fb_assert(writtenBytes == m_messageLength);
 			m_cachedCount++;
 		}
 

src/dsql/DsqlCursor.h

@@ -67,10 +67,11 @@ class DsqlCursor
 
 private:
 	int fetchFromCache(thread_db* tdbb, UCHAR* buffer, FB_UINT64 position);
-	bool cacheInput(thread_db* tdbb, FB_UINT64 position = MAX_UINT64);
+	bool cacheInput(thread_db* tdbb, UCHAR* buffer, FB_UINT64 position = MAX_UINT64);
 
 	DsqlDmlRequest* const m_dsqlRequest;
-	const dsql_msg* const m_message;
+	const USHORT m_message;
+	ULONG m_messageLength;
 	JResultSet* m_resultSet;
 	const ULONG m_flags;
 	TempSpace m_space;