WAL, checkpoints, checksums and durability

crash_session.py

"""A server killed in the middle of work: the committed rows come back from the WAL, the open transaction's do not.

The server is stopped with `pg_ctl -m immediate`, which the manual calls equivalent to a crash: no checkpoint, no
flush of the data pages. The recovery lines of the server log are shown with their log sequence numbers replaced by
"0/…" and the timings after "system usage" removed, because both differ on every run.
"""

import re
import subprocess

from pg_session import BIN, Server

SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'",
            "log_line_prefix": "''", "checkpoint_timeout": "'1d'"}
LOG_KEYS = ("automatic recovery", "redo starts", "redo done")


def control(pg, note):
    """The two lines of pg_controldata that recovery starts from; returns the checkpoint's REDO location."""
    out = subprocess.run([str(BIN / "pg_controldata"), "-D", "data"], cwd=pg.base, env=pg.env, text=True,
                         stdout=subprocess.PIPE).stdout
    print(f"-- {note}")
    print("$ pg_controldata data")
    redo = ""
    for line in out.splitlines():
        if line.startswith(("Database cluster state", "Latest checkpoint's REDO location")):
            print(re.sub(r"0/[0-9A-F]+", "0/…", line))
        if line.startswith("Latest checkpoint's REDO location"):
            redo = line.split()[-1]
    print()
    return redo


def recovery_log(pg, redo):
    """The lines the restart wrote about recovery, and whether redo began at the checkpoint's REDO location."""
    print("-- data.log, the lines about recovery:")
    began = ""
    for line in (pg.base / "data.log").read_text().splitlines():
        if any(key in line for key in LOG_KEYS):
            if "redo starts at" in line:
                began = line.split()[-1]
            line = re.sub(r"0/[0-9A-F]+", "0/…", line)
            print(re.sub(r"(system usage:).*", r"\1 …", line))
    print(f"-- redo began at the REDO location of pg_controldata: {'yes' if began == redo else 'no'}")
    print()


