如何在同一表达式中复用或缓存SQLAlchemy ScalarSelect？

使用SQL Server数据库，驱动为pyodbc。在SQLAlchemy的hybrid expression中，需要根据pm_frequencys表中coefficient和unit_of_time列的值，通过case语句执行不同的func.dateadd或func.datediff计算。

pm_frequencys表部分记录 #

当前实现代码 #

采用case语句实现动态日期计算，但嵌套逻辑导致子查询数量指数级增长，示例代码如下：

frequency_coefficient = (
    select([PmFrequency.coefficient])
    .where(cls.pm_frequency_id == PmFrequency.id)
    .correlate(cls)
    .limit(1)
    .as_scalar()
)
frequency_unit = (
    select([PmFrequency.unit_of_time])
    .where(cls.pm_frequency_id == PmFrequency.id)
    .correlate(cls)
    .limit(1)
    .as_scalar()
)
reference_diff = (
    func.cast(
        func.ceiling(
            func.cast(
                case(
                    [
                        (
                            frequency_unit == "day",
                            func.datediff(
                                text("day"),
                                cls.reference_date,
                                get_date_expression(),
                            ),
                        ),
                        (
                            frequency_unit == "week",
                            func.datediff(
                                text("week"),
                                cls.reference_date,
                                get_date_expression(),
                            ),
                        ),
                        (
                            frequency_unit == "month",
                            func.datediff(
                                text("month"),
                                cls.reference_date,
                                get_date_expression(),
                            ),
                        ),
                        (
                            frequency_unit == "year",
                            func.datediff(
                                text("year"),
                                cls.reference_date,
                                get_date_expression(),
                            ),
                        ),
                    ]
                ),
                Float,
            )
            / frequency_coefficient
        ),
        Integer,
    )
    * frequency_coefficient
)        
# 计算当前日期与参考日期的差值作为起始点
reference_diff_calc = case(
    [
        (
            frequency_unit == "day",
            func.dateadd(
                text("day"),
                case(
                    [
                        (reference_diff > 0, reference_diff),
                    ],
                    else_=0,
                ),
                cls.reference_date,
            ),
        ),
        (
            frequency_unit == "week",
            func.dateadd(
                text("week"),
                case(
                    [
                        (reference_diff > 0, reference_diff),
                    ],
                    else_=0,
                ),
                cls.reference_date,
            ),
        ),
        (
            frequency_unit == "month",
            func.dateadd(
                text("month"),
                case(
                    [
                        (reference_diff > 0, reference_diff),
                    ],
                    else_=0,
                ),
                cls.reference_date,
            ),
        ),
        (
            frequency_unit == "year",
            func.dateadd(
                text("year"),
                case(
                    [
                        (reference_diff > 0, reference_diff),
                    ],
                    else_=0,
                ),
                cls.reference_date,
            ),
        ),
    ],
)
# 通过减去频率值确定当前维护周期的起始边界
next_greatest_cases = case(
    [
        (
            frequency_unit == "day",
            func.dateadd(
                text("day"),
                frequency_coefficient * -1,
                reference_diff_calc,
            ),
        ),
        (
            frequency_unit == "week",
            func.dateadd(
                text("week"),
                frequency_coefficient * -1,
                reference_diff_calc,
            ),
        ),
        (
            frequency_unit == "month",
            func.dateadd(
                text("month"),
                frequency_coefficient * -1,
                reference_diff_calc,
            ),
        ),
        (
            frequency_unit == "year",
            func.dateadd(
                text("year"),
                frequency_coefficient * -1,
                reference_diff_calc,
            ),
        ),
    ]
)
next_greatest_calc = case(
    [
        (next_greatest_cases > cls.reference_date, next_greatest_cases),
    ],
    else_=cls.reference_date,
)

执行问题 #

通过ScalarSelect获取frequency_coefficient和frequency_unit的方式，导致最终执行的SQL包含大量重复子查询，触发DBAPI的子查询数量限制。部分执行SQL片段如下：

FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) ELSE ? END, pm_schedules.reference_date) WHEN ((SELECT TOP 1 pm_frequencys.unit_of_time 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) = ?) THEN dateadd(month, CASE WHEN (CAST(ceiling(CAST(CASE WHEN ((SELECT TOP 1 pm_frequencys.unit_of_time 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) = ?) THEN datediff(month, pm_schedules.reference_date, CAST(GETDATE() AS DATE)) END AS FLOAT) / (SELECT TOP 1 pm_frequencys.coefficient 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id)) AS INTEGER) * (SELECT TOP 1 pm_frequencys.coefficient 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) > ?) THEN CAST(ceiling(CAST(CASE WHEN ((SELECT TOP 1 pm_frequencys.unit_of_time 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) = ?) THEN datediff(month, pm_schedules.reference_date, CAST(GETDATE() AS DATE)) END AS FLOAT) / (SELECT TOP 1 pm_frequencys.coefficient 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id)) AS INTEGER) * (SELECT TOP 1 pm_frequencys.coefficient 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) ELSE ? END, pm_schedules.reference_date) WHEN ((SELECT TOP 1 pm_frequencys.unit_of_time 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) = ?) THEN dateadd(year, CASE WHEN (CAST(ceiling(CAST(CASE WHEN ((SELECT TOP 1 pm_frequencys.unit_of_time 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) = ?) THEN datediff(month, pm_schedules.reference_date, CAST(GETDATE() AS DATE)) END AS FLOAT) / (SELECT TOP 1 pm_frequencys.coefficient 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id)) AS INTEGER) * (SELECT TOP 1 pm_frequencys.coefficient 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) > ?) THEN CAST(ceiling(CAST(CASE WHEN ((SELECT TOP 1 pm_frequencys.unit_of_time 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id) = ?) THEN datediff(month, pm_schedules.reference_date, CAST(GETDATE() AS DATE)) END AS FLOAT) / (SELECT TOP 1 pm_frequencys.coefficient 
FROM pm_frequencys 
WHERE pm_schedules.pm_frequency_id = pm_frequencys.id)) AS INTEGER) * (SELECT TOP 1 pm_frequencys.coefficient 

现状与需求 #

• 仅保留单个case条件时，查询可快速完成
• 保留单个分组全条件时，查询耗时20-30秒
• 全条件触发DBAPI错误
• 由于处于hybrid expression中，无法提前执行查询（如改用.scalar()会触发UnboundExecutionError）

需要找到复用/缓存ScalarSelect结果的方法或其他优化方案，比如使用DECLARE、CTE等。

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