static Source *
CreateRemoteSource(const char *path, TupleDesc desc)
{
	RemoteSource *self = (RemoteSource *) palloc0(sizeof(RemoteSource));
	self->base.close = (SourceCloseProc) RemoteSourceClose;

	if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 3)
	{
		/* new way */
		StringInfoData	buf;
		int16			format;
		int				nattrs;
		int				i;

		self->base.read = (SourceReadProc) RemoteSourceRead;

		/* count valid fields */
		for (nattrs = 0, i = 0; i < desc->natts; i++)
		{
			if (desc->attrs[i]->attisdropped)
				continue;
			nattrs++;
		}

		format = (IsBinaryCopy() ? 1 : 0);
		pq_beginmessage(&buf, 'G');
		pq_sendbyte(&buf, format);		/* overall format */
		pq_sendint(&buf, nattrs, 2);
		for (i = 0; i < nattrs; i++)
			pq_sendint(&buf, format, 2);		/* per-column formats */
		pq_endmessage(&buf);
		self->buffer = makeStringInfo();
	}
	else if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2)
	{
		self->base.read = (SourceReadProc) RemoteSourceReadOld;

		/* old way */
		if (IsBinaryCopy())
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			errmsg("COPY BINARY is not supported to stdout or from stdin")));
		pq_putemptymessage('G');
	}
	else
	{
		self->base.read = (SourceReadProc) RemoteSourceReadOld;

		/* very old way */
		if (IsBinaryCopy())
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			errmsg("COPY BINARY is not supported to stdout or from stdin")));
		pq_putemptymessage('D');
	}
	/* We *must* flush here to ensure FE knows it can send. */
	pq_flush();

	return (Source *) self;
}

Datum
hstore_send(PG_FUNCTION_ARGS)
{
	HStore	   *in = PG_GETARG_HS(0);
	int			i;
	int			count = HS_COUNT(in);
	char	   *base = STRPTR(in);
	HEntry	   *entries = ARRPTR(in);
	StringInfoData buf;

	pq_begintypsend(&buf);

	pq_sendint(&buf, count, 4);

	for (i = 0; i < count; i++)
	{
		int32		keylen = HS_KEYLEN(entries, i);

		pq_sendint(&buf, keylen, 4);
		pq_sendtext(&buf, HS_KEY(entries, base, i), keylen);
		if (HS_VALISNULL(entries, i))
		{
			pq_sendint(&buf, -1, 4);
		}
		else
		{
			int32		vallen = HS_VALLEN(entries, i);

			pq_sendint(&buf, vallen, 4);
			pq_sendtext(&buf, HS_VAL(entries, base, i), vallen);
		}
	}

	PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}

/*
 * SendRowDescriptionMessage --- send a RowDescription message to the frontend
 *
 * Notes: the TupleDesc has typically been manufactured by ExecTypeFromTL()
 * or some similar function; it does not contain a full set of fields.
 * The targetlist will be NIL when executing a utility function that does
 * not have a plan.  If the targetlist isn't NIL then it is a Query node's
 * targetlist; it is up to us to ignore resjunk columns in it.  The formats[]
 * array pointer might be NULL (if we are doing Describe on a prepared stmt);
 * send zeroes for the format codes in that case.
 */
void
SendRowDescriptionMessage(TupleDesc typeinfo, List *targetlist, int16 *formats)
{
	Form_pg_attribute *attrs = typeinfo->attrs;
	int			natts = typeinfo->natts;
	int			proto = PG_PROTOCOL_MAJOR(FrontendProtocol);
	int			i;
	StringInfoData buf;
	ListCell   *tlist_item = list_head(targetlist);

	pq_beginmessage(&buf, 'T'); /* tuple descriptor message type */
	pq_sendint(&buf, natts, 2); /* # of attrs in tuples */

	for (i = 0; i < natts; ++i)
	{
		Oid			atttypid = attrs[i]->atttypid;
		int32		atttypmod = attrs[i]->atttypmod;

		pq_sendstring(&buf, NameStr(attrs[i]->attname));
		/* column ID info appears in protocol 3.0 and up */
		if (proto >= 3)
		{
			/* Do we have a non-resjunk tlist item? */
			while (tlist_item &&
				   ((TargetEntry *) lfirst(tlist_item))->resjunk)
				tlist_item = lnext(tlist_item);
			if (tlist_item)
			{
				TargetEntry *tle = (TargetEntry *) lfirst(tlist_item);

				pq_sendint(&buf, tle->resorigtbl, 4);
				pq_sendint(&buf, tle->resorigcol, 2);
				tlist_item = lnext(tlist_item);
			}
			else
			{
				/* No info available, so send zeroes */
				pq_sendint(&buf, 0, 4);
				pq_sendint(&buf, 0, 2);
			}
		}
		/* If column is a domain, send the base type and typmod instead */
		atttypid = getBaseTypeAndTypmod(atttypid, &atttypmod);
		pq_sendint(&buf, (int) atttypid, sizeof(atttypid));
		pq_sendint(&buf, attrs[i]->attlen, sizeof(attrs[i]->attlen));
		/* typmod appears in protocol 2.0 and up */
		if (proto >= 2)
			pq_sendint(&buf, atttypmod, sizeof(atttypmod));
		/* format info appears in protocol 3.0 and up */
		if (proto >= 3)
		{
			if (formats)
				pq_sendint(&buf, formats[i], 2);
			else
				pq_sendint(&buf, 0, 2);
		}
	}
	pq_endmessage(&buf);
}

