Logical replication data reconstruction for TOAST API

Reference logical replication mechanics could process only data TOASTed
by reference TOAST - using External TOAST Pointers, and data reconstruction
is done inside replication mechanics instead of Toaster used to TOAST data.
This patch hides data reconstruction into Toaster, adding reconstruction
function to TOAST API and moving reference reconstruction code into Generic
Toaster.

Author: Teodor Sigaev <teodor@sigaev.ru>
Author: Oleg Bartunov <obartunov@postgrespro.ru>
Author: Nikita Glukhov <n.gluhov@postgrespro.ru>
Author: Nikita Malakhov <n.malakhov@postgrespro.ru>

Author: Nikita Malakhov

src/backend/access/toast/generic_toaster.c

@@ -52,6 +52,7 @@
 #include "utils/fmgroids.h"
 #include "access/generic_toaster.h"
 #include "access/toast_compression.h"
+#include "replication/reorderbuffer.h"
 
 /*
  * Callback function signatures --- see toaster.sgml for more info.
@@ -63,10 +64,10 @@
  */
 static void
 generic_toast_init(Relation rel, Oid toastoid, Oid toastindexoid, Datum reloptions, LOCKMODE lockmode,
-				 bool check, Oid OIDOldToast)
+		   bool check, Oid OIDOldToast)
 {
 	(void) create_toast_table(rel, toastoid, toastindexoid, reloptions, lockmode,
-							  check, OIDOldToast);
+				  check, OIDOldToast);
 }
 
 
@@ -82,8 +83,8 @@ generic_toast(Relation toast_rel, Oid toasterid, Datum value, Datum oldvalue,
 	Assert(toast_rel != NULL);
 
 	result = toast_save_datum(toast_rel, value,
-							  (struct varlena *) DatumGetPointer(oldvalue),
-							  options);
+				 (struct varlena *) DatumGetPointer(oldvalue),
+				 options);
 	return result;
 }
 
@@ -107,7 +108,7 @@ generic_detoast(Datum toast_ptr, int offset, int length)
 	else
 	{
 		result = toast_fetch_datum_slice(tvalue,
-										 offset, length);
+						offset, length);
 	}
 
 	return PointerGetDatum(result);
@@ -616,6 +617,83 @@ generic_toaster_vtable(Datum toast_ptr)
 	return routine;
 }
 
