[BACKPORT 2024.1][#23661] YSQL: Disable all wal_level checks for logi…

…cal replication

Summary:
##### Backport Description
Clean merge. No conflicts.

##### Original Description
Original commit: 3f64090 / D37567
The wal_level parameter determines what gets logged in the PG WAL. This in-turn determines, what gets streamed during logical replication.

Since, we don't use the PG WAL, this parameter isn't applicable to YSQL. Hence, remove the various checks found across the codebase.
Jira: DB-12572

Test Plan:
Jenkins: test regex: .*ReplicationSlot.*

./yb_build.sh --java-test 'org.yb.pgsql.TestPgReplicationSlot#replicationConnectionCreateDrop'

Reviewers: asrinivasan

Reviewed By: asrinivasan

Subscribers: yql, ycdcxcluster

Differential Revision: https://phorge.dev.yugabyte.com/D37605

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgReplicationSlot.java

@@ -35,6 +35,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized.Parameters;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.yb.YBTestRunner;
@@ -109,16 +110,25 @@ public void createAndDropFromDifferentTservers() throws Exception {
 
   @Test
   public void replicationConnectionCreateDrop() throws Exception {
-    Connection conn =
-        getConnectionBuilder().withTServer(0).replicationConnect();
-    PGReplicationConnection replConnection = conn.unwrap(PGConnection.class).getReplicationAPI();
+    markClusterNeedsRecreation();
 
-    replConnection.createReplicationSlot()
-        .logical()
-        .withSlotName("test_slot_repl_conn")
-        .withOutputPlugin(YB_OUTPUT_PLUGIN_NAME)
-        .make();
-    replConnection.dropReplicationSlot("test_slot_repl_conn");
+    String[] wal_levels = {"minimal", "replica", "logical"};
+    for (String wal_level : wal_levels) {
+      LOG.info("Testing replicationConnectionCreateDrop with wal_level = {}", wal_level);
+      Map<String, String> tserverFlags = super.getTServerFlags();
+      tserverFlags.put("ysql_pg_conf", String.format("wal_level=%s", wal_level));
+      restartClusterWithFlags(Collections.emptyMap(), tserverFlags);
+
+      Connection conn = getConnectionBuilder().withTServer(0).replicationConnect();
+      PGReplicationConnection replConnection = conn.unwrap(PGConnection.class).getReplicationAPI();
+
+      replConnection.createReplicationSlot()
+          .logical()
+          .withSlotName("test_slot_repl_conn")
+          .withOutputPlugin(YB_OUTPUT_PLUGIN_NAME)
+          .make();
+      replConnection.dropReplicationSlot("test_slot_repl_conn");
+    }
   }
 
   @Test

src/postgres/src/backend/postmaster/postmaster.c

@@ -933,7 +933,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 (max_wal_senders > 0 && wal_level == WAL_LEVEL_MINIMAL)
+	/* YB NOTE: wal_level isn't applicable in YSQL since don't use the PG WAL */
+	if (!YBIsEnabledInPostgresEnvVar() && max_wal_senders > 0 &&
+		wal_level == WAL_LEVEL_MINIMAL)
 		ereport(ERROR,
 				(errmsg("WAL streaming (max_wal_senders > 0) requires wal_level \"replica\" or \"logical\"")));
 

src/postgres/src/backend/replication/slot.c

@@ -1156,7 +1156,8 @@ CheckSlotRequirements(void)
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 (errmsg("replication slots can only be used if max_replication_slots > 0"))));
 
-	if (wal_level < WAL_LEVEL_REPLICA)
+	/* YB NOTE: wal_level is not applicable to YSQL. */
+	if (!IsYugaByteEnabled() && wal_level < WAL_LEVEL_REPLICA)
 		ereport(ERROR,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("replication slots can only be used if wal_level >= replica")));
@@ -1563,6 +1564,10 @@ RestoreSlotFromDisk(const char *name)
 	int			readBytes;
 	pg_crc32c	checksum;
 
+	/* Should never be called in YSQL. */
+	if (IsYugaByteEnabled())
+		Assert(false);
+
 	/* no need to lock here, no concurrent access allowed yet */
 
 	/* delete temp file if it exists */