Database Caching Strategies

# Three tables the Python examples read from. Run this once.
docker exec -i pg-cache-demo psql -U shop -d shop -q -v ON_ERROR_STOP=1 \
  --pset linestyle=unicode --pset border=2 <<'SQL'
DROP TABLE IF EXISTS products, categories, users;

CREATE TABLE users (
    id         INT PRIMARY KEY,
    username   TEXT NOT NULL,
    email      TEXT NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

INSERT INTO users (id, username, email) VALUES
    (123, 'alice', 'alice@example.com'),
    (456, 'bob',   'bob@example.com'),
    (789, 'carol', 'carol@example.com');

CREATE TABLE categories (
    id   INT PRIMARY KEY,
    name TEXT NOT NULL
);

INSERT INTO categories VALUES (1, 'Electronics'), (2, 'Books');

CREATE TABLE products (
    id          INT PRIMARY KEY,
    name        TEXT NOT NULL,
    price       NUMERIC(10, 2) NOT NULL,
    category_id INT NOT NULL REFERENCES categories(id)
);

-- 200,000 rows, so the search example later has real work to do.
INSERT INTO products (id, name, price, category_id)
SELECT g,
       CASE WHEN g % 1000 = 0 THEN 'Laptop model ' || g ELSE 'Gadget ' || g END,
       (g % 2000) + 0.99,
       CASE WHEN g % 2 = 0 THEN 1 ELSE 2 END
FROM generate_series(1, 200000) AS g;

ANALYZE;

SELECT (SELECT count(*) FROM users)      AS users,
       (SELECT count(*) FROM categories) AS categories,
       (SELECT count(*) FROM products)   AS products;
SQL

"""Redis basics: connect, SET, GET, INCR."""
import redis

# One connection object per process; redis-py pools connections internally.
cache = redis.Redis(
    host="localhost",
    port=6379,
    db=0,
    decode_responses=True,  # return str instead of bytes
)

cache.flushdb()  # start from a clean database so this script is repeatable

# Strings: SET stores the value, GET reads it back.
cache.set("username", "alice")
cache.set("user:1:email", "alice@example.com")

print("username     :", cache.get("username"))
print("user:1:email :", cache.get("user:1:email"))
print("missing key  :", cache.get("user:2:email"))

# Counters: INCR is atomic and creates the key at 0 if it does not exist.
print("incr #1      :", cache.incr("view_count"))
print("incr #2      :", cache.incr("view_count"))
print("incr by 10   :", cache.incrby("view_count", 10))

# Everything comes back as a string, including numbers.
print("stored value :", repr(cache.get("view_count")))
print("key exists   :", cache.exists("view_count") == 1)

"""Expiration: TTL, EXPIRE, PERSIST."""
import time
from datetime import timedelta

import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
cache.flushdb()

# SET with an expiry. redis-py marks setex() deprecated in favour of
# set(..., ex=...), and ex accepts an int of seconds or a timedelta.
cache.set("session:abc123", "user_data", ex=timedelta(hours=1))
print("ttl right after set :", cache.ttl("session:abc123"), "seconds")

# SET then EXPIRE is the two-step equivalent (the key is briefly immortal).
cache.set("temp:token", "xyz789")
cache.expire("temp:token", 300)
print("ttl of temp:token   :", cache.ttl("temp:token"), "seconds")

# TTL counts down in whole seconds and Redis deletes the key at zero.
cache.set("blink", "gone soon", ex=2)
print("blink now           :", cache.get("blink"), "| ttl", cache.ttl("blink"))
time.sleep(2.1)
print("blink after 2.1s    :", cache.get("blink"), "| ttl", cache.ttl("blink"))

# Two sentinel values worth knowing: -1 = key exists but never expires,
# -2 = key does not exist at all.
cache.persist("session:abc123")
print("ttl after persist   :", cache.ttl("session:abc123"), "(-1 = no expiry)")
print("ttl of unknown key  :", cache.ttl("nope"), "(-2 = key missing)")

"""maxmemory + an eviction policy: the bound that keeps a cache from eating the host."""
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
cache.flushall()

# Redis ships with no memory bound and refuses writes once the host is full.
print("default maxmemory       :", cache.config_get("maxmemory")["maxmemory"])
print("default maxmemory-policy:", cache.config_get("maxmemory-policy")["maxmemory-policy"])

# For a cache, both of these are the wrong defaults. Cap the memory and let
# Redis drop the least recently used keys instead of failing writes.
cache.config_set("maxmemory", "4mb")
cache.config_set("maxmemory-policy", "allkeys-lru")

# evicted_keys counts from server start, so record it before the write.
evicted_before = cache.info("stats")["evicted_keys"]

# Write ~20 MB into a 4 MB cache: 20,000 values of 1 KB.
payload = "x" * 1024
pipe = cache.pipeline(transaction=False)
for i in range(20_000):
    pipe.set(f"pad:{i}", payload)
    if i % 1000 == 999:
        pipe.execute()
pipe.execute()

evicted = cache.info("stats")["evicted_keys"] - evicted_before
print("keys written            : 20000")
print("keys still resident     :", cache.dbsize())
print("keys evicted            :", evicted)
print("used memory             :", cache.info("memory")["used_memory_human"])
print("oldest key still there  :", cache.get("pad:0") is not None)
print("newest key still there  :", cache.get("pad:19999") is not None)

cache.config_set("maxmemory", "0")
cache.config_set("maxmemory-policy", "noeviction")
cache.flushall()

"""Cache-aside (lazy loading): read cache, fall back to the database, backfill."""
import json
from datetime import timedelta

import psycopg
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
db = psycopg.connect("host=localhost port=5499 dbname=shop user=shop password=shop")

cache.flushdb()

db_queries = 0  # so the run below can prove the database was skipped


def get_user(user_id):
    """Return one user, using Redis as a look-aside cache."""
    global db_queries
    cache_key = f"user:{user_id}"

    # 1. Try the cache first.
    cached = cache.get(cache_key)
    if cached is not None:
        print(f"  cache HIT  {cache_key}")
        return json.loads(cached)

    # 2. Miss: the database is the source of truth.
    print(f"  cache MISS {cache_key} -> querying PostgreSQL")
    db_queries += 1
    with db.cursor() as cur:
        cur.execute(
            "SELECT id, username, email FROM users WHERE id = %s", (user_id,)
        )
        row = cur.fetchone()

    if row is None:
        return None  # nothing cached: every request for a missing id hits the DB

    user = {"id": row[0], "username": row[1], "email": row[2]}

    # 3. Backfill the cache so the next reader skips step 2.
    cache.set(cache_key, json.dumps(user), ex=timedelta(hours=1))
    return user


print("first read:")
print("  ->", get_user(123))
print("second read:")
print("  ->", get_user(123))
print("third read:")
print("  ->", get_user(123))
print(f"database queries for 3 reads: {db_queries}")

db.close()

"""Is the cache actually faster? Measure, do not assume.

Compares one indexed primary-key lookup against one Redis GET, 5 repetitions
of 30 runs each, reporting the median of each repetition.
"""
import json
import statistics
import time

import psycopg
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
db = psycopg.connect("host=localhost port=5499 dbname=shop user=shop password=shop")

cache.set("user:123", json.dumps({"id": 123, "username": "alice"}))
cur = db.cursor()


def timed(fn, runs=30):
    samples = []
    for _ in range(runs):
        start = time.perf_counter()
        fn()
        samples.append((time.perf_counter() - start) * 1000)  # ms
    return statistics.median(samples)


def db_read():
    cur.execute("SELECT id, username, email FROM users WHERE id = 123")
    cur.fetchone()


def cache_read():
    json.loads(cache.get("user:123"))


timed(db_read, 50)  # warm up: connection, plan cache, shared buffers
timed(cache_read, 50)

print("rep | db (ms) | redis (ms) | ratio")
for rep in range(1, 6):
    d = timed(db_read)
    c = timed(cache_read)
    print(f"  {rep} |  {d:6.3f} |     {c:6.3f} | {d / c:4.1f}x")

cur.close()
db.close()

"""Cache stampede: what 20 simultaneous misses cost, with and without a lock."""
import json
import threading
import time
from datetime import timedelta

import psycopg
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
DSN = "host=localhost port=5499 dbname=shop user=shop password=shop"

db_calls = 0
db_calls_lock = threading.Lock()


def rebuild_from_db(user_id):
    """The expensive read the cache exists to avoid (pg_sleep fakes the cost)."""
    global db_calls
    with db_calls_lock:
        db_calls += 1
    with psycopg.connect(DSN) as conn, conn.cursor() as cur:
        cur.execute(
            "SELECT id, username, email, pg_sleep(0.2) FROM users WHERE id = %s",
            (user_id,),
        )
        row = cur.fetchone()
    return {"id": row[0], "username": row[1], "email": row[2]}


def get_user_naive(user_id):
    """Cache-aside with no stampede protection."""
    key = f"user:{user_id}"
    cached = cache.get(key)
    if cached is not None:
        return json.loads(cached)
    user = rebuild_from_db(user_id)
    cache.set(key, json.dumps(user), ex=timedelta(hours=1))
    return user


def get_user_locked(user_id, attempts=50):
    """Cache-aside where exactly one caller is allowed to rebuild the entry."""
    key = f"user:{user_id}"
    lock_key = f"lock:{key}"

    for _ in range(attempts):
        cached = cache.get(key)
        if cached is not None:
            return json.loads(cached)

        # SET NX is atomic: exactly one caller gets True. EX is mandatory,
        # otherwise a crash between here and DELETE strands the lock forever.
        if cache.set(lock_key, "1", nx=True, ex=5):
            try:
                user = rebuild_from_db(user_id)
                cache.set(key, json.dumps(user), ex=timedelta(hours=1))
                return user
            finally:
                cache.delete(lock_key)

        time.sleep(0.05)  # someone else is rebuilding: wait, then re-check

    # Lock holder died and the entry is still cold: serve from the database
    # rather than failing the request.
    return rebuild_from_db(user_id)


def storm(fn, threads=20):
    global db_calls
    cache.flushdb()
    db_calls = 0
    workers = [threading.Thread(target=fn, args=(123,)) for _ in range(threads)]
    start = time.perf_counter()
    for w in workers:
        w.start()
    for w in workers:
        w.join()
    elapsed = (time.perf_counter() - start) * 1000
    return db_calls, elapsed


calls, ms = storm(get_user_naive)
print(f"no lock : {calls:2d} database queries for 20 concurrent misses ({ms:.0f} ms)")

calls, ms = storm(get_user_locked)
print(f"with lock: {calls:2d} database query  for 20 concurrent misses ({ms:.0f} ms)")

"""Write-around: write the database, delete the cache entry."""
import json
from datetime import timedelta

import psycopg
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
db = psycopg.connect("host=localhost port=5499 dbname=shop user=shop password=shop")
cache.flushdb()


def get_user(user_id):
    """Cache-aside read (same function as the previous section)."""
    key = f"user:{user_id}"
    cached = cache.get(key)
    if cached is not None:
        return "HIT ", json.loads(cached)

    with db.cursor() as cur:
        cur.execute("SELECT id, username, email FROM users WHERE id = %s", (user_id,))
        row = cur.fetchone()
    user = {"id": row[0], "username": row[1], "email": row[2]}
    cache.set(key, json.dumps(user), ex=timedelta(hours=1))
    return "MISS", user


def update_user_email(user_id, email):
    """Write to the database, then invalidate the cached copy."""
    with db.cursor() as cur:
        cur.execute(
            "UPDATE users SET email = %s, updated_at = now() WHERE id = %s",
            (email, user_id),
        )
    db.commit()  # delete only after the write is durable, never before

    # DEL is idempotent: it returns how many keys it removed, and removing a
    # key that was never cached is not an error.
    removed = cache.delete(f"user:{user_id}")
    print(f"invalidated user:{user_id} (keys removed: {removed})")


print("read 1:", get_user(123))
print("read 2:", get_user(123))

update_user_email(123, "alice.new@example.com")

print("read 3:", get_user(123))
print("read 4:", get_user(123))

# Reset the row so the script can be re-run.
with db.cursor() as cur:
    cur.execute("UPDATE users SET email = 'alice@example.com' WHERE id = 123")
db.commit()
db.close()

"""Write-through: write the database, then refresh the cache with the new row."""
import json
from datetime import timedelta

import psycopg
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
db = psycopg.connect("host=localhost port=5499 dbname=shop user=shop password=shop")
cache.flushdb()


def update_user_email(user_id, email):
    """Update the row and write the returned values straight into the cache."""
    with db.cursor() as cur:
        # RETURNING names the columns explicitly: RETURNING * would make the
        # row unpacking below depend on physical column order.
        cur.execute(
            """
            UPDATE users SET email = %s, updated_at = now()
            WHERE id = %s
            RETURNING id, username, email
            """,
            (email, user_id),
        )
        row = cur.fetchone()
    db.commit()  # refresh the cache only once the write is durable

    user = {"id": row[0], "username": row[1], "email": row[2]}
    cache.set(f"user:{user_id}", json.dumps(user), ex=timedelta(hours=1))
    return user


def get_user(user_id):
    cached = cache.get(f"user:{user_id}")
    if cached is not None:
        return "HIT ", json.loads(cached)
    return "MISS", None


print("before update:", get_user(123))
print("updated      :", update_user_email(123, "alice.new@example.com"))
print("after update :", get_user(123))

with db.cursor() as cur:
    cur.execute("UPDATE users SET email = 'alice@example.com' WHERE id = 123")
db.commit()
db.close()

"""Invalidation pattern 1: let the TTL do it."""
import json
import time

import psycopg
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
db = psycopg.connect("host=localhost port=5499 dbname=shop user=shop password=shop")
cache.flushdb()

with db.cursor() as cur:
    cur.execute("SELECT id, name, price FROM products WHERE id = 1000")
    row = cur.fetchone()

# price is NUMERIC, which psycopg returns as Decimal, and json.dumps() cannot
# serialise Decimal. Cast at the boundary.
product = {"id": row[0], "name": row[1], "price": float(row[2])}

# A 3-second TTL so the effect is visible in one script run; a product page
# in production is closer to 5-15 minutes.
cache.set("product:1000", json.dumps(product), ex=3)
print("cached      :", cache.get("product:1000"))
print("ttl         :", cache.ttl("product:1000"), "seconds")

# The row changes underneath the cache. No invalidation code runs.
with db.cursor() as cur:
    cur.execute("UPDATE products SET price = 1499.99 WHERE id = 1000")
db.commit()

print("db price now: 1499.99")
print("cache still :", json.loads(cache.get("product:1000"))["price"], "(stale, by design)")

time.sleep(3.1)
print("after ttl   :", cache.get("product:1000"), "(expired, next read repopulates)")

with db.cursor() as cur:
    cur.execute("UPDATE products SET price = 1000.99 WHERE id = 1000")
db.commit()
db.close()

"""Invalidation pattern 2: delete every key the write touched."""
import json

import psycopg
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
db = psycopg.connect("host=localhost port=5499 dbname=shop user=shop password=shop")
cache.flushdb()

# Pretend these four entries were built by earlier page renders.
cache.set("product:1000", json.dumps({"id": 1000, "price": 1000.99}), ex=900)
cache.set("product:1000:details", json.dumps({"reviews": 12}), ex=900)
cache.set("category:1:products", json.dumps([1000, 1002, 1004]), ex=900)
cache.set("homepage:featured", json.dumps([1000]), ex=900)


def update_price(product_id, new_price):
    """Update a product and drop every cached view that contains it."""
    with db.cursor() as cur:
        # RETURNING gives the category without a second round trip: the
        # invalidation needs it and the row is already locked here.
        cur.execute(
            "UPDATE products SET price = %s WHERE id = %s RETURNING category_id",
            (new_price, product_id),
        )
        (category_id,) = cur.fetchone()
    db.commit()

    keys = [
        f"product:{product_id}",              # the product itself
        f"product:{product_id}:details",      # its detail page
        f"category:{category_id}:products",   # the listing it appears in
        "homepage:featured",                  # any page that embeds it
    ]
    removed = cache.delete(*keys)  # variadic DEL: one round trip, not four
    print(f"deleted {removed} of {len(keys)} candidate keys: {keys}")


# KEYS scans the whole keyspace and blocks the server: fine for this demo,
# never on a production instance. Use SCAN there.
print("before:", sorted(cache.keys("*")))
update_price(1000, 1499.99)
print("after :", sorted(cache.keys("*")))

with db.cursor() as cur:
    cur.execute("UPDATE products SET price = 1000.99 WHERE id = 1000")
db.commit()
db.close()

"""Invalidation pattern 3: group keys under tags, then delete a whole tag."""
import json

import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
cache.flushdb()


def cache_with_tags(key, data, tags, ttl=900):
    """Store a value and register it under each tag (a Redis SET)."""
    pipe = cache.pipeline()  # one round trip for the value and all tag writes
    pipe.set(key, json.dumps(data), ex=ttl)
    for tag in tags:
        pipe.sadd(f"tag:{tag}", key)
        # Tag sets need their own TTL, or they outlive every value they
        # point at. Give them a margin over the data TTL.
        pipe.expire(f"tag:{tag}", ttl * 2)
    pipe.execute()


def invalidate_by_tag(tag):
    """Delete every value registered under a tag, plus the tag set itself."""
    tag_key = f"tag:{tag}"
    keys = cache.smembers(tag_key)
    if not keys:
        print(f"tag '{tag}': nothing to invalidate")
        return 0

    # Members may already have expired on their own: DEL reports how many
    # keys it actually removed, which is the honest number to log.
    removed = cache.delete(*keys)
    cache.delete(tag_key)
    print(f"tag '{tag}': {len(keys)} member(s), {removed} live key(s) deleted")
    return removed


cache_with_tags(
    "product:1000",
    {"id": 1000, "name": "Laptop model 1000"},
    tags=["product", "category:electronics", "brand:apple"],
)
cache_with_tags(
    "product:2000",
    {"id": 2000, "name": "Laptop model 2000"},
    tags=["product", "category:electronics", "brand:samsung"],
)
cache_with_tags(
    "product:3000",
    {"id": 3000, "name": "Laptop model 3000"},
    tags=["product", "category:books"],
)

print("electronics members:", sorted(cache.smembers("tag:category:electronics")))
invalidate_by_tag("category:electronics")

print("product:1000 ->", cache.get("product:1000"))
print("product:2000 ->", cache.get("product:2000"))
print("product:3000 ->", cache.get("product:3000"))

# The "product" tag still lists the two keys just deleted: tag sets are not
# updated when a value expires or is deleted through another tag.
print("product tag members:", sorted(cache.smembers("tag:product")))
invalidate_by_tag("product")

"""Caching a whole query result: an unindexed search over 200,000 rows."""
import hashlib
import json
import statistics
import time

import psycopg
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
db = psycopg.connect("host=localhost port=5499 dbname=shop user=shop password=shop")
cache.flushdb()


def run_search(term, category=None, min_price=None):
    """The expensive part: ILIKE '%term%' cannot use a b-tree index."""
    sql = """
        SELECT p.id, p.name, p.price, c.name AS category
        FROM products p
        JOIN categories c ON c.id = p.category_id
        WHERE p.name ILIKE %s
    """
    params = [f"%{term}%"]
    if category is not None:
        sql += " AND c.name = %s"
        params.append(category)
    if min_price is not None:
        sql += " AND p.price >= %s"
        params.append(min_price)
    # Without ORDER BY the row order is unspecified, so the cached copy and a
    # fresh query could disagree. Order by the primary key to make both stable.
    sql += " ORDER BY p.id"

    with db.cursor() as cur:
        cur.execute(sql, params)
        rows = cur.fetchall()

    # price is NUMERIC -> Decimal, which json.dumps() refuses to serialise.
    return [
        {"id": r[0], "name": r[1], "price": float(r[2]), "category": r[3]}
        for r in rows
    ]


def search_products(term, category=None, min_price=None, ttl=300):
    """Cache-aside around run_search(), keyed by the search parameters."""
    # The key must depend on every parameter, or two different searches
    # collide. Hash it so the key stays short and free of user input.
    fingerprint = json.dumps([term, category, min_price], sort_keys=True)
    key = "search:" + hashlib.sha256(fingerprint.encode()).hexdigest()[:16]

    cached = cache.get(key)
    if cached is not None:
        return "HIT ", json.loads(cached)

    results = run_search(term, category, min_price)
    cache.set(key, json.dumps(results), ex=ttl)
    return "MISS", results


status, rows = search_products("laptop", category="Electronics", min_price=500)
print(f"{status} {len(rows)} rows, first: {rows[0]}")

status, rows = search_products("laptop", category="Electronics", min_price=500)
print(f"{status} {len(rows)} rows, first: {rows[0]}")

status, rows = search_products("laptop", category="Electronics")
print(f"{status} {len(rows)} rows  (no min_price -> different key, cold again)")


def median_ms(fn, runs=10):
    samples = []
    for _ in range(runs):
        start = time.perf_counter()
        fn()
        samples.append((time.perf_counter() - start) * 1000)
    return statistics.median(samples)


cached_call = lambda: search_products("laptop", category="Electronics", min_price=500)
uncached_call = lambda: run_search("laptop", category="Electronics", min_price=500)

median_ms(uncached_call, 3)  # warm the shared buffers before measuring
print("\nrep | uncached (ms) | cached (ms) | speedup")
for rep in range(1, 6):
    u = median_ms(uncached_call)
    c = median_ms(cached_call)
    print(f"  {rep} |     {u:9.2f} |   {c:9.3f} | {u / c:6.0f}x")

db.close()

"""A leaderboard that never touches the database: Redis sorted sets."""
import redis

cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
cache.flushdb()

KEY = "leaderboard:global"


def set_score(user_id, score):
    """ZADD, O(log N). Re-adding an existing member just moves it."""
    cache.zadd(KEY, {f"user:{user_id}": score})


def add_points(user_id, points):
    """ZINCRBY is atomic, so concurrent scorers cannot lose updates."""
    return cache.zincrby(KEY, points, f"user:{user_id}")


def top(limit=10):
    """ZRANGE ... REV, O(log N + M). ZREVRANGE is deprecated since Redis 6.2."""
    rows = cache.zrange(KEY, 0, limit - 1, desc=True, withscores=True)
    return [
        {"user_id": member.split(":")[1], "score": int(score)}
        for member, score in rows
    ]


def rank_of(user_id):
    """ZREVRANK is 0-based, so add 1 for a human-readable position."""
    rank = cache.zrevrank(KEY, f"user:{user_id}")
    return None if rank is None else rank + 1


set_score(123, 9500)
set_score(456, 12000)
set_score(789, 8000)
set_score(321, 12000)  # deliberate tie with user 456

print("top 3      :", top(3))
print("after +750 :", add_points(789, 750))
print("top 3      :", top(3))
print("rank of 789:", rank_of(789))
print("rank of 999:", rank_of(999), "(not on the board)")
print("board size :", cache.zcard(KEY))
print("score 12000:", sorted(cache.zrangebyscore(KEY, 12000, 12000)))

# The MySQL query cache is the classic example of a database-side result
# cache. Do not take its removal on trust: ask a current server.
docker run -d --name mysql-demo -e MYSQL_ROOT_PASSWORD=demo mysql:8.4

# MYSQL_PWD avoids the "password on the command line is insecure" warning.
docker exec -e MYSQL_PWD=demo mysql-demo mysql -uroot --table \
  -e "SELECT VERSION() AS version;"

# Every query_cache_* variable is gone.
docker exec -e MYSQL_PWD=demo mysql-demo mysql -uroot --table \
  -e "SHOW VARIABLES LIKE 'query_cache%';"

# have_query_cache survives only to answer "no".
docker exec -e MYSQL_PWD=demo mysql-demo mysql -uroot --table \
  -e "SHOW VARIABLES LIKE 'have_query_cache';"

# Trying to size it is an error, not a no-op.
docker exec -e MYSQL_PWD=demo mysql-demo mysql -uroot \
  -e "SET GLOBAL query_cache_size = 1073741824;"

# Clean up:  docker rm -f mysql-demo

# PostgreSQL never caches query results: every query is planned and executed
# again. What it does cache is 8 KB data pages, in shared_buffers, so the
# restart below is what makes the first scan cold.
docker restart pg-cache-demo
until docker exec pg-cache-demo pg_isready -U shop -q; do sleep 1; done

docker exec -i pg-cache-demo psql -U shop -d shop -q \
  --pset linestyle=unicode --pset border=2 <<'SQL'
-- Serial plan, to keep the output easy to read. SET lasts for one session,
-- which here is this single psql invocation.
SET max_parallel_workers_per_gather = 0;
SELECT pg_stat_reset();  -- throwaway container only: this wipes all statistics

SHOW shared_buffers;
SHOW effective_cache_size;

-- Same scan twice. "read" = the block was not in shared_buffers, "hit" = it was.
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT count(*) FROM products WHERE price > 500;

EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT count(*) FROM products WHERE price > 500;
SQL

# The block above runs and exits. To poke around by hand, open a psql
# session inside the container instead: -it allocates the terminal, without
# which there is no prompt.
docker exec -it pg-cache-demo psql -U shop -d shop

# Inside psql, these two meta-commands are what the --pset flags do above.
# SET applies to this session only, so run it and both EXPLAINs before \q:
# the next session plans the same scan in parallel again.

# Buffer cache hit ratio per table. Statistics are reported when a
# transaction ends and the collector can lag by up to a second, so run this
# as its own command, a moment after the queries above. NULLIF guards a table
# that has served no reads at all, where the denominator would be zero.
docker exec pg-cache-demo psql -U shop -d shop \
  --pset linestyle=unicode --pset border=2 -c "
SELECT relname,
       heap_blks_read AS disk_reads,
       heap_blks_hit  AS buffer_hits,
       round(100 * heap_blks_hit::numeric
             / nullif(heap_blks_hit + heap_blks_read, 0), 2) AS hit_pct
FROM pg_statio_user_tables
ORDER BY relname;"