Reading EXPLAIN in MySQL: plans line by line

access_types_session.py

"""Six EXPLAIN plans, one per access type in the type column, against the lab's 50,000 orders and 40,000 payments.

Best to worst: const (primary-key equality), eq_ref (a unique-key lookup inside a join), ref (an equality on a
non-unique index), range (an equality then a half-open range), index (a full index scan), ALL (a table scan). The
same `rows` and `filtered` values appear in two runs because the server reads 300 pages of index statistics instead
of the default 20.

A throwaway MySQL 8.4.11 server (mysql_session.py) loads lab_schema.sql quietly, then runs access_types.sql through
the mysql client.
"""

from mysql_session import Server

with Server(options=["--innodb-stats-persistent-sample-pages=300"]) as my:
    my.up()
    my.sql("CREATE DATABASE shop", flags=())
    my.mysql("lab_schema.sql", db="shop", flags=("-t", "--skip-comments"))
    my.mysql("access_types.sql")

$ mysql -e "CREATE DATABASE shop"

$ mysql -t --skip-comments shop < lab_schema.sql
+----------------+---------+----------+----------+
| Table          | Op      | Msg_type | Msg_text |
+----------------+---------+----------+----------+
| shop.customers | analyze | status   | OK       |
| shop.orders    | analyze | status   | OK       |
| shop.payments  | analyze | status   | OK       |
+----------------+---------+----------+----------+

$ mysql -t -vvv --skip-comments < access_types.sql
--------------
EXPLAIN SELECT id FROM orders WHERE id = 42
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: orders
   partitions: NULL
         type: const
possible_keys: PRIMARY
          key: PRIMARY
      key_len: 4
          ref: const
         rows: 1
     filtered: 100.00
        Extra: Using index
1 row in set, 1 warning

--------------
EXPLAIN SELECT COUNT(*) FROM orders o JOIN payments p ON p.order_id = o.id WHERE o.customer_id = 42
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: p
   partitions: NULL
         type: ALL
possible_keys: NULL
          key: NULL
      key_len: NULL
          ref: NULL
         rows: 40000
     filtered: 100.00
        Extra: NULL
*************************** 2. row ***************************
           id: 1
  select_type: SIMPLE
        table: o
   partitions: NULL
         type: eq_ref
possible_keys: PRIMARY,customer_created
          key: PRIMARY
      key_len: 4
          ref: shop.p.order_id
         rows: 1
     filtered: 5.00
        Extra: Using where
2 rows in set, 1 warning

--------------
EXPLAIN SELECT id, created_at FROM orders WHERE customer_id = 42
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: orders
   partitions: NULL
         type: ref
possible_keys: customer_created
          key: customer_created
      key_len: 4
          ref: const
         rows: 42
     filtered: 100.00
        Extra: Using index
1 row in set, 1 warning

--------------
EXPLAIN SELECT id FROM orders WHERE customer_id = 42 AND created_at >= '2026-03-01'
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: orders
   partitions: NULL
         type: range
possible_keys: customer_created
          key: customer_created
      key_len: 9
          ref: NULL
         rows: 28
     filtered: 100.00
        Extra: Using where; Using index
1 row in set, 1 warning

--------------
EXPLAIN SELECT MAX(created_at) FROM orders
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: orders
   partitions: NULL
         type: index
possible_keys: NULL
          key: customer_created
      key_len: 9
          ref: NULL
         rows: 50000
     filtered: 100.00
        Extra: Using index
1 row in set, 1 warning

--------------
EXPLAIN SELECT COUNT(*) FROM orders WHERE note = 'order note 7'
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: orders
   partitions: NULL
         type: ALL
possible_keys: NULL
          key: NULL
      key_len: NULL
          ref: NULL
         rows: 50000
     filtered: 10.00
        Extra: Using where
1 row in set, 1 warning

Bye

access_types.sql

-- Six EXPLAIN plans against the lab's orders and payments tables, each one producing a different access type in the
-- type column. Best to worst: const, eq_ref, ref, range, index, ALL. The lab's indexes are PRIMARY (id) on all three
-- tables and customer_created (customer_id, created_at) on orders.
USE shop;

-- const: a primary-key equality finds at most one row, so the plan reads one entry of the clustered index.
EXPLAIN SELECT id FROM orders WHERE id = 42\G

-- eq_ref: in a join, each row of the driving table reaches at most one row of this one through a unique key. Here p
-- has no index on order_id, so it scans; its id is a value that o reads through its PRIMARY key, one row at a time.
EXPLAIN SELECT COUNT(*) FROM orders o JOIN payments p ON p.order_id = o.id WHERE o.customer_id = 42\G

-- ref: a non-unique index equality returns every row with the given value. customer_created leads with customer_id,
-- so the seek lands on the first entry with customer_id = 42 and reads until the next value starts.
EXPLAIN SELECT id, created_at FROM orders WHERE customer_id = 42\G

-- range: an equality on the first column of the index, then a half-open range on the second. The seek narrows on
-- both; key_len rises from 4 (customer_id alone) to 9 (customer_id and created_at).
EXPLAIN SELECT id FROM orders WHERE customer_id = 42 AND created_at >= '2026-03-01'\G

-- index: the plan reads every entry of the index because the index holds every column the query needs. Also known
-- as a full index scan, read as an alternative to the table scan below.
EXPLAIN SELECT MAX(created_at) FROM orders\G

-- ALL: no index serves this filter, so the plan reads every row of the table and the server tests the WHERE on each.
EXPLAIN SELECT COUNT(*) FROM orders WHERE note = 'order note 7'\G

lab_schema.sql

-- The lesson lab: 1,200 customers, 50,000 orders and 40,000 payments, built without RAND()
-- so that every run on every machine holds exactly the same rows.
SET SESSION cte_max_recursion_depth = 100000;

CREATE TABLE customers (
  id            INT UNSIGNED NOT NULL AUTO_INCREMENT,
  name          VARCHAR(60)  NOT NULL,
  city          VARCHAR(40)  NOT NULL,
  signed_up_on  DATE         NOT NULL,
  PRIMARY KEY (id),
  KEY city_signed (city, signed_up_on)
) ENGINE=InnoDB;

CREATE TABLE orders (
  id           INT UNSIGNED     NOT NULL AUTO_INCREMENT,
  customer_id  INT UNSIGNED     NOT NULL,
  created_at   DATETIME         NOT NULL,
  status       TINYINT UNSIGNED NOT NULL,
  total_paise  INT UNSIGNED     NOT NULL,
  ship_city    VARCHAR(40)      NOT NULL,
  note         VARCHAR(120)     NOT NULL DEFAULT '',
  PRIMARY KEY (id),
  KEY customer_created (customer_id, created_at)
) ENGINE=InnoDB;

-- payments has no index on order_id: an equi-join to orders has no index to use.
CREATE TABLE payments (
  id            INT UNSIGNED NOT NULL AUTO_INCREMENT,
  order_id      INT UNSIGNED NOT NULL,
  amount_paise  INT UNSIGNED NOT NULL,
  method        VARCHAR(12)  NOT NULL,
  PRIMARY KEY (id)
) ENGINE=InnoDB;

-- 1,200 customers in 8 cities (13 and 8 are coprime, so the cities cycle evenly).
INSERT INTO customers (name, city, signed_up_on)
WITH RECURSIVE seq (n) AS (
  SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 1200
)
SELECT CONCAT('Customer ', LPAD(n, 4, '0')),
       ELT(1 + (n * 13) MOD 8, 'Pune', 'Mumbai', 'Bengaluru', 'Chennai', 'Hyderabad', 'Jaipur', 'Kolkata', 'Kochi'),
       DATE_ADD('2025-01-01', INTERVAL (n * 7) MOD 365 DAY)
FROM seq;

-- 50,000 orders over the first 180 days of 2026. customer_id spreads over all 1,200 customers
-- (7919 is prime, so n * 7919 MOD 1200 cycles); total_paise is skewed, with one order in fifty large.
INSERT INTO orders (customer_id, created_at, status, total_paise, ship_city, note)
WITH RECURSIVE seq (n) AS (
  SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 50000
)
SELECT 1 + (n * 7919) MOD 1200,
       DATE_ADD(DATE_ADD('2026-01-01 00:00:00', INTERVAL (n * 311) MOD 180 DAY),
                INTERVAL (n * 1013) MOD 86400 SECOND),
       ELT(1 + (n * 3) MOD 4, 1, 2, 2, 3),
       CASE WHEN n MOD 50 = 0 THEN 400000 + (n * 97) MOD 200000
            ELSE 5000 + (n * 89) MOD 95000 END,
       ELT(1 + (n * 17) MOD 8, 'Pune', 'Mumbai', 'Bengaluru', 'Chennai', 'Hyderabad', 'Jaipur', 'Kolkata', 'Kochi'),
       CONCAT('order note ', n MOD 997)
