

-- Manual upgrade script as of JULY 2025 from 3.9 to 3.15 - for PostgreSQL databases.
-- STOP the SymmetricDS service before applying these changes!
-- Overview:
-- Stages 100x: Truncate tables with trasitory data, unnecessary for the upgrade itself.
-- Stages 200-400: Migrate 3 biggest tables to new structure and purge in one swoop.
-- Set name of the SymmetricDS catalog (user schema in Oracle):
-- ALTER SESSION SET current_schema = CORP?; 


-- Stage 100. Clear temporary data in tables:
truncate table sym_incoming_batch;
truncate table sym_data_gap;
truncate table sym_node_communication;

-- Stage 110. Backup and rename large tables:
start transaction;

  -- Backup old tables:
alter table sym_outgoing_batch rename to bkpsym_outgoing_batch;
alter table sym_data_event rename to bkpsym_data_event;
alter table sym_data rename to bkpsym_data;

-- Rename indexes on tables to eliminate conflict when creating on the new table.
alter index if exists sym_data_pkey rename to bkpsym_data_pkey;
alter index if exists sym_outgoing_batch_pkey rename to bkpsym_outgoing_batch_pkey;
alter index if exists sym_idx_ob_in_error rename to bkpsym_idx_ob_in_error;
alter index if exists sym_idx_ob_status rename to bkpsym_idx_ob_status;


-- Stage 120. Gather statistics on older tables:
dbms_stats.gather_table_stats(ownname => 'CORP_OWNER'
                                tabname =>'BKPSYM_OUTGOING_BATCH',
                                estimate_percent => 0.1,
                                block_sample => true,
                                cascade => false,
                                method_opt => 'FOR ALL COLUMNS SIZE 1',
                                degree => 8);

dbms_stats.gather_table_stats(ownname => 'CORP_OWNER'
                                tabname =>'SYM_DATA_EVENT',
                                estimate_percent => 0.1,
                                block_sample => true,
                                cascade => false,
                                method_opt => 'FOR ALL COLUMNS SIZE 1',
                                degree => 8);

dbms_stats.gather_table_stats(ownname => 'CORP_OWNER'
                                tabname =>'SYM_DATA',
                                estimate_percent => 0.1,
                                block_sample => true,
                                cascade => false,
                                method_opt => 'FOR ALL COLUMNS SIZE 1',
                                degree => 8);



-- Stage 300. Convert large tables to the format of the 3.15 version of SymmetricDS:
ALTER TABLE SYM_OUTGOING_BATCH RENAME TO BKPSYM_OUTGOING_BATCH;
ALTER TABLE SYM_DATA_EVENT RENAME TO BKPSYM_DATA_EVENT;
ALTER TABLE SYM_DATA RENAME TO BKPSYM_DATA;


