DROP SCHEMA IF EXISTS inplace_upgrade CASCADE; CREATE SCHEMA inplace_upgrade;
CREATE OR REPLACE FUNCTION inplace_upgrade.pgxc_stat_get_wal_senders(OUT nodename text,
OUT sender_pid int, OUT local_role text, OUT peer_role text, OUT peer_state text, OUT state text,
OUT sender_sent_location text, OUT sender_write_location text, OUT sender_flush_location text,
OUT sender_replay_location text, OUT receiver_received_location text, OUT receiver_write_location text,
OUT receiver_flush_location text, OUT receiver_replay_location text)
RETURNS setof record
AS $$
DECLARE
  row_data record;
  row_name record;
  query_str text;
  query_str_nodes text;
BEGIN
  query_str_nodes := 'SELECT node_name FROM pg_catalog.pgxc_node';
  FOR row_name IN EXECUTE(query_str_nodes) LOOP
    query_str := 'EXECUTE DIRECT ON (' || row_name.node_name || ') ''SELECT * FROM pg_catalog.pg_stat_get_wal_senders() where sender_flush_location != receiver_replay_location ''';
    FOR row_data IN EXECUTE(query_str) LOOP
      nodename = row_name.node_name;
      sender_pid = row_data.sender_pid;
      local_role = row_data.local_role;
      peer_role = row_data.peer_role;
      peer_state = row_data.peer_state;
      state = row_data.state;
      sender_sent_location = row_data.sender_sent_location;
      sender_write_location = row_data.sender_write_location;
      sender_flush_location = row_data.sender_flush_location;
      sender_replay_location = row_data.sender_replay_location;
      receiver_received_location = row_data.receiver_received_location;
      receiver_write_location = row_data.receiver_write_location;
      receiver_flush_location = row_data.receiver_flush_location;
      receiver_replay_location = row_data.receiver_replay_location;
      return next;
    END LOOP;
  END LOOP;
  return;
END; $$
LANGUAGE 'plpgsql';