make_fixtures.py

# SPDX-License-Identifier: Apache-2.0
#
# Regenerates the .db fixtures in this directory using only the Python
# standard library (the sqlite3 module IS the reference implementation).
# Content is fixed, so reruns are byte-identical for a given SQLite library
# version (the header embeds the library version number at offset 96).
#
#   python3 make_fixtures.py
#
# All fixtures are synthetic and generated by this script; no third-party or
# customer data. Keep them small.

import os
import sqlite3
import struct
from pathlib import Path

HERE = Path(__file__).parent

print(f"sqlite library {sqlite3.sqlite_version}")


def fresh(name, page_size=None, encoding=None):
    path = HERE / name
    for suffix in ("", "-wal", "-shm", "-journal"):
        p = Path(str(path) + suffix)
        if p.exists():
            p.unlink()
    conn = sqlite3.connect(path)
    if page_size is not None:
        conn.execute(f"PRAGMA page_size = {page_size}")
    if encoding is not None:
        conn.execute(f"PRAGMA encoding = '{encoding}'")
    return conn, path


def header(path):
    with open(path, "rb") as f:
        return f.read(100)


def report(path):
    h = header(path)
    ps = struct.unpack(">H", h[16:18])[0]
    ps = 65536 if ps == 1 else ps
    enc = struct.unpack(">I", h[56:60])[0]
    print(f"{path.name}: {path.stat().st_size} bytes, "
          f"page_size={ps}, encoding={enc}, write_ver={h[18]}")


# --- types.db: every serial type, IPK rowid aliasing, dynamic typing -----------
# 9 rows; iv walks the integer serial types (0/1 constants, 1..8 byte two's
# complement), rv covers doubles including the integral-real optimization
# (0.0, 2.0, -0.0 are stored as integers on disk), tv/bv cover text and blob
# including empty values, nv is always NULL, av holds a different storage class
# in every row. Rowids 1..8 plus 2**40 prove multi-byte varint rowids and the
# INTEGER PRIMARY KEY alias (the id column is stored as NULL serial type 0).

conn, path = fresh("types.db")
conn.execute("""CREATE TABLE Types (
    id INTEGER PRIMARY KEY,
    iv INTEGER,
    rv REAL,
    tv TEXT,
    bv BLOB,
    nv INTEGER,
    av
)""")
conn.execute("CREATE TABLE Empty (id INTEGER PRIMARY KEY, note TEXT)")

TYPES_ROWS = [
    # id, iv, rv, tv, bv, av
    (1, 0, 0.0, "", b"", 42),
    (2, 1, 1.5, "héllo ✓", b"\x00\xff", "text"),
    (3, -1, -2.75, "line\nbreak", b"\x01", b"\x01\x02"),
    (4, 127, 1e300, "tab\there", b"SQLite", 3.5),
    (5, -32768, -1e-300, "日本語", bytes(range(16)), None),
    (6, 8388607, 3.141592653589793, "\U0001f680 rocket", None, 0),
    (7, -2147483648, 2.0, None, b"\x7f\x80", 1),
    (8, 140737488355327, -0.0, "ascii", b"", -1),
    (2 ** 40, 9007199254740993, 0.5, "big rowid", b"\xfe", 9223372036854775807),
]
conn.executemany(
    "INSERT INTO Types (id, iv, rv, tv, bv, av) VALUES (?, ?, ?, ?, ?, ?)",
    TYPES_ROWS)
conn.commit()
conn.close()
assert struct.unpack(">H", header(path)[16:18])[0] == 4096
report(path)

# --- overflow.db: payloads spanning overflow pages ------------------------------
# page_size 512 => usable U = 512, inline max X = U - 35 = 477,
# minimum inline M = ((U - 12) * 32 / 255) - 23 = 39,
# K = M + (P - M) mod (U - 4).
# Blobs: rowid r holds r data bytes of pattern (i + r) mod 251, so the total
# payload P = r + 4 (1-byte header length, serial type 0 for the aliased id,
# 2-byte varint for the blob serial type). The rowid sweep covers, in order:
# fully inline (P <= 477), overflow with K > X so only M = 39 bytes stay inline
# (P 478..546), overflow with K <= X (P 547..985), and the second wrap of the
# mod where K > X again with a 2-page chain (P 986..1054). Rowid 100000 forces
# a long chain (~197 overflow pages) and a 3-byte serial-type varint.
# Texts: multi-byte UTF-8 sequences that straddle overflow page boundaries and
# must be reassembled before decoding.

conn, path = fresh("overflow.db", page_size=512)
conn.execute("CREATE TABLE Blobs (id INTEGER PRIMARY KEY, data BLOB)")
conn.execute("CREATE TABLE Texts (id INTEGER PRIMARY KEY, data TEXT)")

