probe10.py

#!/usr/bin/env python3
"""probe10.py - deleted tuples, and the three objects that describe a relation.

A DELETE does not rewrite Data Blocks. The rows stay where they are, the
`Relation_Header` keeps counting them, and a `Relation_Metadata` records which
ones are gone. Read the blocks and ignore the tombstones and a relation comes
back with its deleted rows still in it - right values, too many of them, which
is exactly the failure mode a spot check survives. The whole corpus had this
hidden, because nothing in it had ever deleted a row: all 111 `Relation_Metadata`
objects were the single byte `00`.

What this probe pins down:

  header      `Relation_Header` is `u8 tuple_count` then **one u4 per block**,
              not a fixed pair of fields. Swept over 1-4 blocks and an empty
              relation, which is the case that proves it is an array: 8 bytes
              and no entries.
  tombstones  the `Relation_Metadata` B-tree, at every depth the engine uses -
              a bare leaf, one internal node, and three levels - checked by
              decoding the relation and comparing to the API.
  boundary    deletions either side of a block boundary, because the row
              indices are relation-global and a block-local reading passes the
              single-block cases and then silently keeps the wrong rows.

  /usr/bin/python3 probe10.py
"""
import os
import shutil
import struct
import sys
from collections import Counter

import hyperparse as hp

try:
    from tableauhyperapi import (HyperProcess, Connection, Telemetry, CreateMode,
                                 TableDefinition, SqlType, NULLABLE, TableName,
                                 Inserter)
except ImportError:
    sys.exit("needs tableauhyperapi; try /usr/bin/python3")

HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "probes10")
CAP = 131072              # rows per block


def build(proc, name, rows, where=None):
    path = os.path.join(OUT, f"{name}.hyper")
    tbl = TableName("E", "T")
    cols = [TableDefinition.Column("c", SqlType.big_int(), NULLABLE),
            TableDefinition.Column("t", SqlType.text(), NULLABLE)]
    with Connection(proc.endpoint, path, CreateMode.CREATE_AND_REPLACE) as c:
        c.catalog.create_schema("E")
        c.catalog.create_table(TableDefinition(tbl, cols))
        if rows:
            with Inserter(c, TableDefinition(tbl, cols)) as ins:
                ins.add_rows([[i, "v%06d" % (i % 997)] for i in range(rows)])
                ins.execute()
        if where:
            c.execute_command(f'DELETE FROM "E"."T" WHERE {where}')
    return path


def headers(path):
    d = open(path, "rb").read()
    f = hp.parse(path)
    out = []
    for e in f.directory.entries:
        if e.category_name == "Relation_Header":
            b = hp.read_object(d, e)
            n = (len(b) - 8) // 4
            out.append((len(b), _u8(b), struct.unpack_from(f"<{n}I", b, 8)))
    return out


def _u8(b):
    return struct.unpack_from("<Q", b, 0)[0]


def rows_match(path, proc):
    """Decode every column and compare whole rows against the API."""
    d = open(path, "rb").read()
    f = hp.parse(path)
    attrs = hp.relations(f.live_catalog)[0]["attributes"]
    cols = []
    for ci, a in enumerate(attrs):
        v, note = hp.decode_relation_column(d, f, 0, ci + 1,
                                            hp.ColType.parse(a.get("type")))
        if note and not v:
            return None, 0, note
        cols.append(v)
    mine = Counter(tuple(c[r] for c in cols) for r in range(len(cols[0])))
    with Connection(proc.endpoint, path) as conn:
        truth = Counter(tuple(r) for r in
                        conn.execute_list_query('SELECT "c","t" FROM "E"."T"'))
    return mine == truth, sum(truth.values()), ""


def main():
    shutil.rmtree(OUT, ignore_errors=True)
    os.makedirs(OUT, exist_ok=True)
    ok = bad = 0
    with HyperProcess(telemetry=Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU,
                      parameters={"default_database_version": "4"}) as proc:
        print("Relation_Header: u8 tuple_count, then one u4 per block")
        for nm, n in (("empty", 0), ("b1", CAP - 1), ("b1x", CAP),
                      ("b2", CAP + 1), ("b3", 2 * CAP + 1), ("b4", 3 * CAP + 1)):
            p = build(proc, f"hdr_{nm}", n)
            blocks = len(hp.column_objects(hp.parse(p), 0, 1))
            (size, rows, per), = headers(p)
            good = len(per) == blocks and rows == n and set(per) <= {2}
            print(f"  {nm:<6} rows={n:<7} blocks={blocks}  header={size:>2}B "
                  f"per-block={per}  {'ok' if good else 'UNEXPECTED'}")
            ok += good
            bad += not good

        print("\nRelation_Metadata: the tombstone B-tree")
        cases = {
            "none":       (100, None),
            "one":        (100, '"c" = 7'),
            "ranges":     (100, '("c" >= 10 AND "c" < 20) OR ("c" >= 50 AND "c" < 55)'),
            "leaf100":    (4000, '"c" % 2 = 0 AND "c" < 200'),
            # 200 singleton ranges: one internal node over three leaves
            "internal":   (4000, '"c" % 2 = 0 AND "c" < 400'),
            "evens":      (5000, '"c" % 2 = 0'),
            # deletions either side of two block boundaries
            "boundary":   (2 * CAP + 10,
                           '"c" IN (5,6,7, 131071,131072,131073, 262143,262144,262145)'),
            # ~87k ranges: three levels
            "deep":       (2 * CAP + 10, '"c" % 3 = 0'),
            "all":        (1000, '1 = 1'),
        }
        for nm, (n, where) in cases.items():
            p = build(proc, f"del_{nm}", n, where)
            d = open(p, "rb").read()
            f = hp.parse(p)
            meta = hp.relation_metadata(f, 0)
            size = len(hp.read_object(d, meta[0])) if meta else 0
            ranges, note = hp.deleted_rows(d, f, 0)
            good, live, err = rows_match(p, proc)
            print(f"  {nm:<10} rows={n:<7} live={live:<7} metadata={size:>9,}B "
                  f"ranges={len(ranges):<6} "
                  f"{'ok' if good else 'FAIL ' + (note or err or 'rows differ')}")
            ok += bool(good)
            bad += not good
    print(f"\nTOTAL: {ok} pass, {bad} fail -> {OUT}")
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())