Inline TOAST: compression

Author: Nikita Glukhov

src/backend/access/common/detoast.c

@@ -49,19 +49,15 @@ create_detoast_iterator(struct varlena *attr)
 		if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
 		{
 			iter->compressed = true;
+			iter->compression_method = VARATT_EXTERNAL_GET_COMPRESS_METHOD(toast_pointer);
 
 			/* prepare buffer to received decompressed data */
 			iter->buf = create_toast_buffer(toast_pointer.va_rawsize, false);
-
-			/* initialize state for pglz_decompress_iterate() */
-			iter->ctrl = 0;
-			iter->ctrlc = INVALID_CTRLC;
-			iter->len = 0;
-			iter->off = 0;
 		}
 		else
 		{
 			iter->compressed = false;
+			iter->compression_method = TOAST_INVALID_COMPRESSION_ID;
 
 			/* point the buffer directly at the raw data */
 			iter->buf = fetch_iter->buf;
@@ -95,16 +91,14 @@ create_detoast_iterator(struct varlena *attr)
 		iter->fetch_datum_iterator->buf = buf = create_toast_buffer(VARSIZE_ANY(attr), true);
 		iter->fetch_datum_iterator->done = true;
 		iter->compressed = true;
+		iter->compression_method = VARDATA_COMPRESSED_GET_COMPRESS_METHOD(attr);
 
 		memcpy((void *) buf->buf, attr, VARSIZE_ANY(attr));
 		buf->limit = (char *) buf->capacity;
 
 		/* prepare buffer to received decompressed data */
 		iter->buf = create_toast_buffer(TOAST_COMPRESS_EXTSIZE(attr) + VARHDRSZ, false);
 
-		/* initialize state for pglz_decompress_iterate() */
-		iter->ctrl = 0;
-		iter->ctrlc = INVALID_CTRLC;
 		return iter;
 	}
 	else

src/backend/access/common/toast_compression.c

@@ -64,6 +64,7 @@ pglz_compress_datum(const struct varlena *value)
 	len = pglz_compress(VARDATA_ANY(value),
 						valsize,
 						(char *) tmp + VARHDRSZ_COMPRESSED,
+						NULL,
 						NULL);
 	if (len < 0)
 	{

src/backend/access/common/toast_internals.c

@@ -1135,28 +1135,12 @@ toast_extract_chunk_fields(Relation toastrel, TupleDesc toasttupDesc,
 	}
 }
 
