Index design: partial, expression, write cost

partial_session.py

"""Index design on PostgreSQL 18.6: partial.sql on the orders of setup.sql."""

from pg_session import Server

# The same plan on every run: autovacuum off, no parallel workers or JIT, and statistics from every row.
SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off",
            "default_statistics_target": 1000, "timezone": "'UTC'"}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("setup.sql", flags=("-q",))
    pg.psql("partial.sql", errors=0)

$ psql -X -q -f setup.sql

$ psql -X -a -f partial.sql
-- A full index on (customer_id, status) against a partial index that holds only the pending orders.
CREATE INDEX orders_customer_status ON orders (customer_id, status);
CREATE INDEX
CREATE INDEX orders_pending ON orders (customer_id) WHERE status = 'pending';
CREATE INDEX
SELECT relname, pg_size_pretty(pg_relation_size(oid)) AS size FROM pg_class
WHERE relname IN ('orders_customer_status', 'orders_pending') ORDER BY relname;
        relname         |  size  
------------------------+--------
 orders_customer_status | 816 kB
 orders_pending         | 16 kB
(2 rows)

-- The query repeats the index's condition, so the planner can prove that the partial index holds every match.
EXPLAIN (COSTS OFF) SELECT id, total FROM orders WHERE status = 'pending' AND customer_id = 42;
                QUERY PLAN                 
-------------------------------------------
 Index Scan using orders_pending on orders
   Index Cond: (customer_id = 42)
(2 rows)

-- A prepared statement planned for any status cannot use it: status could be 'shipped'.
PREPARE by_status(text) AS SELECT id, total FROM orders WHERE status = $1 AND customer_id = 42;
PREPARE
SET plan_cache_mode = force_generic_plan;
SET
EXPLAIN (COSTS OFF) EXECUTE by_status('pending');
                         QUERY PLAN                         
------------------------------------------------------------
 Bitmap Heap Scan on orders
   Recheck Cond: ((customer_id = 42) AND (status = $1))
   ->  Bitmap Index Scan on orders_customer_status
         Index Cond: ((customer_id = 42) AND (status = $1))
(4 rows)

RESET plan_cache_mode;
RESET

partial.sql

-- A full index on (customer_id, status) against a partial index that holds only the pending orders.
CREATE INDEX orders_customer_status ON orders (customer_id, status);
CREATE INDEX orders_pending ON orders (customer_id) WHERE status = 'pending';
SELECT relname, pg_size_pretty(pg_relation_size(oid)) AS size FROM pg_class
WHERE relname IN ('orders_customer_status', 'orders_pending') ORDER BY relname;
-- The query repeats the index's condition, so the planner can prove that the partial index holds every match.
EXPLAIN (COSTS OFF) SELECT id, total FROM orders WHERE status = 'pending' AND customer_id = 42;
-- A prepared statement planned for any status cannot use it: status could be 'shipped'.
PREPARE by_status(text) AS SELECT id, total FROM orders WHERE status = $1 AND customer_id = 42;
SET plan_cache_mode = force_generic_plan;
EXPLAIN (COSTS OFF) EXECUTE by_status('pending');
RESET plan_cache_mode;

setup.sql

-- The shop of this lesson: 100,000 orders, 1 in 100 of them pending, with e-mail addresses in mixed case.
CREATE TABLE orders (
  id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  customer_id int NOT NULL,
  status text NOT NULL,
  total numeric(10,2) NOT NULL,
  email text NOT NULL
);
INSERT INTO orders (customer_id, status, total, email)
SELECT 1 + n % 5000, CASE WHEN n % 100 = 0 THEN 'pending' ELSE 'shipped' END, (n % 900) + 99.5,
       'Customer' || (n % 5000) || '@Example.com'
FROM generate_series(1, 100000) AS n;
VACUUM ANALYZE orders;

pg_session.py