with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("crash.sql")
    open_txn = subprocess.Popen([str(BIN / "psql"), "-X", "-q"], env=pg.env, cwd=pg.base, text=True,
                                stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    open_txn.stdin.write((pg.base / "crash_open.sql").read_text())
    open_txn.stdin.flush()
    pg.wait_for("SELECT count(*) FROM pg_stat_activity WHERE state = 'idle in transaction'", "1")
    pg.show_file("crash_open.sql", note="session 2, still open")
    pg.stop(mode="immediate", show=True)
    open_txn.stdin.close()
    open_txn.wait(timeout=60)
    redo = control(pg, "the server was killed, not shut down")
    pg.start(show=True)
    recovery_log(pg, redo)
    pg.psql("crash_after.sql")

$ psql -X -a -f crash.sql
-- A table, a checkpoint, then one committed transaction of 1,000 rows.
CREATE TABLE orders (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, customer text NOT NULL,
                     total numeric(10,2) NOT NULL);
CREATE TABLE
CHECKPOINT;
CHECKPOINT
INSERT INTO orders (customer, total) SELECT 'customer ' || n, n FROM generate_series(1, 1000) AS n;
INSERT 0 1000
SELECT count(*) AS committed_rows FROM orders;
 committed_rows 
----------------
           1000
(1 row)

-- session 2, still open
$ cat crash_open.sql
-- A second session starts a transaction of 500 rows and never commits it.
BEGIN;
INSERT INTO orders (customer, total) SELECT 'open ' || n, n FROM generate_series(1, 500) AS n;

$ pg_ctl -D data -m immediate -w -t 60 stop
waiting for server to shut down... done
server stopped

-- the server was killed, not shut down
$ pg_controldata data
Database cluster state:               in production
Latest checkpoint's REDO location:    0/…

$ pg_ctl -D data -l data.log -w -t 60 start
waiting for server to start... done
server started

-- data.log, the lines about recovery:
LOG:  database system was not properly shut down; automatic recovery in progress
LOG:  redo starts at 0/…
LOG:  redo done at 0/… system usage: …
-- redo began at the REDO location of pg_controldata: yes

$ psql -X -a -f crash_after.sql
-- After the restart: the committed rows are back, the open transaction's rows are not.
SELECT count(*) AS rows_after_crash,
       count(*) FILTER (WHERE customer LIKE 'open %') AS from_the_open_transaction
  FROM orders;
 rows_after_crash | from_the_open_transaction 
------------------+---------------------------
             1000 |                         0
(1 row)

crash.sql

-- A table, a checkpoint, then one committed transaction of 1,000 rows.
CREATE TABLE orders (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, customer text NOT NULL,
                     total numeric(10,2) NOT NULL);
CHECKPOINT;
INSERT INTO orders (customer, total) SELECT 'customer ' || n, n FROM generate_series(1, 1000) AS n;
SELECT count(*) AS committed_rows FROM orders;

crash_open.sql

-- A second session starts a transaction of 500 rows and never commits it.
BEGIN;
INSERT INTO orders (customer, total) SELECT 'open ' || n, n FROM generate_series(1, 500) AS n;

crash_after.sql

-- After the restart: the committed rows are back, the open transaction's rows are not.
SELECT count(*) AS rows_after_crash,
       count(*) FILTER (WHERE customer LIKE 'open %') AS from_the_open_transaction
  FROM orders;

pg_session.py

"""Throwaway PostgreSQL 18.6 clusters for the examples of this lesson, including standbys and a second cluster.

Every session creates its clusters with initdb in new folders inside the examples folder, starts them, runs psql
scripts and server programs on them, prints what they printed, and stops and deletes them again, also when something
fails. The servers are the pinned PostgreSQL 18.6 from the mise installations next to the Python that runs this file.
Each listens only on 127.0.0.1, an address no other computer can reach, on a port the operating system reports as
unused, with no Unix socket, and asks for a password (scram-sha-256) made up for the run and never printed. A server
that another server must reach (a standby, a subscriber) is given a password file of its own, mode 0600, so no
connection string in the output carries a password.
learn:verify's run sandbox cannot read the home folder, where mise keeps PostgreSQL, so these sessions are recorded
with LEARN_VERIFY_NETWORK=allow; they read only this folder and the toolchain and connect to nothing but 127.0.0.1.

The output is exactly what the programs printed, with five substitutions so that a record never shows the computer it
ran on and is the same on every run: each run folder is shown as ".", the folder of PostgreSQL's programs is left out
of their paths, the operating-system user's name is shown as "learner", the dots pg_ctl prints while it waits (one
about every tenth of a second) as three, and the port the operating system picked for each cluster as a fixed number
(5432 for the first cluster, 5433 for its standby, 5434 for a second cluster), the numbers a reader would see with
PostgreSQL's own defaults. A session may name more patterns (mask=) for values that change on every run, such as a
log sequence number, a timing or a process id; the session's own file says which, and its record shows fixed text in
their place. A session stops with status 1 when a command fails that should have worked, or reports a different
number of errors than the session expects, so a run that went wrong is never recorded as if it had worked.
"""

import os
import pwd
import re
import secrets
import shlex
import shutil
import socket
import subprocess
import sys
import tempfile
import textwrap
import time
from pathlib import Path

PG_VERSION = "18.6"
# .../installs/python/3.14.8/bin/python3 -> .../installs
INSTALLS = Path(sys.executable).resolve().parents[3]
BIN = INSTALLS / "postgres" / PG_VERSION / "bin"
OS_USER = pwd.getpwuid(os.geteuid()).pw_name
# Every cluster of the run, so that each one's output can be cleaned of every run folder and port.
BASES = []
PORTS = {}
# The passwords of this run: no output of a session ever shows one, whatever a program decides to write.
PASSWORDS = []


def free_port():
    """A TCP port on 127.0.0.1 that nothing listens on now."""
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


class Server:
    """One throwaway cluster: initdb, settings, start; stopped and deleted when the with block ends."""

    def __init__(self, settings=None, initdb=(), name="data", shown_port=5432):
        self.settings = dict(settings or {})
        self.initdb_args = list(initdb)
        self.name = name
        self.shown_port = shown_port
        self.running = []

    def __enter__(self):
        if not (BIN / "postgres").exists():
            sys.exit(f"PostgreSQL {PG_VERSION} was not found next to this Python")
        self.base = Path(tempfile.mkdtemp(prefix="pg", dir=".")).resolve()
        self.port = free_port()
        self.password = secrets.token_urlsafe(24)
        (self.base / "pwfile").write_text(self.password + "\n")
        BASES.append(self.base)
        PORTS[self.port] = self.shown_port
        PASSWORDS.append(self.password)
        self.env = {**os.environ, "PGHOST": "127.0.0.1", "PGPORT": str(self.port), "PGUSER": "postgres",
                    "PGPASSWORD": self.password, "PGDATABASE": "postgres", "PGCONNECT_TIMEOUT": "10"}
        for key in ("PGSERVICE", "PGSERVICEFILE", "PGOPTIONS", "PSQLRC"):
            self.env.pop(key, None)
        # The server itself must not inherit this cluster's password: a replication connection to ANOTHER cluster
        # would use it instead of the password file its connection string names.
        self.server_env = {k: v for k, v in self.env.items() if k != "PGPASSWORD"}
        for script in Path(".").glob("*.sql"):
            shutil.copy(script, self.base / script.name)
        for data in Path(".").glob("*.csv"):
            shutil.copy(data, self.base / data.name)
        return self

    def __exit__(self, *exc):
        for folder, _port in reversed(self.running):
            self._call(["pg_ctl", "-D", folder, "-m", "fast", "-w", "-t", "60", "stop"])
        self.running = []
        shutil.rmtree(self.base, ignore_errors=True)

    # ---- the cluster ---------------------------------------------------------------------------------------------
    def initdb(self, show=False):
        args = ["initdb", "-D", self.name, "-U", "postgres", "--auth=scram-sha-256", "--pwfile=pwfile",
                *self.initdb_args]
        if not show:
            args.append("--no-instructions")
        self._expect(args, show=show)
        with open(self.base / self.name / "postgresql.conf", "a") as conf:
            conf.write("\n# Added by the session: 127.0.0.1 only, no Unix socket.\n")
            conf.write(f"listen_addresses = '127.0.0.1'\nport = {self.port}\nunix_socket_directories = ''\n")
            for key, value in self.settings.items():
                conf.write(f"{key} = {value}\n")

    def start(self, show=False, folder=None, port=None, note=None):
        folder = folder or self.name
        if note:
            print(f"-- {note}")
        code, out = self._call(["pg_ctl", "-D", folder, "-l", f"{folder}.log", "-w", "-t", "60", "start"],
                               env=self.server_env)
        if show:
            _show(["pg_ctl", "-D", folder, "-l", f"{folder}.log", "-w", "-t", "60", "start"], out)
        if code != 0:
            if not show:
                print(out)
            log = self.base / f"{folder}.log"
            if log.exists():
                print(self.clean("\n".join(log.read_text().splitlines()[-25:])))
            sys.exit(f"pg_ctl start of {folder} ended with status {code}")
        self.running.append((folder, port or self.port))

    def stop(self, folder=None, show=False, mode="fast"):
        """Stops one cluster of this session (the standby, or the primary of a failover drill)."""
        folder = folder or self.name
        self._expect(["pg_ctl", "-D", folder, "-m", mode, "-w", "-t", "60", "stop"], show=show)
        self.running = [r for r in self.running if r[0] != folder]

    def up(self):
        """initdb and start, quietly."""
        self.initdb()
        self.start()
        return self

    def passfile(self):
        """A libpq password file for this cluster, so a connection string never carries the password."""
        path = self.base / "pgpass"
        if not path.exists():
            path.write_text(f"127.0.0.1:*:*:postgres:{self.password}\n")
            path.chmod(0o600)
        return path

    def conninfo(self, db="postgres", user="postgres", application_name=None, port=None, hosts=None):
        """A connection string another server can use to reach this one (password in a file, never in the text).

        port names another cluster of this run (a promoted standby); hosts is a list of ports for a multi-host
        connection string (host=127.0.0.1,127.0.0.1 port=a,b).
        """
        if hosts:
            where = f"host={','.join('127.0.0.1' for _ in hosts)} port={','.join(str(p) for p in hosts)}"
        else:
            where = f"host=127.0.0.1 port={port or self.port}"
        parts = [f"{where} user={user} dbname={db} passfile={self.passfile()}"]
        if application_name:
            parts.append(f"application_name={application_name}")
        return " ".join(parts)

    def standby(self, folder="standby", slot=None, shown_port=5433, wait=True, extra=(), show=False, settings=None):
        """A streaming standby of this cluster, made with pg_basebackup -R, started on a port of its own.

        slot names a physical replication slot pg_basebackup creates and the standby then uses (-C -S). extra adds
        arguments to pg_basebackup; settings are lines appended to the standby's postgresql.conf. Returns its port.
        """
        args = ["pg_basebackup", "-D", folder, "-R", "-X", "stream", "-c", "fast"]
        if slot:
            args += ["-C", "-S", slot]
        args += list(extra)
        # PGPASSFILE (and no PGPASSWORD) so that the primary_conninfo pg_basebackup -R writes names the password
        # file: the standby then authenticates without a password in any file the record shows.
        self._expect(args, show=show, env={**self.server_env, "PGPASSFILE": str(self.passfile())})
        port = free_port()
        PORTS[port] = shown_port
        with open(self.base / folder / "postgresql.conf", "a") as conf:
            conf.write(f"port = {port}\n")
            for key, value in dict(settings or {}).items():
                conf.write(f"{key} = {value}\n")
        self.start(folder=folder, port=port)
        if wait:
            for _ in range(300):
                if self.query("SELECT count(*) FROM pg_stat_replication WHERE state = 'streaming'") != "0":
                    return port
                time.sleep(0.1)
            sys.exit("the standby did not start streaming")
        return port

    def promote(self, folder="standby", show=True, port=None):
        """Promotes a standby to a primary with pg_ctl promote and waits until it accepts writes."""
        self._expect(["pg_ctl", "-D", folder, "-w", "-t", "60", "promote"], show=show)
        port = port or dict((f, p) for f, p in self.running).get(folder)
        for _ in range(300):
            if self.query("SELECT pg_is_in_recovery()", port=port) == "f":
                return port
            time.sleep(0.1)
        sys.exit(f"{folder} did not leave recovery")

    def rewind(self, folder, source=None, port=None, show=True, extra=(), target_port=None):
        """pg_rewind: resynchronises the stopped cluster in `folder` with a running cluster (a Server or a port).

        It runs without this cluster's password in the environment and with PGPASSFILE set, so a -R run writes a
        primary_conninfo that names the password file instead of a password. pg_rewind copies the source cluster's
        postgresql.conf, which on one machine carries the source's own port, so the target's port is written back
        afterwards (one machine runs both clusters here; on two machines both would listen on 5432).
        """
        who = source or self
        args = ["pg_rewind", "--target-pgdata", folder, f"--source-server={who.conninfo(port=port)}", *extra]
        out = self._expect(args, show=show, env={**self.server_env, "PGPASSFILE": str(who.passfile())})
        mine = target_port or dict((f, p) for f, p in self.running).get(folder) or self.port
        with open(self.base / folder / "postgresql.conf", "a") as conf:
            conf.write(f"port = {mine}\n")
        return out

    # ---- running things ------------------------------------------------------------------------------------------
    def _call(self, args, env=None):
        args = [str(a) for a in args]
        done = subprocess.run([str(BIN / args[0]), *args[1:]], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                              text=True, cwd=self.base, env=env or self.env)
        return done.returncode, self.clean(done.stdout)

    def _expect(self, args, show=False, status=0, mask=(), env=None):
        code, out = self._call(args, env=env)
        out = _masked(out, mask)
        if show:
            _show(args, out)
        if code != status:
            if not show:
                print(out)
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def clean(self, text):
        """Every run folder as ".", the user as learner, pg_ctl's dots as three, each port as a fixed number.

        A password the run made up is never shown, even when a program (pg_rewind -R, for one) writes it into a file
        the session prints.
        """
        text = text.replace(str(BIN) + "/", "")
        for base in BASES:
            text = text.replace(str(base) + "/", "").replace(str(base), ".")
        for real, shown in PORTS.items():
            text = re.sub(rf"\b{real}\b", str(shown), text)
        for password in PASSWORDS:
            text = text.replace(password, "(the password of this run, not shown)")
        text = re.sub(r"(waiting for server to (?:start|shut down|promote))\.+", r"\1...", text)
        return text.replace(f'"{OS_USER}"', '"learner"')

    def wait_for(self, sql, value, port=None):
        """Waits (unprinted) until a query returns the value, for example after a reload."""
        for _ in range(600):
            if self.query(sql, port=port) == value:
                return
            time.sleep(0.1)
        sys.exit(f"{sql} never returned {value}")

    def show_file(self, path, note=None):
        """Prints a text file of the cluster, as `cat` would."""
        if note:
            print(f"-- {note}")
        print(f"$ cat {path}")
        print(self.clean((self.base / path).read_text()).rstrip("\n"))
        print()

    def log_lines(self, path, match, count=1, note=None, mask=()):
        """Waits for and prints the server-log lines that contain `match` (the server log of one cluster)."""
        log = self.base / path
        for _ in range(600):
            found = [ln for ln in log.read_text().splitlines() if match in ln] if log.exists() else []
            if len(found) >= count:
                break
            time.sleep(0.1)
        else:
            sys.exit(f"{path} never showed {count} line(s) with {match!r}")
        print(f"-- {note or f'{path}, the lines about {match}'}:")
        for line in found[:count]:
            print(_masked(self.clean(line), mask))
        print()
        return found[:count]

    def query(self, sql, port=None):
        """One value from one query, unprinted (the session uses it to wait or to decide)."""
        env = {**self.env, "PGPORT": str(port or self.port)}
        code, out = self._call(["psql", "-X", "-At", "-c", sql], env=env)
        if code != 0:
            sys.exit(f"the query failed: {out}")
        return out.strip()

    def silent(self, sql, port=None):
        """Runs a statement without printing it (used only to give a role the run's password)."""
        self.query(sql, port=port)

    def psql(self, script, errors=0, status=None, user=None, db=None, port=None, flags=("-a",), note=None, mask=(),
             variables=()):
        """Runs a psql script with -X (no ~/.psqlrc) and the given flags, and prints the command and its output.
        `errors` is how many ERROR lines the script must print; `status` the exit status it must end with;
        `variables` are (name, value) pairs the script reads as :'name' (a connection string, for example)."""
        env = {**self.env, "PGPORT": str(port or self.port)}
        if user:
            env["PGUSER"] = user
        if db:
            env["PGDATABASE"] = db
        passed = []
        for name, value in variables:
            passed += ["-v", f"{name}={value}"]
        args = ["psql", "-X", *flags, *passed, "-f", script]
        code, out = self._call(args, env=env)
        out = _masked(out, mask)
        shown = ["psql", "-X", *flags, "-f", script]
        if user:
            shown[1:1] = ["-U", user]
        if db:
            shown[1:1] = ["-d", db]
        if note:
            print(f"-- {note}")
        _show(shown, out)
        found = sum(1 for line in out.splitlines() if "ERROR:" in line or ": error:" in line)
        expected_status = 0 if status is None else status
        if found != errors or code != expected_status:
            sys.exit(f"{script}: {found} errors and status {code}; the session expects {errors} and {expected_status}")

    def sessions(self, count):
        """Opens `count` idle psql connections (each one a client backend) until the with block ends."""
        return _Idle(self, count)

    def shell(self, *args, status=0, mask=(), note=None, timeout=None, stdin=None):
        """Runs one of PostgreSQL's programs and prints the command and its output."""
        if note:
            print(f"-- {note}")
        if timeout is None:
            return self._expect(list(args), show=True, status=status, mask=mask)
        return self._timed(list(args), status=status, mask=mask, timeout=timeout, stdin=stdin)

    def _timed(self, args, status, mask, timeout, stdin=None):
        """Runs a program that would not stop by itself (pg_recvlogical) and prints what it printed until timeout."""
        args = [str(a) for a in args]
        proc = subprocess.Popen([str(BIN / args[0]), *args[1:]], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                stdin=subprocess.PIPE if stdin is not None else None, text=True, cwd=self.base,
                                env=self.env)
        try:
            out, _ = proc.communicate(input=stdin, timeout=timeout)
        except subprocess.TimeoutExpired:
            proc.terminate()
            out, _ = proc.communicate()
        out = _masked(self.clean(out), mask)
        _show(args, out)
        return out

    def listing(self, folder, prefix=""):
        """Prints the names in a folder of the cluster (sorted, folders with a slash), as `ls -p` would list them."""
        path = self.base / folder
        names = sorted(n + ("/" if (path / n).is_dir() else "") for n in os.listdir(path) if n.startswith(prefix))
        print(f"-- in {folder}" + (f", the names that start with {prefix}" if prefix else "") + ":")
        print(textwrap.fill("  ".join(names), width=78, initial_indent="   ", subsequent_indent="   ",
                            break_on_hyphens=False))
        print()
        return names


def _masked(text, mask):
    """Replaces what differs on every run (an LSN, a timing, a process id) with fixed text (see the docstring)."""
    for pattern, replacement in mask:
        text = re.sub(pattern, replacement, text, flags=re.M)
    return text


def _arg(a):
    """A command-line argument as a reader would type it: double quotes around text with single quotes in it."""
    a = str(a)
    if re.fullmatch(r"[\w@%+=:,./-]+", a):
        return a
    if not re.search(r"[\"$`\\!]", a):
        return f'"{a}"'
    return shlex.quote(a)


def _show(args, out):
    print("$ " + " ".join(_arg(a) for a in args))
    if out.strip():
        print(out.rstrip("\n"))
    print()


class _Idle:
    """Idle psql connections, closed again when the with block ends."""

    def __init__(self, server, count):
        self.server, self.count, self.procs = server, count, []

    def __enter__(self):
        before = int(self.server.query("SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'client backend'"))
        for _ in range(self.count):
            self.procs.append(subprocess.Popen([str(BIN / "psql"), "-X", "-q"], stdin=subprocess.PIPE,
                                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                                               env=self.server.env, cwd=self.server.base))
        sql = "SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'client backend'"
        for _ in range(300):
            # the query's own connection is one more client backend
            if int(self.server.query(sql)) == before + self.count:
                return self
            time.sleep(0.1)
        sys.exit("the idle connections did not open")

    def __exit__(self, *exc):
        for p in self.procs:
            p.stdin.close()
            p.wait(timeout=30)

lsn_session.py

"""Log sequence numbers: what a literal LSN means, and how a difference measures the WAL a statement wrote.

The first script uses literal LSNs only, so it prints the same on every server. The second prints no LSN of the
run: it keeps two positions in psql variables and shows only their difference in bytes.
"""

from pg_session import Server

SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'",
            "checkpoint_timeout": "'1d'"}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("lsn_literals.sql")
    pg.psql("lsn_measure.sql")

