Add implement for ic interface GetMotionSentRecordTypmod()

Author: zhangyue1818

contrib/udp2/ic_common/udp2/ic_udp2.cpp

@@ -4587,6 +4587,7 @@ UDPConn::UDPConn(TransportEntry *entry)
 	this->remoteContentId      = -1;
 	this->remoteHostAndPort[0] = '\0';
 	this->opaque_data = nullptr;
+	this->sent_record_typmod = 0;
 
 	/* the field of UDPConn */
 	this->capacity  = -1;
@@ -4621,6 +4622,7 @@ UDPConn::UDPConn(TransportEntry *entry)
 	this->stopRequested = false;
 	this->cdbProc = NULL;
 	this->opaque_data = NULL;
+	this->sent_record_typmod = 0;
 
 	/*
 	 * "UDPConn dummyconn(NULL)" will be called by handleMismatch() in rx thread,
@@ -6500,6 +6502,14 @@ CChunkTransportStateImpl::GetOpaqueDataInConn(int16 motNodeID, int16 targetRoute
 	return conn->opaque_data;
 }
 
+int32*
+CChunkTransportStateImpl::GetSentRecordTypmodInConn(int16 motNodeID, int16 targetRoute)
+{
+	TransportEntry *pEntry = this->GetEntry(motNodeID, true);
+	UDPConn *conn = pEntry->GetConn(targetRoute);
+	return &conn->sent_record_typmod;
+}
+
 int
 CChunkTransportStateImpl::GetConnNum(int motNodeID) {
 	TransportEntry *pEntry = this->GetEntry(motNodeID, true);
@@ -6587,6 +6597,11 @@ CChunkTransportState::GetOpaqueDataInConn(int16 motNodeID, int16 targetRoute) {
 	return ToDerived(this)->GetOpaqueDataInConn(motNodeID, targetRoute);
 }
 
+int32*
+CChunkTransportState::GetSentRecordTypmodInConn(int16 motNodeID, int16 targetRoute) {
+	return ToDerived(this)->GetSentRecordTypmodInConn(motNodeID, targetRoute);
+}
+
 int
 CChunkTransportState::GetConnNum(int motNodeID) {
 	return ToDerived(this)->GetConnNum(motNodeID);
@@ -6848,6 +6863,23 @@ UDP2_GetOpaqueDataInConn(ICChunkTransportState *transportStates,
 	return NULL;
 }
 
+int32*
+UDP2_GetSentRecordTypmodInConn(ICChunkTransportState *transportStates,
+								int16 motNodeID,
+								int16 targetRoute)
+{
+	Assert(transportStates);
+
+	try {
+		CChunkTransportState *cstate = static_cast<CChunkTransportState*>(transportStates);
+		return cstate->GetSentRecordTypmodInConn(motNodeID, targetRoute);
+	} catch (...) {
+		handleException();
+	}
+
+	return NULL;
+}
+
 void
 UDP2_InitUDPIFC(GlobalMotionLayerIPCParam *param)
 {

contrib/udp2/ic_common/udp2/ic_udp2.h

@@ -73,6 +73,9 @@ extern void UDP2_ReduceFreeSpace(ICChunkTransportState *transportStates,
 extern void* UDP2_GetOpaqueDataInConn(ICChunkTransportState *transportStates,
 									  int16 motNodeID,
 									  int16 targetRoute);
+extern int32* UDP2_GetSentRecordTypmodInConn(ICChunkTransportState *transportStates,
+									  int16 motNodeID,
+									  int16 targetRoute);
 
 extern uint32 UDP2_GetActiveConns(void);
 extern int    UDP2_GetICHeaderSizeUDP(void);

contrib/udp2/ic_common/udp2/ic_udp2.hpp

@@ -40,6 +40,8 @@ struct CChunkTransportState : public ICChunkTransportState
 	void ReduceFreeSpace(int16 motNodeID, int16 targetRoute, int length);
 
 	void* GetOpaqueDataInConn(int16 motNodeID, int16 targetRoute);
+	int32* GetSentRecordTypmodInConn(int16 motNodeID, int16 targetRoute);
+
 	int  GetConnNum(int motNodeID);
 
 	static CChunkTransportState** GetTransportState();

contrib/udp2/ic_common/udp2/ic_udp2_internal.hpp

@@ -114,6 +114,7 @@ typedef struct MotionConn
 										 * play it safe */
 
 	void *opaque_data;
+	int32 sent_record_typmod;
 
 } MotionConn;
 
@@ -1159,6 +1160,8 @@ class CChunkTransportStateImpl : public CChunkTransportState
 	void ReduceFreeSpace(int16 motNodeID, int16 targetRoute, int length);
 
 	void* GetOpaqueDataInConn(int16 motNodeID, int16 targetRoute);
+	int32* GetSentRecordTypmodInConn(int16 motNodeID, int16 targetRoute);
+
 	int  GetConnNum(int motNodeID);
 
 	TransportEntry* GetEntry(int motNodeID, bool checkValid);

contrib/udp2/ic_modules.c

@@ -54,6 +54,7 @@ MotionIPCLayer udp2_ipc_layer = {
 #endif
 
     .GetMotionConnTupleRemapper = GetMotionConnTupleRemapperUDPIFC2,
+    .GetMotionSentRecordTypmod  = GetMotionSentRecordTypmodUDPIFC2,
 };
 
 void

contrib/udp2/ic_udp2.c

@@ -508,6 +508,32 @@ GetMotionConnTupleRemapperUDPIFC2(ChunkTransportState * transportStates,
 	return remapper;
 }
 
+int32*
+GetMotionSentRecordTypmodUDPIFC2(ChunkTransportState * transportStates,
+								 int16 motNodeID,
+								 int16 targetRoute)
+{
+	int32 *rs = NULL;
+
+	if (!transportStates)
+	{
+		elog(FATAL, "GetMotionConnTupleRemapper: no transport states");
+	}
+
+	if (!transportStates->activated)
+	{
+		elog(FATAL, "GetMotionConnTupleRemapper: inactive transport states");
+	}
+
+	ResetLastError();
+	ICChunkTransportState *udp2_state =
+		(ICChunkTransportState *)transportStates->implement_state;
+	rs = UDP2_GetSentRecordTypmodInConn(udp2_state, motNodeID, targetRoute);
+	HandleLastError();
+
+	return rs;
+}
+
 static void
 SetupGlobalMotionLayerIPCParam(GlobalMotionLayerIPCParam *param)
 {

contrib/udp2/ic_udp2.h

@@ -62,6 +62,10 @@ extern TupleRemapper* GetMotionConnTupleRemapperUDPIFC2(ChunkTransportState *tra
 														int16 motNodeID,
 														int16 targetRoute);
 
+extern int32* GetMotionSentRecordTypmodUDPIFC2(ChunkTransportState * transportStates,
+								 				int16 motNodeID,
+								 				int16 targetRoute);
+
 extern int ic_proxy_server_main(void);
 
 #endif // IC_UDP_H