"""Throwaway PostgreSQL 18.6 servers for the examples of this lesson.

Every session creates a cluster of its own with initdb in a new folder inside the examples folder, starts it, runs
psql scripts and server programs on it, prints what they printed, and stops and deletes it again, also when something
fails. The server is the pinned PostgreSQL 18.6 from the mise installations next to the Python that runs this file.
It 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 it asks for a password (scram-sha-256) made up for the run and never printed.
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 four substitutions so that a record never shows the computer
it ran on and is the same on every run: the run's 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", and the dots pg_ctl prints while it waits
(one about every tenth of a second) as three. The sessions never print the port. 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


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"):
        self.settings = dict(settings or {})
        self.initdb_args = list(initdb)
        self.name = name
        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")
        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)
        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):
        folder = folder or self.name
        self._expect(["pg_ctl", "-D", folder, "-l", f"{folder}.log", "-w", "-t", "60", "start"], show=show)
        self.running.append((folder, port or self.port))

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

    def standby(self, folder="standby"):
        """A streaming standby of this server, made with pg_basebackup -R, started on a port of its own."""
        self._expect(["pg_basebackup", "-D", folder, "-R", "-X", "stream", "-c", "fast"])
        port = free_port()
        with open(self.base / folder / "postgresql.conf", "a") as conf:
            conf.write(f"port = {port}\n")
        self.start(folder=folder, port=port)
        for _ in range(300):
            if self.query("SELECT count(*) FROM pg_stat_replication WHERE state = 'streaming'") == "1":
                return port
            time.sleep(0.1)
        sys.exit("the standby did not start streaming")

    # ---- 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):
        code, out = self._call(args)
        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):
        """The run's folder as ".", the user as learner, pg_ctl's waiting dots as three (see the docstring)."""
        text = text.replace(str(BIN) + "/", "").replace(str(self.base) + "/", "").replace(str(self.base), ".")
        text = re.sub(r"(waiting for server to (?:start|shut down))\.+", r"\1...", text)
        return text.replace(f'"{OS_USER}"', '"learner"')

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

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

    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):
        """Runs a statement without printing it (used only to give a role the run's password)."""
        self.query(sql)

    def psql(self, script, errors=0, status=None, user=None, db=None, port=None, flags=("-a",), note=None):
        """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."""
        env = {**self.env, "PGPORT": str(port or self.port)}
        if user:
            env["PGUSER"] = user
        if db:
            env["PGDATABASE"] = db
        args = ["psql", "-X", *flags, "-f", script]
        code, out = self._call(args, env=env)
        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):
        """Runs one of PostgreSQL's programs and prints the command and its output."""
        return self._expect(list(args), show=True, status=status)

    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 _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)

expression_session.py

"""Index design on PostgreSQL 18.6: expression.sql on the orders of setup.sql."""

from pg_session import Server

# The same plan on every run: autovacuum off, no parallel workers or JIT, and statistics from every row.
SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off",
            "default_statistics_target": 1000, "timezone": "'UTC'"}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("setup.sql", flags=("-q",))
    pg.psql("expression.sql", errors=0)

$ psql -X -q -f setup.sql

$ psql -X -a -f expression.sql
-- Searching case-insensitively: without an index on the expression, every row is lowered and compared.
EXPLAIN SELECT id FROM orders WHERE lower(email) = 'customer42@example.com';
                        QUERY PLAN                         
-----------------------------------------------------------
 Seq Scan on orders  (cost=0.00..2531.00 rows=500 width=8)
   Filter: (lower(email) = 'customer42@example.com'::text)
(2 rows)

CREATE INDEX orders_email_lower ON orders (lower(email));
CREATE INDEX
ANALYZE orders;
ANALYZE
-- The index matches the expression, and ANALYZE now keeps statistics on lower(email) as well.
EXPLAIN SELECT id FROM orders WHERE lower(email) = 'customer42@example.com';
                                    QUERY PLAN                                    
----------------------------------------------------------------------------------
 Bitmap Heap Scan on orders  (cost=4.45..76.39 rows=20 width=8)
   Recheck Cond: (lower(email) = 'customer42@example.com'::text)
   ->  Bitmap Index Scan on orders_email_lower  (cost=0.00..4.44 rows=20 width=0)
         Index Cond: (lower(email) = 'customer42@example.com'::text)
(4 rows)

-- The plain column is a different expression: this query cannot use the index.
EXPLAIN (COSTS OFF) SELECT id FROM orders WHERE email = 'customer42@example.com';
                     QUERY PLAN                     
----------------------------------------------------
 Seq Scan on orders
   Filter: (email = 'customer42@example.com'::text)
(2 rows)

expression.sql

