thorchain: update model logic

dbt_subprojects/daily_spellbook/models/thorchain/gold/thorchain_core_transfers.sql

@@ -0,0 +1,54 @@
+{{ config(
+    schema = 'thorchain',
+    alias = 'core_transfers',
+    materialized = 'incremental',
+    file_format = 'delta',
+    incremental_strategy = 'merge',
+    unique_key = ['day', 'fact_transfers_id'],
+    partition_by = ['day'],
+    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_timestamp')],
+    tags = ['thorchain', 'defi', 'transfers', 'fact'],
+    post_hook='{{ expose_spells(\'["thorchain"]\',
+                              "defi",
+                              "defi_transfers",
+                              \'["krishhh"]\') }}'
+) }}
+
+WITH base AS (
+    SELECT
+        block_id,
+        from_address,
+        to_address,
+        asset,
+        rune_amount,
+        rune_amount_usd,
+        _unique_key,
+        _INSERTED_TIMESTAMP
+    FROM
+    {{ ref('thorchain_silver_transfers') }}
+    {% if is_incremental() %}
+    WHERE {{ incremental_predicate('block_timestamp') }}
+    {% endif %}
+)
+SELECT
+    {{ dbt_utils.generate_surrogate_key(
+        ['a._unique_key']
+    ) }} AS fact_transfers_id,
+    cast(date_trunc('day', b.block_timestamp) AS date) AS day,
+    b.block_timestamp,
+    COALESCE(
+        b.dim_block_id,
+        '-1'
+    ) AS dim_block_id,
+    from_address,
+    to_address,
+    asset,
+    rune_amount,
+    rune_amount_usd,
+    A._inserted_timestamp,
+    current_timestamp AS inserted_timestamp,
+    current_timestamp AS modified_timestamp
+FROM
+    base A
+JOIN {{ ref('thorchain_core_block') }} as b
+    ON A.block_id = b.block_id

dbt_subprojects/daily_spellbook/models/thorchain/gold/thorchain_defi_add_events.sql

@@ -4,9 +4,9 @@
     materialized = 'incremental',
     file_format = 'delta',
     incremental_strategy = 'merge',
-    unique_key = ['block_month', 'tx_id', 'event_id'],
-    partition_by = ['block_month'],
-    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')],
+    unique_key = ['day', 'fact_add_events_id'],
+    partition_by = ['day'],
+    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_timestamp')],
     tags = ['thorchain', 'defi', 'add_events', 'fact'],
     post_hook='{{ expose_spells(\'["thorchain"]\',
                               "defi",
@@ -16,47 +16,56 @@
 
 WITH base AS (
   SELECT
-    raw_block_timestamp as block_timestamp,
-    tx_hash as tx_id,
-    rune_e8,
-    chain as blockchain,
-    asset_e8,
-    pool as pool_name,
-    memo,
-    to_addr as to_address,
-    from_addr as from_address,
-    asset,
-    event_id,
-    tx_type as _tx_type,
-    block_time  -- Keep for incremental predicate
+    e.block_timestamp,
+    e.tx_id,
+    e.rune_e8,
+    e.blockchain,
+    e.asset_e8,
+    e.pool_name,
+    e.memo,
+    e.to_address,
+    e.from_address,
+    e.asset,
+    e.event_id,
+    e._TX_TYPE
   FROM
-    {{ ref('thorchain_silver_add_events') }}
+    {{ ref('thorchain_silver_add_events') }} e
 )
-
 SELECT
-  a.block_time,
-  date(a.block_time) as block_date,
-  date_trunc('month', a.block_time) as block_month,
-  -1 AS block_height,
-  a.tx_id,
-  a.event_id,
-  a.rune_e8,
-  a.blockchain,
-  a.asset_e8,
-  a.pool_name,
-  a.memo,
-  a.to_address,
-  a.from_address,
-  a.asset,
-  a._tx_type,
+  {{ dbt_utils.generate_surrogate_key(
+    ['a.event_id','a.tx_id','a.blockchain','a.from_address','a.to_address','a.asset','a.memo','a.block_timestamp']
+  ) }} AS fact_add_events_id,
+  cast(date_trunc('day', b.block_timestamp) AS date) AS day,
+  b.block_timestamp,
+  COALESCE(
+    b.dim_block_id,
+    '-1'
+  ) AS dim_block_id,
+  A.tx_id,
+  A.rune_e8,
+  A.blockchain,
+  A.asset_e8,
+  A.pool_name,
+  A.memo,
+  A.to_address,
+  A.from_address,
+  A.asset,
+  A._TX_TYPE,
   current_timestamp AS inserted_timestamp,
   current_timestamp AS modified_timestamp
-
-FROM base a
-
+FROM
+  base A
+JOIN {{ ref('thorchain_core_block') }} as b
+  ON A.block_timestamp = b.timestamp
 {% if is_incremental() %}
-WHERE a.block_time >= (
-  SELECT MAX(block_time - INTERVAL '1' HOUR)
-  FROM {{ this }}
-) 
-{% endif %}
+WHERE
+  {{ incremental_predicate('b.block_timestamp') }}
+  OR tx_id IN (
+    SELECT
+      tx_id
+    FROM
+      {{ this }}
+    WHERE
+      dim_block_id = '-1'
+  )
+{% endif %}

dbt_subprojects/daily_spellbook/models/thorchain/gold/thorchain_defi_block_pool_depths.sql

@@ -4,9 +4,8 @@
     materialized = 'incremental',
     file_format = 'delta',
     incremental_strategy = 'merge',