/*
 * Write relation attributes to the stream.
 */
static void
logicalrep_write_attrs(StringInfo out, Relation rel)
{
	TupleDesc	desc;
	int			i;
	uint16		nliveatts = 0;
	Bitmapset  *idattrs = NULL;
	bool		replidentfull;

	desc = RelationGetDescr(rel);

	/* send number of live attributes */
	for (i = 0; i < desc->natts; i++)
	{
		if (TupleDescAttr(desc, i)->attisdropped)
			continue;
		nliveatts++;
	}
	pq_sendint(out, nliveatts, 2);

	/* fetch bitmap of REPLICATION IDENTITY attributes */
	replidentfull = (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL);
	if (!replidentfull)
		idattrs = RelationGetIndexAttrBitmap(rel,
											 INDEX_ATTR_BITMAP_IDENTITY_KEY);

	/* send the attributes */
	for (i = 0; i < desc->natts; i++)
	{
		Form_pg_attribute att = TupleDescAttr(desc, i);
		uint8		flags = 0;

		if (att->attisdropped)
			continue;

		/* REPLICA IDENTITY FULL means all columns are sent as part of key. */
		if (replidentfull ||
			bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,
						  idattrs))
			flags |= LOGICALREP_IS_REPLICA_IDENTITY;

		pq_sendbyte(out, flags);

		/* attribute name */
		pq_sendstring(out, NameStr(att->attname));

		/* attribute type id */
		pq_sendint(out, (int) att->atttypid, sizeof(att->atttypid));

		/* attribute mode */
		pq_sendint(out, att->atttypmod, sizeof(att->atttypmod));
	}

	bms_free(idattrs);
}

static void
putEndLocationReply(XLogRecPtr *endLocation)
{
	StringInfoData buf;
	
	pq_beginmessage(&buf, 's');
	pq_sendint(&buf, endLocation->xlogid, 4);
	pq_sendint(&buf, endLocation->xrecoff, 4);
	pq_endmessage(&buf);
	pq_flush();
}

/*
 * Send a gpdb libpq message.
 */
void
sendQEDetails(void)
{
	StringInfoData buf;

	pq_beginmessage(&buf, 'w');
	pq_sendint(&buf, (int32) ICListenerPort, sizeof(int32));
	pq_sendint(&buf, sizeof(PG_VERSION_STR), sizeof(int32));
	pq_sendbytes(&buf, PG_VERSION_STR, sizeof(PG_VERSION_STR));
	pq_endmessage(&buf);
}

/*
 *		tintervalsend			- converts tinterval to binary format
 */
Datum
tintervalsend(PG_FUNCTION_ARGS)
{
	TimeInterval tinterval = PG_GETARG_TIMEINTERVAL(0);
	StringInfoData buf;

	pq_begintypsend(&buf);
	pq_sendint(&buf, tinterval->status, sizeof(tinterval->status));
	pq_sendint(&buf, tinterval->data[0], sizeof(tinterval->data[0]));
	pq_sendint(&buf, tinterval->data[1], sizeof(tinterval->data[1]));
	PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}

/*
 * Send a gpdb libpq message.
 */
void
sendQEDetails(void)
{
	StringInfoData buf;

	pq_beginmessage(&buf, 'w');
	pq_sendint(&buf, (int32) Gp_listener_port, sizeof(int32));			
	pq_sendint64(&buf, VmemTracker_GetMaxReservedVmemBytes());
	pq_sendint(&buf, sizeof(PG_VERSION_STR), sizeof(int32));
	pq_sendbytes(&buf, PG_VERSION_STR, sizeof(PG_VERSION_STR));
	pq_endmessage(&buf);
}

