如何禁用SQLAlchemy查询缓存？PostgreSQL性能测试遇缓存干扰

问题背景 #

用Python的SQLAlchemy做PostgreSQL性能测试，逻辑是循环迭代测试不同数据量（[100000,150000,200000,250000]行）的查询耗时。第一次迭代正常，但后续迭代中SQLAlchemy会缓存查询，导致测试结果不准。试过session.expires_all()、调整execution_options、创建新Session、修改隔离级别等方法，均无效。

相关代码与日志 #

查询函数示例 #

@get_exec_time
def select_from_data(session_obj, field_name, field_value, mode="json"):
    tgt_table = None

    if mode not in ("json", "jsonb"):
        raise TypeError("Значение mode должно быть равно json/jsonb")

    if mode == "json":
        tgt_table = JsonMedicalData
        filter_expr = tgt_table.json_data.op("->>")(field_name).cast(Integer) == field_value

    elif mode == "jsonb":
        tgt_table = JsonbMedicalData
        filter_expr = tgt_table.jsonb_data.op("@>")({field_name: field_value})

    with session_obj.begin() as session:
        session.expire_all()
        result = session.query(tgt_table).filter(filter_expr).all()

测试流水线代码 #

row_numbers = [100000, 150000, 200000, 250000]

results_dict = {}

# 生成测试数据
json_rows_max, jsonb_rows_max = data_generator(row_numbers[-1], mode="json"), data_generator(row_numbers[-1], mode="jsonb")

for rows in row_numbers:
    json_rows, jsonb_rows = json_rows_max[:rows], jsonb_rows_max[:rows]

    # JSON数据测试流程
    json_insert_time = insert_test(json_rows, Session)
    results_dict_insert(results_dict, key="insert", row_number=rows, mode="json", exec_time=json_insert_time)

    select_where_time = select_from_data(Session, field_name="select_test", field_value=5, mode="json")
    results_dict_insert(results_dict, key="select", row_number=rows, mode="json", exec_time=select_where_time)

    summarize_data_time = summarize_data_field(Session, field_name="зарегистрировано всего", mode="json")
    results_dict_insert(results_dict, key="summarize", row_number=rows, mode="json", exec_time=summarize_data_time)

    json_delete_time = delete_test(JsonMedicalData, Session)
    results_dict_insert(results_dict, key="delete", row_number=rows, mode="json", exec_time=json_delete_time)

    # JSONB数据测试流程
    jsonb_insert_time = insert_test(jsonb_rows, Session)
    results_dict_insert(results_dict, key="insert", row_number=rows, mode="jsonb", exec_time=jsonb_insert_time)

    select_where_time = select_from_data(Session, field_name="select_test", field_value=5, mode="jsonb")
    results_dict_insert(results_dict, key="select", row_number=rows, mode="jsonb", exec_time=select_where_time)

    summarize_data_time = summarize_data_field(Session, field_name="зарегистрировано всего", mode="jsonb")
    results_dict_insert(results_dict, key="summarize", row_number=rows, mode="jsonb", exec_time=summarize_data_time)

    # GIN索引测试流程
    gin_index_create_time = gin_index_create(Session, "gin_index")
    results_dict_insert(results_dict, key="gin_create", row_number=rows, mode="jsonb_gin", exec_time=gin_index_create_time)

    select_where_time = select_from_data(Session, field_name="select_test", field_value=5, mode="jsonb")
    results_dict_insert(results_dict, key="select", row_number=rows, mode="jsonb_gin", exec_time=select_where_time)

    summarize_data_time = summarize_data_field(Session, field_name="зарегистрировано всего", mode="jsonb")
    results_dict_insert(results_dict, key="summarize", row_number=rows, mode="jsonb_gin", exec_time=summarize_data_time)

    gin_index_delete_time = gin_index_delete(Session, index_name="gin_index")
    results_dict_insert(results_dict, key="gin_delete", row_number=rows, mode="jsonb_gin", exec_time=gin_index_delete_time)

    jsonb_delete_time = delete_test(JsonbMedicalData, Session)
    results_dict_insert(results_dict, key="delete", row_number=rows, mode="jsonb", exec_time=jsonb_delete_time)

缓存相关日志片段 #

2023-06-21 12:23:18,175 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-06-21 12:23:18,175 INFO sqlalchemy.engine.Engine SELECT sum(CAST(jsonb_medical_data.jsonb_data ->> %(jsonb_data_1)s AS INTEGER)) AS sum_1 
FROM jsonb_medical_data
2023-06-21 12:23:18,175 INFO sqlalchemy.engine.Engine [cached since 39.2s ago] {'jsonb_data_1': 'зарегистрировано всего'}
2023-06-21 12:23:18,200 INFO sqlalchemy.engine.Engine COMMIT
2023-06-21 12:23:18,201 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-06-21 12:23:18,201 INFO sqlalchemy.engine.Engine DROP INDEX gin_index
2023-06-21 12:23:18,201 INFO sqlalchemy.engine.Engine [cached since 37.55s ago] {}
2023-06-21 12:23:18,201 INFO sqlalchemy.engine.Engine COMMIT
2023-06-21 12:23:18,205 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-06-21 12:23:18,205 INFO sqlalchemy.engine.Engine DELETE FROM jsonb_medical_data
2023-06-21 12:23:18,205 INFO sqlalchemy.engine.Engine [cached since 37.54s ago] {}
2023-06-21 12:23:18,232 INFO sqlalchemy.engine.Engine COMMIT

核心问题分析 #

日志中的[cached since ...]表明，问题根源是SQLAlchemy的语句编译缓存（Statement Cache），而非Session级别的ORM对象缓存。之前尝试的expire_all()等方法只针对ORM对象缓存，无法解决语句缓存的问题。

解决办法 #

1. 关闭SQL语句编译缓存 #

可以全局或单查询禁用语句缓存，强制每次都重新编译并发送SQL到数据库：

• 全局关闭（创建Engine时）：

engine = create_engine(
    "postgresql://user:password@host/database",
    execution_options={"compiled_cache": None}
)

• 单查询禁用：
  在查询语句中添加execution_options(compiled_cache=None)：

result = session.query(tgt_table).filter(filter_expr).execution_options(compiled_cache=None).all()

2. 迭代使用全新Session #

每次测试迭代都创建独立的Session，避免Session级别的对象缓存干扰：

for rows in row_numbers:
    # 每次迭代创建新Session
    with Session() as session:
        json_rows, jsonb_rows = json_rows_max[:rows], jsonb_rows_max[:rows]
        
        # 后续所有测试操作都使用这个新Session
        json_insert_time = insert_test(json_rows, session)
        # ... 其他测试步骤

3. 禁用PostgreSQL查询计划缓存 #

PostgreSQL自身也会缓存查询计划，可通过设置参数强制生成新计划：

# 在每次查询前执行
session.execute("SET plan_cache_mode = 'force_custom_plan';")

4. 彻底清理Session状态 #

如果仍存在ORM对象缓存问题，可在每次测试后执行session.rollback()或session.close()，但最彻底的方式还是使用全新Session。

内容的提问来源于stack exchange，提问作者Nikita Smirnov