FROM seq;

-- 40,000 payments: the orders whose id is not a multiple of five are paid once.
INSERT INTO payments (order_id, amount_paise, method)
SELECT id, total_paise, ELT(1 + (id * 7) MOD 4, 'upi', 'card', 'netbanking', 'wallet')
FROM orders
WHERE id MOD 5 <> 0;

ANALYZE TABLE customers, orders, payments;

mysql_session.py

"""Throwaway MySQL 8.4.11 servers for the examples of this lesson.

Every session creates a server of its own: `mysqld --initialize-insecure` makes a new data directory inside a new
folder in the examples folder, the server starts on it, the session runs `mysql` client scripts and MySQL's programs
against it and prints what they printed, and then the server is stopped and the folder deleted, also when something
fails. The server is the pinned MySQL 8.4.11 from the mise installations next to the Python that runs this file;
Server(basedir=...) runs the programs of another unpacked MySQL tarball instead, for sessions that compare versions.

Safety: the server listens only on 127.0.0.1, an address no other computer can reach, on a port the operating system
reports as unused; the X Protocol listener is off (--mysqlx=OFF); its Unix socket, temporary files, error log and
import/export folder all stay inside the run folder. Right after the first start the session gives root a password made
up for the run (never printed) and connects only with a client option file that holds it, so no password ever appears
on a command line. Every program starts with --no-defaults or --defaults-file, so no option file of the computer
(/etc/my.cnf, ~/.my.cnf) can change a result, and HOME points at the run folder, so ~/.mylogin.cnf is not read either.
A printed mysqld command shows the options that matter to the lesson; the session also passes, on the command line,
the folders it keeps inside the run folder (--basedir, --tmpdir, --log-error, --pid-file, --secure-file-priv) and
the port, which is why performance_schema.variables_info reports those variables as COMMAND_LINE.
A session reads only this folder and the MySQL and Python programs it runs, and connects to nothing but 127.0.0.1.

The output is exactly what the programs printed, with these substitutions so that a record never shows the computer it
ran on and is the same on every run: the run folder is shown as ".", the folder of MySQL's programs is left out of
their paths and MySQL's own folder is shown as <basedir>, the computer's name is shown as "lab-host" and the operating-system user's name as "learner", the time a
statement took (the client's "(0.01 sec)") is left out, the time stamp SET PERSIST writes into mysqld-auto.cnf is
shown as <when it was set>, error-log lines lose their timestamp and thread number, and a session can name more values
that change between runs with my.mask(pattern, replacement), which then prints in their place. A client table whose cells
change is drawn again so that its borders still line up.
The sessions never print the password, and the run's port prints as <port>. A session stops with status 1 when a command that should have worked
fails, or when a script 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

MYSQL_VERSION = "8.4.11"
# .../installs/python/3.14.8/bin/python3 -> .../installs
INSTALLS = Path(sys.executable).resolve().parents[3]
PINNED = INSTALLS / "mysql" / MYSQL_VERSION
OS_USER = pwd.getpwuid(os.geteuid()).pw_name
HOST = socket.gethostname().split(".")[0]
# 2026-10-17T16:50:01.123456Z 0 [ERROR] ... -> [ERROR] ...
LOG_PREFIX = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d+Z \d+ ", re.M)


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 server: initialise, start; stopped and deleted when the with block ends.

    options: extra mysqld options for every start (for example "--sql-mode=" or "--character-set-server=latin1").
    basedir: the folder of an unpacked MySQL tarball to run instead of the pinned 8.4.11.
    masks: (pattern, replacement) pairs, as for mask().
    """

    def __init__(self, options=(), redo=None, basedir=None, masks=()):
        self.options = [str(o) for o in options]
        self.redo = redo
        self.basedir = Path(basedir).resolve() if basedir else PINNED
        self.bin = self.basedir / "bin"
        self.masks = []
        for pattern, replacement in masks:
            self.mask(pattern, replacement)
        self.proc = None

    def mask(self, pattern, replacement):
        """From now on every printed line shows text matching the regular expression as the replacement: for a value
        that changes from run to run (an uptime, a time stamp), so that the recorded output stays the same."""
        self.masks.append((re.compile(pattern), replacement))

    # ---- the folder ------------------------------------------------------------------------------------------------
    def __enter__(self):
        if not (self.bin / "mysqld").exists():
            sys.exit(f"no MySQL server program in {self.basedir}" if self.basedir != PINNED
                     else f"MySQL {MYSQL_VERSION} was not found next to this Python")
        self.base = Path(tempfile.mkdtemp(prefix="my", dir=".")).resolve()
        for sub in ("tmp", "files"):
            (self.base / sub).mkdir(mode=0o700)
        self.port = free_port()
        self.password = secrets.token_urlsafe(24)
        self.client_cnf = self.base / "client.cnf"
        self._write_client_cnf(password=None)
        self.env = {key: value for key, value in os.environ.items()
                    if not key.startswith("MYSQL") and key not in ("TZ", "HOME", "LANG", "LC_ALL")}
        self.env.update({"HOME": str(self.base), "TZ": "UTC", "LANG": "en_US.UTF-8",
                         "MYSQL_TEST_LOGIN_FILE": str(self.base / "no-login-paths.cnf")})
        for pattern in ("*.sql", "*.cnf", "*.csv", "*.txt"):
            for f in Path(".").glob(pattern):
                shutil.copy(f, self.base / f.name)
        return self

    def __exit__(self, *exc):
        if self.proc and self.proc.poll() is None:
            self.stop(quiet=True)
        shutil.rmtree(self.base, ignore_errors=True)

    def _write_client_cnf(self, password):
        lines = ["[client]", "user=root", "host=127.0.0.1", f"port={self.port}", "protocol=TCP"]
        if password:
            lines.append(f"password={password}")
        self.client_cnf.write_text("\n".join(lines) + "\n")
        self.client_cnf.chmod(0o600)

    def server_options(self):
        b = self.base
        return ["--no-defaults", f"--basedir={self.basedir}", f"--datadir={b / 'data'}", f"--tmpdir={b / 'tmp'}",
                f"--log-error={b / 'mysqld.err'}", f"--pid-file={b / 'mysqld.pid'}",
                f"--secure-file-priv={b / 'files'}",
                *([f"--innodb-redo-log-capacity={self.redo}"] if self.redo else [])]

    # ---- the server ------------------------------------------------------------------------------------------------
    def initialize(self, show=False):
        """mysqld --initialize-insecure: a new data directory whose root account has no password yet."""
        args = ["mysqld", *self.server_options(), "--initialize-insecure"]
        code, out = self._call(args)
        if show:
            _show(["mysqld", "--no-defaults", "--initialize-insecure", "--datadir=./data"], out)
        if code != 0:
            print(self.log())
            sys.exit(f"mysqld --initialize-insecure ended with status {code}")

    def start(self, extra=(), defaults_file=None, expect_fail=False, show=False):
        """Starts mysqld on 127.0.0.1 and waits until it answers. With expect_fail, prints the error-log lines of a
        start that must fail, and returns them."""
        log = self.base / "mysqld.err"
        before = log.read_text() if log.exists() else ""
        opts = self.server_options()
        if defaults_file:
            opts[0] = f"--defaults-file={self.base / defaults_file}"
        args = [str(self.bin / "mysqld"), *opts, "--bind-address=127.0.0.1", f"--port={self.port}",
                "--socket=mysql.sock", "--mysqlx=OFF", *self.options, *[str(e) for e in extra]]
        if show:
            shown = ["mysqld", opts[0].replace(str(self.base) + "/", ""), "--datadir=./data", "--bind-address=127.0.0.1",
                     "--port=$PORT", "--socket=mysql.sock", "--mysqlx=OFF", *self.options, *extra]
            print("$ " + " ".join(_arg(a) if a != "--port=$PORT" else a for a in shown) + " &")
            print()
        with open(self.base / "mysqld.out", "a") as out:
            self.proc = subprocess.Popen(args, stdout=out, stderr=subprocess.STDOUT, cwd=self.base, env=self.env)
        for _ in range(600):
            if self.proc.poll() is not None:
                break
            if self._ping():
                break
            time.sleep(0.1)
        if self.proc.poll() is not None:
            new = (log.read_text() if log.exists() else "")[len(before):]
            lines = [l for l in self.clean(LOG_PREFIX.sub("", new)).splitlines() if "[ERROR]" in l]
            if expect_fail:
                print("\n".join(lines))
                print(f"(mysqld ended with status {self.proc.returncode})")
                print()
                self.proc = None
                return lines
            print("\n".join(lines))
            sys.exit(f"mysqld ended with status {self.proc.returncode} while starting")
        if expect_fail:
            sys.exit("mysqld started, but the session expected it to fail")
        if not self._ping():
            sys.exit("mysqld did not answer within 60 seconds")
        return []

    def _ping(self):
        code, _ = self._call(["mysqladmin", f"--defaults-file={self.client_cnf}", "--connect-timeout=2", "ping"])
        return code == 0

    def up(self, extra=(), defaults_file=None, show=False):
        """Initialise, start (with start()'s options), and give root the run's password, quietly."""
        self.initialize()
        self.start(extra=extra, defaults_file=defaults_file, show=show)
        return self.login()

    def login(self):
        """Gives root the run's password (never printed) on a server started from a new data directory, and makes
        every later client of the session use it."""
        self.silent(f"ALTER USER 'root'@'localhost' IDENTIFIED BY '{self.password}'")
        self._write_client_cnf(self.password)
        return self

    def stop(self, quiet=False):
        """A clean shutdown (mysqladmin shutdown), then waits until the process has ended."""
        code, out = self._call(["mysqladmin", f"--defaults-file={self.client_cnf}", "shutdown"])
        if code != 0 and not quiet:
            sys.exit(f"mysqladmin shutdown failed: {out}")
        try:
            self.proc.wait(timeout=120)
        except subprocess.TimeoutExpired:
            self.proc.terminate()
            self.proc.wait(timeout=60)
        self.proc = None

    def restart(self, extra=(), defaults_file=None, show=False):
        self.stop()
        self.start(extra=extra, defaults_file=defaults_file, show=show)

    # ---- running things --------------------------------------------------------------------------------------------
    def _call(self, args, stdin=None, merge=True):
        """Runs one of MySQL's programs in the run folder: (status, cleaned output). merge=False keeps only what it
        wrote to standard output (its warnings on standard error are dropped)."""
        args = [str(a) for a in args]
        done = subprocess.run([str(self.bin / args[0]), *args[1:]], stdout=subprocess.PIPE,
                              stderr=subprocess.STDOUT if merge else subprocess.DEVNULL,
                              text=True, cwd=self.base, env=self.env, input=stdin)
        return done.returncode, self.clean(done.stdout)

    def output(self, *args, connect=False, status=0):
        """Runs one of MySQL's programs without printing anything and returns its standard output, for example the
        SQL that mysql_tzinfo_to_sql writes, which the session saves with write_file(..., show=False) and runs next."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run, merge=False)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def clean(self, text):
        """The substitutions of the docstring. A table of the client whose cells change is drawn again with the
        new column widths, so its borders still line up."""
        lines, out, i = text.split("\n"), [], 0
        while i < len(lines):
            if lines[i].startswith("+-"):
                j = i
                while j < len(lines) and lines[j][:1] in ("+", "|"):
                    j += 1
                block = lines[i:j]
                done = [self._clean(l) for l in block]
                out.extend(done if done == block else _redraw(block, self._clean))
                i = j
            else:
                out.append(self._clean(lines[i]))
                i += 1
        return "\n".join(out)

    def _clean(self, text):
        text = text.replace(str(self.bin) + "/", "").replace(str(self.basedir), "<basedir>")
        text = text.replace(str(self.base) + "/", "./").replace(str(self.base), ".")
        text = text.replace("\x07", "")   # the mysql client rings the terminal bell on an error; drop it
        text = re.sub(r" \(\d+\.\d{2,3} sec\)", "", text)  # 8.4 prints 2 decimals, 9.7 and 26.7 print 3
        text = re.sub(r'"Timestamp": \d+', '"Timestamp": <when it was set>', text)
        text = re.sub(rf"\b{re.escape(HOST)}\b", "lab-host", text)
        text = re.sub(rf"\b{re.escape(OS_USER)}\b", "learner", text)
        text = re.sub(rf"(?<![\d.]){self.port}(?![\d.])", "<port>", text)
        for pattern, replacement in self.masks:
            text = pattern.sub(replacement, text)
        return text

    def log(self, pattern=None):
        """The error log without timestamps and thread numbers, optionally only the lines matching a pattern."""
        text = self.clean(LOG_PREFIX.sub("", (self.base / "mysqld.err").read_text()))
        if pattern:
            text = "\n".join(l for l in text.splitlines() if re.search(pattern, l))
        return text

    def show_log(self, pattern, note=None):
        """Prints the error-log lines that match a pattern, as grep would."""
        print(f"$ grep -E {_arg(pattern)} mysqld.err" + (f"   # {note}" if note else ""))
        print(self.log(pattern))
        print()

    def query(self, sql, db=None):
        """One value from one query, unprinted (the session uses it to wait or to decide)."""
        args = ["mysql", f"--defaults-file={self.client_cnf}", "-N", "-B", *([db] if db else [])]
        code, out = self._call(args, stdin=sql)
        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 for the run's password)."""
        self.query(sql)

    def mysql(self, script, errors=0, status=None, db=None, flags=("-t", "-vvv", "--skip-comments"), note=None, force=None):
        """Runs a client script from standard input and prints the command and its output. `errors` is how many
        ERROR lines the script must print; with errors > 0 the client runs on past an error (--force) and then ends
        with status 0, without --force it stops at the first error with status 1. The default flags print each
        statement (-vvv) as a table (-t) without the script's comments (--skip-comments)."""
        force = errors > 0 if force is None else force
        shown = ["mysql", *flags, *(["--force"] if force else []), *([db] if db else [])]
        args = ["mysql", f"--defaults-file={self.client_cnf}", *shown[1:]]
        code, out = self._call(args, stdin=(self.base / script).read_text())
        if note:
            print(f"-- {note}")
        print("$ " + " ".join(_arg(a) for a in shown) + f" < {script}")
        if out.strip():
            print(out.rstrip("\n"))
        print()
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        expected_status = (1 if errors and not force else 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}")
        return out

    def shell(self, *args, status=0, shown=None, connect=False):
        """Runs one of MySQL's programs and prints the command and its output. connect=True gives a client program
        (mysqladmin, mysqlshow, mysql) the run's client option file, which the printed command leaves out."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run)
        _show(shown or args, out)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def sql(self, statement, flags=("-t",), errors=0, db=None, note=None):
        """Runs one statement with mysql -e and prints the command and its output."""
        shown = ["mysql", *flags, *([db] if db else []), "-e", statement]
        code, out = self._call(["mysql", f"--defaults-file={self.client_cnf}", *shown[1:]])
        if note:
            print(f"-- {note}")
        _show(shown, out)
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        if found != errors or (code != 0) != (errors > 0):
            sys.exit(f"{statement}: {found} errors and status {code}; the session expects {errors}")
        return out

    def write_file(self, path, text, show=True):
        """Writes a text file into the run folder (an option file, a script) and prints it, as cat would."""
        (self.base / path).write_text(textwrap.dedent(text).lstrip("\n"))
        if show:
            self.show_file(path)

    def show_file(self, path):
        print(f"$ cat {path}")
        print(self.clean((self.base / path).read_text()).rstrip("\n"))
        print()

    def listing(self, folder, prefix=""):
        """Prints the names in a folder of the run (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"$ ls -p {folder}" + (f" | grep '^{prefix}'" if prefix else ""))
        print(textwrap.fill("  ".join(names), width=78, break_on_hyphens=False))
        print()
        return names


