Subject: Improve WALRead() to suck data directly from WAL buffers when possible

From: Bharath Rupireddy <[email-masked]>
Date: Fri, 9 Dec 2022 14:33:39 +0530

Hi,

WALRead() currently reads WAL from the WAL file on the disk, which
means, the walsenders serving streaming and logical replication
(callers of WALRead()) will have to hit the disk/OS's page cache for
reading the WAL. This may increase the amount of read IO required for
all the walsenders put together as one typically maintains many
standbys/subscribers on production servers for high availability,
disaster recovery, read-replicas and so on. Also, it may increase
replication lag if all the WAL reads are always hitting the disk.

It may happen that WAL buffers contain the requested WAL, if so, the
WALRead() can attempt to read from the WAL buffers first before
reading from the file. If the read hits the WAL buffers, then reading
from the file on disk is avoided. This mainly reduces the read IO/read
system calls. It also enables us to do other features specified
elsewhere [1].

I'm attaching a patch that implements the idea which is also noted
elsewhere [2]. I've run some tests [3]. The WAL buffers hit ratio with
the patch stood at 95%, in other words, the walsenders avoided 95% of
the time reading from the file. The benefit, if measured in terms of
the amount of data - 79% (13.5GB out of total 17GB) of the requested
WAL is read from the WAL buffers as opposed to 21% from the file. Note
that the WAL buffers hit ratio can be very low for write-heavy
workloads, in which case, file reads are inevitable.

The patch introduces concurrent readers for the WAL buffers, so far
only there are concurrent writers. In the patch, WALRead() takes just
one lock (WALBufMappingLock) in shared mode to enable concurrent
readers and does minimal things - checks if the requested WAL page is
present in WAL buffers, if so, copies the page and releases the lock.
I think taking just WALBufMappingLock is enough here as the concurrent
writers depend on it to initialize and replace a page in WAL buffers.

I'll add this to the next commitfest.

Thoughts?

[1] https://www.postgresql.org/message-id/CALj2ACXCSM%2BsTR%3D5NNRtmSQr3g1Vnr-yR91azzkZCaCJ7u4d4w%40mail.gmail.com

[2]
 * XXX probably this should be improved to suck data directly from the
 * WAL buffers when possible.
 */
