diff --git a/src/bin/pg_basebackup/t/010_pg_basebackup.pl b/src/bin/pg_basebackup/t/010_pg_basebackup.pl
index 3d1a4ddd5c..dbfa35c610 100644
--- a/src/bin/pg_basebackup/t/010_pg_basebackup.pl
+++ b/src/bin/pg_basebackup/t/010_pg_basebackup.pl
@@ -178,6 +178,17 @@ my $baseUnloggedPath = $node->safe_psql('postgres',
 ok(-f "$pgdata/${baseUnloggedPath}_init", 'unlogged init fork in base');
 ok(-f "$pgdata/$baseUnloggedPath",        'unlogged main fork in base');
 
+# Create an compressed table to test that pca and pcd file are copied.
+$node->safe_psql('postgres', 'CREATE TABLE base_compressed (id int) WITH(compresstype=pglz)');
+
+my $baseCompressedPath = $node->safe_psql('postgres',
+	q{select pg_relation_filepath('base_compressed')});
+
+# Make sure compressed relation's address and data files exist
+ok(-f "$pgdata/$baseCompressedPath",       'compressed main fork in base');
+ok(-f "$pgdata/${baseCompressedPath}_pca", 'compressed address file in base');
+ok(-f "$pgdata/${baseCompressedPath}_pcd", 'compressed data file in base');
+
 # Create files that look like temporary relations to ensure they are ignored.
 my $postgresOid = $node->safe_psql('postgres',
 	q{select oid from pg_database where datname = 'postgres'});
@@ -238,6 +249,14 @@ ok(-f "$tempdir/backup/${baseUnloggedPath}_init",
 ok( !-f "$tempdir/backup/$baseUnloggedPath",
 	'unlogged main fork not in backup');
 
+# Compressed relation fork's address and data should be copied
+ok(-f "$tempdir/backup/${baseCompressedPath}",
+	'compressed main fork in backup');
+ok(-f "$tempdir/backup/${baseCompressedPath}_pca",
+	'compressed address file in backup');
+ok(-f "$tempdir/backup/${baseCompressedPath}_pcd",
+	'compressed data file in backup');
+
 # Temp relations should not be copied.
 foreach my $filename (@tempRelationFiles)
 {
diff --git a/src/bin/pg_checksums/pg_checksums.c b/src/bin/pg_checksums/pg_checksums.c
index dc20122c89..ce411c440c 100644
--- a/src/bin/pg_checksums/pg_checksums.c
+++ b/src/bin/pg_checksums/pg_checksums.c
@@ -31,6 +31,8 @@
 #include "storage/bufpage.h"
 #include "storage/checksum.h"
 #include "storage/checksum_impl.h"
+#include "storage/page_compression.h"
+#include "storage/page_compression_impl.h"
 
 
 static int64 files_scanned = 0;
@@ -181,18 +183,67 @@ skipfile(const char *fn)
 }
 
 static void
-scan_file(const char *fn, int segmentno)
+scan_file(const char *fn, int segmentno, bool is_compressed_datafile)
 {
 	PGAlignedBlock buf;
+	PGAlignedCompressChunks pcchunks_buf;
 	PageHeader	header = (PageHeader) buf.data;
-	int			f;
+	int			f,
+				f_pca;
 	BlockNumber blockno;
 	int			flags;
 	int64		blocks_written_in_file = 0;
+	char		fn_pca[MAXPGPATH];
+	PageCompressHeader *pcmap = (PageCompressHeader *) NULL;
+	PageCompressAddr *pcaddr = (PageCompressAddr *) NULL;
 
 	Assert(mode == PG_MODE_ENABLE ||
 		   mode == PG_MODE_CHECK);
 
+	/* init memory map of compression page address file */
+	if (is_compressed_datafile)
+	{
+		int			r;
+		PageCompressHeader pchdr;
+
+		strlcpy(fn_pca, fn, sizeof(fn_pca));
+		fn_pca[strlen(fn_pca) - 1] = 'a';
+
+		f_pca = open(fn_pca, PG_BINARY | O_RDONLY, 0);
+
+		if (f_pca < 0)
+			pg_fatal("could not open file \"%s\": %m", fn_pca);
+
+		r = read(f_pca, &pchdr, sizeof(PageCompressHeader));
+		if (r != sizeof(PageCompressHeader))
+		{
+			if (r < 0)
+				pg_fatal("could not read file \"%s\": %m",
+						 fn_pca);
+			else
+				pg_fatal("could not read file \"%s\": read %d of %zu",
+						 fn_pca, r, sizeof(PageCompressHeader));
+		}
+
+		if (pchdr.chunk_size != BLCKSZ / 2 &&
+			pchdr.chunk_size != BLCKSZ / 4 &&
+			pchdr.chunk_size != BLCKSZ / 8 &&
+			pchdr.chunk_size != BLCKSZ / 16)
+		{
+			pg_fatal("the page compression address file header is corrupted in file \"%s\"",
+					 fn_pca);
+		}
+
+		pcmap = pc_mmap(f_pca, pchdr.chunk_size, true);
+		if (pcmap == MAP_FAILED)
+		{
+			pg_fatal("could not mmap file \"%s\": %m",
+					 fn_pca);
+		}
+
+		close(f_pca);
+	}
+
 	flags = (mode == PG_MODE_ENABLE) ? O_RDWR : O_RDONLY;
 	f = open(fn, PG_BINARY | flags, 0);
 
@@ -204,18 +255,121 @@ scan_file(const char *fn, int segmentno)
 	for (blockno = 0;; blockno++)
 	{
 		uint16		csum;
-		int			r = read(f, buf.data, BLCKSZ);
+		int			r;
+		off_t		first_pc_chunk_pos = -1;
 
-		if (r == 0)
-			break;
-		if (r != BLCKSZ)
+		if (is_compressed_datafile)
 		{
-			if (r < 0)
-				pg_fatal("could not read block %u in file \"%s\": %m",
-						 blockno, fn);
+			char		compressed_data_buf[BLCKSZ + BLCKSZ / 2];
+			char	   *compressed_data = compressed_data_buf;
+			int			nbytes,
+						i,
+						read_amount,
+						range;
+			off_t		seekpos;
+			char	   *buffer_pos;
+
+			if (blockno >= pcmap->nblocks)
+				break;
+
+			pcaddr = GetPageCompressAddr(pcmap, pcmap->chunk_size, blockno);
+
+			/* New pages have no checksum yet */
+			if (pcaddr->nchunks == 0)
+			{
+				blocks_scanned++;
+				continue;
+			}
+
+			/* read all compressed chunk for this block */
+			for (i = 0; i < pcaddr->nchunks; i++)
+			{
+				buffer_pos = pcchunks_buf.data + pcmap->chunk_size * i;
+
+				seekpos = (off_t) OffsetOfPageCompressChunk(pcmap->chunk_size, pcaddr->chunknos[i]);
+				if (i == 0)
+					first_pc_chunk_pos = seekpos;
+
+				range = 1;
+				while (i < pcaddr->nchunks - 1 && pcaddr->chunknos[i + 1] == pcaddr->chunknos[i] + 1)
+				{
+					range++;
+					i++;
+				}
+				read_amount = pcmap->chunk_size * range;
+
+				if (lseek(f, seekpos, SEEK_SET) < 0)
+					pg_fatal("seek failed for block %u in file \"%s\": %m", blockno, fn);
+
+				nbytes = read(f, buffer_pos, read_amount);
+
+				if (nbytes != read_amount)
+				{
+					if (nbytes < 0)
+						pg_fatal("could not read block %u in file \"%s\": %m",
+								 blockno, fn);
+					else
+						pg_fatal("could not read block %u in file \"%s\": read %d of %d",
+								 blockno, fn, nbytes, read_amount);
+				}
+			}
+
+			r = pcmap->chunk_size * pcaddr->nchunks;
+
+			/* build decompress buffer  */
+			if (pcaddr->nchunks > 1)
+			{
+				for (int i = 0; i < pcaddr->nchunks; i++)
+				{
+					memcpy(compressed_data + StoreCapacityPerPageCompressChunk(pcmap->chunk_size) * i,
+						   pcchunks_buf.data + pcmap->chunk_size * i + SizeOfPageCompressChunkHeaderData,
+						   StoreCapacityPerPageCompressChunk(pcmap->chunk_size));
+				}
+			}
 			else
-				pg_fatal("could not read block %u in file \"%s\": read %d of %d",
-						 blockno, fn, r, BLCKSZ);
+				compressed_data = pcchunks_buf.data + SizeOfPageCompressChunkHeaderData;
+
+			/* decompress chunk data */
+			nbytes = decompress_page(compressed_data, buf.data, pcmap->algorithm);
+			if (nbytes != BLCKSZ)
+			{
+				if (nbytes == -2)
+				{
+					pg_fatal("could not recognized compression algorithm %d for file \"%s\"",
+							 pcmap->algorithm, fn);
+				}
+				else
+				{
+					pg_log_error("could not decompress block %u in file \"%s\": decompressed %d of %d bytes",
+								 blockno, fn, nbytes, BLCKSZ);
+
+					if (mode == PG_MODE_ENABLE)
+						exit(1);
+					else
+					{
+						blocks_scanned++;
+						badblocks++;
+						current_size += r;
+						continue;
+					}
+				}
+			}
+		}
+		else
+		{
+			r = read(f, buf.data, BLCKSZ);
+
+			if (r == 0)
+				break;
+			if (r != BLCKSZ)
+			{
+				if (r < 0)
+					pg_fatal("could not read block %u in file \"%s\": %m",
+							 blockno, fn);
+				else
+					pg_fatal("could not read block %u in file \"%s\": read %d of %d",
+							 blockno, fn, r, BLCKSZ);
+			}
 		}
 		blocks_scanned++;
 
@@ -244,7 +398,8 @@ scan_file(const char *fn, int segmentno)
 		}
 		else if (mode == PG_MODE_ENABLE)
 		{
-			int			w;
+			int			w,
+						i;
 
 			/*
 			 * Do not rewrite if the checksum is already set to the expected
@@ -258,20 +413,75 @@ scan_file(const char *fn, int segmentno)
 			/* Set checksum in page header */
 			header->pd_checksum = csum;
 
-			/* Seek back to beginning of block */
-			if (lseek(f, -BLCKSZ, SEEK_CUR) < 0)
-				pg_fatal("seek failed for block %u in file \"%s\": %m", blockno, fn);
+			if (is_compressed_datafile)
+			{
+				/* Seek back to beginning of first chunk */
+				if (lseek(f, first_pc_chunk_pos + SizeOfPageCompressChunkHeaderData, SEEK_SET) < 0)
+				{
+					pg_fatal("seek failed for block %u in file \"%s\": %m", blockno, fn);
+				}
+
+				/* Write block with checksum */
+				w = write(f, header, sizeof(PageHeaderData));
+				if (w != sizeof(PageHeaderData))
+				{
+					if (w < 0)
+						pg_fatal("could not write block %u in file \"%s\": %m",
+								 blockno, fn);
+					else
+						pg_fatal("could not write block %u in file \"%s\": wrote %d of %d",
+								 blockno, fn, w, (int) sizeof(PageHeaderData));
+				}
+
+				/* write checksum of chunks for this block */
+				for (i = 0; i < pcaddr->nchunks; i++)
+				{
+					PageCompressChunk *pcchunk;
+					off_t		seekpos;
+					char	   *buffer_pos;
+
+					/* calculate checksum of chunk */
+					buffer_pos = pcchunks_buf.data + pcmap->chunk_size * i;
+					pcchunk = (PageCompressChunk *) buffer_pos;
 
-			/* Write block with checksum */
-			w = write(f, buf.data, BLCKSZ);
-			if (w != BLCKSZ)
+					csum = pg_checksum_compress_chunk(buffer_pos, pcmap->chunk_size, pcaddr->chunknos[i]);
+					pcchunk->checksum = csum;
+
+					/* Seek back to beginning of chunk */
+					seekpos = (off_t) OffsetOfPageCompressChunk(pcmap->chunk_size, pcaddr->chunknos[i]);
+					if (lseek(f, seekpos, SEEK_SET) < 0)
+						pg_fatal("seek failed for block %u in file \"%s\": %m", blockno, fn);
+
+					/* Write block with checksum */
+					w = write(f, pcchunk, SizeOfPageCompressChunkHeaderData);
+					if (w != SizeOfPageCompressChunkHeaderData)
+					{
+						if (w < 0)
+							pg_fatal("could not write block %u in file \"%s\": %m",
+									 blockno, fn);
+						else
+							pg_fatal("could not write block %u in file \"%s\": wrote %d of %d",
+									 blockno, fn, w, (int) SizeOfPageCompressChunkHeaderData);
+					}
+				}
+			}
+			else
 			{
-				if (w < 0)
-					pg_fatal("could not write block %u in file \"%s\": %m",
-							 blockno, fn);
-				else
-					pg_fatal("could not write block %u in file \"%s\": wrote %d of %d",
-							 blockno, fn, w, BLCKSZ);
+				/* Seek back to beginning of block */
+				if (lseek(f, -BLCKSZ, SEEK_CUR) < 0)
+					pg_fatal("seek failed for block %u in file \"%s\": %m", blockno, fn);
+
+				/* Write block with checksum */
+				w = write(f, buf.data, BLCKSZ);
+				if (w != BLCKSZ)
+				{
+					if (w < 0)
+						pg_fatal("could not write block %u in file \"%s\": %m",
+								 blockno, fn);
+					else
+						pg_fatal("could not write block %u in file \"%s\": wrote %d of %d",
+								 blockno, fn, w, BLCKSZ);
+				}
 			}
 		}
 
@@ -295,6 +505,8 @@ scan_file(const char *fn, int segmentno)
 	}
 
 	close(f);
+	if (is_compressed_datafile)
+		pc_munmap(pcmap);
 }
 
 /*
@@ -346,17 +558,32 @@ scan_directory(const char *basedir, const char *subdir, bool sizeonly)
 			char	   *forkpath,
 					   *segmentpath;
 			int			segmentno = 0;
+			size_t		filenamelength;
+			bool		is_compressed_datafile = false;
 
 			if (skipfile(de->d_name))
 				continue;
 
+			/* check if is address or data file for compressed relation */
+			strlcpy(fnonly, de->d_name, sizeof(fnonly));
+			filenamelength = strlen(fnonly);
+			if (filenamelength >= 4)
+			{
+				if (strncmp(fnonly + filenamelength - 4, "_pca", 4) == 0)
+					continue;
+				else if (strncmp(fnonly + filenamelength - 4, "_pcd", 4) == 0)
+				{
+					is_compressed_datafile = true;
+					fnonly[filenamelength - 4] = '\0';
+				}
+			}
+
 			/*
 			 * Cut off at the segment boundary (".") to get the segment number
 			 * in order to mix it into the checksum. Then also cut off at the
 			 * fork boundary, to get the filenode the file belongs to for
 			 * filtering.
 			 */
-			strlcpy(fnonly, de->d_name, sizeof(fnonly));
 			segmentpath = strchr(fnonly, '.');
 			if (segmentpath != NULL)
 			{
@@ -382,7 +609,7 @@ scan_directory(const char *basedir, const char *subdir, bool sizeonly)
 			 * the items in the data folder.
 			 */
 			if (!sizeonly)
-				scan_file(fn, segmentno);
+				scan_file(fn, segmentno, is_compressed_datafile);
 		}
 #ifndef WIN32
 		else if (S_ISDIR(st.st_mode) || S_ISLNK(st.st_mode))
diff --git a/src/bin/pg_checksums/t/002_actions.pl b/src/bin/pg_checksums/t/002_actions.pl
index 0309cbbaa3..a810f8e1f1 100644
--- a/src/bin/pg_checksums/t/002_actions.pl
+++ b/src/bin/pg_checksums/t/002_actions.pl
@@ -85,6 +85,84 @@ sub check_relation_corruption
 	return;
 }
 
