db_benchmark_suite/
├── README.md
├── requirements.txt
├── config.yaml
├── docker-compose.yml
├── src/
│   ├── __init__.py
│   ├── main.py
│   ├── database/
│   │   ├── __init__.py
│   │   ├── base_connector.py
│   │   ├── postgres_connector.py
│   │   ├── mongodb_connector.py
│   │   ├── redis_connector.py
│   │   ├── oracle_connector.py
│   │   └── sqlite_connector.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── data_models.py
│   ├── benchmarks/
│   │   ├── __init__.py
│   │   ├── base_benchmark.py
│   │   ├── crud_benchmark.py
│   │   └── complex_query_benchmark.py
│   └── utils/
│       ├── __init__.py
│       ├── results_aggregator.py
│       └── visualization.py
├── scripts/
│   ├── init_databases.py
│   └── run_benchmarks.py
├── tests/
│   └── test_connectors.py
├── results/
│   └── .gitkeep
└── .env.example

# 核心异步框架与数据库驱动
asyncio
asyncpg>=0.29.0  # PostgreSQL 高性能异步驱动
motor>=3.3.0     # MongoDB 异步驱动
redis>=5.0.0     # Redis 同步驱动 (aioredis 已合并)
cx_Oracle>=8.3.0 # Oracle 驱动
aiosqlite>=0.19.0 # SQLite 异步驱动

# 配置与数据
pyyaml>=6.0
python-dotenv>=1.0.0
pandas>=2.0.0   # 结果分析

# 可视化
matplotlib>=3.7.0
seaborn>=0.12.0

# 测试
pytest>=7.3.0
pytest-asyncio>=0.21.0

version: '3.8'
services:
  postgres17:
    image: postgres:17-beta  # 使用 beta 版本，正式发布后可切到 :17
    container_name: pg17_benchmark
    environment:
      POSTGRES_USER: bench_user
      POSTGRES_PASSWORD: bench_pass
      POSTGRES_DB: benchmark_db
    ports:

      - "5432:5432"
    volumes:

      - postgres17_data:/var/lib/postgresql/data
      - ./config/postgres.conf:/etc/postgresql/postgresql.conf:ro
    command: >
      postgres -c config_file=/etc/postgresql/postgresql.conf
      -c shared_preload_libraries='pg_stat_statements, auto_explain'
      -c auto_explain.log_min_duration=0
      -c auto_explain.log_analyze=on
    healthcheck:
      test: [ "CMD-SHELL", "pg_isready -U bench_user -d benchmark_db" ]
      interval: 5s
      timeout: 5s
      retries: 5

  mongodb:
    image: mongo:7
    container_name: mongo_benchmark
    environment:
      MONGO_INITDB_ROOT_USERNAME: bench_user
      MONGO_INITDB_ROOT_PASSWORD: bench_pass
    ports:

      - "27017:27017"
    volumes:

      - mongodb_data:/data/db

  redis:
    image: redis:7-alpine
    container_name: redis_benchmark
    ports:

      - "6379:6379"
    volumes:

      - redis_data:/data
    command: redis-server --appendonly yes

  oracle-xe:
    image: gvenzl/oracle-xe:21-slim
    container_name: oracle_benchmark
    environment:
      ORACLE_PASSWORD: bench_pass
      APP_USER: bench_user
      APP_USER_PASSWORD: bench_pass
    ports:

      - "1521:1521"
    volumes:

      - oracle_data:/opt/oracle/oradata

volumes:
  postgres17_data:
  mongodb_data:
  redis_data:
  oracle_data:

databases:
  postgres:
    host: "localhost"
    port: 5432
    user: "bench_user"
    password: "bench_pass"
    database: "benchmark_db"
    min_size: 5
    max_size: 20
    # PostgreSQL 17 特定配置
    server_version: "17"

  mongodb:
    host: "localhost"
    port: 27017
    user: "bench_user"
    password: "bench_pass"
    auth_source: "admin"
    database: "benchmark_db"

  redis:
    host: "localhost"
    port: 6379
    # password: "" # 本例未设密码
    decode_responses: True
    max_connections: 20

  oracle:
    dsn: "localhost:1521/FREEPDB1"
    user: "bench_user"
    password: "bench_pass"
    min: 2
    max: 10
    increment: 1
    encoding: "UTF-8"

  sqlite:
    database_path: "./results/benchmark.db"
    # 内存模式用于极致性能测试
    # database_path: ":memory:"

benchmark:
  data_scale:
    initial_records: 100000
    batch_size: 1000
  iterations: 5
  warmup_runs: 2
  complex_query:
    join_depth: 3
    filter_selectivity: 0.1  # 10%

