#!/usr/bin/env python3
"""
storage_engine comprehensive test suite.

Covers all features before PGXN publication:
  - Extension lifecycle (install, schema, catalog objects)
  - colcompress and rowcompress DML
  - Compression codec options (zstd, lz4, pglz)
  - Sort key (orderby) option
  - Vectorized aggregates — correctness (VEC OFF == VEC ON) for every type
      smallint, integer, bigint, float8, numeric, money, date
  - count(*) and count(col)
  - NULL handling (empty table, all-NULL column, mixed NULLs)
  - EXPLAIN plan verification (StorageEngineVectorAgg node)
  - EXPLAIN ANALYZE correctness (not blocked by IsExplainQuery)
  - Parallel safety (AGGSPLIT_SIMPLE guard)
  - Multi-column aggregate query
  - Upgrade path chain traversability

Usage:
    python3 tests/test_suite.py                     # PG@5432 only
    python3 tests/test_suite.py --pg19              # PG@5432 + PG@5433
    python3 tests/test_suite.py --port 5433         # single custom port
    python3 tests/test_suite.py --port 5432 --pg19 --pg19-port 5433
"""

import argparse
import json
import re
import subprocess
import sys

GREEN  = "\033[32m"
RED    = "\033[31m"
YELLOW = "\033[33m"
BOLD   = "\033[1;36m"
RESET  = "\033[0m"

PASS_LABEL = f"{GREEN}PASS{RESET}"
FAIL_LABEL = f"{RED}FAIL{RESET}"


# ---------------------------------------------------------------------------
# Runner
# ---------------------------------------------------------------------------

