perf(douyu-tasks): 为斗鱼任务增加归属客服索引字段，彻底消除全表 filesort 慢查询

1. 数据库新增迁移 20261003_0040，在 douyu_tasks 表增加 account_assigned_to 字段，并创建 (account_assigned_to, handbook_scope, id) 与 (account_assigned_to, id) 复合覆盖索引； 2. 迁移脚本自动回填历史 24 万条任务的所属客服 ID； 3. 任务服务与账号路由联动：创建任务时自动继承 account_assigned_to，修改账号分配时同步更新关联任务归属； 4. 斗鱼任务查询路由 _visible_tasks_query 改为通过 account_assigned_to 复合索引过滤，消除与 accounts 表的跨表关联与临时表全量文件排序； 5. 新增 test_douyu_task_assigned_optimization 单元测试并通过整链迁移与回归测试。
2026-10-03 13:10:59 +08:00
parent 294c6cd058
commit 7790be4e04
8 changed files with 527 additions and 3 deletions
@@ -0,0 +1,88 @@
+"""为 douyu_tasks 增加 account_assigned_to 字段与复合索引，彻底优化任务全表排序性能。
+
+背景：
+斗鱼任务表 (douyu_tasks) 历史数据已达数十万行。当客服查看任务列表时，原本通过
+JOIN accounts 表并在 accounts.assigned_to 上过滤，导致 MySQL 必须在临时表中对全部
+匹配任务做 filesort 文件排序（Using temporary; Using filesort），单次查询消耗几百毫秒
+并导致 MySQL 容器 CPU 周期性冲顶到 100%+。
+
+本迁移：
+1. 为 douyu_tasks 表新增 account_assigned_to 列；
+2. 批量回填历史任务的归属客服 ID；
+3. 建立 (account_assigned_to, handbook_scope, id) 与 (account_assigned_to, id) 复合索引，
+   使客服任务列表查询从 O(N log N) 临时表全排序降级为 O(1) 索引倒序定位，彻底消除 filesort。
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20261003_0040"
+down_revision: str | None = "20261002_0039"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _columns(bind, table_name: str) -> set[str]:
+    return {col["name"] for col in sa.inspect(bind).get_columns(table_name)}
+
+
+def _indexes(bind, table_name: str) -> set[str]:
+    return {idx["name"] for idx in sa.inspect(bind).get_indexes(table_name)}
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    existing_cols = _columns(bind, "douyu_tasks")
+
+    if "account_assigned_to" not in existing_cols:
+        op.add_column(
+            "douyu_tasks",
+            sa.Column(
+                "account_assigned_to",
+                sa.Integer(),
+                sa.ForeignKey("users.id"),
+                nullable=True,
+            ),
+        )
+
+    # 历史任务回填归属客服 ID
+    op.execute(
+        """
+        UPDATE douyu_tasks dt
+        JOIN accounts a ON a.id = dt.account_id
+        SET dt.account_assigned_to = a.assigned_to
+        WHERE a.assigned_to IS NOT NULL AND dt.account_assigned_to IS NULL;
+        """
+    )
+
+    existing_indexes = _indexes(bind, "douyu_tasks")
+    if "ix_douyu_tasks_account_assigned_to_id" not in existing_indexes:
+        op.create_index(
+            "ix_douyu_tasks_account_assigned_to_id",
+            "douyu_tasks",
+            ["account_assigned_to", "id"],
+        )
+
+    if "ix_douyu_tasks_assigned_scope_id" not in existing_indexes:
+        op.create_index(
+            "ix_douyu_tasks_assigned_scope_id",
+            "douyu_tasks",
+            ["account_assigned_to", "handbook_scope", "id"],
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    existing_indexes = _indexes(bind, "douyu_tasks")
+
+    if "ix_douyu_tasks_assigned_scope_id" in existing_indexes:
+        op.drop_index("ix_douyu_tasks_assigned_scope_id", table_name="douyu_tasks")
+
+    if "ix_douyu_tasks_account_assigned_to_id" in existing_indexes:
+        op.drop_index("ix_douyu_tasks_account_assigned_to_id", table_name="douyu_tasks")
+
+    existing_cols = _columns(bind, "douyu_tasks")
+    if "account_assigned_to" in existing_cols:
+        op.drop_column("douyu_tasks", "account_assigned_to")