Make NFT models incremental (#6199)

* make unperformant models incremental * fix compile * include opensea v1 * fix compile: * add incremental_predicates
duneanalytics · Jun 18, 2024 · 27832d9 · 27832d9
1 parent c18a11b
commit 27832d9
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 15 deletions.
diff --git a/nft/models/_sector/aggregators/chains/nft_ethereum_aggregators_gem.sql b/nft/models/_sector/aggregators/chains/nft_ethereum_aggregators_gem.sql
@@ -1,14 +1,21 @@
 {{config(
-
     schema = 'nft_ethereum',
-    alias = 'aggregators_gem'
-)}}
+    alias = 'aggregators_gem',
+    materialized = 'incremental',
+    file_format = 'delta',
+    incremental_strategy = 'merge',
+    unique_key='contract_address'
+    )
+}}
 WITH vasa_contracts as (
     SELECT distinct
     address AS contract_address
     FROM {{ source('ethereum','creation_traces') }}
     WHERE "from" = 0x073ab1c0cad3677cde9bdb0cdeedc2085c029579
     and block_time >= TIMESTAMP '2021-10-12'
+    {% if is_incremental() %}
+    AND {{incremental_predicate('block_time')}}
+    {% endif %}
 )
 
 
@@ -18,8 +25,14 @@ select
 from vasa_contracts c
 left join {{ source('ethereum','transactions') }} t
 on t.block_time >= CAST('2021-10-12' AS TIMESTAMP) and t.to = c.contract_address
+{% if is_incremental() %}
+AND {{incremental_predicate('t.block_time')}}
+{% endif %}
 left join {{ ref('nft_ethereum_transfers') }} nt
 on t.block_number = nt.block_number and t.hash = nt.tx_hash
+{% if is_incremental() %}
+AND {{incremental_predicate('nt.block_time')}}
+{% endif %}
 group by 1,2
 having count(distinct t.hash) filter(where t."from" != 0x073ab1c0cad3677cde9bdb0cdeedc2085c029579) > 10
     and count(distinct nt.contract_address) > 2
diff --git a/nft/models/_sector/aggregators/chains/nft_ethereum_aggregators_manual.sql b/nft/models/_sector/aggregators/chains/nft_ethereum_aggregators_manual.sql
@@ -1,5 +1,4 @@
 {{config(
-
     schema = 'nft_ethereum',
     alias='aggregators_manual'
 )}}

diff --git a/nft/models/_sector/aggregators/chains/nft_ethereum_aggregators_markers.sql b/nft/models/_sector/aggregators/chains/nft_ethereum_aggregators_markers.sql
@@ -1,13 +1,11 @@
 {{ config(
-
         schema = 'nft_ethereum',
         alias = 'aggregators_markers',
-		materialized = 'table',
-        unique_key='hash_marker',
-        post_hook='{{ expose_spells(\'["ethereum"]\',
-                                    "sector",
-                                    "nft",
-                                    \'["hildobby", "0xRob"]\') }}')
+		materialized = 'incremental',
+		file_format = 'delta',
+		incremental_strategy = 'merge',
+        unique_key='hash_marker'
+        )
 }}
 
  WITH reservoir AS (
@@ -25,6 +23,9 @@
     AND regexp_replace(cast(data as varchar), '^.*00', '') != '1f'
     AND length(regexp_replace(cast(data as varchar), '^.*00', ''))%2 = 0
     AND block_time > TIMESTAMP '2022-10-15'
+    {% if is_incremental() %}
+    AND {{incremental_predicate('block_time')}}
+    {% endif %}
     )
 
   -- needed to eliminate duplicates

diff --git a/nft/models/_sector/trades/chains/ethereum/platforms/opensea_v1_ethereum_base_trades.sql b/nft/models/_sector/trades/chains/ethereum/platforms/opensea_v1_ethereum_base_trades.sql
@@ -1,9 +1,11 @@
 {{ config(
     schema = 'opensea_v1_ethereum',
     alias = 'base_trades',
-    tags =['static'],
-    materialized = 'table',
-    file_format = 'delta'
+    materialized = 'incremental',
+    incremental_strategy = 'merge',
+    file_format = 'delta',
+    unique_key = ['block_number','tx_hash','sub_tx_trade_id'],
+    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')]
     )
 }}
 
@@ -68,9 +70,13 @@ WITH wyvern_call_data as (
     FROM
       {{ source('opensea_ethereum','wyvernexchange_call_atomicmatch_') }} wc
     WHERE 1=1
-        AND (element_at(addrs,4) = {{OS_WALLET}} OR element_at(addrs,11) = {{OS_WALLET}}) -- limit to OpenSea
+    {% if is_incremental() %}
+    AND {{incremental_predicate('call_block_time')}}
+    {% endif %}
+    AND (element_at(addrs,4) = {{OS_WALLET}} OR element_at(addrs,11) = {{OS_WALLET}}) -- limit to OpenSea
     AND call_success = true
     AND call_block_time >= TIMESTAMP '{{START_DATE}}' AND call_block_time <= TIMESTAMP '{{END_DATE}}'
+
 ),
 
 
@@ -85,6 +91,9 @@ order_prices as (
         lag(evt_index) over (partition by evt_block_number, evt_tx_hash order by evt_index asc) as prev_order_evt_index
     from {{ source('opensea_ethereum','wyvernexchange_evt_ordersmatched') }}
     WHERE evt_block_time >= TIMESTAMP '{{START_DATE}}' AND evt_block_time <= TIMESTAMP '{{END_DATE}}'
+    {% if is_incremental() %}
+    AND {{incremental_predicate('evt_block_time')}}
+    {% endif %}
 
 ),
 -- needed to pull token_id, token_amounts, token_standard and nft_contract_address
@@ -102,6 +111,9 @@ nft_transfers as (
         tx_hash
     from {{ ref('nft_ethereum_transfers') }}
     WHERE block_time >= TIMESTAMP '{{START_DATE}}' AND block_time <= TIMESTAMP '{{END_DATE}}'
+    {% if is_incremental() %}
+    AND {{incremental_predicate('block_time')}}
+    {% endif %}
 ),
 
 -- join call and order data