Wait for a command to show up in pg_stat_activity

It takes non-zero amount of time after a command is dispatched from a
client until it appears in pg_stat_activity.  The test must wait
before validating anything based on pg_stat_activity.  The wait logic
was already added for one instance of such validation.  This patch
addres the wait logic for the remaining instance of validation.

Also found a way to avoid creating one table, while at it.

Reviewed-by: Shaoqi Bai and Adam Berlin

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

@@ -21,7 +21,12 @@ select gp_inject_fault_infinite2('walrecv_skip_flush', 'skip', dbid, hostname, p
  Success:                  
 (1 row)
 
--- Generate some WAL to trigger the fault
+begin;
+BEGIN
+create or replace function wait_for_pg_stat_activity(timeout_secs int) returns void as $$ declare c int; /* in func */ i int; /* in func */ begin c := 0; /* in func */ i := 0; /* in func */ while c < 1 and i < timeout_secs*2 loop select count(*) into c from pg_stat_activity where waiting_reason = 'replication'; /* in func */ perform pg_sleep(0.5); /* in func */ perform pg_stat_clear_snapshot(); /* in func */ i := i + 1; /* in func */ end loop; /* in func */ if c < 1 then raise exception 'timeout waiting for command to get blocked'; /* in func */ end if; /* in func */ end; /* in func */ $$ language plpgsql;
+CREATE
+
+-- Flush WAL to trigger the fault on standby.
 checkpoint;
 CHECKPOINT
 
@@ -37,9 +42,11 @@ select gp_wait_until_triggered_fault2('walrecv_skip_flush', 1, dbid, hostname, p
 
 -- The create table command should be seen as blocked.  Wait until
 -- that happens.
-do $$ declare c int; /* in func */ i int; /* in func */ begin c := 0; /* in func */ i := 0; /* in func */ while c < 1 and i < 120 loop select count(*) into c from pg_stat_activity where waiting_reason = 'replication'; /* in func */ perform pg_sleep(0.5); /* in func */ i := i + 1; /* in func */ end loop; /* in func */ if i = 120 then raise exception 'timeout waiting for command to get blocked'; /* in func */ end if; /* in func */ end; /* in func */ $$;
-DO
-
+select wait_for_pg_stat_activity(60);
+ wait_for_pg_stat_activity 
+---------------------------
+                           
+(1 row)
 select datname, waiting_reason, query from pg_stat_activity where waiting_reason = 'replication';
  datname        | waiting_reason | query                                                                  
 ----------------+----------------+------------------------------------------------------------------------
@@ -53,8 +60,8 @@ select gp_inject_fault2('walrecv_skip_flush', 'reset', dbid, hostname, port) fro
 (1 row)
 
 -- Ensure that commits are no longer blocked.
-create table commit_blocking_on_standby_t2 (a int) distributed by (a);
-CREATE
+commit;
+COMMIT
 
 1<:  <... completed>
 CREATE
@@ -165,13 +172,18 @@ select gp_wait_until_triggered_fault2( 'wal_sender_after_caughtup_within_range',
 
 -- Should block because standby is considered to have caughtup within
 -- range.
-1&: create table commit_blocking_on_standby_t3 (a int) distributed by (a);  <waiting ...>
+1&: create table commit_blocking_on_standby_t2 (a int) distributed by (a);  <waiting ...>
 
 -- The create table command should be seen as blocked.
+select wait_for_pg_stat_activity(60);
+ wait_for_pg_stat_activity 
+---------------------------
+                           
+(1 row)
 select datname, waiting_reason, query from pg_stat_activity where waiting_reason = 'replication';
  datname        | waiting_reason | query                                                                  
 ----------------+----------------+------------------------------------------------------------------------
- isolation2test | replication    | create table commit_blocking_on_standby_t3 (a int) distributed by (a); 
+ isolation2test | replication    | create table commit_blocking_on_standby_t2 (a int) distributed by (a); 
 (1 row)
 
 -- Reset faults on primary as well as mirror.
@@ -187,7 +199,7 @@ CREATE
 
 -- Create table transaction must have committed and the table should
 -- be ready for insert.
-insert into commit_blocking_on_standby_t3 values (1);
+insert into commit_blocking_on_standby_t2 values (1);
 INSERT 1
 
 select wait_until_standby_in_state('streaming');

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

@@ -14,37 +14,41 @@ select application_name, state, sync_state from pg_stat_replication;
 select gp_inject_fault_infinite2('walrecv_skip_flush', 'skip', dbid, hostname, port)
 from gp_segment_configuration where content=-1 and role='m';
 
--- Generate some WAL to trigger the fault
-checkpoint;
-
-select gp_wait_until_triggered_fault2('walrecv_skip_flush', 1, dbid, hostname, port)
-from gp_segment_configuration where content=-1 and role='m';
-
--- Should block in commit (SyncrepWaitForLSN()), waiting for commit
--- LSN to be flushed on standby.
-1&: create table commit_blocking_on_standby_t1 (a int) distributed by (a);
-
--- The create table command should be seen as blocked.  Wait until
--- that happens.
-do $$
+begin;
+create or replace function wait_for_pg_stat_activity(timeout_secs int)
+returns void as $$
 declare
   c int; /* in func */
   i int; /* in func */
 begin
   c := 0; /* in func */
   i := 0; /* in func */
-  while c < 1 and i < 120 loop
+  while c < 1 and i < timeout_secs*2 loop
     select count(*) into c from pg_stat_activity
     where waiting_reason = 'replication'; /* in func */
     perform pg_sleep(0.5); /* in func */
+    perform pg_stat_clear_snapshot(); /* in func */
     i := i + 1; /* in func */
   end loop; /* in func */
-  if i = 120 then
+  if c < 1 then
     raise exception 'timeout waiting for command to get blocked'; /* in func */
   end if; /* in func */
 end; /* in func */
-$$;
+$$ language plpgsql;
+
+-- Flush WAL to trigger the fault on standby.
+checkpoint;
 
+select gp_wait_until_triggered_fault2('walrecv_skip_flush', 1, dbid, hostname, port)
+from gp_segment_configuration where content=-1 and role='m';
+
+-- Should block in commit (SyncrepWaitForLSN()), waiting for commit
+-- LSN to be flushed on standby.
+1&: create table commit_blocking_on_standby_t1 (a int) distributed by (a);
+
+-- The create table command should be seen as blocked.  Wait until
+-- that happens.
+select wait_for_pg_stat_activity(60);
 select datname, waiting_reason, query from pg_stat_activity
 where waiting_reason = 'replication';
 
@@ -52,7 +56,7 @@ select gp_inject_fault2('walrecv_skip_flush', 'reset', dbid, hostname, port)
 from gp_segment_configuration where content=-1 and role='m';
 
 -- Ensure that commits are no longer blocked.
-create table commit_blocking_on_standby_t2 (a int) distributed by (a);
+commit;
 
 1<:
 
@@ -140,9 +144,10 @@ select gp_wait_until_triggered_fault2(
 
 -- Should block because standby is considered to have caughtup within
 -- range.
-1&: create table commit_blocking_on_standby_t3 (a int) distributed by (a);
+1&: create table commit_blocking_on_standby_t2 (a int) distributed by (a);
 
 -- The create table command should be seen as blocked.
+select wait_for_pg_stat_activity(60);
 select datname, waiting_reason, query from pg_stat_activity
 where waiting_reason = 'replication';
 
@@ -154,6 +159,6 @@ select gp_inject_fault2('all', 'reset', dbid, hostname, port)
 
 -- Create table transaction must have committed and the table should
 -- be ready for insert.
-insert into commit_blocking_on_standby_t3 values (1);
+insert into commit_blocking_on_standby_t2 values (1);
 
 select wait_until_standby_in_state('streaming');