-- Manual purge script as of OCTOBER 2025 for the version 3.16 powered by the PostgreSQL databases.
-- STOP the Purge Outgoing and Purge Incoming jobs in SymmetricDS before executing Stages 100-199!
-- STOP the SymmetricDS service completely before executing Stages 200+ !
-- Overview:
-- Stages 100x: Set up backup tables (and copy 99% of data).
-- Stages 200-400: Second pass of the backup (the remaining 1%) and truncate 3 biggest tables in one fell swoop.
 
-- Stage 101. Helper table for recording current reference point for First Pass of backups:
CREATE TABLE bkpsym_max_data_batch (
  step_id int8 NOT NULL,
	data_id  int8 NOT NULL,
	batch_id int8 NOT NULL,
	create_time  timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
	CONSTRAINT bkpsym_max_data_batch_pkey PRIMARY KEY (step_id)
);

-- Stage 101. First pass of backup = sym_outgoing_batch (skips completed batches):
CREATE UNLOGGED TABLE bkpsym_outgoing_batch AS 
  SELECT batch_id,node_id,channel_id,status,error_flag,sql_state,sql_code,sql_message,last_update_hostname,last_update_time
    ,create_time,summary,ignore_count,byte_count,load_flag,extract_count,sent_count,load_count,reload_row_count
    ,other_row_count,data_row_count,extract_row_count,load_row_count,data_insert_row_count,data_update_row_count
    ,data_delete_row_count,extract_insert_row_count,extract_update_row_count,extract_delete_row_count
    ,load_insert_row_count,load_update_row_count,load_delete_row_count,network_millis,filter_millis,load_millis
    ,router_millis,extract_millis,transform_extract_millis,transform_load_millis,load_id,thread_id,common_flag
    ,fallback_insert_count,fallback_update_count,conflict_win_count,conflict_lose_count,ignore_row_count
    ,missing_delete_count,skip_count,total_extract_millis,total_load_millis,extract_job_flag,extract_start_time
    ,transfer_start_time,load_start_time,failed_data_id,failed_line_number
  FROM sym_outgoing_batch
  WHERE status<>'OK' AND NOT (channel_id IN ('heartbeat','monitor') AND status='NE');

CREATE UNIQUE INDEX IF NOT EXISTS bkpsym_outgoing_batch_udx ON bkpsym_outgoing_batch (batch_id, node_id);


-- Stage 102. First pass of backup = sym_data_event (skips completed batches):
CREATE UNLOGGED TABLE bkpsym_data_event AS 
  WITH OUTGOING as ( select DISTINCT batch_id from bkpsym_outgoing_batch )
  SELECT e.data_id, e.batch_id, e.create_time
  FROM sym_data_event e
  join OUTGOING ON e.batch_id = OUTGOING.batch_id; 


CREATE INDEX IF NOT EXISTS bkpsym_data_event_data_idx ON bkpsym_data_event (data_id);

CREATE INDEX IF NOT EXISTS bkpsym_data_event_batch_idx ON bkpsym_data_event (batch_id);


-- Stage 103. Record current reference point for the First Pass of backups:
INSERT INTO bkpsym_max_data_batch (step_id, batch_id, data_id) 
  SELECT 1 as step_id, max(batch_id), max(data_id) from bkpsym_data_event;


-- Stage 104. First pass of backup = sym_data:
CREATE UNLOGGED TABLE bkpsym_data AS 
  WITH EVENT as ( select DISTINCT data_id from bkpsym_data_event )
  SELECT  
    d.data_id, d.TABLE_NAME, d.EVENT_TYPE, d.ROW_DATA, d.PK_DATA,
    d.OLD_DATA, d.TRIGGER_HIST_ID, d.CHANNEL_ID, d.TRANSACTION_ID,
    d.SOURCE_NODE_ID, d.EXTERNAL_DATA, d.NODE_LIST,
    d.IS_PREROUTED, d.CREATE_TIME
  FROM EVENT
  JOIN sym_data d ON d.data_id = EVENT.data_id;

CREATE UNIQUE INDEX IF NOT EXISTS bkpsym_data_pkey ON bkpsym_data (data_id);


-- Stage 200. Stop the SymmetricDS service NOW!

select * from bkpsym_max_data_batch;