BLOB_SIZES = (list(range(466, 597))        # P = 470..600: inline / K>X / K<=X
              + list(range(971, 992))      # P = 975..995: second K>X wrap
              + [100000])                  # long overflow chain
for n in BLOB_SIZES:
    conn.execute("INSERT INTO Blobs (id, data) VALUES (?, ?)",
                 (n, bytes((i + n) % 251 for i in range(n))))

TEXTS_ROWS = [
    (1, "a" * 400),                                     # inline
    (2, "β" * 300),                                # 600 bytes, 2-byte chars
    (3, "✓" * 2000),                               # 6000 bytes, 3-byte chars
    (4, "x" * 20000),
    (5, "".join(chr(0x3041 + i % 80) for i in range(3000))),
]
conn.executemany("INSERT INTO Texts (id, data) VALUES (?, ?)", TEXTS_ROWS)
conn.commit()
conn.close()
assert struct.unpack(">H", header(path)[16:18])[0] == 512
report(path)

# --- btree.db: multi-level table b-tree ------------------------------------------
# 5000 small rows on 512-byte pages force leaf -> interior -> root interior
# (three levels). Rows with id % 500 == 0 are then deleted, so traversal must
# skip their slots and return exactly 4990 rows in ascending rowid order.

conn, path = fresh("btree.db", page_size=512)
conn.execute("CREATE TABLE Many (id INTEGER PRIMARY KEY, val TEXT)")
conn.executemany("INSERT INTO Many (id, val) VALUES (?, ?)",
                 [(i, f"row-{i:05d}") for i in range(1, 5001)])
conn.execute("DELETE FROM Many WHERE id % 500 = 0")
conn.commit()
rootpage = conn.execute(
    "SELECT rootpage FROM sqlite_master WHERE name = 'Many'").fetchone()[0]
conn.close()

# confirm the tree really has three levels: interior root over interior children
with open(path, "rb") as f:
    data = f.read()
root_off = (rootpage - 1) * 512
assert data[root_off] == 5, "root of Many is not an interior page"
first_cell = root_off + struct.unpack(">H", data[root_off + 12:root_off + 14])[0]
first_child = struct.unpack(">I", data[first_cell:first_cell + 4])[0]
assert data[(first_child - 1) * 512] == 5, "Many b-tree is only two levels deep"
report(path)

# --- ddl.db: CREATE TABLE parsing edge cases -------------------------------------
# Quoted: all four identifier quoting styles.
# Constraints: parenthesised type, CHECK/DEFAULT with nested parens and commas,
#   and table-level CONSTRAINT/UNIQUE/FOREIGN KEY/CHECK clauses that must not
#   become columns. UNIQUE also creates a sqlite_autoindex_* master entry
#   (type 'index') that must not appear in the navigation table.
# AddedCols: ALTER TABLE ADD COLUMN leaves the first rows as short records.
# Seq: AUTOINCREMENT creates the sqlite_sequence system table, which must be
#   filtered out; multi-line DDL exercises whitespace normalisation.
# Wide: 65 columns; 64 two-byte serial-type varints push the record header
#   length past 127 so the header-length varint itself needs two bytes.
# NoRowid: WITHOUT ROWID, must surface an error in its Data cell.
# A view must not appear in the navigation table.

conn, path = fresh("ddl.db")
conn.execute("""CREATE TABLE Quoted (
    "double quoted" TEXT,
    [bracketed] INTEGER,
    `backticked` REAL,
    'single quoted' BLOB
)""")
conn.execute("INSERT INTO Quoted VALUES ('dq', 1, 1.5, x'AB')")
conn.execute("INSERT INTO Quoted VALUES (NULL, -2, NULL, x'')")

conn.execute("""CREATE TABLE Constraints (
    id INTEGER NOT NULL,
    amount DECIMAL(10,2) CHECK (amount > 0 AND amount < 9999.99),
    label TEXT DEFAULT ('a,b(c)'),
    other INTEGER,
    CONSTRAINT pk PRIMARY KEY (id, other),
    UNIQUE (amount, label),
    FOREIGN KEY (other) REFERENCES Quoted ([bracketed]),
    CHECK (id <> 42)
)""")
conn.execute("INSERT INTO Constraints VALUES (1, 12.5, 'x', 10)")
conn.execute("INSERT INTO Constraints VALUES (2, 0.01, 'a,b(c)', 20)")

conn.execute("CREATE TABLE AddedCols (a INTEGER, b TEXT)")
conn.execute("INSERT INTO AddedCols VALUES (1, 'one')")
conn.execute("INSERT INTO AddedCols VALUES (2, 'two')")
conn.execute("ALTER TABLE AddedCols ADD COLUMN c REAL")
conn.execute("ALTER TABLE AddedCols ADD COLUMN d TEXT")
conn.execute("INSERT INTO AddedCols VALUES (3, 'three', 3.5, 'full')")