-/* ----------
- * create_fetch_datum_iterator -
- *
- * Initialize fetch datum iterator.
- * ----------
- */
-FetchDatumIterator
-create_fetch_datum_iterator(struct varlena *attr)
+static void
+create_fetch_datum_iterator_scan(FetchDatumIterator iter)
 {
 	int			validIndex;
-	FetchDatumIterator iter;
-
-	if (!VARATT_IS_EXTERNAL_ONDISK(attr))
-		elog(ERROR, "create_fetch_datum_iterator shouldn't be called for non-ondisk datums");
-
-	iter = (FetchDatumIterator) palloc0(sizeof(FetchDatumIteratorData));
-
-	/* Must copy to access aligned fields */
-	VARATT_EXTERNAL_GET_POINTER(iter->toast_pointer, attr);
 
-	iter->ressize = VARATT_EXTERNAL_GET_EXTSIZE(iter->toast_pointer);
-	iter->numchunks = ((iter->ressize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
+	MemoryContext oldcxt = MemoryContextSwitchTo(iter->mcxt);
 
 	/*
 	 * Open the toast relation and its indexes
@@ -1189,6 +1173,33 @@ create_fetch_datum_iterator(struct varlena *attr)
 	iter->toastscan = systable_beginscan_ordered(iter->toastrel, iter->toastidxs[validIndex],
 												 &iter->snapshot, 1, &iter->toastkey);
 
+	MemoryContextSwitchTo(oldcxt);
+}
+
+/* ----------
+ * create_fetch_datum_iterator -
+ *
+ * Initialize fetch datum iterator.
+ * ----------
+ */
+FetchDatumIterator
+create_fetch_datum_iterator(struct varlena *attr)
+{
+	FetchDatumIterator iter;
+
+	if (!VARATT_IS_EXTERNAL_ONDISK(attr))
+		elog(ERROR, "create_fetch_datum_iterator shouldn't be called for non-ondisk datums");
+
+	iter = (FetchDatumIterator) palloc0(sizeof(FetchDatumIteratorData));
+
+	iter->mcxt = CurrentMemoryContext;
+
+	/* Must copy to access aligned fields */
+	VARATT_EXTERNAL_GET_POINTER(iter->toast_pointer, attr);
+
+	iter->ressize = VARATT_EXTERNAL_GET_EXTSIZE(iter->toast_pointer);
+	iter->numchunks = ((iter->ressize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
+
 	iter->buf = create_toast_buffer(iter->ressize + VARHDRSZ,
 									VARATT_EXTERNAL_IS_COMPRESSED(iter->toast_pointer));
 
@@ -1204,7 +1215,7 @@ free_fetch_datum_iterator(FetchDatumIterator iter)
 	if (iter == NULL)
 		return;
 
-	if (!iter->done)
+	if (!iter->done && iter->toastscan)
 	{
 		systable_endscan_ordered(iter->toastscan);
 		toast_close_indexes(iter->toastidxs, iter->num_indexes, AccessShareLock);
@@ -1236,6 +1247,9 @@ fetch_datum_iterate(FetchDatumIterator iter)
 
 	Assert(iter != NULL && !iter->done);
 
+	if (!iter->toastscan)
+		create_fetch_datum_iterator_scan(iter);
+
 	ttup = systable_getnext_ordered(iter->toastscan, ForwardScanDirection);
 	if (ttup == NULL)
 	{
@@ -1344,8 +1358,9 @@ ToastBuffer *
 create_toast_buffer(int32 size, bool compressed)
 {
 	ToastBuffer *buf = (ToastBuffer *) palloc0(sizeof(ToastBuffer));
-	buf->buf = (const char *) palloc0(size);
-	if (compressed) {
+	buf->buf = (const char *) palloc(size);
+	if (compressed)
+	{
 		SET_VARSIZE_COMPRESSED(buf->buf, size);
 		/*
 		 * Note the constraint buf->position <= buf->limit may be broken
@@ -1375,6 +1390,7 @@ free_toast_buffer(ToastBuffer *buf)
 	pfree(buf);
 }
 
+#if 0
 /* ----------
  * pglz_decompress_iterate -
  *
@@ -1392,7 +1408,7 @@ free_toast_buffer(ToastBuffer *buf)
  */
 void
 pglz_decompress_iterate(ToastBuffer *source, ToastBuffer *dest,
-						DetoastIterator iter, unsigned char *destend)
+						DetoastIterator iter, const char *destend)
 {
 	const unsigned char *sp;
 	const unsigned char *srcend;
@@ -1525,3 +1541,4 @@ pglz_decompress_iterate(ToastBuffer *source, ToastBuffer *dest,
 	source->position = (char *) sp;
 	dest->limit = (char *) dp;
 }
+#endif

src/backend/access/transam/xloginsert.c

@@ -911,7 +911,7 @@ XLogCompressBackupBlock(char *page, uint16 hole_offset, uint16 hole_length,
 	switch ((WalCompression) wal_compression)
 	{
 		case WAL_COMPRESSION_PGLZ:
-			len = pglz_compress(source, orig_len, dest, PGLZ_strategy_default);
+			len = pglz_compress(source, orig_len, dest, PGLZ_strategy_default, NULL);
 			break;
 
 		case WAL_COMPRESSION_LZ4:

src/common/pg_lzcompress.c

@@ -507,9 +507,10 @@ pglz_find_match(int16 *hstart, const char *input, const char *end,
  */
 int32
 pglz_compress(const char *source, int32 slen, char *dest,
-			  const PGLZ_Strategy *strategy)
+			  const PGLZ_Strategy *strategy, int32 *dlen)
 {
 	unsigned char *bp = (unsigned char *) dest;
+	unsigned char *bend;
 	unsigned char *bstart = bp;
 	int			hist_next = 1;
 	bool		hist_recycle = false;
@@ -545,6 +546,16 @@ pglz_compress(const char *source, int32 slen, char *dest,
 		slen > strategy->max_input_size)
 		return -1;
 
+	if (dlen)
+	{
+		if (*dlen < 4)
+			return -1;
+
+		bend = bstart + *dlen - 4;
+	}
+	else
+		bend = bstart + PGLZ_MAX_OUTPUT(slen);
+
 	/*
 	 * Limit the match parameters to the supported range.
 	 */
@@ -627,6 +638,9 @@ pglz_compress(const char *source, int32 slen, char *dest,
 		if (!found_match && bp - bstart >= strategy->first_success_by)
 			return -1;
 
+		if (bp > bend)
+			break;
+
 		/*
 		 * Try to find a match in the history
 		 */
@@ -671,15 +685,16 @@ pglz_compress(const char *source, int32 slen, char *dest,
 	if (result_size >= result_max)
 		return -1;
 
+	if (dlen)
+		*dlen = dp - source;
+
 	/* success */
 	return result_size;
 }
 
 /* Opaque pglz decompression state */
 typedef struct pglz_state
 {
-	const unsigned char *sp;
-	unsigned char *dp;
 	int32		len;
 	int32		off;
 	int			ctrlc;
@@ -699,28 +714,23 @@ typedef struct pglz_state
  * ----------
  */
 int32
-pglz_decompress_state(const char *source, int32 slen, char *dest,
+pglz_decompress_state(const char *source, int32 *slen, char *dest,
 					  int32 dlen, bool check_complete, bool last_cource_chunk,
 					  void **pstate)
 {
 	pglz_state *state = pstate ? *pstate : NULL;
-	const unsigned char *sp;
-	const unsigned char *srcend;
-	unsigned char *dp;
-	unsigned char *destend;
+	const unsigned char *sp = (const unsigned char *) source;
+	const unsigned char *srcend = sp + *slen;
+	unsigned char *dp = (unsigned char *) dest;
+	unsigned char *destend = dp + dlen;
 	unsigned char ctrl;
 	int			ctrlc;
 	int32		len;
 	int32		remlen;
 	int32		off;
 
-	srcend = ((const unsigned char *) source) + slen;
-	destend = ((unsigned char *) dest) + rawsize;
-
 	if (state)
 	{
-		sp = state->sp;
-		dp = state->dp;
 		ctrl = state->ctrl;
 		ctrlc = state->ctrlc;
 
@@ -742,7 +752,7 @@ pglz_decompress_state(const char *source, int32 slen, char *dest,
 			if (dp >= destend)
 			{
 				state->len = remlen;
-				state->dp = dp;
+				*slen = 0;
 				return (char *) dp - dest;
 			}
 
@@ -757,8 +767,6 @@ pglz_decompress_state(const char *source, int32 slen, char *dest,
 	}
 	else
 	{
-		sp = (const unsigned char *) source;
-		dp = (unsigned char *) dest;
 		ctrl = 0;
 		ctrlc = 8;
 		remlen = 0;
@@ -884,10 +892,10 @@ pglz_decompress_state(const char *source, int32 slen, char *dest,
 
 		state->ctrl = ctrl;
 		state->ctrlc = ctrlc;
-		state->sp = sp;
-		state->dp = dp;
 		state->len = remlen;
 		state->off = off;
+
+		*slen = (const char *) sp - source;
 	}
 
 	/*
@@ -948,5 +956,5 @@ int32
 pglz_decompress(const char *source, int32 slen, char *dest, int32 rawsize,
 				bool check_complete)
 {
-	return pglz_decompress_state(source, slen, dest, rawsize, check_complete, true, NULL);
+	return pglz_decompress_state(source, &slen, dest, rawsize, check_complete, true, NULL);
 }

src/include/access/detoast.h

@@ -47,6 +47,7 @@ do { \
 #include "postgres.h"
 #include "access/genam.h"
 #include "access/toasterapi.h"
+#include "access/toast_compression.h"
 
 /*
  * TOAST buffer is a producer consumer buffer.
@@ -78,10 +79,11 @@ typedef struct FetchDatumIteratorData
 	ToastBuffer	*buf;
 	Relation	toastrel;
 	Relation	*toastidxs;
+	MemoryContext mcxt;
 	SysScanDesc	toastscan;
 	ScanKeyData	toastkey;
 	SnapshotData			snapshot;
-	struct varatt_external	toast_pointer;
+	struct varatt_external toast_pointer;
 	int32		ressize;
 	int32		nextidx;
 	int32		numchunks;
@@ -91,22 +93,13 @@ typedef struct FetchDatumIteratorData
 
 typedef struct FetchDatumIteratorData *FetchDatumIterator;
 
-/*
- * If "ctrlc" field in iterator is equal to INVALID_CTRLC, it means that
- * the field is invalid and need to read the control byte from the
- * source buffer in the next iteration, see pglz_decompress_iterate().
- */
-#define INVALID_CTRLC 8
-
 typedef struct DetoastIteratorData
 {
 	ToastBuffer 		*buf;
 	FetchDatumIterator	fetch_datum_iterator;
-	unsigned char		ctrl;
-	int					ctrlc;
 	int					nrefs;
-	int32				len;
-	int32				off;
+	void			   *decompression_state;
+	ToastCompressionId	compression_method;
 	bool				compressed;		/* toast value is compressed? */
 	bool				done;
 }			DetoastIteratorData;
@@ -118,8 +111,10 @@ extern void free_fetch_datum_iterator(FetchDatumIterator iter);
 extern void fetch_datum_iterate(FetchDatumIterator iter);
 extern ToastBuffer *create_toast_buffer(int32 size, bool compressed);
 extern void free_toast_buffer(ToastBuffer *buf);
+extern void toast_decompress_iterate(ToastBuffer *source, ToastBuffer *dest,
+									 DetoastIterator iter, const char *destend);
 extern void pglz_decompress_iterate(ToastBuffer *source, ToastBuffer *dest,
-									DetoastIterator iter, unsigned char *destend);
+									DetoastIterator iter, char *destend);
 
 /* ----------
  * create_detoast_iterator -
@@ -171,7 +166,7 @@ detoast_iterate(DetoastIterator detoast_iter, const char *destend)
 		fetch_datum_iterate(fetch_iter);
 
 	if (detoast_iter->compressed)
-		pglz_decompress_iterate(fetch_iter->buf, detoast_iter->buf, detoast_iter, (unsigned char *) destend);
+		toast_decompress_iterate(fetch_iter->buf, detoast_iter->buf, detoast_iter, destend);
 
 	if (detoast_iter->buf->limit == detoast_iter->buf->capacity)
 		detoast_iter->done = true;

src/include/access/toast_internals.h

@@ -40,6 +40,9 @@ typedef struct toast_compress_header
 #define TOAST_COMPRESS_METHOD(ptr) \
 	(((toast_compress_header *) (ptr))->tcinfo >> VARLENA_EXTSIZE_BITS)
 
+#define TOAST_COMPRESS_HDRSZ			VARHDRSZ_COMPRESSED
+#define TOAST_COMPRESS_RAWDATA(attr)	((char *)(attr) + TOAST_COMPRESS_HDRSZ)
+
 #define TOAST_COMPRESS_SET_SIZE_AND_COMPRESS_METHOD(ptr, len, cm_method) \
 	do { \
 		Assert((len) > 0 && (len) <= VARLENA_EXTSIZE_MASK); \

src/include/common/pg_lzcompress.h

@@ -84,10 +84,10 @@ extern PGDLLIMPORT const PGLZ_Strategy *const PGLZ_strategy_always;
  * ----------
  */
 extern int32 pglz_compress(const char *source, int32 slen, char *dest,
-						   const PGLZ_Strategy *strategy);
+						   const PGLZ_Strategy *strategy, int32 *dlen);
 extern int32 pglz_decompress(const char *source, int32 slen, char *dest,
 							 int32 rawsize, bool check_complete);
-extern int32 pglz_decompress_state(const char *source, int32 slen,
+extern int32 pglz_decompress_state(const char *source, int32 *slen,
 								   char *dest, int32 dlen,
 								   bool check_complete, bool last_source_chunk,
 								   void **state);