get all distinct txs; null amt/mint bug (#667)

* get all distinct txs; null amt/mint bug

* formatting

data/testing__swaps_intermediate_meteora.csv

@@ -15,4 +15,8 @@ NfvsjHzJx5s5JWDK7pJaWUv7hXJGdWUYwqaoRVLxQ2d5q3Jyrs2uft9FUyeNFVyxVSVZFPdsTv45dpgd
 2CkwiLBE82MALvqW4uHscPLbEdAqgVXhSPsmoeVPq6wzS9iTv1wDfwsE7SBqoR6j7L2KFyi2NMwAy5Yq5a45RuVy,BRentiWgeVVNtzMxgJwR9GhMkozjuYiFCCDuMaSS9o2,0.005,So11111111111111111111111111111111111111112,0.61371687,TNSRxcUxoT9xBG3de7PiJyTDYu7kskLqcpddxnEJAS6,1
 2CkwiLBE82MALvqW4uHscPLbEdAqgVXhSPsmoeVPq6wzS9iTv1wDfwsE7SBqoR6j7L2KFyi2NMwAy5Yq5a45RuVy,BRentiWgeVVNtzMxgJwR9GhMkozjuYiFCCDuMaSS9o2,0.61371687,TNSRxcUxoT9xBG3de7PiJyTDYu7kskLqcpddxnEJAS6,1.278914,85VBFQZC9TZkfaptBWjvUw7YbZjy52A6mjtPGjstQAmQ,2
 2CkwiLBE82MALvqW4uHscPLbEdAqgVXhSPsmoeVPq6wzS9iTv1wDfwsE7SBqoR6j7L2KFyi2NMwAy5Yq5a45RuVy,BRentiWgeVVNtzMxgJwR9GhMkozjuYiFCCDuMaSS9o2,1.278914,85VBFQZC9TZkfaptBWjvUw7YbZjy52A6mjtPGjstQAmQ,0.761483,JUPyiwrYJFskUPiHa7hkeR8VUtAeFoSYbKedZNsDvCN,3
-2CkwiLBE82MALvqW4uHscPLbEdAqgVXhSPsmoeVPq6wzS9iTv1wDfwsE7SBqoR6j7L2KFyi2NMwAy5Yq5a45RuVy,BRentiWgeVVNtzMxgJwR9GhMkozjuYiFCCDuMaSS9o2,0.761483,JUPyiwrYJFskUPiHa7hkeR8VUtAeFoSYbKedZNsDvCN,0.005011026,So11111111111111111111111111111111111111112,4
+2CkwiLBE82MALvqW4uHscPLbEdAqgVXhSPsmoeVPq6wzS9iTv1wDfwsE7SBqoR6j7L2KFyi2NMwAy5Yq5a45RuVy,BRentiWgeVVNtzMxgJwR9GhMkozjuYiFCCDuMaSS9o2,0.761483,JUPyiwrYJFskUPiHa7hkeR8VUtAeFoSYbKedZNsDvCN,0.005011026,So11111111111111111111111111111111111111112,4
+42a7GeWqCVi8p3e5EGqKh6pcYnuqK4c6Kjg9ZmiN7SKe1aFhtvUDQJrPiMinTwvKPeAcTVVYjXE4TH33mMH52yD5,3Cf6RJQrCsaKqZ2Hh9MX5zmZBTvoNP89VW6LaoLacSiC,7,EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v,2.146408,27G8MtK7VtTcCHkpASjSDdkWWYfoqT6ggEuKidVJidD4,1
+42a7GeWqCVi8p3e5EGqKh6pcYnuqK4c6Kjg9ZmiN7SKe1aFhtvUDQJrPiMinTwvKPeAcTVVYjXE4TH33mMH52yD5,3Cf6RJQrCsaKqZ2Hh9MX5zmZBTvoNP89VW6LaoLacSiC,2.146408,27G8MtK7VtTcCHkpASjSDdkWWYfoqT6ggEuKidVJidD4,1.017196,H6JtPso9FsoWj2ni6XzhT4dhQxADxkRsHT6GfVfwdGQP,2
+42a7GeWqCVi8p3e5EGqKh6pcYnuqK4c6Kjg9ZmiN7SKe1aFhtvUDQJrPiMinTwvKPeAcTVVYjXE4TH33mMH52yD5,3Cf6RJQrCsaKqZ2Hh9MX5zmZBTvoNP89VW6LaoLacSiC,1.017196,H6JtPso9FsoWj2ni6XzhT4dhQxADxkRsHT6GfVfwdGQP,7.011509,EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v,3
+5KboaM89D79KwHwmRaoaSA7wD3xduMpdqVnWWxkEYvi4b16dTKBQBtGQYoQ1PgStD66mF95vi7FGNrcfmW3EGTpB,CDReHTkyzsawZpfYhZ1MX7zqpkegzVbyqcWwr48KtFSV,0.0003,BvjkVW3zVRNaeqmx9d6FKeppZaTorJjJe2qeBip5pump,0,So11111111111111111111111111111111111111112,1

models/silver/swaps/meteora/silver__swaps_intermediate_meteora.sql

@@ -11,37 +11,57 @@
 
 {% if execute %}
     {% set base_query %}
-    CREATE OR REPLACE TEMPORARY TABLE silver.swaps_intermediate_meteora__intermediate_tmp AS
-    SELECT
-        *
-    FROM
-        {{ ref('silver__decoded_instructions_combined') }}
-    WHERE
-        program_id IN (
-            'LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo',
-            -- DLMM program
-            'Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB' -- AMM program
+        CREATE OR REPLACE TEMPORARY TABLE silver.swaps_intermediate_meteora__intermediate_tmp AS 
+        WITH distinct_entities AS (
+            SELECT DISTINCT
+                tx_id
+            FROM 
+                {{ ref('silver__decoded_instructions_combined') }} d
+            WHERE
+                program_id IN (
+                'LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo', -- DLMM program
+                'Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB' -- AMM program
+                )
+                AND event_type = 'swap'                
+                AND succeeded
+                {% if is_incremental() %}
+                AND _inserted_timestamp >= (
+                    SELECT
+                        MAX(_inserted_timestamp) - INTERVAL '1 hour'
+                    FROM
+                        {{ this }}
+                )
+                {% endif %}
         )
-        AND event_type = 'swap'
+        /* need to re-select all decoded instructions from all tx_ids in incremental subset 
+        in order for the window function to output accurate values */
+        SELECT 
+            d.block_timestamp,
+            d.block_id,
+            d.tx_id,
+            d.index,
+            d.inner_index,
+            d.signers,
+            d.succeeded,
+            d.program_id,
+            d.decoded_instruction,
+            d.event_type,
+            d._inserted_timestamp
+        FROM 
+            {{ ref('silver__decoded_instructions_combined') }} d
+        JOIN
+            distinct_entities
+            USING(tx_id)
+        WHERE
+             program_id IN (
+            'LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo',
+            'Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB'
+        )
+        AND event_type = 'swap'                
         AND succeeded
-
-    {% if is_incremental() %}
-    AND _inserted_timestamp >= (
-        SELECT
-            MAX(_inserted_timestamp) - INTERVAL '1 hour'
-        FROM
-            {{ this }}
-    )
-    {% else %}
-        AND block_timestamp :: DATE >= '2022-07-14'
-    {% endif %}
     {% endset %}
-    
     {% do run_query(base_query) %}
-    {% set between_stmts = fsc_utils.dynamic_range_predicate(
-        "silver.swaps_intermediate_meteora__intermediate_tmp",
-        "block_timestamp::date"
-    ) %}
+    {% set between_stmts = fsc_utils.dynamic_range_predicate("silver.swaps_intermediate_meteora__intermediate_tmp","block_timestamp::date") %}
 {% endif %}
 
 WITH base AS (
@@ -245,18 +265,63 @@ pre_final AS (
         ROW_NUMBER() over (PARTITION BY A.tx_id, A.index, A.inner_INDEX
             ORDER BY inner_index
         ) = 1
+),
+jup_decoded_swaps as (
+    SELECT 
+        tx_id,
+        index,
+        swap_program_id,
+        to_amount,
+        to_mint,
+        block_timestamp
+    FROM 
+        {{ ref('silver__swaps_inner_intermediate_jupiterv6') }}
+    WHERE
+        swap_program_id = 'Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB'
+        AND {{ between_stmts }}
+),
+swaps as (
+    SELECT
+        p.block_id,
+        p.block_timestamp,
+        p.program_id,
+        p.tx_id,
+        ROW_NUMBER() over (
+            PARTITION BY p.tx_id
+            ORDER BY
+                p.index,
+                p.inner_index
+        ) AS swap_index,
+        p.succeeded,
+        p.swapper,
+        p.from_amt,
+        p.from_mint,
+        COALESCE(
+            p.to_amt,
+            j.to_amount
+        ) AS to_amt,
+        COALESCE(
+            p.to_mint,
+            j.to_mint
+        ) AS to_mint,
+        p._inserted_timestamp
+    FROM
+        pre_final p
+        LEFT JOIN jup_decoded_swaps j
+        ON p.tx_id = j.tx_id
+        AND p.index = j.index
+        AND p.block_timestamp :: DATE = j.block_timestamp :: DATE
+        AND program_id = 'Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB'
+        AND p.to_amt IS NULL
+        AND p.to_mint IS NULL
 )
+
 SELECT
     block_id,
     block_timestamp,
     program_id,
     tx_id,
-    ROW_NUMBER() over (
-        PARTITION BY tx_id
-        ORDER BY
-            INDEX,
-            inner_index
-    ) AS swap_index,
+    swap_index,
     succeeded,
     swapper,
     from_amt,
@@ -269,4 +334,4 @@ SELECT
     SYSDATE() AS modified_timestamp,
     '{{ invocation_id }}' AS invocation_id
 FROM
-    pre_final
+    swaps

models/silver/swaps/meteora/silver__swaps_intermediate_meteora.yml

@@ -26,7 +26,9 @@ models:
           '2eV5PUm4WL7X9T1X4nSbBmAGDLTx9pUiygK34ngMSyR9QwQUS8ZpnzGKTCbuJHeVGYNHjHBGtPh4fLYxkBLGVJeB',
           '35QkSCVWRNFaqFM8DZrDWKWsBdDTrHVMmskd3H9D3pcQstdYsWqZiXVhMpkaCFrBQTi8XZaWgNZaWnQZzwPpZ1dv',
           '2CkwiLBE82MALvqW4uHscPLbEdAqgVXhSPsmoeVPq6wzS9iTv1wDfwsE7SBqoR6j7L2KFyi2NMwAy5Yq5a45RuVy',
-          '64PW2o1TjxU3kJXAWsAZrL6wJAhKeAY2esJNbjp9UfBb4q4mZMF9KdXXqv3DjkAiNZUacitwzU5v8rCCb3zkdvmp')"  
+          '64PW2o1TjxU3kJXAWsAZrL6wJAhKeAY2esJNbjp9UfBb4q4mZMF9KdXXqv3DjkAiNZUacitwzU5v8rCCb3zkdvmp',
+          '42a7GeWqCVi8p3e5EGqKh6pcYnuqK4c6Kjg9ZmiN7SKe1aFhtvUDQJrPiMinTwvKPeAcTVVYjXE4TH33mMH52yD5',
+          '5KboaM89D79KwHwmRaoaSA7wD3xduMpdqVnWWxkEYvi4b16dTKBQBtGQYoQ1PgStD66mF95vi7FGNrcfmW3EGTpB')"  
     columns:
       - name: BLOCK_TIMESTAMP
         description: "{{ doc('block_timestamp') }}"