-    unique_key = ['block_month', 'fact_pool_depths_id'],
-    partition_by = ['block_month'],
-    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')],
+    unique_key = ['fact_pool_depths_id'],
+    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_timestamp')],
     tags = ['thorchain', 'defi', 'pool_depths', 'fact'],
     post_hook='{{ expose_spells(\'["thorchain"]\',
                               "defi",
@@ -20,43 +19,37 @@ WITH base AS (
         asset_e8,
         rune_e8,
         synth_e8,
-        block_time,
-        raw_block_timestamp,
+        block_timestamp,
         _inserted_timestamp
-    FROM {{ ref('thorchain_silver_block_pool_depths') }}
+    FROM
+        {{ ref('thorchain_silver_block_pool_depths') }}
 )
-
 SELECT
-    -- CRITICAL: Generate surrogate key (Trino equivalent of dbt_utils.generate_surrogate_key)
-    to_hex(sha256(to_utf8(concat(
-        COALESCE(a.pool_name, ''),
-        '|',
-        COALESCE(cast(a.block_time as varchar), '')
-    )))) AS fact_pool_depths_id,
-
-    -- CRITICAL: Always include partitioning columns first
-    a.block_time,
-    date(a.block_time) as block_date,
-    date_trunc('month', a.block_time) as block_month,
-    a.raw_block_timestamp,
-
-    -- Block dimension reference (set directly - no JOIN needed)
-    '-1' AS dim_block_id,
-
-    -- Pool depth data
-    a.rune_e8,
-    a.asset_e8,
-    a.synth_e8,
-    a.pool_name,
-
-    -- Audit fields (Trino conversions)
+    {{ dbt_utils.generate_surrogate_key(['a.pool_name','a.block_timestamp']) }} AS fact_pool_depths_id,
+    b.block_timestamp,
+    COALESCE(
+        b.dim_block_id,
+        '-1'
+    ) AS dim_block_id,
+    rune_e8,
+    asset_e8,
+    synth_e8,
+    pool_name,
     a._inserted_timestamp,
-    cast(from_hex(replace(cast(uuid() as varchar), '-', '')) as varchar) AS _audit_run_id,  -- Trino equivalent of invocation_id
-    current_timestamp AS inserted_timestamp,  -- Trino equivalent of SYSDATE()
+    current_timestamp AS inserted_timestamp,
     current_timestamp AS modified_timestamp
-
-FROM base a
-
+FROM
+    base as a
+JOIN {{ ref('thorchain_core_block') }} as b
+    ON a.block_timestamp = b.timestamp
 {% if is_incremental() %}
-WHERE {{ incremental_predicate('a.block_time') }}
-{% endif %}
+WHERE {{ incremental_predicate('b.block_timestamp') }}
+OR pool_name IN (
+    SELECT
+        pool_name
+    FROM
+        {{ this }}
+    WHERE
+        dim_block_id = '-1'
+)
+{% endif %}

dbt_subprojects/daily_spellbook/models/thorchain/gold/thorchain_defi_block_rewards.sql

@@ -4,9 +4,9 @@
     materialized = 'incremental',
     file_format = 'delta',
     incremental_strategy = 'merge',
-    unique_key = ['block_month', 'fact_block_rewards_id'],
-    partition_by = ['block_month'],
-    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_date')],
+    unique_key = ['day', 'fact_block_rewards_id'],
+    partition_by = ['day'],
+    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.day')],
     tags = ['thorchain', 'defi', 'block_rewards', 'fact'],
     post_hook='{{ expose_spells(\'["thorchain"]\',
                               "defi",
@@ -16,42 +16,31 @@
 
 WITH base AS (
     SELECT
-        block_date,
+        day,
         liquidity_fee,
         block_rewards,
         earnings,
         bonding_earnings,
         liquidity_earnings,
         avg_node_count,
-        _inserted_timestamp,
-        block_month
-    FROM {{ ref('thorchain_silver_block_rewards') }}
+        _inserted_timestamp
+    FROM
+        {{ ref('thorchain_silver_block_rewards') }}
+    {% if is_incremental() -%}
+    WHERE {{ incremental_predicate('day') }}
+    {% endif -%}
 )
-
 SELECT
-    -- CRITICAL: Generate surrogate key (Trino equivalent of dbt_utils.generate_surrogate_key)
-    to_hex(sha256(to_utf8(cast(a.block_date as varchar)))) AS fact_block_rewards_id,
-
-    -- CRITICAL: Always include partitioning columns first
-    a.block_date,
-    a.block_month,
-
-    -- Block rewards data
-    a.liquidity_fee,
-    a.block_rewards,
-    a.earnings,
-    a.bonding_earnings,
-    a.liquidity_earnings,
-    a.avg_node_count,
-
-    -- Audit fields (Trino conversions)
+    {{ dbt_utils.generate_surrogate_key(['a.day']) }} AS fact_block_rewards_id,
+    day,
+    liquidity_fee,
+    block_rewards,
+    earnings,
+    bonding_earnings,
+    liquidity_earnings,
+    avg_node_count,
     a._inserted_timestamp,
-    cast(from_hex(replace(cast(uuid() as varchar), '-', '')) as varchar) AS _audit_run_id,  -- Trino equivalent of invocation_id
-    current_timestamp AS inserted_timestamp,  -- Trino equivalent of SYSDATE()
+    current_timestamp AS inserted_timestamp,
     current_timestamp AS modified_timestamp
-
-FROM base a
-
-{% if is_incremental() %}
-WHERE {{ incremental_predicate('a.block_date') }}
-{% endif %}
+FROM
+  base as a