bool
WALRead(XLogReaderState *state,

[3]
1 primary, 1 sync standby, 1 async standby
./pgbench --initialize --scale=300 postgres
./pgbench --jobs=16 --progress=300 --client=32 --time=900
--username=ubuntu postgres

PATCHED:
-[ RECORD 1 ]----------+----------------
application_name       | assb1
wal_read               | 31005
wal_read_bytes         | 3800607104
wal_read_time          | 779.402
wal_read_buffers       | 610611
wal_read_bytes_buffers | 14493226440
wal_read_time_buffers  | 3033.309
sync_state             | async
-[ RECORD 2 ]----------+----------------
application_name       | ssb1
wal_read               | 31027
wal_read_bytes         | 3800932712
wal_read_time          | 696.365
wal_read_buffers       | 610580
wal_read_bytes_buffers | 14492900832
wal_read_time_buffers  | 2989.507
sync_state             | sync

HEAD:
-[ RECORD 1 ]----+----------------
application_name | assb1
wal_read         | 705627
wal_read_bytes   | 18343480640
wal_read_time    | 7607.783
sync_state       | async
-[ RECORD 2 ]----+------------
application_name | ssb1
wal_read         | 705625
wal_read_bytes   | 18343480640
wal_read_time    | 4539.058
sync_state       | sync

--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com

---

From: Kyotaro Horiguchi <[email-masked]>
Date: Mon, 12 Dec 2022 11:57:17 +0900 (JST)

At Fri, 9 Dec 2022 14:33:39 +0530, Bharath Rupireddy <[email-masked]> wrote in 
> I'll add this to the next commitfest.

This adds copying of the whole page (at least) at every WAL *record*
read, fighting all WAL writers by taking WALBufMappingLock on a very
busy page while the copying. I'm a bit doubtful that it results in an
overall improvement. It seems to me almost all pread()s here happens
on file buffer so it is unclear to me that copying a whole WAL page
(then copying the target record again) wins over a pread() call that
copies only the record to read. Do you have an actual number of how
frequent WAL reads go to disk, or the actual number of performance
gain or real I/O reduction this patch offers?

This patch copies the bleeding edge WAL page without recording the
(next) insertion point nor checking whether all in-progress insertion
behind the target LSN have finished. Thus the copied page may have
holes.  That being said, the sequential-reading nature and the fact
that WAL buffers are zero-initialized may make it work for recovery,
but I don't think this also works for replication.

I remember that the one of the advantage of reading the on-memory WAL
records is that that allows walsender to presend the unwritten
records. So perhaps we should manage how far the buffer is filled with
valid content (or how far we can presend) in this feature.

regards.

-- 
Kyotaro Horiguchi
NTT Open Source Software Center

---

From: Kyotaro Horiguchi <[email-masked]>
Date: Mon, 12 Dec 2022 12:06:36 +0900 (JST)

At Mon, 12 Dec 2022 11:57:17 +0900 (JST), Kyotaro Horiguchi <[email-masked]> wrote in 
> but I don't think this also works for replication.

Mmm. I'm a bit dim. Recovery doesn't read concurrently-written
records. Please forget about recovery.

---

From: Kyotaro Horiguchi <[email-masked]>
Date: Mon, 12 Dec 2022 12:08:51 +0900 (JST)

Sorry for the confusion.

At Mon, 12 Dec 2022 12:06:36 +0900 (JST), Kyotaro Horiguchi <[email-masked]> wrote in 
> records. Please forget about recovery.

NO... Logical walsenders do that. So, please forget about this...

regards.

---

From: Bharath Rupireddy <[email-masked]>
Date: Fri, 23 Dec 2022 15:45:55 +0530

On Mon, Dec 12, 2022 at 8:27 AM Kyotaro Horiguchi
<[email-masked]> wrote:

Thanks for providing thoughts.

> read,

In the worst case yes, but that may not always be true. On a typical
production server with decent write traffic, it happens that the
callers of WALRead() read a full WAL page of size XLOG_BLCKSZ bytes or
MAX_SEND_SIZE bytes.

> fighting all WAL writers by taking WALBufMappingLock on a very
> overall improvement.

Well, the tests don't reflect that [1], I've run an insert work load
[2]. The WAL is being read from WAL buffers 99% of the time, which is
pretty cool. If you have any use-cases in mind, please share them
and/or feel free to run at your end.

> It seems to me almost all pread()s here happens
> copies only the record to read.

That's not always guaranteed. Imagine a typical production server with
decent write traffic and heavy analytical queries (which fills OS page
cache with the table pages accessed for the queries), the WAL pread()
calls turn to IOPS. Despite the WAL being present in WAL buffers,
customers will be paying unnecessarily for these IOPS too. With the
patch, we are basically avoiding the pread() system calls which may
turn into IOPS on production servers (99% of the time for the insert
use case [1][2], 95% of the time for pgbench use case specified
upthread). With the patch, WAL buffers can act as L1 cache, if one
calls OS page cache as L2 cache (of course this illustration is not
related to the typical processor L1 and L2 ... caches).

> Do you have an actual number of how

It might be a bit tough to generate such heavy traffic. An idea is to
ensure the WAL page/file goes out of the OS page cache before
WALRead() - these might help here - 0002 patch from
https://www.postgresql.org/message-id/CA%2BhUKGLmeyrDcUYAty90V_YTcoo5kAFfQjRQ-_1joS_%3DX7HztA%40mail.gmail.com
and tool https://github.com/klando/pgfincore.

> but I don't think this also works for replication.

WALRead() callers are smart enough to take the flushed bytes only.
Although they read the whole WAL page, they calculate the valid bytes.

> valid content (or how far we can presend) in this feature.

Yes, the non-flushed WAL can be read and sent across if one wishes to
to make replication faster and parallel flushing on primary and
standbys at the cost of a bit of extra crash handling, that's
mentioned here https://www.postgresql.org/message-id/CALj2ACXCSM%2BsTR%3D5NNRtmSQr3g1Vnr-yR91azzkZCaCJ7u4d4w%40mail.gmail.com.
However, this can be a separate discussion.

I also want to reiterate that the patch implemented a TODO item:

[1]
PATCHED:
1 1470.329907
2 1437.096329
4 2966.096948
8 5978.441040
16 11405.538255
32 22933.546058
64 43341.870038
128 73623.837068
256 104754.248661
512 115746.359530
768 106106.691455
1024 91900.079086
2048 84134.278589
4096 62580.875507

-[ RECORD 1 ]----------+-----------
application_name       | assb1
sent_lsn               | 0/1B8106A8
write_lsn              | 0/1B8106A8
flush_lsn              | 0/1B8106A8
replay_lsn             | 0/1B8106A8
write_lag              |
flush_lag              |
replay_lag             |
wal_read               | 104
wal_read_bytes         | 10733008
wal_read_time          | 1.845
wal_read_buffers       | 76662
wal_read_bytes_buffers | 383598808
wal_read_time_buffers  | 205.418
sync_state             | async

HEAD:
1 1312.054496
2 1449.429321
4 2717.496207
8 5913.361540
16 10762.978907
32 19653.449728
64 41086.124269
128 68548.061171
256 104468.415361
512 114328.943598
768 91751.279309
1024 96403.736757
2048 82155.140270
4096 66160.659511

-[ RECORD 1 ]----+-----------
application_name | assb1
sent_lsn         | 0/1AB5BCB8
write_lsn        | 0/1AB5BCB8
flush_lsn        | 0/1AB5BCB8
replay_lsn       | 0/1AB5BCB8
write_lag        |
flush_lag        |
replay_lag       |
wal_read         | 71967
wal_read_bytes   | 381009080
wal_read_time    | 243.616
sync_state       | async

[2] Test details:
./configure --prefix=$PWD/inst/ CFLAGS="-O3" > install.log && make -j
8 install > install.log 2>&1 &
1 primary, 1 async standby
cd inst/bin
./pg_ctl -D data -l logfile stop
./pg_ctl -D assbdata -l logfile1 stop
rm -rf data assbdata
rm logfile logfile1
free -m
sudo su -c 'sync; echo 3 > /proc/sys/vm/drop_caches'
free -m
./initdb -D data
rm -rf /home/ubuntu/archived_wal
mkdir /home/ubuntu/archived_wal
cat << EOF >> data/postgresql.conf
shared_buffers = '8GB'
wal_buffers = '1GB'
max_wal_size = '16GB'
max_connections = '5000'
archive_mode = 'on'
archive_command='cp %p /home/ubuntu/archived_wal/%f'
track_wal_io_timing = 'on'
EOF
./pg_ctl -D data -l logfile start
./psql -c "select
pg_create_physical_replication_slot('assb1_repl_slot', true, false)"
postgres
./pg_ctl -D data -l logfile restart
./pg_basebackup -D assbdata
./pg_ctl -D data -l logfile stop
cat << EOF >> assbdata/postgresql.conf
port=5433
primary_conninfo='host=localhost port=5432 dbname=postgres user=ubuntu
application_name=assb1'
primary_slot_name='assb1_repl_slot'
restore_command='cp /home/ubuntu/archived_wal/%f %p'
EOF
touch assbdata/standby.signal
./pg_ctl -D data -l logfile start
./pg_ctl -D assbdata -l logfile1 start
./pgbench -i -s 1 -d postgres
./psql -d postgres -c "ALTER TABLE pgbench_accounts DROP CONSTRAINT
pgbench_accounts_pkey;"
cat << EOF >> insert.sql
\set aid random(1, 10 * :scale)
\set delta random(1, 100000 * :scale)
INSERT INTO pgbench_accounts (aid, bid, abalance) VALUES (:aid, :aid, :delta);
EOF
ulimit -S -n 5000
for c in 1 2 4 8 16 32 64 128 256 512 768 1024 2048 4096; do echo -n
"$c ";./pgbench -n -M prepared -U ubuntu postgres -f insert.sql -c$c
-j$c -T5 2>&1|grep '^tps'|awk '{print $3}';done

---

From: Dilip Kumar <[email-masked]>
Date: Sun, 25 Dec 2022 16:55:26 +0530

> callers of WALRead() read a full WAL page of size XLOG_BLCKSZ bytes or

I agree with this.

> Although they read the whole WAL page, they calculate the valid bytes.

Right

On first read the patch looks good, although it needs some more
thoughts on 'XXX' comments in the patch.

And also I do not like that XLogReadFromBuffers() is using 3 bools
hit/partial hit/miss, instead of this we can use an enum or some
tristate variable, I think that will be cleaner.

-- 
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

---

From: Bharath Rupireddy <[email-masked]>
Date: Mon, 26 Dec 2022 14:20:07 +0530

> thoughts on 'XXX' comments in the patch.

Thanks a lot for reviewing.

Here are some open points that I mentioned in v1 patch:

1.
+     * XXX: Perhaps, measuring the immediate lock availability and its impact
+     * on concurrent WAL writers is a good idea here.

It was shown in my testng upthread [1] that the patch does no harm in
this regard. It will be great if other members try testing in their
respective environments and use cases.

2.
+     * XXX: Perhaps, returning if lock is not immediately available a good idea
+     * here. The caller can then go ahead with reading WAL from WAL file.

After thinking a bit more on this, ISTM that doing the above is right
to not cause any contention when the lock is busy. I've done so in the
v2 patch.

3.
+     * XXX: Perhaps, quickly finding if the given WAL record is in WAL buffers
+     * a good idea here. This avoids unnecessary lock acquire-release cycles.
+     * One way to do that is by maintaining oldest WAL record that's currently
+     * present in WAL buffers.

I think by doing the above we might end up creating a new point of
contention. Because shared variables to track min and max available
LSNs in the WAL buffers will need to be protected against all the
concurrent writers. Also, with the change that's done in (2) above,
that is, quickly exiting if the lock was busy, this comment seems
unnecessary to worry about. Hence, I decided to leave it there.

4.
+         * XXX: Perhaps, we can further go and validate the found page header,
+         * record header and record at least in assert builds, something like
+         * the xlogreader.c does and return if any of those validity checks
+         * fail. Having said that, we stick to the minimal checks for now.

I was being over-cautious initially. The fact that we acquire
WALBufMappingLock while reading the needed WAL buffer page itself
guarantees that no one else initializes it/makes it ready for next use
in AdvanceXLInsertBuffer(). The checks that we have for page header
(xlp_magic, xlp_pageaddr and xlp_tli) in the patch are enough for us
to ensure that we're not reading a page that got just initialized. The
callers will anyway perform extensive checks on page and record in
XLogReaderValidatePageHeader() and ValidXLogRecordHeader()
respectively. If any such failures occur after reading WAL from WAL
buffers, then that must be treated as a bug IMO. Hence, I don't think
we need to do the above.

> tristate variable, I think that will be cleaner.

Yeah, that seems more verbose, all that information can be deduced
from requested bytes and read bytes, I've done so in the v2 patch.

Please review the attached v2 patch further.

I'm also attaching two helper patches (as .txt files) herewith for
testing that basically adds WAL read stats -
USE-ON-HEAD-Collect-WAL-read-from-file-stats.txt - apply on HEAD and
monitor pg_stat_replication for per-walsender WAL read from WAL file
stats. USE-ON-PATCH-Collect-WAL-read-from-buffers-and-file-stats.txt -
apply on v2 patch and monitor pg_stat_replication for per-walsender
WAL read from WAL buffers and WAL file stats.

[1] https://www.postgresql.org/message-id/CALj2ACXUbvON86vgwTkum8ab3bf1%3DHkMxQ5hZJZS3ZcJn8NEXQ%40mail.gmail.com

--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
From 6517e50f482f88ea5185609ff4dcf0e0256475d5 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email-masked]>
Date: Mon, 26 Dec 2022 08:14:11 +0000
Subject: [PATCH v2] Collect WAL read from file stats for WAL senders

