SET SQL_MODE='ALLOW_INVALID_DATES';
use kylo;
delimiter //

create procedure update_to_040()

begin

DECLARE output VARCHAR(4000) DEFAULT '';
SELECT CONCAT(' STARTING Update schema') into output;

IF EXISTS(SELECT table_name
            FROM INFORMATION_SCHEMA.COLUMNS
           WHERE table_schema = 'kylo'
             AND table_name = 'BATCH_NIFI_JOB' and COLUMN_NAME ='FLOW_FILE_UUID')
THEN
ALTER TABLE `kylo`.`BATCH_NIFI_JOB`
CHANGE COLUMN `FLOW_FILE_UUID` `FLOW_FILE_ID` VARCHAR(255) NOT NULL DEFAULT '';
END IF;

IF NOT EXISTS(SELECT table_name
            FROM INFORMATION_SCHEMA.COLUMNS
           WHERE table_schema = 'kylo'
             AND table_name = 'BATCH_NIFI_STEP' and column_name = 'FLOW_FILE_ID') then
ALTER TABLE `kylo`.`BATCH_NIFI_STEP`
ADD COLUMN `FLOW_FILE_ID` VARCHAR(45) NULL;
end if;

IF NOT EXISTS(SELECT table_name
            FROM INFORMATION_SCHEMA.COLUMNS
           WHERE table_schema = 'kylo'
             AND table_name = 'BATCH_NIFI_STEP' and column_name = 'JOB_FLOW_FILE_ID') then
ALTER TABLE `kylo`.`BATCH_NIFI_STEP`
ADD COLUMN `JOB_FLOW_FILE_ID` VARCHAR(45) NULL;

end if;


IF NOT EXISTS(SELECT table_name
            FROM INFORMATION_SCHEMA.COLUMNS
           WHERE table_schema = 'kylo'
             AND table_name = 'GENERATED_KEYS' ) THEN

CREATE TABLE IF NOT EXISTS `GENERATED_KEYS` (
  `PK_COLUMN` varchar(255) NOT NULL,
  `VALUE_COLUMN` bigint(20) NOT NULL,
  PRIMARY KEY (`PK_COLUMN`)
) ENGINE=InnoDB;


DELETE FROM kylo.BATCH_NIFI_STEP
WHERE STEP_EXECUTION_ID in (
SELECT y.STEP_EXECUTION_ID FROM (
SELECT MIN(STEP_EXECUTION_ID) STEP_EXECUTION_ID, EVENT_ID from BATCH_NIFI_STEP
WHERE EVENT_ID in(
SELECT x.EVENT_ID FROM (
SELECT event_id,  count(*) FROM kylo.BATCH_NIFI_STEP
group by event_id
having count(*) >1) x
) group by EVENT_ID ) as y
);
SELECT CONCAT(output,'\n','update 3.') into output;

DELETE FROM kylo.BATCH_NIFI_STEP
WHERE STEP_EXECUTION_ID in (
SELECT y.STEP_EXECUTION_ID FROM (
SELECT MIN(STEP_EXECUTION_ID) STEP_EXECUTION_ID, EVENT_ID from BATCH_NIFI_STEP
WHERE STEP_EXECUTION_ID in(
SELECT x.STEP_EXECUTION_ID FROM (
SELECT STEP_EXECUTION_ID,  count(*) FROM kylo.BATCH_NIFI_STEP
group by STEP_EXECUTION_ID
having count(*) >1) x
)
group by EVENT_ID) as y
);



/**
INSERT THE MAX ids into the GENERATED_KEYS table
*/

INSERT INTO `kylo`.`GENERATED_KEYS`
(`PK_COLUMN`,
`VALUE_COLUMN`)
SELECT 'JOB_INSTANCE_ID', COALESCE(MAX(JOB_INSTANCE_ID),0)+1 FROM BATCH_JOB_INSTANCE;

