AN-3939/add fix gap logic and obs models tests (#195)

* add fix gap logic and obs models tests

* block height recency test on bitquery table

* add tag

* upd workflow

* explicit cast as int

.github/workflows/dbt_observability_models.yml

@@ -40,13 +40,9 @@ jobs:
           pip install -r requirements.txt
           dbt deps
 
-      - name: Run DBT Jobs - get_block_tx_count
+      - name: Run DBT Jobs - Bitquery getBlockTxCount and Observability models
         run: |
-          dbt run -s tag:get_block_tx_count --vars 'BITQUERY_API_KEY: ${{ secrets.BITQUERY_API_KEY }}'
-
-      - name: Run DBT Jobs - Observability models
-        run: |
-          dbt run -s tag:observability
+          dbt run -s tag:observability --vars 'BITQUERY_API_KEY: ${{ secrets.BITQUERY_API_KEY }}'
 
       - name: Store logs
         uses: actions/upload-artifact@v3

.github/workflows/dbt_test.yml

@@ -44,7 +44,7 @@ jobs:
 
       - name: Run DBT Jobs
         run: |
-          dbt test -s models/gold
+          dbt test -s models/gold tag:observability
         continue-on-error: true
 
       - name: Log test results

models/descriptions/blocks_impacted_array.md

@@ -0,0 +1,5 @@
+{% docs blocks_impacted_array %}
+
+Array of block numbers that were impacted during the test.
+
+{% enddocs %}

models/descriptions/blocks_impacted_count.md

@@ -0,0 +1,5 @@
+{% docs blocks_impacted_count %}
+
+Number of blocks impacted by a gap in the test.
+
+{% enddocs %}

models/descriptions/blocks_tested.md

@@ -0,0 +1,6 @@
+{% docs blocks_tested %}
+
+Number of blocks tested
+
+{% enddocs %}
+

models/descriptions/max_block.md

@@ -0,0 +1,5 @@
+{% docs max_block %}
+
+The max block height in the test.
+
+{% enddocs %}

models/descriptions/max_block_timestamp.md

@@ -0,0 +1,5 @@
+{% docs max_block_timestamp %}
+
+The max block timestamp in the test.
+
+{% enddocs %}

models/descriptions/min_block.md

@@ -0,0 +1,5 @@
+{% docs min_block %}
+
+The min block in the test.
+
+{% enddocs %}

models/descriptions/min_block_timestamp.md

@@ -0,0 +1,5 @@
+{% docs min_block_timestamp %}
+
+The minimum block timestamp for the test.
+
+{% enddocs %}

models/descriptions/test_name.md

@@ -0,0 +1,5 @@
+{% docs test_name %}
+
+Name of the model being tested.
+
+{% enddocs %}

models/descriptions/test_timestamp.md

@@ -0,0 +1,5 @@
+{% docs test_timestamp %}
+
+Timestamp of the test.
+
+{% enddocs %}

models/silver/_observability/silver_observability__block_tx_count.sql

@@ -1,14 +1,58 @@
 {{ config(
     materialized = 'incremental',
     unique_key = 'block_height',
-    tags = ['get_block_tx_count'],
+    tags = ['get_block_tx_count', 'observability'],
     full_refresh = False
 ) }}
 
-WITH starting_block AS (
+{% if var('FIX_GAPS', False) %}
+    WITH blocks AS (
+        SELECT
+            *
+        FROM
+            {{ this }}
+    ),
+    determine_prior_block AS (
+        SELECT
+            block_height,
+            LAG(block_height) over (
+                ORDER BY
+                    block_height
+            ) AS prev_block_height
+        FROM
+            blocks
+    ),
+    gaps AS (
+        SELECT
+            block_height,
+            prev_block_height,
+            block_height - prev_block_height AS gap
+        FROM
+            determine_prior_block
+        WHERE
+            gap > 1
+        ORDER BY
+            1
+    ),
+    params AS (
+        SELECT
+            'query ($network: FlowNetwork!, $block_height_start: Int!, $block_height_end: Int!) { flow(network: $network) { blocks(height: {gt: $block_height_start, lteq: $block_height_end}) { height transactionsCount } } }' AS query,
+            OBJECT_CONSTRUCT(
+                'network',
+                'flow',
+                'block_height_start',
+                prev_block_height,
+                'block_height_end',
+                (prev_block_height + gap - 1) :: INTEGER
+            ) AS variables,
+            '{{ var('BITQUERY_API_KEY', Null) }}' AS api_key
+        FROM
+            gaps
+    ),
+{% else %}
+    WITH starting_block AS (
 
 {% if is_incremental() %}
-
 SELECT
     MAX(block_height) AS block_height_start, {{ target.database }}.streamline.udf_get_chainhead() AS max_block_height
 FROM
@@ -29,14 +73,16 @@ params AS (
             'block_height_end',
             IFF(
                 block_height_start + 25000 > max_block_height,
-                max_block_height,
+                max_block_height - 500,
                 block_height_start + 25000
-            )
+            ) :: INTEGER
         ) AS variables,
         '{{ var('BITQUERY_API_KEY', Null) }}' AS api_key
     FROM
         starting_block
 ),
+{% endif %}
+
 get_bitquery AS (
     SELECT
         {{ target.database }}.live.udf_api(
@@ -61,7 +107,13 @@ get_bitquery AS (
 )
 SELECT
     VALUE :height :: INTEGER AS block_height,
-    VALUE :transactionsCount :: INTEGER AS transaction_ct
+    VALUE :transactionsCount :: INTEGER AS transaction_ct,
+    SYSDATE() AS _inserted_timestamp
 FROM
     get_bitquery,
-    LATERAL FLATTEN(blocks_res)
+    LATERAL FLATTEN(blocks_res) 
+    qualify ROW_NUMBER() over (
+        PARTITION BY block_height
+        ORDER BY
+            _inserted_timestamp DESC
+    ) = 1

models/silver/_observability/silver_observability__block_tx_count.yml

@@ -0,0 +1,26 @@
+version: 2
+
+models:
+  - name: silver_observability__block_tx_count
+    description: |-
+      Query the Bitquery graphQL API to get transaction count by block number.
+    tests:
+      - dbt_utils.recency:
+          datepart: day
+          field: _inserted_timestamp
+          interval: 1
+      - sequence_gaps:
+          column_name: block_height
+          severity: error
+
+    columns:
+      - name: block_height
+        tests:
+          - unique
+          - not_null
+
+      - name: transaction_ct
+        tests:
+          - not_null
+
+      - name: _inserted_timestamp

models/silver/_observability/silver_observability__blocks_completeness.yml

@@ -0,0 +1,43 @@
+version: 2
+
+models:
+  - name: silver_observability__blocks_completeness
+    description: |-
+      Observability model that queries the blocks table at a designated interval to record and track the completeness of the data.
+    tests:
+      - dbt_utils.recency:
+          datepart: day
+          field: test_timestamp
+          interval: 1
+      - dbt_utils.recency:
+          datepart: hours
+          field: max_block_timestamp
+          interval: 24
+
+    columns:
+      - name: TEST_NAME
+        description: "{{ doc('test_name') }}"
+
+      - name: MIN_BLOCK
+        description: "{{ doc('min_block') }}"
+
+      - name: MAX_BLOCK
+        description: "{{ doc('max_block') }}"
+
+      - name: MIN_BLOCK_TIMESTAMP
+        description: "{{ doc('min_block_timestamp') }}"
+
+      - name: MAX_BLOCK_TIMESTAMP
+        description: "{{ doc('max_block_timestamp') }}"
+
+      - name: BLOCKS_TESTED
+        description: "{{ doc('blocks_tested') }}"
+
+      - name: BLOCKS_IMPACTED_COUNT
+        description: "{{ doc('blocks_impacted_count') }}"
+
+      - name: BLOCKS_IMPACTED_ARRAY
+        description: "{{ doc('blocks_impacted_array') }}"
+
+      - name: TEST_TIMESTAMP
+        description: "{{ doc('test_timestamp') }}"

models/silver/_observability/silver_observability__txs_completeness.yml

@@ -0,0 +1,43 @@
+version: 2
+
+models:
+  - name: silver_observability__txs_completeness
+    description: |-
+      Observability model that queries the txs table at a designated interval to record and track the completeness of the data.
+    tests:
+      - dbt_utils.recency:
+          datepart: day
+          field: test_timestamp
+          interval: 1
+      - dbt_utils.recency:
+          datepart: hours
+          field: max_block_timestamp
+          interval: 24
+
+    columns:
+      - name: TEST_NAME
+        description: "{{ doc('test_name') }}"
+
+      - name: MIN_BLOCK
+        description: "{{ doc('min_block') }}"
+
+      - name: MAX_BLOCK
+        description: "{{ doc('max_block') }}"
+
+      - name: MIN_BLOCK_TIMESTAMP
+        description: "{{ doc('min_block_timestamp') }}"
+
+      - name: MAX_BLOCK_TIMESTAMP
+        description: "{{ doc('max_block_timestamp') }}"
+
+      - name: BLOCKS_TESTED
+        description: "{{ doc('blocks_tested') }}"
+
+      - name: BLOCKS_IMPACTED_COUNT
+        description: "{{ doc('blocks_impacted_count') }}"
+
+      - name: BLOCKS_IMPACTED_ARRAY
+        description: "{{ doc('blocks_impacted_array') }}"
+
+      - name: TEST_TIMESTAMP
+        description: "{{ doc('test_timestamp') }}"

tests/tests__block_tx_count_recency.sql

@@ -0,0 +1,25 @@
+{{ config(
+    severity = 'error',
+    tags = ['observability']
+) }}
+
+WITH check_lag AS (
+
+    SELECT
+        {{ target.database }}.streamline.udf_get_chainhead() AS chainhead,
+        (
+            SELECT
+                MAX(block_height)
+            FROM
+                {{ ref('silver_observability__block_tx_count') }}
+        ) AS max_height,
+        (
+            chainhead - max_height < 25000
+        ) AS is_recent
+)
+SELECT
+    *
+FROM
+    check_lag
+WHERE
+    NOT is_recent