def _redraw(block, clean):
    """Draws a client table again after its cells were cleaned. The columns come from the first border line; the row
    after it is the header (left-aligned); a column whose data cells were padded on the left (numbers) stays
    right-aligned (an empty cell says nothing about its column)."""
    cuts = [k for k, ch in enumerate(block[0]) if ch == "+"]
    rows = [None if line.startswith("+") else [line[x + 1:y] for x, y in zip(cuts, cuts[1:])] for line in block]
    header = next(k for k, r in enumerate(rows) if r is not None)
    ncol = len(cuts) - 1
    right = [any(r is not None and k != header and r[c].strip() and r[c].startswith("  ") for k, r in enumerate(rows))
             for c in range(ncol)]
    cells = [None if r is None else [clean(x.strip()) for x in r] for r in rows]
    widths = [max(len(r[c]) for r in cells if r is not None) for c in range(ncol)]
    drawn = []
    for k, r in enumerate(cells):
        if r is None:
            drawn.append("+" + "+".join("-" * (w + 2) for w in widths) + "+")
        else:
            drawn.append("|" + "|".join(" " + (t.rjust(w) if right[c] and k != header else t.ljust(w)) + " "
                                        for c, (t, w) in enumerate(zip(r, widths))) + "|")
    return drawn


def _arg(a):
    """A command-line argument as a reader would type it: quotes only where the shell needs them."""
    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()