INSERT INTO `kylo`.`GENERATED_KEYS`
(`PK_COLUMN`,
`VALUE_COLUMN`)
SELECT 'JOB_EXECUTION_ID', COALESCE(MAX(JOB_EXECUTION_ID),0)+1 FROM BATCH_JOB_EXECUTION;

INSERT INTO `kylo`.`GENERATED_KEYS`
(`PK_COLUMN`,
`VALUE_COLUMN`)
SELECT 'STEP_EXECUTION_ID', COALESCE(MAX(STEP_EXECUTION_ID),0)+1 FROM BATCH_STEP_EXECUTION;

ELSE

UPDATE `kylo`.`GENERATED_KEYS` SET VALUE_COLUMN = (SELECT COALESCE(MAX(JOB_INSTANCE_ID),0)+1 FROM BATCH_JOB_INSTANCE)
WHERE PK_COLUMN = 'JOB_INSTANCE_ID';

UPDATE `kylo`.`GENERATED_KEYS` SET VALUE_COLUMN = (SELECT COALESCE(MAX(JOB_EXECUTION_ID),0)+1 FROM BATCH_JOB_EXECUTION)
WHERE PK_COLUMN = 'JOB_EXECUTION_ID';

UPDATE `kylo`.`GENERATED_KEYS` SET VALUE_COLUMN = (SELECT COALESCE(MAX(STEP_EXECUTION_ID),0)+1 FROM BATCH_STEP_EXECUTION)
WHERE PK_COLUMN = 'STEP_EXECUTION_ID';

END IF;


CREATE TABLE IF NOT EXISTS `kylo`.`BATCH_JOB_EXECUTION_CTX_VALS` (
  `JOB_EXECUTION_ID` bigint(20) NOT NULL,
  `TYPE_CD` varchar(10) NOT NULL,
  `KEY_NAME` varchar(100) NOT NULL,
  `STRING_VAL` longtext,
  `DATE_VAL` timestamp NULL DEFAULT NULL,
  `LONG_VAL` bigint(20) DEFAULT NULL,
  `DOUBLE_VAL` double DEFAULT NULL,
  `CREATE_DATE` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `ID` varchar(45) NOT NULL,
  UNIQUE KEY `BATCH_JOB_EXEC_CTX_VALS_UQ1` (`JOB_EXECUTION_ID`,`KEY_NAME`),
  KEY `BATCH_JOB_EXEC_CTX_VALS_PK` (`ID`)
) ENGINE=InnoDB;

CREATE TABLE IF NOT EXISTS `kylo`.`BATCH_STEP_EXECUTION_CTX_VALS` (
  `JOB_EXECUTION_ID` bigint(20) NOT NULL,
  `STEP_EXECUTION_ID` bigint(20) NOT NULL DEFAULT '0',
  `EXECUTION_CONTEXT_TYPE` varchar(6) DEFAULT NULL,
  `TYPE_CD` varchar(10) NOT NULL,
  `KEY_NAME` varchar(100) NOT NULL,
  `STRING_VAL` longtext,
  `DATE_VAL` timestamp NULL DEFAULT NULL,
  `LONG_VAL` bigint(20) DEFAULT NULL,
  `DOUBLE_VAL` double DEFAULT NULL,
  `CREATE_DATE` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `ID` varchar(45) DEFAULT NULL,
  KEY `BATCH_STEP_EXEC_CTX_VALS_PK` (`ID`),
  KEY `BATCH_STEP_EXEC_CTX_VALS_UQ1` (`STEP_EXECUTION_ID`,`KEY_NAME`)
) ENGINE=InnoDB;



