vvgsrk

{#
- This macro generate hash value to compare row rank and incremental data 
#}

{%- macro compare_row_rank_query_result_with_incremental_table(history_model, incremental_model) -%}
{%- set history_columns = adapter.get_columns_in_relation(history_model) -%}
{%- set inc_columns = adapter.get_columns_in_relation(incremental_model) -%}
WITH ranked_rows AS
(
SELECT

vvgsrk

CREATE TABLE product_history (
	product_id NUMBER,
	product_name VARCHAR,
	price_per_unit VARCHAR,
	basic_unit VARCHAR,
	is_stock_limited VARCHAR,
	is_active_for_sale VARCHAR,
	__event_ts TIMESTAMP_NTZ(9), 
	__load_ts TIMESTAMP_NTZ(9)
);

vvgsrk

INSERT INTO product_history (product_id, product_name, price_per_unit, basic_unit, is_stock_limited, is_active_for_sale, __event_ts, __load_ts)
VALUES(1009, 'Paris Coffee', '3 EUR', '1 piece', 'No', 'Yes', DATEADD(hour, -1, DATEADD(Day ,-1, CURRENT_TIMESTAMP())), CURRENT_TIMESTAMP());

vvgsrk

merge into dev_edw_dbt_incremental_model_test.dbt_src_product_management.src_product_management_product_latest_incremental as DBT_INTERNAL_DEST
        using dev_edw_dbt_incremental_model_test.dbt_src_product_management.src_product_management_product_latest_incremental__dbt_tmp as DBT_INTERNAL_SOURCE
        on 
            DBT_INTERNAL_SOURCE.__uuid = DBT_INTERNAL_DEST.__uuid       
    
    when matched then update set
        "PRODUCT_ID" = DBT_INTERNAL_SOURCE."PRODUCT_ID","PRODUCT_NAME" = DBT_INTERNAL_SOURCE."PRODUCT_NAME","PRICE_PER_UNIT" = DBT_INTERNAL_SOURCE."PRICE_PER_UNIT","BASIC_UNIT" = DBT_INTERNAL_SOURCE."BASIC_UNIT","IS_STOCK_LIMITED" = DBT_INTERNAL_SOURCE."IS_STOCK_LIMITED","IS_ACTIVE_FOR_SALE" = DBT_INTERNAL_SOURCE."IS_ACTIVE_FOR_SALE","__EVENT_TS" = DBT_INTERNAL_SOURCE."__EVENT_TS","__LOAD_TS" = DBT_INTERNAL_SOURCE."__LOAD_TS","__UUID" = DBT_INTERNAL_SOURCE."__UUID"

    when not matched then insert
        ("PRODUCT_ID", "PRODUCT_NAME", "PRICE_PER_UNIT", "BASIC_UNIT", "IS_STOCK_LIMITED", "IS_

vvgsrk

create or replace temporary table dev_edw_dbt_incremental_model_test.dbt_src_product_management.src_product_management_product_latest_incremental__dbt_tmp  as
      (
    WITH ranked_data AS (
        SELECT
            ROW_NUMBER() over (
                PARTITION BY product_id
                ORDER BY
                    __event_ts DESC
            ) AS row_rank,
            *

vvgsrk

create or replace transient table dev_edw_dbt_incremental_model_test.dbt_src_product_management.src_product_management_product_latest_incremental  as
      (
    WITH ranked_data AS (
        SELECT
            ROW_NUMBER() over (
                PARTITION BY product_id
                ORDER BY
                    __event_ts DESC
            ) AS row_rank,
            *

vvgsrk

SELECT *
  FROM {{ ref('src_product_management_product_latest_incremental') }}

vvgsrk

WITH ranked_data AS (
    SELECT
        ROW_NUMBER() over (
            PARTITION BY product_id
            ORDER BY
                __event_ts DESC
        ) AS row_rank,
        *
      FROM dev_edw_dbt_incremental_model_test.dbt_src_product_management.src_product_management_product_history
    )

vvgsrk

{{ generate_row_rank_using_event_ts_incremental(ref('src_product_management_product_history'), 'product_id') }}

vvgsrk

{{ generate_row_rank_using_event_ts(ref('src_product_management_product_history'), 'product_id') }}