Optimization of vacuum for logical replication

diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index f1ff01e..1009341 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -719,7 +719,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
 	 * We need to log the copied data in WAL iff WAL archiving/streaming is
 	 * enabled AND it's a WAL-logged rel.
 	 */
-	use_wal = XLogIsNeeded() && RelationNeedsWAL(NewHeap);
+	use_wal = XLogPhysicalIsNeeded() && RelationNeedsWAL(NewHeap);
 
 	/* use_wal off requires smgr_targblock be initially invalid */
 	Assert(RelationGetTargetBlock(NewHeap) == InvalidBlockNumber);
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index a3c4a1d..831d87b 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -466,7 +466,7 @@ vacuum_log_cleanup_info(Relation rel, LVRelStats *vacrelstats)
 	 * Skip this for relations for which no WAL is to be written, or if we're
 	 * not trying to support archive recovery.
 	 */
-	if (!RelationNeedsWAL(rel) || !XLogIsNeeded())
+	if (!RelationNeedsWAL(rel) || !XLogPhysicalIsNeeded())
 		return;
 
 	/*
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index ab19692..4ce1632 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -577,7 +577,7 @@ _bt_leafbuild(BTSpool *btspool, BTSpool *btspool2)
 	 * We need to log index creation in WAL iff WAL archiving/streaming is
 	 * enabled UNLESS the index isn't WAL-logged anyway.
 	 */
-	wstate.btws_use_wal = XLogIsNeeded() && RelationNeedsWAL(wstate.index);
+	wstate.btws_use_wal = XLogPhysicalIsNeeded() && RelationNeedsWAL(wstate.index);
 
 	/* reserve the metapage */
 	wstate.btws_pages_alloced = BTREE_METAPAGE + 1;
diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.c
index 33060f3..83c4427 100644
--- a/src/backend/access/rmgrdesc/xlogdesc.c
+++ b/src/backend/access/rmgrdesc/xlogdesc.c
@@ -30,6 +30,7 @@ const struct config_enum_entry wal_level_options[] = {
 	{"archive", WAL_LEVEL_REPLICA, true},	/* deprecated */
 	{"hot_standby", WAL_LEVEL_REPLICA, true},	/* deprecated */
 	{"logical", WAL_LEVEL_LOGICAL, false},
+	{"logical_only", WAL_LEVEL_LOGICAL_ONLY, false},
 	{NULL, 0, false}
 };
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index e651a84..f881ea8 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -6155,7 +6155,13 @@ CheckRequiredParameterValues(void)
 	{
 		ereport(WARNING,
 				(errmsg("WAL was generated with wal_level=minimal, data may be missing"),
-				 errhint("This happens if you temporarily set wal_level=minimal without taking a new base backup.")));
+				 errhint("This happens if you temporarily set wal_level=minimal or logical_only without taking a new base backup.")));
+	}
+	if (ArchiveRecoveryRequested && ControlFile->wal_level == WAL_LEVEL_LOGICAL_ONLY)
+	{
+		ereport(WARNING,
+				(errmsg("WAL was generated with wal_level=logical_only, data may be missing"),
+				 errhint("This happens if you temporarily set wal_level=logical_only or logical_only without taking a new base backup.")));
 	}
 
 	/*
@@ -6164,7 +6170,7 @@ CheckRequiredParameterValues(void)
 	 */
 	if (ArchiveRecoveryRequested && EnableHotStandby)
 	{
-		if (ControlFile->wal_level < WAL_LEVEL_REPLICA)
+		if (ControlFile->wal_level < WAL_LEVEL_REPLICA || ControlFile->wal_level == WAL_LEVEL_LOGICAL_ONLY)
 			ereport(ERROR,
 					(errmsg("hot standby is not possible because wal_level was not set to \"replica\" or higher on the master server"),
 					 errhint("Either set wal_level to \"replica\" on the master, or turn off hot_standby here.")));
@@ -10207,7 +10213,7 @@ do_pg_start_backup(const char *backupidstr, bool fast, TimeLineID *starttli_p,
 	 * During recovery, we don't need to check WAL level. Because, if WAL
 	 * level is not sufficient, it's impossible to get here during recovery.
 	 */
-	if (!backup_started_in_recovery && !XLogIsNeeded())
+	if (!backup_started_in_recovery && !XLogPhysicalIsNeeded())
 		ereport(ERROR,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("WAL level not sufficient for making an online backup"),
@@ -10736,7 +10742,7 @@ do_pg_stop_backup(char *labelfile, bool waitforarchive, TimeLineID *stoptli_p)
 	 * During recovery, we don't need to check WAL level. Because, if WAL
 	 * level is not sufficient, it's impossible to get here during recovery.
 	 */
-	if (!backup_started_in_recovery && !XLogIsNeeded())
+	if (!backup_started_in_recovery && !XLogPhysicalIsNeeded())
 		ereport(ERROR,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("WAL level not sufficient for making an online backup"),
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index 4795c6f..31ab26a 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -316,7 +316,7 @@ pg_create_restore_point(PG_FUNCTION_ARGS)
 				 (errmsg("recovery is in progress"),
 				  errhint("WAL control functions cannot be executed during recovery."))));
 
-	if (!XLogIsNeeded())
+	if (!XLogPhysicalIsNeeded())
 		ereport(ERROR,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("WAL level not sufficient for creating a restore point"),
diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c
index 3cc886f..d8708e9 100644
--- a/src/backend/catalog/storage.c
+++ b/src/backend/catalog/storage.c
@@ -326,7 +326,7 @@ RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
 	 * We need to log the copied data in WAL iff WAL archiving/streaming is
 	 * enabled AND it's a permanent relation.
 	 */
-	use_wal = XLogIsNeeded() &&
+	use_wal = XLogPhysicalIsNeeded() &&
 		(relpersistence == RELPERSISTENCE_PERMANENT || copying_initfork);
 
 	nblocks = smgrnblocks(src, forkNum);
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index f115d4b..7fc0ca9 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -232,7 +232,7 @@ CreatePublication(CreatePublicationStmt *stmt)
 
 	InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0);
 
-	if (wal_level != WAL_LEVEL_LOGICAL)
+	if (wal_level < WAL_LEVEL_LOGICAL)
 	{
 		ereport(WARNING,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index cceefbd..32f75e3 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -4772,7 +4772,7 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
 		bistate = GetBulkInsertState();
 
 		ti_options = TABLE_INSERT_SKIP_FSM;
-		if (!XLogIsNeeded())
+		if (!XLogPhysicalIsNeeded())
 			ti_options |= TABLE_INSERT_SKIP_WAL;
 	}
 	else
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index 3339804..ca01fd7 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -896,6 +896,9 @@ PostmasterMain(int argc, char *argv[])
 	if (XLogArchiveMode > ARCHIVE_MODE_OFF && wal_level == WAL_LEVEL_MINIMAL)
 		ereport(ERROR,
 				(errmsg("WAL archival cannot be enabled when wal_level is \"minimal\"")));
+	if (XLogArchiveMode > ARCHIVE_MODE_OFF && wal_level == WAL_LEVEL_LOGICAL_ONLY)
+		ereport(ERROR,
+				(errmsg("WAL archival cannot be enabled when wal_level is \"logical_only\"")));
 	if (max_wal_senders > 0 && wal_level == WAL_LEVEL_MINIMAL)
 		ereport(ERROR,
 				(errmsg("WAL streaming (max_wal_senders > 0) requires wal_level \"replica\" or \"logical\"")));
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index b1bcd93..7ea26bb 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1523,6 +1523,12 @@ RestoreSlotFromDisk(const char *name)
 				 errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
 						NameStr(cp.slotdata.name)),
 				 errhint("Change wal_level to be logical or higher.")));
