UTM5 log tables rotation (Postgres)

gistfile1.txt

CREATE OR REPLACE FUNCTION log_tables_rotate(v_table_name text)
  RETURNS boolean AS
$BODY$
DECLARE
	rec record;
	rotate record;

	last_rotation_end_date int;
	
	rotate_period_time timestamp with time zone;
	rotate_period_start int;
	rotate_period_end int;
	rotate_period_suffix text;

	new_rotated_table_name text;
	tmp_table_name text;
BEGIN
	SELECT * FROM log_tables_rotation r WHERE r.table_name = v_table_name INTO rotate;
	IF NOT FOUND THEN
		RAISE EXCEPTION 'Wrong table name';
	END IF;

	-- Период: предыдущий месяц
	SELECT now() - '1 month'::interval INTO rotate_period_time;
	SELECT extract('epoch' from date_trunc('month', now() - '1 month'::interval)) INTO rotate_period_start;
	SELECT extract('epoch' from date_trunc('month', now()) - '1 second'::interval) INTO rotate_period_end;
	SELECT '_' || to_char(rotate_period_time, 'YYYY') || '_' || trim(leading '0' FROM to_char(rotate_period_time, 'MM')) INTO rotate_period_suffix;
			
	SELECT max(a.end_date) FROM archives as a WHERE a.table_type = rotate.archives_table_type INTO last_rotation_end_date;

	IF last_rotation_end_date IS NOT NULL THEN
		-- Проверка: была ли уже сделана ротация
		IF last_rotation_end_date > rotate_period_start + 3600 THEN
			RETURN false;
		END IF;

		-- начальное время ротации = время прошлой ротации + 1 сек
		rotate_period_start = last_rotation_end_date + 1;
	END IF;	
	
	-- RAISE NOTICE 'rotate_period_start = %, rotate_period_end = %, rotate_period_suffix = %', to_timestamp(rotate_period_start), to_timestamp(rotate_period_end), rotate_period_suffix;

	tmp_table_name = rotate.table_name || '_prod';
	new_rotated_table_name = rotate.table_name || rotate_period_suffix;
	EXECUTE 'CREATE TABLE ' || quote_ident(tmp_table_name) || ' (LIKE ' || quote_ident(rotate.table_name) || ' INCLUDING ALL)';

	-- Copy owner
	SELECT t.tableowner FROM pg_tables t WHERE t.tablename = rotate.table_name INTO rec;
	EXECUTE 'ALTER TABLE ' || quote_ident(tmp_table_name) || ' OWNER TO ' || rec.tableowner;

	-- Copy grants
	FOR rec IN SELECT rtg.table_name, rtg.grantee, array_agg(privilege_type::text) as priveleges
		FROM information_schema.role_table_grants AS rtg
		WHERE rtg.table_name = rotate.table_name
		GROUP BY rtg.table_name, rtg.grantee
	LOOP
		EXECUTE 'GRANT ' || array_to_string(rec.priveleges, ', ') || ' ON TABLE ' || quote_ident(tmp_table_name) || ' TO ' || rec.grantee;
	END LOOP;

	-- Swap names
	EXECUTE 'ALTER TABLE ' || quote_ident(rotate.table_name) || ' RENAME TO ' || quote_ident(new_rotated_table_name);
	EXECUTE 'ALTER TABLE ' || quote_ident(tmp_table_name) || ' RENAME TO ' || quote_ident(rotate.table_name);

	-- Delete redundant indexes
	FOR rec IN (
		WITH keep_indexes AS (SELECT * FROM unnest(rotate.archive_keep_indexes) as tpl)
		SELECT pg_indexes.*, (tc.constraint_type IS NOT NULL AND tc.constraint_type = 'PRIMARY KEY') as is_pkey
		FROM pg_indexes 
		LEFT JOIN keep_indexes ON pg_indexes.indexdef LIKE keep_indexes.tpl
		LEFT JOIN information_schema.table_constraints tc ON tc.table_name = pg_indexes.tablename AND tc.constraint_name = pg_indexes.indexname
		WHERE pg_indexes.tablename = new_rotated_table_name AND keep_indexes.tpl IS NULL)
	LOOP
		IF rec.is_pkey THEN
			EXECUTE 'ALTER TABLE ' || quote_ident(new_rotated_table_name) || ' DROP CONSTRAINT ' || quote_ident(rec.indexname);
		ELSE
			EXECUTE 'DROP INDEX ' || quote_ident(rec.indexname);
		END IF;
	END LOOP;

	-- Move data		
	EXECUTE 'INSERT INTO ' || quote_ident(rotate.table_name) || ' SELECT * FROM ' || quote_ident(new_rotated_table_name) || ' WHERE ' 
		|| quote_ident(rotate.date_column) || ' < ' || quote_literal(rotate_period_start) || ' OR '
		|| quote_ident(rotate.date_column) || ' > ' || quote_literal(rotate_period_end);
	EXECUTE 'DELETE FROM ' || quote_ident(new_rotated_table_name) || ' WHERE '
		|| quote_ident(rotate.date_column) || ' < ' || quote_literal(rotate_period_start) || ' OR '
		|| quote_ident(rotate.date_column) || ' > ' || quote_literal(rotate_period_end);			

	-- add `archives` rec
	INSERT INTO archives (archive_id, table_type, table_name, start_date, end_date) VALUES(
		(SELECT coalesce(max(archive_id)+1, 1) FROM archives a WHERE a.table_type = rotate.archives_table_type),
		rotate.archives_table_type,
		new_rotated_table_name,
		rotate_period_start,
		rotate_period_end
	);

	RETURN true;
