tokens.transfers_from_traces - fix nulls in value (#9024)

* fix & update & easy dates run

* retrigger

* revert easy dates

* add metadata to final blockchain level macro

* temp: use easy dates

* add transfers start date param for efficient filtering

* include new columns in final view

* update all models to incremental tables

* temp: update base easy dates to true

* cosmetic: lowercase sql

* cleanup new chain mezo to match other base models

* add wrappers, add easy date usage, column aliases

* cosmetic

* Revert "cosmetic"

This reverts commit 688777f.

* cosmetic & visuality

* revert easy dates

---------

Co-authored-by: jeff-dude <[email protected]>

Author: max-morrow

dbt_subprojects/tokens/macros/transfers_from_traces/transfers_from_traces_base_macro.sql

@@ -1,15 +1,21 @@
-{%- macro transfers_from_traces_base_macro(blockchain, easy_dates=false) -%}
+{%- macro
+    transfers_from_traces_base_macro(
+        blockchain,
+        easy_dates=false
+    )
+-%}
 
 -- this stream process all kind of transfers from traces: native, erc20 transfer/transferFrom, mint/burn, wrapped deposit/withdrawal
 
-{%- set null_address = '0x0000000000000000000000000000000000000000' -%}
-{%- set transfer_selector = '0xa9059cbb' -%}
-{%- set transferFrom_selector = '0x23b872dd' -%}
-{%- set mint_selector = '0x40c10f19' -%}{# for DAI, etc. #}
-{%- set burn_selector = '0x9dc29fac' -%}{# for DAI, etc. #}
-{%- set deposit_selector = '0xd0e30db0' -%}{# for wrappers #}
-{%- set withdraw_selector = '0x2e1a7d4d' -%}{# for wrappers #}
-{%- set selector = 'substr(input, 1, 4)' %}
+{%- set null_address = "0x0000000000000000000000000000000000000000" -%}
+{%- set transfer_selector       = "0xa9059cbb" -%}
+{%- set transferFrom_selector   = "0x23b872dd" -%}
+{%- set mint_selector           = "0x40c10f19" -%}{# for DAI, etc. #}
+{%- set burn_selector           = "0x9dc29fac" -%}{# for DAI, etc. #}
+{%- set deposit_selector        = "0xd0e30db0" -%}{# for wrappers #}
+{%- set withdraw_selector       = "0x2e1a7d4d" -%}{# for wrappers #}
+{%- set selector = "substr(input, 1, 4)" -%}
+{%- set value = "coalesce(value, uint256 '0')" %}
 
 -- output --
 
@@ -29,14 +35,14 @@ select
         when {{ selector }} = {{ deposit_selector }} then 'deposit'
         when {{ selector }} = {{ withdraw_selector }} then 'withdraw'
         else 'native'
-    end as type 
-    , if(value > uint256 '0', 'native', 'erc20') as token_standard
-    , if(value > uint256 '0', native_address, "to") as contract_address 
+    end as type
+    , if({{ value }} > uint256 '0', 'native', 'erc20') as token_standard
+    , if({{ value }} > uint256 '0', native_address, "to") as contract_address 
     , case
         when {{ selector }} in ({{ transfer_selector }}, {{ mint_selector }}, {{ burn_selector }}) then bytearray_to_uint256(substr(input, 37, 32)) -- transfer, mint, burn
         when {{ selector }} = {{ transferFrom_selector }} then bytearray_to_uint256(substr(input, 69, 32)) -- transferFrom
         when {{ selector }} = {{ withdraw_selector }} then bytearray_to_uint256(substr(input, 5, 32)) -- withdraw
-        else value -- native, deposit
+        else {{ value }} -- native, deposit
     end as amount_raw
     , case
         when {{ selector }} in ({{ transferFrom_selector }}, {{ burn_selector }}) then substr(input, 17, 20) -- transferFrom, burn
@@ -56,21 +62,21 @@ select
         , cast(block_number as varchar)
         , cast(tx_hash as varchar)
         , array_join(trace_address, ',') -- ',' is necessary to avoid similarities after concatenation // array_join(array[1, 0], '') = array_join(array[10], '')
-        , cast(if(value > uint256 '0', native_address, "to") as varchar)
+        , cast(if({{ value }} > uint256 '0', native_address, "to") as varchar)
     ))) as unique_key
 from {{ source(blockchain, 'traces') }}, (select token_address as native_address from {{ source('dune', 'blockchains') }} where name = '{{ blockchain }}') as meta
 where
     (
         length(input) >= 68 and {{ selector }} in ({{ transfer_selector }}, {{ mint_selector }}, {{ burn_selector }}) -- transfer, mint, burn
         or length(input) >= 100 and {{ selector }} = {{ transferFrom_selector }} -- transferFrom
         or length(input) >= 36 and {{ selector }} = {{ withdraw_selector }} -- withdraw
-        or value > uint256 '0' -- native, deposit
+        or {{ value }} > uint256 '0' -- native, deposit
     )
     and (call_type = 'call' or type = 'create') -- call_type should be only call if present; type = 'create' is contract creation
     and (tx_success or tx_success is null) -- tx_success is null - is for old ethereum data
     and success
-    {% if easy_dates %}and block_date > current_date - interval '10' day{% endif %} -- easy_dates mode for dev, to prevent full scan
-    {% if is_incremental() %}and {{ incremental_predicate('block_time') }}{% endif %}
+    {% if easy_dates -%} and block_date > current_date - interval '10' day {%- endif %} -- easy_dates mode for dev, to prevent full scan
+    {% if is_incremental() -%} and {{ incremental_predicate('block_time') }} {%- endif %}
 
 
 

dbt_subprojects/tokens/macros/transfers_from_traces/transfers_from_traces_base_wrapper_deposits_macro.sql

@@ -1,4 +1,9 @@
-{%- macro transfers_from_traces_base_wrapper_deposits_macro(blockchain, transfers_from_traces_base_table) -%}
+{%- macro
+    transfers_from_traces_base_wrapper_deposits_macro(
+        blockchain,
+        transfers_from_traces_base_table
+    )
+-%}
 
 -- the wrapper deposit includes two transfers: native and wrapped, so we should add second one manually reversing from/to
 -- it's splitted to 2 operations and fetching from pre-materialized table to prevent doubling full-scan of traces
@@ -17,8 +22,8 @@ select
     , 'erc20' as token_standard
     , "to" as contract_address
     , amount_raw
-    , "to" as transfer_from
-    , "from" as transfer_to
+    , "to"
+    , "from"
     , sha1(to_utf8(concat_ws('|'
         , blockchain
         , cast(block_number as varchar)
@@ -34,7 +39,7 @@ join ( -- to leave only real tokens (mostly for wrapped token, but works for rar
 ) using("to")
 where
     type = 'deposit'
-    {% if is_incremental() %}and {{ incremental_predicate('block_time') }}{% endif %}
+    {% if is_incremental() -%} and {{ incremental_predicate('block_time') }} {%- endif %}
 
 
 

dbt_subprojects/tokens/macros/transfers_from_traces/transfers_from_traces_macro.sql

@@ -1,12 +1,123 @@
-{%- macro transfers_from_traces_macro(blockchain) -%}
--- it's splitted to 2 operations and fetching from pre-materialized table to prevent doubling full-scan of traces 
+{%- macro
+    transfers_from_traces_macro(
+        blockchain,
+        transfers_start_date='2000-01-01',
+        easy_dates=false,
+        prices_interval='hour'
+    )
+-%}
 
+{%- if blockchain is none or blockchain == '' -%}
+    {{- exceptions.raise_compiler_error("blockchain parameter cannot be null or empty") -}}
+{%- endif -%}
 
-select * from {{ ref('tokens_' ~ blockchain ~ '_transfers_from_traces_base') }}
 
-union all
--- the wrapper deposit includes two transfers: native and wrapped, so need to add second one manually reversing from/to
-select * from {{ ref('tokens_' ~ blockchain ~ '_transfers_from_traces_base_wrapper_deposits') }}
+
+with
+
+base_tft as (
+    select
+        blockchain
+        , block_month
+        , block_date
+        , block_time
+        , block_number
+        , tx_hash
+        , trace_address
+        , type
+        , token_standard
+        , contract_address
+        , amount_raw
+        , "from"
+        , "to"
+        , unique_key
+        , date_trunc('{{ prices_interval }}', block_time) as timestamp
+    from {{ ref('tokens_' ~ blockchain ~ '_transfers_from_traces_base') }}
+    where true
+        and block_date >= timestamp '{{ transfers_start_date }}'
+        {% if easy_dates -%} and block_date > current_date - interval '10' day {%- endif %} -- easy_dates mode for dev, to prevent full scan
+        {% if is_incremental() -%} and {{ incremental_predicate('block_date') }} {%- endif %}
+)
+
+, base_tft_wrapper_deposits as (
+    select
+        blockchain
+        , block_month
+        , block_date
+        , block_time
+        , block_number
+        , tx_hash
+        , trace_address
+        , type
+        , token_standard
+        , contract_address
+        , amount_raw
+        , "from"
+        , "to"
+        , unique_key
+        , date_trunc('{{ prices_interval }}', block_time) as timestamp
+    from {{ ref('tokens_' ~ blockchain ~ '_transfers_from_traces_base_wrapper_deposits') }}
+    where true
+        and block_date >= timestamp '{{ transfers_start_date }}'
+        {% if easy_dates -%} and block_date > current_date - interval '10' day {%- endif %} -- easy_dates mode for dev, to prevent full scan
+        {% if is_incremental() -%} and {{ incremental_predicate('block_date') }} {%- endif %}
+)
+
+, tft as (
+    select * from base_tft
+    union all
+    select * from base_tft_wrapper_deposits
+)
+
+, tokens as (
+    select
+        contract_address
+        , decimals as token_decimals
+        , symbol as token_symbol
+    from {{ source('tokens', 'erc20') }}
+    where true
+        and blockchain = '{{ blockchain }}'
+)
+
+, prices as (
+    select
+        contract_address
+        , timestamp
+        , decimals
+        , symbol
+        , price
+    from {{ source('prices_external', prices_interval) }}
+    where true
+        and blockchain = '{{ blockchain }}'
+        and timestamp >= timestamp '{{ transfers_start_date }}'
+        {% if easy_dates -%} and timestamp > current_date - interval '10' day {%- endif %} -- easy_dates mode for dev, to prevent full scan
+        {% if is_incremental() -%} and {{ incremental_predicate('timestamp') }} {%- endif %}
+)
+
+-- output --
+
+select
+    blockchain
+    , block_month
+    , block_date
+    , block_time
+    , block_number
+    , tx_hash
+    , trace_address
+    , type
+    , token_standard
+    , contract_address
+    , coalesce(token_symbol, symbol) as symbol
+    , amount_raw
+    , amount_raw / power(10, coalesce(token_decimals, decimals)) as amount
+    , price as price_usd
+    , amount_raw / power(10, coalesce(token_decimals, decimals)) * price as amount_usd
+    , "from"
+    , "to"
+    , unique_key
+from tft
+left join tokens using(contract_address)
+left join prices using(contract_address, timestamp)
 
 
 

dbt_subprojects/tokens/models/transfers_and_balances/abstract/tokens_abstract_transfers_from_traces.sql

@@ -4,8 +4,13 @@
     config(
         schema = 'tokens_' ~ blockchain,
         alias = 'transfers_from_traces',
-        materialized = 'view',
+        partition_by = ['block_month'],
+        materialized = 'incremental',
+        file_format = 'delta',
+        incremental_strategy = 'merge',
+        incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')],
+        unique_key = ['block_date', 'unique_key'],
     )
 }}
 
-{{ transfers_from_traces_macro(blockchain=blockchain) }}
+{{ transfers_from_traces_macro(blockchain=blockchain, transfers_start_date = '2024-10-30') }}

dbt_subprojects/tokens/models/transfers_and_balances/abstract/tokens_abstract_transfers_from_traces_base.sql

@@ -13,4 +13,4 @@
     )
 }}
 
-{{ transfers_from_traces_base_macro(blockchain=blockchain, easy_dates=false) }}
+{{ transfers_from_traces_base_macro(blockchain=blockchain) }}

dbt_subprojects/tokens/models/transfers_and_balances/apechain/tokens_apechain_transfers_from_traces.sql

@@ -4,8 +4,13 @@
     config(
         schema = 'tokens_' ~ blockchain,
         alias = 'transfers_from_traces',
-        materialized = 'view',
+        partition_by = ['block_month'],
+        materialized = 'incremental',
+        file_format = 'delta',
+        incremental_strategy = 'merge',
+        incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')],
+        unique_key = ['block_date', 'unique_key'],
     )
 }}
 
-{{ transfers_from_traces_macro(blockchain=blockchain) }}
+{{ transfers_from_traces_macro(blockchain=blockchain, transfers_start_date='2024-08-28') }}

dbt_subprojects/tokens/models/transfers_and_balances/apechain/tokens_apechain_transfers_from_traces_base.sql

@@ -13,4 +13,4 @@
     )
 }}
 
