vvgsrk

-- A separate new database
CREATE DATABASE refresh_secondary_database;

-- Create a new schema
CREATE SCHEMA refresh_secondary_database.amadeus_dynamic_pricing;

-- Task that refreshes the secondary database on a schedule
CREATE OR REPLACE TASK refresh_secondary_database.amadeus_dynamic_pricing.dynamic_pricing_data_refresh_task
  WAREHOUSE = WH01_XS
  SCHEDULE = 'USING CRON 0 3 * * * UTC'

vvgsrk

-- Create a replica of the primary database in the traget account
CREATE DATABASE AMADEUS_DYNAMIC_PRICING
  AS REPLICA OF ORGANIZATION_CODE.ORGANIZATION_NAME.AMADEUS_DYNAMIC_PRICING;
  
-- Refresh a secondary database
ALTER DATABASE AMADEUS_DYNAMIC_PRICING REFRESH;

vvgsrk

-- Modify an existing permanent database to serve as a primary database using following statement
ALTER DATABASE amadeus_dynamic_pricing ENABLE REPLICATION TO ACCOUNTS ORGANIZATION_CODE.ORGANIZATION_NAME IGNORE EDITION CHECK;

vvgsrk

-- A database, schema and table are required before you can load data

CREATE DATABASE amadeus_dynamic_pricing;

CREATE SCHEMA amadeus_dynamic_pricing.dynamic_pricing;

-- Create a table in newly created permanent database as same as the table from the share
CREATE TABLE amadeus_dynamic_pricing.dynamic_pricing.dp_logs
AS 
SELECT * FROM dynamic_pricing_data_share_by_amadeus.dynamic_pricing.dp_logs WHERE 1=2;

vvgsrk

select
      count(*) as failures,
      count(*) != 0 as should_warn,
      count(*) != 0 as should_error
    from (
    WITH snap AS
(
SELECT
    ROW_NUMBER() OVER (
        PARTITION BY "PRODUCT_ID" 

vvgsrk

version: 2

models:
  - name: src_product_management_product_history
    tests:
      - validate_row_rank_and_incremental_query_result:
          incremental_model: src_product_management_product_latest_incremental
          tags: ["nightly"]
      - validate_row_count_for_lattest_and_incremental:
          incremental_model: src_product_management_product_latest_incremental

vvgsrk

{% test validate_row_count_for_lattest_and_incremental(model, incremental_model) %}

    {{ compare_lattest_and_incremental_row_count(model, ref(incremental_model)) }}
    WHERE diff_count > 0

{% endtest %}

vvgsrk

{#
- This macro generate row counts for lattest data based on history data and incremental data 
- Compares both the row counts and outputs the diffrence
#}

{%- macro compare_lattest_and_incremental_row_count(history_model, incremental_model) -%}
{%- set columns = adapter.get_columns_in_relation(history_model) -%}
WITH snap AS
(
SELECT

vvgsrk

select
      count(*) as failures,
      count(*) != 0 as should_warn,
      count(*) != 0 as should_error
    from (   
    WITH ranked_rows AS
(
SELECT
    ROW_NUMBER() OVER (
        PARTITION BY "PRODUCT_ID" 

vvgsrk

{% test validate_row_rank_and_incremental_query_result(model, incremental_model) %}

    {{ compare_row_rank_query_result_with_incremental_table(model, ref(incremental_model)) }}
    WHERE result > 0

{% endtest %}