tags and uni v4 fix

models/curated_package/contract_reads/streamline/records/aave/silver_reads__aave_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'aave_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH all_tokens AS (

models/curated_package/contract_reads/streamline/records/aave/silver_reads__aave_v2_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'aave_v2_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH all_tokens AS (

models/curated_package/contract_reads/streamline/records/aave/silver_reads__aave_v3_reads.sql

@@ -6,7 +6,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'aave_v3_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH all_tokens AS (

models/curated_package/contract_reads/streamline/records/aerodrome/silver_reads__aerodrome_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'aerodrome_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH liquidity_pools AS (

models/curated_package/contract_reads/streamline/records/binance/silver_reads__binance_v1_reads.sql

@@ -7,7 +7,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'binance_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH contracts AS (

models/curated_package/contract_reads/streamline/records/compound/silver_reads__compound_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'compound_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 -- Compound V1 TVL: balanceOf(MoneyMarket) for each collateral token

models/curated_package/contract_reads/streamline/records/compound/silver_reads__compound_v2_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'compound_v2_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH all_tokens AS (

models/curated_package/contract_reads/streamline/records/compound/silver_reads__compound_v3_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'compound_v3_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH base_tokens AS (

models/curated_package/contract_reads/streamline/records/curve/silver_reads__curve_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'curve_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH liquidity_pools AS (

models/curated_package/contract_reads/streamline/records/eigenlayer/silver_reads__eigenlayer_v1_reads.sql

@@ -6,7 +6,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'eigenlayer_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH strategy_events AS (

models/curated_package/contract_reads/streamline/records/etherfi/silver_reads__etherfi_v1_reads.sql

@@ -7,7 +7,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'etherfi_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH contracts AS (

models/curated_package/contract_reads/streamline/records/lido/silver_reads__lido_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'lido_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH pooled_assets AS (

models/curated_package/contract_reads/streamline/records/morpho/silver_reads__morpho_blue_v1_reads.sql

@@ -6,7 +6,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'morpho_blue_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH market_tokens AS (

models/curated_package/contract_reads/streamline/records/polymarket/silver_reads__polymarket_v1_reads.sql

@@ -7,7 +7,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'polymarket_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH contracts AS (

models/curated_package/contract_reads/streamline/records/rocketpool/silver_reads__rocketpool_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'rocketpool_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 SELECT

models/curated_package/contract_reads/streamline/records/sky/silver_reads__sky_v1_collateral_joins.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'sky_v1_collateral_joins_id',
-    tags = ['silver','contract_reads','sky']
+    tags = ['streamline','contract_reads','records','sky']
 ) }}
 
 -- Sky Protocol (MakerDAO) collateral join discovery

models/curated_package/contract_reads/streamline/records/sky/silver_reads__sky_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'sky_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH collateral_joins AS (

models/curated_package/contract_reads/streamline/records/stablecoins/silver_reads__stablecoins_reads.sql

@@ -10,7 +10,7 @@
     incremental_strategy = 'delete+insert',
     unique_key = 'stablecoins_reads_id',
     post_hook = '{{ unverify_stablecoins() }}',
-    tags = ['silver','contract_reads','heal']
+    tags = ['streamline','contract_reads','records','heal']
 ) }} 
 
 WITH verified_stablecoins AS (

models/curated_package/contract_reads/streamline/records/superchain/silver_reads__superchain_slipstream_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'superchain_slipstream_v1_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH blacklisted_tokens AS (

models/curated_package/contract_reads/streamline/records/tornado_cash/silver_reads__tornado_cash_v1_reads.sql

@@ -8,7 +8,7 @@
     materialized = 'incremental',
     incremental_strategy = 'delete+insert',
     unique_key = 'tornado_cash_reads_id',
-    tags = ['silver','contract_reads']
+    tags = ['streamline','contract_reads','records']
 ) }}
 
 WITH mixers AS (

models/curated_package/contract_reads/streamline/records/uniswap/silver_reads__uniswap_v2_reads.sql

@@ -7,7 +7,7 @@
     incremental_strategy = 'delete+insert',
     unique_key = 'uniswap_v2_reads_id',
     post_hook = '{{ unverify_contract_reads() }}',
-    tags = ['silver','contract_reads','heal']
+    tags = ['streamline','contract_reads','records','heal']
 ) }}
 
 WITH verified_contracts AS (

models/curated_package/contract_reads/streamline/records/uniswap/silver_reads__uniswap_v3_reads.sql

@@ -9,7 +9,7 @@
     incremental_strategy = 'delete+insert',
     unique_key = 'uniswap_v3_reads_id',
     post_hook = '{{ unverify_contract_reads() }}',
-    tags = ['silver','contract_reads','heal']
+    tags = ['streamline','contract_reads','records','heal']
 ) }}
 
 WITH verified_contracts AS (

models/curated_package/contract_reads/streamline/records/uniswap/silver_reads__uniswap_v4_reads.sql

@@ -9,7 +9,7 @@
     incremental_strategy = 'delete+insert',
     unique_key = 'uniswap_v4_reads_id',
     post_hook = '{{ unverify_contract_reads() }}',
-    tags = ['silver','contract_reads','heal']
+    tags = ['streamline','contract_reads','records','heal']
 ) }}
 
 WITH verified_contracts AS (
@@ -20,29 +20,38 @@ WITH verified_contracts AS (
     WHERE
         is_verified
         AND token_address IS NOT NULL
-), 
+),
+{% if is_incremental() %}
+existing_pairs AS (
+    SELECT DISTINCT
+        metadata:token0::STRING AS token0,
+        metadata:token1::STRING AS token1
+    FROM {{ this }}
+),
+{% endif %}
 liquidity_pools AS (
 
     SELECT
-        token0,
-        token1,
-        pool_address AS factory_address,
-        hook_address,
-        protocol,
-        version,
-        platform
+        pools.token0,
+        pools.token1,
+        pools.pool_address AS factory_address,
+        pools.hook_address,
+        pools.protocol,
+        pools.version,
+        pools.platform
     FROM
-        {{ ref('silver_dex__uniswap_v4_pools') }}
-    WHERE token0 IN (SELECT token_address FROM verified_contracts)
-    AND token1 IN (SELECT token_address FROM verified_contracts)
+        {{ ref('silver_dex__uniswap_v4_pools') }} pools
+    {% if is_incremental() %}
+    LEFT JOIN existing_pairs ep
+        ON pools.token0 = ep.token0
+        AND pools.token1 = ep.token1
+    {% endif %}
+    WHERE pools.token0 IN (SELECT token_address FROM verified_contracts)
+    AND pools.token1 IN (SELECT token_address FROM verified_contracts)
     {% if is_incremental() %}
     AND (
-        modified_timestamp > (SELECT MAX(modified_timestamp) FROM {{ this }})
-        OR CONCAT(token0,'-',token1) NOT IN (
-            SELECT CONCAT(metadata:token0::STRING,'-',metadata:token1::STRING) 
-            FROM {{ this }}
-            )
-        -- pull in pools with newly verified tokens
+        pools.modified_timestamp > (SELECT MAX(modified_timestamp) FROM {{ this }})
+        OR ep.token0 IS NULL  -- pull in pools with newly verified tokens
     )
     {% endif %}
 ),