solana utils: refactor token accounts pipeline to account for historical states

dbt_subprojects/solana/macros/_sector/transfers/solana_sol_transfers_macro.sql

@@ -1,67 +1,62 @@
 {% macro solana_sol_transfers_macro(start_date, end_date) %}
 
 WITH transfers AS (
-    SELECT
-        'solana' as blockchain,
-        call_block_time as block_time,
-        cast(date_trunc('day', call_block_time) AS date) AS block_date,
-        cast(date_trunc('month', call_block_time) AS date) AS block_month,
-        call_block_slot as block_slot,
-        call_tx_id as tx_id,
-        call_tx_index as tx_index,
-        call_inner_instruction_index as inner_instruction_index,
-        call_outer_instruction_index as outer_instruction_index,
-        call_tx_signer as tx_signer,
-        COALESCE(tk_from.token_balance_owner, call_account_arguments[1]) AS from_owner, -- if the token account exists, use the owner of that, otherwise if should be an account
-        COALESCE(tk_to.token_balance_owner, call_account_arguments[2]) AS to_owner,
-        CASE WHEN tk_from.address IS NOT NULL THEN tk_from.address ELSE null END as from_token_account, -- if the token account exists, use the address of that, otherwise no token accounts are involved
-        CASE WHEN tk_to.address IS NOT NULL THEN tk_to.address ELSE null END as to_token_account,
-        'native' as token_version,
-        'SOL' as symbol,
-        lamports as amount,
-        lamports / 1e9 as amount_display,
-        'So11111111111111111111111111111111111111112' as token_mint_address,
-        call_outer_executing_account as outer_executing_account,
-        call_inner_executing_account as inner_executing_account,
-        CASE WHEN tk_to.address IS NOT NULL THEN 'wrap' ELSE 'transfer' END as action -- if the token account exists, it's a wrap, otherwise it's a transfer
-    FROM 
-        {{ source('system_program_solana', 'system_program_call_Transfer') }} t
-    LEFT JOIN 
-        {{ ref('solana_utils_token_accounts') }} tk_from 
-        ON tk_from.address = t.call_account_arguments[1]
-    LEFT JOIN 
-        {{ ref('solana_utils_token_accounts') }} tk_to 
-        ON tk_to.address = t.call_account_arguments[2]
-    WHERE
+    select
+        call_block_time as block_time
+        , cast(date_trunc('day', call_block_time) AS date) AS block_date
+        , cast(date_trunc('month', call_block_time) AS date) AS block_month
+        , call_block_slot as block_slot
+        , call_tx_id as tx_id
+        , call_tx_index as tx_index
+        , call_inner_instruction_index as inner_instruction_index
+        , call_outer_instruction_index as outer_instruction_index
+        , call_tx_signer as tx_signer
+        , lamports as amount
+        , lamports / 1e9 as amount_display
+        , call_outer_executing_account as outer_executing_account
+        , call_inner_executing_account as inner_executing_account
+        , substring(call_account_arguments[1], 1, 2) as from_token_account_prefix
+        , call_account_arguments[1] as from_token_account
+        , substring(call_account_arguments[2], 1, 2) as to_token_account_prefix
+        , call_account_arguments[2] as to_token_account
+        , CONCAT(
+            lpad(cast(call_block_slot as varchar), 12, '0'), '-',
+            lpad(cast(call_tx_index as varchar), 6, '0'), '-',
+            lpad(cast(coalesce(call_outer_instruction_index, 0) as varchar), 4, '0'), '-',
+            lpad(cast(coalesce(call_inner_instruction_index, 0) as varchar), 4, '0')
+        ) AS unique_instruction_key --block time is not granular enough, build unique key from block_slot, tx_index, outer_instruction_index, inner_instruction_index
+    from
+        {{ source('system_program_solana', 'system_program_call_Transfer') }} as t
+    where
         1=1
-        {% if is_incremental() %}
-        AND {{incremental_predicate('call_block_time')}}
-        {% else %}
-        AND call_block_time >= {{start_date}}
-        AND call_block_time < {{end_date}}
-        {% endif %}
+        {% if is_incremental() or true -%}
+        and {{incremental_predicate('call_block_time')}}
+        {% else -%}
+        and call_block_time >= {{start_date}}
+        and call_block_time < {{end_date}}
+        {% endif -%}
 )
 , prices AS (
-    SELECT
+    select
         contract_address,
         minute,
         price,
         decimals
-    FROM 
+    from
         {{ source('prices', 'usd_forward_fill') }}
-    WHERE
+    where
         blockchain = 'solana'
-        AND contract_address =  0x069b8857feab8184fb687f634618c035dac439dc1aeb3b5598a0f00000000001 -- SOL address
-        AND minute >= TIMESTAMP '2020-10-02 00:00' --solana start date
-        {% if is_incremental() %}
-        AND {{incremental_predicate('minute')}}
-        {% else %}
-        AND minute >= {{start_date}}
-        AND minute < {{end_date}}
-        {% endif %}
+        and contract_address =  0x069b8857feab8184fb687f634618c035dac439dc1aeb3b5598a0f00000000001 -- SOL address
+        and minute >= TIMESTAMP '2020-10-02 00:00' --solana start date
+        {% if is_incremental() or true -%}
+        and {{incremental_predicate('minute')}}
+        {% else -%}
+        and minute >= {{start_date}}
+        and minute < {{end_date}}
+        {% endif -%}
 )