---
 doc/src/sgml/monitoring.sgml                | 31 +++++++++++
 src/backend/access/transam/xlogreader.c     | 33 ++++++++++--
 src/backend/access/transam/xlogutils.c      |  2 +-
 src/backend/catalog/system_views.sql        |  5 +-
 src/backend/replication/walsender.c         | 58 +++++++++++++++++++--
 src/bin/pg_waldump/pg_waldump.c             |  2 +-
 src/include/access/xlogreader.h             | 21 ++++++--
 src/include/catalog/pg_proc.dat             |  6 +--
 src/include/replication/walsender_private.h |  4 ++
 src/test/regress/expected/rules.out         |  7 ++-
 10 files changed, 151 insertions(+), 18 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 363b183e5f..fdf4c7d774 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -2615,6 +2615,37 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
        Send time of last reply message received from standby server
       </para></entry>
      </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read</structfield> <type>bigint</type>
+      </para>
+      <para>
+       Number of times WAL data is read from disk
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read_bytes</structfield> <type>numeric</type>
+      </para>
+      <para>
+       Total amount of WAL read from disk in bytes
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read_time</structfield> <type>double precision</type>
+      </para>
+      <para>
+       Total amount of time spent reading WAL from disk via
+       <function>WALRead</function> request, in milliseconds
+       (if <xref linkend="guc-track-wal-io-timing"/> is enabled,
+       otherwise zero).
+      </para></entry>
+     </row>
+
     </tbody>
    </tgroup>
   </table>
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index a38a80e049..7453724a07 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -31,6 +31,7 @@
 #include "access/xlogrecord.h"
 #include "catalog/pg_control.h"
 #include "common/pg_lzcompress.h"
+#include "portability/instr_time.h"
 #include "replication/origin.h"
 
 #ifndef FRONTEND
@@ -1489,9 +1490,9 @@ err:
  * WAL buffers when possible.
  */
 bool
