USE DATABASE MYOWN_DB;

CREATE SCHEMA IF NOT EXISTS MYSTREAMS;

CREATE SCHEMA IF NOT EXISTS STAGE_TBLS;

CREATE SCHEMA IF NOT EXISTS INTG_TBLS;

CREATE TABLE STAGE_TBLS.STG_EMPL
( EMPID INT,
EMPNAME VARCHAR(30),
SALARY FLOAT,
AGE INT,
DEPT VARCHAR(10),
LOCATION VARCHAR(20)
);

CREATE STREAM MYSTREAMS.STREAM_EMPL ON TABLE STAGE_TBLS.STG_EMPL;
CREATE STREAM MYSTREAMS.STREAM_EMPL_2 ON TABLE STAGE_TBLS.STG_EMPL;
SHOW STREAMS IN SCHEMA MYSTREAMS;
SELECT * FROM MYSTREAMS.STREAM_EMPL;

CREATE TABLE INTG_TBLS.EMPL
( EMPID INT,EMPNAME VARCHAR(30),SALARY FLOAT,AGE INT,DEPT VARCHAR(15),
LOCATION VARCHAR(20),INSRT_DT DATE,LST_UPDT_DT DATE
);

//Inserts

INSERT INTO STAGE_TBLS.STG_EMPL VALUES
(1, 'Amar', 80000, 35, 'SALES', 'Bangalore'),
(2, 'Bharath', 45000, 26, 'SALES', 'Hyderabad'),
(3, 'Charan', 76000, 34, 'TECHNOLOGY', 'Chennai'),
(4, 'Divya', 52000, 28, 'HR', 'Hyderabad'),
(5, 'Gopal', 24500, 22, 'TECHNOLOGY', 'Bangalore'),
(6, 'Haritha', 42000, 27, 'HR', 'Chennai') ;

SELECT * FROM STAGE_TBLS.STG_EMPL;

SELECT * FROM MYSTREAMS.STREAM_EMPL;

INSERT INTO INTG_TBLS.EMPL
( EMPID, EMPNAME, SALARY, AGE, DEPT, LOCATION, INSRT_DT, LST_UPDT_DT)
SELECT EMPID, EMPNAME, SALARY, AGE, DEPT, LOCATION, CURRENT_DATE, NULL
FROM MYSTREAMS.STREAM_EMPL
WHERE METADATA$ACTION = 'INSERT'
AND METADATA$ISUPDATE = FALSE;

SELECT * FROM INTG_TBLS.EMPL;

SELECT * FROM MYSTREAMS.STREAM_EMPL;

//Updates

SELECT * FROM STAGE_TBLS.STG_EMPL;

UPDATE STAGE_TBLS.STG_EMPL SET SALARY=49000 WHERE EMPID=2;
UPDATE STAGE_TBLS.STG_EMPL SET LOCATION='Pune' WHERE EMPID=5;

SELECT * FROM STAGE_TBLS.STG_EMPL;

SELECT * FROM MYSTREAMS.STREAM_EMPL;

MERGE INTO INTG_TBLS.EMPL E
USING MYSTREAMS.STREAM_EMPL S
ON E.EMPID = S.EMPID
WHEN MATCHED
AND S.METADATA$ACTION ='INSERT'
AND S.METADATA$ISUPDATE ='TRUE'
THEN UPDATE
SET E.EMPNAME = S.EMPNAME,
E.SALARY = S.SALARY,
E.AGE = S.AGE,
E.DEPT = S.DEPT,
E.LOCATION = S.LOCATION,
E.LST_UPDT_DT = CURRENT_DATE;

SELECT * FROM INTG_TBLS.EMPL;

SELECT * FROM MYSTREAMS.STREAM_EMPL;

//Deletes

SELECT * FROM STAGE_TBLS.STG_EMPL;

DELETE FROM STAGE_TBLS.STG_EMPL WHERE EMPID in (3,4);

SELECT * FROM STAGE_TBLS.STG_EMPL;

SELECT * FROM MYSTREAMS.STREAM_EMPL;

MERGE INTO INTG_TBLS.EMPL E
USING MYSTREAMS.STREAM_EMPL S
ON E.EMPID = S.EMPID
WHEN MATCHED
AND S.METADATA$ACTION ='DELETE'
AND S.METADATA$ISUPDATE ='FALSE'
THEN DELETE;