-SELECT
-    t.blockchain
+select
+    'solana' as blockchain
     , t.block_month
     , t.block_date
     , t.block_time
@@ -72,24 +67,35 @@ SELECT
     , coalesce(t.inner_instruction_index, 0) as key_inner_instruction_index
     , t.outer_instruction_index
     , t.tx_signer
-    , t.from_owner
-    , t.to_owner
-    , t.from_token_account
-    , t.to_token_account
-    , t.token_mint_address
-    , t.symbol
+    , COALESCE(tk_from.token_balance_owner, call_account_arguments[1]) AS from_owner -- if the token account exists, use the owner of that, otherwise it should be an account
+    , COALESCE(tk_to.token_balance_owner, call_account_arguments[2]) AS to_owner
+    , tk_from.address as from_token_account, -- if the token account exists, use the address of that, otherwise no token accounts are involved
+    , tk_to.address as to_token_account
+    , 'So11111111111111111111111111111111111111112' as token_mint_address
+    , 'SOL' as symbol
     , t.amount_display
     , t.amount
     , p.price * (t.amount_display) as amount_usd
     , p.price as price_usd
-    , t.action
+    , CASE WHEN tk_to.address IS NOT NULL THEN 'wrap' ELSE 'transfer' END as action -- if the token account exists, it's a wrap, otherwise it's a transfer
     , t.outer_executing_account
     , t.inner_executing_account
-    , t.token_version
-FROM
-    transfers t
-LEFT JOIN
-    prices p
-    ON p.minute = date_trunc('minute', t.block_time)
-
+    , 'native' as token_version
+from
+    transfers as t
+left join 
+    {{ ref('solana_utils_token_accounts_state_history') }} as tk_from    
+    on t.from_token_account_prefix = tk_from.address_prefix
+    and t.from_token_account = tk_from.address
+    and t.unique_instruction_key >= tk_from.valid_from_unique_instruction_key
+    and t.unique_instruction_key < tk_from.valid_to_unique_instruction_key
+left join 
+    {{ ref('solana_utils_token_accounts_state_history') }} as tk_to 
+    on t.to_token_account_prefix = tk_to.address_prefix
+    and t.to_token_account = tk_to.address
+    and t.unique_instruction_key >= tk_to.valid_from_unique_instruction_key
+    and t.unique_instruction_key < tk_to.valid_to_unique_instruction_key
+left join
+    prices as p
+    on p.minute = date_trunc('minute', t.block_time)
 {% endmacro %}

dbt_subprojects/solana/macros/_sector/transfers/solana_spl_transfers_call_transfer_macro.sql

@@ -1,31 +1,38 @@
 {% macro solana_spl_transfers_call_transfer_macro(start_date, end_date) %}
 