-WALRead(XLogReaderState *state,
-		char *buf, XLogRecPtr startptr, Size count, TimeLineID tli,
-		WALReadError *errinfo)
+WALRead(XLogReaderState *state, char *buf, XLogRecPtr startptr, Size count,
+		TimeLineID tli, WALReadError *errinfo, WALReadStats *stats,
+		bool capture_wal_io_timing)
 {
 	char	   *p;
 	XLogRecPtr	recptr;
@@ -1506,6 +1507,7 @@ WALRead(XLogReaderState *state,
 		uint32		startoff;
 		int			segbytes;
 		int			readbytes;
+		instr_time	start;
 
 		startoff = XLogSegmentOffset(recptr, state->segcxt.ws_segsize);
 
@@ -1540,6 +1542,10 @@ WALRead(XLogReaderState *state,
 		else
 			segbytes = nbytes;
 
+		/* Measure I/O timing to read WAL data if requested by the caller. */
+		if (stats != NULL && capture_wal_io_timing)
+			INSTR_TIME_SET_CURRENT(start);
+
 #ifndef FRONTEND
 		pgstat_report_wait_start(WAIT_EVENT_WAL_READ);
 #endif
@@ -1552,6 +1558,27 @@ WALRead(XLogReaderState *state,
 		pgstat_report_wait_end();
 #endif
 
+		/* Collect I/O stats if requested by the caller. */
+		if (stats != NULL)
+		{
+			/* Increment the number of times WAL is read from disk. */
+			stats->wal_read++;
+
+			/* Collect bytes read. */
+			if (readbytes > 0)
+				stats->wal_read_bytes += readbytes;
+
+			/* Increment the I/O timing. */
+			if (capture_wal_io_timing)
+			{
+				instr_time	duration;
+
+				INSTR_TIME_SET_CURRENT(duration);
+				INSTR_TIME_SUBTRACT(duration, start);
+				stats->wal_read_time += INSTR_TIME_GET_MICROSEC(duration);
+			}
+		}
+
 		if (readbytes <= 0)
 		{
 			errinfo->wre_errno = errno;
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index 563cba258d..372de2c7d8 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -1027,7 +1027,7 @@ read_local_xlog_page_guts(XLogReaderState *state, XLogRecPtr targetPagePtr,
 	 * zero-padded up to the page boundary if it's incomplete.
 	 */
 	if (!WALRead(state, cur_page, targetPagePtr, XLOG_BLCKSZ, tli,
-				 &errinfo))
+				 &errinfo, NULL, false))
 		WALReadRaiseError(&errinfo);
 
 	/* number of valid bytes in the buffer */
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 2d8104b090..b47f44a852 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -892,7 +892,10 @@ CREATE VIEW pg_stat_replication AS
             W.replay_lag,
             W.sync_priority,
             W.sync_state,
-            W.reply_time
+            W.reply_time,
+            W.wal_read,
+            W.wal_read_bytes,
+            W.wal_read_time
     FROM pg_stat_get_activity(NULL) AS S
         JOIN pg_stat_get_wal_senders() AS W ON (S.pid = W.pid)
         LEFT JOIN pg_authid AS U ON (S.usesysid = U.oid);
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c11bb3716f..fa02e327f2 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -259,7 +259,7 @@ static bool TransactionIdInRecentPast(TransactionId xid, uint32 epoch);
 
 static void WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo,
 							  TimeLineID *tli_p);
-
+static void WalSndAccumulateWalReadStats(WALReadStats *stats);
 
 /* Initialize walsender process before entering the main command loop */
 void
@@ -907,6 +907,7 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
 	WALReadError errinfo;
 	XLogSegNo	segno;
 	TimeLineID	currTLI = GetWALInsertionTimeLine();
+	WALReadStats	stats;
 
 	/*
 	 * Since logical decoding is only permitted on a primary server, we know
@@ -932,6 +933,8 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
 	else
 		count = flushptr - targetPagePtr;	/* part of the page available */
 
+	MemSet(&stats, 0, sizeof(WALReadStats));
+
 	/* now actually read the data, we know it's there */
 	if (!WALRead(state,
 				 cur_page,
@@ -940,9 +943,13 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
 				 state->seg.ws_tli, /* Pass the current TLI because only
 									 * WalSndSegmentOpen controls whether new
 									 * TLI is needed. */
-				 &errinfo))
+				 &errinfo,
+				 &stats,
+				 track_wal_io_timing))
 		WALReadRaiseError(&errinfo);
 
+	WalSndAccumulateWalReadStats(&stats);
+
 	/*
 	 * After reading into the buffer, check that what we read was valid. We do
 	 * this after reading, because even though the segment was present when we
@@ -2610,6 +2617,9 @@ InitWalSenderSlot(void)
 			walsnd->sync_standby_priority = 0;
 			walsnd->latch = &MyProc->procLatch;
 			walsnd->replyTime = 0;
+			walsnd->wal_read_stats.wal_read = 0;
+			walsnd->wal_read_stats.wal_read_bytes = 0;
+			walsnd->wal_read_stats.wal_read_time = 0;
 			SpinLockRelease(&walsnd->mutex);
 			/* don't need the lock anymore */
 			MyWalSnd = (WalSnd *) walsnd;
@@ -2730,6 +2740,7 @@ XLogSendPhysical(void)
 	Size		nbytes;
 	XLogSegNo	segno;
 	WALReadError errinfo;
+	WALReadStats stats;
 
 	/* If requested switch the WAL sender to the stopping state. */
 	if (got_STOPPING)
@@ -2945,6 +2956,8 @@ XLogSendPhysical(void)
 	enlargeStringInfo(&output_message, nbytes);
 
 retry:
+	MemSet(&stats, 0, sizeof(WALReadStats));
+
 	if (!WALRead(xlogreader,
 				 &output_message.data[output_message.len],
 				 startptr,
@@ -2952,9 +2965,13 @@ retry:
 				 xlogreader->seg.ws_tli,	/* Pass the current TLI because
 											 * only WalSndSegmentOpen controls
 											 * whether new TLI is needed. */
-				 &errinfo))
+				 &errinfo,
+				 &stats,
+				 track_wal_io_timing))
 		WALReadRaiseError(&errinfo);
 
+	WalSndAccumulateWalReadStats(&stats);
+
 	/* See logical_read_xlog_page(). */
 	XLByteToSeg(startptr, segno, xlogreader->segcxt.ws_segsize);
 	CheckXLogRemoved(segno, xlogreader->seg.ws_tli);
@@ -3458,7 +3475,7 @@ offset_to_interval(TimeOffset offset)
 Datum
 pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_WAL_SENDERS_COLS	12
