From 5a52eb6768dc34db0eceea1e21523b4acbcc01bd Mon Sep 17 00:00:00 2001
From: Daniel Gustafsson <dgustafsson@postgresql.org>
Date: Tue, 2 Jul 2024 15:20:43 +0200
Subject: [PATCH v2] Support checksum enable/disable in a running cluster

This allows data checksums to be enabled, or disabled, in a running
cluster without restricting access to the cluster during processing.

Data checksums could prior to this only be enabled at initdb time, or
when the cluster is offline using pg_checksums. This commit introduce
functionality to enable, and disable, data checksums without the need
for turning off the cluster.

A dynamic background worker is responsible for launching a per-database
worker which will mark all buffers dirty for all relation with storage
in order for them to have data checksums on write. Once all relations
in all databases have been processed, the data_checksums state can be
set to "on" and the cluster will at that point be identical to one
which had checksums enabled from the start.

While the cluster is writing checksums on existing buffers, checksums
are written but not verified during reading to avoid false negatives.
Disabling checksums will not touch any buffers (but existing checksums
cannot be re-used in case checksums are immediately re-enabled). While
disabling, checksums are again written but not verified to ensure that
concurrent backends which haven't started disabling checksums will
incur a verification error.

New in-progress states are introduced for data_checksums which during
processing ensures that backends know whether to verify and write
checksums. All state changes across backends are synchronized using
procsignalbarriers

Earlier versions of this patch were reviewed by Heikki Linnakangas,
Robert Haas, Andres Freund, Tomas Vondra, Michael Banck and Andrey
Borodin.

Authors: Daniel Gustafsson, Magnus Hagander
Discussion: https://postgr.es/m/CABUevExz9hUUOLnJVr2kpw9Cx=o4MCr1SVKwbupzuxP7ckNutA@mail.gmail.com
Discussion: https://postgr.es/m/20181030051643.elbxjww5jjgnjaxg@alap3.anarazel.de
Discussion: https://postgr.es/m/CABUevEwE3urLtwxxqdgd5O2oQz9J717ZzMbh+ziCSa5YLLU_BA@mail.gmail.com
---
 doc/src/sgml/func.sgml                        |   71 +
 doc/src/sgml/monitoring.sgml                  |  206 ++-
 doc/src/sgml/ref/pg_checksums.sgml            |    6 +
 doc/src/sgml/wal.sgml                         |   57 +-
 src/backend/access/heap/heapam.c              |    1 +
 src/backend/access/rmgrdesc/xlogdesc.c        |   24 +
 src/backend/access/transam/xlog.c             |  455 +++++-
 src/backend/access/transam/xlogfuncs.c        |   41 +
 src/backend/backup/basebackup.c               |    6 +-
 src/backend/catalog/system_functions.sql      |    7 +
 src/backend/catalog/system_views.sql          |   21 +
 src/backend/postmaster/Makefile               |    1 +
 src/backend/postmaster/bgworker.c             |    7 +
 src/backend/postmaster/datachecksumsworker.c  | 1404 +++++++++++++++++
 src/backend/postmaster/meson.build            |    1 +
 src/backend/replication/logical/decode.c      |    1 +
 src/backend/storage/buffer/bufmgr.c           |    4 +-
 src/backend/storage/ipc/ipci.c                |    3 +
 src/backend/storage/ipc/procsignal.c          |   13 +
 src/backend/storage/page/README               |    4 +-
 src/backend/storage/page/bufpage.c            |    6 +-
 src/backend/storage/smgr/bulk_write.c         |    2 +
 src/backend/utils/activity/pgstat.c           |    1 -
 src/backend/utils/activity/pgstat_io.c        |    2 +
 .../utils/activity/wait_event_names.txt       |    3 +
 src/backend/utils/adt/pgstatfuncs.c           |    8 +-
 src/backend/utils/init/miscinit.c             |    9 +-
 src/backend/utils/init/postinit.c             |    7 +-
 src/backend/utils/misc/guc_tables.c           |   32 +-
 src/bin/pg_checksums/pg_checksums.c           |    2 +-
 src/bin/pg_upgrade/controldata.c              |    9 +
 src/include/access/xlog.h                     |   16 +-
 src/include/access/xlog_internal.h            |    7 +
 src/include/catalog/pg_control.h              |    1 +
 src/include/catalog/pg_proc.dat               |   17 +
 src/include/commands/progress.h               |   18 +
 src/include/miscadmin.h                       |    4 +
 src/include/postmaster/datachecksumsworker.h  |   31 +
 src/include/storage/bufpage.h                 |   10 +
 src/include/storage/checksum.h                |    8 +
 src/include/storage/lwlocklist.h              |    1 +
 src/include/storage/procsignal.h              |    5 +
 src/include/utils/backend_progress.h          |    1 +
 src/test/Makefile                             |   10 +-
 src/test/checksum/.gitignore                  |    2 +
 src/test/checksum/Makefile                    |   23 +
 src/test/checksum/README                      |   22 +
 src/test/checksum/meson.build                 |   15 +
 src/test/checksum/t/001_basic.pl              |   85 +
 src/test/checksum/t/002_restarts.pl           |   90 ++
 src/test/checksum/t/003_standby_restarts.pl   |  130 ++
 src/test/checksum/t/004_offline.pl            |  100 ++
 src/test/meson.build                          |    1 +
 src/test/perl/PostgreSQL/Test/Cluster.pm      |   34 +
 src/tools/pgindent/typedefs.list              |    5 +
 55 files changed, 3001 insertions(+), 49 deletions(-)
 create mode 100644 src/backend/postmaster/datachecksumsworker.c
 create mode 100644 src/include/postmaster/datachecksumsworker.h
 create mode 100644 src/test/checksum/.gitignore
 create mode 100644 src/test/checksum/Makefile
 create mode 100644 src/test/checksum/README
 create mode 100644 src/test/checksum/meson.build
 create mode 100644 src/test/checksum/t/001_basic.pl
 create mode 100644 src/test/checksum/t/002_restarts.pl
 create mode 100644 src/test/checksum/t/003_standby_restarts.pl
 create mode 100644 src/test/checksum/t/004_offline.pl

diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index d6acdd3059..dff3f1ab08 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -29715,6 +29715,77 @@ DETAIL:  Make sure pg_wal_replay_wait() isn't called within a transaction with a
 
   </sect2>
 
+  <sect2 id="functions-admin-checksum">
+   <title>Data Checksum Functions</title>
+
+   <para>
+    The functions shown in <xref linkend="functions-checksums-table" /> can
+    be used to enable or disable data checksums in a running cluster.
+    See <xref linkend="checksums" /> for details.
+   </para>
+
+   <table id="functions-checksums-table">
+    <title>Data Checksum Functions</title>
+    <tgroup cols="1">
+     <thead>
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        Function
+       </para>
+       <para>
+        Description
+       </para></entry>
+      </row>
+     </thead>
+
+     <tbody>
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>pg_enable_data_checksums</primary>
+        </indexterm>
+        <function>pg_enable_data_checksums</function> ( <optional><parameter>cost_delay</parameter> <type>int</type>, <parameter>cost_limit</parameter> <type>int</type></optional> )
+        <returnvalue>void</returnvalue>
+       </para>
+       <para>
+        Initiates data checksums for the cluster. This will switch the data
+        checksums mode to <literal>inprogress-on</literal> as well as start a
+        background worker that will process all pages in the database and
+        enable checksums on them. When all data pages have had checksums
+        enabled, the cluster will automatically switch data checksums mode to
+        <literal>on</literal>.
+       </para>
+       <para>
+        If <parameter>cost_delay</parameter> and <parameter>cost_limit</parameter> are
+        specified, the speed of the process is throttled using the same principles as
+        <link linkend="runtime-config-resource-vacuum-cost">Cost-based Vacuum Delay</link>.
+       </para></entry>
+      </row>
+
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>pg_disable_data_checksums</primary>
+        </indexterm>
+        <function>pg_disable_data_checksums</function> ()
+        <returnvalue>void</returnvalue>
+       </para>
+       <para>
+        Disables data checksum validation and calculation for the cluster. This
+        will switch the data checksum mode to <literal>inprogress-off</literal>
+        while data checksums are being disabled. When all active backends have
+        stopped validating data checksums, the data checksum mode will be
+        changed to <literal>off</literal>.  At this point the data pages will
+        still have checksums recorded but they are not updated when pages are
+        modified.
+       </para></entry>
+      </row>
+     </tbody>
+    </tgroup>
+   </table>
+
+  </sect2>
+
   <sect2 id="functions-admin-dbobject">
    <title>Database Object Management Functions</title>
 
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 48ffe87241..5c61e8bbc1 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -3487,8 +3487,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para>
       <para>
        Number of data page checksum failures detected in this
-       database (or on a shared object), or NULL if data checksums are not
-       enabled.
+       database. Detected failures are reported regardless of the
+       <xref linkend="guc-data-checksums"/> setting.
       </para></entry>
      </row>
 
@@ -3498,8 +3498,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para>
       <para>
        Time at which the last data page checksum failure was detected in
-       this database (or on a shared object), or NULL if data checksums are not
-       enabled.
+       this database (or on a shared object). Last failure is reported
+       regardless of the <xref linkend="guc-data-checksums"/> setting.
       </para></entry>
      </row>
 
@@ -6661,6 +6661,204 @@ FROM pg_stat_get_backend_idset() AS backendid;
 
  </sect2>
 
+ <sect2 id="data-checksum-progress-reporting">
+  <title>Data Checksum Progress Reporting</title>
+
+  <indexterm>
+   <primary>pg_stat_progress_datachecksums</primary>
+  </indexterm>
+
+  <para>
+   When data checksums are being enabled on a running cluster, the
+   <structname>pg_stat_progress_datachecksums</structname> view will contain
+   a row for each background worker which is currently calculating checksums
+   for the data pages.
+  </para>
+
+  <table id="pg-stat-progress-datachecksums-view" xreflabel="pg_stat_progress_datachecksums">
+   <title><structname>pg_stat_progress_datachecksums</structname> View</title>
+   <tgroup cols="1">
+    <thead>
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        Column Type
+       </para>
+       <para>
+        Description>
+       </para>
+      </entry>
+     </row>
+    </thead>
+
+    <tbody>
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>pid</structfield> <type>integer</type>
+       </para>
+       <para>
+        Process ID of a datachecksumworker process.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>phase</structfield> <type>text</type>
+       </para>
+       <para>
+        Current processing phase, see <xref linkend="datachecksum-phases"/>
+        for description of the phases.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>databases_total</structfield> <type>integer</type>
+       </para>
+       <para>
+        The total number of database which will be processed.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>relations_total</structfield> <type>integer</type>
+       </para>
+       <para>
+        The total number of relations which will be processed, or
+        <literal>NULL</literal> if the datachecksumsworker hasn't calcuated
+        the number of relations yet.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>databases_processed</structfield> <type>integer</type>
+       </para>
+       <para>
+        The number of databases which have been processed.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>relations_processed</structfield> <type>integer</type>
+       </para>
+       <para>
+        The number of relations which have been processed.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>databases_current</structfield> <type>oid</type>
+       </para>
+       <para>
+        OID of the database currently being processed.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>relation_current</structfield> <type>oid</type>
+       </para>
+       <para>
+        OID of the relation currently being processed.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>relation_current_blocks</structfield> <type>integer</type>
+       </para>
+       <para>
+        The total number of blocks in the relation currently being processed.
+       </para>
+      </entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry">
+       <para role="column_definition">
+        <structfield>relation_current_blocks_processed</structfield> <type>integer</type>
+       </para>
+       <para>
+        The number of blocks which have been processed in the relation currently
+        being processed.
+       </para>
+      </entry>
+     </row>
+
+    </tbody>
+   </tgroup>
+  </table>
+
+  <table id="datachecksum-phases">
+   <title>Data Checksum Phases</title>
+   <tgroup cols="2">
+    <colspec colname="col1" colwidth="1*"/>
+    <colspec colname="col2" colwidth="2*"/>
+    <thead>
+     <row>
+      <entry>Phase</entry>
+      <entry>Description</entry>
+     </row>
+    </thead>
+    <tbody>
+     <row>
+      <entry><literal>enabling</literal></entry>
+      <entry>
+       The command is currently enabling data checksums on the cluster.
+      </entry>
+     </row>
+     <row>
+      <entry><literal>disabling</literal></entry>
+      <entry>
+       The command is currently disabling data checksums on the cluster.
+      </entry>
+     </row>
+     <row>
+      <entry><literal>waiting on backends</literal></entry>
+      <entry>
+       The command is currently waiting for backends to acknowledge the data
+       checksum operation.
+      </entry>
+     </row>
+     <row>
+      <entry><literal>waiting on temporary tables</literal></entry>
+      <entry>
+       The command is currently waiting for all temporary tables which existed
+       at the time the command was started to be removed.
+      </entry>
+     </row>
+     <row>
+      <entry><literal>done</literal></entry>
+      <entry>
+       The command has finished processing and is exiting.
+      </entry>
+     </row>
+    </tbody>
+   </tgroup>
+  </table>
+ </sect2>
+
  </sect1>
 
  <sect1 id="dynamic-trace">
diff --git a/doc/src/sgml/ref/pg_checksums.sgml b/doc/src/sgml/ref/pg_checksums.sgml
index 95043aa329..0343710af5 100644
--- a/doc/src/sgml/ref/pg_checksums.sgml
+++ b/doc/src/sgml/ref/pg_checksums.sgml
@@ -45,6 +45,12 @@ PostgreSQL documentation
    exit status is nonzero if the operation failed.
   </para>
 
+  <para>
+   When enabling checksums, if checksums were in the process of being enabled
+   when the cluster was shut down, <application>pg_checksums</application>
+   will still process all relations regardless of the online processing.
+  </para>
+
   <para>
    When verifying checksums, every file in the cluster is scanned. When
    enabling checksums, each relation file block with a changed checksum is
diff --git a/doc/src/sgml/wal.sgml b/doc/src/sgml/wal.sgml
index 0ba0c930b7..5061372c27 100644
--- a/doc/src/sgml/wal.sgml
+++ b/doc/src/sgml/wal.sgml
@@ -248,9 +248,10 @@
   <para>
    Checksums are normally enabled when the cluster is initialized using <link
    linkend="app-initdb-data-checksums"><application>initdb</application></link>.
-   They can also be enabled or disabled at a later time as an offline
-   operation. Data checksums are enabled or disabled at the full cluster
-   level, and cannot be specified individually for databases or tables.
+   They can also be enabled or disabled at a later time either as an offline
+   operation or online in a running cluster allowing concurrent access. Data
+   checksums are enabled or disabled at the full cluster level, and cannot be
+   specified individually for databases or tables.
   </para>
 
   <para>
@@ -267,7 +268,7 @@
   </para>
 
   <sect2 id="checksums-offline-enable-disable">
-   <title>Off-line Enabling of Checksums</title>
+   <title>Offline Enabling of Checksums</title>
 
    <para>
     The <link linkend="app-pgchecksums"><application>pg_checksums</application></link>
@@ -276,6 +277,54 @@
    </para>
 
   </sect2>
+
+  <sect2 id="checksums-online-enable-disable">
+   <title>Online Enabling of Checksums</title>
+
+   <para>
+    Checksums can be enabled or disabled online, by calling the appropriate
+    <link linkend="functions-admin-checksum">functions</link>.
+   </para>
+
+   <para>
+    Enabling checksums will put the cluster checksum mode in
+    <literal>inprogress-on</literal> mode.  During this time, checksums will be
+    written but not verified. In addition to this, a background worker process
+    is started that enables checksums on all existing data in the cluster. Once
+    this worker has completed processing all databases in the cluster, the
+    checksum mode will automatically switch to <literal>on</literal>. The
+    processing will consume two background worker processes, make sure that
+    <varname>max_worker_processes</varname> allows for at least two more
+    additional processes.
+   </para>
+
+   <para>
+    The process will initially wait for all open transactions to finish before
+    it starts, so that it can be certain that there are no tables that have been
+    created inside a transaction that has not committed yet and thus would not
+    be visible to the process enabling checksums. It will also, for each database,
+    wait for all pre-existing temporary tables to get removed before it finishes.
+    If long-lived temporary tables are used in the application it may be necessary
+    to terminate these application connections to allow the process to complete.
+   </para>
+
+   <para>
+    If the cluster is stopped while in <literal>inprogress-on</literal> mode, for
+    any reason, then this process must be restarted manually. To do this,
+    re-execute the function <function>pg_enable_data_checksums()</function>
+    once the cluster has been restarted. The background worker will attempt
+    to resume the work from where it was interrupted.
+   </para>
+
+   <note>
+    <para>
+     Enabling checksums can cause significant I/O to the system, as most of the
+     database pages will need to be rewritten, and will be written both to the
+     data files and the WAL.
+    </para>
+   </note>
+
+  </sect2>
  </sect1>
 
   <sect1 id="wal-intro">
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index da5e656a08..0a5516f45d 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -8576,6 +8576,7 @@ log_heap_visible(Relation rel, Buffer heap_buffer, Buffer vm_buffer,
 	XLogRegisterBuffer(0, vm_buffer, 0);
 
 	flags = REGBUF_STANDARD;
+
 	if (!XLogHintBitIsNeeded())
 		flags |= REGBUF_NO_IMAGE;
 	XLogRegisterBuffer(1, heap_buffer, flags);
diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.c
index 363294d623..465153e5d1 100644
--- a/src/backend/access/rmgrdesc/xlogdesc.c
+++ b/src/backend/access/rmgrdesc/xlogdesc.c
@@ -18,6 +18,7 @@
 #include "access/xlog.h"
 #include "access/xlog_internal.h"
 #include "catalog/pg_control.h"
+#include "storage/bufpage.h"
 #include "utils/guc.h"
 #include "utils/timestamp.h"
 
@@ -167,6 +168,26 @@ xlog_desc(StringInfo buf, XLogReaderState *record)
 		memcpy(&wal_level, rec, sizeof(int));
 		appendStringInfo(buf, "wal_level %s", get_wal_level_string(wal_level));
 	}