-- Stage 201. Second pass of backup and purge for SYM_OUTGOING_BATCH (run commands closely together for data integrity!):
INSERT INTO bkpsym_outgoing_batch 
    (batch_id,node_id,channel_id,status,error_flag,sql_state,sql_code,sql_message,last_update_hostname,last_update_time
    ,create_time,summary,ignore_count,byte_count,load_flag,extract_count,sent_count,load_count,reload_row_count
    ,other_row_count,data_row_count,extract_row_count,load_row_count,data_insert_row_count,data_update_row_count
    ,data_delete_row_count,extract_insert_row_count,extract_update_row_count,extract_delete_row_count
    ,load_insert_row_count,load_update_row_count,load_delete_row_count,network_millis,filter_millis,load_millis
    ,router_millis,extract_millis,transform_extract_millis,transform_load_millis,load_id,thread_id,common_flag
    ,fallback_insert_count,fallback_update_count,conflict_win_count,conflict_lose_count,ignore_row_count
    ,missing_delete_count,skip_count,total_extract_millis,total_load_millis,extract_job_flag,extract_start_time
    ,transfer_start_time,load_start_time,failed_data_id,failed_line_number)
  SELECT
    batch_id,node_id,channel_id,status,error_flag,sql_state,sql_code,sql_message,last_update_hostname,last_update_time
    ,create_time,summary,ignore_count,byte_count,load_flag,extract_count,sent_count,load_count,reload_row_count
    ,other_row_count,data_row_count,extract_row_count,load_row_count,data_insert_row_count,data_update_row_count
    ,data_delete_row_count,extract_insert_row_count,extract_update_row_count,extract_delete_row_count
    ,load_insert_row_count,load_update_row_count,load_delete_row_count,network_millis,filter_millis,load_millis
    ,router_millis,extract_millis,transform_extract_millis,transform_load_millis,load_id,thread_id,common_flag
    ,fallback_insert_count,fallback_update_count,conflict_win_count,conflict_lose_count,ignore_row_count
    ,missing_delete_count,skip_count,total_extract_millis,total_load_millis,extract_job_flag,extract_start_time
    ,transfer_start_time,load_start_time,failed_data_id,failed_line_number
  FROM sym_outgoing_batch
  WHERE batch_id > (select max(batch_id) from bkpsym_max_data_batch);

TRUNCATE TABLE sym_outgoing_batch;

INSERT INTO sym_outgoing_batch 
    (batch_id,node_id,channel_id,status,error_flag,sql_state,sql_code,sql_message,last_update_hostname,last_update_time
    ,create_time,summary,ignore_count,byte_count,load_flag,extract_count,sent_count,load_count,reload_row_count
    ,other_row_count,data_row_count,extract_row_count,load_row_count,data_insert_row_count,data_update_row_count
    ,data_delete_row_count,extract_insert_row_count,extract_update_row_count,extract_delete_row_count
    ,load_insert_row_count,load_update_row_count,load_delete_row_count,network_millis,filter_millis,load_millis
    ,router_millis,extract_millis,transform_extract_millis,transform_load_millis,load_id,thread_id,common_flag
    ,fallback_insert_count,fallback_update_count,conflict_win_count,conflict_lose_count,ignore_row_count
    ,missing_delete_count,skip_count,total_extract_millis,total_load_millis,extract_job_flag,extract_start_time
    ,transfer_start_time,load_start_time,failed_data_id,failed_line_number)
  SELECT
    batch_id,node_id,channel_id,status,error_flag,sql_state,sql_code,sql_message,last_update_hostname,last_update_time
    ,create_time,summary,ignore_count,byte_count,load_flag,extract_count,sent_count,load_count,reload_row_count
    ,other_row_count,data_row_count,extract_row_count,load_row_count,data_insert_row_count,data_update_row_count
    ,data_delete_row_count,extract_insert_row_count,extract_update_row_count,extract_delete_row_count
    ,load_insert_row_count,load_update_row_count,load_delete_row_count,network_millis,filter_millis,load_millis
    ,router_millis,extract_millis,transform_extract_millis,transform_load_millis,load_id,thread_id,common_flag
    ,fallback_insert_count,fallback_update_count,conflict_win_count,conflict_lose_count,ignore_row_count
    ,missing_delete_count,skip_count,total_extract_millis,total_load_millis,extract_job_flag,extract_start_time
    ,transfer_start_time,load_start_time,failed_data_id,failed_line_number
  FROM bkpsym_outgoing_batch; 