+#define PG_STAT_GET_WAL_SENDERS_COLS	15
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	SyncRepStandbyData *sync_standbys;
 	int			num_standbys;
@@ -3487,9 +3504,13 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 		WalSndState state;
 		TimestampTz replyTime;
 		bool		is_sync_standby;
+		int64		wal_read;
+		uint64		wal_read_bytes;
+		int64		wal_read_time;
 		Datum		values[PG_STAT_GET_WAL_SENDERS_COLS];
 		bool		nulls[PG_STAT_GET_WAL_SENDERS_COLS] = {0};
 		int			j;
+		char		buf[256];
 
 		/* Collect data from shared memory */
 		SpinLockAcquire(&walsnd->mutex);
@@ -3509,6 +3530,9 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 		applyLag = walsnd->applyLag;
 		priority = walsnd->sync_standby_priority;
 		replyTime = walsnd->replyTime;
+		wal_read = walsnd->wal_read_stats.wal_read;
+		wal_read_bytes = walsnd->wal_read_stats.wal_read_bytes;
+		wal_read_time = walsnd->wal_read_stats.wal_read_time;
 		SpinLockRelease(&walsnd->mutex);
 
 		/*
@@ -3605,6 +3629,18 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 				nulls[11] = true;
 			else
 				values[11] = TimestampTzGetDatum(replyTime);
+
+			values[12] = Int64GetDatum(wal_read);
+
+			/* Convert to numeric. */
+			snprintf(buf, sizeof buf, UINT64_FORMAT, wal_read_bytes);
+			values[13] = DirectFunctionCall3(numeric_in,
+											 CStringGetDatum(buf),
+											 ObjectIdGetDatum(0),
+											 Int32GetDatum(-1));
+
+			/* Convert counter from microsec to millisec for display. */
+			values[14] = Float8GetDatum(((double) wal_read_time) / 1000.0);
 		}
 
 		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
@@ -3849,3 +3885,17 @@ LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now)
 	Assert(time != 0);
 	return now - time;
 }
+
+/*
+ * Function to accumulate WAL Read stats for WAL sender.
+ */
+static void
+WalSndAccumulateWalReadStats(WALReadStats *stats)
+{
+	/* Collect I/O stats for walsender. */
+	SpinLockAcquire(&MyWalSnd->mutex);
+	MyWalSnd->wal_read_stats.wal_read += stats->wal_read;
+	MyWalSnd->wal_read_stats.wal_read_bytes += stats->wal_read_bytes;
+	MyWalSnd->wal_read_stats.wal_read_time += stats->wal_read_time;
+	SpinLockRelease(&MyWalSnd->mutex);
+}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..698ce1e9f7 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -364,7 +364,7 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
 	}
 
 	if (!WALRead(state, readBuff, targetPagePtr, count, private->timeline,
-				 &errinfo))
+				 &errinfo, NULL, false))
 	{
 		WALOpenSegment *seg = &errinfo.wre_seg;
 		char		fname[MAXPGPATH];
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index e87f91316a..26a2c975de 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -389,9 +389,24 @@ typedef struct WALReadError
 	WALOpenSegment wre_seg;		/* Segment we tried to read from. */
 } WALReadError;
 
-extern bool WALRead(XLogReaderState *state,
-					char *buf, XLogRecPtr startptr, Size count,
-					TimeLineID tli, WALReadError *errinfo);
+/*
+ * WAL read stats from WALRead that the callers can use.
+ */
+typedef struct WALReadStats
+{
+	/* Number of times WAL read from disk. */
+	int64	wal_read;
+
+	/* Total amount of WAL read from disk in bytes. */
+	uint64	wal_read_bytes;
+
+	/* Total amount of time spent reading WAL from disk. */
+	int64 	wal_read_time;
+} WALReadStats;
+
+extern bool WALRead(XLogReaderState *state, char *buf, XLogRecPtr startptr,
+					Size count, TimeLineID tli, WALReadError *errinfo,
+					WALReadStats *stats, bool capture_wal_io_timing);
 
 /* Functions for decoding an XLogRecord */
 
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 7056c95371..18320cf846 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5391,9 +5391,9 @@
   proname => 'pg_stat_get_wal_senders', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => '',
-  proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz}',
-  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time}',
+  proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz,int8,numeric,float8}',
+  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time,wal_read,wal_read_bytes,wal_read_time}',
   prosrc => 'pg_stat_get_wal_senders' },
 { oid => '3317', descr => 'statistics: information about WAL receiver',
   proname => 'pg_stat_get_wal_receiver', proisstrict => 'f', provolatile => 's',
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index 7897c74589..35413ea0d2 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -13,6 +13,7 @@
 #define _WALSENDER_PRIVATE_H
 
 #include "access/xlog.h"
+#include "access/xlogreader.h"
 #include "nodes/nodes.h"
 #include "replication/syncrep.h"
 #include "storage/latch.h"
@@ -78,6 +79,9 @@ typedef struct WalSnd
 	 * Timestamp of the last message received from standby.
 	 */
 	TimestampTz replyTime;
+
+	/* WAL read stats for walsender. */
+	WALReadStats wal_read_stats;
 } WalSnd;
 
 extern PGDLLIMPORT WalSnd *MyWalSnd;
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index fb9f936d43..fd9d298e79 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2054,9 +2054,12 @@ pg_stat_replication| SELECT s.pid,
     w.replay_lag,
     w.sync_priority,
     w.sync_state,
-    w.reply_time
+    w.reply_time,
+    w.wal_read,
+    w.wal_read_bytes,
+    w.wal_read_time
    FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid, query_id)