+struct varlena *
+generic_toaster_reconstruct(Relation toastrel, struct varlena *varlena,
+                                HTAB *toast_hash)
+{
+       struct varatt_external toast_pointer;
+       struct varlena *reconstructed;
+       ReorderBufferToastEnt *ent;
+       dlist_iter      it;
+       Size            data_done = 0;
+       TupleDesc       toast_desc;
+
+       if (!VARATT_IS_EXTERNAL_ONDISK(varlena))
+               return NULL;
+
+       if (!toast_hash)
+               return NULL;
+
+       VARATT_EXTERNAL_GET_POINTER(toast_pointer, varlena);
+
+       /*
+        * Check whether the toast tuple changed, replace if so.
+        */
+       ent = (ReorderBufferToastEnt *)
+               hash_search(toast_hash,
+                                       (void *) &toast_pointer.va_valueid,
+                                       HASH_FIND,
+                                       NULL);
+
+       if (ent == NULL)
+               return NULL;
+
+       reconstructed = palloc0(toast_pointer.va_rawsize);
+
+       ent->reconstructed = reconstructed;
+
+       toast_desc = RelationGetDescr(toastrel);
+
+       /* stitch toast tuple back together from its parts */
+       dlist_foreach(it, &ent->chunks)
+       {
+               bool            isnull;
+               ReorderBufferChange *cchange;
+               ReorderBufferTupleBuf *ctup;
+               Pointer         chunk;
+
+               cchange = dlist_container(ReorderBufferChange, node, it.cur);
+               ctup = cchange->data.tp.newtuple;
+               chunk = DatumGetPointer(fastgetattr(&ctup->tuple, 3, toast_desc, &isnull));
+
+               Assert(!isnull);
+               Assert(!VARATT_IS_EXTERNAL(chunk));
+               Assert(!VARATT_IS_SHORT(chunk));
+
+               memcpy(VARDATA(reconstructed) + data_done,
+                          VARDATA(chunk),
+                          VARSIZE(chunk) - VARHDRSZ);
+               data_done += VARSIZE(chunk) - VARHDRSZ;
+       }
+       Assert(data_done == VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer));
+
+       /* make sure its marked as compressed or not */
+       if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
+               SET_VARSIZE_COMPRESSED(reconstructed, data_done + VARHDRSZ);
+       else
+               SET_VARSIZE(reconstructed, data_done + VARHDRSZ);
+
+       return reconstructed;
+}
+
+static Datum
+generic_reconstruct(Relation toastrel, struct varlena *varlena,
+                                       HTAB *toast_hash, bool *need_free)
+{
+       *need_free = false;
+       return PointerGetDatum(generic_toaster_reconstruct(toastrel, varlena, toast_hash));
+}
+
 Datum
 default_toaster_handler(PG_FUNCTION_ARGS)
 {
@@ -628,6 +706,7 @@ default_toaster_handler(PG_FUNCTION_ARGS)
 	tsrroutine->update_toast = NULL;
 	tsrroutine->copy_toast = NULL;
 	tsrroutine->get_vtable = generic_toaster_vtable;
+	tsrroutine->reconstruct = generic_reconstruct;
 	tsrroutine->toastervalidate = generic_validate;
 
 	PG_RETURN_POINTER(tsrroutine);

src/backend/replication/logical/proto.c

@@ -19,6 +19,7 @@
 #include "replication/logicalproto.h"
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
+#include "access/toasterapi.h"
 
 /*
  * Protocol message flags.
@@ -814,7 +815,9 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
 			continue;
 		}
 
-		if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(values[i]))
+        if (att->attlen == -1 &&
+           (VARATT_IS_EXTERNAL_ONDISK(values[i]) ||
+            VARATT_IS_CUSTOM(values[i])))
 		{
 			/*
 			 * Unchanged toasted datum.  (Note that we don't promise to detect
@@ -830,6 +833,31 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
 			elog(ERROR, "cache lookup failed for type %u", att->atttypid);
 		typclass = (Form_pg_type) GETSTRUCT(typtup);
 
+       	if (att->attlen == -1 && VARATT_IS_EXTERNAL_INDIRECT(values[i]))
+       	{
+            struct varatt_indirect redirect;
+            struct varlena *attr;
+
+            VARATT_EXTERNAL_GET_POINTER(redirect, values[i]);
+            attr = (struct varlena *) redirect.pointer;
+
+            /* Send type diff, if it is a custom pointer. */
+            if (VARATT_IS_CUSTOM(attr))
+            {
+                int len = VARATT_CUSTOM_GET_DATA_SIZE(attr);
+
+                if (!OidIsValid(typclass->typapplydiff))
+                    elog(ERROR, "toaster returned diff, but datatype does not support diffs");
+
+                pq_sendbyte(out, LOGICALREP_COLUMN_DIFF);
+                pq_sendint(out, len, 4);        /* length */
+                pq_sendbytes(out, VARATT_CUSTOM_GET_DATA(attr), len);   /* data */
+
+                ReleaseSysCache(typtup);
+                continue;
+            }
+	    }
+
 		/*
 		 * Send in binary if requested and type has suitable send function.
 		 */
@@ -907,6 +935,7 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
 				value->maxlen = len;
 				break;
 			case LOGICALREP_COLUMN_BINARY:
+			case LOGICALREP_COLUMN_DIFF:
 				len = pq_getmsgint(in, 4);	/* read length */
 
 				/* and data */

src/backend/replication/logical/reorderbuffer.c

@@ -87,6 +87,7 @@
 #include "access/toasterapi.h"
 #include "access/heapam.h"
 #include "access/rewriteheap.h"
+#include "access/toast_helper.h"
 #include "access/transam.h"
 #include "access/xact.h"
 #include "access/xlog_internal.h"
@@ -157,19 +158,6 @@ typedef struct ReorderBufferIterTXNState
 	ReorderBufferIterTXNEntry entries[FLEXIBLE_ARRAY_MEMBER];
 } ReorderBufferIterTXNState;
 