CREATE TABLE IF NOT EXISTS `kylo`.`NIFI_EVENT` (
  `EVENT_ID` bigint(20) NOT NULL,
  `FLOW_FILE_ID` varchar(45) NOT NULL,
  `FM_FEED_NAME` varchar(255) DEFAULT NULL,
  `EVENT_TIME` timestamp NULL DEFAULT NULL,
  `EVENT_DURATION_MILLIS` bigint(20) DEFAULT NULL,
  `EVENT_TYPE` varchar(45) DEFAULT NULL,
  `SOURCE_CONNECTION_ID` varchar(45) DEFAULT NULL,
  `PROCESSOR_ID` varchar(45) DEFAULT NULL,
  `FEED_PROCESS_GROUP_ID` varchar(45) DEFAULT NULL,
  `FILE_SIZE` varchar(100) DEFAULT NULL,
  `FILE_SIZE_BYTES` bigint(20) DEFAULT NULL,
  `PARENT_FLOW_FILE_IDS` varchar(255) DEFAULT NULL,
  `CHILD_FLOW_FILE_IDS` varchar(255) DEFAULT NULL,
  `ATTRIBUTES_JSON` longtext,
  `EVENT_DETAILS` mediumtext,
  `PROCESSOR_NAME` varchar(255) DEFAULT NULL,
  `JOB_FLOW_FILE_ID` varchar(45) DEFAULT NULL,
  `IS_END_OF_JOB` varchar(1) DEFAULT NULL,
  `IS_START_OF_JOB` varchar(1) DEFAULT NULL,
  `IS_BATCH_JOB` varchar(1) DEFAULT NULL,
  `IS_FAILURE` varchar(1) DEFAULT NULL,
  `IS_FINAL_JOB_EVENT` varchar(45) DEFAULT NULL COMMENT 'Y/N .  This will be set to Y when this is the last event for any and all related parent root flow files.  This indicator marks the true complete end of a large job that may have merged many files together to a single event.',
  `HAS_FAILURE_EVENTS` varchar(1) DEFAULT NULL,
  `VERSION` int(11) DEFAULT NULL,
  `CREATED_TIME` timestamp NULL DEFAULT NULL,
  `MODIFIED_TIME` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`EVENT_ID`,`FLOW_FILE_ID`)
) ENGINE=InnoDB;

CREATE TABLE IF NOT EXISTS `kylo`.`NIFI_FEED_PROCESSOR_STATS` (
  `FM_FEED_NAME` varchar(45) NOT NULL,
  `NIFI_PROCESSOR_ID` varchar(45) NOT NULL,
  `NIFI_FEED_PROCESS_GROUP_ID` varchar(45) DEFAULT NULL,
  `COLLECTION_TIME` timestamp NULL DEFAULT NULL,
  `TOTAL_EVENTS` int(11) DEFAULT NULL,
  `DURATION_MILLIS` bigint(19) DEFAULT NULL,
  `BYTES_IN` bigint(19) DEFAULT NULL,
  `BYTES_OUT` bigint(19) DEFAULT NULL,
  `MIN_EVENT_TIME` timestamp NULL DEFAULT NULL,
  `MAX_EVENT_TIME` timestamp NULL DEFAULT NULL,
  `JOBS_STARTED` bigint(20) DEFAULT NULL,
  `JOBS_FINISHED` bigint(20) DEFAULT NULL,
  `JOBS_FAILED` bigint(20) DEFAULT NULL,
  `PROCESSORS_FAILED` bigint(20) DEFAULT NULL,
  `FLOW_FILES_STARTED` bigint(20) DEFAULT NULL,
  `FLOW_FILES_FINISHED` bigint(20) DEFAULT NULL,
  `COLLECTION_ID` varchar(45) DEFAULT NULL,
  `COLLECTION_INTERVAL_SEC` bigint(20) DEFAULT NULL,
  `ID` varchar(45) NOT NULL,
  `PROCESSOR_NAME` varchar(255) DEFAULT NULL,
  `JOB_DURATION` bigint(20) DEFAULT NULL,
  `SUCCESSFUL_JOB_DURATION` bigint(20) DEFAULT NULL,
  PRIMARY KEY (`ID`),
  KEY `FM_FEED_NAME` (`FM_FEED_NAME`),
  KEY `MIN_EVENT_TIME` (`MIN_EVENT_TIME`),
  KEY `MAX_EVENT_TIME` (`MAX_EVENT_TIME`)
) ENGINE=InnoDB;