-WITH base AS (
+WITH transfers AS (
     SELECT
         call_block_time as block_time
         , cast(date_trunc('day', call_block_time) as date) as block_date
         , cast(date_trunc('month', call_block_time) as date) as block_month
         , call_block_slot as block_slot
-        , 'transfer' as action
         , amount
+        , substring(account_source, 1, 2) as from_token_account_prefix
         , account_source as from_token_account
+        , substring(account_destination, 1, 2) as to_token_account_prefix
         , account_destination as to_token_account
-        , 'spl_token' as token_version
         , call_tx_signer as tx_signer
         , call_tx_id as tx_id
+        , call_tx_index as tx_index
         , call_outer_instruction_index as outer_instruction_index
         , COALESCE(call_inner_instruction_index,0) as inner_instruction_index
         , call_outer_executing_account as outer_executing_account
+        , CONCAT(
+            lpad(cast(call_block_slot as varchar), 12, '0'), '-',
+            lpad(cast(call_tx_index as varchar), 6, '0'), '-',
+            lpad(cast(coalesce(call_outer_instruction_index, 0) as varchar), 4, '0'), '-',
+            lpad(cast(coalesce(call_inner_instruction_index, 0) as varchar), 4, '0')
+        ) AS unique_instruction_key --block time is not granular enough, build unique key from block_slot, tx_index, outer_instruction_index, inner_instruction_index
     FROM 
         {{ source('spl_token_solana','spl_token_call_transfer') }}
     WHERE 
         1=1
-        {% if is_incremental() %}
+        {% if is_incremental() or true -%}
         AND {{incremental_predicate('call_block_time')}}
-        {% else %}
+        {% else -%}
         AND call_block_time >= {{start_date}}
         AND call_block_time < {{end_date}}
-        {% endif %}
+        {% endif -%}
 )
 , prices AS (
     SELECT
@@ -39,56 +46,62 @@ WITH base AS (
     WHERE 
         blockchain = 'solana'
         AND minute >= TIMESTAMP '2020-10-02 00:00' --solana start date
-        {% if is_incremental() %}
+        {% if is_incremental() or true -%}
         AND {{incremental_predicate('minute')}}
-        {% else %}
+        {% else -%}
         AND minute >= {{start_date}}
         AND minute < {{end_date}}
-        {% endif %}
+        {% endif -%}
 )
-
 SELECT
-    b.block_time
-    , b.block_date
-    , b.block_month
-    , b.block_slot
-    , b.action
-    , b.amount
+    t.block_time
+    , t.block_date
+    , t.block_month
+    , t.block_slot
+    , 'transfer' as action
+    , t.amount
     , CASE  
         WHEN p.decimals is null THEN null
-        WHEN p.decimals = 0 THEN b.amount
-        ELSE b.amount / power(10, p.decimals)
+        WHEN p.decimals = 0 THEN t.amount
+        ELSE t.amount / power(10, p.decimals)
       END as amount_display
-    , b.from_token_account
-    , b.to_token_account
+    , t.from_token_account
+    , t.to_token_account
     , tk_s.token_balance_owner as from_owner
     , tk_d.token_balance_owner as to_owner
-    , b.token_version
-    , b.tx_signer
-    , b.tx_id
-    , b.outer_instruction_index
-    , b.inner_instruction_index
-    , b.outer_executing_account
+    , 'spl_token' as token_version
+    , t.tx_signer
+    , t.tx_id
+    , t.tx_index
+    , t.outer_instruction_index
+    , t.inner_instruction_index
+    , t.outer_executing_account
     , COALESCE(tk_s.token_mint_address, tk_d.token_mint_address) as token_mint_address
     , p.price as price_usd
     , CASE 
         WHEN p.decimals is null THEN null
-        WHEN p.decimals = 0 THEN p.price * b.amount
-        ELSE p.price * b.amount / power(10, p.decimals)
+        WHEN p.decimals = 0 THEN p.price * t.amount
+        ELSE p.price * t.amount / power(10, p.decimals)
       END as amount_usd
     , p.symbol as symbol
-FROM base b
+FROM transfers as t
 LEFT JOIN 
-    {{ ref('solana_utils_token_accounts') }} tk_s 
-    ON tk_s.address = b.from_token_account
+    {{ ref('solana_utils_token_accounts_state_history') }} tk_s 
+    ON t.from_token_account_prefix = tk_s.address_prefix
+    and t.from_token_account = tk_s.address
+    and t.unique_instruction_key >= tk_s.valid_from_unique_instruction_key
+    and t.unique_instruction_key < tk_s.valid_to_unique_instruction_key
 LEFT JOIN 
-    {{ ref('solana_utils_token_accounts') }} tk_d 
-    ON tk_d.address = b.to_token_account
+    {{ ref('solana_utils_token_accounts_state_history') }} tk_d 
+    ON t.to_token_account_prefix = tk_d.address_prefix
+    and t.to_token_account = tk_d.address
+    and t.unique_instruction_key >= tk_d.valid_from_unique_instruction_key
+    and t.unique_instruction_key < tk_d.valid_to_unique_instruction_key
 LEFT JOIN 
     {{ ref('solana_utils_token_address_mapping') }} tk_m
     ON tk_m.base58_address = COALESCE(tk_s.token_mint_address, tk_d.token_mint_address)
 LEFT JOIN prices p
     ON p.contract_address = tk_m.binary_address
-    AND p.minute = date_trunc('minute', b.block_time)
+    AND p.minute = date_trunc('minute', t.block_time)
 
 {% endmacro %}