# ====================================================================================================================
# Concurrency for the MVCC and locking lessons (undo-logs-mvcc-and-purge, row-gap-next-key-locks).
# Two or more persistent `mysql` clients keep their own transactions open at the same time, so a session can show what
# one transaction sees of another's uncommitted or committed changes, and the record, gap and next-key locks a
# statement takes. One statement can be started without waiting (it blocks on a lock); an observer connection polls
# performance_schema until the wait is visible; the blocked statement then finishes (with its result or its
# ERROR 1205 / ERROR 1213). Output is masked so two runs are byte-identical.
# ====================================================================================================================
import threading
import pty
import termios
import struct
import fcntl

_PROMPT = "READYmk> "   # a distinctive mysql prompt we use as the boundary between statements


class Client:
    """A persistent interactive mysql client on a pseudo-terminal: one connection whose transaction lives across
    calls. Built by Server.client(name). run() sends one statement, waits for the next prompt and prints the result;
    start() sends a statement expected to block and returns at once; finish() waits for that blocked statement.
    A pseudo-terminal is used because the mysql client block-buffers its output to a plain pipe, which would hide the
    result until the process ends; on a terminal it flushes a prompt after every statement. Terminal echo is off (the
    statement is printed by the session, not by the terminal) and the window is made wide so tables never wrap."""

    def __init__(self, server, name):
        self.server = server
        self.name = name
        self.pty_fd, tty_fd = pty.openpty()
        attrs = termios.tcgetattr(tty_fd)
        attrs[3] = attrs[3] & ~termios.ECHO          # lflags: no echo of what we type
        termios.tcsetattr(tty_fd, termios.TCSANOW, attrs)
        fcntl.ioctl(tty_fd, termios.TIOCSWINSZ, struct.pack("HHHH", 100, 1000, 0, 0))  # 1000 columns, no wrap
        args = [str(server.bin / "mysql"), f"--defaults-file={server.client_cnf}",
                f"--prompt={_PROMPT}", "--skip-auto-rehash", "-t"]
        self.proc = subprocess.Popen(args, stdin=tty_fd, stdout=tty_fd, stderr=tty_fd,
                                     cwd=server.base, env=server.env, close_fds=True)
        os.close(tty_fd)
        self._buf = ""
        self._consumed = 0
        self._lock = threading.Lock()
        self._reader = threading.Thread(target=self._read, daemon=True)
        self._reader.start()
        self._read_until_prompt()          # discard the welcome banner up to the first prompt
        self.conn_id = self._value("SELECT CONNECTION_ID()")

    def _read(self):
        while True:
            try:
                data = os.read(self.pty_fd, 4096)
            except OSError:
                break
            if not data:
                break
            with self._lock:
                self._buf += data.decode("utf-8", "replace").replace("\r\n", "\n").replace("\r", "")

    def _send(self, sql):
        os.write(self.pty_fd, (sql.rstrip().rstrip(";") + ";\n").encode())

    def _read_until_prompt(self, timeout=60):
        deadline = time.time() + timeout
        while time.time() < deadline:
            with self._lock:
                idx = self._buf.find(_PROMPT, self._consumed)
                if idx != -1:
                    block = self._buf[self._consumed:idx]
                    self._consumed = idx + len(_PROMPT)
                    return block
            time.sleep(0.02)
        return None

    def _collect(self, sql, timeout=60):
        self._send(sql)
        block = self._read_until_prompt(timeout=timeout)
        if block is None:
            sys.exit(f"{self.name}: statement did not finish within {timeout}s: {sql}")
        return block

    def _value(self, sql):
        text = self.server.clean(self._collect(sql))
        cells = [l for l in text.splitlines() if l.startswith("|")]
        return cells[-1].split("|")[1].strip() if len(cells) >= 2 else ""

    def run(self, sql, note=None, errors=0):
        """Run a statement to completion and print it as the mysql client showed it."""
        out = self._collect(sql)
        self._print(sql, out, note)
        self._check(sql, out, errors)
        return out

    def start(self, sql, note=None):
        """Send a statement expected to block on a lock; return at once. finish() collects its output."""
        self._pending = (sql, note)
        self._send(sql)

    def finish(self, errors=0, timeout=60):
        """Wait for the statement start() sent and print it (its result, or its ERROR 1205/1213)."""
        sql, note = self._pending
        block = self._read_until_prompt(timeout=timeout)
        if block is None:
            sys.exit(f"{self.name}: blocked statement did not finish within {timeout}s: {sql}")
        self._print(sql, block, note)
        self._check(sql, block, errors)
        return block

    def _check(self, sql, out, errors):
        found = sum(1 for l in out.splitlines() if l.startswith("ERROR "))
        if found != errors:
            sys.exit(f"{self.name}: {found} errors on {sql!r}, expected {errors}")

    def _print(self, sql, out, note):
        print(f"-- {self.name}" + (f": {note}" if note else ""))
        print(f"mysql> {sql.rstrip().rstrip(';')};")
        text = self.server.clean(out).strip("\n")
        if text:
            print(text)
        print()

    def close(self):
        try:
            os.write(self.pty_fd, b"\\q\n")
        except OSError:
            pass
        try:
            self.proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            self.proc.terminate()
        try:
            os.close(self.pty_fd)
        except OSError:
            pass


# ---- Server helpers for the concurrency lessons --------------------------------------------------------------------
def _server_client(self, name):
    """A new persistent connection whose transaction lives across statements."""
    if not hasattr(self, "_clients"):
        self._clients = []
    c = Client(self, name)
    self._clients.append(c)
    return c


def _server_wait_for(self, sql, want, timeout=20):
    """Poll a single-value query (through a short-lived connection) until it returns `want` (as text)."""
    want = str(want)
    deadline = time.time() + timeout
    while time.time() < deadline:
        if self.query(sql) == want:
            return True
        time.sleep(0.05)
    sys.exit(f"wait_for timed out: {sql} never became {want}")


def _server_locks(self, note=None, where="1=1"):
    """Print performance_schema.data_locks, only the columns that do not change between runs, ordered stably."""
    sql = ("SELECT INDEX_NAME, LOCK_TYPE, LOCK_MODE, LOCK_STATUS, LOCK_DATA "
           "FROM performance_schema.data_locks "
           f"WHERE OBJECT_SCHEMA='test' AND ({where}) "
           "ORDER BY INDEX_NAME, FIELD(LOCK_MODE,'X','X,GAP','X,REC_NOT_GAP','X,GAP,INSERT_INTENTION','S','S,GAP'), "
           "LOCK_DATA")
    self.sql(sql, note=note)