+	else if (info == XLOG_CHECKSUMS)
+	{
+		xl_checksum_state xlrec;
+
+		memcpy(&xlrec, rec, sizeof(xl_checksum_state));
+		switch (xlrec.new_checksumtype)
+		{
+			case PG_DATA_CHECKSUM_VERSION:
+				appendStringInfoString(buf, "on");
+				break;
+			case PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION:
+				appendStringInfoString(buf, "inprogress-off");
+				break;
+			case PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION:
+				appendStringInfoString(buf, "inprogress-on");
+				break;
+			default:
+				appendStringInfoString(buf, "off");
+		}
+	}
 }
 
 const char *
@@ -218,6 +239,9 @@ xlog_identify(uint8 info)
 		case XLOG_CHECKPOINT_REDO:
 			id = "CHECKPOINT_REDO";
 			break;
+		case XLOG_CHECKSUMS:
+			id = "CHECKSUMS";
+			break;
 	}
 
 	return id;
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 64304d77d3..ecded32414 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -646,6 +646,16 @@ static XLogRecPtr LocalMinRecoveryPoint;
 static TimeLineID LocalMinRecoveryPointTLI;
 static bool updateMinRecoveryPoint = true;
 
+/*
+ * Local state fror Controlfile data_checksum_version. After initialization
+ * this is only updated when absorbing a procsignal barrier during interrupt
+ * processing. The reason for keeping a copy in backend-private memory is to
+ * avoid locking for interrogating checksum state. Possible values are the
+ * checksum versions defined in storage/bufpage.h as well as zero when data
+ * checksums are disabled.
+ */
+static uint32 LocalDataChecksumVersion = 0;
+
 /* For WALInsertLockAcquire/Release functions */
 static int	MyLockNo = 0;
 static bool holdingAllLocks = false;
@@ -714,6 +724,8 @@ static void WALInsertLockAcquireExclusive(void);
 static void WALInsertLockRelease(void);
 static void WALInsertLockUpdateInsertingAt(XLogRecPtr insertingAt);
 
+static void XLogChecksums(ChecksumType new_type);
+
 /*
  * Insert an XLOG record represented by an already-constructed chain of data
  * chunks.  This is a low-level routine; to construct the WAL record header
@@ -827,9 +839,10 @@ XLogInsertRecord(XLogRecData *rdata,
 		 * only happen just after a checkpoint, so it's better to be slow in
 		 * this case and fast otherwise.
 		 *
-		 * Also check to see if fullPageWrites was just turned on or there's a
-		 * running backup (which forces full-page writes); if we weren't
-		 * already doing full-page writes then go back and recompute.
+		 * Also check to see if fullPageWrites was just turned on, there's a
+		 * running backup or if checksums are enabled (all of which forces
+		 * full-page writes); if we weren't already doing full-page writes
+		 * then go back and recompute.
 		 *
 		 * If we aren't doing full-page writes then RedoRecPtr doesn't
 		 * actually affect the contents of the XLOG record, so we'll update
@@ -842,7 +855,9 @@ XLogInsertRecord(XLogRecData *rdata,
 			Assert(RedoRecPtr < Insert->RedoRecPtr);
 			RedoRecPtr = Insert->RedoRecPtr;
 		}
-		doPageWrites = (Insert->fullPageWrites || Insert->runningBackups > 0);
+		doPageWrites = (Insert->fullPageWrites ||
+						Insert->runningBackups > 0 ||
+						DataChecksumsNeedWrite());
 
 		if (doPageWrites &&
 			(!prevDoPageWrites ||
@@ -4539,9 +4554,7 @@ ReadControlFile(void)
 
 	CalculateCheckpointSegments();
 
-	/* Make the initdb settings visible as GUC variables, too */
-	SetConfigOption("data_checksums", DataChecksumsEnabled() ? "yes" : "no",
-					PGC_INTERNAL, PGC_S_DYNAMIC_DEFAULT);
+	LocalDataChecksumVersion = ControlFile->data_checksum_version;
 }
 
 /*
@@ -4575,13 +4588,349 @@ GetMockAuthenticationNonce(void)
 }
 
 /*
- * Are checksums enabled for data pages?
+ * DataChecksumsNeedWrite
+ *		Returns whether data checksums must be written or not
+ *
+ * Returns true iff data checksums are enabled or are in the process of being
+ * enabled. During "inprogress-on" and "inprogress-off" states checksums must
+ * be written even though they are not verified (see datachecksumsworker.c for
+ * a longer discussion).
+ *
+ * This function is intended for callsites which are about to write a data page
+ * to storage, and need to know whether to re-calculate the checksum for the
+ * page header. Calling this function must be performed as close to the write
+ * operation as possible to keep the critical section short.
+ */
+bool
+DataChecksumsNeedWrite(void)
+{
+	return (LocalDataChecksumVersion == PG_DATA_CHECKSUM_VERSION ||
+			LocalDataChecksumVersion == PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION ||
+			LocalDataChecksumVersion == PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION);
+}
+
+/*
+ * DataChecksumsNeedVerify
+ *		Returns whether data checksums must be verified or not
+ *
+ * Data checksums are only verified if they are fully enabled in the cluster.
+ * During the "inprogress-on" and "inprogress-off" states they are only
+ * updated, not verified (see datachecksumsworker.c for a longer discussion).
+ *
+ * This function is intended for callsites which have read data and are about
+ * to perform checksum validation based on the result of this.  Calling this
+ * function must be performed as close to the validation call as possible to
+ * keep the critical section short. This is in order to protect against time of
+ * check/time of use situations around data checksum validation.
+ */
+bool
+DataChecksumsNeedVerify(void)
+{
+	return (LocalDataChecksumVersion == PG_DATA_CHECKSUM_VERSION);
+}
+
+/*
+ * DataChecksumsOnInProgress
+ *		Returns whether data checksums are being enabled
+ *
+ * Most operations don't need to worry about the "inprogress" states, and
+ * should use DataChecksumsNeedVerify() or DataChecksumsNeedWrite(). The
+ * "inprogress-on" state for enabling checksums is used when the checksum
+ * worker is setting checksums on all pages, it can thus be used to check for
+ * aborted checksum processing which need to be restarted.
+ */
+inline bool
+DataChecksumsOnInProgress(void)
+{
+	return (LocalDataChecksumVersion == PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION);
+}
+
+/*
+ * DataChecksumsOffInProgress
+ *		Returns whether data checksums are being disabled
+ *
+ * The "inprogress-off" state for disabling checksums is used for when the
+ * worker resets the catalog state.  DataChecksumsNeedVerify() or
+ * DataChecksumsNeedWrite() should be used for deciding whether to read/write
+ * checksums.
  */
 bool
-DataChecksumsEnabled(void)
+DataChecksumsOffInProgress(void)
+{
+	return (LocalDataChecksumVersion == PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION);
+}
+
+/*
+ * SetDataChecksumsOnInProgress
+ *		Sets the data checksum state to "inprogress-on" to enable checksums
+ *
+ * To start the process of enabling data checksums in a running cluster the
+ * data_checksum_version state must be changed to "inprogress-on". See
+ * SetDataChecksumsOn below for a description on how this state change works.
+ * This function blocks until all backends in the cluster have acknowledged the
+ * state transition.
+ */
+void
+SetDataChecksumsOnInProgress(void)
+{
+	uint64		barrier;
+
+	Assert(ControlFile != NULL);
+
+	/*
+	 * The state transition is performed in a critical section with
+	 * checkpoints held off to provide crash safety.
+	 */
+	MyProc->delayChkptFlags |= DELAY_CHKPT_START;
+	START_CRIT_SECTION();
+
+	XLogChecksums(PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION);
+
+	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+	ControlFile->data_checksum_version = PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION;
+	UpdateControlFile();
+	LWLockRelease(ControlFileLock);
+
+	barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON);
+
+	END_CRIT_SECTION();
+	MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
+
+	/*
+	 * Await state change in all backends to ensure that all backends are in
+	 * "inprogress-on". Once done we know that all backends are writing data
+	 * checksums.
+	 */
+	WaitForProcSignalBarrier(barrier);
+}
+
+/*
+ * SetDataChecksumsOn
+ *		Enables data checksums cluster-wide
+ *
+ * Enabling data checksums is performed using two barriers, the first one to
+ * set the checksums state to "inprogress-on" (which is performed by
+ * SetDataChecksumsOnInProgress()) and the second one to set the state to "on"
+ * (performed here).
+ *
+ * To start the process of enabling data checksums in a running cluster the
+ * data_checksum_version state must be changed to "inprogress-on".  This state
+ * requires data checksums to be written but not verified. This ensures that
+ * all data pages can be checksummed without the risk of false negatives in
+ * validation during the process.  When all existing pages are guaranteed to
+ * have checksums, and all new pages will be initiated with checksums, the
+ * state can be changed to "on". Once the state is "on" checksums will be both
+ * written and verified. See datachecksumsworker.c for a longer discussion on
+ * how data checksums can be enabled in a running cluster.
+ *
+ * This function blocks until all backends in the cluster have acknowledged the
+ * state transition.
+ */
+void
+SetDataChecksumsOn(void)
 {
+	uint64		barrier;
+
 	Assert(ControlFile != NULL);
-	return (ControlFile->data_checksum_version > 0);
+
+	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+
+	/*
+	 * The only allowed state transition to "on" is from "inprogress-on" since
+	 * that state ensures that all pages will have data checksums written.
+	 */
+	if (ControlFile->data_checksum_version != PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION)
+	{
+		LWLockRelease(ControlFileLock);
+		elog(ERROR, "checksums not in \"inprogress-on\" mode");
+	}
+
+	LWLockRelease(ControlFileLock);
+
+	MyProc->delayChkptFlags |= DELAY_CHKPT_START;
+	START_CRIT_SECTION();
+
+	XLogChecksums(PG_DATA_CHECKSUM_VERSION);
+
+	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+	ControlFile->data_checksum_version = PG_DATA_CHECKSUM_VERSION;
+	UpdateControlFile();
+	LWLockRelease(ControlFileLock);
+
+	barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_ON);
+
+	END_CRIT_SECTION();
+	MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
+
+	/*
+	 * Await state transition of "on" in all backends. When done we know that
+	 * data checksums are enabled in all backends and data checksums are both
+	 * written and verified.
+	 */
+	WaitForProcSignalBarrier(barrier);
+}
+
+/*
+ * SetDataChecksumsOff
+ *		Disables data checksums cluster-wide
+ *
+ * Disabling data checksums must be performed with two sets of barriers, each
+ * carrying a different state. The state is first set to "inprogress-off"
+ * during which checksums are still written but not verified. This ensures that
+ * backends which have yet to observe the state change from "on" won't get
+ * validation errors on concurrently modified pages. Once all backends have
+ * changed to "inprogress-off", the barrier for moving to "off" can be emitted.
+ * This function blocks until all backends in the cluster have acknowledged the
+ * state transition.
+ */
+void
+SetDataChecksumsOff(void)
+{
+	uint64		barrier;
+
+	Assert(ControlFile);
+
+	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+
+	/* If data checksums are already disabled there is nothing to do */
+	if (ControlFile->data_checksum_version == 0)
+	{
+		LWLockRelease(ControlFileLock);
+		return;
+	}
+
+	/*
+	 * If data checksums are currently enabled we first transition to the
+	 * "inprogress-off" state during which backends continue to write
+	 * checksums without verifying them. When all backends are in
+	 * "inprogress-off" the next transition to "off" can be performed, after
+	 * which all data checksum processing is disabled.
+	 */
+	if (ControlFile->data_checksum_version == PG_DATA_CHECKSUM_VERSION)
+	{
+		LWLockRelease(ControlFileLock);
+
+		MyProc->delayChkptFlags |= DELAY_CHKPT_START;
+		START_CRIT_SECTION();
+
+		XLogChecksums(PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION);
+
+		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+		ControlFile->data_checksum_version = PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION;
+		UpdateControlFile();
+		LWLockRelease(ControlFileLock);
+
+		barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF);
+
+		END_CRIT_SECTION();
+		MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
+
+
+		/*
+		 * Update local state in all backends to ensure that any backend in
+		 * "on" state is changed to "inprogress-off".
+		 */
+		WaitForProcSignalBarrier(barrier);
+
+		/*
+		 * At this point we know that no backends are verifying data checksums
+		 * during reading. Next, we can safely move to state "off" to also
+		 * stop writing checksums.
+		 */
+	}
+	else
+	{
+		/*
+		 * Ending up here implies that the checksums state is "inprogress-on"
+		 * or "inprogress-off" and we can transition directly to "off" from
+		 * there.
+		 */
+		LWLockRelease(ControlFileLock);
+	}
+
+	/*
+	 * Ensure that we don't incur a checkpoint during disabling checksums.
+	 */
+	MyProc->delayChkptFlags |= DELAY_CHKPT_START;
+	START_CRIT_SECTION();
+
+	XLogChecksums(0);
+
+	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+	ControlFile->data_checksum_version = 0;
+	UpdateControlFile();
+	LWLockRelease(ControlFileLock);
+
+	barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_OFF);
+
+	END_CRIT_SECTION();
+	MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
+
+	WaitForProcSignalBarrier(barrier);
+}
+
+/*
+ * ProcSignalBarrier absorption functions for enabling and disabling data
+ * checksums in a running cluster. The procsignalbarriers are emitted in the
+ * SetDataChecksums* functions.
+ */
+bool
+AbsorbChecksumsOnInProgressBarrier(void)
+{
+	LocalDataChecksumVersion = PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION;
+	return true;
+}
+
+bool
+AbsorbChecksumsOnBarrier(void)
+{
+	Assert(LocalDataChecksumVersion == PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION);
+	LocalDataChecksumVersion = PG_DATA_CHECKSUM_VERSION;
+	return true;
+}
+
+bool
+AbsorbChecksumsOffInProgressBarrier(void)
+{
+	LocalDataChecksumVersion = PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION;
+	return true;
+}
+
+bool
+AbsorbChecksumsOffBarrier(void)
+{
+	LocalDataChecksumVersion = 0;
+	return true;
+}
+
+/*
+ * InitLocalControlData
+ *
+ * Set up backend local caches of controldata variables which may change at
+ * any point during runtime and thus require special cased locking. So far
+ * this only applies to data_checksum_version, but it's intended to be general
+ * purpose enough to handle future cases.
+ */
+void
+InitLocalControldata(void)
+{
+	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+	LocalDataChecksumVersion = ControlFile->data_checksum_version;
+	LWLockRelease(ControlFileLock);
+}
+
+/* guc hook */
+const char *
+show_data_checksums(void)
+{
+	if (LocalDataChecksumVersion == PG_DATA_CHECKSUM_VERSION)
+		return "on";
+	else if (LocalDataChecksumVersion == PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION)
+		return "inprogress-on";
+	else if (LocalDataChecksumVersion == PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION)
+		return "inprogress-off";
+	else
+		return "off";
 }
 
 /*
@@ -6141,6 +6490,33 @@ StartupXLOG(void)
 	 */
 	CompleteCommitTsInitialization();
 
+	/*
+	 * If we reach this point with checksums being enabled ("inprogress-on"
+	 * state), we notify the user that they need to manually restart the
+	 * process to enable checksums. This is because we cannot launch a dynamic
+	 * background worker directly from here, it has to be launched from a
+	 * regular backend.
+	 */
+	if (ControlFile->data_checksum_version == PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION)
+		ereport(WARNING,
+				(errmsg("data checksums are being enabled, but no worker is running"),
+				 errhint("If checksums were being enabled during shutdown then processing must be manually restarted.")));
+
+	/*
+	 * If data checksums were being disabled when the cluster was shut down,
+	 * we know that we have a state where all backends have stopped validating
+	 * checksums and we can move to off instead of prompting the user to
+	 * perform any action.
+	 */
+	if (ControlFile->data_checksum_version == PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION)
+	{
+		XLogChecksums(0);
+
+		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+		ControlFile->data_checksum_version = 0;
+		LWLockRelease(ControlFileLock);
+	}
+
 	/*
 	 * All done with end-of-recovery actions.
 	 *
@@ -8152,6 +8528,24 @@ XLogReportParameters(void)
 	}
 }
 
+/*
+ * Log the new state of checksums
+ */
+static void
+XLogChecksums(ChecksumType new_type)
+{
+	xl_checksum_state xlrec;
+	XLogRecPtr	recptr;
+
+	xlrec.new_checksumtype = new_type;
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, sizeof(xl_checksum_state));
+
+	recptr = XLogInsert(RM_XLOG_ID, XLOG_CHECKSUMS);
+	XLogFlush(recptr);
+}
+
 /*
  * Update full_page_writes in shared memory, and write an
  * XLOG_FPW_CHANGE record if necessary.
@@ -8580,6 +8974,47 @@ xlog_redo(XLogReaderState *record)
 	{
 		/* nothing to do here, just for informational purposes */
 	}