$ psql -X -a -f lsn_literals.sql
-- An LSN is a byte position in the WAL. These literals give the same answers on every server.
SELECT '0/3521448'::pg_lsn AS lsn;
    lsn    
-----------
 0/3521448
(1 row)

SELECT pg_walfile_name('0/3521448') AS segment_file;
       segment_file       
--------------------------
 000000010000000000000003
(1 row)

SELECT * FROM pg_walfile_name_offset('0/3521448');
        file_name         | file_offset 
--------------------------+-------------
 000000010000000000000003 |     5379144
(1 row)

SELECT * FROM pg_split_walfile_name('000000010000000000000003');
 segment_number | timeline_id 
----------------+-------------
              3 |           1
(1 row)

SELECT pg_wal_lsn_diff('0/3521448', '0/3500000') AS bytes_between;
 bytes_between 
---------------
        136264
(1 row)

SELECT pg_walfile_name('0/2FFFFFF') AS last_byte_of_2, pg_walfile_name('0/3000000') AS first_byte_of_3;
      last_byte_of_2      |     first_byte_of_3      
--------------------------+--------------------------
 000000010000000000000002 | 000000010000000000000003
(1 row)

SELECT pg_walfile_name('1/0') AS first_byte_after_4_gib;
  first_byte_after_4_gib  
