Correct and stabilize some replication tests

Adding pg_stat_clear_snapshot() in functions looping over
gp_stat_replication / pg_stat_replication to refresh result everytime
the query is run as part of same transaction. Without
pg_stat_clear_snapshot() query result is not refreshed for
pg_stat_activity neither for xx_stat_replication functions on multiple
invocations inside a transaction. So, in absence of it the tests
become flaky.

Also, tests commit_blocking_on_standby and dtx_recovery_wait_lsn were
initially committed with wrong expectations, hence were missing to
test the intended behavior. Now reflect the correct expectation.

(cherry picked from commit c565e988)

src/test/isolation2/expected/reader_waits_for_lock.out

@@ -3,7 +3,7 @@
 -- lock was held by some other session.
 
 -- setup
-CREATE or REPLACE FUNCTION check_readers_are_blocked () RETURNS bool AS $$ declare retries int; /* in func */ begin retries := 1200; /* in func */ loop if (SELECT count(*) > 0 as reader_waits from pg_locks l join pg_stat_activity a on a.pid = l.pid and a.query like '%reader_waits_for_lock_table%' and not a.pid = pg_backend_pid() and l.granted = false and l.mppiswriter = false) then return true; /* in func */ end if; /* in func */ if retries <= 0 then return false; /* in func */ end if; /* in func */ perform pg_sleep(0.1); /* in func */ retries := retries - 1; /* in func */ end loop; /* in func */ end; /* in func */ $$ language plpgsql;
+CREATE or REPLACE FUNCTION check_readers_are_blocked () RETURNS bool AS $$ declare retries int; /* in func */ begin retries := 1200; /* in func */ loop if (SELECT count(*) > 0 as reader_waits from pg_locks l join pg_stat_activity a on a.pid = l.pid and a.query like '%reader_waits_for_lock_table%' and not a.pid = pg_backend_pid() and l.granted = false and l.mppiswriter = false) then return true; /* in func */ end if; /* in func */ if retries <= 0 then return false; /* in func */ end if; /* in func */ perform pg_sleep(0.1); /* in func */ perform pg_stat_clear_snapshot(); /* in func */ retries := retries - 1; /* in func */ end loop; /* in func */ end; /* in func */ $$ language plpgsql;
 CREATE
 1: create table reader_waits_for_lock_table(a int, b int) distributed by (a);
 CREATE

src/test/isolation2/expected/segwalrep/commit_blocking_on_standby.out

@@ -8,6 +8,9 @@
 
 -- Check that are starting with a clean slate, standby must be in sync
 -- with master.
+include: helpers/server_helpers.sql;
+CREATE
+
 select application_name, state from pg_stat_replication;
  application_name | state     
 ------------------+-----------
@@ -121,13 +124,10 @@ show repl_catchup_within_range;
 begin;
 BEGIN
 
-create or replace function wait_until_standby_in_state(targetstate text) returns void as $$ declare replstate text; /* in func */ begin loop select state into replstate from pg_stat_replication; /* in func */ exit when replstate = targetstate; /* in func */ perform pg_sleep(0.1); /* in func */ end loop; /* in func */ end; /* in func */ $$ language plpgsql;
-CREATE
-
 select wait_until_standby_in_state('catchup');
  wait_until_standby_in_state 
 -----------------------------
-                             
+ catchup                     
 (1 row)
 
 select gp_wait_until_triggered_fault('wal_sender_loop', 1, dbid) from gp_segment_configuration where content=-1 and role='p';
@@ -205,5 +205,5 @@ INSERT 1
 select wait_until_standby_in_state('streaming');
  wait_until_standby_in_state 
 -----------------------------
-                             
+ streaming                   
 (1 row)

src/test/isolation2/expected/segwalrep/dtm_recovery_on_standby.out

@@ -121,8 +121,8 @@ select reinitialize_standby();
 -- end_ignore
 
 -- Sync state between master and standby must be restored at the end.
-select wait_until_master_standby_insync();
- wait_until_master_standby_insync 
-----------------------------------
- OK                               
+select wait_until_standby_in_state('streaming');
+ wait_until_standby_in_state 
+-----------------------------
+ streaming                   
 (1 row)

src/test/isolation2/helpers/server_helpers.sql

@@ -41,7 +41,6 @@ returns text as $$
         raise PgCtlError(stdout+'|'+stderr)
 $$ language plpythonu;
 
