将数据点和聚合数据点放在同一个表中不是理想的设计模式，最好将它们分开到两个表中.

One table

如果您仍然只想使用一个表，那么可以使用下面的代码.

使用Created_at筛选

添加一个名为created_at的timestamp列，其缺省值为NOW().请注意，值NOW()在postgres中的事务内不会更改，并且它还具有读一致性(快照隔离)，从而防止在该事务开始后由其他进程插入/修改的行被包括在该事务中.

-- Start a transaction
BEGIN;

-- Aggregate
INSERT INTO test (contract, amount, aggregated, is_aggregate)
SELECT contract,
       SUM(amount),
       NULL,
       1
FROM test
WHERE aggregated IS NULL
    OR aggregated = 0
GROUP BY contract
HAVING COUNT(*) > 1;

-- Mark aggregated rows as aggregated
UPDATE test
SET aggregated = 1
WHERE id IN (
    SELECT id
    FROM test
    WHERE (aggregated IS NULL
        OR aggregated = 0)
        AND created_at <> NOW()
    GROUP BY contract
    HAVING COUNT(*) > 1
);

-- Commit the transaction
COMMIT;

使用显式行锁

添加一个名为is_locked的boolean列，其缺省值为NULL.这是因为聚合条件在聚合前后返回不同的行.另一种方法是使用插入行的时间戳在最后的更新中过滤掉它们，但我不确定哪些列可用，因为表定义是部分的.

这段代码不是并行安全的--在任何给定时间只运行一个聚合进程.

-- Start a transaction
BEGIN;

-- Lock the rows you want to aggregate
UPDATE test
SET is_locked = TRUE
WHERE id IN (
    SELECT id
    FROM test
    WHERE is_locked IS NULL
      AND (aggregated IS NULL OR aggregated = 0)
      AND contract IN (
        SELECT contract
        FROM test
        WHERE is_locked IS NULL
          AND (aggregated IS NULL OR aggregated = 0)
        GROUP BY contract
        HAVING COUNT(*) > 1
      )
);

-- Insert the data from the locked rows
INSERT INTO test (contract, amount, aggregated, is_aggregate)
SELECT contract,
       SUM(amount) AS sum_amount,
       NULL,
       1
FROM test
WHERE is_locked = TRUE
GROUP BY contract;

-- Update the aggregated column to set it to 1 for the locked rows
-- and remove the locks
UPDATE test
SET aggregated = 1,
    is_locked = NULL
WHERE is_locked = TRUE;

-- Commit the transaction
COMMIT;

两张桌子

如果您想使用单独的聚合表，请参见下面的内容.

将插入的时间戳添加到测试表中，并将该时间戳用于聚合目的.

Conceptual solution proposal below. The exact syntax depends on which database engine you are using.个

修改后的测试表定义:

 create table test (
    id serial, 
    contract varchar, 
    amount int, 
    created_at timestamp DEFAULT CURRENT_TIMESTAMP
);

新的集结表:

create table test_agg (
    contract varchar, 
    amount int, 
    aggregated_to timestamp
);
CREATE UNIQUE INDEX test_agg_contract_uq ON test_agg (contract);

修改后的SQL语句，如果不存在则插入一行或更新聚合表中的现有行.

INSERT INTO test_agg (contract, amount, aggregated_to)
SELECT t.contract,
       SUM(t.amount),
       MAX(t.created_at)
FROM test t,
LEFT OUTER JOIN test_agg ta ON ta.contract = t.contract
WHERE
    ta.aggregated_to IS NULL
    OR t.created_at > ta.aggregated_to
GROUP BY t.contract
ON DUPLICATE KEY UPDATE
    amount = amount + VALUES(amount),
    aggregated_to = VALUES(aggregated_to);