-- Searching case-insensitively: without an index on the expression, every row is lowered and compared.
EXPLAIN SELECT id FROM orders WHERE lower(email) = 'customer42@example.com';
CREATE INDEX orders_email_lower ON orders (lower(email));
ANALYZE orders;
-- The index matches the expression, and ANALYZE now keeps statistics on lower(email) as well.
EXPLAIN SELECT id FROM orders WHERE lower(email) = 'customer42@example.com';
-- The plain column is a different expression: this query cannot use the index.
EXPLAIN (COSTS OFF) SELECT id FROM orders WHERE email = 'customer42@example.com';

unique_session.py

"""Index design on PostgreSQL 18.6: unique.sql on the orders of setup.sql."""

from pg_session import Server

# The same plan on every run: autovacuum off, no parallel workers or JIT, and statistics from every row.
SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off",
            "default_statistics_target": 1000, "timezone": "'UTC'"}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("setup.sql", flags=("-q",))
    pg.psql("unique.sql", errors=1)

$ psql -X -q -f setup.sql

$ psql -X -a -f unique.sql
-- Any number of addresses per customer, but only one default address: a partial unique index.
CREATE TABLE addresses (
  id int GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  customer_id int NOT NULL,
  city text NOT NULL,
  is_default boolean NOT NULL
);
CREATE TABLE
CREATE UNIQUE INDEX one_default_address ON addresses (customer_id) WHERE is_default;
CREATE INDEX
INSERT INTO addresses (customer_id, city, is_default)
VALUES (1, 'Pune', true), (1, 'Mumbai', false), (1, 'Goa', false), (2, 'Kochi', true);
INSERT 0 4
-- A second default for customer 1 is refused.
INSERT INTO addresses (customer_id, city, is_default) VALUES (1, 'Nashik', true);
psql:unique.sql:12: ERROR:  duplicate key value violates unique constraint "one_default_address"
DETAIL:  Key (customer_id)=(1) already exists.
-- Moving the default is two changes in one statement order: clear the old one, then set the new one.
BEGIN;
BEGIN
UPDATE addresses SET is_default = false WHERE customer_id = 1 AND is_default;
UPDATE 1
UPDATE addresses SET is_default = true WHERE customer_id = 1 AND city = 'Mumbai';
UPDATE 1
COMMIT;
COMMIT
SELECT customer_id, city, is_default FROM addresses ORDER BY customer_id, id;
 customer_id |  city  | is_default 
-------------+--------+------------
           1 | Pune   | f
           1 | Mumbai | t
           1 | Goa    | f
           2 | Kochi  | t
(4 rows)

unique.sql

-- Any number of addresses per customer, but only one default address: a partial unique index.
CREATE TABLE addresses (
  id int GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  customer_id int NOT NULL,
  city text NOT NULL,
  is_default boolean NOT NULL
);
CREATE UNIQUE INDEX one_default_address ON addresses (customer_id) WHERE is_default;
INSERT INTO addresses (customer_id, city, is_default)
VALUES (1, 'Pune', true), (1, 'Mumbai', false), (1, 'Goa', false), (2, 'Kochi', true);
-- A second default for customer 1 is refused.
INSERT INTO addresses (customer_id, city, is_default) VALUES (1, 'Nashik', true);
-- Moving the default is two changes in one statement order: clear the old one, then set the new one.
BEGIN;
UPDATE addresses SET is_default = false WHERE customer_id = 1 AND is_default;
UPDATE addresses SET is_default = true WHERE customer_id = 1 AND city = 'Mumbai';
COMMIT;
SELECT customer_id, city, is_default FROM addresses ORDER BY customer_id, id;

writes_session.py

"""Index design on PostgreSQL 18.6: writes.sql on the orders of setup.sql."""

from pg_session import Server

# The same plan on every run: autovacuum off, no parallel workers or JIT, and statistics from every row.
SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off",
            "default_statistics_target": 1000, "timezone": "'UTC'"}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("setup.sql", flags=("-q",))
    pg.psql("writes.sql", errors=0)

$ psql -X -q -f setup.sql

$ psql -X -a -f writes.sql
-- 10,000 stock rows with room left on every page (fillfactor 85), and a quantity that changes all day.
CREATE TABLE stock (id int PRIMARY KEY, sku text NOT NULL, shelf int NOT NULL, supplier int NOT NULL,
                    qty int NOT NULL) WITH (fillfactor = 85);