--------------------------
 000000010000000100000000
(1 row)

$ psql -X -a -f lsn_measure.sql
-- On the live server: which file is the WAL in now, and what does one small commit add to it?
SELECT name, size FROM pg_ls_waldir() ORDER BY name;
           name           |   size   
--------------------------+----------
 000000010000000000000001 | 16777216
(1 row)

CREATE TABLE notes (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, body text NOT NULL);
CREATE TABLE
INSERT INTO notes (body) VALUES ('warm-up: the first row also creates the first pages');
INSERT 0 1
SELECT pg_current_wal_lsn() AS before \gset
INSERT INTO notes (body) VALUES ('one more row');
INSERT 0 1
SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), :'before') AS small \gset
SELECT :small AS bytes_for_one_row, 16 * 1024 * 1024 / :small AS commits_per_segment;
 bytes_for_one_row | commits_per_segment 
-------------------+---------------------
               184 |               91180
(1 row)

SELECT pg_current_wal_lsn() AS before \gset
INSERT INTO notes (body) VALUES (repeat('x', 1000));
INSERT 0 1
SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), :'before') AS bytes_for_a_1000_byte_row;
 bytes_for_a_1000_byte_row 
---------------------------
                      1176
(1 row)

lsn_literals.sql

-- An LSN is a byte position in the WAL. These literals give the same answers on every server.
SELECT '0/3521448'::pg_lsn AS lsn;
SELECT pg_walfile_name('0/3521448') AS segment_file;
SELECT * FROM pg_walfile_name_offset('0/3521448');
SELECT * FROM pg_split_walfile_name('000000010000000000000003');
SELECT pg_wal_lsn_diff('0/3521448', '0/3500000') AS bytes_between;
SELECT pg_walfile_name('0/2FFFFFF') AS last_byte_of_2, pg_walfile_name('0/3000000') AS first_byte_of_3;
SELECT pg_walfile_name('1/0') AS first_byte_after_4_gib;

lsn_measure.sql

-- On the live server: which file is the WAL in now, and what does one small commit add to it?
SELECT name, size FROM pg_ls_waldir() ORDER BY name;
CREATE TABLE notes (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, body text NOT NULL);
INSERT INTO notes (body) VALUES ('warm-up: the first row also creates the first pages');
SELECT pg_current_wal_lsn() AS before \gset
INSERT INTO notes (body) VALUES ('one more row');
SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), :'before') AS small \gset
SELECT :small AS bytes_for_one_row, 16 * 1024 * 1024 / :small AS commits_per_segment;
SELECT pg_current_wal_lsn() AS before \gset
INSERT INTO notes (body) VALUES (repeat('x', 1000));
SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), :'before') AS bytes_for_a_1000_byte_row;

volume_session.py

"""The WAL that rows cost: 10,000 inserts into a logged and an UNLOGGED table, read back with pg_walinspect.

Only sizes are printed, never a log sequence number of the run: each position stays in a psql variable and the
script shows the difference. The server never checkpoints by itself during the session (checkpoint_timeout is one day).
The UNLOGGED insert is read with the insert position, because its commit, the only record it writes, is not flushed
at once; one logged commit after it flushes the WAL so that pg_walinspect can read that record.
"""

from pg_session import Server

SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'",
            "checkpoint_timeout": "'1d'", "max_wal_size": "'4GB'", "bgwriter_lru_maxpages": 0}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("volume.sql")

$ psql -X -a -f volume.sql
-- The same 10,000 rows into a logged table and into an UNLOGGED one, and the WAL records each wrote.
CREATE EXTENSION pg_walinspect;
CREATE EXTENSION
CREATE TABLE orders (id bigint PRIMARY KEY, customer_id int NOT NULL, total numeric(12,2) NOT NULL);
CREATE TABLE
CREATE UNLOGGED TABLE orders_scratch (id bigint PRIMARY KEY, customer_id int NOT NULL, total numeric(12,2) NOT NULL);
CREATE TABLE
VACUUM;   -- sets the hint bits of the catalog rows made above, so that only the inserts show in the WAL below
VACUUM
CHECKPOINT;
CHECKPOINT
SELECT pg_current_wal_lsn() AS a \gset
INSERT INTO orders SELECT g, 1 + g % 500, 100 + g % 900 FROM generate_series(1, 10000) g;
INSERT 0 10000
SELECT pg_current_wal_lsn() AS b \gset
SELECT pg_size_pretty(pg_wal_lsn_diff(:'b', :'a')) AS logged_insert,
       round(pg_wal_lsn_diff(:'b', :'a') / 10000) AS bytes_per_row;
 logged_insert | bytes_per_row 