-{{ transfers_from_traces_base_macro(blockchain=blockchain, easy_dates=false) }}
+{{ transfers_from_traces_base_macro(blockchain=blockchain) }}

dbt_subprojects/tokens/models/transfers_and_balances/arbitrum/tokens_arbitrum_transfers_from_traces.sql

@@ -4,8 +4,13 @@
     config(
         schema = 'tokens_' ~ blockchain,
         alias = 'transfers_from_traces',
-        materialized = 'view',
+        partition_by = ['block_month'],
+        materialized = 'incremental',
+        file_format = 'delta',
+        incremental_strategy = 'merge',
+        incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')],
+        unique_key = ['block_date', 'unique_key'],
     )
 }}
 
-{{ transfers_from_traces_macro(blockchain=blockchain) }}
+{{ transfers_from_traces_macro(blockchain=blockchain, transfers_start_date='2021-05-29') }}

dbt_subprojects/tokens/models/transfers_and_balances/avalanche_c/tokens_avalanche_c_transfers_from_traces.sql

@@ -4,8 +4,13 @@
     config(
         schema = 'tokens_' ~ blockchain,
         alias = 'transfers_from_traces',
-        materialized = 'view',
+        partition_by = ['block_month'],
+        materialized = 'incremental',
+        file_format = 'delta',
+        incremental_strategy = 'merge',
+        incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')],
+        unique_key = ['block_date', 'unique_key'],
     )
 }}
 
-{{ transfers_from_traces_macro(blockchain=blockchain) }}
+{{ transfers_from_traces_macro(blockchain=blockchain, transfers_start_date='2020-09-23') }}

dbt_subprojects/tokens/models/transfers_and_balances/b3/tokens_b3_transfers_from_traces.sql

@@ -4,8 +4,13 @@
     config(
         schema = 'tokens_' ~ blockchain,
         alias = 'transfers_from_traces',
-        materialized = 'view',
+        partition_by = ['block_month'],
+        materialized = 'incremental',
+        file_format = 'delta',
+        incremental_strategy = 'merge',
+        incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')],
+        unique_key = ['block_date', 'unique_key'],
     )
 }}
 
-{{ transfers_from_traces_macro(blockchain=blockchain) }}
+{{ transfers_from_traces_macro(blockchain=blockchain, transfers_start_date='2024-07-31') }}