informix-db/tests/benchmarks/compare/ifxpy_bench.py

"""IfxPy comparison benchmark.

Runs the same workloads as ``tests/benchmarks/test_*_perf.py`` against
the same dev-container Informix instance, but using IfxPy (the C-bound
PyPI driver) instead of ``informix-db``. Numbers go straight to stdout;
the host parses them and produces a side-by-side table.

Workloads:
* ``select_one_row`` — single-row SELECT round-trip latency
* ``select_systables_first_10`` — small server-side query
* ``select_bench_table_all`` — 1k-row sustained fetch
* ``executemany_1000_rows_in_txn`` — bulk INSERT throughput
* ``cold_connect_disconnect`` — login handshake cost

Each workload runs N times; we report mean and stddev.
"""

from __future__ import annotations

import statistics
import sys
import time
from collections.abc import Callable

import IfxPy

# Connect string — mirrors the conftest.py defaults the host uses.
CONN_STR = (
    "SERVER=informix;"
    "DATABASE=sysmaster;"
    "HOST=127.0.0.1;"
    "SERVICE=9088;"
    "UID=informix;"
    "PWD=in4mix;"
    "PROTOCOL=onsoctcp"
)

ROUNDS_FAST = 100  # for sub-millisecond ops
ROUNDS_MED = 20  # for 1-100ms ops
ROUNDS_SLOW = 3  # for >1s ops


def measure(name: str, rounds: int, body: Callable[[], None]) -> dict:
    """Run ``body`` ``rounds`` times; return mean/stddev/min/max in seconds."""
    timings = []
    for _ in range(rounds):
        t0 = time.perf_counter()
        body()
        t1 = time.perf_counter()
        timings.append(t1 - t0)
    return {
        "name": name,
        "rounds": rounds,
        "mean_s": statistics.mean(timings),
        "stddev_s": statistics.stdev(timings) if len(timings) > 1 else 0.0,
        "min_s": min(timings),
        "max_s": max(timings),
    }


def bench_select_one_row(conn) -> dict:
    def run() -> None:
        stmt = IfxPy.exec_immediate(
            conn, "SELECT 1 FROM systables WHERE tabid = 1"
        )
        IfxPy.fetch_tuple(stmt)
        IfxPy.free_stmt(stmt)

    return measure("select_one_row", ROUNDS_FAST, run)


def bench_select_systables_first_10(conn) -> dict:
    def run() -> None:
        stmt = IfxPy.exec_immediate(
            conn,
            "SELECT FIRST 10 tabname, owner, tabid, ncols FROM systables",
        )
        while IfxPy.fetch_tuple(stmt):
            pass
        IfxPy.free_stmt(stmt)

    return measure("select_systables_first_10", ROUNDS_FAST, run)


def bench_select_bench_table_all(conn) -> dict:
    """Requires p21_bench table to exist (created by host-side fixture)."""
    # Probe whether the table exists; if not, skip
    try:
        stmt = IfxPy.exec_immediate(conn, "SELECT COUNT(*) FROM p21_bench")
        row = IfxPy.fetch_tuple(stmt)
        IfxPy.free_stmt(stmt)
        if not row or row[0] == 0:
            return {"name": "select_bench_table_all", "skipped": "p21_bench empty"}
    except Exception as e:
        return {"name": "select_bench_table_all", "skipped": f"p21_bench missing: {e}"}

    def run() -> None:
        stmt = IfxPy.exec_immediate(conn, "SELECT * FROM p21_bench")
        while IfxPy.fetch_tuple(stmt):
            pass
        IfxPy.free_stmt(stmt)

    return measure("select_bench_table_all", ROUNDS_MED, run)


def bench_executemany_1000_rows_in_txn() -> dict:
    """Open a connection on testdb, autocommit OFF, executemany 1000."""
    try:
        conn = IfxPy.connect(
            CONN_STR.replace("DATABASE=sysmaster", "DATABASE=testdb"), "", ""
        )
    except Exception as e:
        return {"name": "executemany_1000_rows_in_txn", "skipped": f"testdb: {e}"}
    IfxPy.autocommit(conn, IfxPy.SQL_AUTOCOMMIT_OFF)

    table = "p21_ifxpy_bench"
    try:
        try:
            IfxPy.exec_immediate(conn, f"DROP TABLE {table}")
            IfxPy.commit(conn)
        except Exception:
            pass
        IfxPy.exec_immediate(
            conn, f"CREATE TABLE {table} (id INT, name VARCHAR(64), value FLOAT)"
        )
        IfxPy.commit(conn)

        counter = [0]

        def run() -> None:
            counter[0] += 1
            base = counter[0] * 1000
            stmt = IfxPy.prepare(
                conn, f"INSERT INTO {table} VALUES (?, ?, ?)"
            )
            for i in range(1000):
                IfxPy.execute(stmt, (base + i, f"row_{base + i}", float(base + i)))
            IfxPy.free_stmt(stmt)
            IfxPy.commit(conn)

        result = measure("executemany_1000_rows_in_txn", ROUNDS_SLOW, run)
        return result
    finally:
        try:
            IfxPy.exec_immediate(conn, f"DROP TABLE {table}")
            IfxPy.commit(conn)
        except Exception:
            pass
        IfxPy.close(conn)


def bench_cold_connect_disconnect() -> dict:
    def run() -> None:
        conn = IfxPy.connect(CONN_STR, "", "")
        IfxPy.close(conn)

    return measure("cold_connect_disconnect", ROUNDS_SLOW, run)


def main() -> None:
    print("# IfxPy benchmark results", file=sys.stderr)
    print(f"# IfxPy version: {IfxPy.__version__ if hasattr(IfxPy, '__version__') else 'unknown'}", file=sys.stderr)

    # Persistent connection for the read-mostly benchmarks
    conn = IfxPy.connect(CONN_STR, "", "")

    results = []
    results.append(bench_select_one_row(conn))
    results.append(bench_select_systables_first_10(conn))
    results.append(bench_select_bench_table_all(conn))
    IfxPy.close(conn)

    results.append(bench_executemany_1000_rows_in_txn())
    results.append(bench_cold_connect_disconnect())

    # Emit machine-parseable lines on stdout
    for r in results:
        if r.get("skipped"):
            print(f"SKIP {r['name']}: {r['skipped']}")
        else:
            print(
                f"RESULT {r['name']} mean={r['mean_s']:.6f}s "
                f"stddev={r['stddev_s']:.6f}s min={r['min_s']:.6f}s "
                f"max={r['max_s']:.6f}s rounds={r['rounds']}"
            )


if __name__ == "__main__":
    main()