---------------+---------------
 1352 kB       |           138
(1 row)

SELECT resource_manager, record_type, count(*) AS records, sum(record_length) AS bytes
  FROM pg_get_wal_records_info(:'a', :'b') GROUP BY 1, 2 ORDER BY 3 DESC, 1;
 resource_manager | record_type  | records | bytes  
------------------+--------------+---------+--------
 Btree            | INSERT_LEAF  |    9973 | 638272
 Heap             | INSERT       |    9936 | 715392
 Heap             | INSERT+INIT  |      64 |   4608
 Btree            | SPLIT_R      |      27 |  20304
 Btree            | INSERT_UPPER |      26 |   1872
 Btree            | NEWROOT      |       2 |    212
 Transaction      | COMMIT       |       1 |     34
(7 rows)

-- The UNLOGGED insert is measured with the insert position: a commit that wrote no other WAL is not flushed at once.
SELECT pg_current_wal_insert_lsn() AS a \gset
INSERT INTO orders_scratch SELECT g, 1 + g % 500, 100 + g % 900 FROM generate_series(1, 10000) g;
INSERT 0 10000
SELECT pg_current_wal_insert_lsn() AS b \gset
INSERT INTO orders VALUES (0, 1, 100);   -- a logged commit flushes the WAL, and pg_walinspect reads only flushed WAL
INSERT 0 1
SELECT resource_manager, record_type, count(*) AS records, sum(record_length) AS bytes
  FROM pg_get_wal_records_info(:'a', :'b') GROUP BY 1, 2 ORDER BY 3 DESC, 1;
 resource_manager | record_type | records | bytes 
------------------+-------------+---------+-------
 Transaction      | COMMIT      |       1 |    34
(1 row)

volume.sql

-- The same 10,000 rows into a logged table and into an UNLOGGED one, and the WAL records each wrote.
CREATE EXTENSION pg_walinspect;
CREATE TABLE orders (id bigint PRIMARY KEY, customer_id int NOT NULL, total numeric(12,2) NOT NULL);
CREATE UNLOGGED TABLE orders_scratch (id bigint PRIMARY KEY, customer_id int NOT NULL, total numeric(12,2) NOT NULL);
VACUUM;   -- sets the hint bits of the catalog rows made above, so that only the inserts show in the WAL below
CHECKPOINT;
SELECT pg_current_wal_lsn() AS a \gset
INSERT INTO orders SELECT g, 1 + g % 500, 100 + g % 900 FROM generate_series(1, 10000) g;
SELECT pg_current_wal_lsn() AS b \gset
SELECT pg_size_pretty(pg_wal_lsn_diff(:'b', :'a')) AS logged_insert,
       round(pg_wal_lsn_diff(:'b', :'a') / 10000) AS bytes_per_row;
SELECT resource_manager, record_type, count(*) AS records, sum(record_length) AS bytes
  FROM pg_get_wal_records_info(:'a', :'b') GROUP BY 1, 2 ORDER BY 3 DESC, 1;
-- The UNLOGGED insert is measured with the insert position: a commit that wrote no other WAL is not flushed at once.
SELECT pg_current_wal_insert_lsn() AS a \gset
INSERT INTO orders_scratch SELECT g, 1 + g % 500, 100 + g % 900 FROM generate_series(1, 10000) g;
SELECT pg_current_wal_insert_lsn() AS b \gset
INSERT INTO orders VALUES (0, 1, 100);   -- a logged commit flushes the WAL, and pg_walinspect reads only flushed WAL
SELECT resource_manager, record_type, count(*) AS records, sum(record_length) AS bytes
  FROM pg_get_wal_records_info(:'a', :'b') GROUP BY 1, 2 ORDER BY 3 DESC, 1;

fpi_session.py

"""The full page image: the first change to a page after a checkpoint writes the whole page into the WAL.

One row is updated four times; pg_walinspect reads back the records each update wrote. Only sizes are printed.
"""

from pg_session import Server

SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'",
            "checkpoint_timeout": "'1d'", "max_wal_size": "'4GB'", "bgwriter_lru_maxpages": 0}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("fpi.sql")

$ psql -X -a -f fpi.sql
-- One account row updated four times: what each update wrote, read back from the WAL with pg_walinspect.
CREATE EXTENSION pg_walinspect;
CREATE EXTENSION
CREATE TABLE accounts (id bigint PRIMARY KEY, balance numeric(12,2) NOT NULL) WITH (fillfactor = 50);
CREATE TABLE
INSERT INTO accounts SELECT g, 1000 FROM generate_series(1, 1000) g;
INSERT 0 1000
SELECT count(*) AS accounts FROM accounts;   -- reads every row once, which sets their hint bits
 accounts 
----------
     1000
(1 row)

CHECKPOINT;
CHECKPOINT
SELECT pg_current_wal_lsn() AS s1 \gset
UPDATE accounts SET balance = balance + 1 WHERE id = 42;
UPDATE 1
SELECT pg_current_wal_lsn() AS e1 \gset
UPDATE accounts SET balance = balance + 1 WHERE id = 42;
UPDATE 1
SELECT pg_current_wal_lsn() AS e2 \gset
CHECKPOINT;
CHECKPOINT
SELECT pg_current_wal_lsn() AS s3 \gset
UPDATE accounts SET balance = balance + 1 WHERE id = 42;
UPDATE 1
SELECT pg_current_wal_lsn() AS e3 \gset
CHECKPOINT;
CHECKPOINT
SET wal_compression = pglz;
SET
SELECT pg_current_wal_lsn() AS s4 \gset
UPDATE accounts SET balance = balance + 1 WHERE id = 42;
UPDATE 1
SELECT pg_current_wal_lsn() AS e4 \gset
RESET wal_compression;
RESET
SELECT n, step, records, bytes, full_page_image_bytes
  FROM (VALUES (1, 'first update after a checkpoint', :'s1'::pg_lsn, :'e1'::pg_lsn),
               (2, 'second update, same page', :'e1'::pg_lsn, :'e2'::pg_lsn),
               (3, 'first update after the next checkpoint', :'s3'::pg_lsn, :'e3'::pg_lsn),
               (4, 'the same, with wal_compression = pglz', :'s4'::pg_lsn, :'e4'::pg_lsn)) AS u(n, step, s, e)
  CROSS JOIN LATERAL (SELECT count(*) AS records, sum(record_length) AS bytes,
                             coalesce(sum(fpi_length), 0) AS full_page_image_bytes
                        FROM pg_get_wal_records_info(u.s, u.e)) AS w
 ORDER BY n;
 n |                  step                  | records | bytes | full_page_image_bytes 
---+----------------------------------------+---------+-------+-----------------------
 1 | first update after a checkpoint        |       2 |  4215 |                  4116
 2 | second update, same page               |       3 |   162 |                     0
 3 | first update after the next checkpoint |       4 |  4335 |                  4120
 4 | the same, with wal_compression = pglz  |       4 |  1406 |                  1189
(4 rows)

SELECT resource_manager, record_type, count(*) AS records, sum(record_length) AS bytes,
       sum(fpi_length) AS full_page_image_bytes
  FROM pg_get_wal_records_info(:'s1', :'e1') GROUP BY 1, 2 ORDER BY 1, 2;
 resource_manager | record_type | records | bytes | full_page_image_bytes 
