在Synapse SQL中实现无循环的任务分配至最少任务用户

解决方案代码 #

-- 模拟用户表和任务表
WITH Users AS (
    SELECT 'A3' AS UserID, 12 AS TotalTasks UNION ALL
    SELECT 'A4' AS UserID, 14 AS TotalTasks UNION ALL
    SELECT 'A5' AS UserID, 11 AS TotalTasks
),
Tasks AS (
    SELECT 1 AS TaskID, 4 AS NewTask UNION ALL
    SELECT 2 AS TaskID, 5 AS NewTask UNION ALL
    SELECT 3 AS TaskID, 3 AS NewTask UNION ALL
    SELECT 4 AS TaskID, 2 AS NewTask
),
-- 给任务按ID排序，生成处理顺序
OrderedTasks AS (
    SELECT *, 
           ROW_NUMBER() OVER (ORDER BY TaskID) AS TaskSeq
    FROM Tasks
),
-- 递归CTE处理任务分配
TaskAssignments AS (
    -- 锚点：处理第一个任务
    SELECT 
        ot.TaskID,
        ot.NewTask,
        u.UserID,
        u.TotalTasks + ot.NewTask AS UpdatedTasks,
        ot.TaskSeq,
        -- 保存当前所有用户的最新任务量，用JSON格式方便后续提取
        JSON_OBJECT_AGG(u.UserID, CASE WHEN u.UserID = selected.UserID THEN u.TotalTasks + ot.NewTask ELSE u.TotalTasks END) AS UserTaskState
    FROM OrderedTasks ot
    CROSS JOIN (
        SELECT TOP 1 UserID, TotalTasks
        FROM Users
        ORDER BY TotalTasks, UserID
    ) selected
    JOIN Users u ON u.UserID = selected.UserID
    WHERE ot.TaskSeq = 1

    UNION ALL

    -- 递归：处理后续任务
    SELECT 
        ot.TaskID,
        ot.NewTask,
        selected.UserID,
        current_user_tasks.TotalTasks + ot.NewTask AS UpdatedTasks,
        ot.TaskSeq,
        JSON_OBJECT_AGG(u.UserID, CASE WHEN u.UserID = selected.UserID THEN current_user_tasks.TotalTasks + ot.NewTask ELSE current_user_tasks.TotalTasks END) AS UserTaskState
    FROM TaskAssignments ta
    JOIN OrderedTasks ot ON ot.TaskSeq = ta.TaskSeq + 1
    -- 解析上一轮的用户任务状态，得到当前所有用户的最新任务量
    CROSS APPLY OPENJSON(ta.UserTaskState)
        WITH (UserID VARCHAR(10) '$', TotalTasks INT '$.value') AS current_user_tasks
    JOIN Users u ON u.UserID = current_user_tasks.UserID
    -- 筛选当前任务量最少的用户（并列时按UserID取第一个）
    CROSS JOIN (
        SELECT TOP 1 UserID, TotalTasks
        FROM OPENJSON(ta.UserTaskState)
            WITH (UserID VARCHAR(10) '$', TotalTasks INT '$.value')
        ORDER BY TotalTasks, UserID
    ) selected
    WHERE current_user_tasks.UserID = selected.UserID
)
-- 输出最终结果
SELECT TaskID, NewTask, UserID, UpdatedTasks
FROM TaskAssignments
ORDER BY TaskID;

代码说明 #

1. 模拟测试数据：用CTE构造Users和Tasks表，与输入数据完全匹配。
2. 任务排序：OrderedTasks给任务按TaskID生成顺序编号，确保严格按要求的顺序处理任务。
3. 递归CTE核心逻辑：
   • 锚点成员：处理第一个任务，从初始用户列表中选出总任务量最小的用户（若有并列，按UserID排序取第一个），计算更新后的任务量，同时用JSON_OBJECT_AGG把所有用户的最新任务状态打包成JSON，供递归步骤读取。
   • 递归成员：每次读取上一轮保存的用户任务状态JSON，解析出当前所有用户的任务量，再筛选出任务量最小的用户分配当前任务，更新该用户的任务量后，重新生成新的用户任务状态JSON。
4. 结果输出：从递归结果中提取所需字段，按TaskID排序得到符合要求的输出格式。

关键注意事项 #

• 用JSON_OBJECT_AGG和OPENJSON传递用户任务状态，避免使用循环，完全适配Synapse SQL的递归CTE规则。
• 当多个用户任务量相同时，通过ORDER BY TotalTasks, UserID保证分配逻辑的稳定性，与示例输出一致。
• 方案支持任意数量的用户和任务，无需修改核心逻辑即可扩展。

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