multisource curated model

FlipsideCrypto · Dec 16, 2024 · 5fd462e · 5fd462e
1 parent 33656e4
commit 5fd462e
Show file tree

Hide file tree

Showing 6 changed files with 96 additions and 27 deletions.
diff --git a/models/silver/curated/defi/silver__dex_swaps_v2.sql b/models/silver/curated/defi/silver__dex_swaps_v2.sql
@@ -9,6 +9,62 @@
     tags = ['curated','scheduled_non_core'],
 ) }}
 {# Note - multisource model #}
+-- depends on {{ ref('silver__logs_s3') }}
+-- depends on {{ ref('silver__streamline_receipts_final') }}
+{% if execute %}
+
+    {% if is_incremental() and not var("MANUAL_FIX") %}
+    {% do log("Incremental and not MANUAL_FIX", info=True) %}
+    {% set max_mod_query %}
+
+    SELECT
+        MAX(modified_timestamp) modified_timestamp
+    FROM
+        {{ this }}
+
+    {% endset %}
+
+        {% set max_mod = run_query(max_mod_query) [0] [0] %}
+        {% if not max_mod or max_mod == 'None' %}
+            {% set max_mod = '2099-01-01' %}
+        {% endif %}
+
+        {% do log("max_mod: " ~ max_mod, info=True) %}
+
+        {% set min_block_date_query %}
+    SELECT
+        MIN(
+            block_timestamp :: DATE
+        )
+    FROM
+        (
+            SELECT
+                MIN(block_timestamp) block_timestamp
+            FROM
+                {{ ref('silver__logs_s3') }} A
+            WHERE
+                modified_timestamp >= '{{max_mod}}'
+            UNION ALL
+            SELECT
+                MIN(block_timestamp) block_timestamp
+            FROM
+                {{ ref('silver__streamline_receipts_final') }} A
+            WHERE
+                modified_timestamp >= '{{max_mod}}'
+        ) 
+    {% endset %}
+
+        {% set min_bd = run_query(min_block_date_query) [0] [0] %}
+        {% if not min_bd or min_bd == 'None' %}
+            {% set min_bd = '2099-01-01' %}
+        {% endif %}
+
+        {% do log("min_bd: " ~ min_bd, info=True) %}
+
+    {% endif %}
+
+{% endif %}
+
 WITH swap_logs AS (
 
     SELECT
@@ -21,26 +77,22 @@ WITH swap_logs AS (
         log_index,
         clean_log,
         _partition_by_block_number,
-        _inserted_timestamp
+        _inserted_timestamp,
+        modified_timestamp
     FROM
         {{ ref('silver__logs_s3') }}
     WHERE
         receipt_succeeded
         AND clean_log LIKE 'Swapped%'
         AND receiver_id NOT LIKE '%dragon_bot.near' 
 
-        {% if var("MANUAL_FIX") %}
-            AND {{ partition_load_manual('no_buffer') }}
-        {% else %}
-{% if is_incremental() %}
-AND modified_timestamp >= (
-    SELECT
-        MAX(modified_timestamp)
-    FROM
-        {{ this }}
-)
-{% endif %}
-{% endif %}
+    {% if var("MANUAL_FIX") %}
+        AND {{ partition_load_manual('no_buffer') }}
+    {% else %}
+        {% if is_incremental() %}
+            AND block_timestamp :: DATE >= '{{min_bd}}'
+        {% endif %}
+    {% endif %}
 ),
 receipts AS (
     SELECT
@@ -49,7 +101,8 @@ receipts AS (
         receiver_id,
         signer_id,
         _partition_by_block_number,
-        _inserted_timestamp
+        _inserted_timestamp,
+        modified_timestamp
     FROM
         {{ ref('silver__streamline_receipts_final') }}
     WHERE
@@ -59,18 +112,14 @@ receipts AS (
             FROM
                 swap_logs
         ) 
-        {% if var("MANUAL_FIX") %}
-            AND {{ partition_load_manual('no_buffer') }}
-        {% else %}
-{% if is_incremental() %}
-AND modified_timestamp >= (
-    SELECT
-        MAX(modified_timestamp)
-    FROM
-        {{ this }}
-)
-{% endif %}
-{% endif %}
+
+    {% if var("MANUAL_FIX") %}
+        AND {{ partition_load_manual('no_buffer') }}
+    {% else %}
+        {% if is_incremental() %}
+        AND block_timestamp :: DATE >= '{{min_bd}}'
+        {% endif %}
+    {% endif %}
 ),
 swap_outcome AS (
     SELECT
@@ -107,7 +156,8 @@ swap_outcome AS (
             '\\1'
         ) :: STRING AS token_out,
         _partition_by_block_number,
-        _inserted_timestamp
+        _inserted_timestamp,
+        modified_timestamp
     FROM
         swap_logs
 ),
@@ -166,6 +216,14 @@ parse_actions AS (
     FROM
         swap_outcome o
         LEFT JOIN receipts r USING (receipt_object_id)
+
+    {% if is_incremental() and not var("MANUAL_FIX") %}
+        WHERE
+            GREATEST(
+                COALESCE(o.modified_timestamp, '1970-01-01'),
+                COALESCE(r.modified_timestamp, '1970-01-01')   
+            ) >= '{{max_mod}}'
+    {% endif %}
 ),
 FINAL AS (
     SELECT

diff --git a/models/silver/curated/nft/silver__standard_nft_mint_s3.sql b/models/silver/curated/nft/silver__standard_nft_mint_s3.sql
@@ -9,6 +9,9 @@
     tags = ['curated','scheduled_non_core']
 ) }}
 {# Note - multisource model #}
+-- NOTE - does not need to depend on txs final after ez actions table is deployed
+-- will update with if_execute block after that is merged to migrate this model to new ez_actions
+
 WITH logs AS (
 
     SELECT

diff --git a/models/silver/curated/staking/silver__staking_pools_s3.sql b/models/silver/curated/staking/silver__staking_pools_s3.sql
@@ -7,6 +7,8 @@
     tags = ['curated','scheduled_non_core']
 ) }}
 {# Note - multisource model #}
+-- TODO ez_actions refactor
+
 WITH txs AS (
 
     SELECT

diff --git a/models/silver/horizon/silver_horizon__decoded_actions.sql b/models/silver/horizon/silver_horizon__decoded_actions.sql
@@ -6,6 +6,8 @@
     tags = ['curated', 'horizon','scheduled_non_core']
 ) }}
 {# Note - multisource model #}
+-- TODO ez_actions refactor
+
 WITH all_horizon_receipts AS (
 
     SELECT

diff --git a/models/silver/social/silver_social__addkey.sql b/models/silver/social/silver_social__addkey.sql
@@ -7,6 +7,8 @@
     tags = ['curated', 'social','scheduled_non_core']
 ) }}
 {# Note - multisource model #}
+-- TODO ez_actions refactor
+
 WITH receipts AS (
 
     SELECT

diff --git a/models/silver/social/silver_social__decoded_actions.sql b/models/silver/social/silver_social__decoded_actions.sql
@@ -6,6 +6,8 @@
     tags = ['curated', 'social','scheduled_non_core']
 ) }}
 {# Note - multisource model #}
+-- TODO ez_actions refactor
+
 WITH all_social_receipts AS (
 
     SELECT