PostgreSQL/flow_runs/filter 对大型流运算表进行状态过滤的 START_TIME_ASC 查询超时

from datetime import datetime, timedelta, timezone
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterStartTime,
    FlowRunFilterState,
    FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType
from prefect.client.schemas.sorting import FlowRunSort
async with get_client() as client:
    runs = await client.read_flow_runs(
        flow_run_filter=FlowRunFilter(
            state=FlowRunFilterState(
                type=FlowRunFilterStateType(any_=[StateType.RUNNING])
            ),
            start_time=FlowRunFilterStartTime(
                before_=datetime.now(timezone.utc) - timedelta(minutes=15)
            ),
        ),
        sort=FlowRunSort.START_TIME_ASC,
        limit=200,
        offset=0,
    )
我们的 `flow_run` 表包含大约 635 万行（26 GB），其中包括大约 27,000 个 `RUNNING` 行。
Prefect 生成了以下查询形状：

```sql
SELECT ...
FROM flow_run
WHERE flow_run.state_type IN ('RUNNING')
  AND coalesce(flow_run.start_time, flow_run.expected_start_time) <= :cutoff
ORDER BY coalesce(flow_run.start_time, flow_run.expected_start_time) ASC
LIMIT 200 OFFSET 0;
现有的 `(state_type, start_time)` 索引无法服务于 `coalesce(...)` 表达式。现有的 `coalesce(start_time, expected_start_time)` 索引不包含 `state_type`。
因此，我们顺序扫描了表达式索引，然后按状态进行过滤：

```text
Limit  (actual time=59751.738..61319.498 rows=200)
  ->  Index Scan Backward using ix_flow_run__coalesce_start_time_expected_start_time_desc
        Filter: (state_type = 'RUNNING'::state_type)
        Rows Removed by Filter: 5497752
Execution Time: 61319.544 ms
最终，API 请求返回了：

```text
PrefectHTTPStatusError: Server error '500 Internal Server Error'
for url '/api/flow_runs/filter'
Response: {'exception_message': 'Internal Server Error'}
对应的服务器异常如下：

```text
File "prefect/server/api/flow_runs.py", line 551, in read_flow_runs
    db_flow_runs = await models.flow_runs.read_flow_runs(...)
File "prefect/server/models/flow_runs.py", line 341, in read_flow_runs
    result = await session.execute(query)
...
File "asyncpg/prepared_stmt.py", line 177, in fetch
    data = await self.__bind_execute(args, 0, timeout)
...
TimeoutError
我们验证了添加以下索引可解决病理计划问题：

```sql
CREATE INDEX CONCURRENTLY ix_flow_run__state_type_coalesce_start_time
ON flow_run (
    state_type,
    (coalesce(start_time, expected_start_time))
);
添加后，相同的 API 请求在 0.385 秒内返回 200 行。一个以前在第一页上失败的流程随后检索了 4,000 个匹配的行。