Datum
ipaddr_send(PG_FUNCTION_ARGS)
{
    IP_P arg1 = PG_GETARG_IP_P(0);
    StringInfoData buf;
	IP ip;
	int af = ip_unpack(arg1, &ip);

    pq_begintypsend(&buf);
	pq_sendbyte(&buf, af);
	pq_sendbyte(&buf, ip_maxbits(af));
	pq_sendbyte(&buf, 1);
	pq_sendbyte(&buf, ip_sizeof(af));

	switch (af)
	{
		case PGSQL_AF_INET:
			pq_sendint(&buf, ip.ip4, sizeof(IP4));
			break;

		case PGSQL_AF_INET6:
			pq_sendint64(&buf, ip.ip6.bits[0]);
			pq_sendint64(&buf, ip.ip6.bits[1]);
			break;
	}

    PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}

/*
 * Write relation description to the output stream.
 */
static void
pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel)
{
	const char *nspname;
	uint8		nspnamelen;
	const char *relname;
	uint8		relnamelen;
	Oid         relid;
    if (MtmIsFilteredTxn) { 
		return;
	}
	
	relid = RelationGetRelid(rel);
	pq_sendbyte(out, 'R');		/* sending RELATION */	
	pq_sendint(out, relid, sizeof relid); /* use Oid as relation identifier */
	
	nspname = get_namespace_name(rel->rd_rel->relnamespace);
	if (nspname == NULL)
		elog(ERROR, "cache lookup failed for namespace %u",
				 rel->rd_rel->relnamespace);
	nspnamelen = strlen(nspname) + 1;
	
	relname = NameStr(rel->rd_rel->relname);
	relnamelen = strlen(relname) + 1;
	
	pq_sendbyte(out, nspnamelen);		/* schema name length */
	pq_sendbytes(out, nspname, nspnamelen);
	
	pq_sendbyte(out, relnamelen);		/* table name length */
	pq_sendbytes(out, relname, relnamelen);
}

/*
 *		gpxloglocsend			- converts xlog location to binary format
 */
Datum
gpxloglocsend(PG_FUNCTION_ARGS)
{
	XLogRecPtr *xlogLoc = PG_GETARG_XLOGLOC(0);
	uint32		send_xlogid;
	uint32		send_xrecoff;
	StringInfoData buf;

	send_xlogid = xlogLoc->xlogid;
	send_xrecoff = xlogLoc->xrecoff;

	pq_begintypsend(&buf);
	pq_sendint(&buf, send_xlogid, sizeof(send_xlogid));
	pq_sendint(&buf, send_xrecoff, sizeof(send_xrecoff));
	PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}

/* ----------------
 *		AddQEWriterTransactionInfo - Add QE writer transction information.
 * ----------------
 */
void
AddQEWriterTransactionInfo(StringInfo buf)
{
	DistributedTransactionId	QEDistributedTransactionId;
	CommandId					QECommandId;
	bool						QEDirty;

	TransactionInformationQEWriter(&QEDistributedTransactionId, &QECommandId, &QEDirty);

	elog(DEBUG5,"QEWriterTransactionInfo: (DistributedTransactionId = %u, CommandId = %u, and Dirty = %s)",
	     QEDistributedTransactionId, QECommandId, (QEDirty ? "true" : "false"));

	pq_sendint(buf, QEDistributedTransactionId, 4);
	pq_sendint(buf, QECommandId, 4);
	pq_sendbyte(buf, (QEDirty ? 'T' : 'F'));
}

/*
 * Send NOTIFY message to my front end.
 */
static void
NotifyMyFrontEnd(char *relname, int32 listenerPID)
{
	if (whereToSendOutput == DestRemote)
	{
		StringInfoData buf;

		pq_beginmessage(&buf, 'A');
		pq_sendint(&buf, listenerPID, sizeof(int32));
		pq_sendstring(&buf, relname);
		if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 3)
		{
			/* XXX Add parameter string here later */
			pq_sendstring(&buf, "");
		}
		pq_endmessage(&buf);

		/*
		 * NOTE: we do not do pq_flush() here.	For a self-notify, it will
		 * happen at the end of the transaction, and for incoming notifies
		 * ProcessIncomingNotify will do it after finding all the notifies.
		 */
	}
	else
		elog(INFO, "NOTIFY for %s", relname);
}

/*
 * Write UPDATE to the output stream.
 */
