DROP TABLE IF EXISTS ADJUSTMENT;
DROP TABLE IF EXISTS staging_ADJUSTMENT;
----- CREATE TEMPORARY STAGING TABLE TO LOAD CSV FORMATTED DATA -----
CREATE EXTERNAL TABLE staging_ADJUSTMENT
    USING com.databricks.spark.csv OPTIONS (path ':dataLocation/ADJUSTMENT.dat', header 'true', inferSchema 'true', nullValue 'NULL', maxCharsPerColumn '4096');
CREATE TABLE ADJUSTMENT USING column OPTIONS(partition_by 'BILL_ENT_ID',buckets '32',key_columns 'CLIENT_ID,BILL_ENT_ID,ADJ_ID ' ) AS (SELECT * FROM staging_ADJUSTMENT);

 DROP TABLE IF EXISTS AGREEMENT;
 DROP TABLE IF EXISTS staging_AGREEMENT;
   ----- CREATE TEMPORARY STAGING TABLE TO LOAD CSV FORMATTED DATA -----
 CREATE EXTERNAL TABLE staging_AGREEMENT
     USING com.databricks.spark.csv OPTIONS (path ':dataLocation/AGREEMENT.dat', header 'true', inferSchema 'true', nullValue 'NULL', maxCharsPerColumn '4096');
 CREATE TABLE AGREEMENT USING column OPTIONS(partition_by 'AGREE_ID', buckets '32',key_columns 'CLIENT_ID,AGREE_ID ' ) AS (SELECT * FROM staging_AGREEMENT);

 DROP TABLE IF EXISTS BANK;
 DROP TABLE IF EXISTS staging_BANK;
 ----- CREATE TEMPORARY STAGING TABLE TO LOAD CSV FORMATTED DATA -----
 CREATE EXTERNAL TABLE staging_BANK
     USING com.databricks.spark.csv OPTIONS (path ':dataLocation/BANK.dat', header 'true', inferSchema 'true', nullValue 'NULL', maxCharsPerColumn '4096');
 CREATE TABLE BANK USING column OPTIONS(partition_by 'BNK_ORG_ID', buckets '32',key_columns 'CLIENT_ID,BNK_ORG_ID,BNK_ID ' ) AS (SELECT * FROM staging_BANK);

 DROP TABLE IF EXISTS BENEFIT_PACKAGE;
 DROP TABLE IF EXISTS staging_BENEFIT_PACKAGE;
 CREATE EXTERNAL TABLE IF NOT EXISTS  staging_BENEFIT_PACKAGE
     USING com.databricks.spark.csv OPTIONS (path ':dataLocation/BENEFIT_PACKAGE.dat', header 'true', inferSchema 'true', nullValue 'NULL', maxCharsPerColumn '4096');
 CREATE TABLE IF NOT EXISTS BENEFIT_PACKAGE USING column OPTIONS(partition_by 'BENE_PKG_ID',redundancy '1', buckets '32',key_columns 'CLIENT_ID,BENE_PKG_ID' ) AS (SELECT * FROM staging_BENEFIT_PACKAGE);

 DROP TABLE IF EXISTS BANK_ACCOUNT;
 DROP TABLE IF EXISTS staging_BANK_ACCOUNT;
 ----- CREATE TEMPORARY STAGING TABLE TO LOAD CSV FORMATTED DATA -----
 CREATE EXTERNAL TABLE IF NOT EXISTS  staging_BANK_ACCOUNT
     USING com.databricks.spark.csv OPTIONS (path ':dataLocation/BANK_ACCOUNT.dat', header 'true', inferSchema 'true', nullValue 'NULL', maxCharsPerColumn '4096');
  CREATE TABLE IF NOT EXISTS BANK_ACCOUNT USING column OPTIONS(partition_by 'BNK_ORG_ID',redundancy '1', buckets '32',key_columns 'CLIENT_ID,BNK_ORG_ID,BNK_ID,BNK_ACCT_ID' ) AS (SELECT * FROM staging_BANK_ACCOUNT);

 DROP TABLE IF EXISTS BENEFIT_GROUP_NAME;
 DROP TABLE IF EXISTS staging_BENEFIT_GROUP_NAME;
 ----- CREATE TEMPORARY STAGING TABLE TO LOAD CSV FORMATTED DATA -----
 CREATE EXTERNAL TABLE IF NOT EXISTS  staging_BENEFIT_GROUP_NAME
     USING com.databricks.spark.csv OPTIONS (path ':dataLocation/BENEFIT_GROUP_NAME.dat', header 'true', inferSchema 'true', nullValue 'NULL', maxCharsPerColumn '4096');
 CREATE TABLE IF NOT EXISTS BENEFIT_GROUP_NAME(
   GRP_ID BIGINT NOT NULL,
   BENE_GRP_ID BIGINT NOT NULL,
   BENE_GRP_NM_ID BIGINT   NOT NULL,
   VER BIGINT NOT NULL,
   CLIENT_ID BIGINT NOT NULL,
   BENE_GRP_NM VARCHAR(8),
   DESCR VARCHAR(15),
   EFF_DT DATE,
   EXPR_DT DATE,
   VLD_FRM_DT date NOT NULL,
   VLD_TO_DT date,
   SRC_SYS_REF_ID VARCHAR(10) NOT NULL,
   SRC_SYS_REC_ID VARCHAR(15),
   PRIMARY KEY (CLIENT_ID,GRP_ID,BENE_GRP_ID,BENE_GRP_NM_ID)
   )
  USING row OPTIONS(partition_by 'GRP_ID', buckets '32',redundancy '1');
   INSERT INTO BENEFIT_GROUP_NAME SELECT * FROM staging_BENEFIT_GROUP_NAME;

 DROP TABLE IF EXISTS BENEFIT_GROUPS;
 DROP TABLE IF EXISTS staging_BENEFIT_GROUPS;
 ----- CREATE TEMPORARY STAGING TABLE TO LOAD CSV FORMATTED DATA -----
 CREATE EXTERNAL TABLE IF NOT EXISTS  staging_BENEFIT_GROUPS
     USING com.databricks.spark.csv OPTIONS (path ':dataLocation/BENEFIT_GROUPS.dat', header 'true', inferSchema 'true', nullValue 'NULL', maxCharsPerColumn '4096');
 CREATE TABLE IF NOT EXISTS BENEFIT_GROUPS(
   GRP_ID BIGINT NOT NULL,
   BENE_PKG_ID BIGINT NOT NULL,
   BENE_GRP_ID BIGINT   NOT NULL,
   VER BIGINT NOT NULL,
   CLIENT_ID BIGINT NOT NULL,
   BENE_GRP_CD VARCHAR(5),
   EFF_DT DATE,
   EXPR_DT DATE,
   VLD_FRM_DT date NOT NULL,
   VLD_TO_DT date,
   SRC_SYS_REF_ID VARCHAR(10) NOT NULL,
   SRC_SYS_REC_ID VARCHAR(15),
   PRIMARY KEY (CLIENT_ID,GRP_ID,BENE_PKG_ID,BENE_GRP_ID)
   )
  USING row OPTIONS(partition_by 'GRP_ID', buckets '32',redundancy '1');
  INSERT INTO BENEFIT_GROUPS SELECT * FROM staging_BENEFIT_GROUPS;