PostgreSQL按thing_id批量更新：限数量+按日期排序需求及问题

需求概述 #

• 开发PostgreSQL查询，更新最多LIMIT个唯一thing.thing_id对应的所有行
• 按status_1_date列排序，旧数据优先更新
• thing_id非主键，代表由多行组成的实体
• 需处理大型数据库，限制处理的实体总数，且保证并发执行时不重复更新

示例表结构 #

CREATE TABLE IF NOT EXISTS thing (
  name VARCHAR(255) PRIMARY KEY,
  thing_id VARCHAR(255),
  c_id VARCHAR(255),
  status VARCHAR(255),
  etag VARCHAR(255),
  last_modified VARCHAR(255),
  size VARCHAR(255),
  status_1_date DATE
);

示例数据 #

INSERT INTO thing (name,thing_id, c_id,status, status_1_date)
values 
('protocol://full/path/to/thing/thing_1.file1', 'thing_1','c_id', 'status_1', '2023-09-29 09:00:01'),
('protocol://full/path/to/thing/thing_1.file2', 'thing_1','c_id', 'status_1', '2023-09-29 09:00:02'),
('protocol://full/path/to/thing/thing_2.file5', 'thing_2','c_id', 'status_1', '2023-09-29 09:00:02.5'),
('protocol://something else', 'thing_1','c_id', 'status_1', '2023-09-29 09:00:02.8'),
('protocol://full/path/to/thing/thing_2.file1', 'thing_2','c_id', 'status_1', '2023-09-29 09:00:03'),
('protocol://full/path/to/thing/thing_2.file2', 'thing_2','c_id', 'status_1', '2023-09-29 09:00:04'),
('protocol://full/path/to/thing/thing_2.file3', 'thing_2','c_id', 'status_1', '2023-09-29 09:00:05'),
('protocol://full/path/to/thing/thing_2.file4', 'thing_2','different', 'status_1', '2023-09-29 09:00:05'),
('protocol://full/path/to/thing/thing_3.file1', 'thing_3','c_id', 'status_1', '2023-09-29 09:00:06'),
('protocol://full/path/to/thing/thing_3.file2', 'thing_3','c_id', 'status_1', '2023-09-29 09:00:06.2'),
('protocol://full/path/to/thing/thing_4.file1', 'thing_4','c_id', 'status_1', '2023-09-29 09:00:06.1'),
('protocol://full/path/to/thing/thing_4.file2', 'thing_4','c_id', 'status_1', '2023-09-29 09:00:06.3'),
('protocol://full/path/to/thing/thing_5.file1', 'thing_5','c_id', 'status_1', '2023-09-29 09:00:06.4'),
('protocol://full/path/to/thing/thing_5.file2', 'thing_5','c_id', 'status_1', '2023-09-29 09:00:06.5'),
('protocol://full/path/to/thing/thing_5.file3', 'thing_5','c_id', 'status_1', '2023-09-29 09:00:06.6'),
('protocol://full/path/to/thing/thing_6.file1', 'thing_6','c_id', 'status_1', '2023-09-29 09:00:06.7');

期望结果 #

当设置LIMIT 3时，更新thing_id为thing_1、thing_2、thing_3对应行（行1-9）中符合WHERE条件的所有行。

现有方案问题 #

• 参考答案未按最旧的thing_id优先更新
• 并发执行相同条件查询时会重复更新，海量数据场景下不可接受

尝试方案及问题 #

尝试1 #

尝试用FOR UPDATE避免重复更新，但窗口函数无法与FOR UPDATE共存：

with ranked_matching_things as (
  select
    dense_rank() over (order by thing_id) as thing_rank,
    name
  from thing
  where c_id = 'c_id'
    and name like 'protocol://full/path/to/thing/%'
  order by status_1_date
  for update
)
update thing
set status = 'CHANGED'
from ranked_matching_things
where thing.name = ranked_matching_things.name
  and ranked_matching_things.thing_rank <= 3

错误提示：ERROR: FOR UPDATE is not allowed with window functions

尝试2 #

存在排序错误（未按最旧thing_id优先），且WHERE子句重复：

UPDATE thing
SET status = 'CHANGED'
FROM (
    SELECT name 
    FROM thing
    WHERE thing.thing_id in (
        SELECT DISTINCT thing.thing_id
        FROM thing
        WHERE c_id = 'c_id' AND name like 'protocol://full/path/to/thing/%' AND status = 'status_1'
        ORDER BY thing.thing_id
        LIMIT (3)
    ) AND thing.c_id = 'c_id' AND thing.name like 'protocol://full/path/to/thing/%' AND thing.status = 'status_1'
    ORDER BY thing.status_1_date
    FOR UPDATE
) AS rows
WHERE thing.name = rows.name 
RETURNING *

解决方案 #

并发安全的正确查询 #

要满足按status_1_date排序选择最旧的N个thing_id，同时并发更新不重复，可以分两步锁定目标thing_id，再更新对应行：

WITH target_thing_ids AS (
    -- 先锁定要处理的thing_id，按最小的status_1_date排序，取前3个
    SELECT DISTINCT thing_id
    FROM thing
    WHERE c_id = 'c_id'
      AND name LIKE 'protocol://full/path/to/thing/%'
      AND status = 'status_1'
    ORDER BY MIN(status_1_date) OVER (PARTITION BY thing_id)
    LIMIT 3
    FOR UPDATE SKIP LOCKED -- 跳过已被锁定的行，避免并发重复处理
),
target_rows AS (
    -- 获取这些thing_id对应的所有符合条件的行
    SELECT name
    FROM thing
    WHERE thing_id IN (SELECT thing_id FROM target_thing_ids)
      AND c_id = 'c_id'
      AND name LIKE 'protocol://full/path/to/thing/%'
      AND status = 'status_1'
    FOR UPDATE
)
UPDATE thing
SET status = 'CHANGED'
FROM target_rows
WHERE thing.name = target_rows.name
RETURNING *;

方案说明 #

1. target_thing_ids CTE：

   • 按每个thing_id的最小status_1_date排序，确保最旧的实体优先被选中
   • 使用FOR UPDATE SKIP LOCKED，并发执行时会跳过已被其他事务锁定的thing_id，避免重复更新
   • 限制返回最多3个唯一thing_id

2. target_rows CTE：

   • 获取选中thing_id对应的所有符合条件的行，并锁定这些行，防止其他事务修改

3. UPDATE语句：

   • 更新锁定的目标行，确保数据一致性

内容的提问来源于stack exchange，提问作者Cogito Ergo Sum