hold

dbt_project.yml

@@ -53,7 +53,7 @@ vars:
   OBSERV_FULL_TEST: False
   START_GHA_TASKS: False
   STREAMLINE_INVOKE_STREAMS: False
-  STREAMLINE_USE_DEV_FOR_EXTERNAL_TABLES: False
+  STREAMLINE_USE_DEV_FOR_EXTERNAL_TABLES: True
   STREAMLINE_RUN_HISTORY: False
   STREAMLINE_RETRY_UNKNOWN: False
   UPDATE_SNOWFLAKE_TAGS: True

models/defillama/silver/silver__defillama_protocol_tvl.sql

@@ -10,7 +10,7 @@ WITH FINAL AS (
     SELECT
         timestamp,
         protocol_id,
-        category,
+        TO_VARIANT(category) AS category,
         protocol,
         NULL AS market_cap,
         symbol,

models/defillama/streamline/streamline__defillama_protocol_historical.sql

@@ -5,8 +5,8 @@
         target = "{{this.schema}}.{{this.identifier}}",
         params ={ "external_table" :"defillama_protocol_historical",
         "sql_limit" :"10000",
-        "producer_batch_size" :"10",
-        "worker_batch_size" :"1",
+        "producer_batch_size" :"10000",
+        "worker_batch_size" :"2000",
         "async_concurrent_requests" :"1",
         "sql_source" :"{{this.identifier}}" }
     ),
@@ -20,6 +20,29 @@ WITH base AS (
         protocol_id
     FROM
         {{ ref('defillama__dim_protocols') }}
+),
+recent_pulls AS (
+    SELECT
+        protocol_id,
+        _inserted_timestamp AS last_pulled_at
+    FROM
+        {{ ref('streamline__defillama_protocol_historical_complete') }}
+    QUALIFY ROW_NUMBER() OVER (
+        PARTITION BY protocol_id
+        ORDER BY _inserted_timestamp DESC
+    ) = 1
+),
+eligible_protocols AS (
+    SELECT
+        b.protocol_id,
+        b.protocol_slug
+    FROM
+        base b
+        LEFT JOIN recent_pulls r
+          ON r.protocol_id = b.protocol_id
+    WHERE
+        r.last_pulled_at IS NULL
+        OR r.last_pulled_at < DATEADD('hour', -4, SYSDATE())
 )
 SELECT
     protocol_id,
@@ -36,7 +59,7 @@ SELECT
         'Vault/prod/external/defillama'
     ) AS request
 FROM
-    base
+    eligible_protocols
 ORDER BY
     protocol_id::int ASC