CREATE TABLE IF NOT EXISTS `NIFI_RELATED_ROOT_FLOW_FILES` (
  `FLOW_FILE_ID` varchar(45) NOT NULL,
  `RELATION_ID` varchar(45) NOT NULL,
  `EVENT_ID` bigint(20) DEFAULT NULL,
  `EVENT_FLOW_FILE_ID` varchar(45) DEFAULT NULL,
  PRIMARY KEY (`FLOW_FILE_ID`)
) ENGINE=InnoDB;

/**
Remove any orphan records so FKs work
 */
DELETE FROM BATCH_NIFI_JOB
WHERE JOB_EXECUTION_ID NOT IN(SELECT JOB_EXECUTION_ID from BATCH_JOB_EXECUTION);

DELETE FROM BATCH_NIFI_STEP
WHERE STEP_EXECUTION_ID NOT IN(SELECT STEP_EXECUTION_ID from BATCH_STEP_EXECUTION);





CREATE TABLE IF NOT EXISTS `FEED` (
  `id` binary(16) NOT NULL,
  `name` varchar(100) NOT NULL,
`description` varchar(255) DEFAULT NULL,
FEED_TYPE varchar(45) NULL,
`created_time` timestamp,
  `modified_time` timestamp,
 PRIMARY KEY (`id`)
 ) ENGINE=InnoDB;

IF NOT EXISTS(SELECT table_name
            FROM INFORMATION_SCHEMA.COLUMNS
           WHERE table_schema = 'kylo'
             AND table_name = 'FEED' and COLUMN_NAME ='FEED_TYPE')
THEN
ALTER TABLE `kylo`.`FEED`
ADD COLUMN `FEED_TYPE` VARCHAR(45) NULL;

END IF;




CREATE TABLE IF NOT EXISTS `KYLO_VERSION` (
   ID binary(16) NOT NULL,
  `MAJOR_VERSION` varchar(45) NOT NULL,
  `MINOR_VERSION` varchar(45) DEFAULT NULL,
  `DESCRIPTION` varchar(255) DEFAULT NULL,
  `CREATED_TIME` timestamp NULL DEFAULT NULL,
  `MODIFIED_TIME` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`ID`)
) ENGINE=InnoDB;



CREATE TABLE IF NOT EXISTS `SLA_ASSESSMENT` (
  `ID` binary(16) NOT NULL,
  `SLA_ID` varchar(55) DEFAULT NULL,
  `MESSAGE` varchar(255) DEFAULT NULL,
  `RESULT` varchar(255) DEFAULT NULL,
  `CREATED_TIME` timestamp NULL DEFAULT NULL,
  `MODIFIED_TIME` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`ID`)
) ENGINE=InnoDB;

CREATE TABLE IF NOT EXISTS `SLA_METRIC_ASSESSMENT` (
  `ID` binary(16) NOT NULL,
  `MESSAGE` varchar(255) DEFAULT NULL,
  `RESULT` varchar(255) DEFAULT NULL,
  `CREATED_TIME` timestamp NULL DEFAULT NULL,
  `MODIFIED_TIME` timestamp NULL DEFAULT NULL,
  `SLA_OBLIGATION_ASSESSMENT_ID` binary(16) DEFAULT NULL,
  `COMPARABLES` longtext,
  `DATA` longtext,
  `METRIC_TYPE` varchar(255) DEFAULT NULL,
  METRIC_DESCRIPTION varchar(255) DEFAULT NULL,
  PRIMARY KEY (`ID`)
) ENGINE=InnoDB;