-/* toast datastructures */
-typedef struct ReorderBufferToastEnt
-{
-	Oid			chunk_id;		/* toast_table.chunk_id */
-	int32		last_chunk_seq; /* toast_table.chunk_seq of the last chunk we
-								 * have seen */
-	Size		num_chunks;		/* number of chunks we've already seen */
-	Size		size;			/* combined size of chunks seen */
-	dlist_head	chunks;			/* linked list of chunks */
-	struct varlena *reconstructed;	/* reconstructed varlena now pointed to in
-									 * main tup */
-} ReorderBufferToastEnt;
-
 /* Disk serialization support datastructures */
 typedef struct ReorderBufferDiskChange
 {
@@ -4724,15 +4712,16 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	bool	   *free;
 	HeapTuple	tmphtup;
 	Relation	toast_rel;
-	TupleDesc	toast_desc;
 	MemoryContext oldcontext;
 	ReorderBufferTupleBuf *newtup;
 	Size		old_size;
 
 	/* no toast tuples changed */
-	if (txn->toast_hash == NULL)
-		return;
-
+    if (!change->data.tp.newtuple ||
+        !HeapTupleHasExternal(&change->data.tp.newtuple->tuple))
+	{
+        return;
+	}
 	/*
 	 * We're going to modify the size of the change. So, to make sure the
 	 * accounting is correct we record the current change size and then after
@@ -4757,8 +4746,6 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		elog(ERROR, "could not open toast relation with OID %u (base relation \"%s\")",
 			 relation->rd_rel->reltoastrelid, RelationGetRelationName(relation));
 
-	toast_desc = RelationGetDescr(toast_rel);
-
 	/* should we allocate from stack instead? */
 	attrs = palloc0(sizeof(Datum) * desc->natts);
 	isnull = palloc0(sizeof(bool) * desc->natts);
@@ -4771,16 +4758,14 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	for (natt = 0; natt < desc->natts; natt++)
 	{
 		Form_pg_attribute attr = TupleDescAttr(desc, natt);
-		ReorderBufferToastEnt *ent;
+		TsrRoutine *toaster;
 		struct varlena *varlena;
 
 		/* va_rawsize is the size of the original datum -- including header */
-		struct varatt_external toast_pointer;
 		struct varatt_indirect redirect_pointer;
 		struct varlena *new_datum = NULL;
 		struct varlena *reconstructed;
-		dlist_iter	it;
-		Size		data_done = 0;
+		bool        need_free;
 
 		/* system columns aren't toasted */
 		if (attr->attnum < 0)
@@ -4801,68 +4786,40 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		varlena = (struct varlena *) DatumGetPointer(attrs[natt]);
 
 		/* no need to do anything if the tuple isn't external */
-		if (!VARATT_IS_EXTERNAL(varlena))
-			continue;
-
-		VARATT_EXTERNAL_GET_POINTER(toast_pointer, varlena);
-
-		/*
-		 * Check whether the toast tuple changed, replace if so.
-		 */
-		ent = (ReorderBufferToastEnt *)
-			hash_search(txn->toast_hash,
-						(void *) &toast_pointer.va_valueid,
-						HASH_FIND,
-						NULL);
-		if (ent == NULL)
+        if (!VARATT_IS_EXTERNAL_ONDISK(varlena) &&
+            !VARATT_IS_CUSTOM(varlena))
 			continue;
 
-		new_datum =
-			(struct varlena *) palloc0(INDIRECT_POINTER_SIZE);
-
-		free[natt] = true;
-
-		reconstructed = palloc0(toast_pointer.va_rawsize);
-
-		ent->reconstructed = reconstructed;
-
-		/* stitch toast tuple back together from its parts */
-		dlist_foreach(it, &ent->chunks)
+        toaster = SearchTsrCache(attr->atttoaster);
+		if (toaster->reconstruct)
 		{
-			bool		isnull;
-			ReorderBufferChange *cchange;
-			ReorderBufferTupleBuf *ctup;
-			Pointer		chunk;
-
-			cchange = dlist_container(ReorderBufferChange, node, it.cur);
-			ctup = cchange->data.tp.newtuple;
-			chunk = DatumGetPointer(fastgetattr(&ctup->tuple, 3, toast_desc, &isnull));
-
-			Assert(!isnull);
-			Assert(!VARATT_IS_EXTERNAL(chunk));
-			Assert(!VARATT_IS_SHORT(chunk));
-
-			memcpy(VARDATA(reconstructed) + data_done,
-				   VARDATA(chunk),
-				   VARSIZE(chunk) - VARHDRSZ);
-			data_done += VARSIZE(chunk) - VARHDRSZ;
+			reconstructed = (struct varlena *) DatumGetPointer(
+				toaster->reconstruct(toast_rel, varlena, txn->toast_hash, &need_free));
 		}
-		Assert(data_done == VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer));
-
-		/* make sure its marked as compressed or not */
-		if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
-			SET_VARSIZE_COMPRESSED(reconstructed, data_done + VARHDRSZ);
 		else
-			SET_VARSIZE(reconstructed, data_done + VARHDRSZ);
+        {
+            ereport(ERROR,
+                   (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+                    errmsg("TOASTER does not support reconstruction of values")));
+        }
+
+        if (!reconstructed)
+            continue;
+
+        if (need_free)
+            txn->toast_reconstructed = lappend(txn->toast_reconstructed, reconstructed);
 
 		memset(&redirect_pointer, 0, sizeof(redirect_pointer));
 		redirect_pointer.pointer = reconstructed;
 
+        new_datum = (struct varlena *) palloc0(INDIRECT_POINTER_SIZE);
+
 		SET_VARTAG_EXTERNAL(new_datum, VARTAG_INDIRECT);
 		memcpy(VARDATA_EXTERNAL(new_datum), &redirect_pointer,
 			   sizeof(redirect_pointer));
 
 		attrs[natt] = PointerGetDatum(new_datum);
+		free[natt] = true;
 	}
 
 	/*
@@ -4910,6 +4867,9 @@ ReorderBufferToastReset(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	HASH_SEQ_STATUS hstat;
 	ReorderBufferToastEnt *ent;
 
+    list_free_deep(txn->toast_reconstructed);
+    txn->toast_reconstructed = NIL;
+
 	if (txn->toast_hash == NULL)
 		return;