+	else if (wal_level == WAL_LEVEL_LOGICAL_ONLY)
+		ereport(FATAL,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("physical replication slot \"%s\" exists, but wal_level = logical_only",
+						NameStr(cp.slotdata.name)),
+				 errhint("Change wal_level to be replica or logical.")));
 	else if (wal_level < WAL_LEVEL_REPLICA)
 		ereport(FATAL,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 23870a2..9dd5e2f 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -542,6 +542,11 @@ StartReplication(StartReplicationCmd *cmd)
 	StringInfoData buf;
 	XLogRecPtr	FlushPtr;
 
+	if (wal_level == WAL_LEVEL_LOGICAL_ONLY)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 (errmsg("cannot start physical replication in logical_only WAl mode"))));
+
 	if (ThisTimeLineID == 0)
 		ereport(ERROR,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index ff17804..d584dcf 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -80,6 +80,8 @@ wal_level_str(WalLevel wal_level)
 			return "replica";
 		case WAL_LEVEL_LOGICAL:
 			return "logical";
+		case WAL_LEVEL_LOGICAL_ONLY:
+			return "logical_only";
 	}
 	return _("unrecognized wal_level");
 }
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index d519252..a44c5b3 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -161,17 +161,18 @@ typedef enum WalLevel
 {
 	WAL_LEVEL_MINIMAL = 0,
 	WAL_LEVEL_REPLICA,
-	WAL_LEVEL_LOGICAL
+	WAL_LEVEL_LOGICAL,
+	WAL_LEVEL_LOGICAL_ONLY
 } WalLevel;
 
 extern PGDLLIMPORT int wal_level;
 
 /* Is WAL archiving enabled (always or only while server is running normally)? */
 #define XLogArchivingActive() \
-	(AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF)
+	(AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || XLogPhysicalIsNeeded()), XLogArchiveMode > ARCHIVE_MODE_OFF)
 /* Is WAL archiving enabled always (even during recovery)? */
 #define XLogArchivingAlways() \
-	(AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode == ARCHIVE_MODE_ALWAYS)
+	(AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || XLogPhysicalIsNeeded()), XLogArchiveMode == ARCHIVE_MODE_ALWAYS)
 #define XLogArchiveCommandSet() (XLogArchiveCommand[0] != '\0')
 
 /*
@@ -181,6 +182,11 @@ extern PGDLLIMPORT int wal_level;
 #define XLogIsNeeded() (wal_level >= WAL_LEVEL_REPLICA)
 
 /*
+ * Is WAL-logging necessary for archival or physical replication?
+ */
+#define XLogPhysicalIsNeeded() (XLogIsNeeded() && wal_level != WAL_LEVEL_LOGICAL_ONLY)
+
+/*
  * Is a full-page image needed for hint bit updates?
  *
  * Normally, we don't WAL-log hint bit updates, but if checksums are enabled,