-- Stage 310. Convert table SYM_OUTGOING_BATCH
CREATE TABLE SYM_OUTGOING_BATCH(
    "BATCH_ID" NUMBER NOT NULL,
    "NODE_ID" VARCHAR2(50) NOT NULL,
    "CHANNEL_ID" VARCHAR2(128),
    "STATUS" CHAR(2),
    "ERROR_FLAG" NUMBER(1) DEFAULT 0,
    "SQL_STATE" VARCHAR2(10),
    "SQL_CODE" NUMBER DEFAULT 0 NOT NULL,
    "SQL_MESSAGE" CLOB,
    "LAST_UPDATE_HOSTNAME" VARCHAR2(255),
    "LAST_UPDATE_TIME" TIMESTAMP(6),
    "CREATE_TIME" TIMESTAMP(6),
    "SUMMARY" VARCHAR2(255),
    "IGNORE_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "BYTE_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "LOAD_FLAG" NUMBER(1) DEFAULT 0,
    "EXTRACT_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "SENT_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "LOAD_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "RELOAD_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "OTHER_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "DATA_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "EXTRACT_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "LOAD_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "DATA_INSERT_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "DATA_UPDATE_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "DATA_DELETE_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "EXTRACT_INSERT_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "EXTRACT_UPDATE_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "EXTRACT_DELETE_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "LOAD_INSERT_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "LOAD_UPDATE_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "LOAD_DELETE_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "NETWORK_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "FILTER_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "LOAD_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "ROUTER_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "EXTRACT_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "TRANSFORM_EXTRACT_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "TRANSFORM_LOAD_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "LOAD_ID" NUMBER,
    "COMMON_FLAG" NUMBER(1) DEFAULT 0,
    "FALLBACK_INSERT_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "FALLBACK_UPDATE_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "CONFLICT_WIN_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "CONFLICT_LOSE_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "IGNORE_ROW_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "MISSING_DELETE_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "SKIP_COUNT" NUMBER DEFAULT 0 NOT NULL,
    "TOTAL_EXTRACT_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "TOTAL_LOAD_MILLIS" NUMBER DEFAULT 0 NOT NULL,
    "EXTRACT_JOB_FLAG" NUMBER(1) DEFAULT 0,
    "EXTRACT_START_TIME" TIMESTAMP(6),
    "TRANSFER_START_TIME" TIMESTAMP(6),
    "LOAD_START_TIME" TIMESTAMP(6),
    "FAILED_DATA_ID" NUMBER DEFAULT 0 NOT NULL,
    "FAILED_LINE_NUMBER" NUMBER DEFAULT 0 NOT NULL,
    "CREATE_BY" VARCHAR2(255),
    "BULK_LOADER_FLAG" NUMBER(1) DEFAULT 0
) NOLOGGING;


INSERT INTO /*+ APPEND PARALLEL(SYM_OUTGOING_BATCH,8) */ SYM_OUTGOING_BATCH
  SELECT /*+ PARALLEL(BKPSYM_OUTGOING_BATCH,8) */
    BATCH_ID,
    NODE_ID,
    CHANNEL_ID,
    STATUS,
    ERROR_FLAG ,
    NULL AS SQL_STATE,
    0 AS SQL_CODE,
    NULL AS SQL_MESSAGE,
    LAST_UPDATE_HOSTNAME,
    LAST_UPDATE_TIME,
    CREATE_TIME ,
    'OLDER_SYMMETRIC_VERSION' AS SUMMARY,
    0 AS IGNORE_COUNT ,
    BYTE_COUNT ,
    0 AS LOAD_FLAG,
    0 AS EXTRACT_COUNT,
    0 AS SENT_COUNT,
    0 AS LOAD_COUNT,
    0 AS RELOAD_ROW_COUNT,
    0 AS OTHER_ROW_COUNT,
    0 AS DATA_ROW_COUNT,
    0 AS EXTRACT_ROW_COUNT,
    0 AS LOAD_ROW_COUNT,
    0 AS DATA_INSERT_ROW_COUNT,
    0 AS DATA_UPDATE_ROW_COUNT,
    0 AS DATA_DELETE_ROW_COUNT,
    0 AS EXTRACT_INSERT_ROW_COUNT,
    0 AS EXTRACT_UPDATE_ROW_COUNT,
    0 AS EXTRACT_DELETE_ROW_COUNT,
    0 AS LOAD_INSERT_ROW_COUNT,
    0 AS LOAD_UPDATE_ROW_COUNT,
    0 AS LOAD_DELETE_ROW_COUNT,
    NETWORK_MILLIS,
    FILTER_MILLIS,
    0 AS LOAD_MILLIS,
    0 AS ROUTER_MILLIS,
    0 AS EXTRACT_MILLIS,
    0 AS TRANSFORM_EXTRACT_MILLIS,
    0 AS TRANSFORM_LOAD_MILLIS,
    0 AS LOAD_ID,
    0 AS COMMON_FLAG,
    0 AS FALLBACK_INSERT_COUNT,
    0 AS FALLBACK_UPDATE_COUNT,
    0 AS CONFLICT_WIN_COUNT,
    0 AS CONFLICT_LOSE_COUNT,
    0 AS IGNORE_ROW_COUNT,
    MISSING_DELETE_COUNT,
    SKIP_COUNT,
    0 AS TOTAL_EXTRACT_MILLIS,
    0 AS TOTAL_LOAD_MILLIS,
    0 AS EXTRACT_JOB_FLAG,
    CREATE_TIME AS EXTRACT_START_TIME,
    CREATE_TIME AS TRANSFER_START_TIME,
    CREATE_TIME AS LOAD_START_TIME,
    0 AS FAILED_DATA_ID,
    FAILED_LINE_NUMBER,
    'OLDER_SYMMETRIC_VERSION' AS CREATE_BY,
    0 AS BULK_LOADER_FLAG
  FROM BKPSYM_OUTGOING_BATCH
  WHERE STATUS != 'OK'
