/* SANDBOX.DQ_LOG - TABLE TO STORE DQ LOG */

CREATE TABLE IF NOT EXISTS SANDBOX.DQ_LOG (
	LOAD_TS TIMESTAMP,
	OBJECT VARCHAR(256),
    DQ_NAME VARCHAR(256),
	METRIC FLOAT,
    STATUS VARCHAR(64),
	DESCRIPTION VARCHAR(2048)
)
ORDER BY LOAD_TS, OBJECT
UNSEGMENTED ALL NODES
;

/* SANDBOX.STG_COMMENTS - TABLE WITH DATA (STAGE TABLE) */


CREATE TABLE IF NOT EXISTS SANDBOX.STG_COMMENTS (
	SRC_SYSTEM VARCHAR(64) NOT NULL DEFAULT 'EXCEL FILE',
	SCENARIO VARCHAR(64) NOT NULL DEFAULT 'COMMENTS ON SHARED FOLDER W:',
	BUSINESS_DT DATE NOT NULL DEFAULT GETDATE(),
	ID VARCHAR(12) NOT NULL,
	DT DATE NOT NULL,
	PLANT VARCHAR(12),
	SUBPLANT VARCHAR(12),
	COMMENT VARCHAR(12000),
	TEMPERATURE NUMERIC(8,2),
	PRIMARY KEY (ID) DISABLED
)
ORDER BY ID
SEGMENTED BY HASH(MD5(ID::VARCHAR)) ALL NODES
;

/* DQ CHECK #1 - SANDBOX.STG_COMMENTS_DQ_SOURCE_TARGET 
	Check if source.rowcount (from log) == stage.rowcount (fact count in dwh)
*/

INSERT INTO SANDBOX.DQ_LOG (
	LOAD_TS,
	OBJECT,
    DQ_NAME,
	METRIC,
    STATUS,
	DESCRIPTION
)
SELECT DISTINCT
	GETDATE()
	, 'STG_COMMENTS' AS OBJECT
	, 'COMPARE SOURCE AND STAGE' AS DQ_NAME
	, SOURCE_COUNT.ROWCOUNT - STAGE_COUNT.ROWCOUNT AS METRIC
	, CASE (SOURCE_COUNT.ROWCOUNT - STAGE_COUNT.ROWCOUNT) WHEN 0 THEN 'OK' ELSE 'WARN' END AS STATUS
	, 'SOURCE.ROWCOUNT - STAGE.ROWCOUNT. OK=0. COMPARE NUMBER OF ROWS IN SOURCE AND NUMBER OF ROWS LOADED TO STAGE' AS DESCRIPTION
FROM SANDBOX.ETL_LOAD_LOG
	CROSS JOIN (SELECT count(1) AS ROWCOUNT FROM SANDBOX.STG_COMMENTS) AS STAGE_COUNT
	CROSS JOIN (SELECT ROWCOUNT FROM SANDBOX.ETL_LOAD_LOG
				WHERE OBJECT = 'STG_COMMENTS'
				AND STEP = 'LOAD FILE'
				LIMIT 1 OVER (PARTITION BY OBJECT ORDER BY LOAD_TS DESC)) AS SOURCE_COUNT
;

/* DQ CHECK #2 - SANDBOX.STG_COMMENTS_DQ_FILE_EXISTS 
	Check if file from this.source_system at this.business_date already been loaded
*/

INSERT INTO SANDBOX.DQ_LOG (
	LOAD_TS,
	OBJECT,
    DQ_NAME,
	METRIC,
    STATUS,
	DESCRIPTION
)
SELECT DISTINCT
	GETDATE()
	, FILE_REG.STAGE_TABLE AS OBJECT
	, 'CHECK IF FILE ALREADY EXISTS' AS DQ_NAME
	, FILE_REG.FILE_COUNT AS METRIC
	, CASE FILE_REG.FILE_COUNT WHEN 0 THEN 'OK' ELSE 'WARN' END AS STATUS
	, 'CHECK IF FILE FOR BUSINESS DATE FROM GIVEN DATA SOURCE ALREADY BEEN REGISTERED. OK=0 (NEW FILE)' AS DESCRIPTION
FROM (SELECT
			'STG_COMMENTS'	AS STAGE_TABLE,
			COUNT(DISTINCT 1) AS FILE_COUNT
		FROM SANDBOX.STG_COMMENTS src
			INNER JOIN SANDBOX.ETL_FILE_LOAD fl
				ON fl.SRC_SYSTEM = src.SRC_SYSTEM
					AND fl.FILE_NAME = src.SCENARIO
					AND fl.BUSINESS_DT = src.BUSINESS_DT) AS FILE_REG

;

/* SANDBOX.STG_COMMENTS_ETL_FILE_LOAD 
	Register file in dwh (id stamp with metadata)
*/

INSERT INTO SANDBOX.ETL_FILE_LOAD (
	SRC_SYSTEM ,
	FILE_NAME ,
	BUSINESS_DT ,
	LOAD_TS
)
SELECT DISTINCT
	src.SRC_SYSTEM AS SRC_SYSTEM ,
	src.SCENARIO AS FILE_NAME ,
	src.BUSINESS_DT AS BUSINESS_DT ,
	GETDATE()
FROM SANDBOX.STG_COMMENTS src
	LEFT JOIN SANDBOX.ETL_FILE_LOAD fl
		ON fl.SRC_SYSTEM = src.SRC_SYSTEM
			AND fl.FILE_NAME = src.SCENARIO
			AND fl.BUSINESS_DT = src.BUSINESS_DT
WHERE fl.FILE_ID IS NULL
;

/* DQ CHECK #3 - SANDBOX.STG_COMMENTS_DQ_FILE_REGISTERED 
	Check if file's metadata has been registered (there is row for this.file)
*/

INSERT INTO SANDBOX.DQ_LOG (
	LOAD_TS,
	OBJECT,
    DQ_NAME,
	METRIC,
    STATUS,
	DESCRIPTION
)
SELECT DISTINCT
	GETDATE()
	, FILE_REG.STAGE_TABLE AS OBJECT
	, 'CHECK FILE REGISTERED' AS DQ_NAME
	, FILE_REG.FILE_COUNT AS METRIC
	, CASE FILE_REG.FILE_COUNT WHEN 1 THEN 'OK' ELSE 'WARN' END AS STATUS
	, 'CHECK IF FILE HAS BEEN REGISTERED. OK=1 (FILE FOUND)' AS DESCRIPTION
FROM (SELECT
			'STG_COMMENTS'	AS STAGE_TABLE,
			COUNT(DISTINCT 1) AS FILE_COUNT
		FROM SANDBOX.STG_COMMENTS src
			INNER JOIN SANDBOX.ETL_FILE_LOAD fl
				ON fl.SRC_SYSTEM = src.SRC_SYSTEM
					AND fl.FILE_NAME = src.SCENARIO
					AND fl.BUSINESS_DT = src.BUSINESS_DT) AS FILE_REG
;


/* SANDBOX.STG_COMMENTS_DQ_SOURCE_TARGET */

/* SANDBOX.STG_COMMENTS_DQ_SOURCE_TARGET */