class TestRunner:
    def __init__(self, port: int, label: str = ""):
        self.port    = port
        self.label   = label or f"PG@{port}"
        self.dbname  = "storage_engine_test"
        self.passed  = 0
        self.failed  = 0
        self.failures: list[str] = []

    # ------------------------------------------------------------------ psql

    def _run(self, sql: str, dbname: str | None = None,
             tuples_only: bool = True) -> tuple[str, int, str]:
        """Execute SQL via psql. Returns (stdout, returncode, stderr)."""
        db   = dbname or self.dbname
        flag = ["-Atc"] if tuples_only else ["-v", "ON_ERROR_STOP=1", "-c"]
        cmd  = ["sudo", "-u", "postgres", "psql",
            "-p", str(self.port), "-d", db, "--no-psqlrc"] + flag + [sql]
        r    = subprocess.run(cmd, capture_output=True, text=True)
        # Use rstrip('\n') not strip(): preserves the empty line that represents
        # a NULL value in tuples-only mode (strip() would destroy it).
        return r.stdout.rstrip("\n"), r.returncode, r.stderr.strip()

    def q(self, sql: str, dbname: str | None = None) -> str:
        """Return all output rows joined by newline."""
        out, _rc, _err = self._run(sql, dbname=dbname)
        return "\n".join(line for line in out.split("\n") if line != "SET")

    def q1(self, sql: str, dbname: str | None = None) -> str:
        """Return first column of LAST row (empty string for NULL).

        Using the last line handles the common pattern of prepending SET
        statements to a SELECT: psql outputs one 'SET' line per GUC before
        the actual query result, so taking the last line gives the result.
        """
        out = self.q(sql, dbname=dbname)
        if not out:
            return ""
        lines = out.split("\n")
        return lines[-1].split("|")[0]

    def exec(self, sql: str, dbname: str | None = None) -> tuple[int, str]:
        """Execute DDL/DML. Returns (returncode, stderr)."""
        _out, rc, err = self._run(sql, dbname=dbname, tuples_only=False)
        return rc, err

    def _execution_time_ms(self, explain_output: str) -> float | None:
        """Extract execution time in ms from EXPLAIN ANALYZE output."""
        # Prefer JSON plans for locale-independent parsing.
        lines = [ln for ln in explain_output.splitlines() if ln and ln != "SET"]
        if lines:
            raw = "\n".join(lines)
            try:
                payload = json.loads(raw)
                if isinstance(payload, list) and payload and isinstance(payload[0], dict):
                    t = payload[0].get("Execution Time")
                    if isinstance(t, (int, float)):
                        return float(t)
            except Exception:
                pass

        patterns = [
            r"Execution Time:\\s*([0-9]+(?:[\\.,][0-9]+)?)\\s*ms",
            r"Execution time:\\s*([0-9]+(?:[\\.,][0-9]+)?)\\s*ms",
            r"Total runtime:\\s*([0-9]+(?:[\\.,][0-9]+)?)\\s*ms",
            r"Tempo de Execu(?:c|ç)[aã]o:\\s*([0-9]+(?:[\\.,][0-9]+)?)\\s*ms",
            r"Tempo de execu(?:c|ç)[aã]o:\\s*([0-9]+(?:[\\.,][0-9]+)?)\\s*ms",
        ]
        for pat in patterns:
            m = re.search(pat, explain_output)
            if m:
                return float(m.group(1).replace(",", "."))
        return None

    # ------------------------------------------------------------------ assertions

    def check(self, name: str, cond: bool, detail: str = "") -> None:
        if cond:
            print(f"  [{PASS_LABEL}] {name}")
            self.passed += 1
        else:
            detail_str = f"\n         {detail}" if detail else ""
            print(f"  [{FAIL_LABEL}] {name}{detail_str}")
            self.failed += 1
            tag = f"{self.label}: {name}"
            self.failures.append(tag + (f" — {detail}" if detail else ""))

    def check_eq(self, name: str, got: str, expected: str) -> None:
        self.check(name, got.strip() == expected.strip(),
                   f"got={got!r}  expected={expected!r}")

    def agg_ok(self, label: str, sql: str) -> None:
        """Assert that VEC OFF and VEC ON produce identical output."""
        pfx_off = ("SET max_parallel_workers_per_gather=0; "
                   "SET storage_engine.enable_vectorization=off;")
        pfx_on  = ("SET max_parallel_workers_per_gather=0; "
                   "SET storage_engine.enable_vectorization=on;")
        off = self.q(f"{pfx_off} {sql}")
        on  = self.q(f"{pfx_on}  {sql}")
        self.check(label, off == on, f"OFF={off!r}  ON={on!r}")

    def section(self, title: str) -> None:
        print(f"\n{BOLD}{'─' * 62}{RESET}")
        print(f"{BOLD}  {self.label}: {title}{RESET}")
        print(f"{BOLD}{'─' * 62}{RESET}")

    # ------------------------------------------------------------------ setup / teardown

    def setup(self) -> None:
        self._run(f"DROP DATABASE IF EXISTS {self.dbname}", dbname="postgres")
        self._run(f"CREATE DATABASE {self.dbname}", dbname="postgres")
        rc, err = self.exec("CREATE EXTENSION storage_engine")
        if rc != 0:
            print(f"{RED}FATAL: CREATE EXTENSION failed — {err}{RESET}")
            sys.exit(1)

    def teardown(self) -> None:
        self._run(f"DROP DATABASE IF EXISTS {self.dbname}", dbname="postgres")

    # ================================================================== tests

    def test_extension_lifecycle(self) -> None:
        self.section("Extension Lifecycle")

        # Version
        ver = self.q1("SELECT extversion FROM pg_extension WHERE extname='storage_engine'")
        self.check("extension version = 1.3.4", ver == "1.3.4", f"got {ver!r}")

        # Schema
        ns = self.q1("SELECT nspname FROM pg_namespace WHERE nspname='engine'")
        self.check("schema engine exists", ns == "engine")

        # Catalog tables
        for tbl in ("col_options", "stripe", "chunk_group", "chunk"):
            cnt = self.q1(
                f"SELECT count(*) FROM pg_class c "
                f"JOIN pg_namespace n ON n.oid = c.relnamespace "
                f"WHERE c.relname = '{tbl}' AND n.nspname = 'engine'"
            )
            self.check(f"catalog table engine.{tbl} exists", cnt == "1")

        # Table access methods
        for am in ("colcompress", "rowcompress"):
            cnt = self.q1(f"SELECT count(*) FROM pg_am WHERE amname = '{am}'")
            self.check(f"access method {am} registered", cnt == "1")

        # Key aggregate names (spot-check: one per type group)
        spot_aggs = [
            ("vcount",  "count"),
            ("vmin",    "integer"),
            ("vmax",    "smallint"),
            ("vsum",    "bigint"),
            ("vavg",    "bigint"),
            ("vmin",    "date"),
            ("vsum",    "double precision"),
            ("vavg",    "double precision"),
            ("vmin",    "numeric"),
            ("vavg",    "numeric"),
            ("vsum",    "money"),
            ("vmin",    "money"),
            ("vmax",    "money"),
        ]
        for agg_name, arg_type in spot_aggs:
            arg_clause = (
                "p.pronargs = 0"
                if arg_type == "count"
                else f"p.proargtypes::regtype[]::text LIKE '%{arg_type}%'"
            )
            cnt = self.q1(
                f"SELECT count(*) FROM pg_proc p "
                f"JOIN pg_namespace n ON n.oid = p.pronamespace "
                f"WHERE n.nspname = 'engine' AND p.proname = '{agg_name}' "
                f"  AND {arg_clause} "
                f"  AND p.oid IN (SELECT aggfnoid FROM pg_aggregate)"
            )
            lbl = f"aggregate engine.{agg_name}({arg_type}) registered"
            self.check(lbl, cnt != "0", f"count={cnt}")

        # No engine.vavg(money) — PostgreSQL has no avg(money)
        cnt = self.q1(
            "SELECT count(*) FROM pg_proc p "
            "JOIN pg_namespace n ON n.oid = p.pronamespace "
            "WHERE n.nspname = 'engine' AND p.proname = 'vavg' "
            "  AND p.proargtypes::regtype[]::text LIKE '%money%' "
            "  AND p.oid IN (SELECT aggfnoid FROM pg_aggregate)"
        )
        self.check("no engine.vavg(money) — PostgreSQL has no avg(money)", cnt == "0", f"count={cnt}")

    # ------------------------------------------------------------------ DML

    def test_colcompress_dml(self) -> None:
        self.section("colcompress DML")
        N = 50_000

        rc, err = self.exec(f"""
            DROP TABLE IF EXISTS _tc;
            CREATE TABLE _tc (id int, val bigint, label text, d date) USING colcompress;
            INSERT INTO _tc
                SELECT i, i*3, 'label_'||i, '2020-01-01'::date + i
                FROM generate_series(1,{N}) i;
        """)
        self.check("CREATE TABLE + INSERT colcompress", rc == 0, err[:200] if err else "")

        cnt = self.q1("SELECT count(*) FROM _tc")
        self.check_eq(f"row count after INSERT = {N}", cnt, str(N))

        cnt2 = self.q1(f"SELECT count(*) FROM _tc WHERE id > {N // 2}")
        self.check_eq("SELECT with WHERE filter", cnt2, str(N // 2))

        rc, err = self.exec("UPDATE _tc SET label = 'updated' WHERE id = 1")
        self.check("UPDATE single row", rc == 0, err[:200] if err else "")

        upd = self.q1("SELECT label FROM _tc WHERE id = 1")
        self.check_eq("UPDATE value persisted", upd, "updated")

        rc, err = self.exec(f"DELETE FROM _tc WHERE id > {N - 10}")
        self.check("DELETE 10 rows", rc == 0, err[:200] if err else "")

        cnt3 = self.q1("SELECT count(*) FROM _tc")
        self.check_eq(f"count after DELETE = {N - 10}", cnt3, str(N - 10))

        rc, err = self.exec("TRUNCATE _tc")
        self.check("TRUNCATE", rc == 0, err[:200] if err else "")

        cnt4 = self.q1("SELECT count(*) FROM _tc")
        self.check_eq("count after TRUNCATE = 0", cnt4, "0")

        # Re-populate for later aggregate tests
        rc, err = self.exec(f"""
            INSERT INTO _tc
                SELECT i, i*3, 'label_'||i, '2020-01-01'::date + i
                FROM generate_series(1,{N}) i;
        """)
        self.check("re-INSERT after TRUNCATE", rc == 0, err[:200] if err else "")

    def test_rowcompress_dml(self) -> None:
        self.section("rowcompress DML")
        N = 10_000

        rc, err = self.exec(f"""
            DROP TABLE IF EXISTS _tr;
            CREATE TABLE _tr (id int, val bigint, label text) USING rowcompress;
            INSERT INTO _tr
                SELECT i, i * 2, 'row_'||i
                FROM generate_series(1,{N}) i;
        """)
        self.check("CREATE TABLE + INSERT rowcompress", rc == 0, err[:200] if err else "")

        cnt = self.q1("SELECT count(*) FROM _tr")
        self.check_eq(f"row count = {N}", cnt, str(N))

        cnt2 = self.q1(f"SELECT count(*) FROM _tr WHERE val > {N}")
        self.check_eq("SELECT with WHERE filter (val > N → N/2 rows)", cnt2, str(N // 2))

        rc, err = self.exec("UPDATE _tr SET label = 'changed' WHERE id = 42")
        self.check("UPDATE rowcompress", rc == 0, err[:200] if err else "")

        rc, err = self.exec("DELETE FROM _tr WHERE id > 9000")
        self.check("DELETE rowcompress", rc == 0, err[:200] if err else "")

    # ------------------------------------------------------------------ options

    def test_compression_options(self) -> None:
        self.section("Compression Options")

        # Options are set via engine.alter_colcompress_table_set(), not WITH clause.
        for codec in ("zstd", "lz4", "pglz"):
            rc, err = self.exec(f"""
                DROP TABLE IF EXISTS _tcodec;
                CREATE TABLE _tcodec (id int, payload text) USING colcompress;
                SELECT engine.alter_colcompress_table_set('_tcodec', compression => '{codec}');
                INSERT INTO _tcodec
                    SELECT i, repeat('storage_engine_', 10)
                    FROM generate_series(1, 2000) i;
            """)
            self.check(f"colcompress: alter_colcompress_table_set compression='{codec}'",
                       rc == 0, err[:200] if err else "")

            cnt = self.q1("SELECT count(*) FROM _tcodec")
            self.check_eq(f"  row count with {codec} = 2000", cnt, "2000")

            saved = self.q1(
                "SELECT compression FROM engine.col_options "
                "WHERE regclass = '_tcodec'::regclass"
            )
            self.check(f"  engine.col_options.compression = '{codec}'",
                       saved == codec, f"got {saved!r}")

    def test_sort_key(self) -> None:
        self.section("Sort Key (orderby) Option")

        # orderby is set via engine.alter_colcompress_table_set() after CREATE.
        rc, err = self.exec("""
            DROP TABLE IF EXISTS _tsort;
            CREATE TABLE _tsort (ts date, val integer) USING colcompress;
            SELECT engine.alter_colcompress_table_set('_tsort', orderby => 'ts');
            INSERT INTO _tsort
                SELECT '2020-01-01'::date + (i % 3650), i
                FROM generate_series(1, 100000) i;
        """)
        self.check("alter_colcompress_table_set orderby='ts'", rc == 0, err[:200] if err else "")

        saved = self.q1(
            "SELECT orderby FROM engine.col_options "
            "WHERE regclass = '_tsort'::regclass"
        )
        self.check("engine.col_options.orderby = 'ts'", saved == "ts", f"got {saved!r}")

        cnt = self.q1("SELECT count(*) FROM _tsort")
        self.check_eq("all rows readable after sorted INSERT", cnt, "100000")

    # ------------------------------------------------------------------ vectorized aggregates

    def test_vect_int_aggregates(self) -> None:
        self.section("Vectorized Aggregates — smallint / integer / bigint")

        self.exec("""
            DROP TABLE IF EXISTS _tagg_int;
            CREATE TABLE _tagg_int (i2 smallint, i4 integer, i8 bigint) USING colcompress;
            INSERT INTO _tagg_int
                SELECT (i % 32767)::smallint, i, i::bigint * 10
                FROM generate_series(1, 100000) i;
        """)

        for col, typ in (("i2", "smallint"), ("i4", "integer"), ("i8", "bigint")):
            self.agg_ok(f"count(*)   [{typ}]",
                        f"SELECT count(*) FROM _tagg_int")
            self.agg_ok(f"count(col) [{typ}]",
                        f"SELECT count({col}) FROM _tagg_int")
            self.agg_ok(f"min({typ})",
                        f"SELECT min({col}) FROM _tagg_int")
            self.agg_ok(f"max({typ})",
                        f"SELECT max({col}) FROM _tagg_int")
            self.agg_ok(f"sum({typ})",
                        f"SELECT sum({col}) FROM _tagg_int")
            self.agg_ok(f"avg({typ})",
                        f"SELECT round(avg({col})::numeric, 6) FROM _tagg_int")

    def test_vect_float8_aggregates(self) -> None:
        self.section("Vectorized Aggregates — float8 (double precision)")

        self.exec("""
            DROP TABLE IF EXISTS _tagg_f8;
            CREATE TABLE _tagg_f8 (val double precision) USING colcompress;
            INSERT INTO _tagg_f8
                SELECT 1.0 + (i % 9999) * 0.01
                FROM generate_series(1, 100000) i;
        """)

        self.agg_ok("count(*)   [float8]",
                    "SELECT count(*) FROM _tagg_f8")
        self.agg_ok("count(col) [float8]",
                    "SELECT count(val) FROM _tagg_f8")
        self.agg_ok("min(float8)",
                    "SELECT min(val)::text FROM _tagg_f8")
        self.agg_ok("max(float8)",
                    "SELECT max(val)::text FROM _tagg_f8")
        self.agg_ok("sum(float8)",
                    "SELECT round(sum(val)::numeric, 4) FROM _tagg_f8")
        self.agg_ok("avg(float8)",
                    "SELECT round(avg(val)::numeric, 6) FROM _tagg_f8")

    def test_vect_numeric_aggregates(self) -> None:
        self.section("Vectorized Aggregates — numeric")

        self.exec("""
            DROP TABLE IF EXISTS _tagg_num;
            CREATE TABLE _tagg_num (val numeric(14, 4)) USING colcompress;
            INSERT INTO _tagg_num
                SELECT (1000 + (i % 9000))::numeric(14,4) / 100
                FROM generate_series(1, 100000) i;
        """)

        self.agg_ok("count(*)   [numeric]",
                    "SELECT count(*) FROM _tagg_num")
        self.agg_ok("count(col) [numeric]",
                    "SELECT count(val) FROM _tagg_num")
        self.agg_ok("min(numeric)",
                    "SELECT min(val) FROM _tagg_num")
        self.agg_ok("max(numeric)",
                    "SELECT max(val) FROM _tagg_num")
        self.agg_ok("sum(numeric)",
                    "SELECT sum(val) FROM _tagg_num")
        self.agg_ok("avg(numeric)",
                    "SELECT round(avg(val), 6) FROM _tagg_num")

    def test_vect_money_aggregates(self) -> None:
        self.section("Vectorized Aggregates — money")

        self.exec("""
            DROP TABLE IF EXISTS _tagg_money;
            CREATE TABLE _tagg_money (val money) USING colcompress;
            INSERT INTO _tagg_money
                SELECT ((100 + i % 9900)::numeric / 100)::money
                FROM generate_series(1, 100000) i;
        """)

        self.agg_ok("count(*)   [money]",
                    "SELECT count(*) FROM _tagg_money")
        self.agg_ok("min(money)",
                    "SELECT min(val) FROM _tagg_money")
        self.agg_ok("max(money)",
                    "SELECT max(val) FROM _tagg_money")
        self.agg_ok("sum(money)",
                    "SELECT sum(val) FROM _tagg_money")

    def test_vect_date_aggregates(self) -> None:
        self.section("Vectorized Aggregates — date")

        self.exec("""
            DROP TABLE IF EXISTS _tagg_date;
            CREATE TABLE _tagg_date (val date) USING colcompress;
            INSERT INTO _tagg_date
                SELECT '2000-01-01'::date + (i % 10000)
                FROM generate_series(1, 100000) i;
        """)

        self.agg_ok("count(*)   [date]",
                    "SELECT count(*) FROM _tagg_date")
        self.agg_ok("count(col) [date]",
                    "SELECT count(val) FROM _tagg_date")
        self.agg_ok("min(date)",
                    "SELECT min(val) FROM _tagg_date")
        self.agg_ok("max(date)",
                    "SELECT max(val) FROM _tagg_date")

    # ------------------------------------------------------------------ multi-column

    def test_multi_column_aggregates(self) -> None:
        self.section("Multi-column / Multi-type Aggregate Query")

        self.exec("""
            DROP TABLE IF EXISTS _tmulti;
            CREATE TABLE _tmulti (
                i2  smallint,
                i4  integer,
                i8  bigint,
                f8  double precision,
                num numeric(12, 4),
                d   date,
                m   money
            ) USING colcompress;
            INSERT INTO _tmulti
            SELECT
                (i % 32767)::smallint,
                i,
                i::bigint * 2,
                i * 0.01,
                (i * 0.01)::numeric(12, 4),
                '2020-01-01'::date + (i % 3650),
                (i * 0.01)::numeric::money
            FROM generate_series(1, 100000) i;
        """)

        self.agg_ok(
            "multi-type query: count/min/max for all columns",
            "SELECT count(*), "
            "min(i2), max(i2), "
            "min(i4), max(i4), "
            "min(i8), max(i8), "
            "round(min(f8)::numeric, 4), round(max(f8)::numeric, 4), "
            "min(num), max(num), "
            "min(d), max(d), "
            "min(m), max(m) "
            "FROM _tmulti",
        )
        self.agg_ok(
            "multi-type query: sum/avg for numeric types",
            "SELECT sum(i4), round(avg(i4)::numeric, 6), "
            "sum(i8), "
            "round(sum(f8)::numeric, 2), round(avg(f8)::numeric, 6), "
            "sum(num), round(avg(num), 6), "
            "sum(m) "
            "FROM _tmulti",
        )

    # ------------------------------------------------------------------ NULL handling

    def test_null_handling(self) -> None:
        self.section("NULL Handling")

        # Note: direct q1() calls below do NOT prepend SET statements so the
        # last-line heuristic in q1() always returns the SELECT result.
        # All NULL-vs-expected correctness is verified via agg_ok() which
        # compares VEC OFF against VEC ON (both sides produce the same NULL).

        # --- empty table ---
        self.exec("""
            DROP TABLE IF EXISTS _tnull;
            CREATE TABLE _tnull (
                i2  smallint,
                i4  integer,
                f8  double precision,
                num numeric,
                d   date,
                m   money
            ) USING colcompress;
        """)

        cnt = self.q1("SELECT count(*) FROM _tnull")
        self.check_eq("empty table: count(*) = 0", cnt, "0")

        # Correctness: VEC OFF == VEC ON for NULL-returning aggregates
        for col, agg in (
            ("i2",  "min"), ("i2",  "max"), ("i2",  "sum"),
            ("i4",  "min"), ("i4",  "max"), ("i4",  "sum"),
            ("f8",  "min"), ("f8",  "max"),
            ("num", "min"), ("num", "max"),
            ("d",   "min"), ("d",   "max"),
            ("m",   "min"), ("m",   "max"),
        ):
            self.agg_ok(f"empty table: {agg}({col}) VEC==OFF",
                        f"SELECT {agg}({col}) FROM _tnull")

        # Verify the result is actually NULL (not some sentinel like INT_MIN/0)
        for col, agg in (("i2", "sum"), ("i4", "sum"), ("i4", "min"), ("f8", "max"), ("m", "sum")):
            val = self.q1(f"SELECT {agg}({col}) FROM _tnull")
            self.check(f"empty table: {agg}({col}) IS NULL (VEC default)",
                       val == "", f"got {val!r}")

        # --- all-NULL column ---
        self.exec("""
            INSERT INTO _tnull
                SELECT i, i, NULL, NULL, NULL, NULL
                FROM generate_series(1, 1000) i;
        """)

        star = self.q1("SELECT count(*) FROM _tnull")
        self.check_eq("all-NULL cols: count(*) = 1000", star, "1000")

        col_cnt = self.q1("SELECT count(f8) FROM _tnull")
        self.check_eq("all-NULL col: count(f8) = 0", col_cnt, "0")

        self.agg_ok("all-NULL col: min(f8) VEC==OFF", "SELECT min(f8) FROM _tnull")
        self.agg_ok("all-NULL col: max(num) VEC==OFF", "SELECT max(num) FROM _tnull")
        self.agg_ok("all-NULL col: sum(money) VEC==OFF", "SELECT sum(m) FROM _tnull")

        min_null = self.q1("SELECT min(f8) FROM _tnull")
        self.check("all-NULL col: min(f8) IS NULL", min_null == "", f"got {min_null!r}")

        max_null = self.q1("SELECT max(num) FROM _tnull")
        self.check("all-NULL col: max(num) IS NULL", max_null == "", f"got {max_null!r}")

        # --- mixed NULLs (every 3rd row NULL) ---
        self.exec("""
            DROP TABLE IF EXISTS _tmixed;
            CREATE TABLE _tmixed (val integer, fval double precision) USING colcompress;
            INSERT INTO _tmixed
                SELECT
                    CASE WHEN i % 3 = 0 THEN NULL ELSE i END,
                    CASE WHEN i % 3 = 0 THEN NULL ELSE i * 0.01 END
                FROM generate_series(1, 30000) i;
        """)

        self.agg_ok("mixed NULLs: count(*)",       "SELECT count(*)    FROM _tmixed")
        self.agg_ok("mixed NULLs: count(val)",     "SELECT count(val)  FROM _tmixed")
        self.agg_ok("mixed NULLs: min(integer)",   "SELECT min(val)    FROM _tmixed")
        self.agg_ok("mixed NULLs: max(integer)",   "SELECT max(val)    FROM _tmixed")
        self.agg_ok("mixed NULLs: sum(integer)",   "SELECT sum(val)    FROM _tmixed")
        self.agg_ok("mixed NULLs: avg(integer)",
                    "SELECT round(avg(val)::numeric, 6) FROM _tmixed")
        self.agg_ok("mixed NULLs: sum(float8)",
                    "SELECT round(sum(fval)::numeric, 4) FROM _tmixed")
        self.agg_ok("mixed NULLs: avg(float8)",
                    "SELECT round(avg(fval)::numeric, 6) FROM _tmixed")

        # single non-NULL row
        self.exec("""
            DROP TABLE IF EXISTS _tsingle;
            CREATE TABLE _tsingle (val integer) USING colcompress;
            INSERT INTO _tsingle VALUES (42);
        """)
        self.agg_ok("single row: min/max/sum/count",
                    "SELECT min(val), max(val), sum(val), count(*) FROM _tsingle")

    # ------------------------------------------------------------------ EXPLAIN

    def test_explain_plan(self) -> None:
        self.section("EXPLAIN Plan Verification")

        self.exec("""
            DROP TABLE IF EXISTS _texpl;
            CREATE TABLE _texpl (val integer) USING colcompress;
            INSERT INTO _texpl SELECT i FROM generate_series(1, 100000) i;
        """)

        # VEC ON → StorageEngineVectorAgg present
        plan_on = self.q(
            "SET storage_engine.enable_vectorization=on; "
            "SET max_parallel_workers_per_gather=0; "
            "EXPLAIN SELECT min(val), max(val), sum(val), count(*) FROM _texpl"
        )
        self.check("VEC ON: EXPLAIN shows StorageEngineVectorAgg",
                   "StorageEngineVectorAgg" in plan_on, plan_on[:400])
        self.check("VEC ON: EXPLAIN shows 'Engine Vectorized Aggregate: enabled'",
                   "Engine Vectorized Aggregate: enabled" in plan_on, plan_on[:400])

        # VEC OFF → no StorageEngineVectorAgg
        plan_off = self.q(
            "SET storage_engine.enable_vectorization=off; "
            "SET max_parallel_workers_per_gather=0; "
            "EXPLAIN SELECT min(val) FROM _texpl"
        )
        self.check("VEC OFF: EXPLAIN has no StorageEngineVectorAgg",
                   "StorageEngineVectorAgg" not in plan_off, plan_off[:400])

        # EXPLAIN ANALYZE should complete without error and show VectorAgg node
        out, rc, err = self._run(
            "SET storage_engine.enable_vectorization=on; "
            "SET max_parallel_workers_per_gather=0; "
            "EXPLAIN ANALYZE SELECT min(val), max(val), count(*) FROM _texpl"
        )
        self.check("EXPLAIN ANALYZE: no error (VEC ON)", rc == 0, err[:200] if err else "")
        self.check("EXPLAIN ANALYZE: shows StorageEngineVectorAgg",
                   "StorageEngineVectorAgg" in out, out[:400])

        # Plain EXPLAIN (no ANALYZE) on a filter query
        plan_filter = self.q(
            "SET storage_engine.enable_vectorization=on; "
            "EXPLAIN SELECT val FROM _texpl WHERE val > 50000"
        )
        self.check("EXPLAIN SELECT (filter): no crash, returns plan",
                   len(plan_filter) > 0, plan_filter[:200])

        self.exec("""
            DROP TABLE IF EXISTS _texpl_mixed;
            CREATE TABLE _texpl_mixed (
                i4  integer,
                num numeric(12, 4),
                m   money
            ) USING colcompress;
            INSERT INTO _texpl_mixed
                SELECT
                    i,
                    (i * 0.01)::numeric(12, 4),
                    (i * 0.01)::numeric::money
                FROM generate_series(1, 100000) i;
        """)

        plan_mixed = self.q(
            "SET storage_engine.enable_vectorization=on; "
            "SET max_parallel_workers_per_gather=0; "
            "EXPLAIN SELECT sum(i4), round(avg(num), 6), sum(m) FROM _texpl_mixed"
        )
        self.check("mixed numeric+money: EXPLAIN falls back to regular Agg",
                   "StorageEngineVectorAgg" not in plan_mixed, plan_mixed[:400])

        self.agg_ok("mixed numeric+money: VEC ON falls back without changing result",
                    "SELECT sum(i4), round(avg(num), 6), sum(m) FROM _texpl_mixed")

        out, rc, err = self._run(
            "SET storage_engine.enable_vectorization=on; "
            "SET max_parallel_workers_per_gather=0; "
            "SELECT sum(i4), round(avg(num), 6), sum(m) FROM _texpl_mixed"
        )
        self.check("mixed numeric+money: VEC ON query runs without crash",
                   rc == 0, err[:200] if err else out[:200])

    # ------------------------------------------------------------------ parallel safety

    def test_parallel_safety(self) -> None:
        self.section("Parallel Safety (AGGSPLIT_SIMPLE guard)")

        self.exec("""
            DROP TABLE IF EXISTS _tpar;
            CREATE TABLE _tpar (val integer) USING colcompress;
            INSERT INTO _tpar SELECT i FROM generate_series(1, 100000) i;
        """)

        pfx_serial   = ("SET max_parallel_workers_per_gather=0; "
                        "SET storage_engine.enable_vectorization=on;")
        pfx_parallel = ("SET max_parallel_workers_per_gather=4; "
                        "SET storage_engine.enable_vectorization=on;")

        agg_sql = "SELECT min(val), max(val), sum(val), count(*) FROM _tpar"

        r_serial   = self.q(f"{pfx_serial}   {agg_sql}")
        r_parallel = self.q(f"{pfx_parallel} {agg_sql}")
        self.check("parallel=4: result equals serial",
                   r_serial == r_parallel,
                   f"serial={r_serial!r} parallel={r_parallel!r}")

        _, rc, err = self._run(f"{pfx_parallel} {agg_sql}")
        self.check("parallel=4: no crash / error", rc == 0, err[:200] if err else "")

        # Also test with float8, numeric, money under parallelism
        self.exec("""
            DROP TABLE IF EXISTS _tpar_typed;
            CREATE TABLE _tpar_typed (
                f8  double precision,
                num numeric(12,4),
                m   money
            ) USING colcompress;
            INSERT INTO _tpar_typed
                SELECT i * 0.01, (i * 0.01)::numeric(12,4), (i * 0.01)::numeric::money
                FROM generate_series(1, 100000) i;
        """)

        typed_sql = ("SELECT round(sum(f8)::numeric,2), "
                     "sum(num), sum(m) FROM _tpar_typed")

        r_s = self.q(f"{pfx_serial}   {typed_sql}")
        r_p = self.q(f"{pfx_parallel} {typed_sql}")
        self.check("parallel=4 float8+numeric+money: result equals serial",
                   r_s == r_p, f"serial={r_s!r} parallel={r_p!r}")

        # --------------------------------------------------------------
        # Plan-shape gates for real vectorized parallel execution.
        # --------------------------------------------------------------
        self.exec("""
            DROP TABLE IF EXISTS _tpar_group;
            CREATE TABLE _tpar_group (
                grp integer,
                v   integer,
                vn  numeric(12,2)
            ) USING colcompress;
            INSERT INTO _tpar_group
                SELECT i % 16, i, (i * 0.01)::numeric(12,2)
                FROM generate_series(1, 500000) i;
        """)

        pfx_parallel_vec = (
            "SET storage_engine.enable_vectorization=on; "
            "SET storage_engine.enable_vectorized_groupagg=on; "
            "SET max_parallel_workers_per_gather=4; "
            "SET parallel_setup_cost=0; "
            "SET parallel_tuple_cost=0; "
            "SET min_parallel_table_scan_size=0; "
            "SET min_parallel_index_scan_size=0;"
        )
        pfx_parallel_native = (
            "SET storage_engine.enable_vectorization=off; "
            "SET storage_engine.enable_vectorized_groupagg=off; "
            "SET max_parallel_workers_per_gather=4; "
            "SET parallel_setup_cost=0; "
            "SET parallel_tuple_cost=0; "
            "SET min_parallel_table_scan_size=0; "
            "SET min_parallel_index_scan_size=0;"
        )

        vec_group_sql = (
            "SELECT grp, count(*), sum(v), min(v), max(v) "
            "FROM _tpar_group GROUP BY grp ORDER BY grp"
        )

        plan_vec = self.q(f"{pfx_parallel_vec} EXPLAIN {vec_group_sql}")
        self.check(
            "parallel grouped count/sum/min/max: EXPLAIN shows StorageEngineVectorGroupAgg",
            "StorageEngineVectorGroupAgg" in plan_vec,
            plan_vec[:500],
        )
        self.check(
            "parallel grouped count/sum/min/max: EXPLAIN is parallel",
            "Parallel" in plan_vec,
            plan_vec[:500],
        )

        avg_group_sql = "SELECT grp, avg(v::float8) FROM _tpar_group GROUP BY grp"
        plan_avg = self.q(f"{pfx_parallel_vec} EXPLAIN {avg_group_sql}")
        server_version_num = self.q1("SHOW server_version_num")
        avg_groupagg_expected = (
            server_version_num and
            (int(server_version_num) < 200000)
        )
        if avg_groupagg_expected:
            self.check(
                "parallel grouped avg(v::float8): EXPLAIN shows StorageEngineVectorGroupAgg",
                "StorageEngineVectorGroupAgg" in plan_avg,
                plan_avg[:500],
            )
        else:
            self.check(
                "parallel grouped avg(v::float8): EXPLAIN falls back (version-limited)",
                "StorageEngineVectorGroupAgg" not in plan_avg,
                plan_avg[:500],
            )

        avg_group_int4_sql = "SELECT grp, avg(v) FROM _tpar_group GROUP BY grp"
        plan_avg_int4 = self.q(f"{pfx_parallel_vec} EXPLAIN {avg_group_int4_sql}")
        if avg_groupagg_expected:
            self.check(
                "parallel grouped avg(int4): EXPLAIN shows StorageEngineVectorGroupAgg",
                "StorageEngineVectorGroupAgg" in plan_avg_int4,
                plan_avg_int4[:500],
            )
        else:
            self.check(
                "parallel grouped avg(int4): EXPLAIN falls back (version-limited)",
                "StorageEngineVectorGroupAgg" not in plan_avg_int4,
                plan_avg_int4[:500],
            )

        plan_avg_int4_serial = self.q(
            "SET storage_engine.enable_vectorization=on; "
            "SET storage_engine.enable_vectorized_groupagg=on; "
            "SET max_parallel_workers_per_gather=0; "
            "EXPLAIN SELECT grp, avg(v) FROM _tpar_group GROUP BY grp"
        )
        if avg_groupagg_expected:
            self.check(
                "serial grouped avg(int4): EXPLAIN shows StorageEngineVectorGroupAgg",
                "StorageEngineVectorGroupAgg" in plan_avg_int4_serial,
                plan_avg_int4_serial[:500],
            )
        else:
            self.check(
                "serial grouped avg(int4): EXPLAIN falls back (version-limited)",
                "StorageEngineVectorGroupAgg" not in plan_avg_int4_serial,
                plan_avg_int4_serial[:500],
            )

        # PG15-only safety gate: numeric plain aggregate must fallback.
        server_version_num = self.q1("SHOW server_version_num")
        if server_version_num and int(server_version_num) < 160000:
            self.exec("""
                DROP TABLE IF EXISTS _tpar_num_plain;
                CREATE TABLE _tpar_num_plain (val numeric(14,4)) USING colcompress;
                INSERT INTO _tpar_num_plain
                    SELECT (1000 + (i % 9000))::numeric(14,4) / 100
                    FROM generate_series(1, 100000) i;
            """)
            plan_num_pg15 = self.q(
                "SET storage_engine.enable_vectorization=on; "
                "SET max_parallel_workers_per_gather=0; "
                "EXPLAIN SELECT count(val) FROM _tpar_num_plain"
            )
            self.check(
                "PG15 numeric plain aggregate: EXPLAIN falls back (no StorageEngineVectorAgg)",
                "StorageEngineVectorAgg" not in plan_num_pg15,
                plan_num_pg15[:500],
            )

        # Performance sanity gate (wide threshold to catch catastrophic regressions only).
        out_vec, rc_vec, err_vec = self._run(
            f"{pfx_parallel_vec} "
            f"EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY ON, FORMAT JSON) {vec_group_sql}"
        )
        out_nat, rc_nat, err_nat = self._run(
            f"{pfx_parallel_native} "
            f"EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY ON, FORMAT JSON) {vec_group_sql}"
        )

        self.check("parallel perf sanity: vectorized EXPLAIN ANALYZE executes", rc_vec == 0,
                   err_vec[:200] if err_vec else "")
        self.check("parallel perf sanity: native EXPLAIN ANALYZE executes", rc_nat == 0,
                   err_nat[:200] if err_nat else "")

        t_vec = self._execution_time_ms(out_vec) if rc_vec == 0 else None
        t_nat = self._execution_time_ms(out_nat) if rc_nat == 0 else None
        self.check("parallel perf sanity: execution times parsed",
                   t_vec is not None and t_nat is not None,
                   f"vec={t_vec!r} native={t_nat!r}")
        if t_vec is not None and t_nat is not None:
            self.check(
                "parallel perf sanity: vectorized <= 5x native",
                t_vec <= (t_nat * 5.0),
                f"vectorized={t_vec:.3f}ms native={t_nat:.3f}ms",
            )

    # ------------------------------------------------------------------ VecGroupAgg

    def test_vecgroupagg(self) -> None:
        self.section("VecGroupAgg — GROUP BY correctness")

        pfx_on  = ("SET storage_engine.enable_vectorization=on; "
                   "SET storage_engine.enable_vectorized_groupagg=on; "
                   "SET max_parallel_workers_per_gather=4; "
                   "SET parallel_setup_cost=0; SET parallel_tuple_cost=0; "
                   "SET min_parallel_table_scan_size=0; "
                   "SET min_parallel_index_scan_size=0;")
        pfx_off = ("SET storage_engine.enable_vectorization=off; "
                   "SET storage_engine.enable_vectorized_groupagg=off; "
                   "SET max_parallel_workers_per_gather=0;")

        # --- setup ---
        self.exec("""
            DROP TABLE IF EXISTS _tgrp;
            CREATE TABLE _tgrp (
                country char(2),
                cat     integer,
                val     float8,
                m       money
            ) USING colcompress;
            INSERT INTO _tgrp
            SELECT
                (ARRAY['BR','US','JP'])[1 + (i % 3)],
                1 + (i % 5),
                CASE WHEN i % 7  = 0 THEN NULL ELSE i * 1.5       END,
                CASE WHEN i % 11 = 0 THEN NULL ELSE (i * 0.99)::money END
            FROM generate_series(1, 1500000) i;
            ANALYZE _tgrp;

            DROP TABLE IF EXISTS _tgrp2;
            CREATE TABLE _tgrp2 (k1 integer, k2 integer, v bigint) USING colcompress;
            INSERT INTO _tgrp2
                SELECT i % 50, i % 20, i FROM generate_series(1, 2000000) i;
            ANALYZE _tgrp2;
        """)

        # 1. Single-key GROUP BY — correctness VEC ON == VEC OFF
        single_sql = ("SELECT country, count(*), count(val), "
                      "min(val), max(val), round(sum(val)::numeric, 2), "
                      "min(m), max(m), sum(m) "
                      "FROM _tgrp GROUP BY country ORDER BY country")
        r_off = self.q(f"{pfx_off} {single_sql}")
        r_on  = self.q(f"{pfx_on}  {single_sql}")
        self.check("single-key GROUP BY: VEC ON == VEC OFF", r_off == r_on,
                   f"OFF={r_off[:200]!r}  ON={r_on[:200]!r}")

        # 2. Single-key: EXPLAIN shows VecGroupAgg
        # Force HashAgg: disable sort so planner can't pick GroupAggregate
        plan = self.q(f"{pfx_on} SET enable_sort=off; EXPLAIN {single_sql}")
        self.check("single-key GROUP BY: EXPLAIN shows StorageEngineVectorGroupAgg",
                   "StorageEngineVectorGroupAgg" in plan, plan[:400])

        # 3. COUNT(col) with NULLs — val has NULLs every 7th row
        null_sql = ("SELECT country, count(*) AS cs, count(val) AS cv "
                    "FROM _tgrp GROUP BY country ORDER BY country")
        r_off2 = self.q(f"{pfx_off} {null_sql}")
        r_on2  = self.q(f"{pfx_on}  {null_sql}")
        self.check("COUNT(col) NULL-aware GROUP BY: VEC ON == VEC OFF",
                   r_off2 == r_on2, f"OFF={r_off2!r}  ON={r_on2!r}")

        # COUNT(*) > COUNT(col) because val has NULLs
        r_diff = self.q(
            f"{pfx_on} SELECT bool_and(cs > cv) FROM ({null_sql}) x"
        )
        self.check("COUNT(col) NULL-aware: count(*) > count(val) for all groups",
                   r_diff.strip() == "t", f"got {r_diff!r}")

        # 4. Composite GROUP BY (2 keys) — correctness
        composite_sql = ("SELECT country, cat, count(*), count(val), "
                         "round(sum(val)::numeric, 2), sum(m) "
                         "FROM _tgrp GROUP BY country, cat ORDER BY country, cat")
        r_off3 = self.q(f"{pfx_off} {composite_sql}")
        r_on3  = self.q(f"{pfx_on}  {composite_sql}")
        self.check("composite GROUP BY (2 keys): VEC ON == VEC OFF",
                   r_off3 == r_on3, f"OFF={r_off3[:200]!r}  ON={r_on3[:200]!r}")

        # 5. Composite GROUP BY: EXPLAIN shows VecGroupAgg
        # Disable parallel to get a plain HashAggregate (not Finalize/Partial split)
        # which the hook can intercept on all PG versions
        plan2 = self.q(f"{pfx_on} SET max_parallel_workers_per_gather=0; EXPLAIN "
                       "SELECT k1, k2, count(*), sum(v) FROM _tgrp2 GROUP BY k1, k2")
        self.check("composite GROUP BY (2 int keys): EXPLAIN shows StorageEngineVectorGroupAgg",
                   "StorageEngineVectorGroupAgg" in plan2, plan2[:400])

        # 6. HAVING — must still produce correct results (applied by Finalize node)
        having_sql = ("SELECT country, count(*) FROM _tgrp "
                      "GROUP BY country HAVING count(*) > 50000 ORDER BY country")
        r_off4 = self.q(f"{pfx_off} {having_sql}")
        r_on4  = self.q(f"{pfx_on}  {having_sql}")
        self.check("HAVING: VEC ON == VEC OFF (correct results)",
                   r_off4 == r_on4, f"OFF={r_off4!r}  ON={r_on4!r}")

        # HAVING must not be empty (all 3 groups have 100k rows > 50k)
        rows = [ln for ln in r_on4.strip().split("\n") if ln]
        self.check("HAVING: returns 3 groups (count > 50000)", len(rows) == 3,
                   f"rows={rows!r}")

        # 7. money (CASHOID) GROUP BY — correctness
        money_sql = ("SELECT country, count(m), min(m), max(m), sum(m) "
                     "FROM _tgrp GROUP BY country ORDER BY country")
        r_off5 = self.q(f"{pfx_off} {money_sql}")
        r_on5  = self.q(f"{pfx_on}  {money_sql}")
        self.check("money (CASHOID) GROUP BY: VEC ON == VEC OFF",
                   r_off5 == r_on5, f"OFF={r_off5[:200]!r}  ON={r_on5[:200]!r}")

        # 8. Composite GROUP BY (3 keys)
        three_key_sql = ("SELECT country, cat, val IS NULL AS vn, count(*) "
                         "FROM _tgrp GROUP BY country, cat, val IS NULL "
                         "ORDER BY country, cat, vn")
        r_off6 = self.q(f"{pfx_off} {three_key_sql}")
        r_on6  = self.q(f"{pfx_on}  {three_key_sql}")
        self.check("composite GROUP BY (3 keys): VEC ON == VEC OFF",
                   r_off6 == r_on6, f"OFF={r_off6[:200]!r}  ON={r_on6[:200]!r}")

    # ------------------------------------------------------------------ upgrade path

    def test_upgrade_path(self) -> None:
        self.section("Upgrade Path Chain")

        # Latest version available
        ver = self.q1(
            "SELECT max(version) FROM pg_available_extension_versions "
            "WHERE name = 'storage_engine'"
        )
        self.check("latest available version = 1.3.4", ver == "1.3.4", f"got {ver!r}")

        # Complete upgrade path from 1.0 to 1.3.4 exists
        path = self.q1(
            "SELECT path FROM pg_extension_update_paths('storage_engine') "
            "WHERE source = '1.0' AND target = '1.3.4'"
        )
        self.check("upgrade path 1.0 \u2192 1.3.4 exists", path != "", f"path={path!r}")

        # Each individual upgrade step
        steps = [
            ("1.0", "1.1"),
            ("1.1", "1.2.0"),
            ("1.2.0", "1.2.1"),
            ("1.2.1", "1.2.2"),
            ("1.2.2", "1.2.3"),
            ("1.2.3", "1.2.4"),
            ("1.2.4", "1.2.5"),
            ("1.2.5", "1.2.6"),
            ("1.2.6", "1.2.7"),
            ("1.2.7", "1.2.8"),
            ("1.2.8", "1.2.9"),
            ("1.2.9", "1.3.0"),
            ("1.3.0", "1.3.1"),
            ("1.3.1", "1.3.2"),
            ("1.3.2", "1.3.3"),
            ("1.3.3", "1.3.4"),
        ]
        for src, tgt in steps:
            p = self.q1(
                f"SELECT path FROM pg_extension_update_paths('storage_engine') "
                f"WHERE source = '{src}' AND target = '{tgt}'"
            )
            self.check(f"upgrade step {src} → {tgt}", p != "", f"path={p!r}")

    # ================================================================== run all

    def run_all(self) -> bool:
        print(f"\n{'=' * 62}")
        print(f"  storage_engine test suite — {self.label}")
        print(f"{'=' * 62}")

        print(f"\n[setup] creating test database '{self.dbname}'…")
        self.setup()

        self.test_extension_lifecycle()
        self.test_colcompress_dml()
        self.test_rowcompress_dml()
        self.test_compression_options()
        self.test_sort_key()
        self.test_vect_int_aggregates()
        self.test_vect_float8_aggregates()
        self.test_vect_numeric_aggregates()
        self.test_vect_money_aggregates()
        self.test_vect_date_aggregates()
        self.test_multi_column_aggregates()
        self.test_null_handling()
        self.test_explain_plan()
        self.test_parallel_safety()
        self.test_vecgroupagg()
        self.test_upgrade_path()

        self.teardown()

        total = self.passed + self.failed
        print(f"\n{'=' * 62}")
        if self.failed == 0:
            print(f"  {self.label}: {GREEN}ALL {total} TESTS PASSED{RESET}")
        else:
            print(f"  {self.label}: {RED}{self.failed} FAILED{RESET} / {total} total")
            print(f"\n  Failures:")
            for f in self.failures:
                print(f"    • {f}")
        print(f"{'=' * 62}\n")
        return self.failed == 0


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(description="storage_engine test suite")
    parser.add_argument("--port",     type=int, default=5432,
                        help="Primary PostgreSQL port (default: 5432)")
    parser.add_argument("--pg19",     action="store_true",
                        help="Also run tests on the secondary PostgreSQL instance")
    parser.add_argument("--pg19-port", type=int, default=5433, dest="pg19_port",
                        help="Secondary PostgreSQL port (default: 5433)")
    parser.add_argument("--ports",    type=str, default=None,
                        help="Comma-separated list of PostgreSQL ports to test "
                             "(overrides --port / --pg19, e.g. 5436,5434,5435,5432,5433)")
    args = parser.parse_args()

    all_ok = True

    if args.ports:
        ports = [int(p.strip()) for p in args.ports.split(",") if p.strip()]
        for port in ports:
            r = TestRunner(port, label=f"PG@{port}")
            all_ok = r.run_all() and all_ok
    else:
        r1 = TestRunner(args.port, label=f"PG@{args.port}")
        all_ok = r1.run_all() and all_ok

        if args.pg19:
            r2 = TestRunner(args.pg19_port, label=f"PG@{args.pg19_port}")
            all_ok = r2.run_all() and all_ok

    sys.exit(0 if all_ok else 1)


if __name__ == "__main__":
    main()
