checkrows.py

#!/usr/bin/env python3
"""checkrows.py - validate hyperparse's row decoder against the Hyper API.

Run with the interpreter that has tableauhyperapi (here: /usr/bin/python3).

Compares per-column value *multisets*, not sequences: Hyper reorders rows when
it freezes a block, so positional comparison is meaningless. A multiset match
over every column of every relation is still a strong check - it catches wrong
code widths, wrong bit order, off-by-one dictionary indexing and misplaced
NULLs, all of which change the histogram.

  /usr/bin/python3 checkrows.py FILE...
"""
import decimal
import os
import sys
from collections import Counter

import hyperparse as hp

try:
    from tableauhyperapi import HyperProcess, Connection, Telemetry
except ImportError:
    sys.exit("needs tableauhyperapi; try /usr/bin/python3")


def normalise(v):
    """Make oracle and parser values comparable across type representations."""
    if v is None:
        return None
    if isinstance(v, bool):
        return bool(v)
    if isinstance(v, float):
        return round(v, 6)
    # hyperparse keeps the column's declared scale (a NUMERIC(18,4) reads back
    # as -1000.5000); the API's binding normalises to -1000.5. Same number, so
    # compare numerically rather than by str().
    if isinstance(v, decimal.Decimal):
        return v.normalize()
    # Build the oracle's Date/Timestamp canonically from their fields. Never
    # str() them: the binding renders via datetime.datetime, which caps at year
    # 9999 and raises on the wider range Hyper itself accepts - and it prints a
    # year below 1000 unpadded, where ISO 8601 wants four digits.
    if all(hasattr(v, k) for k in ("year", "month", "day")):
        out = f"{v.year:04d}-{v.month:02d}-{v.day:02d}"
        if not hasattr(v, "hour"):
            return out
        out += f" {v.hour:02d}:{v.minute:02d}:{v.second:02d}"
        if v.microsecond:
            out += f".{v.microsecond:06d}"
        try:
            if v.tzinfo is not None:
                out += "+00:00"
        except Exception:
            pass
        return out
    # The API hands back its own Interval objects; compare them structurally
    # against hyperparse's IntervalValue rather than through str().
    if all(hasattr(v, k) for k in ("months", "days", "microseconds")):
        return (v.months, v.days, v.microseconds)
    if isinstance(v, int):
        return int(v)
    return str(v)


def check(path, proc):
    d = open(path, "rb").read()
    f = hp.parse(path, want_blocks=True)
    cat = f.live_catalog
    if not cat or not f.directory:
        print(f"{path}: no catalog/directory"); return 0, 0
    rels = hp.relations(cat)
    ok = bad = 0
    with Connection(proc.endpoint, path) as conn:
        for ri, rel in enumerate(rels):
            attrs = rel.get("attributes", [])
            tbl = f'"{rel["_schema"]}"."{rel["name"]}"'
            for ci, a in enumerate(attrs):
                if not hp.column_objects(f, ri, ci + 1):
                    # An empty relation has no Data Block at all - the engine
                    # drops them, both for a table never inserted into and for
                    # one whose rows were all deleted. Zero rows is the right
                    # answer there, not a missing block.
                    n = conn.execute_scalar_query(f"SELECT COUNT(*) FROM {tbl}")
                    if n == 0:
                        ok += 1
                    else:
                        print(f"  MISS  {tbl}.{a['name']}: no block for {n} rows")
                        bad += 1
                    continue
                sql_type = (a.get("type") or [None])[0]
                vals, note = hp.decode_relation_column(
                    d, f, ri, ci + 1, hp.ColType.parse(a.get("type")))
                if note and not vals:
                    print(f"  SKIP  {tbl}.{a['name']} ({sql_type}): {note}")
                    bad += 1
                    continue
                col = f'"{a["name"]}"'
                truth = [r[0] for r in conn.execute_list_query(
                    f"SELECT {col} FROM {tbl}")]
                got, want = Counter(map(normalise, vals)), Counter(map(normalise, truth))
                if got == want:
                    ok += 1
                else:
                    bad += 1
                    only_got = (got - want).most_common(3)
                    only_want = (want - got).most_common(3)
                    schemes = sorted(set(
                        hp.parse_block_index(hp.read_object(d, e)).scheme
                        for e in hp.column_objects(f, ri, ci + 1)))
                    print(f"  DIFF  {tbl}.{a['name']} ({sql_type}, "
                          f"schemes={schemes}): {len(vals)} vs {len(truth)} rows; "
                          f"parser-only={only_got} oracle-only={only_want}")
    return ok, bad


def main():
    tot_ok = tot_bad = 0
    with HyperProcess(telemetry=Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU) as proc:
        for p in sys.argv[1:]:
            print(f"\n{p}")
            fn = check_tuples if os.environ.get("TUPLES") else check
            o, b = fn(p, proc)
            print(f"  -> {o} columns match, {b} fail")
            tot_ok += o
            tot_bad += b
    print(f"\nTOTAL: {tot_ok} {'relations' if os.environ.get('TUPLES') else 'columns'} match, {tot_bad} fail")
    return 1 if tot_bad else 0



def check_tuples(path, proc):
    """Stronger check: whole rows, not per-column multisets. Column-wise
    histograms can all match while the columns are mis-aligned into rows."""
    d = open(path, "rb").read()
    f = hp.parse(path, want_blocks=True)
    cat = f.live_catalog
    if not cat or not f.directory:
        return 0, 0
    ok = bad = 0
    with Connection(proc.endpoint, path) as conn:
        for ri, rel in enumerate(hp.relations(cat)):
            attrs = rel.get("attributes", [])
            tbl = f'"{rel["_schema"]}"."{rel["name"]}"'
            cols = []
            empty = False
            for ci, a in enumerate(attrs):
                if not hp.column_objects(f, ri, ci + 1):
                    # see check(): an empty relation legitimately has no blocks
                    empty = conn.execute_scalar_query(
                        f"SELECT COUNT(*) FROM {tbl}") == 0
                    cols = None
                    break
                v, note = hp.decode_relation_column(
                    d, f, ri, ci + 1, hp.ColType.parse(a.get("type")))
                if note and not v:
                    cols = None; break
                cols.append(v)
            if not cols:
                if empty:
                    ok += 1
                else:
                    print(f"  TUPLE-SKIP {tbl}"); bad += 1
                continue
            mine = Counter(tuple(normalise(c[r]) for c in cols)
                           for r in range(len(cols[0])))
            sel = ", ".join(f'"{a["name"]}"' for a in attrs)
            truth = Counter(tuple(map(normalise, r))
                            for r in conn.execute_list_query(f"SELECT {sel} FROM {tbl}"))
            if mine == truth:
                ok += 1
            else:
                bad += 1
                print(f"  TUPLE-DIFF {tbl}: {sum((mine - truth).values())} rows differ")
                for t in list((mine - truth))[:2]:
                    print(f"      parser: {t}")
                for t in list((truth - mine))[:2]:
                    print(f"      oracle: {t}")
    return ok, bad

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