-- 1. Пытаемся вставить строку в table_replica, если не получилось - обновляем
CREATE OR REPLACE FUNCTION upsert_table_replica()
RETURNS TRIGGER AS $$
BEGIN
-- Вставка записи в table_replica с обновлением в случае конфликта по id
INSERT INTO table_replica (id, field_1, ...)
VALUES (NEW.id, NEW.field_1, ...)
ON CONFLICT (id)
DO UPDATE SET
field_1 = EXCLUDED.field_1,
...
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
-- 2. Создаем триггер типа AFTER на создание и обновление строк в table
CREATE TRIGGER trigger_upsert_table
AFTER INSERT OR UPDATE ON table
FOR EACH ROW
EXECUTE FUNCTION upsert_table_replica();-- 3. Создаем триггер на удаление для таблицы table
CREATE OR REPLACE FUNCTION delete_table_replica()
RETURNS TRIGGER AS $$
BEGIN
-- Удаление записи из table_replica по id
DELETE FROM table_replica
WHERE id = OLD.id;
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trigger_delete_table
AFTER DELETE ON table
FOR EACH ROW
EXECUTE FUNCTION delete_table_replica();drop function if exists copy_rows_from_table(size integer);
create or replace function copy_rows_from_table(size integer) returns bool as $$
declare
all_duplicated bool;
begin
with batch as (
select
*
from table as ts
where ts.id > (select last_id from dup_id) and ts.id < (select max_id from dup_id)
order by ts.id asc
limit size
for update -- чтобы в момент переноса ничего не происходило со строками из батча
), duplicated as (
insert into table_replica select * from batch on conflict(id) do nothing
), duplication_offset as (
-- Запоминаем последний id из батча
-- если в батче для переноса ничего нет, то оставляем текущий id
update dup_ids set last_id = COALESCE((select max(id) from batch), last_id)
)
-- Запрос необходимо повторять до тех пор, пока значение all_done не станет true
-- если в батче переноса ничего нет, то продолжать нет смысла
select count(*)::integer = 0 as all_done into all_duplicated from batch;
return all_duplicated;
end;
$$ language plpgsql;
Дубликатор + триггеры в работе
До перестановки
Поломка вставки после перестановки
begin;
set local lock_timeout to '5s';
set local statement_timeout to '10s';
lock table table in ACCESS EXCLUSIVE MODE;
lock table table_replica in ACCESS EXCLUSIVE MODE;
alter table table rename to table_temp;
alter table table_replica rename to table;
-- дополнительный шаг, так как мы используем сиквенс оригинала
-- без выполнения этого шага не удастся удалить table_temp
alter sequence table_id_seq owned by table.id;
commit;SELECT
t.relname AS table_name,
i.relname AS index_name,
s.idx_scan,
s.idx_tup_read,
s.idx_tup_fetch,
pg_size_pretty(pg_relation_size(s.indexrelid)) as size
FROM
pg_stat_user_indexes s
JOIN
pg_index x ON s.indexrelid = x.indexrelid
JOIN
pg_class t ON x.indrelid = t.oid
JOIN
pg_class i ON s.indexrelid = i.oid
where
s.schemaname = 'public' and s.idx_scan = 0 and s.idx_tup_read = 0 and s.idx_tup_fetch = 0
order by
i.relname desc;