+	else if (info == XLOG_CHECKSUMS)
+	{
+		xl_checksum_state state;
+		uint64		barrier;
+
+		memcpy(&state, XLogRecGetData(record), sizeof(xl_checksum_state));
+
+		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+		ControlFile->data_checksum_version = state.new_checksumtype;
+		UpdateControlFile();
+		LWLockRelease(ControlFileLock);
+
+		/*
+		 * Block on a procsignalbarrier to await all processes having seen the
+		 * change to checksum status. Once the barrier has been passed we can
+		 * initiate the corresponding processing.
+		 */
+		switch (state.new_checksumtype)
+		{
+			case PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION:
+				barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON);
+				WaitForProcSignalBarrier(barrier);
+				break;
+
+			case PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION:
+				barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF);
+				WaitForProcSignalBarrier(barrier);
+				break;
+
+			case PG_DATA_CHECKSUM_VERSION:
+				barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_ON);
+				WaitForProcSignalBarrier(barrier);
+				break;
+
+			default:
+				Assert(state.new_checksumtype == 0);
+				barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_OFF);
+				WaitForProcSignalBarrier(barrier);
+				break;
+		}
+	}
 }
 
 /*
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index 3e3d2bb618..c1b22e0b9c 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -27,6 +27,7 @@
 #include "funcapi.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "postmaster/datachecksumsworker.h"
 #include "replication/walreceiver.h"
 #include "storage/fd.h"
 #include "storage/proc.h"
@@ -803,3 +804,43 @@ pg_wal_replay_wait(PG_FUNCTION_ARGS)
 
 	PG_RETURN_VOID();
 }
+
+/*
+ * Disables data checksums for the cluster, if applicable. Starts a background
+ * worker which turns off the data checksums.
+ */
+Datum
+disable_data_checksums(PG_FUNCTION_ARGS)
+{
+	if (!superuser())
+		ereport(ERROR, errmsg("must be superuser"));
+
+	StartDataChecksumsWorkerLauncher(false, 0, 0, false);
+	PG_RETURN_VOID();
+}
+
+/*
+ * Enables data checksums for the cluster, if applicable.  Supports vacuum-
+ * like cost based throttling to limit system load. Starts a background worker
+ * which updates data checksums on existing data.
+ */
+Datum
+enable_data_checksums(PG_FUNCTION_ARGS)
+{
+	int			cost_delay = PG_GETARG_INT32(0);
+	int			cost_limit = PG_GETARG_INT32(1);
+	bool		fast = PG_GETARG_BOOL(2);
+
+	if (!superuser())
+		ereport(ERROR, errmsg("must be superuser"));
+
+	if (cost_delay < 0)
+		ereport(ERROR, errmsg("cost delay cannot be a negative value"));
+
+	if (cost_limit <= 0)
+		ereport(ERROR, errmsg("cost limit must be greater than zero"));
+
+	StartDataChecksumsWorkerLauncher(true, cost_delay, cost_limit, fast);
+
+	PG_RETURN_VOID();
+}
diff --git a/src/backend/backup/basebackup.c b/src/backend/backup/basebackup.c
index 14e5ba72e9..3c64413dd4 100644
--- a/src/backend/backup/basebackup.c
+++ b/src/backend/backup/basebackup.c
@@ -1612,7 +1612,8 @@ sendFile(bbsink *sink, const char *readfilename, const char *tarfilename,
 	 * enabled for this cluster, and if this is a relation file, then verify
 	 * the checksum.
 	 */
-	if (!noverify_checksums && DataChecksumsEnabled() &&
+	if (!noverify_checksums &&
+		DataChecksumsNeedWrite() &&
 		RelFileNumberIsValid(relfilenumber))
 		verify_checksum = true;
 
@@ -2005,6 +2006,9 @@ verify_page_checksum(Page page, XLogRecPtr start_lsn, BlockNumber blkno,
 	if (PageIsNew(page) || PageGetLSN(page) >= start_lsn)
 		return true;
 
+	if (!DataChecksumsNeedVerify())
+		return true;
+
 	/* Perform the actual checksum calculation. */
 	checksum = pg_checksum_page(page, blkno);
 
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql
index b0d0de051e..16c77db056 100644
--- a/src/backend/catalog/system_functions.sql
+++ b/src/backend/catalog/system_functions.sql
@@ -639,6 +639,11 @@ LANGUAGE INTERNAL
 CALLED ON NULL INPUT VOLATILE PARALLEL SAFE
 AS 'pg_stat_reset_slru';
 
+CREATE OR REPLACE FUNCTION
+  pg_enable_data_checksums(cost_delay integer DEFAULT 0, cost_limit integer DEFAULT 100, fast boolean DEFAULT false)
+  RETURNS void STRICT VOLATILE LANGUAGE internal AS 'enable_data_checksums'
+  PARALLEL RESTRICTED;
+
 --
 -- The default permissions for functions mean that anyone can execute them.
 -- A number of functions shouldn't be executable by just anyone, but rather
@@ -760,6 +765,8 @@ REVOKE EXECUTE ON FUNCTION pg_ls_logicalmapdir() FROM PUBLIC;
 
 REVOKE EXECUTE ON FUNCTION pg_ls_replslotdir(text) FROM PUBLIC;
 
+REVOKE EXECUTE ON FUNCTION pg_enable_data_checksums(integer, integer, boolean) FROM public;
+
 --
 -- We also set up some things as accessible to standard roles.
 --
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 49109dbdc8..e9d23d6826 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1322,6 +1322,27 @@ CREATE VIEW pg_stat_progress_copy AS
     FROM pg_stat_get_progress_info('COPY') AS S
         LEFT JOIN pg_database D ON S.datid = D.oid;
 
+CREATE VIEW pg_stat_progress_datachecksums AS
+	SELECT
+		S.pid AS pid, S.datid AS datid, D.datname AS datname,
+		CASE S.param1 WHEN 0 THEN 'enabling'
+                      WHEN 1 THEN 'disabling'
+					  WHEN 2 THEN 'waiting'
+					  WHEN 3 THEN 'waiting on backends'
+					  WHEN 4 THEN 'waiting on temporary tables'
+					  WHEN 5 THEN 'done'
+					  END AS phase,
+		CASE S.param2 WHEN -1 THEN NULL ELSE S.param2 END AS databases_total,
+		CASE S.param3 WHEN -1 THEN NULL ELSE S.param3 END AS relations_total,
+		S.param4 AS databases_processed,
+		S.param5 AS relations_processed,
+		S.param6 AS databases_current,
+		S.param7 AS relation_current,
+		S.param8 AS relation_current_blocks,
+		S.param9 AS relation_current_blocks_processed
+	FROM pg_stat_get_progress_info('DATACHECKSUMS') AS S
+		LEFT JOIN pg_database D ON S.datid = D.oid;
+
 CREATE VIEW pg_user_mappings AS
     SELECT
         U.oid       AS umid,
diff --git a/src/backend/postmaster/Makefile b/src/backend/postmaster/Makefile
index db08543d19..e112d4b53e 100644
--- a/src/backend/postmaster/Makefile
+++ b/src/backend/postmaster/Makefile
@@ -18,6 +18,7 @@ OBJS = \
 	bgworker.o \
 	bgwriter.o \
 	checkpointer.o \
+	datachecksumsworker.o \
 	fork_process.o \
 	interrupt.o \
 	launch_backend.o \
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 07bc5517fc..662cd12681 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -18,6 +18,7 @@
 #include "pgstat.h"
 #include "port/atomics.h"
 #include "postmaster/bgworker_internals.h"
+#include "postmaster/datachecksumsworker.h"
 #include "postmaster/postmaster.h"
 #include "replication/logicallauncher.h"
 #include "replication/logicalworker.h"
@@ -132,6 +133,12 @@ static const struct
 	},
 	{
 		"TablesyncWorkerMain", TablesyncWorkerMain
+	},
+	{
+		"DataChecksumsWorkerLauncherMain", DataChecksumsWorkerLauncherMain
+	},
+	{
+		"DataChecksumsWorkerMain", DataChecksumsWorkerMain
 	}
 };
 
diff --git a/src/backend/postmaster/datachecksumsworker.c b/src/backend/postmaster/datachecksumsworker.c
new file mode 100644
index 0000000000..72c75a092d
--- /dev/null
+++ b/src/backend/postmaster/datachecksumsworker.c
@@ -0,0 +1,1404 @@
+/*-------------------------------------------------------------------------
+ *
+ * datachecksumsworker.c
+ *	  Background worker for enabling or disabling data checksums online
+ *
+ * When enabling data checksums on a database at initdb time or when shut down
+ * with pg_checksums, no extra process is required as each page is checksummed,
+ * and verified, when accessed.  When enabling checksums on an already running
+ * cluster, this worker will ensure that all pages are checksummed before
+ * verification of the checksums is turned on. In the case of disabling
+ * checksums, the state transition is performed only in the control file, no
+ * changes are performed on the data pages.
+ *
+ * Checksums can be either enabled or disabled cluster-wide, with on/off being
+ * the end state for data_checksums.
+ *
+ * Enabling checksums
+ * ------------------
+ * When enabling checksums in an online cluster, data_checksums will be set to
+ * "inprogress-on" which signals that write operations MUST compute and write
+ * the checksum on the data page, but during reading the checksum SHALL NOT be
+ * verified. This ensures that all objects created during checksumming will
+ * have checksums set, but no reads will fail due to incorrect checksum. The
+ * DataChecksumsWorker will compile a list of databases which exist at the
+ * start of checksumming, and all of these which haven't been dropped during
+ * the processing MUST have been processed successfully in order for checksums
+ * to be enabled. Any new relation created during processing will see the
+ * in-progress state and will automatically be checksummed.
+ *
+ * For each database, all relations which have storage are read and every data
+ * page is marked dirty to force a write with the checksum. This will generate
+ * a lot of WAL as the entire database is read and written.
+ *
+ * If the processing is interrupted by a cluster restart, it will be restarted
+ * from the beginning again as state isn't persisted.
+ *
+ * Disabling checksums
+ * -------------------
+ * When disabling checksums, data_checksums will be set to "inprogress-off"
+ * which signals that checksums are written but no longer verified. This ensure
+ * that backends which have yet to move from the "on" state will still be able
+ * to process data checksum validation.
+ *
+ * Synchronization and Correctness
+ * -------------------------------
+ * The processes involved in enabling, or disabling, data checksums in an
+ * online cluster must be properly synchronized with the normal backends
+ * serving concurrent queries to ensure correctness. Correctness is defined
+ * as the following:
+ *
+ *    - Backends SHALL NOT violate local data_checksums state
+ *    - Data checksums SHALL NOT be considered enabled cluster-wide until all
+ *      currently connected backends have the local state "enabled"
+ *
+ * There are two levels of synchronization required for enabling data checksums
+ * in an online cluster: (i) changing state in the active backends ("on",
+ * "off", "inprogress-on" and "inprogress-off"), and (ii) ensuring no
+ * incompatible objects and processes are left in a database when workers end.
+ * The former deals with cluster-wide agreement on data checksum state and the
+ * latter with ensuring that any concurrent activity cannot break the data
+ * checksum contract during processing.
+ *
+ * Synchronizing the state change is done with procsignal barriers, where the
+ * WAL logging backend updating the global state in the controlfile will wait
+ * for all other backends to absorb the barrier. Barrier absorption will happen
+ * during interrupt processing, which means that connected backends will change
+ * state at different times. To prevent data checksum state changes when
+ * writing and verifying checksums, interrupts shall be held off before
+ * interrogating state and resumed when the IO operation has been performed.
+ *
+ *   When Enabling Data Checksums
+ *   ----------------------------
+ *   A process which fails to observe data checksums being enabled can induce
+ *   two types of errors: failing to write the checksum when modifying the page
+ *   and failing to validate the data checksum on the page when reading it.
+ *
+ *   When processing starts all backends belong to one of the below sets, with
+ *   one set being empty:
+ *
+ *   Bd: Backends in "off" state
+ *   Bi: Backends in "inprogress-on" state
+ *
+ *   If processing is started in an online cluster then all backends are in Bd.
+ *   If processing was halted by the cluster shutting down, the controlfile
+ *   state "inprogress-on" will be observed on system startup and all backends
+ *   will be in Bd. Backends transition Bd -> Bi via a procsignalbarrier.  When
+ *   the DataChecksumsWorker has finished writing checksums on all pages and
+ *   enables data checksums cluster-wide, there are four sets of backends where
+ *   Bd shall be an empty set:
+ *
+ *   Bg: Backend updating the global state and emitting the procsignalbarrier
+ *   Bd: Backends in "off" state
+ *   Be: Backends in "on" state
+ *   Bi: Backends in "inprogress-on" state
+ *
+ *   Backends in Bi and Be will write checksums when modifying a page, but only
+ *   backends in Be will verify the checksum during reading. The Bg backend is
+ *   blocked waiting for all backends in Bi to process interrupts and move to
+ *   Be. Any backend starting while Bg is waiting on the procsignalbarrier will
+ *   observe the global state being "on" and will thus automatically belong to
+ *   Be.  Checksums are enabled cluster-wide when Bi is an empty set. Bi and Be
+ *   are compatible sets while still operating based on their local state as
+ *   both write data checksums.
+ *
+ *   When Disabling Data Checksums
+ *   -----------------------------
+ *   A process which fails to observe that data checksums have been disabled
+ *   can induce two types of errors: writing the checksum when modifying the
+ *   page and validating a data checksum which is no longer correct due to
+ *   modifications to the page.
+ *
+ *   Bg: Backend updating the global state and emitting the procsignalbarrier
+ *   Bd: Backends in "off" state
+ *   Be: Backends in "on" state
+ *   Bo: Backends in "inprogress-off" state
+ *
+ *   Backends transition from the Be state to Bd like so: Be -> Bo -> Bd
+ *
+ *   The goal is to transition all backends to Bd making the others empty sets.
+ *   Backends in Bo write data checksums, but don't validate them, such that
+ *   backends still in Be can continue to validate pages until the barrier has
+ *   been absorbed such that they are in Bo. Once all backends are in Bo, the
+ *   barrier to transition to "off" can be raised and all backends can safely
+ *   stop writing data checksums as no backend is enforcing data checksum
+ *   validation any longer.
+ *
+ *
+ * Potential optimizations
+ * -----------------------
+ * Below are some potential optimizations and improvements which were brought
+ * up during reviews of this feature, but which weren't implemented in the
+ * initial version. These are ideas listed without any validation on their
+ * feasibility or potential payoff. More discussion on these can be found on
+ * the -hackers threads linked to in the commit message of this feature.
+ *
+ *   * Launching datachecksumsworker for resuming operation from the startup
+ *     process: Currently users have to restart processing manually after a
+ *     restart since dynamic background worker cannot be started from the
+ *     postmaster. Changing the startup process could make restarting the
+ *     processing automatic on cluster restart.
+ *   * Avoid dirtying the page when checksums already match: Iff the checksum
+ *     on the page happens to already match we still dirty the page. It should
+ *     be enough to only do the log_newpage_buffer() call in that case.
+ *   * Invent a lightweight WAL record that doesn't contain the full-page
+ *     image but just the block number: On replay, the redo routine would read
+ *     the page from disk.
+ *   * Teach pg_checksums to avoid checksummed pages when pg_checksums is used
+ *     to enable checksums on a cluster which is in inprogress-on state and
+ *     may have checksummed pages (make pg_checksums be able to resume an
+ *     online operation).
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/postmaster/datachecksumsworker.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/genam.h"
+#include "access/heapam.h"
+#include "access/htup_details.h"
+#include "access/xact.h"
+#include "access/xloginsert.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_class.h"
+#include "catalog/pg_database.h"
+#include "commands/progress.h"
+#include "commands/vacuum.h"
+#include "common/relpath.h"
+#include "miscadmin.h"
+#include "pgstat.h"
+#include "postmaster/bgworker.h"
+#include "postmaster/bgwriter.h"
+#include "postmaster/datachecksumsworker.h"
+#include "storage/bufmgr.h"
+#include "storage/checksum.h"
+#include "storage/ipc.h"
+#include "storage/lmgr.h"
+#include "storage/lwlock.h"
+#include "storage/procarray.h"
+#include "storage/smgr.h"
+#include "tcop/tcopprot.h"
+#include "utils/fmgroids.h"
+#include "utils/lsyscache.h"
+#include "utils/ps_status.h"
+#include "utils/syscache.h"
+
+/*
+ * Number of times we retry to open a database before giving up and consider
+ * it to have failed processing.
+ */
+#define DATACHECKSUMSWORKER_MAX_DB_RETRIES 5
+
+typedef enum
+{
+	DATACHECKSUMSWORKER_SUCCESSFUL = 0,
+	DATACHECKSUMSWORKER_ABORTED,
+	DATACHECKSUMSWORKER_FAILED,
+	DATACHECKSUMSWORKER_RETRYDB,
+} DataChecksumsWorkerResult;
+
+/*
+ * Signaling between backends calling pg_enable/disable_data_checksums, the
+ * checksums launcher process, and the checksums worker process.
+ *
+ * This struct is protected by DataChecksumsWorkerLock
+ */
+typedef struct DataChecksumsWorkerShmemStruct
+{
+	/*
+	 * These are set by pg_enable/disable_data_checksums, to tell the launcher
+	 * what the target state is.
+	 */
+	bool		launch_enable_checksums;	/* True if checksums are being
+											 * enabled, else false */
+	int			launch_cost_delay;
+	int			launch_cost_limit;
+	bool		launch_fast;
+
+	/*
+	 * Is a launcher process is currently running?
+	 *
+	 * This is set by the launcher process, after it has read the above
+	 * launch_* parameters.
+	 */
+	bool		launcher_running;
+
+	/*
+	 * These fields indicate the target state that the launcher is currently
+	 * working towards. They can be different from the corresponding launch_*
+	 * fields, if a new pg_enable/disable_data_checksums() call was made while
+	 * the launcher/worker was already running.
+	 *
+	 * The below members are set when the launcher starts, and are only
+	 * accessed read-only by the single worker. Thus, we can access these
+	 * without a lock. If multiple workers, or dynamic cost parameters, are
+	 * supported at some point then this would need to be revisited.
+	 */
+	bool		enabling_checksums; /* True if checksums are being enabled,
+									 * else false */
+	int			cost_delay;
+	int			cost_limit;
+	bool		immediate_checkpoint;
+
+	/*
+	 * Signaling between the launcher and the worker process.
+	 *
+	 * As there is only a single worker, and the launcher won't read these
+	 * until the worker exits, they can be accessed without the need for a
+	 * lock. If multiple workers are supported then this will have to be
+	 * revisited.
+	 */
+
+	/* result, set by worker before exiting */
+	DataChecksumsWorkerResult success;
+
+	/*
+	 * tells the worker process whether it should also process the shared
+	 * catalogs
+	 */
+	bool		process_shared_catalogs;
+} DataChecksumsWorkerShmemStruct;
+
+/* Shared memory segment for datachecksumsworker */
+static DataChecksumsWorkerShmemStruct *DataChecksumsWorkerShmem;
+
+/* Bookkeeping for work to do */
+typedef struct DataChecksumsWorkerDatabase
+{
+	Oid			dboid;
+	char	   *dbname;
+} DataChecksumsWorkerDatabase;
+
+typedef struct DataChecksumsWorkerResultEntry
+{
+	Oid			dboid;
+	DataChecksumsWorkerResult result;
+	int			retries;
+} DataChecksumsWorkerResultEntry;
+
+
+/*
+ * Flag set by the interrupt handler
+ */
+static volatile sig_atomic_t abort_requested = false;
+
+/*
+ * Have we set the DataChecksumsWorkerShmemStruct->launcher_running flag?
+ * If we have, we need to clear it before exiting!
+ */
+static volatile sig_atomic_t launcher_running = false;
+
+/*
+ * Are we enabling data checksums, or disabling them?
+ */
+static bool enabling_checksums;
+
+/* Prototypes */
+static List *BuildDatabaseList(void);
+static List *BuildRelationList(bool temp_relations, bool include_shared);
+static void FreeDatabaseList(List *dblist);
+static DataChecksumsWorkerResult ProcessDatabase(DataChecksumsWorkerDatabase *db);
+static bool ProcessAllDatabases(bool immediate_checkpoint);
+static bool ProcessSingleRelationFork(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy);
+static void launcher_cancel_handler(SIGNAL_ARGS);
+static void WaitForAllTransactionsToFinish(void);
+
+/*
+ * StartDataChecksumsWorkerLauncher
+ *		Main entry point for datachecksumsworker launcher process
+ *
+ * The main entrypoint for starting data checksums processing for enabling as
+ * well as disabling.
+ */
+void
+StartDataChecksumsWorkerLauncher(bool enable_checksums,
+								 int cost_delay,
+								 int cost_limit,
+								 bool fast)
+{
+	BackgroundWorker bgw;
+	BackgroundWorkerHandle *bgw_handle;
+	bool		launcher_running;
+
+	/* the cost delay settings have no effect when disabling */
+	Assert(enable_checksums || cost_delay == 0);
+	Assert(enable_checksums || cost_limit == 0);
+
+	/* store the desired state in shared memory */
+	LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
+
+	DataChecksumsWorkerShmem->launch_enable_checksums = enable_checksums;
+	DataChecksumsWorkerShmem->launch_cost_delay = cost_delay;
+	DataChecksumsWorkerShmem->launch_cost_limit = cost_limit;
+	DataChecksumsWorkerShmem->launch_fast = fast;
+
+	/* is the launcher already running? */
+	launcher_running = DataChecksumsWorkerShmem->launcher_running;
+
+	LWLockRelease(DataChecksumsWorkerLock);
+
+	/*
+	 * Launch a new launcher process, if it's not running already.
+	 *
+	 * If the launcher is currently busy enabling the checksums, and we want
+	 * them disabled (or vice versa), the launcher will notice that at latest
+	 * when it's about to exit, and will loop back process the new request. So
+	 * if the launcher is already running, we don't need to do anything more
+	 * here to abort it.
+	 *
+	 * If you call pg_enable/disable_data_checksums() twice in a row, before
+	 * the launcher has had a chance to start up, we still end up launching it
+	 * twice.  That's OK, the second invocation will see that a launcher is
+	 * already running and exit quickly.
+	 *
+	 * TODO: We could optimize here and skip launching the launcher, if we are
+	 * already in the desired state, i.e. if the checksums are already enabled
+	 * and you call pg_enable_data_checksums().
+	 */
+	if (!launcher_running)
+	{
+		/*
+		 * Prepare the BackgroundWorker and launch it.
+		 */
+		memset(&bgw, 0, sizeof(bgw));
+		bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
+		bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
+		snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
+		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "DataChecksumsWorkerLauncherMain");
+		snprintf(bgw.bgw_name, BGW_MAXLEN, "datachecksumsworker launcher");
+		snprintf(bgw.bgw_type, BGW_MAXLEN, "datachecksumsworker launcher");
+		bgw.bgw_restart_time = BGW_NEVER_RESTART;
+		bgw.bgw_notify_pid = MyProcPid;
+		bgw.bgw_main_arg = (Datum) 0;
+
+		if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
+			ereport(ERROR,
+					(errmsg("failed to start background worker to process data checksums")));
+	}
+}
+
+/*
+ * ProcessSingleRelationFork
+ *		Enable data checksums in a single relation/fork.
+ *
+ * Returns true if successful, and false if *aborted*. On error, an actual
+ * error is raised in the lower levels.
+ */
+static bool
+ProcessSingleRelationFork(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy)
+{
+	BlockNumber numblocks = RelationGetNumberOfBlocksInFork(reln, forkNum);
+	BlockNumber blknum;
+	char		activity[NAMEDATALEN * 2 + 128];
+	char	   *relns;
+
+	relns = get_namespace_name(RelationGetNamespace(reln));
+
+	if (!relns)
+		return false;
+
+	pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_CUR_REL_TOTAL_BLOCKS,
+								 numblocks);
+
+	/* Report the current relation to pgstat_activity */
+	snprintf(activity, sizeof(activity) - 1, "processing: %s.%s (%s, %dblocks)",
+			 relns, RelationGetRelationName(reln), forkNames[forkNum], numblocks);
+	pgstat_report_activity(STATE_RUNNING, activity);
+
+	/*
+	 * We are looping over the blocks which existed at the time of process
+	 * start, which is safe since new blocks are created with checksums set
+	 * already due to the state being "inprogress-on".
+	 */
+	for (blknum = 0; blknum < numblocks; blknum++)
+	{
+		Buffer		buf = ReadBufferExtended(reln, forkNum, blknum, RBM_NORMAL, strategy);
+
+		/* Progress report the current block */
+		pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_CUR_REL_PROCESSED_BLOCKS,
+									 blknum);
+
+		/* Need to get an exclusive lock before we can flag as dirty */
+		LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+
+		/*
+		 * Mark the buffer as dirty and force a full page write.  We have to
+		 * re-write the page to WAL even if the checksum hasn't changed,
+		 * because if there is a replica it might have a slightly different
+		 * version of the page with an invalid checksum, caused by unlogged
+		 * changes (e.g. hintbits) on the master happening while checksums
+		 * were off. This can happen if there was a valid checksum on the page
+		 * at one point in the past, so only when checksums are first on, then
+		 * off, and then turned on again. Iff wal_level is set to "minimal",
+		 * this could be avoided iff the checksum is calculated to be correct.
+		 */
+		START_CRIT_SECTION();
+		MarkBufferDirty(buf);
+		log_newpage_buffer(buf, false);
+		END_CRIT_SECTION();
+
+		UnlockReleaseBuffer(buf);
+
+		/*
+		 * This is the only place where we check if we are asked to abort, the
+		 * abortion will bubble up from here. It's safe to check this without
+		 * a lock, because if we miss it being set, we will try again soon.
+		 */
+		Assert(enabling_checksums);
+		if (!DataChecksumsWorkerShmem->launch_enable_checksums)
+			abort_requested = true;
+		if (abort_requested)
+			return false;
+
+		vacuum_delay_point();
+	}
+
+	pfree(relns);
+	return true;
+}
+
+/*
+ * ProcessSingleRelationByOid
+ *		Process a single relation based on oid.
+ *
+ * Returns true if successful, and false if *aborted*. On error, an actual
+ * error is raised in the lower levels.
+ */
+static bool
+ProcessSingleRelationByOid(Oid relationId, BufferAccessStrategy strategy)
+{
+	Relation	rel;
+	ForkNumber	fnum;
+	bool		aborted = false;
+
+	StartTransactionCommand();
+
+	pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_CUR_REL, relationId);
+
+	rel = try_relation_open(relationId, AccessShareLock);
+	if (rel == NULL)
+	{
+		/*
+		 * Relation no longer exists. We don't consider this an error since
+		 * there are no pages in it that need data checksums, and thus return
+		 * true. The worker operates off a list of relations generated at the
+		 * start of processing, so relations being dropped in the meantime is
+		 * to be expected.
+		 */
+		CommitTransactionCommand();
+		pgstat_report_activity(STATE_IDLE, NULL);
+		return true;
+	}
+	RelationGetSmgr(rel);
+
+	for (fnum = 0; fnum <= MAX_FORKNUM; fnum++)
+	{
+		if (smgrexists(rel->rd_smgr, fnum))
+		{
+			if (!ProcessSingleRelationFork(rel, fnum, strategy))
+			{
+				aborted = true;
+				break;
+			}
+		}
+	}
+	relation_close(rel, AccessShareLock);
+	elog(DEBUG2,
+		 "data checksum processing done for relation with OID %u: %s",
+		 relationId, (aborted ? "aborted" : "finished"));
+
+	CommitTransactionCommand();
+
+	pgstat_report_activity(STATE_IDLE, NULL);
+
+	return !aborted;
+}
+
+/*
+ * ProcessDatabase
+ *		Enable data checksums in a single database.
+ *
+ * We do this by launching a dynamic background worker into this database, and
+ * waiting for it to finish.  We have to do this in a separate worker, since
+ * each process can only be connected to one database during its lifetime.
+ */
+static DataChecksumsWorkerResult
+ProcessDatabase(DataChecksumsWorkerDatabase *db)
+{
+	BackgroundWorker bgw;
+	BackgroundWorkerHandle *bgw_handle;
+	BgwHandleStatus status;
+	pid_t		pid;
+	char		activity[NAMEDATALEN + 64];
+
+	DataChecksumsWorkerShmem->success = DATACHECKSUMSWORKER_FAILED;
+
+	memset(&bgw, 0, sizeof(bgw));
+	bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
+	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
+	snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
+	snprintf(bgw.bgw_function_name, BGW_MAXLEN, "%s", "DataChecksumsWorkerMain");
+	snprintf(bgw.bgw_name, BGW_MAXLEN, "datachecksumsworker worker");
+	snprintf(bgw.bgw_type, BGW_MAXLEN, "datachecksumsworker worker");
+	bgw.bgw_restart_time = BGW_NEVER_RESTART;
+	bgw.bgw_notify_pid = MyProcPid;
+	bgw.bgw_main_arg = ObjectIdGetDatum(db->dboid);
+
+	/*
+	 * If there are no worker slots available, make sure we retry processing
+	 * this database. This will make the datachecksumsworker move on to the
+	 * next database and quite likely fail with the same problem. TODO: Maybe
+	 * we need a backoff to avoid running through all the databases here in
+	 * short order.
+	 */
+	if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
+	{
+		ereport(WARNING,
+				(errmsg("failed to start worker for enabling data checksums in database \"%s\", retrying",
+						db->dbname),
+				 errhint("The max_worker_processes setting might be too low.")));
+		return DATACHECKSUMSWORKER_RETRYDB;
+	}
+
+	status = WaitForBackgroundWorkerStartup(bgw_handle, &pid);
+	if (status == BGWH_STOPPED)
+	{
+		ereport(WARNING,
+				(errmsg("could not start background worker for enabling data checksums in database \"%s\"",
+						db->dbname),
+				 errhint("More details on the error might be found in the server log.")));
+		return DATACHECKSUMSWORKER_FAILED;
+	}
+
+	/*
+	 * If the postmaster crashed we cannot end up with a processed database so
+	 * we have no alternative other than exiting. When enabling checksums we
+	 * won't at this time have changed the pg_control version to enabled so
+	 * when the cluster comes back up processing will have to be restarted.
+	 * When disabling, the pg_control version will be set to off before this
+	 * so when the cluster comes up checksums will be off as expected.
+	 */
+	if (status == BGWH_POSTMASTER_DIED)
+		ereport(FATAL,
+				(errmsg("cannot enable data checksums without the postmaster process"),
+				 errhint("Restart the database and restart data checksum processing by calling pg_enable_data_checksums().")));
+
+	Assert(status == BGWH_STARTED);
+	ereport(DEBUG1,
+			(errmsg("initiating data checksum processing in database \"%s\"",
+					db->dbname)));
+
+	snprintf(activity, sizeof(activity) - 1,
+			 "Waiting for worker in database %s (pid %d)", db->dbname, pid);
+	pgstat_report_activity(STATE_RUNNING, activity);
+
+	status = WaitForBackgroundWorkerShutdown(bgw_handle);
+	if (status == BGWH_POSTMASTER_DIED)
+		ereport(FATAL,
+				(errmsg("postmaster exited during data checksum processing in \"%s\"",
+						db->dbname),
+				 errhint("Restart the database and restart data checksum processing by calling pg_enable_data_checksums().")));
+
+	if (DataChecksumsWorkerShmem->success == DATACHECKSUMSWORKER_ABORTED)
+		ereport(LOG,
+				(errmsg("data checksums processing was aborted in database \"%s\"",
+						db->dbname)));
+
+	pgstat_report_activity(STATE_IDLE, NULL);
+
+	return DataChecksumsWorkerShmem->success;
+}
+
+/*
+ * launcher_exit
+ *
+ * Internal routine for cleaning up state when the launcher process exits. We
+ * need to clean up the abort flag to ensure that processing can be restarted
+ * again after it was previously aborted.
+ */
+static void
+launcher_exit(int code, Datum arg)
+{
+	if (launcher_running)
+	{
+		LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
+		launcher_running = false;
+		DataChecksumsWorkerShmem->launcher_running = false;
+		LWLockRelease(DataChecksumsWorkerLock);
+	}
+}
+
+/*
+ * launcher_cancel_handler
+ *
+ * Internal routine for reacting to SIGINT and flagging the worker to abort.
+ * The worker won't be interrupted immediately but will check for abort flag
+ * between each block in a relation.
+ */
+static void
+launcher_cancel_handler(SIGNAL_ARGS)
+{
+	int			save_errno = errno;
+
+	abort_requested = true;
+
+	/*
+	 * There is no sleeping in the main loop, the flag will be checked
+	 * periodically in ProcessSingleRelationFork. The worker does however
+	 * sleep when waiting for concurrent transactions to end so we still need
+	 * to set the latch.
+	 */
+	SetLatch(MyLatch);
+
+	errno = save_errno;
+}
+
+/*
+ * WaitForAllTransactionsToFinish
+ *		Blocks awaiting all current transactions to finish
+ *
+ * Returns when all transactions which are active at the call of the function
+ * have ended, or if the postmaster dies while waiting. If the postmaster dies
+ * the abort flag will be set to indicate that the caller of this shouldn't
+ * proceed.
+ *
+ * NB: this will return early, if aborted by SIGINT or if the target state
+ * is changed while we're running.
+ */
+static void
+WaitForAllTransactionsToFinish(void)
+{
+	TransactionId waitforxid;
+
+	LWLockAcquire(XidGenLock, LW_SHARED);
+	waitforxid = XidFromFullTransactionId(TransamVariables->nextXid);
+	LWLockRelease(XidGenLock);
+
+	while (TransactionIdPrecedes(GetOldestActiveTransactionId(), waitforxid))
+	{
+		char		activity[64];
+		int			rc;
+
+		/* Oldest running xid is older than us, so wait */
+		snprintf(activity,
+				 sizeof(activity),
+				 "Waiting for current transactions to finish (waiting for %u)",
+				 waitforxid);
+		pgstat_report_activity(STATE_RUNNING, activity);
+
+		/* Retry every 5 seconds */
+		ResetLatch(MyLatch);
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   5000,
+					   WAIT_EVENT_CHECKSUM_ENABLE_STARTCONDITION);
+
+		/*
+		 * If the postmaster died we won't be able to enable checksums
+		 * cluster-wide so abort and hope to continue when restarted.
+		 */
+		if (rc & WL_POSTMASTER_DEATH)
+			ereport(FATAL,
+					(errmsg("postmaster exited during data checksum processing"),
+					 errhint("Restart the database and restart data checksum processing by calling pg_enable_data_checksums().")));
+
+		LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);
+		if (DataChecksumsWorkerShmem->launch_enable_checksums != enabling_checksums)
+			abort_requested = true;
+		LWLockRelease(DataChecksumsWorkerLock);
+		if (abort_requested)
+			break;
+	}
+
+	pgstat_report_activity(STATE_IDLE, NULL);
+	return;
+}
+
+/*
+ * DataChecksumsWorkerLauncherMain
+ *
+ * Main function for launching dynamic background workers for processing data
+ * checksums in databases. This function has the bgworker management, with
+ * ProcessAllDatabases being responsible for looping over the databases and
+ * initiating processing.
+ */
+void
+DataChecksumsWorkerLauncherMain(Datum arg)
+{
+	on_shmem_exit(launcher_exit, 0);
+
+	ereport(DEBUG1,
+			errmsg("background worker \"datachecksumsworker\" launcher started"));
+
+	pqsignal(SIGTERM, die);
+	pqsignal(SIGINT, launcher_cancel_handler);
+
+	BackgroundWorkerUnblockSignals();
+
+	MyBackendType = B_DATACHECKSUMSWORKER_LAUNCHER;
+	init_ps_display(NULL);
+
+	LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
+
+	if (DataChecksumsWorkerShmem->launcher_running)
+	{
+		/* Launcher was already running, let it finish */
+		LWLockRelease(DataChecksumsWorkerLock);
+		return;
+	}
+
+	launcher_running = true;
+
+	/*
+	 * Initialize a connection to shared catalogs only.
+	 */
+	BackgroundWorkerInitializeConnectionByOid(InvalidOid, InvalidOid, 0);
+
+	/* Initialize backend status information */
+	pgstat_bestart();
+
+	enabling_checksums = DataChecksumsWorkerShmem->launch_enable_checksums;
+	DataChecksumsWorkerShmem->launcher_running = true;
+	DataChecksumsWorkerShmem->enabling_checksums = enabling_checksums;
+	DataChecksumsWorkerShmem->cost_delay = DataChecksumsWorkerShmem->launch_cost_delay;
+	DataChecksumsWorkerShmem->cost_limit = DataChecksumsWorkerShmem->launch_cost_limit;
+	DataChecksumsWorkerShmem->immediate_checkpoint = DataChecksumsWorkerShmem->launch_fast;
+	LWLockRelease(DataChecksumsWorkerLock);
+
+	/*
+	 * The target state can change while we are busy enabling/disabling
+	 * checksums, if the user calls pg_disable/enable_data_checksums() before
+	 * we are finished with the previous request. In that case, we will loop
+	 * back here, to process the new request.
+	 */
+again:
+
+	/*
+	 * If we're asked to enable checksums, we need to check if processing was
+	 * previously interrupted such that we should resume rather than start
+	 * from scratch.
+	 */
+	if (enabling_checksums)
+	{
+		/*
+		 * If we are asked to enable checksums in a cluster which already has
+		 * checksums enabled, exit immediately as there is nothing more to do.
+		 * Hold interrupts to make sure state doesn't change during checking.
+		 */
+		HOLD_INTERRUPTS();
+		if (DataChecksumsNeedVerify())
+		{
+			RESUME_INTERRUPTS();
+			goto done;
+		}
+		RESUME_INTERRUPTS();
+
+		/*
+		 * Initialize progress and indicate that we are waiting on the other
+		 * backends to clear the procsignalbarrier.
+		 */
+		pgstat_progress_start_command(PROGRESS_COMMAND_DATACHECKSUMS,
+									  InvalidOid);
+		pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
+									 PROGRESS_DATACHECKSUMS_PHASE_WAITING_BACKENDS);
+
+		/*
+		 * Set the state to inprogress-on and wait on the procsignal barrier.
+		 */
+		SetDataChecksumsOnInProgress();
+
+		if (!ProcessAllDatabases(DataChecksumsWorkerShmem->immediate_checkpoint))
+		{
+			/*
+			 * If the target state changed during processing then it's not a
+			 * failure, so restart processing instead.
+			 */
+			LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
+			if (DataChecksumsWorkerShmem->launch_enable_checksums != enabling_checksums)
+			{
+				LWLockRelease(DataChecksumsWorkerLock);
+				goto done;
+			}
+			LWLockRelease(DataChecksumsWorkerLock);
+			ereport(ERROR,
+					(errmsg("unable to enable data checksums in cluster")));
+		}
+
+		SetDataChecksumsOn();
+	}
+	else
+	{
+		SetDataChecksumsOff();
+	}
+
+done:
+
+	/*
+	 * All done. But before we exit, check if the target state was changed
+	 * while we were running. In that case we will have to start all over
+	 * again.
+	 */
+	LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
+	if (DataChecksumsWorkerShmem->launch_enable_checksums != enabling_checksums)
+	{
+		DataChecksumsWorkerShmem->enabling_checksums = DataChecksumsWorkerShmem->launch_enable_checksums;
+		enabling_checksums = DataChecksumsWorkerShmem->launch_enable_checksums;
+		DataChecksumsWorkerShmem->cost_delay = DataChecksumsWorkerShmem->launch_cost_delay;
+		DataChecksumsWorkerShmem->cost_limit = DataChecksumsWorkerShmem->launch_cost_limit;
+		LWLockRelease(DataChecksumsWorkerLock);
+		goto again;
+	}
+
+	/* Shut down progress reporting as we are done */
+	pgstat_progress_end_command();
+
+	launcher_running = false;
+	DataChecksumsWorkerShmem->launcher_running = false;
+	LWLockRelease(DataChecksumsWorkerLock);
+}
+
+/*
+ * ProcessAllDatabases
+ *		Compute the list of all databases and process checksums in each
+ *
+ * This will repeatedly generate a list of databases to process for enabling
+ * checksums. Until no new databases are found, this will loop around computing
+ * a new list and comparing it to the already seen ones.
+ *
+ * If immediate_checkpoint is set to true then a CHECKPOINT_IMMEDIATE will be
+ * issued. This is useful for testing but should be avoided in production use
+ * as it may affect cluster performance drastically.
+ */
+static bool
+ProcessAllDatabases(bool immediate_checkpoint)
+{
+	List	   *DatabaseList;
+	HTAB	   *ProcessedDatabases = NULL;
+	ListCell   *lc;
+	HASHCTL		hash_ctl;
+	bool		found_failed = false;
+	int			flags;
+
+	/* Initialize a hash tracking all processed databases */
+	memset(&hash_ctl, 0, sizeof(hash_ctl));
+	hash_ctl.keysize = sizeof(Oid);
+	hash_ctl.entrysize = sizeof(DataChecksumsWorkerResultEntry);
+	ProcessedDatabases = hash_create("Processed databases",
+									 64,
+									 &hash_ctl,
+									 HASH_ELEM | HASH_BLOBS);
+	/*
+	 * Set up so first run processes shared catalogs, but not once in every
+	 * db.
+	 */
+	DataChecksumsWorkerShmem->process_shared_catalogs = true;
+
+	/*
+	 * Get a list of all databases to process. This may include databases that
+	 * were created during our runtime.  Since a database can be created as a
+	 * copy of any other database (which may not have existed in our last
+	 * run), we have to repeat this loop until no new databases show up in the
+	 * list.
+	 */
+	DatabaseList = BuildDatabaseList();
+
+	/*
+	 * Update progress reporting with the total number of databases we need
+	 * to process. This number should not be changed during processing, the
+	 * columns for processed databases is instead increased such that it can
+	 * be compared against the total.
+	 */
+	pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_TOTAL_DB,
+								 list_length(DatabaseList));
+
+	while (true)
+	{
+		int			processed_databases = 0;
+
+		foreach(lc, DatabaseList)
+		{
+			DataChecksumsWorkerDatabase *db = (DataChecksumsWorkerDatabase *) lfirst(lc);
+			DataChecksumsWorkerResult result;
+			DataChecksumsWorkerResultEntry *entry;
+			bool		found;
+
+			/*
+			 * Indicate which database is being processed set the number of
+			 * relations to -1 to clear field from previous values. -1 will
+			 * translate to NULL in the progress view.
+			 */
+			pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_CUR_DB, db->dboid);
+			pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_TOTAL_REL, -1);
+
+			/*
+			 * Check if this database has been processed already, and if so
+			 * whether it should be retried or skipped.
+			 */
+			entry = (DataChecksumsWorkerResultEntry *) hash_search(ProcessedDatabases, &db->dboid,
+																   HASH_FIND, NULL);
+
+			if (entry)
+			{
+				if (entry->result == DATACHECKSUMSWORKER_RETRYDB)
+				{
+					/*
+					 * Limit the number of retries to avoid infinite looping
+					 * in case there simply won't be enough workers in the
+					 * cluster to finish this operation.
+					 */
+					if (entry->retries > DATACHECKSUMSWORKER_MAX_DB_RETRIES)
+						entry->result = DATACHECKSUMSWORKER_FAILED;
+				}
+
+				/* Skip if this database has been processed already */
+				if (entry->result != DATACHECKSUMSWORKER_RETRYDB)
+					continue;
+			}
+
+			result = ProcessDatabase(db);
+			processed_databases++;
+
+			if (result == DATACHECKSUMSWORKER_SUCCESSFUL)
+			{
+				/*
+				 * If one database has completed shared catalogs, we don't
+				 * have to process them again.
+				 */
+				if (DataChecksumsWorkerShmem->process_shared_catalogs)
+					DataChecksumsWorkerShmem->process_shared_catalogs = false;
+			}
+			else if (result == DATACHECKSUMSWORKER_ABORTED)
+			{
+				/* Abort flag set, so exit the whole process */
+				return false;
+			}
+
+			entry = hash_search(ProcessedDatabases, &db->dboid, HASH_ENTER, &found);
+			entry->dboid = db->dboid;
+			entry->result = result;
+			if (!found)
+				entry->retries = 0;
+			else
+				entry->retries++;
+		}
+
+		elog(DEBUG1,
+			 "%i databases processed for data checksum enabling, %s",
+			 processed_databases,
+			 (processed_databases ? "process with restart" : "process completed"));
+
+		FreeDatabaseList(DatabaseList);
+
+		/*
+		 * If no databases were processed in this run of the loop, we have now
+		 * finished all databases and no concurrently created ones can exist.
+		 */
+		if (processed_databases == 0)
+			break;
+
+		/*
+		 * Re-generate the list of databases for another pass. Since we wait
+		 * for all pre-existing transactions finish, this way we can be
+		 * certain that there are no databases left without checksums.
+		 */
+		WaitForAllTransactionsToFinish();
+		DatabaseList = BuildDatabaseList();
+	}
+
+	/*
+	 * ProcessedDatabases now has all databases and the results of their
+	 * processing. Failure to enable checksums for a database can be because
+	 * they actually failed for some reason, or because the database was
+	 * dropped between us getting the database list and trying to process it.
+	 * Get a fresh list of databases to detect the second case where the
+	 * database was dropped before we had started processing it. If a database
+	 * still exists, but enabling checksums failed then we fail the entire
+	 * checksumming process and exit with an error.
+	 */
+	WaitForAllTransactionsToFinish();
+	DatabaseList = BuildDatabaseList();
+
+	foreach(lc, DatabaseList)
+	{
+		DataChecksumsWorkerDatabase *db = (DataChecksumsWorkerDatabase *) lfirst(lc);
+		DataChecksumsWorkerResult *entry;
+		bool		found;
+
+		entry = hash_search(ProcessedDatabases, (void *) &db->dboid,
+							HASH_FIND, &found);
+
+		/*
+		 * We are only interested in the databases where the failed database
+		 * still exists.
+		 */
+		if (found && *entry == DATACHECKSUMSWORKER_FAILED)
+		{
+			ereport(WARNING,
+					(errmsg("failed to enable data checksums in \"%s\"",
+							db->dbname)));
+			found_failed = found;
+			continue;
+		}
+	}
+
+	FreeDatabaseList(DatabaseList);
+
+	if (found_failed)
+	{
+		/* Disable checksums on cluster, because we failed */
+		SetDataChecksumsOff();
+		ereport(ERROR,
+				(errmsg("data checksums failed to get enabled in all databases, aborting"),
+				 errhint("The server log might have more information on the cause of the error.")));
+	}
+
+	/*
+	 * Force a checkpoint to get everything out to disk. The use of immediate
+	 * checkpoints is for running tests, as they would otherwise not execute
+	 * in such a way that they can reliably be placed under timeout control.
+	 */
+	flags = CHECKPOINT_FORCE | CHECKPOINT_WAIT;
+	if (immediate_checkpoint)
+		flags = flags | CHECKPOINT_IMMEDIATE;
+	RequestCheckpoint(flags);
+
+	return true;
+}
+
+/*
+ * DataChecksumsWorkerShmemSize
+ *		Compute required space for datachecksumsworker-related shared memory
+ */
+Size
+DataChecksumsWorkerShmemSize(void)
+{
+	Size		size;
+
+	size = sizeof(DataChecksumsWorkerShmemStruct);
+	size = MAXALIGN(size);
+
+	return size;
+}
+
+/*
+ * DataChecksumsWorkerShmemInit
+ *		Allocate and initialize datachecksumsworker-related shared memory
+ */
+void
+DataChecksumsWorkerShmemInit(void)
+{
+	bool		found;
+
+	DataChecksumsWorkerShmem = (DataChecksumsWorkerShmemStruct *)
+		ShmemInitStruct("DataChecksumsWorker Data",
+						DataChecksumsWorkerShmemSize(),
+						&found);
+
+	MemSet(DataChecksumsWorkerShmem, 0, DataChecksumsWorkerShmemSize());
+
+	/*
+	 * Even if this is a redundant assignment, we want to be explicit about
+	 * our intent for readability, since we want to be able to query this
+	 * state in case of restartability.
+	 */
+	DataChecksumsWorkerShmem->launch_enable_checksums = false;
+	DataChecksumsWorkerShmem->launcher_running = false;
+	DataChecksumsWorkerShmem->launch_fast = false;
+}
+
+/*
+ * BuildDatabaseList
+ *		Compile a list of all currently available databases in the cluster
+ *
+ * This creates the list of databases for the datachecksumsworker workers to
+ * add checksums to. If the caller wants to ensure that no concurrently
+ * running CREATE DATABASE calls exist, this needs to be preceded by a call
+ * to WaitForAllTransactionsToFinish().
+ */
+static List *
+BuildDatabaseList(void)
+{
+	List	   *DatabaseList = NIL;
+	Relation	rel;
+	TableScanDesc scan;
+	HeapTuple	tup;
+	MemoryContext ctx = CurrentMemoryContext;
+	MemoryContext oldctx;
+
+	StartTransactionCommand();
+
+	rel = table_open(DatabaseRelationId, AccessShareLock);
+	scan = table_beginscan_catalog(rel, 0, NULL);
+
+	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
+	{
+		Form_pg_database pgdb = (Form_pg_database) GETSTRUCT(tup);
+		DataChecksumsWorkerDatabase *db;
+
+		oldctx = MemoryContextSwitchTo(ctx);
+
+		db = (DataChecksumsWorkerDatabase *) palloc0(sizeof(DataChecksumsWorkerDatabase));
+
+		db->dboid = pgdb->oid;
+		db->dbname = pstrdup(NameStr(pgdb->datname));
+
+		DatabaseList = lappend(DatabaseList, db);
+
+		MemoryContextSwitchTo(oldctx);
+	}
+
+	table_endscan(scan);
+	table_close(rel, AccessShareLock);
+
+	CommitTransactionCommand();
+
+	return DatabaseList;
+}
+
+static void
+FreeDatabaseList(List *dblist)
+{
+	ListCell   *lc;
+
+	if (!dblist)
+		return;
+
+	foreach(lc, dblist)
+	{
+		DataChecksumsWorkerDatabase *db = lfirst(lc);
+
+		if (db->dbname != NULL)
+			pfree(db->dbname);
+	}
+
+	list_free_deep(dblist);
+}
+
+/*
+ * BuildRelationList
+ *		Compile a list of relations in the database
+ *
+ * Returns a list of OIDs for the request relation types. If temp_relations
+ * is True then only temporary relations are returned. If temp_relations is
+ * False then non-temporary relations which have data checksums are returned.
+ * If include_shared is True then shared relations are included as well in a
+ * non-temporary list. include_shared has no relevance when building a list of
+ * temporary relations.
+ */
+static List *
+BuildRelationList(bool temp_relations, bool include_shared)
+{
+	List	   *RelationList = NIL;
+	Relation	rel;
+	TableScanDesc scan;
+	HeapTuple	tup;
+	MemoryContext ctx = CurrentMemoryContext;
+	MemoryContext oldctx;
+
+	StartTransactionCommand();
+
+	rel = table_open(RelationRelationId, AccessShareLock);
+	scan = table_beginscan_catalog(rel, 0, NULL);
+
+	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
+	{
+		Form_pg_class pgc = (Form_pg_class) GETSTRUCT(tup);
+
+		/*
+		 * Only include temporary relations when asked for a temp relation
+		 * list.
+		 */
+		if (pgc->relpersistence == RELPERSISTENCE_TEMP)
+		{
+			if (!temp_relations)
+				continue;
+		}
+		else
+		{
+			/*
+			 * If we are only interested in temp relations then continue
+			 * immediately as the current relation isn't a temp relation.
+			 */
+			if (temp_relations)
+				continue;
+
+			if (!RELKIND_HAS_STORAGE(pgc->relkind))
+				continue;
+
+			if (pgc->relisshared && !include_shared)
+				continue;
+		}
+
+		oldctx = MemoryContextSwitchTo(ctx);
+		RelationList = lappend_oid(RelationList, pgc->oid);
+		MemoryContextSwitchTo(oldctx);
+	}
+
+	table_endscan(scan);
+	table_close(rel, AccessShareLock);
+
+	CommitTransactionCommand();
+
+	return RelationList;
+}
+
+/*
+ * DataChecksumsWorkerMain
+ *
+ * Main function for enabling checksums in a single database, This is the
+ * function set as the bgw_function_name in the dynamic background worker
+ * process initiated for each database by the worker launcher. After enabling
+ * data checksums in each applicable relation in the database, it will wait for
+ * all temporary relations that were present when the function started to
+ * disappear before returning. This is required since we cannot rewrite
+ * existing temporary relations with data checksums.
+ */
+void
+DataChecksumsWorkerMain(Datum arg)
+{
+	Oid			dboid = DatumGetObjectId(arg);
+	List	   *RelationList = NIL;
+	List	   *InitialTempTableList = NIL;
+	ListCell   *lc;
+	BufferAccessStrategy strategy;
+	bool		aborted = false;
+
+	enabling_checksums = true;
+
+	pqsignal(SIGTERM, die);
+
+	BackgroundWorkerUnblockSignals();
+
+	MyBackendType = B_DATACHECKSUMSWORKER_WORKER;
+	init_ps_display(NULL);
+
+	pgstat_progress_start_command(PROGRESS_COMMAND_DATACHECKSUMS, dboid);
+
+	BackgroundWorkerInitializeConnectionByOid(dboid, InvalidOid,
+											  BGWORKER_BYPASS_ALLOWCONN);
+
+	/*
+	 * Get a list of all temp tables present as we start in this database. We
+	 * need to wait until they are all gone until we are done, since we cannot
+	 * access these relations and modify them.
+	 */
+	InitialTempTableList = BuildRelationList(true, false);
+
+	/*
+	 * Enable vacuum cost delay, if any.
+	 */
+	Assert(DataChecksumsWorkerShmem->enabling_checksums);
+	VacuumCostDelay = DataChecksumsWorkerShmem->cost_delay;
+	VacuumCostLimit = DataChecksumsWorkerShmem->cost_limit;
+	VacuumCostActive = (VacuumCostDelay > 0);
+	VacuumCostBalance = 0;
+	VacuumCostPageHit = 0;
+	VacuumCostPageMiss = 0;
+	VacuumCostPageDirty = 0;
+
+	/*
+	 * Create and set the vacuum strategy as our buffer strategy.
+	 */
+	strategy = GetAccessStrategy(BAS_VACUUM);
+
+	RelationList = BuildRelationList(false,
+									 DataChecksumsWorkerShmem->process_shared_catalogs);
+	pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_TOTAL_REL,
+								 list_length(RelationList));
+	foreach(lc, RelationList)
+	{
+		Oid			reloid = lfirst_oid(lc);
+
+		if (!ProcessSingleRelationByOid(reloid, strategy))
+		{
+			aborted = true;
+			break;
+		}
+	}
+	list_free(RelationList);
+
+	if (aborted)
+	{
+		DataChecksumsWorkerShmem->success = DATACHECKSUMSWORKER_ABORTED;
+		ereport(DEBUG1,
+				(errmsg("data checksum processing aborted in database OID %u",
+						dboid)));
+		return;
+	}
+
+	/*
+	 * Wait for all temp tables that existed when we started to go away. This
+	 * is necessary since we cannot "reach" them to enable checksums. Any temp
+	 * tables created after we started will already have checksums in them
+	 * (due to the "inprogress-on" state), so no need to wait for those.
+	 */
+	for (;;)
+	{
+		List	   *CurrentTempTables;
+		ListCell   *lc;
+		int			numleft;
+		char		activity[64];
+		const int index[] = {
+			PROGRESS_DATACHECKSUMS_PHASE,
+			PROGRESS_DATACHECKSUMS_TOTAL_REL
+		};
+		int64 vals[2];
+
+		CurrentTempTables = BuildRelationList(true, false);
+		numleft = 0;
+		foreach(lc, InitialTempTableList)
+		{
+			if (list_member_oid(CurrentTempTables, lfirst_oid(lc)))
+				numleft++;
+		}
+		list_free(CurrentTempTables);
+
+		if (numleft == 0)
+			break;
+
+		/*
+		 * At least one temp table is left to wait for, indicate in pgstat
+		 * activity and progress reporting.
+		 */
+		snprintf(activity,
+				 sizeof(activity),
+				 "Waiting for %d temp tables to be removed", numleft);
+		pgstat_report_activity(STATE_RUNNING, activity);
+		vals[0] = PROGRESS_DATACHECKSUMS_PHASE_WAITING_TEMPREL;
+		vals[1] = numleft;
+		pgstat_progress_update_multi_param(2, index, vals);
+
+		/* Retry every 5 seconds */
+		ResetLatch(MyLatch);
+		(void) WaitLatch(MyLatch,
+						 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+						 5000,
+						 WAIT_EVENT_CHECKSUM_ENABLE_FINISHCONDITION);
+
+		LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
+		aborted = DataChecksumsWorkerShmem->launch_enable_checksums != enabling_checksums;
+		LWLockRelease(DataChecksumsWorkerLock);
+
+		if (aborted || abort_requested)
+		{
+			DataChecksumsWorkerShmem->success = DATACHECKSUMSWORKER_ABORTED;
+			ereport(DEBUG1,
+					(errmsg("data checksum processing aborted in database OID %u",
+							dboid)));
+			return;
+		}
+	}
+
+	list_free(InitialTempTableList);
+
+	DataChecksumsWorkerShmem->success = DATACHECKSUMSWORKER_SUCCESSFUL;
+
+	pgstat_progress_end_command();
+}
diff --git a/src/backend/postmaster/meson.build b/src/backend/postmaster/meson.build
index 0ea4bbe084..3aacb2e0e7 100644
--- a/src/backend/postmaster/meson.build
+++ b/src/backend/postmaster/meson.build
@@ -6,6 +6,7 @@ backend_sources += files(
   'bgworker.c',
   'bgwriter.c',
   'checkpointer.c',
+  'datachecksumsworker.c',
   'fork_process.c',
   'interrupt.c',
   'launch_backend.c',
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index d687ceee33..6c38a57a20 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -186,6 +186,7 @@ xlog_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		case XLOG_FPW_CHANGE:
 		case XLOG_FPI_FOR_HINT:
 		case XLOG_FPI:
+		case XLOG_CHECKSUMS:
 		case XLOG_OVERWRITE_CONTRECORD:
 		case XLOG_CHECKPOINT_REDO:
 			break;
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 4852044300..93e198f19e 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -1516,7 +1516,9 @@ WaitReadBuffers(ReadBuffersOperation *operation)
 				bufBlock = BufHdrGetBlock(bufHdr);
 			}
 