-     JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid)))
+     JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time, wal_read, wal_read_bytes, wal_read_time) ON ((s.pid = w.pid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_replication_slots| SELECT s.slot_name,
     s.spill_txns,
-- 
2.34.1

From f90dfcbd1968280feec6d116568697225854ac40 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email-masked]>
Date: Mon, 26 Dec 2022 08:13:07 +0000
Subject: [PATCH v2] Collect WAL read from buffers and file stats for WAL
 senders

---
 doc/src/sgml/monitoring.sgml                | 61 +++++++++++++++
 src/backend/access/transam/xlogreader.c     | 56 +++++++++++++-
 src/backend/access/transam/xlogutils.c      |  2 +-
 src/backend/catalog/system_views.sql        |  8 +-
 src/backend/replication/walsender.c         | 85 ++++++++++++++++++++-
 src/bin/pg_waldump/pg_waldump.c             |  2 +-
 src/include/access/xlogreader.h             | 30 +++++++-
 src/include/catalog/pg_proc.dat             |  6 +-
 src/include/replication/walsender_private.h |  4 +
 src/test/regress/expected/rules.out         | 10 ++-
 10 files changed, 246 insertions(+), 18 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 363b183e5f..239e0b0db9 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -2615,6 +2615,67 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
        Send time of last reply message received from standby server
       </para></entry>
      </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read</structfield> <type>bigint</type>
+      </para>
+      <para>
+       Number of times WAL data is read from disk
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read_bytes</structfield> <type>numeric</type>
+      </para>
+      <para>
+       Total amount of WAL read from disk in bytes
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read_time</structfield> <type>double precision</type>
+      </para>
+      <para>
+       Total amount of time spent reading WAL from disk via
+       <function>WALRead</function> request, in milliseconds
+       (if <xref linkend="guc-track-wal-io-timing"/> is enabled,
+       otherwise zero).
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read_buffers</structfield> <type>bigint</type>
+      </para>
+      <para>
+       Number of times WAL data is read from WAL buffers
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read_bytes_buffers</structfield> <type>numeric</type>
+      </para>
+      <para>
+       Total amount of WAL read from WAL buffers in bytes
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>wal_read_time_buffers</structfield> <type>double precision</type>
+      </para>
+      <para>
+       Total amount of time spent reading WAL from WAL buffers via
+       <function>WALRead</function> request, in milliseconds
+       (if <xref linkend="guc-track-wal-io-timing"/> is enabled,
+       otherwise zero).
+      </para></entry>
+     </row>
+
     </tbody>
    </tgroup>
   </table>
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 4a2e7af169..b9dfd4fde7 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -31,6 +31,7 @@
 #include "access/xlogrecord.h"
 #include "catalog/pg_control.h"
 #include "common/pg_lzcompress.h"
+#include "portability/instr_time.h"
 #include "replication/origin.h"
 
 #ifndef FRONTEND
@@ -1488,9 +1489,9 @@ err:
  * When possible, this function reads data directly from WAL buffers.
  */
 bool
-WALRead(XLogReaderState *state,
-		char *buf, XLogRecPtr startptr, Size count, TimeLineID tli,
-		WALReadError *errinfo)
+WALRead(XLogReaderState *state, char *buf, XLogRecPtr startptr, Size count,
+		TimeLineID tli, WALReadError *errinfo, WALReadStats *stats,
+		bool capture_wal_io_timing)
 {
 	char	   *p;
 	XLogRecPtr	recptr;
@@ -1510,10 +1511,33 @@ WALRead(XLogReaderState *state,
 	if (!RecoveryInProgress() &&
 		tli == GetWALInsertionTimeLine())
 	{
+		instr_time      start;
+
+		/* Measure I/O timing to read WAL data if requested by the caller. */
+		if (stats != NULL && capture_wal_io_timing)
+			INSTR_TIME_SET_CURRENT(start);
+
 		pgstat_report_wait_start(WAIT_EVENT_WAL_READ);
 		XLogReadFromBuffers(startptr, tli, count, buf, &read_bytes);
 		pgstat_report_wait_end();
 
+		/* Collect I/O stats if requested by the caller. */
+		if (stats != NULL && read_bytes > 0)
+		{
+			stats->wal_read_buffers++;
+			stats->wal_read_bytes_buffers += read_bytes;
+
+			/* Increment the I/O timing. */
+			if (capture_wal_io_timing)
+			{
+				instr_time      duration;
+
+				INSTR_TIME_SET_CURRENT(duration);
+				INSTR_TIME_SUBTRACT(duration, start);
+				stats->wal_read_time_buffers += INSTR_TIME_GET_MICROSEC(duration);
+			}
+		}
+
 		/*
 		 * Check if we have read fully (hit), partially (partial hit) or
 		 * nothing (miss) from WAL buffers. If we have read either partially or
@@ -1549,6 +1573,7 @@ WALRead(XLogReaderState *state,
 		uint32		startoff;
 		int			segbytes;
 		int			readbytes;
+		instr_time	start;
 
 		startoff = XLogSegmentOffset(recptr, state->segcxt.ws_segsize);
 
@@ -1583,6 +1608,10 @@ WALRead(XLogReaderState *state,
 		else
 			segbytes = nbytes;
 
+		/* Measure I/O timing to read WAL data if requested by the caller. */
+		if (stats != NULL && capture_wal_io_timing)
+			INSTR_TIME_SET_CURRENT(start);
+
 #ifndef FRONTEND
 		pgstat_report_wait_start(WAIT_EVENT_WAL_READ);
 #endif
@@ -1595,6 +1624,27 @@ WALRead(XLogReaderState *state,
 		pgstat_report_wait_end();
 #endif
 
+		/* Collect I/O stats if requested by the caller. */
+		if (stats != NULL)
+		{
+			/* Increment the number of times WAL is read from disk. */
+			stats->wal_read++;
+
+			/* Collect bytes read. */
+			if (readbytes > 0)
+				stats->wal_read_bytes += readbytes;
+
+			/* Increment the I/O timing. */
+			if (capture_wal_io_timing)
+			{
+				instr_time	duration;
+
+				INSTR_TIME_SET_CURRENT(duration);
+				INSTR_TIME_SUBTRACT(duration, start);
+				stats->wal_read_time += INSTR_TIME_GET_MICROSEC(duration);
+			}
+		}
+
 		if (readbytes <= 0)
 		{
 			errinfo->wre_errno = errno;
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index 563cba258d..372de2c7d8 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -1027,7 +1027,7 @@ read_local_xlog_page_guts(XLogReaderState *state, XLogRecPtr targetPagePtr,
 	 * zero-padded up to the page boundary if it's incomplete.
 	 */
 	if (!WALRead(state, cur_page, targetPagePtr, XLOG_BLCKSZ, tli,
-				 &errinfo))
+				 &errinfo, NULL, false))
 		WALReadRaiseError(&errinfo);
 
 	/* number of valid bytes in the buffer */
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 2d8104b090..bf6315df27 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -892,7 +892,13 @@ CREATE VIEW pg_stat_replication AS
             W.replay_lag,
             W.sync_priority,
             W.sync_state,
-            W.reply_time
+            W.reply_time,
+            W.wal_read,
+            W.wal_read_bytes,
+            W.wal_read_time,
+            W.wal_read_buffers,
+            W.wal_read_bytes_buffers,
+            W.wal_read_time_buffers
     FROM pg_stat_get_activity(NULL) AS S
         JOIN pg_stat_get_wal_senders() AS W ON (S.pid = W.pid)
         LEFT JOIN pg_authid AS U ON (S.usesysid = U.oid);
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c11bb3716f..d3393b2b63 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -259,7 +259,7 @@ static bool TransactionIdInRecentPast(TransactionId xid, uint32 epoch);
 
 static void WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo,
 							  TimeLineID *tli_p);
-
+static void WalSndAccumulateWalReadStats(WALReadStats *stats);
 
 /* Initialize walsender process before entering the main command loop */
 void
@@ -907,6 +907,7 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
 	WALReadError errinfo;
 	XLogSegNo	segno;
 	TimeLineID	currTLI = GetWALInsertionTimeLine();
+	WALReadStats	stats;
 
 	/*
 	 * Since logical decoding is only permitted on a primary server, we know
@@ -932,6 +933,8 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
 	else
 		count = flushptr - targetPagePtr;	/* part of the page available */
 
+	MemSet(&stats, 0, sizeof(WALReadStats));
+
 	/* now actually read the data, we know it's there */
 	if (!WALRead(state,
 				 cur_page,
@@ -940,9 +943,13 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
 				 state->seg.ws_tli, /* Pass the current TLI because only
 									 * WalSndSegmentOpen controls whether new
 									 * TLI is needed. */
-				 &errinfo))
+				 &errinfo,
+				 &stats,
+				 track_wal_io_timing))
 		WALReadRaiseError(&errinfo);
 
+	WalSndAccumulateWalReadStats(&stats);
+
 	/*
 	 * After reading into the buffer, check that what we read was valid. We do
 	 * this after reading, because even though the segment was present when we
@@ -2610,6 +2617,12 @@ InitWalSenderSlot(void)
 			walsnd->sync_standby_priority = 0;
 			walsnd->latch = &MyProc->procLatch;
 			walsnd->replyTime = 0;
+			walsnd->wal_read_stats.wal_read = 0;
+			walsnd->wal_read_stats.wal_read_bytes = 0;
+			walsnd->wal_read_stats.wal_read_time = 0;
+			walsnd->wal_read_stats.wal_read_buffers = 0;
+			walsnd->wal_read_stats.wal_read_bytes_buffers = 0;
+			walsnd->wal_read_stats.wal_read_time_buffers = 0;
 			SpinLockRelease(&walsnd->mutex);
 			/* don't need the lock anymore */
 			MyWalSnd = (WalSnd *) walsnd;
@@ -2730,6 +2743,7 @@ XLogSendPhysical(void)
 	Size		nbytes;
 	XLogSegNo	segno;
 	WALReadError errinfo;
+	WALReadStats stats;
 
 	/* If requested switch the WAL sender to the stopping state. */
 	if (got_STOPPING)
@@ -2945,6 +2959,8 @@ XLogSendPhysical(void)
 	enlargeStringInfo(&output_message, nbytes);
 
 retry:
+	MemSet(&stats, 0, sizeof(WALReadStats));
+
 	if (!WALRead(xlogreader,
 				 &output_message.data[output_message.len],
 				 startptr,
@@ -2952,9 +2968,13 @@ retry:
 				 xlogreader->seg.ws_tli,	/* Pass the current TLI because
 											 * only WalSndSegmentOpen controls
 											 * whether new TLI is needed. */
-				 &errinfo))
+				 &errinfo,
+				 &stats,
+				 track_wal_io_timing))
 		WALReadRaiseError(&errinfo);
 