CREATE TABLE
INSERT INTO stock SELECT n, 'SKU-' || n, n % 400, n % 60, 100 FROM generate_series(1, 10000) AS n;
INSERT 0 10000
VACUUM ANALYZE stock;
VACUUM
-- 1. Only the primary key: an update of qty changes no indexed column.
EXPLAIN (ANALYZE, WAL, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) UPDATE stock SET qty = qty - 1 WHERE id % 10 = 0;
                      QUERY PLAN                       
-------------------------------------------------------
 Update on stock (actual rows=0.00 loops=1)
   WAL: records=1000 bytes=71688
   ->  Seq Scan on stock (actual rows=1000.00 loops=1)
         Filter: ((id % 10) = 0)
         Rows Removed by Filter: 9000
(5 rows)

-- 2. Three more indexes, none on qty.
CREATE INDEX stock_sku ON stock (sku);
CREATE INDEX
CREATE INDEX stock_shelf ON stock (shelf);
CREATE INDEX
CREATE INDEX stock_supplier ON stock (supplier);
CREATE INDEX
EXPLAIN (ANALYZE, WAL, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) UPDATE stock SET qty = qty - 1 WHERE id % 10 = 1;
                      QUERY PLAN                       
-------------------------------------------------------
 Update on stock (actual rows=0.00 loops=1)
   WAL: records=1085 bytes=79554
   ->  Seq Scan on stock (actual rows=1000.00 loops=1)
         Filter: ((id % 10) = 1)
         Rows Removed by Filter: 9000
         WAL: records=85 bytes=8554
(6 rows)

-- 3. An index on qty itself.
CREATE INDEX stock_qty ON stock (qty);
CREATE INDEX
EXPLAIN (ANALYZE, WAL, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) UPDATE stock SET qty = qty - 1 WHERE id % 10 = 2;
                      QUERY PLAN                       
-------------------------------------------------------
 Update on stock (actual rows=0.00 loops=1)
   WAL: records=6260 bytes=461184
   ->  Seq Scan on stock (actual rows=1000.00 loops=1)
         Filter: ((id % 10) = 2)
         Rows Removed by Filter: 9000
         WAL: records=85 bytes=8554
(6 rows)

SELECT pg_stat_force_next_flush();
 pg_stat_force_next_flush 
--------------------------
 
(1 row)

SELECT n_tup_upd AS updates, n_tup_hot_upd AS hot_updates FROM pg_stat_user_tables WHERE relname = 'stock';
 updates | hot_updates 
---------+-------------
    3000 |        2000
(1 row)

writes.sql

-- 10,000 stock rows with room left on every page (fillfactor 85), and a quantity that changes all day.
CREATE TABLE stock (id int PRIMARY KEY, sku text NOT NULL, shelf int NOT NULL, supplier int NOT NULL,
                    qty int NOT NULL) WITH (fillfactor = 85);
INSERT INTO stock SELECT n, 'SKU-' || n, n % 400, n % 60, 100 FROM generate_series(1, 10000) AS n;
VACUUM ANALYZE stock;
-- 1. Only the primary key: an update of qty changes no indexed column.
EXPLAIN (ANALYZE, WAL, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) UPDATE stock SET qty = qty - 1 WHERE id % 10 = 0;
-- 2. Three more indexes, none on qty.
CREATE INDEX stock_sku ON stock (sku);
CREATE INDEX stock_shelf ON stock (shelf);
CREATE INDEX stock_supplier ON stock (supplier);
EXPLAIN (ANALYZE, WAL, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) UPDATE stock SET qty = qty - 1 WHERE id % 10 = 1;
-- 3. An index on qty itself.
CREATE INDEX stock_qty ON stock (qty);
EXPLAIN (ANALYZE, WAL, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) UPDATE stock SET qty = qty - 1 WHERE id % 10 = 2;
SELECT pg_stat_force_next_flush();
SELECT n_tup_upd AS updates, n_tup_hot_upd AS hot_updates FROM pg_stat_user_tables WHERE relname = 'stock';

find_session.py

"""Index design on PostgreSQL 18.6: find.sql, unused, twin and invalid indexes of orders."""

from pg_session import Server

