tpcc-mysql测试工具

import mysql.connector
import random

def run_tpcc_test(host, user, password, database):
    conn = mysql.connector.connect(host=host, user=user, password=password, database=database)
    cursor = conn.cursor()

    # 创建测试表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS warehouse (
            w_id SMALLINT PRIMARY KEY,
            w_ytd DECIMAL(12,2),
            w_tax DECIMAL(4,4),
            w_name VARCHAR(10),
            w_street_1 VARCHAR(20),
            w_street_2 VARCHAR(20),
            w_city VARCHAR(20),
            w_state CHAR(2),
            w_zip CHAR(9)
        );
    """)

    # 插入测试数据
    for w_id in range(1, 11):
        cursor.execute("""
            INSERT INTO warehouse (w_id, w_ytd, w_tax, w_name, w_street_1, w_street_2, w_city, w_state, w_zip)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s);
        """, (w_id, random.randint(0, 100000), round(random.uniform(0, 0.05), 4), f'Warehouse {w_id}', f'Street {w_id}', f'Street {w_id+10}', f'City {w_id}', f'ST{w_id}', f'12345{w_id}'))

    conn.commit()
    cursor.close()
    conn.close()

if __name__ == "__main__":
    run_tpcc_test('localhost', 'user', 'password', 'tpcc_test')