+	WalSndAccumulateWalReadStats(&stats);
+
 	/* See logical_read_xlog_page(). */
 	XLByteToSeg(startptr, segno, xlogreader->segcxt.ws_segsize);
 	CheckXLogRemoved(segno, xlogreader->seg.ws_tli);
@@ -3458,7 +3478,7 @@ offset_to_interval(TimeOffset offset)
 Datum
 pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_WAL_SENDERS_COLS	12
+#define PG_STAT_GET_WAL_SENDERS_COLS	18
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	SyncRepStandbyData *sync_standbys;
 	int			num_standbys;
@@ -3487,9 +3507,16 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 		WalSndState state;
 		TimestampTz replyTime;
 		bool		is_sync_standby;
+		int64		wal_read;
+		uint64		wal_read_bytes;
+		int64		wal_read_time;
+		int64		wal_read_buffers;
+		uint64		wal_read_bytes_buffers;
+		int64		wal_read_time_buffers;
 		Datum		values[PG_STAT_GET_WAL_SENDERS_COLS];
 		bool		nulls[PG_STAT_GET_WAL_SENDERS_COLS] = {0};
 		int			j;
+		char		buf[256];
 
 		/* Collect data from shared memory */
 		SpinLockAcquire(&walsnd->mutex);
@@ -3509,6 +3536,12 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 		applyLag = walsnd->applyLag;
 		priority = walsnd->sync_standby_priority;
 		replyTime = walsnd->replyTime;
