diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 9f417de..514c166 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -61,6 +61,9 @@ #include "utils/builtins.h" #include "utils/guc.h" #include "utils/memutils.h" +#include "utils/pg_lzcompress.h" +#include "utils/pg_snappy.h" +#include "utils/pg_lz4.h" #include "utils/ps_status.h" #include "utils/relmapper.h" #include "utils/snapmgr.h" @@ -85,6 +88,7 @@ bool XLogArchiveMode = false; char *XLogArchiveCommand = NULL; bool EnableHotStandby = false; bool fullPageWrites = true; +int full_page_compression = BACKUP_BLOCK_COMPRESSION_OFF; bool wal_log_hints = false; bool log_checkpoints = false; int sync_method = DEFAULT_SYNC_METHOD; @@ -741,6 +745,10 @@ static bool holdingAllLocks = false; static MemoryContext walDebugCxt = NULL; #endif +/* For storing full page image after compression */ +static char *compressed_pages[XLR_MAX_BKP_BLOCKS]; +static bool compressed_pages_allocated = false; + static void readRecoveryCommandFile(void); static void exitArchiveRecovery(TimeLineID endTLI, XLogSegNo endLogSegNo); static bool recoveryStopsBefore(XLogRecord *record); @@ -813,6 +821,8 @@ static void ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, static bool ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, XLogRecPtr *PrevPtr); static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto); +static char *CompressBackupBlock(char *page, uint32 orig_len, char *dest, uint32 *len); +static void CompressBackupBlockPagesAlloc(void); static char *GetXLogBuffer(XLogRecPtr ptr); static XLogRecPtr XLogBytePosToRecPtr(uint64 bytepos); static XLogRecPtr XLogBytePosToEndRecPtr(uint64 bytepos); @@ -1020,6 +1030,27 @@ begin:; rdt->next = &(dtbuf_rdt2[i]); rdt = rdt->next; + if (full_page_compression != BACKUP_BLOCK_COMPRESSION_OFF) + { + /* Compress the backup block before including it in rdata chain */ + rdt->data = CompressBackupBlock(page, BLCKSZ - bkpb->hole_length, + compressed_pages[i], &(rdt->len)); + if (rdt->data != NULL) + { + /* + * write_len is the length of compressed block and its varlena + * header + */ + write_len += rdt->len; + bkpb->hole_length = BLCKSZ - rdt->len; + + /* Adding information about compression in the backup block header */ + bkpb->block_compression = full_page_compression; + rdt->next = NULL; + continue; + } + } + bkpb->block_compression = BACKUP_BLOCK_COMPRESSION_OFF; if (bkpb->hole_length == 0) { rdt->data = page; @@ -1774,6 +1805,65 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto) } /* + * Create a compressed version of a backup block + * + * If successful, return a compressed result and set 'len' to its length. + * Otherwise (ie, compressed result is actually bigger than original), + * return NULL. + */ +static char * +CompressBackupBlock(char *page, uint32 orig_len, char *dest, uint32 *len) +{ + struct varlena *buf = (struct varlena *) dest; + + if (full_page_compression == BACKUP_BLOCK_COMPRESSION_SNAPPY) + { + int ret; + ret = pg_snappy_compress(page,BLCKSZ,buf); + /* EIO is returned for incompressible data */ + if (ret == EIO) + return NULL; + } + else if (full_page_compression == BACKUP_BLOCK_COMPRESSION_LZ4) + { + int ret; + ret = pg_LZ4_compress(page,BLCKSZ,buf); + /* Zero is returned for incompressible data */ + if (ret == 0) + return NULL; + } + else if (full_page_compression == BACKUP_BLOCK_COMPRESSION_PGLZ) + { + bool ret; + ret = pglz_compress(page, BLCKSZ, + (PGLZ_Header *) buf, PGLZ_strategy_default); + /* Zero is returned for incompressible data */ + if(!ret) + return NULL; + } + else + elog(ERROR, "Wrong value for full_page_compression GUC"); + + /* + * We recheck the actual size even if pglz_compress(),pg_LZ4_compress() and + * pg_snappy_compress report success, + * because it might be satisfied with having saved as little as one byte + * in the compressed data --- which could turn into a net loss once you + * consider header and alignment padding. Worst case, the compressed + * format might require three padding bytes (plus header, which is + * included in VARSIZE(buf)), whereas the uncompressed format would take + * only one header byte and no padding if the value is short enough. So + * we insist on a savings of more than 2 bytes to ensure we have a gain. + */ + if(VARSIZE(buf) >= orig_len - 2) + { + return NULL; + } + *len = VARSIZE(buf); + return (char *) buf; +} + +/* * Get a pointer to the right location in the WAL buffer containing the * given XLogRecPtr. * @@ -4069,6 +4159,50 @@ RestoreBackupBlockContents(XLogRecPtr lsn, BkpBlock bkpb, char *blk, { memcpy((char *) page, blk, BLCKSZ); } + + /* Decompress if backup block is compressed */ + + else if (bkpb.block_compression != BACKUP_BLOCK_COMPRESSION_OFF) + { + if (bkpb.block_compression == BACKUP_BLOCK_COMPRESSION_SNAPPY) + { + int ret; + size_t compressed_length = VARSIZE((struct varlena *) blk) - VARHDRSZ; + char *compressed_data = (char *)VARDATA((struct varlena *) blk); + size_t s_uncompressed_length; + + ret = snappy_uncompressed_length(compressed_data,compressed_length, + &s_uncompressed_length); + if (!ret) + elog(ERROR, "snappy: failed to determine compression length"); + if (BLCKSZ != s_uncompressed_length) + elog(ERROR, "snappy: compression size mismatch %d != %zu", + BLCKSZ, s_uncompressed_length); + + ret = snappy_uncompress(compressed_data, + compressed_length, + page); + if (ret != 0) + elog(ERROR, "snappy: decompression failed: %d", ret); + } + else if (bkpb.block_compression == BACKUP_BLOCK_COMPRESSION_LZ4) + { + int ret; + size_t compressed_length = VARSIZE((struct varlena *) blk) - VARHDRSZ; + char *compressed_data = (char *)VARDATA((struct varlena *) blk); + ret = LZ4_decompress_fast(compressed_data, page,BLCKSZ); + if (ret != compressed_length) + elog(ERROR, "lz4: decompression size mismatch: %d vs %zu", ret, + compressed_length); + } + else if (bkpb.block_compression == BACKUP_BLOCK_COMPRESSION_PGLZ) + { + /* Checks to see if decompression is successful is made inside the function */ + pglz_decompress((PGLZ_Header *) blk, (char *) page); + } + else + elog(ERROR, "Wrong value for compression algorithm in block header"); + } else { memcpy((char *) page, blk, bkpb.hole_offset); @@ -7233,6 +7367,9 @@ StartupXLOG(void) */ if (fast_promoted) RequestCheckpoint(CHECKPOINT_FORCE); + + /* Allocate pages to store compressed backup blocks */ + CompressBackupBlockPagesAlloc(); } /* @@ -7626,9 +7763,48 @@ InitXLOGAccess(void) /* Use GetRedoRecPtr to copy the RedoRecPtr safely */ (void) GetRedoRecPtr(); + + /* Allocate pages to store compressed backup blocks */ + CompressBackupBlockPagesAlloc(); } /* + * Allocate pages to store compressed backup blocks once per backend. + * Size of pages depend on the compression algorithm used. These pages + * persist till the end of the backend process. If memory allocation + * fails we disable compression of backup blocks entirely. + */ +void +CompressBackupBlockPagesAlloc(void) +{ + + if (full_page_compression != BACKUP_BLOCK_COMPRESSION_OFF && + !compressed_pages_allocated) + { + size_t buffer_size = VARHDRSZ; + int j; + if (full_page_compression == BACKUP_BLOCK_COMPRESSION_SNAPPY) + buffer_size += snappy_max_compressed_length(BLCKSZ); + else if (full_page_compression == BACKUP_BLOCK_COMPRESSION_LZ4) + buffer_size += LZ4_compressBound(BLCKSZ); + else if (full_page_compression == BACKUP_BLOCK_COMPRESSION_PGLZ) + buffer_size += PGLZ_MAX_OUTPUT(BLCKSZ); + for (j = 0; j < XLR_MAX_BKP_BLOCKS; j++) + { + compressed_pages[j] = (char *) malloc(buffer_size); + if(compressed_pages[j] == NULL) + { + full_page_compression = BACKUP_BLOCK_COMPRESSION_OFF; + break; + } + } + if(j == XLR_MAX_BKP_BLOCKS) + { + compressed_pages_allocated=true; + } + } +} +/* * Return the current Redo pointer from shared memory. * * As a side-effect, the local RedoRecPtr copy is updated. diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 3a31a75..93bc994 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -403,6 +403,22 @@ static const struct config_enum_entry huge_pages_options[] = { {NULL, 0, false} }; + +/* + * Although only "off" in documented, we accept all likely variants of "off". + * "on" is indicated by specifying name of compression algorithm. + */ +static const struct config_enum_entry backup_block_compression_options[] = { + {"off", BACKUP_BLOCK_COMPRESSION_OFF, false}, + {"false", BACKUP_BLOCK_COMPRESSION_OFF, true}, + {"no", BACKUP_BLOCK_COMPRESSION_OFF, true}, + {"0", BACKUP_BLOCK_COMPRESSION_OFF, true}, + {"pglz", BACKUP_BLOCK_COMPRESSION_PGLZ, true}, + {"snappy", BACKUP_BLOCK_COMPRESSION_SNAPPY, true}, + {"lz4", BACKUP_BLOCK_COMPRESSION_LZ4, true}, + {NULL, 0, false} +}; + /* * Options for enum values stored in other modules */ @@ -3510,6 +3526,16 @@ static struct config_enum ConfigureNamesEnum[] = NULL, NULL, NULL }, + { + {"full_page_compression", PGC_SIGHUP, WAL_SETTINGS, + gettext_noop("Compress backup block in WAL using specified compression algorithm."), + NULL + }, + &full_page_compression, + BACKUP_BLOCK_COMPRESSION_OFF, backup_block_compression_options, + NULL, NULL, NULL + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, 0, NULL, NULL, NULL, NULL diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 85f9cb7..6f4de44 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -177,6 +177,14 @@ typedef enum RECOVERY_TARGET_IMMEDIATE } RecoveryTargetType; +typedef enum +{ + BACKUP_BLOCK_COMPRESSION_OFF, + BACKUP_BLOCK_COMPRESSION_PGLZ, + BACKUP_BLOCK_COMPRESSION_SNAPPY, + BACKUP_BLOCK_COMPRESSION_LZ4 +} BackupBlockCompressionAlgorithm; + extern XLogRecPtr XactLastRecEnd; extern bool reachedConsistency; @@ -190,6 +198,7 @@ extern bool XLogArchiveMode; extern char *XLogArchiveCommand; extern bool EnableHotStandby; extern bool fullPageWrites; +extern int full_page_compression; extern bool wal_log_hints; extern bool log_checkpoints; extern int num_xloginsert_locks; diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h index 8c8de38..17db3ee 100644 --- a/src/include/access/xlog_internal.h +++ b/src/include/access/xlog_internal.h @@ -48,7 +48,7 @@ typedef struct BkpBlock BlockNumber block; /* block number */ uint16 hole_offset; /* number of bytes before "hole" */ uint16 hole_length; /* number of bytes in "hole" */ - + uint16 block_compression; /*information about compression of block */ /* ACTUAL BLOCK DATA FOLLOWS AT END OF STRUCT */ } BkpBlock;