-			/* check for garbage data */
+			/*
+			 * Check for garbage data.
+			 */
 			if (!PageIsVerifiedExtended((Page) bufBlock, io_first_block + j,
 										PIV_LOG_WARNING | PIV_REPORT_STAT))
 			{
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 10fc18f252..b9d98d0ada 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -31,6 +31,7 @@
 #include "postmaster/autovacuum.h"
 #include "postmaster/bgworker_internals.h"
 #include "postmaster/bgwriter.h"
+#include "postmaster/datachecksumsworker.h"
 #include "postmaster/postmaster.h"
 #include "postmaster/walsummarizer.h"
 #include "replication/logicallauncher.h"
@@ -152,6 +153,7 @@ CalculateShmemSize(int *num_semaphores)
 	size = add_size(size, InjectionPointShmemSize());
 	size = add_size(size, SlotSyncShmemSize());
 	size = add_size(size, WaitLSNShmemSize());
+	size = add_size(size, DataChecksumsWorkerShmemSize());
 
 	/* include additional requested shmem from preload libraries */
 	size = add_size(size, total_addin_request);
@@ -334,6 +336,7 @@ CreateOrAttachShmemStructs(void)
 	PgArchShmemInit();
 	ApplyLauncherShmemInit();
 	SlotSyncShmemInit();
+	DataChecksumsWorkerShmemInit();
 
 	/*
 	 * Set up other modules that need some shared memory space
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index 87027f27eb..e51cfb3a98 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -18,6 +18,7 @@
 #include <unistd.h>
 
 #include "access/parallel.h"
+#include "access/xlog.h"
 #include "commands/async.h"
 #include "miscadmin.h"
 #include "pgstat.h"
@@ -573,6 +574,18 @@ ProcessProcSignalBarrier(void)
 					case PROCSIGNAL_BARRIER_SMGRRELEASE:
 						processed = ProcessBarrierSmgrRelease();
 						break;
+					case PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON:
+						processed = AbsorbChecksumsOnInProgressBarrier();
+						break;
+					case PROCSIGNAL_BARRIER_CHECKSUM_ON:
+						processed = AbsorbChecksumsOnBarrier();
+						break;
+					case PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF:
+						processed = AbsorbChecksumsOffInProgressBarrier();
+						break;
+					case PROCSIGNAL_BARRIER_CHECKSUM_OFF:
+						processed = AbsorbChecksumsOffBarrier();
+						break;
 				}
 
 				/*
diff --git a/src/backend/storage/page/README b/src/backend/storage/page/README
index e30d7ac59a..73c36a6390 100644
--- a/src/backend/storage/page/README
+++ b/src/backend/storage/page/README
@@ -10,7 +10,9 @@ http://www.cs.toronto.edu/~bianca/papers/sigmetrics09.pdf, discussed
 2010/12/22 on -hackers list.
 
 Current implementation requires this be enabled system-wide at initdb time, or
-by using the pg_checksums tool on an offline cluster.
+by using the pg_checksums tool on an offline cluster.  Checksums can also be
+enabled at runtime using pg_enable_data_checksums(), and disabled by using
+pg_disable_data_checksums().
 
 The checksum is not valid at all times on a data page!!
 The checksum is valid when the page leaves the shared pool and is checked
diff --git a/src/backend/storage/page/bufpage.c b/src/backend/storage/page/bufpage.c
index be6f1f62d2..112c05ee7e 100644
--- a/src/backend/storage/page/bufpage.c
+++ b/src/backend/storage/page/bufpage.c
@@ -100,7 +100,7 @@ PageIsVerifiedExtended(Page page, BlockNumber blkno, int flags)
 	 */
 	if (!PageIsNew(page))
 	{
-		if (DataChecksumsEnabled())
+		if (DataChecksumsNeedVerify())
 		{
 			checksum = pg_checksum_page((char *) page, blkno);
 
@@ -1512,7 +1512,7 @@ PageSetChecksumCopy(Page page, BlockNumber blkno)
 	static char *pageCopy = NULL;
 
 	/* If we don't need a checksum, just return the passed-in data */
-	if (PageIsNew(page) || !DataChecksumsEnabled())
+	if (PageIsNew(page) || !DataChecksumsNeedWrite())
 		return (char *) page;
 
 	/*
@@ -1542,7 +1542,7 @@ void
 PageSetChecksumInplace(Page page, BlockNumber blkno)
 {
 	/* If we don't need a checksum, just return */
-	if (PageIsNew(page) || !DataChecksumsEnabled())
+	if (PageIsNew(page) || !DataChecksumsNeedWrite())
 		return;
 
 	((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blkno);
diff --git a/src/backend/storage/smgr/bulk_write.c b/src/backend/storage/smgr/bulk_write.c
index 1a5f3ce96e..1981ed768d 100644
--- a/src/backend/storage/smgr/bulk_write.c
+++ b/src/backend/storage/smgr/bulk_write.c
@@ -36,6 +36,7 @@
 
 #include "access/xloginsert.h"
 #include "access/xlogrecord.h"
+#include "miscadmin.h"
 #include "storage/bufmgr.h"
 #include "storage/bufpage.h"
 #include "storage/bulk_write.h"
@@ -303,6 +304,7 @@ smgr_bulk_flush(BulkWriteState *bulkstate)
 		}
 		else
 			smgrwrite(bulkstate->smgr, bulkstate->forknum, blkno, page, true);
+
 		pfree(page);
 	}
 
diff --git a/src/backend/utils/activity/pgstat.c b/src/backend/utils/activity/pgstat.c
index d1768a89f6..b612d9d0fc 100644
--- a/src/backend/utils/activity/pgstat.c
+++ b/src/backend/utils/activity/pgstat.c
@@ -353,7 +353,6 @@ static const PgStat_KindInfo pgstat_kind_builtin_infos[PGSTAT_KIND_BUILTIN_SIZE]
 		.reset_timestamp_cb = pgstat_subscription_reset_timestamp_cb,
 	},
 
-
 	/* stats for fixed-numbered (mostly 1) objects */
 
 	[PGSTAT_KIND_ARCHIVER] = {
diff --git a/src/backend/utils/activity/pgstat_io.c b/src/backend/utils/activity/pgstat_io.c
index cc2ffc78aa..2b5671e2ca 100644
--- a/src/backend/utils/activity/pgstat_io.c
+++ b/src/backend/utils/activity/pgstat_io.c
@@ -359,6 +359,8 @@ pgstat_tracks_io_bktype(BackendType bktype)
 		case B_WAL_SUMMARIZER:
 			return false;
 
+		case B_DATACHECKSUMSWORKER_LAUNCHER:
+		case B_DATACHECKSUMSWORKER_WORKER:
 		case B_AUTOVAC_LAUNCHER:
 		case B_AUTOVAC_WORKER:
 		case B_BACKEND:
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index 8efb4044d6..4ed6ec157a 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -115,6 +115,8 @@ CHECKPOINT_DELAY_COMPLETE	"Waiting for a backend that blocks a checkpoint from c
 CHECKPOINT_DELAY_START	"Waiting for a backend that blocks a checkpoint from starting."
 CHECKPOINT_DONE	"Waiting for a checkpoint to complete."
 CHECKPOINT_START	"Waiting for a checkpoint to start."
+CHECKSUM_ENABLE_STARTCONDITION	"Waiting for data checksums enabling to start."
+CHECKSUM_ENABLE_FINISHCONDITION	"Waiting for data checksums to be enabled."
 EXECUTE_GATHER	"Waiting for activity from a child process while executing a <literal>Gather</literal> plan node."
 HASH_BATCH_ALLOCATE	"Waiting for an elected Parallel Hash participant to allocate a hash table."
 HASH_BATCH_ELECT	"Waiting to elect a Parallel Hash participant to allocate a hash table."
@@ -347,6 +349,7 @@ DSMRegistry	"Waiting to read or update the dynamic shared memory registry."
 InjectionPoint	"Waiting to read or update information related to injection points."
 SerialControl	"Waiting to read or update shared <filename>pg_serial</filename> state."
 WaitLSN	"Waiting to read or update shared Wait-for-LSN state."
+DataChecksumsWorker	"Waiting for data checksumsworker."
 
 #
 # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE)
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 17b0fc02ef..bcc6da7996 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -246,6 +246,8 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS)
 		cmdtype = PROGRESS_COMMAND_BASEBACKUP;
 	else if (pg_strcasecmp(cmd, "COPY") == 0)
 		cmdtype = PROGRESS_COMMAND_COPY;
+	else if (pg_strcasecmp(cmd, "DATACHECKSUMS") == 0)
+		cmdtype = PROGRESS_COMMAND_DATACHECKSUMS;
 	else
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
@@ -1115,9 +1117,6 @@ pg_stat_get_db_checksum_failures(PG_FUNCTION_ARGS)
 	int64		result;
 	PgStat_StatDBEntry *dbentry;
 
-	if (!DataChecksumsEnabled())
-		PG_RETURN_NULL();
-
 	if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
 		result = 0;
 	else
@@ -1133,9 +1132,6 @@ pg_stat_get_db_checksum_last_failure(PG_FUNCTION_ARGS)
 	TimestampTz result;
 	PgStat_StatDBEntry *dbentry;
 
-	if (!DataChecksumsEnabled())
-		PG_RETURN_NULL();
-
 	if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
 		result = 0;
 	else
diff --git a/src/backend/utils/init/miscinit.c b/src/backend/utils/init/miscinit.c
index 537d92c0cf..6571c187cc 100644
--- a/src/backend/utils/init/miscinit.c
+++ b/src/backend/utils/init/miscinit.c
@@ -290,6 +290,12 @@ GetBackendTypeDesc(BackendType backendType)
 		case B_CHECKPOINTER:
 			backendDesc = "checkpointer";
 			break;
+		case B_DATACHECKSUMSWORKER_LAUNCHER:
+			backendDesc = "datachecksumsworker launcher";
+			break;
+		case B_DATACHECKSUMSWORKER_WORKER:
+			backendDesc = "datachecksumsworker worker";
+			break;
 		case B_LOGGER:
 			backendDesc = "logger";
 			break;
@@ -841,7 +847,8 @@ InitializeSessionUserIdStandalone(void)
 	 * workers, in slot sync worker and in background workers.
 	 */
 	Assert(!IsUnderPostmaster || AmAutoVacuumWorkerProcess() ||
-		   AmLogicalSlotSyncWorkerProcess() || AmBackgroundWorkerProcess());
+		   AmLogicalSlotSyncWorkerProcess() || AmBackgroundWorkerProcess() ||
+		   AmDataChecksumsWorkerProcess());
 
 	/* call only once */
 	Assert(!OidIsValid(AuthenticatedUserId));
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index a024b1151d..513b163ed7 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -720,6 +720,11 @@ InitPostgres(const char *in_dbname, Oid dboid,
 	 */
 	SharedInvalBackendInit(false);
 
+	/*
+	 * Set up backend local cache of Controldata values.
+	 */
+	InitLocalControldata();
+
 	ProcSignalInit(MyCancelKeyValid, MyCancelKey);
 
 	/*
@@ -864,7 +869,7 @@ InitPostgres(const char *in_dbname, Oid dboid,
 					 errhint("You should immediately run CREATE USER \"%s\" SUPERUSER;.",
 							 username != NULL ? username : "postgres")));
 	}
-	else if (AmBackgroundWorkerProcess())
+	else if (AmBackgroundWorkerProcess() || AmDataChecksumsWorkerProcess())
 	{
 		if (username == NULL && !OidIsValid(useroid))
 		{
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 686309db58..e56e43f701 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -474,6 +474,14 @@ static const struct config_enum_entry wal_compression_options[] = {
 	{NULL, 0, false}
 };
 
+static const struct config_enum_entry data_checksums_options[] = {
+	{"on", DATA_CHECKSUMS_ON, true},
+	{"off", DATA_CHECKSUMS_OFF, true},
+	{"inprogress-on", DATA_CHECKSUMS_INPROGRESS_ON, true},
+	{"inprogress-off", DATA_CHECKSUMS_INPROGRESS_OFF, true},
+	{NULL, 0, false}
+};
+
 /*
  * Options for enum values stored in other modules
  */
@@ -599,7 +607,7 @@ static int	shared_memory_size_mb;
 static int	shared_memory_size_in_huge_pages;
 static int	wal_block_size;
 static int	num_os_semaphores;
-static bool data_checksums;
+static int data_checksums;
 static bool integer_datetimes;
 
 #ifdef USE_ASSERT_CHECKING
@@ -1919,17 +1927,6 @@ struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 
-	{
-		{"data_checksums", PGC_INTERNAL, PRESET_OPTIONS,
-			gettext_noop("Shows whether data checksums are turned on for this cluster."),
-			NULL,
-			GUC_NOT_IN_SAMPLE | GUC_DISALLOW_IN_FILE | GUC_RUNTIME_COMPUTED
-		},
-		&data_checksums,
-		false,
-		NULL, NULL, NULL
-	},
-
 	{
 		{"syslog_sequence_numbers", PGC_SIGHUP, LOGGING_WHERE,
 			gettext_noop("Add sequence number to syslog messages to avoid duplicate suppression."),
@@ -5196,6 +5193,17 @@ struct config_enum ConfigureNamesEnum[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"data_checksums", PGC_INTERNAL, PRESET_OPTIONS,
+			gettext_noop("Shows whether data checksums are turned on for this cluster."),
+			NULL,
+			GUC_NOT_IN_SAMPLE | GUC_DISALLOW_IN_FILE
+		},
+		&data_checksums,
+		DATA_CHECKSUMS_OFF, data_checksums_options,
+		NULL, NULL, show_data_checksums
+	},
+
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, 0, NULL, NULL, NULL, NULL
diff --git a/src/bin/pg_checksums/pg_checksums.c b/src/bin/pg_checksums/pg_checksums.c
index f5f7ff1045..3bda3adb04 100644
--- a/src/bin/pg_checksums/pg_checksums.c
+++ b/src/bin/pg_checksums/pg_checksums.c
@@ -577,7 +577,7 @@ main(int argc, char *argv[])
 		mode == PG_MODE_DISABLE)
 		pg_fatal("data checksums are already disabled in cluster");
 
-	if (ControlFile->data_checksum_version > 0 &&
+	if (ControlFile->data_checksum_version == DATA_CHECKSUMS_ON &&
 		mode == PG_MODE_ENABLE)
 		pg_fatal("data checksums are already enabled in cluster");
 
diff --git a/src/bin/pg_upgrade/controldata.c b/src/bin/pg_upgrade/controldata.c
index 854c6887a2..527c807f1c 100644
--- a/src/bin/pg_upgrade/controldata.c
+++ b/src/bin/pg_upgrade/controldata.c
@@ -693,6 +693,15 @@ check_control_data(ControlData *oldctrl,
 	 * check_for_isn_and_int8_passing_mismatch().
 	 */
 
+	/*
+	 * If data checksums have been turned on in the old cluster, but the
+	 * datachecksumsworker have yet to finish, then disallow the upgrade. The
+	 * user should either let the process finish, or turn off checksums,
+	 * before retrying.
+	 */
+	if (oldctrl->data_checksum_version == 2)
+		pg_fatal("checksums are being enabled in the old cluster");
+
 	/*
 	 * We might eventually allow upgrades from checksum to no-checksum
 	 * clusters.
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 36f6e4e4b4..0e2b53c8d2 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -117,7 +117,7 @@ extern PGDLLIMPORT int wal_level;
  * of the bits make it to disk, but the checksum wouldn't match.  Also WAL-log
  * them if forced by wal_log_hints=on.
  */
-#define XLogHintBitIsNeeded() (DataChecksumsEnabled() || wal_log_hints)
+#define XLogHintBitIsNeeded() (wal_log_hints || DataChecksumsNeedWrite())
 
 /* Do we need to WAL-log information required only for Hot Standby and logical replication? */
 #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA)
@@ -229,7 +229,19 @@ extern XLogRecPtr GetXLogWriteRecPtr(void);
 
 extern uint64 GetSystemIdentifier(void);
 extern char *GetMockAuthenticationNonce(void);
-extern bool DataChecksumsEnabled(void);
+extern bool DataChecksumsNeedWrite(void);
+extern bool DataChecksumsNeedVerify(void);
+extern bool DataChecksumsOnInProgress(void);
+extern bool DataChecksumsOffInProgress(void);
+extern void SetDataChecksumsOnInProgress(void);
+extern void SetDataChecksumsOn(void);
+extern void SetDataChecksumsOff(void);
+extern bool AbsorbChecksumsOnInProgressBarrier(void);
+extern bool AbsorbChecksumsOffInProgressBarrier(void);
+extern bool AbsorbChecksumsOnBarrier(void);
+extern bool AbsorbChecksumsOffBarrier(void);
+extern const char *show_data_checksums(void);
+extern void InitLocalControldata(void);
 extern XLogRecPtr GetFakeLSNForUnloggedRel(void);
 extern Size XLOGShmemSize(void);
 extern void XLOGShmemInit(void);
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index 5ef244bcdb..21edc8b737 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -25,6 +25,7 @@
 #include "lib/stringinfo.h"
 #include "pgtime.h"
 #include "storage/block.h"
+#include "storage/checksum.h"
 #include "storage/relfilelocator.h"
 
 
@@ -289,6 +290,12 @@ typedef struct xl_restore_point
 	char		rp_name[MAXFNAMELEN];
 } xl_restore_point;
 
+/* Information logged when data checksum level is changed */
+typedef struct xl_checksum_state
+{
+	ChecksumType new_checksumtype;
+}			xl_checksum_state;
+
 /* Overwrite of prior contrecord */
 typedef struct xl_overwrite_contrecord
 {
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index e80ff8e414..17da375aa6 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -80,6 +80,7 @@ typedef struct CheckPoint
 /* 0xC0 is used in Postgres 9.5-11 */
 #define XLOG_OVERWRITE_CONTRECORD		0xD0
 #define XLOG_CHECKPOINT_REDO			0xE0
+#define XLOG_CHECKSUMS					0xF0
 
 
 /*
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 05fcbf7515..c2703308eb 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -12140,6 +12140,23 @@
   proname => 'jsonb_subscript_handler', prorettype => 'internal',
   proargtypes => 'internal', prosrc => 'jsonb_subscript_handler' },
 
+# data checksum management functions
+{ oid => '9258',
+  descr => 'disable data checksums',
+  proname => 'pg_disable_data_checksums', provolatile => 'v', prorettype => 'void',
+  proparallel => 'r',
+  proargtypes => '',
+  prosrc => 'disable_data_checksums' },
+
+{ oid => '9257',
+  descr => 'enable data checksums',
+  proname => 'pg_enable_data_checksums', provolatile => 'v', prorettype => 'void',
+  proparallel => 'r',
+  proargtypes => 'int4 int4 bool', proallargtypes => '{int4,int4,bool}',
+  proargmodes => '{i,i,i}',
+  proargnames => '{cost_delay,cost_limit,fast}',
+  prosrc => 'enable_data_checksums' },
+
 # collation management functions
 { oid => '3445', descr => 'import collations from operating system',
   proname => 'pg_import_system_collations', procost => '100',
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 5616d64523..8e3c180392 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -155,4 +155,22 @@
 #define PROGRESS_COPY_TYPE_PIPE 3
 #define PROGRESS_COPY_TYPE_CALLBACK 4
 
+/* Progress parameters for PROGRESS_DATACHECKSUMS */
+#define PROGRESS_DATACHECKSUMS_PHASE 0
+#define PROGRESS_DATACHECKSUMS_TOTAL_DB 1
+#define PROGRESS_DATACHECKSUMS_TOTAL_REL 2
+#define PROGRESS_DATACHECKSUMS_PROCESSED_DB 3
+#define PROGRESS_DATACHECKSUMS_PROCESSED_REL 4
+#define PROGRESS_DATACHECKSUMS_CUR_DB 5
+#define PROGRESS_DATACHECKSUMS_CUR_REL 6
+#define PROGRESS_DATACHECKSUMS_CUR_REL_TOTAL_BLOCKS 7
+#define PROGRESS_DATACHECKSUMS_CUR_REL_PROCESSED_BLOCKS 8
+
+/* Phases of datachecksumsworker operation */
+#define PROGRESS_DATACHECKSUMS_PHASE_ENABLING 0
+#define PROGRESS_DATACHECKSUMS_PHASE_DISABLING 1
+#define PROGRESS_DATACHECKSUMS_PHASE_WAITING_BACKENDS 2
+#define PROGRESS_DATACHECKSUMS_PHASE_WAITING_TEMPREL 3
+#define PROGRESS_DATACHECKSUMS_DONE 4
+
 #endif
diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h
index e26d108a47..55e507683a 100644
--- a/src/include/miscadmin.h
+++ b/src/include/miscadmin.h
@@ -357,6 +357,9 @@ typedef enum BackendType
 	B_WAL_SUMMARIZER,
 	B_WAL_WRITER,
 
+	B_DATACHECKSUMSWORKER_LAUNCHER,
+	B_DATACHECKSUMSWORKER_WORKER,
+
 	/*
 	 * Logger is not connected to shared memory and does not have a PGPROC
 	 * entry.
@@ -380,6 +383,7 @@ extern PGDLLIMPORT BackendType MyBackendType;
 #define AmWalReceiverProcess()		(MyBackendType == B_WAL_RECEIVER)
 #define AmWalSummarizerProcess()	(MyBackendType == B_WAL_SUMMARIZER)
 #define AmWalWriterProcess()		(MyBackendType == B_WAL_WRITER)
+#define AmDataChecksumsWorkerProcess()	(MyBackendType == B_DATACHECKSUMSWORKER_LAUNCHER || MyBackendType == B_DATACHECKSUMSWORKER_WORKER)
 
 extern const char *GetBackendTypeDesc(BackendType backendType);
 
diff --git a/src/include/postmaster/datachecksumsworker.h b/src/include/postmaster/datachecksumsworker.h
new file mode 100644
index 0000000000..59c9000d64
--- /dev/null
+++ b/src/include/postmaster/datachecksumsworker.h
@@ -0,0 +1,31 @@
+/*-------------------------------------------------------------------------
+ *
+ * datachecksumsworker.h
+ *	  header file for checksum helper background worker
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/postmaster/datachecksumsworker.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef DATACHECKSUMSWORKER_H
+#define DATACHECKSUMSWORKER_H
+
+/* Shared memory */
+extern Size DataChecksumsWorkerShmemSize(void);
+extern void DataChecksumsWorkerShmemInit(void);
+
+/* Start the background processes for enabling or disabling checksums */
+void		StartDataChecksumsWorkerLauncher(bool enable_checksums,
+											 int cost_delay,
+											 int cost_limit,
+											 bool fast);
+
+/* Background worker entrypoints */
+void		DataChecksumsWorkerLauncherMain(Datum arg);
+void		DataChecksumsWorkerMain(Datum arg);
+
+#endif							/* DATACHECKSUMSWORKER_H */
diff --git a/src/include/storage/bufpage.h b/src/include/storage/bufpage.h
index 6222d46e53..9ff4127e58 100644
--- a/src/include/storage/bufpage.h
+++ b/src/include/storage/bufpage.h
@@ -204,7 +204,17 @@ typedef PageHeaderData *PageHeader;
  * handling pages.
  */
 #define PG_PAGE_LAYOUT_VERSION		4
+
+/*
+ * Checksum version 0 is used for when data checksums are disabled.
+ * PG_DATA_CHECKSUM_VERSION defines that data checksums are enabled in the
+ * cluster and PG_DATA_CHECKSUM_INPROGRESS_{ON|OFF}_VERSION defines that data
+ * checksums are either currently being enabled or disabled.
+ */
 #define PG_DATA_CHECKSUM_VERSION	1
+#define PG_DATA_CHECKSUM_INPROGRESS_ON_VERSION	2
+#define PG_DATA_CHECKSUM_INPROGRESS_OFF_VERSION	3
+
 
 /* ----------------------------------------------------------------
  *						page support functions
diff --git a/src/include/storage/checksum.h b/src/include/storage/checksum.h
index 08e9d598ce..e3e6ec3d4a 100644
--- a/src/include/storage/checksum.h
+++ b/src/include/storage/checksum.h
@@ -15,6 +15,14 @@
 
 #include "storage/block.h"
 
+typedef enum ChecksumType
+{
+	DATA_CHECKSUMS_OFF = 0,
+	DATA_CHECKSUMS_ON,
+	DATA_CHECKSUMS_INPROGRESS_ON,
+	DATA_CHECKSUMS_INPROGRESS_OFF
+} ChecksumType;
+
 /*
  * Compute the checksum for a Postgres page.  The page must be aligned on a
  * 4-byte boundary.
diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h
index 88dc79b2bd..c3b3011f72 100644
--- a/src/include/storage/lwlocklist.h
+++ b/src/include/storage/lwlocklist.h
@@ -84,3 +84,4 @@ PG_LWLOCK(50, DSMRegistry)
 PG_LWLOCK(51, InjectionPoint)
 PG_LWLOCK(52, SerialControl)
 PG_LWLOCK(53, WaitLSN)
+PG_LWLOCK(54, DataChecksumsWorker)
diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h
index f94c11a9a8..bba00f9d19 100644
--- a/src/include/storage/procsignal.h
+++ b/src/include/storage/procsignal.h
@@ -54,6 +54,11 @@ typedef enum
 typedef enum
 {
 	PROCSIGNAL_BARRIER_SMGRRELEASE, /* ask smgr to close files */
+
+	PROCSIGNAL_BARRIER_CHECKSUM_OFF,
+	PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON,
+	PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF,
+	PROCSIGNAL_BARRIER_CHECKSUM_ON,
 } ProcSignalBarrierType;
 
 /*
diff --git a/src/include/utils/backend_progress.h b/src/include/utils/backend_progress.h
index 7b63d38f97..063d3ef86c 100644
--- a/src/include/utils/backend_progress.h
+++ b/src/include/utils/backend_progress.h
@@ -28,6 +28,7 @@ typedef enum ProgressCommandType
 	PROGRESS_COMMAND_CREATE_INDEX,
 	PROGRESS_COMMAND_BASEBACKUP,
 	PROGRESS_COMMAND_COPY,
+	PROGRESS_COMMAND_DATACHECKSUMS,
 } ProgressCommandType;
 
 #define PGSTAT_NUM_PROGRESS_PARAM	20
diff --git a/src/test/Makefile b/src/test/Makefile
index dbd3192874..36023c1878 100644
--- a/src/test/Makefile
+++ b/src/test/Makefile
@@ -12,7 +12,15 @@ subdir = src/test
 top_builddir = ../..
 include $(top_builddir)/src/Makefile.global
 
-SUBDIRS = perl regress isolation modules authentication recovery subscription
+SUBDIRS = \
+		perl \
+		regress \
+		isolation \
+		modules \
+		authentication \
+		recovery \
+		subscription \
+		checksum
 
 ifeq ($(with_icu),yes)
 SUBDIRS += icu
diff --git a/src/test/checksum/.gitignore b/src/test/checksum/.gitignore
new file mode 100644
index 0000000000..871e943d50
--- /dev/null
+++ b/src/test/checksum/.gitignore
@@ -0,0 +1,2 @@
+# Generated by test suite
+/tmp_check/
diff --git a/src/test/checksum/Makefile b/src/test/checksum/Makefile
new file mode 100644
index 0000000000..fd03bf73df
--- /dev/null
+++ b/src/test/checksum/Makefile
@@ -0,0 +1,23 @@
+#-------------------------------------------------------------------------
+#
+# Makefile for src/test/checksum
+#
+# Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+# Portions Copyright (c) 1994, Regents of the University of California
+#
+# src/test/checksum/Makefile
+#
+#-------------------------------------------------------------------------
+
+subdir = src/test/checksum
+top_builddir = ../../..
+include $(top_builddir)/src/Makefile.global
+
+check:
+	$(prove_check)
+
+installcheck:
+	$(prove_installcheck)
+
+clean distclean maintainer-clean:
+	rm -rf tmp_check
diff --git a/src/test/checksum/README b/src/test/checksum/README
new file mode 100644
index 0000000000..0f0317060b
--- /dev/null
+++ b/src/test/checksum/README
@@ -0,0 +1,22 @@
+src/test/checksum/README
+
+Regression tests for data checksums
+===================================
+
+This directory contains a test suite for enabling data checksums
+in a running cluster.
+
+Running the tests
+=================
+
+    make check
+
+or
+
+    make installcheck
+
+NOTE: This creates a temporary installation (in the case of "check"),
+with multiple nodes, be they master or standby(s) for the purpose of
+the tests.
+
+NOTE: This requires the --enable-tap-tests argument to configure.
diff --git a/src/test/checksum/meson.build b/src/test/checksum/meson.build
new file mode 100644
index 0000000000..5f96b5c246
--- /dev/null
+++ b/src/test/checksum/meson.build
@@ -0,0 +1,15 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+tests += {
+  'name': 'checksums',
+  'sd': meson.current_source_dir(),
+  'bd': meson.current_build_dir(),
+  'tap': {
+    'tests': [
+      't/001_basic.pl',
+      't/002_restarts.pl',
+      't/003_standby_restarts.pl',
+      't/004_offline.pl',
+    ],
+  },
+}
diff --git a/src/test/checksum/t/001_basic.pl b/src/test/checksum/t/001_basic.pl
new file mode 100644
index 0000000000..f16cf78b91
--- /dev/null
+++ b/src/test/checksum/t/001_basic.pl
@@ -0,0 +1,85 @@
+
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test suite for testing enabling data checksums in an online cluster
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# pg_enable_checksums take three params: cost_delay, cost_limit and fast. For
+# testing we always want to override the default value for 'fast' with True
+# which will cause immediate checkpoints. 0 and 100 are the defaults for
+# cost_delay and cost_limit which are fine to use for testing so let's keep
+# them.
+my $enable_params = '0, 100, true';
+
+# Initialize node with checksums disabled.
+my $node = PostgreSQL::Test::Cluster->new('main');
+$node->init();
+$node->start();
+
+# Create some content to have un-checksummed data in the cluster
+$node->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;");
+
+# Ensure that checksums are turned off
+my $result = $node->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, 'off', 'ensure checksums are disabled');
+
+# Enable data checksums
+$node->safe_psql('postgres', "SELECT pg_enable_data_checksums($enable_params);");
+
+# Wait for checksums to become enabled
+$result = $node->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'on');
+is($result, 1, 'ensure checksums are enabled');
+
+# Run a dummy query just to make sure we can read back some data
+$result = $node->safe_psql('postgres', "SELECT count(*) FROM t");
+is($result, '10000', 'ensure checksummed pages can be read back');
+
+# Enable data checksums again which should be a no-op..
+$node->safe_psql('postgres', "SELECT pg_enable_data_checksums($enable_params);");
+# ..and make sure we can still read/write data
+$node->safe_psql('postgres', "UPDATE t SET a = a + 1;");
+$result = $node->safe_psql('postgres', "SELECT count(*) FROM t");
+is($result, '10000', 'ensure checksummed pages can be read back');
+
+# Disable checksums again
+$node->safe_psql('postgres', "SELECT pg_disable_data_checksums();");
+
+$result = $node->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'off');
+is($result, 1, 'ensure checksums are disabled');
+
+# Test reading again
+$result = $node->safe_psql('postgres', "SELECT count(*) FROM t");
+is($result, '10000', 'ensure previously checksummed pages can be read back');
+
+# Re-enable checksums and make sure that the underlying data has changed to
+# ensure that checksums will be different.
+$node->safe_psql('postgres', "UPDATE t SET a = a + 1;");
+
+$node->safe_psql('postgres', "SELECT pg_enable_data_checksums($enable_params);");
+$result = $node->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'on');
+is($result, 1, 'ensure checksums are enabled');
+
+# Run a dummy query just to make sure we can read back the data
+$result = $node->safe_psql('postgres', "SELECT count(*) FROM t");
+is($result, '10000', 'ensure checksummed pages can be read back');
+
+$node->stop;
+
+done_testing();
diff --git a/src/test/checksum/t/002_restarts.pl b/src/test/checksum/t/002_restarts.pl
new file mode 100644
index 0000000000..dea0ec31df
--- /dev/null
+++ b/src/test/checksum/t/002_restarts.pl
@@ -0,0 +1,90 @@
+
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test suite for testing enabling data checksums in an online cluster with
+# restarting the processing
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# pg_enable_checksums take three params: cost_delay, cost_limit and fast. For
+# testing we always want to override the default value for 'fast' with True
+# which will cause immediate checkpoints. 0 and 100 are the defaults for
+# cost_delay and cost_limit which are fine to use for testing so let's keep
+# them.
+my $enable_params = '0, 100, true';
+
+# Initialize node with checksums disabled.
+my $node = PostgreSQL::Test::Cluster->new('main');
+$node->init;
+$node->start;
+
+# Create some content to have un-checksummed data in the cluster
+$node->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;");
+
+# Ensure that checksums are disabled
+my $result = $node->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, 'off', 'ensure checksums are disabled');
+
+# Create a barrier for checksumming to block on, in this case a pre-existing
+# temporary table which is kept open while processing is started. We can
+# accomplish this by setting up an interactive psql process which keeps the
+# temporary table created as we enable checksums in another psql process.
+
+my $bsession = $node->background_psql('postgres');
+$bsession->query_safe('CREATE TEMPORARY TABLE tt (a integer);');
+
+# In another session, make sure we can see the blocking temp table but start
+# processing anyways and check that we are blocked with a proper wait event.
+$result = $node->safe_psql('postgres',
+	"SELECT relpersistence FROM pg_catalog.pg_class WHERE relname = 'tt';");
+is($result, 't', 'ensure we can see the temporary table');
+
+$node->safe_psql('postgres', "SELECT pg_enable_data_checksums($enable_params);");
+
+$result = $node->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'inprogress-on');
+is($result, '1', "ensure checksums aren't enabled yet");
+
+$bsession->quit;
+$node->stop;
+$node->start;
+
+$result = $node->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, 'inprogress-on', "ensure checksums aren't enabled yet");
+
+$node->safe_psql('postgres', "SELECT pg_enable_data_checksums($enable_params);");
+
+$result = $node->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'on');
+is($result, 1, 'ensure checksums are turned on');
+
+$result = $node->safe_psql('postgres', "SELECT count(*) FROM t");
+is($result, '10000', 'ensure checksummed pages can be read back');
+
+$result = $node->poll_query_until(
+	'postgres',
+	"SELECT count(*) FROM pg_stat_activity WHERE backend_type LIKE 'datachecksumsworker%';",
+	'0');
+is($result, 1, 'await datachecksums worker/launcher termination');
+
+$result = $node->safe_psql('postgres', "SELECT pg_disable_data_checksums();");
+$result = $node->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'off');
+is($result, 1, 'ensure checksums are turned off');
+
+done_testing();
diff --git a/src/test/checksum/t/003_standby_restarts.pl b/src/test/checksum/t/003_standby_restarts.pl
new file mode 100644
index 0000000000..26ad93f86e
--- /dev/null
+++ b/src/test/checksum/t/003_standby_restarts.pl
@@ -0,0 +1,130 @@
+
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test suite for testing enabling data checksums in an online cluster with
+# streaming replication
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# pg_enable_checksums take three params: cost_delay, cost_limit and fast. For
+# testing we always want to override the default value for 'fast' with True
+# which will cause immediate checkpoints. 0 and 100 are the defaults for
+# cost_delay and cost_limit which are fine to use for testing so let's keep
+# them.
+my $enable_params = '0, 100, true';
+
+# Initialize primary node
+my $node_primary = PostgreSQL::Test::Cluster->new('primary');
+$node_primary->init(allows_streaming => 1);
+$node_primary->start;
+my $backup_name = 'my_backup';
+
+# Take backup
+$node_primary->backup($backup_name);
+
+# Create streaming standby linking to primary
+my $node_standby_1 = PostgreSQL::Test::Cluster->new('standby_1');
+$node_standby_1->init_from_backup($node_primary, $backup_name,
+	has_streaming => 1);
+$node_standby_1->start;
+
+# Create some content on the primary to have un-checksummed data in the cluster
+$node_primary->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;");
+
+# Wait for standbys to catch up
+$node_primary->wait_for_catchup($node_standby_1, 'replay',
+	$node_primary->lsn('insert'));
+
+# Check that checksums are turned off on all nodes
+my $result = $node_primary->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, "off", 'ensure checksums are turned off on primary');
+
+$result = $node_standby_1->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, "off", 'ensure checksums are turned off on standby_1');
+
+# Enable checksums for the cluster
+$node_primary->safe_psql('postgres', "SELECT pg_enable_data_checksums($enable_params);");
+
+# Ensure that the primary switches to "inprogress-on"
+$result = $node_primary->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	"inprogress-on");
+is($result, 1, 'ensure checksums are in progress on primary');
+
+# Wait for checksum enable to be replayed
+$node_primary->wait_for_catchup($node_standby_1, 'replay');
+
+# Ensure that the standby has switched to "inprogress-on" or "on".  Normally it
+# would be "inprogress-on", but it is theoretically possible for the primary to
+# complete the checksum enabling *and* have the standby replay that record
+# before we reach the check below.
+$result = $node_standby_1->poll_query_until(
+	'postgres',
+	"SELECT setting = 'off' FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'f');
+is($result, 1, 'ensure standby has absorbed the inprogress-on barrier');
+$result = $node_standby_1->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+
+is(($result eq 'inprogress-on' || $result eq 'on'),
+	1, 'ensure checksums are on, or in progress, on standby_1');
+
+# Insert some more data which should be checksummed on INSERT
+$node_primary->safe_psql('postgres',
+	"INSERT INTO t VALUES (generate_series(1, 10000));");
+
+# Wait for checksums enabled on the primary
+$result = $node_primary->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'on');
+is($result, 1, 'ensure checksums are enabled on the primary');
+
+# Wait for checksums enabled on the standby
+$result = $node_standby_1->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'on');
+is($result, 1, 'ensure checksums are enabled on the standby');
+
+$result = $node_primary->safe_psql('postgres', "SELECT count(a) FROM t");
+is($result, '20000', 'ensure we can safely read all data with checksums');
+
+$result = $node_primary->poll_query_until(
+	'postgres',
+	"SELECT count(*) FROM pg_stat_activity WHERE backend_type LIKE 'datachecksumsworker%';",
+	'0');
+is($result, 1, 'await datachecksums worker/launcher termination');
+
+# Disable checksums and ensure it's propagated to standby and that we can
+# still read all data
+$node_primary->safe_psql('postgres', "SELECT pg_disable_data_checksums();");
+# Wait for checksum disable to be replayed
+$node_primary->wait_for_catchup($node_standby_1, 'replay');
+$result = $node_primary->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'off');
+is($result, 1, 'ensure data checksums are disabled on the primary 2');
+
+# Ensure that the standby has switched to off
+$result = $node_standby_1->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'off');
+is($result, 1, 'ensure checksums are off on standby_1');
+
+$result = $node_primary->safe_psql('postgres', "SELECT count(a) FROM t");
+is($result, "20000", 'ensure we can safely read all data without checksums');
+
+done_testing();
diff --git a/src/test/checksum/t/004_offline.pl b/src/test/checksum/t/004_offline.pl
new file mode 100644
index 0000000000..b1f585ec7c
--- /dev/null
+++ b/src/test/checksum/t/004_offline.pl
@@ -0,0 +1,100 @@
+
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test suite for testing enabling data checksums offline from various states
+# of checksum processing
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# pg_enable_checksums take three params: cost_delay, cost_limit and fast. For
+# testing we always want to override the default value for 'fast' with True
+# which will cause immediate checkpoints. 0 and 100 are the defaults for
+# cost_delay and cost_limit which are fine to use for testing so let's keep
+# them.
+my $enable_params = '0, 100, true';
+
+# Initialize node with checksums disabled.
+my $node = PostgreSQL::Test::Cluster->new('main');
+$node->init();
+$node->start();
+
+# Create some content to have un-checksummed data in the cluster
+$node->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;");
+
+# Ensure that checksums are disabled
+my $result = $node->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, 'off', 'ensure checksums are disabled');
+
+# Enable checksums offline using pg_checksums
+$node->stop;
+$node->checksum_enable_offline;
+$node->start;
+
+# Ensure that checksums are enabled
+$result = $node->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, 'on', 'ensure checksums are enabled');
+
+# Run a dummy query just to make sure we can read back some data
+$result = $node->safe_psql('postgres', "SELECT count(*) FROM t");
+is($result, '10000', 'ensure checksummed pages can be read back');
+
+# Disable checksums offline again using pg_checksums
+$node->stop;
+$node->checksum_disable_offline;
+$node->start;
+
+# Ensure that checksums are disabled
+$result = $node->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, 'off', 'ensure checksums are disabled');
+
+# Create a barrier for checksumming to block on, in this case a pre-existing
+# temporary table which is kept open while processing is started. We can
+# accomplish this by setting up an interactive psql process which keeps the
+# temporary table created as we enable checksums in another psql process.
+
+my $bsession = $node->background_psql('postgres');
+$bsession->query_safe('CREATE TEMPORARY TABLE tt (a integer);');
+
+# In another session, make sure we can see the blocking temp table but start
+# processing anyways and check that we are blocked with a proper wait event.
+$result = $node->safe_psql('postgres',
+	"SELECT relpersistence FROM pg_catalog.pg_class WHERE relname = 'tt';");
+is($result, 't', 'ensure we can see the temporary table');
+
+$node->safe_psql('postgres', "SELECT pg_enable_data_checksums($enable_params);");
+
+$result = $node->poll_query_until(
+	'postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';",
+	'inprogress-on');
+is($result, 1, 'ensure checksums are in the process of being enabled');
+
+# Turn the cluster off and enable checksums offline, then start back up
+$bsession->quit;
+$node->stop;
+$node->checksum_enable_offline;
+$node->start;
+
+# Ensure that checksums are now enabled even though processing wasn't
+# restarted
+$result = $node->safe_psql('postgres',
+	"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"
+);
+is($result, 'on', 'ensure checksums are enabled');
+
+# Run a dummy query just to make sure we can read back some data
+$result = $node->safe_psql('postgres', "SELECT count(*) FROM t");
+is($result, '10000', 'ensure checksummed pages can be read back');
+
+done_testing();
diff --git a/src/test/meson.build b/src/test/meson.build
index c3d0dfedf1..b07d5d2d00 100644
--- a/src/test/meson.build
+++ b/src/test/meson.build
@@ -7,6 +7,7 @@ subdir('authentication')
 subdir('recovery')
 subdir('subscription')
 subdir('modules')
+subdir('checksum')
 
 if ssl.found()
   subdir('ssl')
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm
index 90a842f96a..c008459ae8 100644
--- a/src/test/perl/PostgreSQL/Test/Cluster.pm
+++ b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -3483,6 +3483,40 @@ sub advance_wal
 	}
 }
 
+=item $node->checksum_enable_offline()
+
+Enable data page checksums in an offline cluster with B<pg_checksums>. The
+caller is responsible for ensuring that the cluster is in the right state for
+this operation.
+
+=cut
+
+sub checksum_enable_offline
+{
+	my ($self) = @_;
+
+	print "### Enabling checksums in \"$self->data_dir\"\n";
+	PostgreSQL::Test::Utils::system_or_bail('pg_checksums', '-D', $self->data_dir, '-e');
+	return;
+}
+
+=item checksum_disable_offline
+
+Disable data page checksums in an offline cluster with B<pg_checksums>. The
+caller is responsible for ensuring that the cluster is in the right state for
+this operation.
+
+=cut
+
+sub checksum_disable_offline
+{
+	my ($self) = @_;
+
+	print "### Disabling checksums in \"$self->data_dir\"\n";
+	PostgreSQL::Test::Utils::system_or_bail('pg_checksums', '-D', $self->data_dir, '-d');
+	return;
+}
+
 =pod
 
 =back
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 5fabb127d7..929232bf5b 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -396,6 +396,7 @@ CheckPointStmt
 CheckpointStatsData
 CheckpointerRequest
 CheckpointerShmemStruct
+ChecksumType
 Chromosome
 CkptSortItem
 CkptTsStatus
@@ -582,6 +583,10 @@ DataPageDeleteStack
 DataTypesUsageChecks
 DataTypesUsageVersionCheck
 DatabaseInfo
+DataChecksumsWorkerDatabase
+DataChecksumsWorkerResult
+DataChecksumsWorkerResultEntry
+DataChecksumsWorkerShmemStruct
 DateADT
 DateTimeErrorExtra
 Datum
-- 
2.39.3 (Apple Git-146)

