src/backend/replication/logical/proto.c
/*
 * Write a tuple to the outputstream, in the most efficient format possible.
 */
Write a tuple to the outputstream, in the most efficient format possible.
Write a tuple to the outputstream in the most efficient format possible .
static void
logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binary)
{
	TupleDesc	desc;
	Datum		values[MaxTupleAttributeNumber];
	bool		isnull[MaxTupleAttributeNumber];
	int			i;
	uint16		nliveatts = 0;
	desc = RelationGetDescr(rel);
	for (i = 0; i < desc->natts; i++)
	{
		if (TupleDescAttr(desc, i)->attisdropped || TupleDescAttr(desc, i)->attgenerated)
			continue;
		nliveatts++;
	}
	pq_sendint16(out, nliveatts);
	enlargeStringInfo(out, tuple->t_len +
					  nliveatts * (1 + 4));
	heap_deform_tuple(tuple, desc, values, isnull);
	for (i = 0; i < desc->natts; i++)
	{
		HeapTuple	typtup;
		Form_pg_type typclass;
		Form_pg_attribute att = TupleDescAttr(desc, i);
		char	   *outputstr;
		if (att->attisdropped || att->attgenerated)
			continue;
		if (isnull[i])
		{
			pq_sendbyte(out, LOGICALREP_COLUMN_NULL);
			continue;
		}
		if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(values[i]))
		{
			pq_sendbyte(out, LOGICALREP_COLUMN_UNCHANGED);
			continue;
		}
		typtup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(att->atttypid));
		if (!HeapTupleIsValid(typtup))
			elog(ERROR, "cache lookup failed for type %u", att->atttypid);
		typclass = (Form_pg_type) GETSTRUCT(typtup);
		if (binary && OidIsValid(typclass->typsend))
		{
			bytea	   *outputbytes;
			int			len;
			pq_sendbyte(out, LOGICALREP_COLUMN_BINARY);
			outputbytes = OidSendFunctionCall(typclass->typsend, values[i]);
			len = VARSIZE(outputbytes) - VARHDRSZ;
			pq_sendint(out, len, 4);	
			pq_sendbytes(out, VARDATA(outputbytes), len);	
			pfree(outputbytes);
		}
		else
		{
			pq_sendbyte(out, LOGICALREP_COLUMN_TEXT);
			outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]);
			pq_sendcountedtext(out, outputstr, strlen(outputstr), false);
			pfree(outputstr);
		}
		ReleaseSysCache(typtup);
	}
}
static void logicalrep_write_tuple ( StringInfo out , Relation rel , HeapTuple tuple , bool binary ) { TupleDesc desc ; Datum values [ MaxTupleAttributeNumber ] ; bool isnull [ MaxTupleAttributeNumber ] ; int i ; uint16 nliveatts = 0 ; desc = RelationGetDescr ( rel ) ; for ( i = 0 ; i < desc -> natts ; i ++ ) { if ( TupleDescAttr ( desc , i ) -> attisdropped || TupleDescAttr ( desc , i ) -> attgenerated ) continue ; nliveatts ++ ; } pq_sendint16 ( out , nliveatts ) ; enlargeStringInfo ( out , tuple -> t_len + nliveatts * ( 1 + 4 ) ) ; heap_deform_tuple ( tuple , desc , values , isnull ) ; for ( i = 0 ; i < desc -> natts ; i ++ ) { HeapTuple typtup ; Form_pg_type typclass ; Form_pg_attribute att = TupleDescAttr ( desc , i ) ; char * outputstr ; if ( att -> attisdropped || att -> attgenerated ) continue ; if ( isnull [ i ] ) { pq_sendbyte ( out , LOGICALREP_COLUMN_NULL ) ; continue ; } if ( att -> attlen == -1 && VARATT_IS_EXTERNAL_ONDISK ( values [ i ] ) ) { pq_sendbyte ( out , LOGICALREP_COLUMN_UNCHANGED ) ; continue ; } typtup = SearchSysCache1 ( TYPEOID , ObjectIdGetDatum ( att -> atttypid ) ) ; if ( ! HeapTupleIsValid ( typtup ) ) elog ( ERROR , " " , att -> atttypid ) ; typclass = ( Form_pg_type ) GETSTRUCT ( typtup ) ; if ( binary && OidIsValid ( typclass -> typsend ) ) { bytea * outputbytes ; int len ; pq_sendbyte ( out , LOGICALREP_COLUMN_BINARY ) ; outputbytes = OidSendFunctionCall ( typclass -> typsend , values [ i ] ) ; len = VARSIZE ( outputbytes ) - VARHDRSZ ; pq_sendint ( out , len , 4 ) ; pq_sendbytes ( out , VARDATA ( outputbytes ) , len ) ; pfree ( outputbytes ) ; } else { pq_sendbyte ( out , LOGICALREP_COLUMN_TEXT ) ; outputstr = OidOutputFunctionCall ( typclass -> typoutput , values [ i ] ) ; pq_sendcountedtext ( out , outputstr , strlen ( outputstr ) , false ) ; pfree ( outputstr ) ; } ReleaseSysCache ( typtup ) ; } }
Write a tuple to the outputstream, in the most efficient format possible.
Write a tuple to the outputstream in the most efficient format possible .
/* try to allocate enough memory from the get-go */ /* Write the values */ /*
			 * Unchanged toasted datum.  (Note that we don't promise to detect
			 * unchanged data in general; this is just a cheap check to avoid
			 * sending large values unnecessarily.)
			 */ /*
		 * Send in binary if requested and type has suitable send function.
		 */ /* length */ /* data */