-
 --
 -- pg_ctl_start:
 --
@@ -130,7 +129,7 @@ $$ language sql;
 
 create or replace function wait_until_segment_synchronized(segment_number int) returns text as $$
 begin
-	for i in 1..600 loop
+	for i in 1..1200 loop
 		if (select count(*) = 0 from gp_segment_configuration where content = segment_number and mode != 's') then
 			return 'OK';
 		end if;
@@ -143,7 +142,7 @@ $$ language plpgsql;
 
 create or replace function wait_until_all_segments_synchronized() returns text as $$
 begin
-	for i in 1..600 loop
+	for i in 1..1200 loop
 		if (select count(*) = 0 from gp_segment_configuration where content != -1 and mode != 's') then
 			return 'OK';
 		end if;
@@ -154,18 +153,6 @@ begin
 end;
 $$ language plpgsql;
 
-create or replace function wait_until_master_standby_insync() returns text as $$
-begin
-	for i in 1..1200 loop
-		if (select count(*) = 1 from pg_stat_replication) then
-			return 'OK';
-		end if;
-		perform pg_sleep(0.1);
-	end loop;
-	return 'Fail';
-end;
-$$ language plpgsql;
-
 create or replace function wait_for_replication_replay (segid int, retries int) returns bool as
 $$
 declare
@@ -184,7 +171,28 @@ begin
 		   return false;
 		end if;
 		perform pg_sleep(0.1);
+		perform pg_stat_clear_snapshot();
 		i := i + 1;
 	end loop;
 end;
 $$ language plpgsql;
+
+create or replace function wait_until_standby_in_state(targetstate text)
+returns text as $$
+declare
+   replstate text;
+   i int;
+begin
+   i := 0;
+   while i < 1200 loop
+      select state into replstate from pg_stat_replication;
+      if replstate = targetstate then
+          return replstate;
+      end if;
+      perform pg_sleep(0.1);
+      perform pg_stat_clear_snapshot();
+      i := i + 1;
+   end loop;
+   return replstate;
+end;
+$$ language plpgsql;

src/test/isolation2/sql/reader_waits_for_lock.sql

@@ -22,6 +22,7 @@ begin
       return false; /* in func */
     end if; /* in func */
     perform pg_sleep(0.1); /* in func */
+    perform pg_stat_clear_snapshot(); /* in func */
     retries := retries - 1; /* in func */
   end loop; /* in func */
 end; /* in func */

src/test/isolation2/sql/segwalrep/commit_blocking_on_standby.sql

@@ -8,6 +8,8 @@
 
 -- Check that are starting with a clean slate, standby must be in sync
 -- with master.
+include: helpers/server_helpers.sql;
+
 select application_name, state from pg_stat_replication;
 
 -- Inject fault on standby to skip WAL flush.
@@ -99,19 +101,6 @@ show repl_catchup_within_range;
 -- not block.
 begin;
 
-create or replace function wait_until_standby_in_state(targetstate text)
-returns void as $$
-declare
-   replstate text; /* in func */
-begin
-   loop
-      select state into replstate from pg_stat_replication; /* in func */
-      exit when replstate = targetstate; /* in func */
-      perform pg_sleep(0.1); /* in func */
-   end loop; /* in func */
-end; /* in func */
-$$ language plpgsql;
-
 select wait_until_standby_in_state('catchup');
 
 select gp_wait_until_triggered_fault('wal_sender_loop', 1, dbid)

src/test/isolation2/sql/segwalrep/dtm_recovery_on_standby.sql

@@ -98,4 +98,4 @@ select reinitialize_standby();
 -- end_ignore
 
 -- Sync state between master and standby must be restored at the end.
-select wait_until_master_standby_insync();
+select wait_until_standby_in_state('streaming');

src/test/regress/expected/mirror_replay.out

@@ -39,6 +39,7 @@ begin
       return false;
     end if;
     perform pg_sleep(0.1);
+    perform pg_stat_clear_snapshot();
     i := i + 1;
   end loop;
 end;

src/test/regress/sql/mirror_replay.sql

@@ -42,6 +42,7 @@ begin
       return false;
     end if;
     perform pg_sleep(0.1);
+    perform pg_stat_clear_snapshot();
     i := i + 1;
   end loop;
 end;