;
COMMIT;


CREATE INDEX SYM_IDX_OB_NODE_STATUS ON SYM_OUTGOING_BATCH ("NODE_ID", "STATUS", "CHANNEL_ID");
COMMIT;

CREATE INDEX SYM_IDX_OB_STATUS ON SYM_OUTGOING_BATCH("STATUS");
COMMIT;

CREATE INDEX SYM_IDX_OB_IN_ERROR ON SYM_OUTGOING_BATCH("ERROR_FLAG");
COMMIT;

ALTER TABLE SYM_OUTGOING_BATCH logging;
COMMIT;

ALTER TABLE SYM_OUTGOING_BATCH ADD CONSTRAINT "SYM_OUTGOING_BATCH_PK" PRIMARY KEY ("BATCH_ID", "NODE_ID");
COMMIT;



-- Gather statistics on new table:
dbms_stats.gather_table_stats(ownname => 'CORP_OWNER'
                                tabname =>'BKPSYM_OUTGOING_BATCH',
                                estimate_percent => 0.1,
                                block_sample => true,
                                cascade => false,
                                method_opt => 'FOR ALL COLUMNS SIZE 1',
                                degree => 8);


ALTER INDEX SYM_IDX_OB_NODE_STATUS logging;
COMMIT;


ALTER INDEX SYM_IDX_OB_STATUS logging;
COMMIT;


ALTER INDEX SYM_IDX_OB_IN_ERROR logging;
COMMIT;



-- Stage 320. Convert table SYM_DATA_EVENT:
CREATE TABLE SYM_DATA_EVENT(
    "DATA_ID" NUMBER NOT NULL,
    "BATCH_ID" NUMBER NOT NULL,
    "CREATE_TIME" TIMESTAMP(6)
) NOLOGGING;


INSERT INTO /*+ APPEND PARALLEL(SYM_DATA_EVENT,8) */ SYM_DATA_EVENT
  WITH OUTGOING as ( select /*+ INLINE */ DISTINCT BATCH_ID from SYM_OUTGOING_BATCH )
  SELECT /*+ PARALLEL(8) */
    be.DATA_ID,
    be.BATCH_ID,
    be.CREATE_TIME
  FROM BKPSYM_DATA_EVENT be
  join OUTGOING ON be.BATCH_ID = OUTGOING.BATCH_ID;
 ;
COMMIT;


CREATE INDEX SYM_IDX_DE_BATCHID ON SYM_DATA_EVENT ("BATCH_ID")
;
COMMIT;

ALTER TABLE SYM_DATA_EVENT logging
;
COMMIT;


ALTER TABLE SYM_DATA_EVENT ADD CONSTRAINT "SYM_DATA_EVENT_PK" PRIMARY KEY ("DATA_ID", "BATCH_ID")
;
COMMIT;


