国产数据库进入“份额、AI、开源”三线竞争期

python -m venv .venv
source .venv/bin/activate
pip install psycopg[binary]

export DB_DSN="postgresql://user:password@127.0.0.1:5432/testdb"
python db_smoke_test.py

import os
import statistics
import time

import psycopg

DSN = os.environ["DB_DSN"]

DDL = """
DROP TABLE IF EXISTS db_eval_orders;
CREATE TABLE db_eval_orders (
    id BIGINT PRIMARY KEY,
    user_id BIGINT NOT NULL,
    amount NUMERIC(12, 2) NOT NULL,
    status VARCHAR(20) NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_db_eval_orders_user_id ON db_eval_orders(user_id);
"""

INSERT_SQL = """
INSERT INTO db_eval_orders(id, user_id, amount, status)
VALUES (%s, %s, %s, %s)
"""

QUERY_SQL = """
SELECT user_id, COUNT(*) AS order_count, SUM(amount) AS total_amount
FROM db_eval_orders
WHERE user_id = %s
GROUP BY user_id
"""


def timed(fn):
    start = time.perf_counter()
    result = fn()
    return time.perf_counter() - start, result


def main():
    with psycopg.connect(DSN) as conn:
        with conn.cursor() as cur:
            cur.execute(DDL)
        conn.commit()

        def insert_rows():
            with conn.cursor() as cur:
                for i in range(1, 5001):
                    cur.execute(
                        INSERT_SQL,
                        (i, i % 100, round((i % 97) * 1.23, 2), "PAID"),
                    )
            conn.commit()

        insert_seconds, _ = timed(insert_rows)

        latencies_ms = []
        with conn.cursor() as cur:
            for user_id in range(100):
                seconds, _ = timed(lambda uid=user_id: cur.execute(QUERY_SQL, (uid,)).fetchone())
                latencies_ms.append(seconds * 1000)

        try:
            with conn.transaction():
                with conn.cursor() as cur:
                    cur.execute(INSERT_SQL, (999999, 1, 10.00, "ROLLBACK_TEST"))
                    raise RuntimeError("force rollback")
        except RuntimeError:
            pass

        with conn.cursor() as cur:
            cur.execute("SELECT COUNT(*) FROM db_eval_orders WHERE id = 999999")
            rollback_count = cur.fetchone()[0]

    print(f"insert_5000_rows_seconds={insert_seconds:.3f}")
    print(f"query_latency_p50_ms={statistics.median(latencies_ms):.3f}")
    print(f"query_latency_p95_ms={statistics.quantiles(latencies_ms, n=20)[18]:.3f}")
    print(f"rollback_verified={rollback_count == 0}")


if __name__ == "__main__":
    main()