# The same plan on every run: autovacuum off, no parallel workers or JIT, and statistics from every row.
SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off",
            "default_statistics_target": 1000, "timezone": "'UTC'",
            # one process builds each index, so the failed build always reports the same duplicate
            "max_parallel_maintenance_workers": 0}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("setup.sql", flags=("-q",))
    pg.psql("find.sql", errors=1)

$ psql -X -q -f setup.sql

$ psql -X -a -f find.sql
-- Five indexes on orders: one used, one never used, one that duplicates another, and one left by a failed build.
CREATE INDEX orders_customer ON orders (customer_id);
CREATE INDEX
CREATE INDEX orders_customer_again ON orders (customer_id);
CREATE INDEX
CREATE INDEX orders_total ON orders (total);
CREATE INDEX
CREATE UNIQUE INDEX CONCURRENTLY orders_email_key ON orders (email);
psql:find.sql:5: ERROR:  could not create unique index "orders_email_key"
DETAIL:  Key (email)=(Customer1@Example.com) is duplicated.
SELECT count(*) FROM orders WHERE customer_id = 42;
 count 
-------
    20
(1 row)

SELECT pg_stat_force_next_flush();
 pg_stat_force_next_flush 
--------------------------
 
(1 row)

-- One catalog query: scans since the statistics were reset, duplicates by definition, and validity.
SELECT s.indexrelname AS index, s.idx_scan AS scans, i.indisvalid AS valid, i.indisunique AS is_unique,
       (SELECT string_agg(d.indexrelid::regclass::text, ', ') FROM pg_index AS d
        WHERE d.indrelid = i.indrelid AND d.indexrelid <> i.indexrelid AND d.indkey = i.indkey
          AND d.indclass = i.indclass AND d.indcollation = i.indcollation AND d.indoption = i.indoption
          AND coalesce(d.indpred::text, '') = coalesce(i.indpred::text, '')
          AND coalesce(d.indexprs::text, '') = coalesce(i.indexprs::text, '')) AS same_as,
       pg_size_pretty(pg_relation_size(s.indexrelid)) AS size
FROM pg_stat_user_indexes AS s JOIN pg_index AS i ON i.indexrelid = s.indexrelid
WHERE s.relname = 'orders' ORDER BY s.indexrelname;
         index         | scans | valid | is_unique |        same_as        |  size   
-----------------------+-------+-------+-----------+-----------------------+---------
 orders_customer       |     0 | t     | f         | orders_customer_again | 776 kB
 orders_customer_again |     1 | t     | f         | orders_customer       | 776 kB
 orders_email_key      |     0 | f     | t         |                       | 0 bytes
 orders_pkey           |     0 | t     | t         |                       | 2208 kB
 orders_total          |     0 | t     | f         |                       | 2224 kB
(5 rows)

find.sql

-- Five indexes on orders: one used, one never used, one that duplicates another, and one left by a failed build.
CREATE INDEX orders_customer ON orders (customer_id);
CREATE INDEX orders_customer_again ON orders (customer_id);
CREATE INDEX orders_total ON orders (total);
CREATE UNIQUE INDEX CONCURRENTLY orders_email_key ON orders (email);
SELECT count(*) FROM orders WHERE customer_id = 42;
SELECT pg_stat_force_next_flush();
-- One catalog query: scans since the statistics were reset, duplicates by definition, and validity.
SELECT s.indexrelname AS index, s.idx_scan AS scans, i.indisvalid AS valid, i.indisunique AS is_unique,
       (SELECT string_agg(d.indexrelid::regclass::text, ', ') FROM pg_index AS d
        WHERE d.indrelid = i.indrelid AND d.indexrelid <> i.indexrelid AND d.indkey = i.indkey
          AND d.indclass = i.indclass AND d.indcollation = i.indcollation AND d.indoption = i.indoption
          AND coalesce(d.indpred::text, '') = coalesce(i.indpred::text, '')
          AND coalesce(d.indexprs::text, '') = coalesce(i.indexprs::text, '')) AS same_as,
       pg_size_pretty(pg_relation_size(s.indexrelid)) AS size
FROM pg_stat_user_indexes AS s JOIN pg_index AS i ON i.indexrelid = s.indexrelid
WHERE s.relname = 'orders' ORDER BY s.indexrelname;