Creating Final State Tables

-- Replace the placeholders in the query
-- DBNAME, SCHEMANAME, WAREHOUSE_NAME
-------------------------------------------------------
-- Create a `TABLE_REFRESH_HISTORY` table to store the timestamp of the last inserted timestamp. This will ensure that only the latest data is pulled during subsequent runs. Additionally, set up a task to run after the main `<TABLENAME>_TASK` task, which will update the timestamp in the `<TASK_TABLENAME>_TASK_RUN_HISTORY` table.

CREATE OR REPLACE TASK <DBNAME>.<SCHEMANAME>.<TASK_TABLENAME>_TASK_RUN_HISTORY
  WAREHOUSE = WAREHOUSE_NAME
  AFTER <DBNAME>.<SCHEMANAME>.<TABLENAME>_TASK
AS
INSERT INTO <DBNAME>.<SCHEMANAME>.TABLE_REFRESH_HISTORY (TABLE_NAME, MAX_TIMESTAMP, REFRESH_TIME)
SELECT 
    '<DBNAME>.<SCHEMANAME>.<TABLENAME_DEDUP>', 
    MAX(_streamkap_ts_ms) AS max_ts,
    CURRENT_TIMESTAMP AS refresh_time
FROM <DBNAME>.<SCHEMANAME>.<TABLENAME_DEDUP>;

---------------------------------------------------------
-- `MERGE` statement will merge data into the target table. Additionally, a configurable task will be created to run every 15 minutes.

CREATE TASK <DBNAME>.<SCHEMANAME>.<TABLENAME>_TASK
  WAREHOUSE = <WAREHOUSE_NAME>  
  SCHEDULE = '15 MINUTE'
AS
MERGE INTO <DBNAME>.<SCHEMANAME>.<TABLENAME_DEDUP> AS target
USING (
    SELECT *
    FROM (
        SELECT *,
               ROW_NUMBER() OVER (PARTITION BY ID ORDER BY _streamkap_ts_ms DESC, _streamkap_offset DESC) AS dedupe_id
        FROM <DBNAME>.<SCHEMANAME>.<TABLENAME>
        WHERE _streamkap_ts_ms >= 
        COALESCE((SELECT MAX(MAX_TIMESTAMP) FROM <DBNAME>.<SCHEMANAME>.TABLE_REFRESH_HISTORY
        WHERE TABLE_NAME='<DBNAME>.<SCHEMANAME>.<TABLENAME_DEDUP>'),
        (DATE_PART('EPOCH', CURRENT_TIMESTAMP()) * 1000) - (8 * 24 * 60 * 60 * 1000))
    ) AS subquery
    WHERE dedupe_id = 1 AND __deleted = 'false'
) AS source
ON target.ID = source.ID
WHEN MATCHED THEN
    UPDATE SET 
        target.RECORD_METADATA = source.RECORD_METADATA,
        <values>,
        target.__DELETED = source.__DELETED,
        target._STREAMKAP_SOURCE_TS_MS = source._STREAMKAP_SOURCE_TS_MS,
        target._STREAMKAP_TS_MS = source._STREAMKAP_TS_MS,
        target._STREAMKAP_OFFSET = source._STREAMKAP_OFFSET
WHEN NOT MATCHED THEN
    INSERT (
  			<values>,
        __DELETED,
        _STREAMKAP_SOURCE_TS_MS,
        _STREAMKAP_TS_MS,
        _STREAMKAP_OFFSET
    )
    VALUES (
        <values>,
        source.__DELETED,
        source._STREAMKAP_SOURCE_TS_MS,
        source._STREAMKAP_TS_MS,
        source._STREAMKAP_OFFSET
    );

-- Replace the placeholders in the query
-- DYNAMIC_TABLE_NAME, TARGET_LAG, WAREHOUSE_NAME, PRIMARY_KEY, SOURCE_TABLE_NAME

CREATE OR REPLACE DYNAMIC TABLE <DYNAMIC_TABLE_NAME> (
    {{column_1}},
    {{column_2}},
    {{column_3}},
    {{column_4}},
    {{column_5}},
    .
    .
    .
    .
    __DELETED,
    _STREAMKAP_SOURCE_TS_MS,
    _STREAMKAP_TS_MS,
    _STREAMKAP_OFFSET
)
TARGET_LAG = '<TARGET_LAG>'  -- Example: '15 minutes'
REFRESH_MODE = AUTO 		 
INITIALIZE =  ON_CREATE
WAREHOUSE = <WAREHOUSE_NAME>
AS
SELECT * EXCLUDE dedupe_id
FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY <PRIMARY_KEY> ORDER BY _streamkap_ts_ms DESC, _streamkap_offset DESC) AS dedupe_id 
    FROM <SOURCE_TABLE_NAME>
)
WHERE dedupe_id = 1 AND __deleted = 'false';

{{
	config(
		materialized='incremental',
		unique_key=<your_primary_key>,
		post_hook ='DELETE FROM {{ this }} WHERE __deleted = TRUE',
	)
}}


WITH cte AS (
    SELECT * FROM source_table
    {% if is_incremental() %}
    WHERE
			_streamkap_ts_ms >= (SELECT MAX(_streamkap_ts_ms) FROM {{ this }})
	{% endif %}
    QUALIFY RANK() OVER (PARTITION BY <your_primary_key> ORDER BY _streamkap_ts_ms DESC, _streamkap_offset DESC) = 1
)

SELECT
    *
FROM
    cte

{% if is_incremental() %}

	WHERE
		_streamkap_ts_ms >= (SELECT MAX(this._streamkap_ts_ms) FROM {{ this }} AS this)

{% endif %}

import os
import yaml


class IncrementalScriptGenerator:
    def __init__(self, relative_input_path, relative_output_dir):
        self.home_directory = os.path.expanduser("~")
        self.relative_input_path = relative_input_path
        self.relative_output_dir = relative_output_dir
        self.prod_tables = []

    def load_yaml_file(self):
        full_path = os.path.join(self.home_directory, self.relative_input_path)
        with open(full_path, "r") as file:
            self.data = yaml.safe_load(file)

    def extract_prod_tables(self):
        for source in self.data.get("sources", []):
            if source.get("name") == "prod":
                tables = source.get("tables", [])
                for table in tables:
                    table_name = table.get("name")
                    if table_name:
                        self.prod_tables.append(table_name)

    def generate_incremental_scripts(self):
        for prod_table in self.prod_tables:
            incremental_script = """
            {{
                config(
                    incremental_strategy='merge',
                    unique_key='id'
                )
            }}
            """

            incremental_script += f"""
            select * 
            from {{{{ source('prod', '{prod_table}') }}}}
            """

            incremental_script += """
            {% if is_incremental() %}
            where
                _streamkap_ts_ms >= (select max(_streamkap_ts_ms) from {{ this }})
            {% endif %}

            qualify rank() over (partition by id order by _streamkap_ts_ms desc, _streamkap_offset desc) = 1
            """

            relative_output_path = (
                f"{self.relative_output_dir}/sk_{prod_table.lower()}.sql"
            )
            final_output_path = os.path.join(self.home_directory, relative_output_path)

            with open(final_output_path, "w") as file:
                file.write(incremental_script)

            print(f"Written to {final_output_path}")


if __name__ == "__main__":
    relative_input_path = "dae-dbt/models/source.yml"
    relative_output_dir = "dae-dbt/models/streamkap/incremental"
    processor = IncrementalScriptGenerator(relative_input_path, relative_output_dir)
    processor.load_yaml_file()
    processor.extract_prod_tables()
    processor.generate_incremental_scripts()