add/beacon-blocks-retry (#749)

* retry for withdrawals

* test

* number seq

* simplify

* space

* updated test

macros/tests/withdrawal_index_gaps.sql

@@ -0,0 +1,37 @@
+{% test withdrawal_index_gaps(
+    model,
+    column_name
+) %}
+WITH base AS (
+    SELECT
+        {{ column_name }} AS withdrawal_index,
+        LEAD(withdrawal_index) over (
+            ORDER BY
+                withdrawal_index
+        ) AS next_index,
+        slot_number AS start_slot_number,
+        LEAD(slot_number) over (
+            ORDER BY
+                withdrawal_index
+        ) AS end_slot_number
+    FROM
+        {{ model }}
+),
+gaps AS (
+    SELECT
+        withdrawal_index,
+        next_index,
+        withdrawal_index + 1 AS expected_index,
+        start_slot_number,
+        end_slot_number
+    FROM
+        base
+    WHERE
+        next_index IS NOT NULL
+        AND expected_index <> next_index
+)
+SELECT
+    next_index - withdrawal_index AS gaps
+FROM
+    gaps 
+{% endtest %}

models/beacon_chain/silver/silver__beacon_withdrawals.yml

@@ -53,6 +53,10 @@ models:
               column_type_list:
                 - NUMBER
                 - FLOAT
+          - withdrawal_index_gaps:
+              config:
+                severity: error
+                error_if: ">0"
       - name: VALIDATOR_INDEX
         tests:
           - not_null

models/streamline/silver/beacon/realtime/streamline__beacon_blocks_realtime.sql

@@ -7,6 +7,8 @@
     tags = ['streamline_beacon_realtime']
 ) }}
 
+WITH to_do AS (
+
 SELECT
     {{ dbt_utils.generate_surrogate_key(
         ['slot_number']
@@ -25,3 +27,15 @@ FROM
     {{ ref("streamline__complete_beacon_blocks") }}
 WHERE
     slot_number > 5000000
+)
+
+SELECT 
+    id,
+    slot_number
+FROM to_do
+UNION
+SELECT
+    id,
+    slot_number
+FROM
+    {{ ref("_missing_withdrawals") }}

models/streamline/silver/beacon/retry/_missing_withdrawals.sql

@@ -0,0 +1,58 @@
+{{ config (
+    materialized = "ephemeral"
+) }}
+
+WITH base AS (
+
+    SELECT
+        INDEX AS withdrawal_index,
+        LEAD(withdrawal_index) over (
+            ORDER BY
+                withdrawal_index
+        ) AS next_index,
+        slot_number AS start_slot_number,
+        LEAD(slot_number) over (
+            ORDER BY
+                withdrawal_index
+        ) AS end_slot_number
+    FROM
+        {{ ref('silver__beacon_withdrawals') }}
+),
+gaps AS (
+    SELECT
+        withdrawal_index,
+        next_index,
+        withdrawal_index + 1 AS expected_index,
+        start_slot_number,
+        end_slot_number
+    FROM
+        base
+    WHERE
+        next_index IS NOT NULL
+        AND expected_index <> next_index
+),
+FINAL AS (
+    SELECT
+        withdrawal_index,
+        next_index,
+        expected_index,
+        start_slot_number,
+        end_slot_number,
+        start_slot_number + _id AS missing_slot_number
+    FROM
+        gaps
+        JOIN {{ ref('silver__number_sequence') }}
+        ON _id BETWEEN 1
+        AND (
+            end_slot_number - start_slot_number - 1
+        )
+    WHERE
+        missing_slot_number < end_slot_number
+)
+SELECT
+    DISTINCT missing_slot_number AS slot_number,
+    {{ dbt_utils.generate_surrogate_key(
+        ['slot_number']
+    ) }} AS id
+FROM
+    FINAL