CREATE TABLE IF NOT EXISTS `SLA_OBLIGATION_ASSESSMENT` (
  `ID` binary(16) NOT NULL,
  `SLA_OBLIGATION_ID` varchar(255) DEFAULT NULL,
  `RESULT` varchar(255) DEFAULT NULL,
  `CREATED_TIME` timestamp NULL DEFAULT NULL,
  `MODIFIED_TIME` timestamp NULL DEFAULT NULL,
  `SLA_ASSESSMENT_ID` binary(16) DEFAULT NULL,
  `COMPARABLES` longtext,
  `MESSAGE` varchar(255) DEFAULT NULL,
  `OBLIGATION_ID` varchar(45) DEFAULT NULL,
  PRIMARY KEY (`ID`)
) ENGINE=InnoDB;


IF NOT EXISTS(SELECT table_name
            FROM INFORMATION_SCHEMA.COLUMNS
           WHERE table_schema = 'kylo'
             AND table_name = 'SLA_METRIC_ASSESSMENT' and COLUMN_NAME ='METRIC_DESCRIPTION')
THEN
ALTER TABLE `kylo`.`SLA_METRIC_ASSESSMENT`
ADD COLUMN `METRIC_DESCRIPTION` VARCHAR(255) NULL;

END IF;



ALTER TABLE `kylo`.`BATCH_JOB_EXECUTION_PARAMS`
CHANGE COLUMN `IDENTIFYING` `IDENTIFYING` CHAR(1) NOT NULL DEFAULT 'N';

IF NOT EXISTS(SELECT CONSTRAINT_NAME
            FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS
            WHERE TABLE_NAME = 'BATCH_NIFI_JOB'
            AND CONSTRAINT_NAME = 'fk_JOB_EXECUTION')
THEN
alter table BATCH_NIFI_JOB
ADD CONSTRAINT fk_JOB_EXECUTION
FOREIGN KEY(JOB_EXECUTION_ID)
REFERENCES BATCH_JOB_EXECUTION(JOB_EXECUTION_ID);
end if;

IF NOT EXISTS(SELECT CONSTRAINT_NAME
            FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS
            WHERE TABLE_NAME = 'BATCH_NIFI_STEP'
            AND CONSTRAINT_NAME = 'fk_STEP_EXECUTION')
THEN
alter table  BATCH_NIFI_STEP
ADD CONSTRAINT fk_STEP_EXECUTION
FOREIGN KEY(STEP_EXECUTION_ID)
REFERENCES BATCH_STEP_EXECUTION(STEP_EXECUTION_ID);
end if;

IF NOT EXISTS(SELECT CONSTRAINT_NAME
            FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS
            WHERE TABLE_NAME = 'BATCH_NIFI_JOB'
            AND CONSTRAINT_NAME = 'PRIMARY')
THEN
ALTER TABLE `kylo`.`BATCH_NIFI_JOB`
ADD PRIMARY KEY (`EVENT_ID`, `FLOW_FILE_ID`)  ,
ADD UNIQUE INDEX `BATCH_NIFI_JOB_UQ1` (`JOB_EXECUTION_ID` ASC)  ;
END IF;

IF NOT EXISTS(SELECT CONSTRAINT_NAME
            FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS
            WHERE TABLE_NAME = 'BATCH_NIFI_STEP'
            AND CONSTRAINT_NAME = 'PRIMARY')
THEN
ALTER TABLE `kylo`.`BATCH_NIFI_STEP`
ADD PRIMARY KEY (`EVENT_ID`, `FLOW_FILE_ID`)  ,
ADD UNIQUE INDEX `BATCH_NIFI_STEP_UQ1` (`STEP_EXECUTION_ID` ASC),
ADD UNIQUE INDEX `BATCH_NIFI_STEP_UQ2` (`COMPONENT_ID` ASC, `JOB_FLOW_FILE_ID` ASC),
ADD INDEX `BATCH_NIFI_STEP_IDX1` (`JOB_FLOW_FILE_ID` ASC);
END IF;

END//

delimiter ;

-- Execute the procedure
call update_to_040();

-- Drop the procedure
drop procedure update_to_040;