-- Gather statistics on new table:
dbms_stats.gather_table_stats(ownname => 'CORP_OWNER'
                                tabname =>'SYM_DATA_EVENT',
                                estimate_percent => 0.1,
                                block_sample => true,
                                cascade => false,
                                method_opt => 'FOR ALL COLUMNS SIZE 1',
                                degree => 8);


ALTER INDEX SYM_IDX_DE_BATCHID logging
;
COMMIT;


-- Stage 330. Convert table SYM_DATA:
CREATE TABLE SYM_DATA(
    "DATA_ID" NUMBER NOT NULL ,
    "TABLE_NAME" VARCHAR2(255) NOT NULL,
    "EVENT_TYPE" CHAR(1) NOT NULL,
    "ROW_DATA" CLOB,
    "PK_DATA" CLOB,
    "OLD_DATA" CLOB,
    "TRIGGER_HIST_ID" NUMBER NOT NULL,
    "CHANNEL_ID" VARCHAR2(128),
    "TRANSACTION_ID" VARCHAR2(255),
    "SOURCE_NODE_ID" VARCHAR2(50),
    "EXTERNAL_DATA" VARCHAR2(50),
    "NODE_LIST" VARCHAR2(255),
    "IS_PREROUTED" NUMBER(1) DEFAULT 0 NOT NULL,
    "CREATE_TIME" TIMESTAMP(6)
) NOLOGGING;


INSERT INTO /*+ APPEND PARALLEL(SYM_DATA,8) */ SYM_DATA
  WITH EVENT as ( select /*+ INLINE */ DISTINCT DATA_ID from SYM_DATA_EVENT )
  SELECT /*+ PARALLEL(8) */
    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,
    0 AS IS_PREROUTED,
    d.CREATE_TIME
  FROM BKPSYM_DATA d
  JOIN EVENT ON d.DATA_ID = EVENT.DATA_ID
;
COMMIT;


CREATE UNIQUE INDEX "SYM_IDX_D_CHANNEL_ID" ON SYM_DATA ("DATA_ID", "CHANNEL_ID")
;
COMMIT;

ALTER TABLE SYM_DATA logging
;
COMMIT;


ALTER INDEX SYM_IDX_D_CHANNEL_ID logging
;
COMMIT;


ALTER TABLE SYM_DATA ADD CONSTRAINT "SYM_DATA_PK" PRIMARY KEY ("DATA_ID")
;
COMMIT;

 
-- Gather statistics on new table:
dbms_stats.gather_table_stats(ownname => 'CORP_OWNER'
                                tabname =>'SYM_DATA',
                                estimate_percent => 0.1,
                                block_sample => true,
                                cascade => false,
                                method_opt => 'FOR ALL COLUMNS SIZE 1',
                                degree => 8);




-- Stage 500. Validation:
SELECT * FROM (
	SELECT 'SYM_OUTGOING_BATCH' AS table_name, Count(*) AS total_rows FROM SYM_OUTGOING_BATCH
	UNION ALL
	SELECT 'BKPSYM_OUTGOING_BATCH not OK' AS table_name, Count(*) AS total_rows FROM BKPSYM_OUTGOING_BATCH WHERE STATUS != 'OK'
	UNION ALL
	SELECT 'SYM_DATA' AS table_name, Count(*) AS total_rows FROM SYM_DATA
	UNION ALL
	SELECT 'BKPSYM_DATA' AS table_name, Count(*) AS total_rows FROM BKPSYM_DATA
	UNION ALL
	SELECT 'SYM_DATA_EVENT' AS table_name, Count(*) AS total_rows FROM SYM_DATA_EVENT
	UNION ALL
	SELECT 'BKPSYM_DATA_EVENT' AS table_name, Count(*) AS total_rows FROM BKPSYM_DATA_EVENT
);


-- Stage 800. Database performance tuning
-- TODO: Refresh statictics on all SYM_% tables which contain more than 1000 records ( manually)


