#!/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 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] + 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 out

    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

    # ------------------------------------------------------------------ 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.1", ver == "1.3.1", 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])

    # ------------------------------------------------------------------ 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}")

    # ------------------------------------------------------------------ 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.1", ver == "1.3.1", f"got {ver!r}")

        # Complete upgrade path from 1.0 to 1.3.1 exists
        path = self.q1(
            "SELECT path FROM pg_extension_update_paths('storage_engine') "
            "WHERE source = '1.0' AND target = '1.3.1'"
        )
        self.check("upgrade path 1.0 \u2192 1.3.1 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"),
        ]
        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_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)")
    args = parser.parse_args()

    all_ok = True

    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()