------------------+-------------+---------+-------+-----------------------
 Heap             | HOT_UPDATE  |       1 |  4181 |                  4116
 Transaction      | COMMIT      |       1 |    34 |                     0
(2 rows)

fpi.sql

-- One account row updated four times: what each update wrote, read back from the WAL with pg_walinspect.
CREATE EXTENSION pg_walinspect;
CREATE TABLE accounts (id bigint PRIMARY KEY, balance numeric(12,2) NOT NULL) WITH (fillfactor = 50);
INSERT INTO accounts SELECT g, 1000 FROM generate_series(1, 1000) g;
SELECT count(*) AS accounts FROM accounts;   -- reads every row once, which sets their hint bits
CHECKPOINT;
SELECT pg_current_wal_lsn() AS s1 \gset
UPDATE accounts SET balance = balance + 1 WHERE id = 42;
SELECT pg_current_wal_lsn() AS e1 \gset
UPDATE accounts SET balance = balance + 1 WHERE id = 42;
SELECT pg_current_wal_lsn() AS e2 \gset
CHECKPOINT;
SELECT pg_current_wal_lsn() AS s3 \gset
UPDATE accounts SET balance = balance + 1 WHERE id = 42;
SELECT pg_current_wal_lsn() AS e3 \gset
CHECKPOINT;
SET wal_compression = pglz;
SELECT pg_current_wal_lsn() AS s4 \gset
UPDATE accounts SET balance = balance + 1 WHERE id = 42;
SELECT pg_current_wal_lsn() AS e4 \gset
RESET wal_compression;
SELECT n, step, records, bytes, full_page_image_bytes
  FROM (VALUES (1, 'first update after a checkpoint', :'s1'::pg_lsn, :'e1'::pg_lsn),
               (2, 'second update, same page', :'e1'::pg_lsn, :'e2'::pg_lsn),
               (3, 'first update after the next checkpoint', :'s3'::pg_lsn, :'e3'::pg_lsn),
               (4, 'the same, with wal_compression = pglz', :'s4'::pg_lsn, :'e4'::pg_lsn)) AS u(n, step, s, e)
  CROSS JOIN LATERAL (SELECT count(*) AS records, sum(record_length) AS bytes,
                             coalesce(sum(fpi_length), 0) AS full_page_image_bytes
                        FROM pg_get_wal_records_info(u.s, u.e)) AS w
 ORDER BY n;
SELECT resource_manager, record_type, count(*) AS records, sum(record_length) AS bytes,
       sum(fpi_length) AS full_page_image_bytes
  FROM pg_get_wal_records_info(:'s1', :'e1') GROUP BY 1, 2 ORDER BY 1, 2;

durable_session.py

"""Two commits, two crashes: a commit made with synchronous_commit = off can be lost, one made with on cannot.

The wal_writer_delay of 10 s keeps the WAL writer from flushing the asynchronous commit between the client's
"committed" and the crash, so the loss shows on every run; with the default of 200 ms the window is real but short.
The crash is `pg_ctl -m immediate`.
"""

from pg_session import Server

SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'",
            "checkpoint_timeout": "'1d'", "wal_writer_delay": "'10s'", "wal_writer_flush_after": "'1GB'"}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("durable_setup.sql")
    pg.psql("durable_off.sql")
    pg.stop(mode="immediate", show=True)
    pg.start(show=True, note="after the crash")
    pg.psql("durable_after.sql")
    pg.psql("durable_on.sql")
    pg.stop(mode="immediate", show=True)
    pg.start(show=True, note="after the second crash")
    pg.psql("durable_after.sql")

$ psql -X -a -f durable_setup.sql
-- One payment, committed the normal way, and the settings that decide when a commit is safe.
CREATE TABLE payments (id bigint PRIMARY KEY, note text NOT NULL);
CREATE TABLE
INSERT INTO payments VALUES (1, 'committed with synchronous_commit = on');
INSERT 0 1
SELECT current_setting('synchronous_commit') AS synchronous_commit,
       current_setting('wal_writer_delay') AS wal_writer_delay,
       current_setting('fsync') AS fsync;
 synchronous_commit | wal_writer_delay | fsync 
--------------------+------------------+-------
 on                 | 10s              | on
(1 row)

$ psql -X -a -f durable_off.sql
-- The client is told "committed" for payment 2 without waiting for the WAL to reach the disk.
SET synchronous_commit = off;
SET
INSERT INTO payments VALUES (2, 'committed with synchronous_commit = off');
INSERT 0 1
SELECT id, note FROM payments ORDER BY id;
 id |                  note                   
----+-----------------------------------------
  1 | committed with synchronous_commit = on
  2 | committed with synchronous_commit = off
(2 rows)

$ pg_ctl -D data -m immediate -w -t 60 stop
waiting for server to shut down... done
server stopped

-- after the crash
$ pg_ctl -D data -l data.log -w -t 60 start
waiting for server to start... done
server started

$ psql -X -a -f durable_after.sql
SELECT id, note FROM payments ORDER BY id;
 id |                  note                  
----+----------------------------------------
  1 | committed with synchronous_commit = on
(1 row)

$ psql -X -a -f durable_on.sql
-- Payment 3 waits for the flush, as every commit does by default.
INSERT INTO payments VALUES (3, 'committed with synchronous_commit = on');
INSERT 0 1

$ pg_ctl -D data -m immediate -w -t 60 stop
waiting for server to shut down... done
server stopped

-- after the second crash
$ pg_ctl -D data -l data.log -w -t 60 start
waiting for server to start... done
server started

$ psql -X -a -f durable_after.sql
SELECT id, note FROM payments ORDER BY id;
 id |                  note                  
----+----------------------------------------
  1 | committed with synchronous_commit = on
  3 | committed with synchronous_commit = on
(2 rows)

durable_setup.sql

-- One payment, committed the normal way, and the settings that decide when a commit is safe.
CREATE TABLE payments (id bigint PRIMARY KEY, note text NOT NULL);
INSERT INTO payments VALUES (1, 'committed with synchronous_commit = on');
SELECT current_setting('synchronous_commit') AS synchronous_commit,
       current_setting('wal_writer_delay') AS wal_writer_delay,
       current_setting('fsync') AS fsync;

durable_off.sql

-- The client is told "committed" for payment 2 without waiting for the WAL to reach the disk.
SET synchronous_commit = off;
INSERT INTO payments VALUES (2, 'committed with synchronous_commit = off');
SELECT id, note FROM payments ORDER BY id;

durable_on.sql

-- Payment 3 waits for the flush, as every commit does by default.
INSERT INTO payments VALUES (3, 'committed with synchronous_commit = on');

durable_after.sql

SELECT id, note FROM payments ORDER BY id;

checksum_catch_session.py

"""Data checksums catch a damaged page: pg_checksums finds it offline, and the server refuses to read it.

One byte of the table's file is flipped while the server is stopped, as a failing disk might. The checksum values in
the messages are replaced by "N" because they depend on the page's contents and position in the WAL.
"""

import re

from pg_session import Server

SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'",
            "checkpoint_timeout": "'1d'", "log_line_prefix": "''"}