-- Stage 202. Second pass of backup and purge for SYM_DATA_EVENT (run commands closely together for data integrity!):
INSERT INTO bkpsym_data_event 
  SELECT e.data_id, e.batch_id, e.create_time
  FROM sym_data_event e 
  WHERE batch_id > (select max(batch_id) from bkpsym_max_data_batch)
    and data_id > (select max(data_id) from bkpsym_max_data_batch); 

TRUNCATE TABLE sym_data_event;

INSERT INTO sym_data_event
  SELECT data_id, batch_id, create_time
  FROM bkpsym_data_event; 


-- Stage 203. Second pass of backup and purge for SYM_DATA (run commands closely together for data integrity!):
INSERT INTO bkpsym_data 
  ( DATA_ID, TABLE_NAME, EVENT_TYPE, ROW_DATA, PK_DATA,
  OLD_DATA, TRIGGER_HIST_ID, CHANNEL_ID, TRANSACTION_ID,
  SOURCE_NODE_ID, EXTERNAL_DATA, NODE_LIST,
  IS_PREROUTED, CREATE_TIME) 
SELECT  
  d.DATA_ID, d.TABLE_NAME, d.EVENT_TYPE, d.ROW_DATA, d.PK_DATA,
  d.OLD_DATA, d.TRIGGER_HIST_ID, d.CHANNEL_ID, d.TRANSACTION_ID,
  d.SOURCE_NODE_ID, d.EXTERNAL_DATA, d.NODE_LIST,
  d.IS_PREROUTED, d.CREATE_TIME
FROM sym_data d
  WHERE data_id > (select max(data_id) from bkpsym_max_data_batch); 

TRUNCATE TABLE sym_data; 

INSERT INTO sym_data 
  ( DATA_ID, TABLE_NAME, EVENT_TYPE, ROW_DATA, PK_DATA,
  OLD_DATA, TRIGGER_HIST_ID, CHANNEL_ID, TRANSACTION_ID,
  SOURCE_NODE_ID, EXTERNAL_DATA, NODE_LIST,
  IS_PREROUTED, CREATE_TIME)
SELECT  
  DATA_ID, TABLE_NAME, EVENT_TYPE, ROW_DATA, PK_DATA,
  OLD_DATA, TRIGGER_HIST_ID, CHANNEL_ID, TRANSACTION_ID,
  SOURCE_NODE_ID, EXTERNAL_DATA, NODE_LIST,
  IS_PREROUTED, CREATE_TIME
FROM bkpsym_data;


-- Stage 204. Record current reference point for the Second Pass of backups:
INSERT INTO bkpsym_max_data_batch (step_id, batch_id, data_id) 
  SELECT 2 as step_id, max(batch_id), max(data_id) from bkpsym_data_event;


-- Stage 301. Truncate SYM_DATA_GAP and reset purge job variables to scan the entire data table:
TRUNCATE TABLE SYM_DATA_GAP;

DELETE from sym_context where name in ('purge.last.event.batch.id',
	'purge.last.data.id', 'purge.last.batch.id'); 



-- Stage 400. Start the SymmetricDS service back up NOW!


-- Stage 500. Review:
SELECT * FROM (
	SELECT 0 as ord, 'bkpsym_max_data_batch' AS table_name, Count(*) AS total_rows FROM bkpsym_max_data_batch
	UNION ALL
	SELECT 1 as ord, 'SYM_OUTGOING_BATCH' AS table_name, Count(*) AS total_rows FROM SYM_OUTGOING_BATCH
	UNION ALL
	SELECT 2 as ord, 'BKPSYM_OUTGOING_BATCH' AS table_name, Count(*) AS total_rows FROM BKPSYM_OUTGOING_BATCH
	UNION ALL
	SELECT 3 as ord, 'SYM_DATA' AS table_name, Count(*) AS total_rows FROM SYM_DATA
	UNION ALL
	SELECT 4 as ord, 'BKPSYM_DATA' AS table_name, Count(*) AS total_rows FROM BKPSYM_DATA
	UNION ALL
	SELECT 5 as ord, 'SYM_DATA_EVENT' AS table_name, Count(*) AS total_rows FROM SYM_DATA_EVENT
	UNION ALL
	SELECT 6 as ord, 'BKPSYM_DATA_EVENT' AS table_name, Count(*) AS total_rows FROM BKPSYM_DATA_EVENT
) report order by ord;