+# Utility routine to create and check a compressed table with corrupted checksums
+# on a wanted tablespace.  Note that this stops and starts the node
+# multiple times to perform the checks, leaving the node started
+# at the end.
+sub check_compressed_relation_corruption
+{
+	my $node       = shift;
+	my $table      = shift;
+	my $tablespace = shift;
+	my $pgdata     = $node->data_dir;
+
+	$node->safe_psql(
+		'postgres',
+		"CREATE TABLE $table (a int) WITH(compresstype=pglz, autovacuum_enabled=false);
+		 INSERT INTO $table SELECT a FROM generate_series(1,10000)a;");
+
+	$node->safe_psql('postgres',
+		"ALTER TABLE " . $table . " SET TABLESPACE " . $tablespace . ";");
+
+	my $file_corrupted =
+	  $node->safe_psql('postgres', "SELECT pg_relation_filepath('$table');") . "_pcd";
+	my $relfilenode_corrupted = $node->safe_psql('postgres',
+		"SELECT relfilenode FROM pg_class WHERE relname = '$table';");
+
+	# Set page header and block size
+	my $pageheader_size = 24;
+	my $block_size = $node->safe_psql('postgres', 'SHOW block_size;');
+	$node->stop;
+
+	# Checksums are correct for single relfilenode as the table is not
+	# corrupted yet.
+	command_ok(
+		[
+			'pg_checksums', '--check',
+			'-D',           $pgdata,
+			'--filenode',   $relfilenode_corrupted
+		],
+		"succeeds for single relfilenode on tablespace $tablespace with offline cluster"
+	);
+
+	# Time to create some corruption
+	open my $file, '+<', "$pgdata/$file_corrupted";
+	seek($file, $pageheader_size, 0);
+	syswrite($file, "\0\0\0\0\0\0\0\0\0");
+	close $file;
+
+	# Checksum checks on single relfilenode fail
+	$node->command_checks_all(
+		[
+			'pg_checksums', '--check',
+			'-D',           $pgdata,
+			'--filenode',   $relfilenode_corrupted
+		],
+		1,
+		[qr/Bad checksums:.*1/],
+		[qr/could not decompress block/],
+		"fails with corrupted data for single relfilenode on tablespace $tablespace"
+	);
+
+	# Global checksum checks fail as well
+	$node->command_checks_all(
+		[ 'pg_checksums', '--check', '-D', $pgdata ],
+		1,
+		[qr/Bad checksums:.*1/],
+		[qr/could not decompress block/],
+		"fails with corrupted data on tablespace $tablespace");
+
+	# Drop corrupted table again and make sure there is no more corruption.
+	$node->start;
+	$node->safe_psql('postgres', "DROP TABLE $table;");
+	$node->stop;
+	$node->command_ok([ 'pg_checksums', '--check', '-D', $pgdata ],
+		"succeeds again after table drop on tablespace $tablespace");
+
+	$node->start;
+	return;
+}
+
 # Initialize node with checksums disabled.
 my $node = PostgreSQL::Test::Cluster->new('node_checksum');
 $node->init();
@@ -114,6 +192,15 @@ append_to_file "$pgdata/global/pgsql_tmp/1.1",        "foo";
 append_to_file "$pgdata/global/pg_internal.init",     "foo";
 append_to_file "$pgdata/global/pg_internal.init.123", "foo";
 
+# Create a compressed table and index before vertify enable checksums
+$node->start;
+$node->safe_psql(
+	'postgres',
+	"CREATE TABLE compressed_table(a int) WITH(compresstype=pglz, autovacuum_enabled=false);
+	 INSERT INTO compressed_table SELECT a FROM generate_series(1,10000)a;
+	 CREATE INDEX ON compressed_table(a) WITH(compresstype=pglz);");
+$node->stop;
+
 # Enable checksums.
 command_ok([ 'pg_checksums', '--enable', '--no-sync', '-D', $pgdata ],
 	"checksums successfully enabled in cluster");
@@ -195,6 +282,7 @@ command_fails([ 'pg_checksums', '--check', '-D', $pgdata ],
 
 # Check corruption of table on default tablespace.
 check_relation_corruption($node, 'corrupt1', 'pg_default');
+check_compressed_relation_corruption($node, 'compressed_corrupt1', 'pg_default');
 
 # Create tablespace to check corruptions in a non-default tablespace.
 my $basedir        = $node->basedir;
@@ -203,6 +291,7 @@ mkdir($tablespace_dir);
 $node->safe_psql('postgres',
 	"CREATE TABLESPACE ts_corrupt LOCATION '$tablespace_dir';");
 check_relation_corruption($node, 'corrupt2', 'ts_corrupt');
+check_compressed_relation_corruption($node, 'compressed_corrupt2', 'ts_corrupt');
 
 # Utility routine to check that pg_checksums is able to detect
 # correctly-named relation files filled with some corrupted data.
diff --git a/src/bin/pg_rewind/t/010_page_compression.pl b/src/bin/pg_rewind/t/010_page_compression.pl
new file mode 100644
index 0000000000..59a3dbb238
--- /dev/null
+++ b/src/bin/pg_rewind/t/010_page_compression.pl
@@ -0,0 +1,204 @@
+
+# Copyright (c) 2021-2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use RewindTest;
+
+sub run_test
+{
+	my $test_mode = shift;
+
+	RewindTest::setup_cluster($test_mode);
+	RewindTest::start_primary();
+
+	# Create a test table and insert a row in primary.
+	primary_psql("CREATE TABLE tbl1 (d text) WITH(compresstype=pglz, compress_chunk_size=1024, compress_prealloc_chunks=2)");
+	primary_psql("INSERT INTO tbl1 VALUES ('in primary')");
+
+	# This test table will be used to test truncation, i.e. the table
+	# is extended in the old primary after promotion
+	primary_psql("CREATE TABLE trunc_tbl (d text) WITH(compresstype=pglz, compress_chunk_size=1024, compress_prealloc_chunks=2)");
+	primary_psql("INSERT INTO trunc_tbl VALUES ('in primary')");
+
+	# This test table will be used to test the "copy-tail" case, i.e. the
+	# table is truncated in the old primary after promotion
+	primary_psql("CREATE TABLE tail_tbl (id integer, d text) WITH(compresstype=pglz, compress_chunk_size=1024, compress_prealloc_chunks=2)");
+	primary_psql("INSERT INTO tail_tbl VALUES (0, 'in primary')");
+
+	# This test table is dropped in the old primary after promotion.
+	primary_psql("CREATE TABLE drop_tbl (d text) WITH(compresstype=pglz, compress_chunk_size=1024, compress_prealloc_chunks=2)");
+	primary_psql("INSERT INTO drop_tbl VALUES ('in primary')");
+
+	primary_psql("CHECKPOINT");
+
+	RewindTest::create_standby($test_mode);
+
+	# Insert additional data on primary that will be replicated to standby
+	primary_psql("INSERT INTO tbl1 values ('in primary, before promotion')");
+	primary_psql(
+		"INSERT INTO trunc_tbl values ('in primary, before promotion')");
+	primary_psql(
+		"INSERT INTO tail_tbl SELECT g, 'in primary, before promotion: ' || g FROM generate_series(1, 10000) g"
+	);
+
+	primary_psql('CHECKPOINT');
+
+	RewindTest::promote_standby();
+
+	# Insert a row in the old primary. This causes the primary and standby
+	# to have "diverged", it's no longer possible to just apply the
+	# standy's logs over primary directory - you need to rewind.
+	primary_psql("INSERT INTO tbl1 VALUES ('in primary, after promotion')");
+
+	# Also insert a new row in the standby, which won't be present in the
+	# old primary.
+	standby_psql("INSERT INTO tbl1 VALUES ('in standby, after promotion')");
+
+	# Insert enough rows to trunc_tbl to extend the file. pg_rewind should
+	# truncate it back to the old size.
+	primary_psql(
+		"INSERT INTO trunc_tbl SELECT 'in primary, after promotion: ' || g FROM generate_series(1, 10000) g"
+	);
+
+	# Truncate tail_tbl. pg_rewind should copy back the truncated part
+	# (We cannot use an actual TRUNCATE command here, as that creates a
+	# whole new relfilenode)
+	primary_psql("DELETE FROM tail_tbl WHERE id > 10");
+	primary_psql("VACUUM tail_tbl");
+
+	# Drop drop_tbl. pg_rewind should copy it back.
+	primary_psql(
+		"insert into drop_tbl values ('in primary, after promotion')");
+	primary_psql("DROP TABLE drop_tbl");
+
+	# Create new_tbl in standby. pg_rewind should copy it from source.
+	standby_psql("CREATE TABLE new_tbl (d text) WITH(compresstype=pglz, compress_chunk_size=1024, compress_prealloc_chunks=2)");
+	standby_psql("INSERT INTO new_tbl VALUES ('in standby')");
+
+	# Before running pg_rewind, do a couple of extra tests with several
+	# option combinations.  As the code paths taken by those tests
+	# do not change for the "local" and "remote" modes, just run them
+	# in "local" mode for simplicity's sake.
+	if ($test_mode eq 'local')
+	{
+		my $primary_pgdata  = $node_primary->data_dir;
+		my $standby_pgdata = $node_standby->data_dir;
+
+		# First check that pg_rewind fails if the target cluster is
+		# not stopped as it fails to start up for the forced recovery
+		# step.
+		command_fails(
+			[
+				'pg_rewind',       '--debug',
+				'--source-pgdata', $standby_pgdata,
+				'--target-pgdata', $primary_pgdata,
+				'--no-sync'
+			],
+			'pg_rewind with running target');
+
+		# Again with --no-ensure-shutdown, which should equally fail.
+		# This time pg_rewind complains without attempting to perform
+		# recovery once.
+		command_fails(
+			[
+				'pg_rewind',       '--debug',
+				'--source-pgdata', $standby_pgdata,
+				'--target-pgdata', $primary_pgdata,
+				'--no-sync',       '--no-ensure-shutdown'
+			],
+			'pg_rewind --no-ensure-shutdown with running target');
+
+		# Stop the target, and attempt to run with a local source
+		# still running.  This fails as pg_rewind requires to have
+		# a source cleanly stopped.
+		$node_primary->stop;
+		command_fails(
+			[
+				'pg_rewind',       '--debug',
+				'--source-pgdata', $standby_pgdata,
+				'--target-pgdata', $primary_pgdata,
+				'--no-sync',       '--no-ensure-shutdown'
+			],
+			'pg_rewind with unexpected running source');
+
+		# Stop the target cluster cleanly, and run again pg_rewind
+		# with --dry-run mode.  If anything gets generated in the data
+		# folder, the follow-up run of pg_rewind will most likely fail,
+		# so keep this test as the last one of this subset.
+		$node_standby->stop;
+		command_ok(
+			[
+				'pg_rewind',       '--debug',
+				'--source-pgdata', $standby_pgdata,
+				'--target-pgdata', $primary_pgdata,
+				'--no-sync',       '--dry-run'
+			],
+			'pg_rewind --dry-run');
+
+		# Both clusters need to be alive moving forward.
+		$node_standby->start;
+		$node_primary->start;
+	}
+
+	RewindTest::run_pg_rewind($test_mode);
+
+	check_query(
+		'SELECT * FROM tbl1',
+		qq(in primary
+in primary, before promotion
+in standby, after promotion
+),
+		'table content');
+
+	check_query(
+		'SELECT * FROM trunc_tbl',
+		qq(in primary
+in primary, before promotion
+),
+		'truncation');
+
+	check_query(
+		'SELECT count(*) FROM tail_tbl',
+		qq(10001
+),
+		'tail-copy');
+
+	check_query(
+		'SELECT * FROM drop_tbl',
+		qq(in primary
+),
+		'drop');
+
+	check_query(
+		'SELECT * FROM new_tbl',
+		qq(in standby
+),
+		'new');
+
+	# Permissions on PGDATA should be default
+  SKIP:
+	{
+		skip "unix-style permissions not supported on Windows", 1
+		  if ($windows_os);
+
+		ok(check_mode_recursive($node_primary->data_dir(), 0700, 0600),
+			'check PGDATA permissions');
+	}
+
+	RewindTest::clean_rewind_test();
+	return;
+}
+
+# Run the test in both modes
+run_test('local');
+run_test('remote');
+run_test('archive');
+
+done_testing();
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index f265e043e9..5743c28914 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1278,6 +1278,10 @@ static const char *const table_storage_parameters[] = {
 	"autovacuum_vacuum_insert_threshold",
 	"autovacuum_vacuum_scale_factor",
 	"autovacuum_vacuum_threshold",
+	"compress_chunk_size",
+	"compress_prealloc_chunks",
+	"compresslevel",
+	"compresstype",
 	"fillfactor",
 	"log_autovacuum_min_duration",
 	"parallel_workers",
@@ -2000,6 +2004,8 @@ psql_completion(const char *text, int start, int end)
 	/* ALTER INDEX <foo> SET|RESET ( */
 	else if (Matches("ALTER", "INDEX", MatchAny, "RESET", "("))
 		COMPLETE_WITH("fillfactor",
+					  "compress_prealloc_chunks",
+					  "compresslevel",
 					  "deduplicate_items",	/* BTREE */
 					  "fastupdate", "gin_pending_list_limit",	/* GIN */
 					  "buffering",	/* GiST */
@@ -2007,6 +2013,8 @@ psql_completion(const char *text, int start, int end)
 			);
 	else if (Matches("ALTER", "INDEX", MatchAny, "SET", "("))
 		COMPLETE_WITH("fillfactor =",
+					  "compress_prealloc_chunks =",
+					  "compresslevel =",
 					  "deduplicate_items =",	/* BTREE */
 					  "fastupdate =", "gin_pending_list_limit =",	/* GIN */
 					  "buffering =",	/* GiST */