SUMMARY = [(r"(calculated checksum )\w+( but block contains )\w+", r"\1N\2N"),
           (r"(calculated checksum )\d+( but expected )\d+", r"\1N\2N")]

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("checksum_setup.sql", flags=("-a", "-t"))
    path = pg.query("SELECT pg_relation_filepath('accounts')")
    pg.stop()
    pg.shell("pg_checksums", "--check", "-D", "data", mask=SUMMARY, note="the server is stopped: check every page")
    with open(pg.base / "data" / path, "r+b") as table:
        table.seek(8000)
        byte = table.read(1)
        table.seek(8000)
        table.write(bytes([byte[0] ^ 0xFF]))
    print(f"-- one byte flipped at offset 8000 of data/{path}\n")
    pg.shell("pg_checksums", "--check", "-D", "data", status=1, mask=SUMMARY, note="the same check again")
    pg.start()
    pg.psql("checksum_read.sql", errors=1, status=0, mask=SUMMARY)
    pg.log_lines("data.log", "page verification failed", mask=SUMMARY, note="the server log")
    pg.wait_for("SELECT checksum_failures FROM pg_stat_database WHERE datname = 'postgres'", "1")
    pg.psql("checksum_failures.sql")

$ psql -X -a -t -f checksum_setup.sql
-- A table of 100 accounts, written to disk by a checkpoint.
CREATE TABLE accounts (id bigint PRIMARY KEY, owner text NOT NULL, balance numeric(12,2) NOT NULL);
CREATE TABLE
INSERT INTO accounts SELECT g, 'owner ' || g, 1000 FROM generate_series(1, 100) g;
INSERT 0 100
VACUUM accounts;
VACUUM
CHECKPOINT;
CHECKPOINT
SELECT pg_relation_filepath('accounts') AS file;
 base/5/16384

-- the server is stopped: check every page
$ pg_checksums --check -D data
Checksum operation completed
Files scanned:   956
Blocks scanned:  2928
Bad checksums:  0
Data checksum version: 1

-- one byte flipped at offset 8000 of data/base/5/16384

-- the same check again
$ pg_checksums --check -D data
pg_checksums: error: checksum verification failed in file "data/base/5/16384", block 0: calculated checksum N but block contains N
Checksum operation completed
Files scanned:   956
Blocks scanned:  2928
Bad checksums:  1
Data checksum version: 1

$ psql -X -a -f checksum_read.sql
-- Reading the damaged table: the server refuses the page instead of returning wrong rows.
SELECT count(*) FROM accounts;
psql:checksum_read.sql:2: ERROR:  invalid page in block 0 of relation "base/5/16384"

-- the server log:
LOG:  page verification failed, calculated checksum N but expected N

$ psql -X -a -f checksum_failures.sql
SELECT checksum_failures FROM pg_stat_database WHERE datname = 'postgres';
 checksum_failures 
-------------------
                 1
(1 row)

checksum_setup.sql

-- A table of 100 accounts, written to disk by a checkpoint.
CREATE TABLE accounts (id bigint PRIMARY KEY, owner text NOT NULL, balance numeric(12,2) NOT NULL);
INSERT INTO accounts SELECT g, 'owner ' || g, 1000 FROM generate_series(1, 100) g;
VACUUM accounts;
CHECKPOINT;
SELECT pg_relation_filepath('accounts') AS file;

checksum_read.sql

-- Reading the damaged table: the server refuses the page instead of returning wrong rows.
SELECT count(*) FROM accounts;

checksum_failures.sql

SELECT checksum_failures FROM pg_stat_database WHERE datname = 'postgres';

checksum_cost_session.py

"""The price of data checksums: the same SELECT on a cluster with checksums (the default) and one without.

Two throwaway clusters in turn. Only sizes are printed, never a log sequence number of the run. The second cluster is
then stopped, converted with pg_checksums --enable (an offline operation) and started again.
"""

from pg_session import Server

SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'",
            "checkpoint_timeout": "'1d'", "max_wal_size": "'4GB'", "bgwriter_lru_maxpages": 0}

for note, initdb in (("initdb with its defaults", []), ("initdb --no-data-checksums", ["--no-data-checksums"])):
    print(f"-- cluster made with {note}")
    with Server(settings=SETTINGS, initdb=initdb) as pg:
        pg.up()
        pg.psql("checksum_cost.sql")
        if initdb:
            pg.stop()
            pg.shell("pg_checksums", "--enable", "-D", "data", note="the server is stopped: turn checksums on")
            pg.start()
            pg.psql("checksum_show.sql")

-- cluster made with initdb with its defaults
$ psql -X -a -f checksum_cost.sql
-- Does a cluster with data checksums write more WAL? A SELECT after a checkpoint is the test. The insert position
-- counts WAL that is still in memory; the write position, pg_current_wal_lsn(), would miss it.
SHOW data_checksums;
 data_checksums 
----------------
 on
(1 row)

CREATE TABLE readings (id bigint PRIMARY KEY, v int NOT NULL) WITH (fillfactor = 50);
CREATE TABLE
VACUUM;   -- sets the hint bits of the catalog rows made above, so that only the SELECT shows in the WAL below
VACUUM
INSERT INTO readings SELECT g, g FROM generate_series(1, 10000) g;
INSERT 0 10000
CHECKPOINT;
CHECKPOINT
SELECT pg_current_wal_insert_lsn() AS a \gset
SELECT count(*) AS rows_read FROM readings;
 rows_read 
-----------
     10000
(1 row)

SELECT pg_current_wal_insert_lsn() AS b \gset
SELECT pg_wal_lsn_diff(:'b', :'a') AS wal_bytes_written_by_the_select;
 wal_bytes_written_by_the_select 
---------------------------------
                          450040
(1 row)

SELECT current_setting('data_checksums') = 'on' AS has_checksums \gset
\if :has_checksums
  CREATE EXTENSION pg_walinspect;
CREATE EXTENSION
  SELECT resource_manager, record_type, count(*) AS records, sum(fpi_length) AS full_page_image_bytes
    FROM pg_get_wal_records_info(:'a', :'b') GROUP BY 1, 2;
 resource_manager | record_type  | records | full_page_image_bytes 
------------------+--------------+---------+-----------------------
 XLOG             | FPI_FOR_HINT |     109 |                442616
(1 row)

\endif

-- cluster made with initdb --no-data-checksums
$ psql -X -a -f checksum_cost.sql
-- Does a cluster with data checksums write more WAL? A SELECT after a checkpoint is the test. The insert position
-- counts WAL that is still in memory; the write position, pg_current_wal_lsn(), would miss it.
SHOW data_checksums;
 data_checksums 
----------------
 off
(1 row)

CREATE TABLE readings (id bigint PRIMARY KEY, v int NOT NULL) WITH (fillfactor = 50);
CREATE TABLE
VACUUM;   -- sets the hint bits of the catalog rows made above, so that only the SELECT shows in the WAL below
VACUUM
INSERT INTO readings SELECT g, g FROM generate_series(1, 10000) g;
INSERT 0 10000
CHECKPOINT;
CHECKPOINT
SELECT pg_current_wal_insert_lsn() AS a \gset
SELECT count(*) AS rows_read FROM readings;
 rows_read 
-----------
     10000
(1 row)

SELECT pg_current_wal_insert_lsn() AS b \gset
SELECT pg_wal_lsn_diff(:'b', :'a') AS wal_bytes_written_by_the_select;
 wal_bytes_written_by_the_select 