conn.execute("""CREATE TABLE Seq (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    v TEXT
)""")
conn.execute("INSERT INTO Seq (v) VALUES ('first')")
conn.execute("INSERT INTO Seq (v) VALUES ('second')")
conn.execute("DELETE FROM Seq WHERE id = 2")
conn.execute("INSERT INTO Seq (v) VALUES ('third')")   # takes id 3, not 2

wide_cols = ", ".join(f"c{i:02d} TEXT" for i in range(1, 65))
conn.execute(f"CREATE TABLE Wide (id INTEGER PRIMARY KEY, {wide_cols})")
for r in range(1, 4):
    vals = [f"r{r}c{i:02d}".ljust(60, ".") for i in range(1, 65)]
    conn.execute(
        "INSERT INTO Wide VALUES (?" + ", ?" * 64 + ")", [r] + vals)

conn.execute(
    "CREATE TABLE NoRowid (k TEXT PRIMARY KEY, v INTEGER) WITHOUT ROWID")
conn.execute("INSERT INTO NoRowid VALUES ('a', 1)")

conn.execute("CREATE VIEW SeqView AS SELECT id, v FROM Seq")
conn.commit()
conn.close()
report(path)

# --- utf16le.db / utf16be.db: non-UTF-8 database encodings -----------------------
# Same logical content in both. The DDL text in sqlite_master is stored in the
# database encoding too, so column-name parsing exercises the decoder as well.

U_ROWS = [
    (1, "plain ascii"),
    (2, "héllo ✓"),
    (3, "日本語テキスト"),
    (4, "emoji \U0001f680\U0001f30d"),        # surrogate pairs in UTF-16
    (5, ""),
    (6, None),
]
for name, enc, encid in [("utf16le.db", "UTF-16le", 2),
                         ("utf16be.db", "UTF-16be", 3)]:
    conn, path = fresh(name, page_size=512, encoding=enc)
    conn.execute("CREATE TABLE U (id INTEGER PRIMARY KEY, txt TEXT)")
    conn.executemany("INSERT INTO U (id, txt) VALUES (?, ?)", U_ROWS)
    conn.commit()
    conn.close()
    assert struct.unpack(">I", header(path)[56:60])[0] == encid, name
    report(path)

# --- wal.db: WAL-mode database ---------------------------------------------------
# journal_mode=WAL sets the header read/write version bytes to 2, which the
# reader surfaces as the SQLite.WalMode metadata flag. The WAL is checkpointed
# and removed on close, so the main file holds all committed data.

conn, path = fresh("wal.db", page_size=512)
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("CREATE TABLE Log (id INTEGER PRIMARY KEY, msg TEXT)")
conn.executemany("INSERT INTO Log (id, msg) VALUES (?, ?)",
                 [(i, f"entry {i}") for i in range(1, 6)])
conn.commit()
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
conn.close()
assert not (HERE / "wal.db-wal").exists()
h = header(path)
assert h[18] == 2 and h[19] == 2, "wal.db header is not marked WAL"
report(path)

# --- empty.db: database whose schema is empty ------------------------------------
# A table is created and dropped, leaving page 1 as a leaf with zero cells.
# The navigation table must come back with zero rows, not an error.

conn, path = fresh("empty.db", page_size=512)
conn.execute("CREATE TABLE Tmp (x INTEGER)")
conn.execute("INSERT INTO Tmp VALUES (1)")
conn.execute("DROP TABLE Tmp")
conn.commit()
conn.execute("VACUUM")
conn.close()
report(path)

# --- round-trip check with the reference implementation --------------------------

EXPECT = {
    "types.db": {"Types": 9, "Empty": 0},
    "overflow.db": {"Blobs": len(BLOB_SIZES), "Texts": 5},
    "btree.db": {"Many": 4990},
    "ddl.db": {"Quoted": 2, "Constraints": 2, "AddedCols": 3, "Seq": 2,
               "Wide": 3, "NoRowid": 1},
    "utf16le.db": {"U": 6},
    "utf16be.db": {"U": 6},
    "wal.db": {"Log": 5},
    "empty.db": {},
}
for name, tables in EXPECT.items():
    conn = sqlite3.connect(HERE / name)
    found = {r[0] for r in conn.execute(
        "SELECT name FROM sqlite_master WHERE type = 'table'"
        " AND name NOT LIKE 'sqlite_%'")}
    assert found == set(tables), (name, found)
    for t, n in tables.items():
        got = conn.execute(f'SELECT count(*) FROM "{t}"').fetchone()[0]
        assert got == n, (name, t, got)
    conn.close()
print("round-trip OK")