END
$BODY$
  LANGUAGE plpgsql VOLATILE SECURITY DEFINER
  COST 100;

-------------------------------------

CREATE TABLE log_tables_rotation
(
  table_name text NOT NULL,
  archives_table_type integer NOT NULL, -- ID в таблицы `archives.table_type`
  date_column text NOT NULL,
  archive_keep_indexes text[] NOT NULL DEFAULT ARRAY[]::text[],
  CONSTRAINT log_tables_rotation_pkey PRIMARY KEY (table_name)
)
WITH (
  OIDS=FALSE
);
COMMENT ON COLUMN log_tables_rotation.archives_table_type IS 'ID в таблицы `archives.table_type`';

--------------------------------------

INSERT INTO log_tables_rotation (table_name, archives_table_type, date_column, archive_keep_indexes) VALUES ('tel_sessions_log', 3, 'recv_date', '{"CREATE INDEX % ON % USING btree (session_start_date)"}');
INSERT INTO log_tables_rotation (table_name, archives_table_type, date_column, archive_keep_indexes) VALUES ('payment_transactions', 7, 'payment_enter_date', '{}');
INSERT INTO log_tables_rotation (table_name, archives_table_type, date_column, archive_keep_indexes) VALUES ('tel_sessions_detail', 4, 'recv_date', '{"CREATE INDEX % ON % USING btree (dhs_sess_id)"}');
INSERT INTO log_tables_rotation (table_name, archives_table_type, date_column, archive_keep_indexes) VALUES ('dhs_sessions_detail', 6, 'recv_date', '{}');
INSERT INTO log_tables_rotation (table_name, archives_table_type, date_column, archive_keep_indexes) VALUES ('dhs_sessions_log', 5, 'recv_date', '{}');
INSERT INTO log_tables_rotation (table_name, archives_table_type, date_column, archive_keep_indexes) VALUES ('discount_transactions_iptraffic_all', 2, 'discount_date', '{}');
INSERT INTO log_tables_rotation (table_name, archives_table_type, date_column, archive_keep_indexes) VALUES ('discount_transactions_all', 1, 'discount_date', '{"CREATE INDEX % ON % USING btree (account_id)","CREATE UNIQUE INDEX % ON % USING btree (id)"}');