SELECT * FROM INTG_TBLS.EMPL;
SELECT * FROM MYSTREAMS.STREAM_EMPL;

//All changes at a time

SELECT * FROM STAGE_TBLS.STG_EMPL;

INSERT INTO STAGE_TBLS.STG_EMPL VALUES
(7, 'Janaki', 61000, 29, 'SALES', 'Pune'),
(8, 'Kamal', 92000, 33, 'TECHNOLOGY', 'Bangalore');

UPDATE STAGE_TBLS.STG_EMPL
SET SALARY=85000, LOCATION='Hyderabad' WHERE EMPID=1;

DELETE FROM STAGE_TBLS.STG_EMPL WHERE EMPID in (6);

SELECT * FROM STAGE_TBLS.STG_EMPL;

SELECT * FROM MYSTREAMS.STREAM_EMPL;

MERGE INTO INTG_TBLS.EMPL T
USING MYSTREAMS.STREAM_EMPL S
ON T.EMPID = S.EMPID
WHEN MATCHED
AND S.METADATA$ACTION ='DELETE'
AND S.METADATA$ISUPDATE = 'FALSE'
THEN DELETE
WHEN MATCHED
AND S.METADATA$ACTION ='INSERT'
AND S.METADATA$ISUPDATE = 'TRUE'
THEN UPDATE
SET T.EMPNAME = S.EMPNAME,
T.SALARY = S.SALARY,
T.AGE = S.AGE,
T.DEPT = S.DEPT,
T.LOCATION = S.LOCATION,
T.LST_UPDT_DT = CURRENT_DATE
WHEN NOT MATCHED
AND S.METADATA$ACTION ='INSERT'
AND S.METADATA$ISUPDATE = 'FALSE'
THEN INSERT( EMPID, EMPNAME, SALARY, AGE, DEPT, LOCATION, INSRT_DT, LST_UPDT_DT)
VALUES(S.EMPID, S.EMPNAME, S.SALARY, S.AGE, S.DEPT, S.LOCATION, CURRENT_DATE, NULL);
SELECT * FROM INTG_TBLS.EMPL;

SELECT * FROM MYSTREAMS.STREAM_EMPL;

//With Tasks

CREATE OR REPLACE TASK MYTASKS.TASK_EMPL_DATA_LOAD
WAREHOUSE = MYOWN_WH
SCHEDULE = '5 MINUTES'
WHEN SYSTEM$STREAM_HAS_DATA('MYSTREAMS.STREAM_EMPL')
AS
MERGE INTO INTG_TBLS.EMPL T
USING MYSTREAMS.STREAM_EMPL S
ON T.EMPID = S.EMPID
WHEN MATCHED
AND S.METADATA$ACTION ='DELETE'
AND S.METADATA$ISUPDATE = 'FALSE'
THEN DELETE
WHEN MATCHED
AND S.METADATA$ACTION ='INSERT'
AND S.METADATA$ISUPDATE = 'TRUE'
THEN UPDATE
SET T.EMPNAME = S.EMPNAME,
T.SALARY = S.SALARY,
T.AGE = S.AGE,
T.DEPT = S.DEPT,
T.LOCATION = S.LOCATION,
T.LST_UPDT_DT = CURRENT_DATE
WHEN NOT MATCHED
AND S.METADATA$ACTION ='INSERT'
AND S.METADATA$ISUPDATE = 'FALSE'
THEN INSERT( EMPID, EMPNAME, SALARY, AGE, DEPT, LOCATION, INSRT_DT, LST_UPDT_DT)
VALUES(S.EMPID, S.EMPNAME, S.SALARY, S.AGE, S.DEPT, S.LOCATION, CURRENT_DATE, NULL);

ALTER TASK MYTASKS.TASK_EMPL_DATA_LOAD RESUME;
SELECT * FROM STAGE_TBLS.STG_EMPL;

INSERT INTO STAGE_TBLS.STG_EMPL VALUES
(9, 'Latha', 47000, 25, 'HR', 'Chennai');

UPDATE STAGE_TBLS.STG_EMPL
SET SALARY=67000 WHERE EMPID=7;

DELETE FROM STAGE_TBLS.STG_EMPL WHERE EMPID in (8);

SELECT * FROM MYSTREAMS.STREAM_EMPL;

SELECT * FROM INTG_TBLS.EMPL;

SELECT * FROM MYSTREAMS.STREAM_EMPL;

свернуть