Fix issues with wide tuples being updated and REPLICA IDENTITY FULL.

When replica identity full is being used with a wide tuple (above 2^16
bytes after compression) it lead to errors and/or crashes during
decoding because the length field used to store such tuples doesn't
fit into the variable used to store the width in the WAL record.

To fix, discontinue use of xl_heap_header_len.t_len when decoding the
old tuple version, instead compute length of the old tuple by
subtracting the new tuple's length from the record length.

In newer version of postgres this issue is moot because the length is
stored by the new WAL machinery, instead of a xl_heap_header_len
struct.  A separate commit will forward-patch the regression test.

Reported-By: "anderson"
Discussion: http://postgr.es/m/20170105144819.f6i5o64vfvy4bn5i@alap3.anarazel.de

1 files changed, 16 insertions, 3 deletions

diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 3a6d6ffab1a..09a334232e1 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -653,6 +653,7 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	xl_heap_update *xlrec;
 	ReorderBufferChange *change;
 	char	   *data;
+	size_t		remlen = r->xl_len;
 
 	xlrec = (xl_heap_update *) buf->record_data;
 
@@ -666,6 +667,7 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 	/* caution, remaining data in record is not aligned */
 	data = buf->record_data + SizeOfHeapUpdate;
+	remlen -= SizeOfHeapUpdate;
 
 	if (xlrec->flags & XLOG_HEAP_CONTAINS_NEW_TUPLE)
 	{
@@ -677,6 +679,7 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 		memcpy(&xlhdr, data, sizeof(xlhdr));
 		data += offsetof(xl_heap_header_len, header);
+		remlen -= offsetof(xl_heap_header_len, header);
 
 		datalen = xlhdr.t_len + SizeOfHeapHeader;
 		tuplelen = xlhdr.t_len;
@@ -687,8 +690,10 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
 		/* skip over the rest of the tuple header */
 		data += SizeOfHeapHeader;
+		remlen -= SizeOfHeapHeader;
 		/* skip over the tuple data */
 		data += xlhdr.t_len;
+		remlen -= xlhdr.t_len;
 	}
 
 	if (xlrec->flags & XLOG_HEAP_CONTAINS_OLD)
@@ -699,10 +704,17 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 		memcpy(&xlhdr, data, sizeof(xlhdr));
 		data += offsetof(xl_heap_header_len, header);
+		remlen -= offsetof(xl_heap_header_len, header);
 
-		/* t_len is inconsistent with other cases, see log_heap_update */
-		tuplelen = xlhdr.t_len - offsetof(HeapTupleHeaderData, t_bits);
-		datalen = tuplelen + SizeOfHeapHeader;
+		/*
+		 * NB: Even though xl_heap_header_len contains the tuple's length,
+		 * it's length field is not wide enough. Use the whole record length
+		 * minus the new tuple's length instead. We can't remove the record
+		 * length from the WAL record format in 9.4 due to compatibility
+		 * concerns - later versions don't have it anyway.
+		 */
+		datalen = remlen;
+		tuplelen = datalen - SizeOfHeapHeader;
 
 		change->data.tp.oldtuple =
 			ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
@@ -710,6 +722,7 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
 #ifdef NOT_USED
 		data += datalen;
+		remlen -= datalen;
 #endif
 	}