---------------------------------
                               0
(1 row)

SELECT current_setting('data_checksums') = 'on' AS has_checksums \gset
\if :has_checksums
  CREATE EXTENSION pg_walinspect;
  SELECT resource_manager, record_type, count(*) AS records, sum(fpi_length) AS full_page_image_bytes
    FROM pg_get_wal_records_info(:'a', :'b') GROUP BY 1, 2;
\endif

-- the server is stopped: turn checksums on
$ pg_checksums --enable -D data
Checksum operation completed
Files scanned:   953
Blocks scanned:  3062
Files written:  784
Blocks written: 3060
pg_checksums: syncing data directory
pg_checksums: updating control file
Checksums enabled in cluster

$ psql -X -a -f checksum_show.sql
SHOW data_checksums;
 data_checksums 
----------------
 on
(1 row)

checksum_cost.sql

-- Does a cluster with data checksums write more WAL? A SELECT after a checkpoint is the test. The insert position
-- counts WAL that is still in memory; the write position, pg_current_wal_lsn(), would miss it.
SHOW data_checksums;
CREATE TABLE readings (id bigint PRIMARY KEY, v int NOT NULL) WITH (fillfactor = 50);
VACUUM;   -- sets the hint bits of the catalog rows made above, so that only the SELECT shows in the WAL below
INSERT INTO readings SELECT g, g FROM generate_series(1, 10000) g;
CHECKPOINT;
SELECT pg_current_wal_insert_lsn() AS a \gset
SELECT count(*) AS rows_read FROM readings;
SELECT pg_current_wal_insert_lsn() AS b \gset
SELECT pg_wal_lsn_diff(:'b', :'a') AS wal_bytes_written_by_the_select;
SELECT current_setting('data_checksums') = 'on' AS has_checksums \gset
\if :has_checksums
  CREATE EXTENSION pg_walinspect;
  SELECT resource_manager, record_type, count(*) AS records, sum(fpi_length) AS full_page_image_bytes
    FROM pg_get_wal_records_info(:'a', :'b') GROUP BY 1, 2;
\endif

checksum_show.sql

SHOW data_checksums;

checkpoint_session.py

"""Sizing max_wal_size: the same 130 MB of WAL under a small and under the default max_wal_size.

checkpoint_timeout is one day, so no checkpoint is due to time: every checkpoint of the session is one that the
amount of WAL asked for. The checkpoint log lines are shown without the "seconds apart" figure, which differs per run.
"""

import re

from pg_session import Server

BASE = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'",
        "checkpoint_timeout": "'1d'", "min_wal_size": "'32MB'", "log_line_prefix": "''"}

for size, expected in (("'96MB'", 2), ("'1GB'", 0)):
    print(f"-- a cluster with max_wal_size = {size.strip(chr(39))}")
    with Server(settings={**BASE, "max_wal_size": size}) as pg:
        pg.up()
        pg.psql("checkpoint_load.sql")
        if expected:
            pg.wait_for("SELECT num_done FROM pg_stat_checkpointer", str(expected))
            pg.log_lines("data.log", "checkpoint starting", count=expected, note="the server log")
        pg.psql("checkpoint_after.sql")

-- a cluster with max_wal_size = 96MB
$ psql -X -a -f checkpoint_load.sql
-- The distance between size-triggered checkpoints, from the two settings that decide it.
SELECT current_setting('max_wal_size') AS max_wal_size,
       current_setting('checkpoint_completion_target') AS completion_target,
       floor(pg_size_bytes(current_setting('max_wal_size')) / (1 + current_setting('checkpoint_completion_target')::numeric)
             / (16 * 1024 * 1024)) AS segments_between_checkpoints;
 max_wal_size | completion_target | segments_between_checkpoints 
--------------+-------------------+------------------------------
 96MB         | 0.9               |                            3
(1 row)

SELECT num_timed, num_requested, num_done FROM pg_stat_checkpointer;
 num_timed | num_requested | num_done 
-----------+---------------+----------
         0 |             0 |        0
(1 row)

-- A bulk load, stood in for by 130 WAL messages of about 1 MB each.
SELECT pg_current_wal_insert_lsn() AS a \gset
SELECT count(pg_logical_emit_message(false, 'filler', repeat('x', 1000000))) AS messages FROM generate_series(1, 130);
 messages 
----------
      130
(1 row)

SELECT pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_insert_lsn(), :'a')) AS wal_written;
 wal_written 
-------------
 124 MB
(1 row)

-- the server log:
LOG:  checkpoint starting: wal
LOG:  checkpoint starting: wal

$ psql -X -a -f checkpoint_after.sql
SELECT num_timed, num_requested, num_done FROM pg_stat_checkpointer;
 num_timed | num_requested | num_done 
-----------+---------------+----------
         0 |             2 |        2
(1 row)

-- a cluster with max_wal_size = 1GB
$ psql -X -a -f checkpoint_load.sql
-- The distance between size-triggered checkpoints, from the two settings that decide it.
SELECT current_setting('max_wal_size') AS max_wal_size,
       current_setting('checkpoint_completion_target') AS completion_target,
       floor(pg_size_bytes(current_setting('max_wal_size')) / (1 + current_setting('checkpoint_completion_target')::numeric)
             / (16 * 1024 * 1024)) AS segments_between_checkpoints;
 max_wal_size | completion_target | segments_between_checkpoints 
--------------+-------------------+------------------------------
 1GB          | 0.9               |                           33
(1 row)

SELECT num_timed, num_requested, num_done FROM pg_stat_checkpointer;
 num_timed | num_requested | num_done 
-----------+---------------+----------
         0 |             0 |        0
(1 row)

-- A bulk load, stood in for by 130 WAL messages of about 1 MB each.
SELECT pg_current_wal_insert_lsn() AS a \gset
SELECT count(pg_logical_emit_message(false, 'filler', repeat('x', 1000000))) AS messages FROM generate_series(1, 130);
 messages 
----------
      130
(1 row)

SELECT pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_insert_lsn(), :'a')) AS wal_written;
 wal_written 
-------------
 124 MB
(1 row)

$ psql -X -a -f checkpoint_after.sql
SELECT num_timed, num_requested, num_done FROM pg_stat_checkpointer;
 num_timed | num_requested | num_done 
-----------+---------------+----------
         0 |             0 |        0
(1 row)

checkpoint_load.sql

-- The distance between size-triggered checkpoints, from the two settings that decide it.
SELECT current_setting('max_wal_size') AS max_wal_size,
       current_setting('checkpoint_completion_target') AS completion_target,
       floor(pg_size_bytes(current_setting('max_wal_size')) / (1 + current_setting('checkpoint_completion_target')::numeric)
             / (16 * 1024 * 1024)) AS segments_between_checkpoints;
SELECT num_timed, num_requested, num_done FROM pg_stat_checkpointer;
-- A bulk load, stood in for by 130 WAL messages of about 1 MB each.
SELECT pg_current_wal_insert_lsn() AS a \gset
SELECT count(pg_logical_emit_message(false, 'filler', repeat('x', 1000000))) AS messages FROM generate_series(1, 130);
SELECT pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_insert_lsn(), :'a')) AS wal_written;

checkpoint_after.sql

SELECT num_timed, num_requested, num_done FROM pg_stat_checkpointer;