perf(chainlink_polygon_vrf_request_fulfilled_daily): push incremental time bounds below the aggregation (#9766)

* perf(chainlink_polygon_vrf_request_fulfilled_daily): push incremental time bounds below the aggregation

* materialize v1 request logs; drop the 7-day lookback assumption

* Address review: align schema tests with vrf_v1_random_request_logs merge key

unique_combination_of_columns was on (blockchain, block_number, tx_hash, index)
but the model's incremental unique_key is [tx_hash, index]. Drop the extra
columns and add not_null tests on tx_hash and index.

Author: a-monteiro

dbt_subprojects/daily_spellbook/models/chainlink/polygon/chainlink_polygon_schema.yml

@@ -918,8 +918,6 @@ models:
       - dbt_utils.unique_combination_of_columns:
           arguments:
             combination_of_columns:
-              - blockchain
-              - block_number
               - tx_hash
               - index
     columns:
@@ -931,10 +929,16 @@ models:
       - *topic1
       - *topic2
       - *topic3
-      - *tx_hash
+      - name: tx_hash
+        description: "Transaction Hash"
+        data_tests:
+          - not_null
       - *block_number
       - *block_time
-      - *index
+      - name: index
+        description: "Index"
+        data_tests:
+          - not_null
       - *tx_index
 
   - name: chainlink_polygon_vrf_v2_random_fulfilled_logs

dbt_subprojects/daily_spellbook/models/chainlink/polygon/chainlink_polygon_vrf_request_fulfilled_daily.sql

@@ -11,6 +11,56 @@
   )
 }}
 
+{%- set incremental_lower_bound -%}
+date_trunc('{{ var("DBT_ENV_INCREMENTAL_TIME_UNIT") }}', now() - interval '{{ var("DBT_ENV_INCREMENTAL_TIME") }}' {{ var("DBT_ENV_INCREMENTAL_TIME_UNIT") }})
+{%- endset -%}
+
+-- The upstream view chainlink_polygon_vrf_request_fulfilled computes
+-- evt_block_time = MAX(block_time) above a GROUP BY, so filtering its output
+-- with incremental_predicate('evt_block_time') cannot reach the polygon.logs
+-- scans (3 full scans, ~2.4 TiB/run, to merge a handful of rows). The view
+-- body is inlined here so the time bounds apply BELOW the aggregations and
+-- prune the Delta scans via block_time file skipping.
+WITH vrf_request_fulfilled AS (
+  SELECT
+    'polygon' as blockchain,
+    MAX(bytearray_to_uint256(bytearray_substring(v1_request.data, 110, 19)) / 1e18) AS token_value,
+    MAX(v1_fulfilled.tx_from) as operator_address,
+    MAX(v1_fulfilled.block_time) as evt_block_time
+  FROM
+    {{ ref('chainlink_polygon_vrf_v1_random_request_logs') }} v1_request
+    INNER JOIN {{ ref('chainlink_polygon_vrf_v1_random_fulfilled_logs') }} v1_fulfilled ON bytearray_substring(v1_fulfilled.data, 1, 32) = bytearray_substring(v1_request.data, 129, 32)
+  {% if is_incremental() %}
+  -- The request side is a materialized ~6.3M-row table read in full, so no
+  -- request-side lookback (or delay assumption) is needed. The fulfilled-side
+  -- bound is exact: a group passes the post-agg evt_block_time predicate iff
+  -- its MAX(fulfilled.block_time) row is itself in the window, and dropping
+  -- older sibling fulfillments cannot change MAX() or the request-side values.
+  WHERE {{ incremental_predicate('v1_fulfilled.block_time') }}
+  {% endif %}
+  GROUP BY
+    v1_request.tx_hash,
+    v1_fulfilled.tx_from
+
+  UNION
+
+  SELECT
+    'polygon' as blockchain,
+    MAX(bytearray_to_uint256(bytearray_substring(v2_fulfilled.data, 33, 32)) / 1e18) AS token_value,
+    MAX(v2_fulfilled.tx_from) as operator_address,
+    MAX(v2_fulfilled.block_time) as evt_block_time
+  FROM
+    {{ ref('chainlink_polygon_vrf_v2_random_fulfilled_logs') }} v2_fulfilled
+  {% if is_incremental() %}
+  -- exact: each group is a single log row (tx_hash + index), so the pre-agg
+  -- bound is equivalent to the post-agg evt_block_time predicate
+  WHERE {{ incremental_predicate('v2_fulfilled.block_time') }}
+  {% endif %}
+  GROUP BY
+    v2_fulfilled.tx_hash,
+    v2_fulfilled.tx_from,
+    v2_fulfilled.index
+)
 
 SELECT
   'polygon' as blockchain,
@@ -19,11 +69,11 @@ SELECT
   vrf_request_fulfilled.operator_address,
   SUM(token_value) as token_amount
 FROM
-  {{ref('chainlink_polygon_vrf_request_fulfilled')}} vrf_request_fulfilled
+  vrf_request_fulfilled
 {% if is_incremental() %}
   WHERE {{ incremental_predicate('evt_block_time') }}
 {% endif %}
 GROUP BY
   2, 4
 ORDER BY
-  2, 4
+  2, 4

dbt_subprojects/daily_spellbook/models/chainlink/polygon/chainlink_polygon_vrf_v1_random_request_logs.sql

@@ -1,11 +1,19 @@
 {{
   config(
-    
+
     alias='vrf_v1_random_request_logs',
-    materialized='view'
+    materialized='incremental',
+    file_format='delta',
+    incremental_strategy='merge',
+    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')],
+    unique_key=['tx_hash', 'index']
   )
 }}
 
+-- Materialized (was a view) so downstream consumers join the full ~6.3M-row
+-- request history without rescanning polygon.logs (50B+ rows); logs are
+-- append-only, so an incremental block_time window is exact.
+
 SELECT
   'polygon' as blockchain,
   block_hash,
@@ -23,4 +31,7 @@ SELECT
 FROM
   {{ source('polygon', 'logs') }} logs
 WHERE
-  topic0 = 0x56bd374744a66d531874338def36c906e3a6cf31176eb1e9afd9f1de69725d51 -- RandomnessRequest
+  topic0 = 0x56bd374744a66d531874338def36c906e3a6cf31176eb1e9afd9f1de69725d51 -- RandomnessRequest
+{% if is_incremental() %}
+  AND {{ incremental_predicate('block_time') }}
+{% endif %}