+		wal_read = walsnd->wal_read_stats.wal_read;
+		wal_read_bytes = walsnd->wal_read_stats.wal_read_bytes;
+		wal_read_time = walsnd->wal_read_stats.wal_read_time;
+		wal_read_buffers = walsnd->wal_read_stats.wal_read_buffers;
+		wal_read_bytes_buffers = walsnd->wal_read_stats.wal_read_bytes_buffers;
+		wal_read_time_buffers = walsnd->wal_read_stats.wal_read_time_buffers;
 		SpinLockRelease(&walsnd->mutex);
 
 		/*
@@ -3605,6 +3638,31 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 				nulls[11] = true;
 			else
 				values[11] = TimestampTzGetDatum(replyTime);
+
+			values[12] = Int64GetDatum(wal_read);
+
+			/* Convert to numeric. */
+			snprintf(buf, sizeof buf, UINT64_FORMAT, wal_read_bytes);
+			values[13] = DirectFunctionCall3(numeric_in,
+											 CStringGetDatum(buf),
+											 ObjectIdGetDatum(0),
+											 Int32GetDatum(-1));
+
+			/* Convert counter from microsec to millisec for display. */
+			values[14] = Float8GetDatum(((double) wal_read_time) / 1000.0);
+
+			values[15] = Int64GetDatum(wal_read_buffers);
+
+			/* Convert to numeric. */
+			MemSet(buf, '\0', sizeof buf);
+			snprintf(buf, sizeof buf, UINT64_FORMAT, wal_read_bytes_buffers);
+			values[16] = DirectFunctionCall3(numeric_in,
+											 CStringGetDatum(buf),
+											 ObjectIdGetDatum(0),
+											 Int32GetDatum(-1));
+
+			/* Convert counter from microsec to millisec for display. */
+			values[17] = Float8GetDatum(((double) wal_read_time_buffers) / 1000.0);
 		}
 
 		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
@@ -3849,3 +3907,22 @@ LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now)
 	Assert(time != 0);
 	return now - time;
 }
+
+/*
+ * Function to accumulate WAL Read stats for WAL sender.
+ */
+static void
+WalSndAccumulateWalReadStats(WALReadStats *stats)
+{
+	/* Collect I/O stats for walsender. */
+	SpinLockAcquire(&MyWalSnd->mutex);
+	MyWalSnd->wal_read_stats.wal_read += stats->wal_read;
+	MyWalSnd->wal_read_stats.wal_read_bytes += stats->wal_read_bytes;
+	MyWalSnd->wal_read_stats.wal_read_time += stats->wal_read_time;
+	MyWalSnd->wal_read_stats.wal_read_buffers += stats->wal_read_buffers;
+	MyWalSnd->wal_read_stats.wal_read_bytes_buffers +=
+									stats->wal_read_bytes_buffers;
+	MyWalSnd->wal_read_stats.wal_read_time_buffers +=
+									stats->wal_read_time_buffers;
+	SpinLockRelease(&MyWalSnd->mutex);
+}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..698ce1e9f7 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -364,7 +364,7 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
 	}
 
 	if (!WALRead(state, readBuff, targetPagePtr, count, private->timeline,
-				 &errinfo))
+				 &errinfo, NULL, false))
 	{
 		WALOpenSegment *seg = &errinfo.wre_seg;
 		char		fname[MAXPGPATH];
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index e87f91316a..9287114779 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -389,9 +389,33 @@ typedef struct WALReadError
 	WALOpenSegment wre_seg;		/* Segment we tried to read from. */
 } WALReadError;
 
-extern bool WALRead(XLogReaderState *state,
-					char *buf, XLogRecPtr startptr, Size count,
-					TimeLineID tli, WALReadError *errinfo);
+/*
+ * WAL read stats from WALRead that the callers can use.
+ */
+typedef struct WALReadStats
+{
+	/* Number of times WAL read from disk. */
+	int64	wal_read;
+
+	/* Total amount of WAL read from disk in bytes. */
+	uint64	wal_read_bytes;
+
+	/* Total amount of time spent reading WAL from disk. */
+	int64 	wal_read_time;
+
+	/* Number of times WAL read from WAL buffers. */
+	int64   wal_read_buffers;
+
+	/* Total amount of WAL read from WAL buffers in bytes. */
+	uint64  wal_read_bytes_buffers;
+
+	/* Total amount of time spent reading WAL from WAL buffers. */
+	int64   wal_read_time_buffers;
+} WALReadStats;
+
+extern bool WALRead(XLogReaderState *state, char *buf, XLogRecPtr startptr,
+					Size count, TimeLineID tli, WALReadError *errinfo,
+					WALReadStats *stats, bool capture_wal_io_timing);
 
 /* Functions for decoding an XLogRecord */
 
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 7056c95371..706a005c2b 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5391,9 +5391,9 @@
   proname => 'pg_stat_get_wal_senders', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => '',
-  proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz}',
-  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time}',
+  proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz,int8,numeric,float8,int8,numeric,float8}',
+  proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time,wal_read,wal_read_bytes,wal_read_time,wal_read_buffers,wal_read_bytes_buffers,wal_read_time_buffers}',
   prosrc => 'pg_stat_get_wal_senders' },
 { oid => '3317', descr => 'statistics: information about WAL receiver',
   proname => 'pg_stat_get_wal_receiver', proisstrict => 'f', provolatile => 's',
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index 7897c74589..35413ea0d2 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -13,6 +13,7 @@
 #define _WALSENDER_PRIVATE_H
 
 #include "access/xlog.h"
+#include "access/xlogreader.h"
 #include "nodes/nodes.h"
 #include "replication/syncrep.h"
 #include "storage/latch.h"
@@ -78,6 +79,9 @@ typedef struct WalSnd
 	 * Timestamp of the last message received from standby.
 	 */
 	TimestampTz replyTime;
+
+	/* WAL read stats for walsender. */
+	WALReadStats wal_read_stats;
 } WalSnd;
 
 extern PGDLLIMPORT WalSnd *MyWalSnd;
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index fb9f936d43..6ae65981c2 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2054,9 +2054,15 @@ pg_stat_replication| SELECT s.pid,
     w.replay_lag,
     w.sync_priority,
     w.sync_state,
-    w.reply_time
+    w.reply_time,
+    w.wal_read,
+    w.wal_read_bytes,
+    w.wal_read_time,
+    w.wal_read_buffers,
+    w.wal_read_bytes_buffers,
+    w.wal_read_time_buffers
    FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid, query_id)
-     JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid)))
+     JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time, wal_read, wal_read_bytes, wal_read_time, wal_read_buffers, wal_read_bytes_buffers, wal_read_time_buffers) ON ((s.pid = w.pid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_replication_slots| SELECT s.slot_name,
     s.spill_txns,
-- 
2.34.1