def _server_status_deadlock(self, note=None):
    """The LATEST DETECTED DEADLOCK section of SHOW ENGINE INNODB STATUS, trimmed to the parts that do not change
    between runs: the two transactions, the locks each holds and waits for, and which one was rolled back. The
    per-record hex dumps and the run-varying ids, times and thread handles are left out or masked."""
    out = self.output("mysql", "-E", "-e", "SHOW ENGINE INNODB STATUS", connect=True)
    lines = out.splitlines()
    try:
        a = next(i for i, l in enumerate(lines) if "LATEST DETECTED DEADLOCK" in l)
        b = next(i for i in range(a + 2, len(lines)) if lines[i].startswith("-----------"))
    except StopIteration:
        a, b = 1, len(lines)
    kept, i = [], a - 1
    while i < b:
        l = lines[i]
        if l.startswith("Record lock, heap no"):      # skip the record's hex field dump up to the blank line
            while i < b and lines[i].strip():
                i += 1
            continue
        kept.append(l)
        i += 1
    text = "\n".join(kept)
    text = re.sub(r"^\d{4}-\d\d-\d\d \d\d:\d\d:\d\d .*$", "<date and time> <thread>", text, flags=re.M)
    text = re.sub(r"TRANSACTION \d+,", "TRANSACTION <id>,", text)
    text = re.sub(r"ACTIVE \d+ sec", "ACTIVE <n> sec", text)
    text = re.sub(r"heap size \d+", "heap size <n>", text)
    text = re.sub(r"MySQL thread id \d+, OS thread handle \w+, query id \d+",
                  "MySQL thread id <tid>, OS thread handle <h>, query id <q>", text)
    text = re.sub(r"trx id \d+", "trx id <id>", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    if note:
        print(f"-- {note}")
    print("mysql> SHOW ENGINE INNODB STATUS\\G   (LATEST DETECTED DEADLOCK section)")
    print(self.clean(text).strip("\n"))
    print()


Server.client = _server_client
Server.wait_for = _server_wait_for
Server.locks = _server_locks
Server.status_deadlock = _server_status_deadlock

key_bytes_session.py

"""key and key_len column by column, for two indexes of the lab.

customer_created (customer_id INT NOT NULL + created_at DATETIME NOT NULL): seek on customer_id alone reports
key_len 4; seek on both reports key_len 9. city_signed (city VARCHAR(40) NOT NULL utf8mb4 + signed_up_on DATE
NOT NULL): seek on city alone reports key_len 162 (40 characters times 4 bytes per character, plus a 2-byte length
prefix); seek on both reports key_len 165. Every byte of the index that narrows the seek is in key_len; every column
that comes after a gap or a range column is not.

A throwaway MySQL 8.4.11 server (mysql_session.py) loads lab_schema.sql quietly, then runs key_bytes.sql through
the mysql client.
"""

from mysql_session import Server

with Server(options=["--innodb-stats-persistent-sample-pages=300"]) as my:
    my.up()
    my.sql("CREATE DATABASE shop", flags=())
    my.mysql("lab_schema.sql", db="shop", flags=("-t", "--skip-comments"))
    my.mysql("key_bytes.sql")

$ mysql -e "CREATE DATABASE shop"

$ mysql -t --skip-comments shop < lab_schema.sql
+----------------+---------+----------+----------+
| Table          | Op      | Msg_type | Msg_text |
+----------------+---------+----------+----------+
| shop.customers | analyze | status   | OK       |
| shop.orders    | analyze | status   | OK       |
| shop.payments  | analyze | status   | OK       |
+----------------+---------+----------+----------+

$ mysql -t -vvv --skip-comments < key_bytes.sql
--------------
EXPLAIN SELECT id FROM orders WHERE customer_id = 42
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: orders
   partitions: NULL
         type: ref
possible_keys: customer_created
          key: customer_created
      key_len: 4
          ref: const
         rows: 42
     filtered: 100.00
        Extra: Using index
1 row in set, 1 warning

--------------
EXPLAIN SELECT id FROM orders WHERE customer_id = 42 AND created_at >= '2026-03-01'
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: orders
   partitions: NULL
         type: range
possible_keys: customer_created
          key: customer_created
      key_len: 9
          ref: NULL
         rows: 28
     filtered: 100.00
        Extra: Using where; Using index
1 row in set, 1 warning

--------------
EXPLAIN SELECT id FROM customers WHERE city = 'Pune'
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: customers
   partitions: NULL
         type: ref
possible_keys: city_signed
          key: city_signed
      key_len: 162
          ref: const
         rows: 150
     filtered: 100.00
        Extra: Using index
1 row in set, 1 warning

--------------
EXPLAIN SELECT id FROM customers WHERE city = 'Pune' AND signed_up_on >= '2025-07-01'
--------------

*************************** 1. row ***************************
           id: 1
  select_type: SIMPLE
        table: customers
   partitions: NULL
         type: range
possible_keys: city_signed
          key: city_signed
      key_len: 165
          ref: NULL
         rows: 77
     filtered: 100.00
        Extra: Using where; Using index
1 row in set, 1 warning

Bye

key_bytes.sql

-- key and key_len for two indexes of the lab, read column by column.
--   customer_created on orders is on (customer_id, created_at): customer_id is INT UNSIGNED NOT NULL (4 bytes),
--   created_at is DATETIME NOT NULL (5 bytes), so a seek on customer_id alone reports key_len 4, and a seek that
--   adds created_at reports key_len 9.
--   city_signed on customers is on (city, signed_up_on): city is VARCHAR(40) NOT NULL in utf8mb4 (40 characters
--   store up to 160 bytes, plus a 2-byte length prefix = 162), signed_up_on is DATE NOT NULL (3 bytes), so a seek
--   on city alone reports key_len 162, and a seek that adds signed_up_on reports key_len 165.
USE shop;

EXPLAIN SELECT id FROM orders WHERE customer_id = 42\G
EXPLAIN SELECT id FROM orders WHERE customer_id = 42 AND created_at >= '2026-03-01'\G
EXPLAIN SELECT id FROM customers WHERE city = 'Pune'\G
EXPLAIN SELECT id FROM customers WHERE city = 'Pune' AND signed_up_on >= '2025-07-01'\G

lab_schema.sql

-- The lesson lab: 1,200 customers, 50,000 orders and 40,000 payments, built without RAND()
-- so that every run on every machine holds exactly the same rows.
SET SESSION cte_max_recursion_depth = 100000;

CREATE TABLE customers (
  id            INT UNSIGNED NOT NULL AUTO_INCREMENT,
  name          VARCHAR(60)  NOT NULL,
  city          VARCHAR(40)  NOT NULL,
  signed_up_on  DATE         NOT NULL,
  PRIMARY KEY (id),
  KEY city_signed (city, signed_up_on)
) ENGINE=InnoDB;

CREATE TABLE orders (
  id           INT UNSIGNED     NOT NULL AUTO_INCREMENT,
  customer_id  INT UNSIGNED     NOT NULL,
  created_at   DATETIME         NOT NULL,
  status       TINYINT UNSIGNED NOT NULL,
  total_paise  INT UNSIGNED     NOT NULL,
  ship_city    VARCHAR(40)      NOT NULL,
  note         VARCHAR(120)     NOT NULL DEFAULT '',
  PRIMARY KEY (id),
  KEY customer_created (customer_id, created_at)
) ENGINE=InnoDB;

-- payments has no index on order_id: an equi-join to orders has no index to use.
CREATE TABLE payments (
  id            INT UNSIGNED NOT NULL AUTO_INCREMENT,
  order_id      INT UNSIGNED NOT NULL,
  amount_paise  INT UNSIGNED NOT NULL,
  method        VARCHAR(12)  NOT NULL,
  PRIMARY KEY (id)
) ENGINE=InnoDB;

-- 1,200 customers in 8 cities (13 and 8 are coprime, so the cities cycle evenly).
INSERT INTO customers (name, city, signed_up_on)
WITH RECURSIVE seq (n) AS (
  SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 1200
)
SELECT CONCAT('Customer ', LPAD(n, 4, '0')),
       ELT(1 + (n * 13) MOD 8, 'Pune', 'Mumbai', 'Bengaluru', 'Chennai', 'Hyderabad', 'Jaipur', 'Kolkata', 'Kochi'),
       DATE_ADD('2025-01-01', INTERVAL (n * 7) MOD 365 DAY)
FROM seq;

-- 50,000 orders over the first 180 days of 2026. customer_id spreads over all 1,200 customers
-- (7919 is prime, so n * 7919 MOD 1200 cycles); total_paise is skewed, with one order in fifty large.
INSERT INTO orders (customer_id, created_at, status, total_paise, ship_city, note)
WITH RECURSIVE seq (n) AS (
  SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 50000
)
SELECT 1 + (n * 7919) MOD 1200,
       DATE_ADD(DATE_ADD('2026-01-01 00:00:00', INTERVAL (n * 311) MOD 180 DAY),
                INTERVAL (n * 1013) MOD 86400 SECOND),
       ELT(1 + (n * 3) MOD 4, 1, 2, 2, 3),
       CASE WHEN n MOD 50 = 0 THEN 400000 + (n * 97) MOD 200000
            ELSE 5000 + (n * 89) MOD 95000 END,
       ELT(1 + (n * 17) MOD 8, 'Pune', 'Mumbai', 'Bengaluru', 'Chennai', 'Hyderabad', 'Jaipur', 'Kolkata', 'Kochi'),
       CONCAT('order note ', n MOD 997)
FROM seq;

-- 40,000 payments: the orders whose id is not a multiple of five are paid once.
INSERT INTO payments (order_id, amount_paise, method)
SELECT id, total_paise, ELT(1 + (id * 7) MOD 4, 'upi', 'card', 'netbanking', 'wallet')
FROM orders
WHERE id MOD 5 <> 0;

ANALYZE TABLE customers, orders, payments;

mysql_session.py

"""Throwaway MySQL 8.4.11 servers for the examples of this lesson.

Every session creates a server of its own: `mysqld --initialize-insecure` makes a new data directory inside a new
folder in the examples folder, the server starts on it, the session runs `mysql` client scripts and MySQL's programs
against it and prints what they printed, and then the server is stopped and the folder deleted, also when something
fails. The server is the pinned MySQL 8.4.11 from the mise installations next to the Python that runs this file;
Server(basedir=...) runs the programs of another unpacked MySQL tarball instead, for sessions that compare versions.

Safety: the server listens only on 127.0.0.1, an address no other computer can reach, on a port the operating system
reports as unused; the X Protocol listener is off (--mysqlx=OFF); its Unix socket, temporary files, error log and
import/export folder all stay inside the run folder. Right after the first start the session gives root a password made
up for the run (never printed) and connects only with a client option file that holds it, so no password ever appears
on a command line. Every program starts with --no-defaults or --defaults-file, so no option file of the computer
(/etc/my.cnf, ~/.my.cnf) can change a result, and HOME points at the run folder, so ~/.mylogin.cnf is not read either.
A printed mysqld command shows the options that matter to the lesson; the session also passes, on the command line,
the folders it keeps inside the run folder (--basedir, --tmpdir, --log-error, --pid-file, --secure-file-priv) and
the port, which is why performance_schema.variables_info reports those variables as COMMAND_LINE.
A session reads only this folder and the MySQL and Python programs it runs, and connects to nothing but 127.0.0.1.

The output is exactly what the programs printed, with these substitutions so that a record never shows the computer it
ran on and is the same on every run: the run folder is shown as ".", the folder of MySQL's programs is left out of
their paths and MySQL's own folder is shown as <basedir>, the computer's name is shown as "lab-host" and the operating-system user's name as "learner", the time a
statement took (the client's "(0.01 sec)") is left out, the time stamp SET PERSIST writes into mysqld-auto.cnf is
shown as <when it was set>, error-log lines lose their timestamp and thread number, and a session can name more values
that change between runs with my.mask(pattern, replacement), which then prints in their place. A client table whose cells
change is drawn again so that its borders still line up.
The sessions never print the password, and the run's port prints as <port>. A session stops with status 1 when a command that should have worked
fails, or when a script 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

MYSQL_VERSION = "8.4.11"
# .../installs/python/3.14.8/bin/python3 -> .../installs
INSTALLS = Path(sys.executable).resolve().parents[3]
PINNED = INSTALLS / "mysql" / MYSQL_VERSION
OS_USER = pwd.getpwuid(os.geteuid()).pw_name
HOST = socket.gethostname().split(".")[0]
# 2026-10-17T16:50:01.123456Z 0 [ERROR] ... -> [ERROR] ...
LOG_PREFIX = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d+Z \d+ ", re.M)


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 server: initialise, start; stopped and deleted when the with block ends.

    options: extra mysqld options for every start (for example "--sql-mode=" or "--character-set-server=latin1").
    basedir: the folder of an unpacked MySQL tarball to run instead of the pinned 8.4.11.
    masks: (pattern, replacement) pairs, as for mask().
    """

    def __init__(self, options=(), redo=None, basedir=None, masks=()):
        self.options = [str(o) for o in options]
        self.redo = redo
        self.basedir = Path(basedir).resolve() if basedir else PINNED
        self.bin = self.basedir / "bin"
        self.masks = []
        for pattern, replacement in masks:
            self.mask(pattern, replacement)
        self.proc = None

    def mask(self, pattern, replacement):
        """From now on every printed line shows text matching the regular expression as the replacement: for a value
        that changes from run to run (an uptime, a time stamp), so that the recorded output stays the same."""
        self.masks.append((re.compile(pattern), replacement))

    # ---- the folder ------------------------------------------------------------------------------------------------
    def __enter__(self):
        if not (self.bin / "mysqld").exists():
            sys.exit(f"no MySQL server program in {self.basedir}" if self.basedir != PINNED
                     else f"MySQL {MYSQL_VERSION} was not found next to this Python")
        self.base = Path(tempfile.mkdtemp(prefix="my", dir=".")).resolve()
        for sub in ("tmp", "files"):
            (self.base / sub).mkdir(mode=0o700)
        self.port = free_port()
        self.password = secrets.token_urlsafe(24)
        self.client_cnf = self.base / "client.cnf"
        self._write_client_cnf(password=None)
        self.env = {key: value for key, value in os.environ.items()
                    if not key.startswith("MYSQL") and key not in ("TZ", "HOME", "LANG", "LC_ALL")}
        self.env.update({"HOME": str(self.base), "TZ": "UTC", "LANG": "en_US.UTF-8",
                         "MYSQL_TEST_LOGIN_FILE": str(self.base / "no-login-paths.cnf")})
        for pattern in ("*.sql", "*.cnf", "*.csv", "*.txt"):
            for f in Path(".").glob(pattern):
                shutil.copy(f, self.base / f.name)
        return self

    def __exit__(self, *exc):
        if self.proc and self.proc.poll() is None:
            self.stop(quiet=True)
        shutil.rmtree(self.base, ignore_errors=True)

    def _write_client_cnf(self, password):
        lines = ["[client]", "user=root", "host=127.0.0.1", f"port={self.port}", "protocol=TCP"]
        if password:
            lines.append(f"password={password}")
        self.client_cnf.write_text("\n".join(lines) + "\n")
        self.client_cnf.chmod(0o600)

    def server_options(self):
        b = self.base
        return ["--no-defaults", f"--basedir={self.basedir}", f"--datadir={b / 'data'}", f"--tmpdir={b / 'tmp'}",
                f"--log-error={b / 'mysqld.err'}", f"--pid-file={b / 'mysqld.pid'}",
                f"--secure-file-priv={b / 'files'}",
                *([f"--innodb-redo-log-capacity={self.redo}"] if self.redo else [])]

    # ---- the server ------------------------------------------------------------------------------------------------
    def initialize(self, show=False):
        """mysqld --initialize-insecure: a new data directory whose root account has no password yet."""
        args = ["mysqld", *self.server_options(), "--initialize-insecure"]
        code, out = self._call(args)
        if show:
            _show(["mysqld", "--no-defaults", "--initialize-insecure", "--datadir=./data"], out)
        if code != 0:
            print(self.log())
            sys.exit(f"mysqld --initialize-insecure ended with status {code}")

    def start(self, extra=(), defaults_file=None, expect_fail=False, show=False):
        """Starts mysqld on 127.0.0.1 and waits until it answers. With expect_fail, prints the error-log lines of a
        start that must fail, and returns them."""
        log = self.base / "mysqld.err"
        before = log.read_text() if log.exists() else ""
        opts = self.server_options()
        if defaults_file:
            opts[0] = f"--defaults-file={self.base / defaults_file}"
        args = [str(self.bin / "mysqld"), *opts, "--bind-address=127.0.0.1", f"--port={self.port}",
                "--socket=mysql.sock", "--mysqlx=OFF", *self.options, *[str(e) for e in extra]]
        if show:
            shown = ["mysqld", opts[0].replace(str(self.base) + "/", ""), "--datadir=./data", "--bind-address=127.0.0.1",
                     "--port=$PORT", "--socket=mysql.sock", "--mysqlx=OFF", *self.options, *extra]
            print("$ " + " ".join(_arg(a) if a != "--port=$PORT" else a for a in shown) + " &")
            print()
        with open(self.base / "mysqld.out", "a") as out:
            self.proc = subprocess.Popen(args, stdout=out, stderr=subprocess.STDOUT, cwd=self.base, env=self.env)
        for _ in range(600):
            if self.proc.poll() is not None:
                break
            if self._ping():
                break
            time.sleep(0.1)
        if self.proc.poll() is not None:
            new = (log.read_text() if log.exists() else "")[len(before):]
            lines = [l for l in self.clean(LOG_PREFIX.sub("", new)).splitlines() if "[ERROR]" in l]
            if expect_fail:
                print("\n".join(lines))
                print(f"(mysqld ended with status {self.proc.returncode})")
                print()
                self.proc = None
                return lines
            print("\n".join(lines))
            sys.exit(f"mysqld ended with status {self.proc.returncode} while starting")
        if expect_fail:
            sys.exit("mysqld started, but the session expected it to fail")
        if not self._ping():
            sys.exit("mysqld did not answer within 60 seconds")
        return []

    def _ping(self):
        code, _ = self._call(["mysqladmin", f"--defaults-file={self.client_cnf}", "--connect-timeout=2", "ping"])
        return code == 0

    def up(self, extra=(), defaults_file=None, show=False):
        """Initialise, start (with start()'s options), and give root the run's password, quietly."""
        self.initialize()
        self.start(extra=extra, defaults_file=defaults_file, show=show)
        return self.login()

    def login(self):
        """Gives root the run's password (never printed) on a server started from a new data directory, and makes
        every later client of the session use it."""
        self.silent(f"ALTER USER 'root'@'localhost' IDENTIFIED BY '{self.password}'")
        self._write_client_cnf(self.password)
        return self

    def stop(self, quiet=False):
        """A clean shutdown (mysqladmin shutdown), then waits until the process has ended."""
        code, out = self._call(["mysqladmin", f"--defaults-file={self.client_cnf}", "shutdown"])
        if code != 0 and not quiet:
            sys.exit(f"mysqladmin shutdown failed: {out}")
        try:
            self.proc.wait(timeout=120)
        except subprocess.TimeoutExpired:
            self.proc.terminate()
            self.proc.wait(timeout=60)
        self.proc = None

    def restart(self, extra=(), defaults_file=None, show=False):
        self.stop()
        self.start(extra=extra, defaults_file=defaults_file, show=show)

    # ---- running things --------------------------------------------------------------------------------------------
    def _call(self, args, stdin=None, merge=True):
        """Runs one of MySQL's programs in the run folder: (status, cleaned output). merge=False keeps only what it
        wrote to standard output (its warnings on standard error are dropped)."""
        args = [str(a) for a in args]
        done = subprocess.run([str(self.bin / args[0]), *args[1:]], stdout=subprocess.PIPE,
                              stderr=subprocess.STDOUT if merge else subprocess.DEVNULL,
                              text=True, cwd=self.base, env=self.env, input=stdin)
        return done.returncode, self.clean(done.stdout)

    def output(self, *args, connect=False, status=0):
        """Runs one of MySQL's programs without printing anything and returns its standard output, for example the
        SQL that mysql_tzinfo_to_sql writes, which the session saves with write_file(..., show=False) and runs next."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run, merge=False)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def clean(self, text):
        """The substitutions of the docstring. A table of the client whose cells change is drawn again with the
        new column widths, so its borders still line up."""
        lines, out, i = text.split("\n"), [], 0
        while i < len(lines):
            if lines[i].startswith("+-"):
                j = i
                while j < len(lines) and lines[j][:1] in ("+", "|"):
                    j += 1
                block = lines[i:j]
                done = [self._clean(l) for l in block]
                out.extend(done if done == block else _redraw(block, self._clean))
                i = j
            else:
                out.append(self._clean(lines[i]))
                i += 1
        return "\n".join(out)

    def _clean(self, text):
        text = text.replace(str(self.bin) + "/", "").replace(str(self.basedir), "<basedir>")
        text = text.replace(str(self.base) + "/", "./").replace(str(self.base), ".")
        text = text.replace("\x07", "")   # the mysql client rings the terminal bell on an error; drop it
        text = re.sub(r" \(\d+\.\d{2,3} sec\)", "", text)  # 8.4 prints 2 decimals, 9.7 and 26.7 print 3
        text = re.sub(r'"Timestamp": \d+', '"Timestamp": <when it was set>', text)
        text = re.sub(rf"\b{re.escape(HOST)}\b", "lab-host", text)
        text = re.sub(rf"\b{re.escape(OS_USER)}\b", "learner", text)
        text = re.sub(rf"(?<![\d.]){self.port}(?![\d.])", "<port>", text)
        for pattern, replacement in self.masks:
            text = pattern.sub(replacement, text)
        return text

    def log(self, pattern=None):
        """The error log without timestamps and thread numbers, optionally only the lines matching a pattern."""
        text = self.clean(LOG_PREFIX.sub("", (self.base / "mysqld.err").read_text()))
        if pattern:
            text = "\n".join(l for l in text.splitlines() if re.search(pattern, l))
        return text

    def show_log(self, pattern, note=None):
        """Prints the error-log lines that match a pattern, as grep would."""
        print(f"$ grep -E {_arg(pattern)} mysqld.err" + (f"   # {note}" if note else ""))
        print(self.log(pattern))
        print()

    def query(self, sql, db=None):
        """One value from one query, unprinted (the session uses it to wait or to decide)."""
        args = ["mysql", f"--defaults-file={self.client_cnf}", "-N", "-B", *([db] if db else [])]
        code, out = self._call(args, stdin=sql)
        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 for the run's password)."""
        self.query(sql)

    def mysql(self, script, errors=0, status=None, db=None, flags=("-t", "-vvv", "--skip-comments"), note=None, force=None):
        """Runs a client script from standard input and prints the command and its output. `errors` is how many
        ERROR lines the script must print; with errors > 0 the client runs on past an error (--force) and then ends
        with status 0, without --force it stops at the first error with status 1. The default flags print each
        statement (-vvv) as a table (-t) without the script's comments (--skip-comments)."""
        force = errors > 0 if force is None else force
        shown = ["mysql", *flags, *(["--force"] if force else []), *([db] if db else [])]
        args = ["mysql", f"--defaults-file={self.client_cnf}", *shown[1:]]
        code, out = self._call(args, stdin=(self.base / script).read_text())
        if note:
            print(f"-- {note}")
        print("$ " + " ".join(_arg(a) for a in shown) + f" < {script}")
        if out.strip():
            print(out.rstrip("\n"))
        print()
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        expected_status = (1 if errors and not force else 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}")
        return out

    def shell(self, *args, status=0, shown=None, connect=False):
        """Runs one of MySQL's programs and prints the command and its output. connect=True gives a client program
        (mysqladmin, mysqlshow, mysql) the run's client option file, which the printed command leaves out."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run)
        _show(shown or args, out)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def sql(self, statement, flags=("-t",), errors=0, db=None, note=None):
        """Runs one statement with mysql -e and prints the command and its output."""
        shown = ["mysql", *flags, *([db] if db else []), "-e", statement]
        code, out = self._call(["mysql", f"--defaults-file={self.client_cnf}", *shown[1:]])
        if note:
            print(f"-- {note}")
        _show(shown, out)
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        if found != errors or (code != 0) != (errors > 0):
            sys.exit(f"{statement}: {found} errors and status {code}; the session expects {errors}")
        return out

    def write_file(self, path, text, show=True):
        """Writes a text file into the run folder (an option file, a script) and prints it, as cat would."""
        (self.base / path).write_text(textwrap.dedent(text).lstrip("\n"))
        if show:
            self.show_file(path)

    def show_file(self, path):
        print(f"$ cat {path}")
        print(self.clean((self.base / path).read_text()).rstrip("\n"))
        print()

    def listing(self, folder, prefix=""):
        """Prints the names in a folder of the run (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"$ ls -p {folder}" + (f" | grep '^{prefix}'" if prefix else ""))
        print(textwrap.fill("  ".join(names), width=78, break_on_hyphens=False))
        print()
        return names


def _redraw(block, clean):
    """Draws a client table again after its cells were cleaned. The columns come from the first border line; the row
    after it is the header (left-aligned); a column whose data cells were padded on the left (numbers) stays
    right-aligned (an empty cell says nothing about its column)."""
    cuts = [k for k, ch in enumerate(block[0]) if ch == "+"]
    rows = [None if line.startswith("+") else [line[x + 1:y] for x, y in zip(cuts, cuts[1:])] for line in block]
    header = next(k for k, r in enumerate(rows) if r is not None)
    ncol = len(cuts) - 1
    right = [any(r is not None and k != header and r[c].strip() and r[c].startswith("  ") for k, r in enumerate(rows))
             for c in range(ncol)]
    cells = [None if r is None else [clean(x.strip()) for x in r] for r in rows]
    widths = [max(len(r[c]) for r in cells if r is not None) for c in range(ncol)]
    drawn = []
    for k, r in enumerate(cells):
        if r is None:
            drawn.append("+" + "+".join("-" * (w + 2) for w in widths) + "+")
        else:
            drawn.append("|" + "|".join(" " + (t.rjust(w) if right[c] and k != header else t.ljust(w)) + " "
                                        for c, (t, w) in enumerate(zip(r, widths))) + "|")
    return drawn


def _arg(a):
    """A command-line argument as a reader would type it: quotes only where the shell needs them."""
    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()


# ====================================================================================================================
# Concurrency for the MVCC and locking lessons (undo-logs-mvcc-and-purge, row-gap-next-key-locks).
# Two or more persistent `mysql` clients keep their own transactions open at the same time, so a session can show what
# one transaction sees of another's uncommitted or committed changes, and the record, gap and next-key locks a
# statement takes. One statement can be started without waiting (it blocks on a lock); an observer connection polls
# performance_schema until the wait is visible; the blocked statement then finishes (with its result or its
# ERROR 1205 / ERROR 1213). Output is masked so two runs are byte-identical.
# ====================================================================================================================
import threading
import pty
import termios
import struct
import fcntl

_PROMPT = "READYmk> "   # a distinctive mysql prompt we use as the boundary between statements


class Client:
    """A persistent interactive mysql client on a pseudo-terminal: one connection whose transaction lives across
    calls. Built by Server.client(name). run() sends one statement, waits for the next prompt and prints the result;
    start() sends a statement expected to block and returns at once; finish() waits for that blocked statement.
    A pseudo-terminal is used because the mysql client block-buffers its output to a plain pipe, which would hide the
    result until the process ends; on a terminal it flushes a prompt after every statement. Terminal echo is off (the
    statement is printed by the session, not by the terminal) and the window is made wide so tables never wrap."""

    def __init__(self, server, name):
        self.server = server
        self.name = name
        self.pty_fd, tty_fd = pty.openpty()
        attrs = termios.tcgetattr(tty_fd)
        attrs[3] = attrs[3] & ~termios.ECHO          # lflags: no echo of what we type
        termios.tcsetattr(tty_fd, termios.TCSANOW, attrs)
        fcntl.ioctl(tty_fd, termios.TIOCSWINSZ, struct.pack("HHHH", 100, 1000, 0, 0))  # 1000 columns, no wrap
        args = [str(server.bin / "mysql"), f"--defaults-file={server.client_cnf}",
                f"--prompt={_PROMPT}", "--skip-auto-rehash", "-t"]
        self.proc = subprocess.Popen(args, stdin=tty_fd, stdout=tty_fd, stderr=tty_fd,
                                     cwd=server.base, env=server.env, close_fds=True)
        os.close(tty_fd)
        self._buf = ""
        self._consumed = 0
        self._lock = threading.Lock()
        self._reader = threading.Thread(target=self._read, daemon=True)
        self._reader.start()
        self._read_until_prompt()          # discard the welcome banner up to the first prompt
        self.conn_id = self._value("SELECT CONNECTION_ID()")

    def _read(self):
        while True:
            try:
                data = os.read(self.pty_fd, 4096)
            except OSError:
                break
            if not data:
                break
            with self._lock:
                self._buf += data.decode("utf-8", "replace").replace("\r\n", "\n").replace("\r", "")

    def _send(self, sql):
        os.write(self.pty_fd, (sql.rstrip().rstrip(";") + ";\n").encode())

    def _read_until_prompt(self, timeout=60):
        deadline = time.time() + timeout
        while time.time() < deadline:
            with self._lock:
                idx = self._buf.find(_PROMPT, self._consumed)
                if idx != -1:
                    block = self._buf[self._consumed:idx]
                    self._consumed = idx + len(_PROMPT)
                    return block
            time.sleep(0.02)
        return None

    def _collect(self, sql, timeout=60):
        self._send(sql)
        block = self._read_until_prompt(timeout=timeout)
        if block is None:
            sys.exit(f"{self.name}: statement did not finish within {timeout}s: {sql}")
        return block

    def _value(self, sql):
        text = self.server.clean(self._collect(sql))
        cells = [l for l in text.splitlines() if l.startswith("|")]
        return cells[-1].split("|")[1].strip() if len(cells) >= 2 else ""

    def run(self, sql, note=None, errors=0):
        """Run a statement to completion and print it as the mysql client showed it."""
        out = self._collect(sql)
        self._print(sql, out, note)
        self._check(sql, out, errors)
        return out

    def start(self, sql, note=None):
        """Send a statement expected to block on a lock; return at once. finish() collects its output."""
        self._pending = (sql, note)
        self._send(sql)

    def finish(self, errors=0, timeout=60):
        """Wait for the statement start() sent and print it (its result, or its ERROR 1205/1213)."""
        sql, note = self._pending
        block = self._read_until_prompt(timeout=timeout)
        if block is None:
            sys.exit(f"{self.name}: blocked statement did not finish within {timeout}s: {sql}")
        self._print(sql, block, note)
        self._check(sql, block, errors)
        return block

    def _check(self, sql, out, errors):
        found = sum(1 for l in out.splitlines() if l.startswith("ERROR "))
        if found != errors:
            sys.exit(f"{self.name}: {found} errors on {sql!r}, expected {errors}")

    def _print(self, sql, out, note):
        print(f"-- {self.name}" + (f": {note}" if note else ""))
        print(f"mysql> {sql.rstrip().rstrip(';')};")
        text = self.server.clean(out).strip("\n")
        if text:
            print(text)
        print()

    def close(self):
        try:
            os.write(self.pty_fd, b"\\q\n")
        except OSError:
            pass
        try:
            self.proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            self.proc.terminate()
        try:
            os.close(self.pty_fd)
        except OSError:
            pass


# ---- Server helpers for the concurrency lessons --------------------------------------------------------------------
def _server_client(self, name):
    """A new persistent connection whose transaction lives across statements."""
    if not hasattr(self, "_clients"):
        self._clients = []
    c = Client(self, name)
    self._clients.append(c)
    return c


def _server_wait_for(self, sql, want, timeout=20):
    """Poll a single-value query (through a short-lived connection) until it returns `want` (as text)."""
    want = str(want)
    deadline = time.time() + timeout
    while time.time() < deadline:
        if self.query(sql) == want:
            return True
        time.sleep(0.05)
    sys.exit(f"wait_for timed out: {sql} never became {want}")


def _server_locks(self, note=None, where="1=1"):
    """Print performance_schema.data_locks, only the columns that do not change between runs, ordered stably."""
    sql = ("SELECT INDEX_NAME, LOCK_TYPE, LOCK_MODE, LOCK_STATUS, LOCK_DATA "
           "FROM performance_schema.data_locks "
           f"WHERE OBJECT_SCHEMA='test' AND ({where}) "
           "ORDER BY INDEX_NAME, FIELD(LOCK_MODE,'X','X,GAP','X,REC_NOT_GAP','X,GAP,INSERT_INTENTION','S','S,GAP'), "
           "LOCK_DATA")
    self.sql(sql, note=note)


def _server_status_deadlock(self, note=None):
    """The LATEST DETECTED DEADLOCK section of SHOW ENGINE INNODB STATUS, trimmed to the parts that do not change
    between runs: the two transactions, the locks each holds and waits for, and which one was rolled back. The
    per-record hex dumps and the run-varying ids, times and thread handles are left out or masked."""
    out = self.output("mysql", "-E", "-e", "SHOW ENGINE INNODB STATUS", connect=True)
    lines = out.splitlines()
    try:
        a = next(i for i, l in enumerate(lines) if "LATEST DETECTED DEADLOCK" in l)
        b = next(i for i in range(a + 2, len(lines)) if lines[i].startswith("-----------"))
    except StopIteration:
        a, b = 1, len(lines)
    kept, i = [], a - 1
    while i < b:
        l = lines[i]
        if l.startswith("Record lock, heap no"):      # skip the record's hex field dump up to the blank line
            while i < b and lines[i].strip():
                i += 1
            continue
        kept.append(l)
        i += 1
    text = "\n".join(kept)
    text = re.sub(r"^\d{4}-\d\d-\d\d \d\d:\d\d:\d\d .*$", "<date and time> <thread>", text, flags=re.M)
    text = re.sub(r"TRANSACTION \d+,", "TRANSACTION <id>,", text)
    text = re.sub(r"ACTIVE \d+ sec", "ACTIVE <n> sec", text)
    text = re.sub(r"heap size \d+", "heap size <n>", text)
    text = re.sub(r"MySQL thread id \d+, OS thread handle \w+, query id \d+",
                  "MySQL thread id <tid>, OS thread handle <h>, query id <q>", text)
    text = re.sub(r"trx id \d+", "trx id <id>", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    if note:
        print(f"-- {note}")
    print("mysql> SHOW ENGINE INNODB STATUS\\G   (LATEST DETECTED DEADLOCK section)")
    print(self.clean(text).strip("\n"))
    print()


Server.client = _server_client
Server.wait_for = _server_wait_for
Server.locks = _server_locks
Server.status_deadlock = _server_status_deadlock