void
logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple,
						HeapTuple newtuple)
{
	pq_sendbyte(out, 'U');		/* action UPDATE */

	Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
		   rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
		   rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);

	/* use Oid as relation identifier */
	pq_sendint(out, RelationGetRelid(rel), 4);

	if (oldtuple != NULL)
	{
		if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
			pq_sendbyte(out, 'O');	/* old tuple follows */
		else
			pq_sendbyte(out, 'K');	/* old key follows */
		logicalrep_write_tuple(out, rel, oldtuple);
	}

	pq_sendbyte(out, 'N');		/* new tuple follows */
	logicalrep_write_tuple(out, rel, newtuple);
}

/**
 *  svec_send - converts text to binary format
 */
Datum svec_send(PG_FUNCTION_ARGS)
{
	StringInfoData buf;
	SvecType *svec = PG_GETARG_SVECTYPE_P(0);
	SparseData sdata = sdata_from_svec(svec);

	pq_begintypsend(&buf);
	pq_sendint(&buf,sdata->type_of_data,sizeof(Oid));
	pq_sendint(&buf,sdata->unique_value_count,sizeof(int));
	pq_sendint(&buf,sdata->total_value_count,sizeof(int));
	pq_sendint(&buf,sdata->vals->len,sizeof(int));
	pq_sendint(&buf,sdata->index->len,sizeof(int));
	pq_sendbytes(&buf,sdata->vals->data,sdata->vals->len);
	pq_sendbytes(&buf,sdata->index->data,sdata->index->len);

	PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}

static void
SendResultDescriptionMessage(AttributeDefinition *attrs, int natts)
{
	int			proto = PG_PROTOCOL_MAJOR(FrontendProtocol);
	int			i;
	StringInfoData buf;

	pq_beginmessage(&buf, 'T'); /* tuple descriptor message type */
	pq_sendint(&buf, natts, 2);	/* # of attrs in tuples */

	for (i = 0; i < natts; ++i)
	{
		pq_sendstring(&buf, attrs[i].name);
		/* column ID info appears in protocol 3.0 and up */
		if (proto >= 3)
		{
			pq_sendint(&buf, 0, 4);
			pq_sendint(&buf, 0, 2);
		}
		/* If column is a domain, send the base type and typmod instead */
		pq_sendint(&buf, attrs[i].typid, sizeof(Oid));
		pq_sendint(&buf, attrs[i].typlen, sizeof(int16));
		/* typmod appears in protocol 2.0 and up */
		if (proto >= 2)
			pq_sendint(&buf, attrs[i].typmod, sizeof(int32));
		/* format info appears in protocol 3.0 and up */
		if (proto >= 3)
			pq_sendint(&buf, 0, 2);
	}

	pq_endmessage(&buf);
}

/*
 * Write BEGIN to the output stream.
 */
static void
pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
					  ReorderBufferTXN *txn)
{
	bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
	csn_t csn = MtmTransactionSnapshot(txn->xid);
	MTM_LOG2("%d: pglogical_write_begin XID=%d node=%d CSN=%ld recovery=%d", MyProcPid, txn->xid, MtmReplicationNodeId, csn, isRecovery);
	
	if (csn == INVALID_CSN && !isRecovery) { 
		MtmIsFilteredTxn = true;
	} else { 
		pq_sendbyte(out, 'B');		/* BEGIN */
		pq_sendint(out, MtmNodeId, 4);
		pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
		pq_sendint64(out, csn);
		MtmIsFilteredTxn = false;
	}
}

/*
 *		int2send			- converts int2 to binary format
 */
datum_t int2send(PG_FUNC_ARGS)
{
	int16 arg1 = ARG_INT16(0);
	struct string buf;

	pq_begintypsend(&buf);
	pq_sendint(&buf, arg1, sizeof(int16));
	RET_BYTEA_P(pq_endtypsend(&buf));
}

/*
 *		tidsend			- converts tid to binary format
 */
Datum
tidsend(PG_FUNCTION_ARGS)
{
	ItemPointer itemPtr = PG_GETARG_ITEMPOINTER(0);
	BlockId		blockId;
	BlockNumber blockNumber;
	OffsetNumber offsetNumber;
	StringInfoData buf;

	blockId = &(itemPtr->ip_blkid);
	blockNumber = BlockIdGetBlockNumber(blockId);
	offsetNumber = itemPtr->ip_posid;

	pq_begintypsend(&buf);
	pq_sendint(&buf, blockNumber, sizeof(blockNumber));
	pq_sendint(&buf, offsetNumber, sizeof(offsetNumber));
	PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}

/*
 *		oidsend			- converts oid to binary format
 */
Datum
oidsend(PG_FUNCTION_ARGS)
{
	Oid			arg1 = PG_GETARG_OID(0);
	StringInfoData buf;

	pq_begintypsend(&buf);
	pq_sendint(&buf, arg1, sizeof(Oid));
	PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}