# 复制此文件为 .env 并填写敏感信息（如生产环境密码）
# POSTGRES_PASSWORD=your_strong_password_here
# ORACLE_PASSWORD=your_strong_password_here

import abc
import asyncio
import time
from typing import Any, Dict, List, Optional
import pandas as pd


class BaseDatabaseConnector(abc.ABC):
    """所有数据库连接器的抽象基类，定义统一接口。"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.connection_pool = None
        self._query_stats: List[Dict] = []

    @abc.abstractmethod
    async def connect(self):
        """建立连接池。"""
        pass

    @abc.abstractmethod
    async def disconnect(self):
        """关闭连接池。"""
        pass

    @abc.abstractmethod
    async def execute_query(self, query: str, params: Optional[tuple] = None) -> Any:
        """执行查询并返回结果。"""
        pass

    @abc.abstractmethod
    async def insert_bulk(self, data: List[Dict], table_or_collection: str) -> int:
        """批量插入数据。"""
        pass

    @abc.abstractmethod
    async def create_test_schema(self):
        """创建测试所需的表/集合与索引。"""
        pass

    @abc.abstractmethod
    async def drop_test_schema(self):
        """清理测试模式。"""
        pass

    def _record_query_stat(self, query_type: str, duration_ms: float, rows_affected: int = 0):
        """记录查询统计信息，用于后续分析。"""
        self._query_stats.append({
            "timestamp": time.time(),
            "query_type": query_type,
            "duration_ms": duration_ms,
            "rows_affected": rows_affected,
            "connector": self.__class__.__name__
        })

    def get_query_stats(self) -> pd.DataFrame:
        """返回记录的查询统计数据为DataFrame。"""
        return pd.DataFrame(self._query_stats)

    def reset_stats(self):
        """重置统计信息。"""
        self._query_stats.clear()

import asyncio
import asyncpg
from typing import Any, Dict, List, Optional
from .base_connector import BaseDatabaseConnector
import logging

logger = logging.getLogger(__name__)


class PostgresConnector(BaseDatabaseConnector):
    """PostgreSQL 17 连接器，利用 asyncpg 实现高性能异步操作。"""

    async def connect(self):
        """创建 PostgreSQL 连接池。"""
        try:
            self.connection_pool = await asyncpg.create_pool(
                host=self.config['host'],
                port=self.config['port'],
                user=self.config['user'],
                password=self.config['password'],
                database=self.config['database'],
                min_size=self.config.get('min_size', 5),
                max_size=self.config.get('max_size', 20),
                server_settings={ # PostgreSQL 17 相关优化设置
                    'application_name': 'pg17_benchmark',
                    'jit': 'off',  # 对于短查询，JIT可能引入开销
                }
            )
            logger.info("PostgreSQL connection pool created.")
        except Exception as e:
            logger.error(f"Failed to create PostgreSQL connection pool: {e}")
            raise

    async def disconnect(self):
        if self.connection_pool:
            await self.connection_pool.close()
            logger.info("PostgreSQL connection pool closed.")

    async def execute_query(self, query: str, params: Optional[tuple] = None) -> List[asyncpg.Record]:
        start = asyncio.get_event_loop().time()
        async with self.connection_pool.acquire() as conn:
            # 使用 prepared statement 提升重复查询性能
            if params:
                result = await conn.fetch(query, *params)
            else:
                result = await conn.fetch(query)
        duration = (asyncio.get_event_loop().time() - start) * 1000  # ms
        self._record_query_stat("SELECT", duration, len(result))
        return result

    async def insert_bulk(self, data: List[Dict], table_name: str) -> int:
        """使用 COPY 命令进行高性能批量插入（PostgreSQL 优化）。"""
        if not data:
            return 0
        columns = data[0].keys()
        column_names = ', '.join(columns)
        # 构建 VALUES 占位符 ($1, $2, ...)
        value_placeholders = ', '.join([f'${i+1}' for i in range(len(columns))])
        query = f"""
            INSERT INTO {table_name} ({column_names})
            VALUES ({value_placeholders})
        """
        start = asyncio.get_event_loop().time()
        async with self.connection_pool.acquire() as conn:
            # 使用 executemany 进行批量插入，asyncpg 内部会优化
            await conn.executemany(query, [tuple(record.values()) for record in data])
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_query_stat("BULK_INSERT", duration, len(data))
        return len(data)

    async def create_test_schema(self):
        """创建测试表、索引，并启用扩展以利用 PostgreSQL 17 特性。"""
        async with self.connection_pool.acquire() as conn:
            # 启用关键扩展
            await conn.execute("""
                CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
                CREATE EXTENSION IF NOT EXISTS btree_gin; -- 用于多列索引
            """)
            # 创建主表
            await conn.execute("""
                DROP TABLE IF EXISTS orders CASCADE;
                CREATE TABLE orders (
                    order_id BIGSERIAL PRIMARY KEY,
                    customer_id INT NOT NULL,
                    order_date DATE NOT NULL DEFAULT CURRENT_DATE,
                    total_amount DECIMAL(12,2) NOT NULL,
                    status VARCHAR(20) NOT NULL,
                    -- PostgreSQL 17 新特性：支持在分区键上创建哈希分区？
                    -- 注意：哈希分区在更早版本已引入，但17可能有增强。
                    region VARCHAR(10) NOT NULL,
                    metadata JSONB
                ) PARTITION BY LIST (region);
                -- 创建两个分区以演示分区修剪
                CREATE TABLE orders_east PARTITION OF orders FOR VALUES IN ('EAST');
                CREATE TABLE orders_west PARTITION OF orders FOR VALUES IN ('WEST');
            """)
            # 创建二级表用于连接查询
            await conn.execute("""
                DROP TABLE IF EXISTS order_items;
                CREATE TABLE order_items (
                    item_id BIGSERIAL PRIMARY KEY,
                    order_id BIGINT NOT NULL REFERENCES orders(order_id) ON DELETE CASCADE,
                    product_id INT NOT NULL,
                    quantity INT NOT NULL,
                    unit_price DECIMAL(10,2) NOT NULL
                );
            """)
            # 创建索引策略：B-tree, BRIN, GIN, 复合索引
            await conn.execute("""
                -- B-tree 用于等值查询和范围扫描
                CREATE INDEX idx_orders_customer_id ON orders(customer_id);
                -- BRIN 索引对按时间顺序插入的大表非常高效
                CREATE INDEX idx_orders_order_date_brin ON orders USING BRIN(order_date);
                -- GIN 索引用于 JSONB 字段的快速查询
                CREATE INDEX idx_orders_metadata_gin ON orders USING GIN(metadata);
                -- 复合索引覆盖常见查询
                CREATE INDEX idx_orders_status_date ON orders(status, order_date);
                -- 对 order_items 的连接字段创建索引
                CREATE INDEX idx_order_items_order_id ON order_items(order_id);
                -- PostgreSQL 17 增强：并行创建索引（CREATE INDEX CONCURRENTLY 的改进）
            """)
            logger.info("PostgreSQL test schema created with partitions and advanced indexes.")

    async def drop_test_schema(self):
        async with self.connection_pool.acquire() as conn:
            await conn.execute("DROP TABLE IF EXISTS order_items CASCADE;")
            await conn.execute("DROP TABLE IF EXISTS orders CASCADE;")
            logger.info("PostgreSQL test schema dropped.")

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
from typing import Any, Dict, List, Optional
from .base_connector import BaseDatabaseConnector
import logging

logger = logging.getLogger(__name__)


class MongodbConnector(BaseDatabaseConnector):
    """MongoDB 连接器，使用 Motor 异步驱动。"""

    async def connect(self):
        """创建 MongoDB 客户端。"""
        uri = f"mongodb://{self.config['user']}:{self.config['password']}@{self.config['host']}:{self.config['port']}/?authSource={self.config['auth_source']}"
        self.client = AsyncIOMotorClient(uri, maxPoolSize=self.config.get('max_pool_size', 100))
        self.db = self.client[self.config['database']]
        logger.info("MongoDB client connected.")

    async def disconnect(self):
        self.client.close()
        logger.info("MongoDB client closed.")

    async def execute_query(self, query: Dict, collection_name: str) -> List[Dict]:
        """执行 MongoDB 查询。query 参数是一个 MongoDB 查询字典。"""
        start = asyncio.get_event_loop().time()
        collection: AsyncIOMotorCollection = self.db[collection_name]
        cursor = collection.find(query)
        result = await cursor.to_list(length=None)  # 获取所有文档
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_query_stat("FIND", duration, len(result))
        return result

    async def insert_bulk(self, data: List[Dict], collection_name: str) -> int:
        """使用 insert_many 进行批量插入。"""
        if not data:
            return 0
        start = asyncio.get_event_loop().time()
        collection: AsyncIOMotorCollection = self.db[collection_name]
        result = await collection.insert_many(data, ordered=False)  # 无序插入更快
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_query_stat("BULK_INSERT", duration, len(result.inserted_ids))
        return len(result.inserted_ids)

    async def create_test_schema(self):
        """创建集合和索引。"""
        # 创建 orders 集合
        orders_collection: AsyncIOMotorCollection = self.db['orders']
        # 如果集合存在，清空
        await orders_collection.drop()
        await self.db.create_collection('orders')
        # 创建索引
        await orders_collection.create_index([('customer_id', 1)])
        await orders_collection.create_index([('order_date', 1)])
        await orders_collection.create_index([('status', 1), ('order_date', -1)])
        await orders_collection.create_index([('region', 1)])  # 模拟分区
        # 对于 JSONB 的模拟，MongoDB 文档原生支持，但可对子字段创建索引
        # 例如，如果 metadata 是一个子文档，可以创建索引: await orders_collection.create_index([('metadata.some_field', 1)])

        # 创建 order_items 集合
        items_collection: AsyncIOMotorCollection = self.db['order_items']
        await items_collection.drop()
        await self.db.create_collection('order_items')
        await items_collection.create_index([('order_id', 1)])
        logger.info("MongoDB test collections and indexes created.")

    async def drop_test_schema(self):
        await self.db['orders'].drop()
        await self.db['order_items'].drop()
        logger.info("MongoDB test collections dropped.")

import asyncio
import redis.asyncio as aioredis
from typing import Any, Dict, List, Optional
import json
import pickle
from .base_connector import BaseDatabaseConnector
import logging

logger = logging.getLogger(__name__)


class RedisConnector(BaseDatabaseConnector):
    """Redis 连接器，使用 redis-py 的异步接口。"""
    # Redis 是键值存储，数据模型与其他数据库差异较大。
    # 我们将模拟类似的结构：使用 Hash 存储订单，Set 存储订单项 ID，List 用于排序。

    async def connect(self):
        """创建 Redis 连接池。"""
        self.connection_pool = aioredis.ConnectionPool.from_url(
            f"redis://{self.config['host']}:{self.config['port']}",
            max_connections=self.config.get('max_connections', 20),
            decode_responses=self.config.get('decode_responses', True)
        )
        self.client = aioredis.Redis(connection_pool=self.connection_pool)
        logger.info("Redis connection pool created.")

    async def disconnect(self):
        await self.client.close()
        await self.connection_pool.disconnect()
        logger.info("Redis connection closed.")

    async def execute_query(self, query: str, params: Optional[tuple] = None) -> Any:
        """Redis 查询较为特殊。这里 query 可以是命令，如 'HGETALL'。"""
        # 这是一个简化示例。实际基准测试中，我们会定义特定的操作模式。
        start = asyncio.get_event_loop().time()
        if query == 'HGETALL':
            key = params[0] if params else ''
            result = await self.client.hgetall(key)
        elif query == 'ZRANGEBYSCORE':
            key, min_score, max_score = params
            result = await self.client.zrangebyscore(key, min_score, max_score)
        else:
            result = await self.client.execute_command(query, *params) if params else await self.client.execute_command(query)
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_query_stat("COMMAND", duration, len(result) if isinstance(result, (list, dict)) else 1)
        return result

    async def insert_bulk(self, data: List[Dict], key_prefix: str) -> int:
        """批量插入数据到 Redis。使用 pipeline 提升性能。"""
        if not data:
            return 0
        start = asyncio.get_event_loop().time()
        async with self.client.pipeline(transaction=False) as pipe:
            for i, record in enumerate(data):
                key = f"{key_prefix}:{i}"
                # 将记录存储为 Hash
                await pipe.hset(key, mapping=record)
                # 例如，按日期添加到 Sorted Set 以支持范围查询
                if 'order_date' in record:
                    # 简化：将日期转换为分数
                    score = float(record['order_date'].replace('-', '')) if isinstance(record['order_date'], str) else 0
                    await pipe.zadd(f"idx:{key_prefix}:by_date", {key: score})
            results = await pipe.execute()
        duration = (asyncio.get_event_loop().time() - start) * 1000
        inserted = sum(1 for r in results if r)
        self._record_query_stat("BULK_PIPELINE", duration, inserted)
        return inserted

    async def create_test_schema(self):
        """Redis 没有模式，但可以初始化索引结构。"""
        # 清空当前数据库（DB 0）中的测试键
        await self.client.flushdb()
        logger.info("Redis database flushed for test.")

    async def drop_test_schema(self):
        # 与 create 相同，清空数据库
        await self.client.flushdb()
        logger.info("Redis test data dropped.")

import random
from datetime import date, timedelta
from typing import List, Dict
import json


def generate_order_record(order_id_start: int, customer_id_range: tuple, date_range: tuple, regions: List[str]) -> Dict:
    """生成一条模拟订单记录。"""
    order_date = date_range[0] + timedelta(days=random.randint(0, (date_range[1] - date_range[0]).days))
    return {
        'order_id': order_id_start,  # 将在插入时由序列生成
        'customer_id': random.randint(*customer_id_range),
        'order_date': order_date.isoformat(),
        'total_amount': round(random.uniform(10.0, 10000.0), 2),
        'status': random.choice(['PENDING', 'PROCESSING', 'SHIPPED', 'DELIVERED', 'CANCELLED']),
        'region': random.choice(regions),
        'metadata': json.dumps({
            'source': random.choice(['web', 'mobile', 'in-store']),
            'priority': random.choice(['low', 'medium', 'high']),
            'tags': random.sample(['electronics', 'clothing', 'books', 'home'], k=random.randint(1,3))
        })
    }


def generate_order_item_record(order_id: int, item_id: int) -> Dict:
    """生成一条模拟订单项记录。"""
    return {
        'item_id': item_id,
        'order_id': order_id,
        'product_id': random.randint(1000, 9999),
        'quantity': random.randint(1, 10),
        'unit_price': round(random.uniform(1.0, 1000.0), 2)
    }


def generate_bulk_data(initial_records: int, batch_size: int) -> List[List[Dict]]:
    """生成批量数据，用于初始化数据库。"""
    all_batches = []
    customer_range = (1, 50000)
    date_range = (date(2023, 1, 1), date(2024, 12, 31))
    regions = ['EAST', 'WEST']

    # 生成订单批次
    order_batch_count = (initial_records + batch_size - 1) // batch_size
    order_item_ratio = 5  # 平均每个订单有5个订单项
    order_item_id_counter = 1

    for batch_idx in range(order_batch_count):
        order_batch = []
        items_batch = []
        records_in_this_batch = min(batch_size, initial_records - batch_idx * batch_size)
        for i in range(records_in_this_batch):
            order_id_placeholder = batch_idx * batch_size + i + 1
            order = generate_order_record(order_id_placeholder, customer_range, date_range, regions)
            order_batch.append(order)
            # 为每个订单生成订单项
            num_items = random.randint(1, order_item_ratio * 2)  # 1-10个订单项
            for _ in range(num_items):
                item = generate_order_item_record(order_id_placeholder, order_item_id_counter)
                items_batch.append(item)
                order_item_id_counter += 1
        all_batches.append(('orders', order_batch))
        all_batches.append(('order_items', items_batch))
    return all_batches

import abc
import asyncio
import time
from typing import List, Dict, Any
import pandas as pd
import numpy as np
from ..database.base_connector import BaseDatabaseConnector


class BaseBenchmark(abc.ABC):
    """基准测试的基类。"""

    def __init__(self, connector: BaseDatabaseConnector, config: Dict[str, Any]):
        self.connector = connector
        self.config = config
        self.results: List[Dict] = []

    @abc.abstractmethod
    async def run(self):
        """执行具体的基准测试。"""
        pass

    def _record_result(self, operation: str, duration_ms: float, iteration: int, **kwargs):
        """记录单次操作结果。"""
        self.results.append({
            "operation": operation,
            "duration_ms": duration_ms,
            "iteration": iteration,
            "timestamp": time.time(),
            "connector": self.connector.__class__.__name__,
            **kwargs
        })

    def get_results_df(self) -> pd.DataFrame:
        """返回结果为 DataFrame。"""
        return pd.DataFrame(self.results)

    def get_summary_stats(self) -> Dict[str, Any]:
        """计算汇总统计：平均耗时、P95、P99、吞吐量等。"""
        df = self.get_results_df()
        summary = {}
        if df.empty:
            return summary
        for op in df['operation'].unique():
            op_df = df[df['operation'] == op]
            durs = op_df['duration_ms']
            summary[f'{op}_mean_ms'] = durs.mean()
            summary[f'{op}_p95_ms'] = np.percentile(durs, 95)
            summary[f'{op}_p99_ms'] = np.percentile(durs, 99)
            summary[f'{op}_throughput_ops_per_sec'] = 1000 / durs.mean() if durs.mean() > 0 else 0
        return summary

import asyncio
import random
from typing import List, Dict, Any
from .base_benchmark import BaseBenchmark
from ..models.data_models import generate_order_record, generate_order_item_record


class CRUDBenchmark(BaseBenchmark):
    """测试 CRUD（创建、读取、更新、删除）操作的性能。"""

    async def run(self):
        """执行 CRUD 基准测试序列。"""
        iterations = self.config.get('iterations', 5)
        warmup = self.config.get('warmup_runs', 2)
        data_scale = self.config.get('data_scale', {})
        total_records = data_scale.get('initial_records', 100000)
        # 选择一小部分数据进行 CRUD 测试
        sample_size = min(1000, total_records // 100)

        # 预热运行（不记录结果）
        for _ in range(warmup):
            await self._run_single_iteration(sample_size, iteration=-1)

        # 正式测试运行
        for i in range(iterations):
            await self._run_single_iteration(sample_size, iteration=i)
            await asyncio.sleep(0.5)  # 迭代间短暂间隔

    async def _run_single_iteration(self, sample_size: int, iteration: int):
        """单次迭代：包含 SELECT, UPDATE, DELETE, INSERT 操作。"""
        # 1. 随机 SELECT 查询
        start = asyncio.get_event_loop().time()
        # 示例：查询特定客户最近的订单
        customer_id = random.randint(1, 50000)
        if isinstance(self.connector.__class__.__name__, str) and 'Postgres' in self.connector.__class__.__name__:
            result = await self.connector.execute_query(
                """SELECT * FROM orders WHERE customer_id = $1 ORDER BY order_date DESC LIMIT 10""",
                (customer_id,)
            )
        # ... 其他数据库的查询语法适配（此处略，实际项目需完整实现）
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_result("SELECT_BY_CUSTOMER", duration, iteration, rows_returned=len(result) if 'result' in locals() else 0)

        # 2. 范围 SELECT 查询（利用索引）
        start = asyncio.get_event_loop().time()
        date_threshold = '2024-06-01'
        if 'Postgres' in self.connector.__class__.__name__:
            result = await self.connector.execute_query(
                """SELECT COUNT(*) FROM orders WHERE order_date > $1 AND status = $2""",
                (date_threshold, 'SHIPPED')
            )
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_result("SELECT_RANGE", duration, iteration)

        # 3. UPDATE 操作
        start = asyncio.get_event_loop().time()
        update_id = random.randint(1, sample_size)
        if 'Postgres' in self.connector.__class__.__name__:
            await self.connector.execute_query(
                """UPDATE orders SET status = $1 WHERE order_id = $2""",
                ('PROCESSING', update_id)
            )
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_result("UPDATE_SINGLE", duration, iteration)

        # 4. DELETE 操作
        start = asyncio.get_event_loop().time()
        delete_id = random.randint(1, sample_size)
        if 'Postgres' in self.connector.__class__.__name__:
            await self.connector.execute_query(
                """DELETE FROM orders WHERE order_id = $1""",
                (delete_id,)
            )
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_result("DELETE_SINGLE", duration, iteration)

        # 5. INSERT 单条操作（补回删除的记录）
        start = asyncio.get_event_loop().time()
        new_order = generate_order_record(delete_id, (1, 50000), (date(2023,1,1), date(2024,12,31)), ['EAST', 'WEST'])
        # 注意：实际 INSERT 语句需要适配不同数据库（此处为示例）
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_result("INSERT_SINGLE", duration, iteration)

import asyncio
from .base_benchmark import BaseBenchmark


class ComplexQueryBenchmark(BaseBenchmark):
    """测试复杂查询（多表连接、聚合、子查询、JSON 查询）的性能。"""

    async def run(self):
        iterations = self.config.get('iterations', 5)
        warmup = self.config.get('warmup_runs', 2)
        complex_config = self.config.get('complex_query', {})

        # 预热
        for _ in range(warmup):
            await self._run_complex_queries(iteration=-1, config=complex_config)

        for i in range(iterations):
            await self._run_complex_queries(iteration=i, config=complex_config)
            await asyncio.sleep(1)

    async def _run_complex_queries(self, iteration: int, config: Dict):
        """执行一组复杂查询。"""
        # 1. 多表 JOIN 聚合查询（分析型查询）
        start = asyncio.get_event_loop().time()
        if 'Postgres' in self.connector.__class__.__name__:
            result = await self.connector.execute_query("""
                SELECT
                    o.region,
                    o.status,
                    DATE_TRUNC('month', o.order_date) as month,
                    COUNT(DISTINCT o.order_id) as order_count,
                    SUM(oi.quantity * oi.unit_price) as revenue,
                    AVG(o.total_amount) as avg_order_value
                FROM orders o
                JOIN order_items oi ON o.order_id = oi.order_id
                WHERE o.order_date >= DATE '2024-01-01'
                GROUP BY o.region, o.status, DATE_TRUNC('month', o.order_date)
                ORDER BY month, region, revenue DESC
                LIMIT 100;
            """)
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_result("JOIN_AGGREGATE", duration, iteration, rows_returned=len(result) if 'result' in locals() else 0)

        # 2. 窗口函数查询（排名、移动平均）
        start = asyncio.get_event_loop().time()
        if 'Postgres' in self.connector.__class__.__name__:
            result = await self.connector.execute_query("""
                SELECT
                    customer_id,
                    order_date,
                    total_amount,
                    AVG(total_amount) OVER (
                        PARTITION BY customer_id
                        ORDER BY order_date
                        ROWS BETWEEN 3 PRECEDING AND CURRENT ROW
                    ) as moving_avg_4,
                    RANK() OVER (
                        PARTITION BY region, DATE_TRUNC('month', order_date)
                        ORDER BY total_amount DESC
                    ) as regional_monthly_rank
                FROM orders
                WHERE region = 'EAST' AND order_date >= '2024-01-01'
                ORDER BY customer_id, order_date
                LIMIT 200;
            """)
        duration = (asyncio.get_event_loop().time() - start) * 1000
        self._record_result("WINDOW_FUNCTION", duration, iteration)

        # 3. JSONB 字段查询（PostgreSQL 特有）
        start = asyncio.get_event_loop().time()
        if 'Postgres' in self.connector.__class__.__name__:
            result = await self.connector.execute_query("""
                SELECT order_id, customer_id, metadata->>'source' as source
                FROM orders
                WHERE metadata @> '{"tags": ["electronics"]}'::jsonb
                  AND metadata->>'priority' = 'high'
                ORDER BY order_date DESC
                LIMIT 50;
            """)
            duration = (asyncio.get_event_loop().time() - start) * 1000
            self._record_result("JSONB_QUERY", duration, iteration)

        # 4. 公共表表达式 (CTE) 与递归查询（可选）
        # 5. 利用分区修剪的查询（针对 PostgreSQL 分区表）
        start = asyncio.get_event_loop().time()
        if 'Postgres' in self.connector.__class__.__name__:
            result = await self.connector.execute_query(
                "SELECT COUNT(*) FROM orders WHERE region = $1",
                ('WEST',)
            )
            duration = (asyncio.get_event_loop().time() - start) * 1000
            self._record_result("PARTITION_PRUNING", duration, iteration)

import asyncio
import yaml
import logging
from pathlib import Path
from datetime import datetime
import pandas as pd

from database import (
    PostgresConnector,
    MongodbConnector,
    RedisConnector,
    OracleConnector,
    SqliteConnector
)
from benchmarks import CRUDBenchmark, ComplexQueryBenchmark
from utils.results_aggregator import aggregate_all_results
from utils.visualization import create_performance_dashboard

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


async def main():
    """主协调函数：初始化数据库，运行基准测试，聚合结果，生成报告。"""
    # 加载配置
    config_path = Path(__file__).parent.parent / 'config.yaml'
    with open(config_path, 'r') as f:
        config = yaml.safe_load(f)

    connectors = []
    benchmarks_to_run = []
    results_dir = Path('results')
    results_dir.mkdir(exist_ok=True)
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')

    # 1. 初始化所有数据库连接器并创建模式
    db_configs = config['databases']
    connector_classes = [
        (PostgresConnector, db_configs['postgres']),
        (MongodbConnector, db_configs['mongodb']),
        (RedisConnector, db_configs['redis']),
        (OracleConnector, db_configs['oracle']),
        (SqliteConnector, db_configs['sqlite'])
    ]

    for ConnectorClass, db_config in connector_classes:
        try:
            connector = ConnectorClass(db_config)
            await connector.connect()
            connectors.append(connector)
            logger.info(f"Connected to {connector.__class__.__name__}")
        except Exception as e:
            logger.error(f"Failed to connect to {ConnectorClass.__name__}: {e}")
            # 继续其他数据库
            continue

    if not connectors:
        logger.error("No database connectors initialized. Exiting.")
        return

    # 2. 为每个连接器创建测试模式并插入初始数据
    from models.data_models import generate_bulk_data
    data_batches = generate_bulk_data(
        config['benchmark']['data_scale']['initial_records'],
        config['benchmark']['data_scale']['batch_size']
    )
    for connector in connectors:
        try:
            await connector.create_test_schema()
            # 批量插入数据（示例：仅插入订单数据）
            for table_name, batch in data_batches:
                if table_name == 'orders':  # 简化处理
                    await connector.insert_bulk(batch, 'orders' if not isinstance(connector, RedisConnector) else 'order')
            logger.info(f"Initial data loaded for {connector.__class__.__name__}")
        except Exception as e:
            logger.error(f"Failed to init schema/data for {connector.__class__.__name__}: {e}")

    # 3. 为每个连接器运行基准测试套件
    benchmark_config = config['benchmark']
    for connector in connectors:
        logger.info(f"Running benchmarks for {connector.__class__.__name__}...")
        # CRUD 基准测试
        crud_bench = CRUDBenchmark(connector, benchmark_config)
        await crud_bench.run()
        # 复杂查询基准测试
        complex_bench = ComplexQueryBenchmark(connector, benchmark_config)
        await complex_bench.run()
        benchmarks_to_run.append((connector, crud_bench, complex_bench))

    # 4. 聚合结果并保存
    all_results_df, summary_df = aggregate_all_results(connectors, benchmarks_to_run)
    results_file = results_dir / f'benchmark_results_{timestamp}.parquet'
    summary_file = results_dir / f'benchmark_summary_{timestamp}.csv'
    all_results_df.to_parquet(results_file)
    summary_df.to_csv(summary_file, index=False)
    logger.info(f"Results saved to {results_file} and {summary_file}")

    # 5. 生成可视化报告
    dashboard_path = results_dir / f'performance_dashboard_{timestamp}.html'
    create_performance_dashboard(all_results_df, summary_df, dashboard_path)
    logger.info(f"Dashboard generated at {dashboard_path}")

    # 6. 清理
    for connector in connectors:
        try:
            await connector.drop_test_schema()
            await connector.disconnect()
        except Exception as e:
            logger.warning(f"Error during cleanup for {connector.__class__.__name__}: {e}")

    logger.info("Benchmark suite completed.")


if __name__ == '__main__':
    asyncio.run(main())

import pytest
import pytest_asyncio
import asyncio
from src.database.postgres_connector import PostgresConnector
from src.database.mongodb_connector import MongodbConnector
import yaml
import os

@pytest.fixture(scope="module")
def config():
    with open('config.yaml', 'r') as f:
        return yaml.safe_load(f)

@pytest_asyncio.fixture
async def postgres_connector(config):
    connector = PostgresConnector(config['databases']['postgres'])
    await connector.connect()
    yield connector
    await connector.drop_test_schema()
    await connector.disconnect()

@pytest.mark.asyncio
async def test_postgres_crud(postgres_connector):
    """测试 PostgreSQL 连接器的基本 CRUD 功能。"""
    await postgres_connector.create_test_schema()
    # 测试插入
    test_data = [{'customer_id': 99999, 'order_date': '2024-01-01', 'total_amount': 100.0, 'status': 'TEST', 'region': 'EAST', 'metadata': '{}'}]
    inserted = await postgres_connector.insert_bulk(test_data, 'orders_east')  # 插入到分区
    assert inserted == 1
    # 测试查询
    result = await postgres_connector.execute_query("SELECT * FROM orders_east WHERE customer_id = 99999")
    assert len(result) == 1
    assert result[0]['status'] == 'TEST'
    # 测试更新
    await postgres_connector.execute_query("UPDATE orders_east SET status = $1 WHERE customer_id = $2", ('UPDATED', 99999))
    result = await postgres_connector.execute_query("SELECT status FROM orders_east WHERE customer_id = 99999")
    assert result[0]['status'] == 'UPDATED'
    # 清理
    await postgres_connector.execute_query("DELETE FROM orders_east WHERE customer_id = 99999")

git clone <your-repo-url> db_benchmark_suite
cd db_benchmark_suite
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或 venv\Scripts\activate  # Windows
pip install --upgrade pip
pip install -r requirements.txt
cp .env.example .env  # 如需，修改 .env 中的密码

docker-compose up -d
# 等待所有容器健康检查通过（尤其是 Oracle，启动较慢）
docker-compose logs -f  # 查看日志

python scripts/init_databases.py
# 此脚本将依次连接每个数据库，创建表/集合，并插入初始的 100,000 条订单记录（及其订单项）。

python scripts/run_benchmarks.py
# 或直接运行主模块
python -m src.main

pytest tests/ -v

# postgresql.conf 关键优化 (基于 32GB RAM, 16 核的专用数据库服务器)
    shared_buffers = 8GB                 # 通常为内存的 25%
    effective_cache_size = 24GB          # 优化器假设的可用文件系统缓存
    maintenance_work_mem = 1GB           # 用于 VACUUM, CREATE INDEX
    max_parallel_workers_per_gather = 4  # 根据实际 CPU 核心数调整
    max_parallel_workers = 16
    wal_buffers = 16MB
    default_statistics_target = 500      # 更详细的统计信息，优化器更精准
    random_page_cost = 1.1               # 对于 SSD 存储
    effective_io_concurrency = 200
    # 负载管理（PostgreSQL 17 增强）
    pg_stat_statements.max = 10000
    pg_stat_statements.track = all
    # 日志记录用于分析
    shared_preload_libraries = 'pg_stat_statements, auto_explain'
    auto_explain.log_min_duration = 1000  # 记录超过 1秒的查询计划
    auto_explain.log_analyze = on