Migrate blocks to streamline 2.0 (#675)

* bronze streamline blocks

* complete blocks and streamline requests

* add workflow

* revert

* update cutoff

.github/workflows/dbt_run_streamline_blocks.yml

@@ -0,0 +1,45 @@
+name: dbt_run_streamline_blocks
+run-name: dbt_run_streamline_blocks
+
+on:
+  workflow_dispatch:
+    branches:
+      - "main"
+    
+env:
+  DBT_PROFILES_DIR: "${{ vars.DBT_PROFILES_DIR }}"
+
+  ACCOUNT: "${{ vars.ACCOUNT }}"
+  ROLE: "${{ vars.ROLE }}"
+  USER: "${{ vars.USER }}"
+  PASSWORD: "${{ secrets.PASSWORD }}"
+  REGION: "${{ vars.REGION }}"
+  DATABASE: "${{ vars.DATABASE }}"
+  WAREHOUSE: "${{ vars.WAREHOUSE }}"
+  SCHEMA: "${{ vars.SCHEMA }}"
+
+concurrency:
+  group: ${{ github.workflow }}
+
+jobs:
+  run_dbt_jobs:
+    runs-on: ubuntu-latest
+    environment: 
+      name: workflow_prod
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - uses: actions/setup-python@v4
+        with:
+          python-version: "${{ vars.PYTHON_VERSION }}"
+          cache: "pip"
+
+      - name: install dependencies
+        run: |
+          pip install -r requirements.txt
+          dbt deps
+      - name: Run DBT Jobs
+        run: |
+          dbt run -s models/streamline/core/complete/streamline__complete_blocks_2.sql
+          dbt run --vars '{"STREAMLINE_INVOKE_STREAMS": True}' -s streamline__blocks_2

data/github_actions__workflows.csv

@@ -7,4 +7,5 @@ dbt_run_nft_compressed_mints_realtime,"*/15 * * * *"
 dbt_run_nft_compressed_mints_sales_realtime,"55 * * * *"
 dbt_run_decode_logs_orchestrator,"*/15 * * * *"
 dbt_run_streamline_block_rewards,"*/15 * * * *"
-dbt_run_streamline_helius_cnft_metadata,"*/10 * * * *"
+dbt_run_streamline_blocks,"*/5 * * * *"
+dbt_run_streamline_helius_cnft_metadata,"*/10 * * * *"

models/bronze/streamline/bronze__streamline_FR_blocks_2.sql

@@ -0,0 +1,12 @@
+{{ config (
+    materialized = 'view'
+) }}
+
+{% set model = "blocks_2" %}
+{{ streamline_external_table_FR_query_v2(
+    model,
+    partition_function = "split_part(file_name, '/', 3)",
+    partition_name = "_partition_by_created_date",
+    unique_key = "block_id",
+    other_cols="error"
+) }}

models/bronze/streamline/bronze__streamline_blocks_2.sql

@@ -0,0 +1,12 @@
+{{ config (
+    materialized = 'view'
+) }}
+
+{% set model = "blocks_2" %}
+{{ streamline_external_table_query_v2(
+    model,
+    partition_function = "split_part(file_name, '/', 3)",
+    partition_name = "_partition_by_created_date",
+    unique_key = "block_id",
+    other_cols="error"
+) }}

models/sources.yml

@@ -58,6 +58,7 @@ sources:
       - name: stake_program_accounts_2
       - name: block_rewards_2
       - name: helius_nft_metadata
+      - name: blocks_2
   - name: bronze_api
     schema: bronze_api
     tables:
@@ -99,6 +100,7 @@ sources:
     tables:
       - name: complete_block_txs
       - name: complete_block_rewards
+      - name: complete_blocks
   - name: bronze
     schema: bronze
     tables:

models/streamline/core/complete/streamline__complete_blocks_2.sql

@@ -0,0 +1,32 @@
+-- depends_on: {{ ref('bronze__streamline_blocks_2') }}
+-- depends_on: {{ ref('bronze__streamline_FR_blocks_2') }}
+
+{{ config (
+    materialized = "incremental",
+    unique_key = 'block_id',
+    cluster_by = "ROUND(block_id, -6)",
+) }}
+
+SELECT
+    block_id,
+    error,
+    _partition_by_created_date,
+    _inserted_timestamp
+FROM
+    {% if is_incremental() %}
+    {{ ref('bronze__streamline_blocks_2') }}
+    WHERE
+        _inserted_timestamp >= (
+            SELECT
+                coalesce(max(_inserted_timestamp), '1970-01-01'::DATE) max_inserted_timestamp
+            FROM
+                {{ this }}
+        )
+    {% else %}
+    {{ ref('bronze__streamline_FR_blocks_2') }}
+    {% endif %}
+QUALIFY
+    row_number() OVER (
+        PARTITION BY block_id
+        ORDER BY _inserted_timestamp DESC
+    ) = 1

models/streamline/core/realtime/streamline__blocks_2.sql

@@ -0,0 +1,79 @@
+{{ config (
+    materialized = "view",
+    post_hook = fsc_utils.if_data_call_function_v2(
+        func = 'streamline.udf_bulk_rest_api_v2',
+        target = "{{this.schema}}.{{this.identifier}}",
+        params ={ "external_table" :"blocks_2",
+        "sql_limit" :"25000",
+        "producer_batch_size" :"25000",
+        "worker_batch_size" :"10000",
+        "sql_source" :"{{this.identifier}}",
+        "order_by_column": "block_id", }
+    )
+) }}
+
+{% if execute %}
+    {% set next_batch_num_query %}
+    SELECT
+        greatest(
+            295976123,
+            (SELECT coalesce(max(block_id),0) FROM {{ ref('streamline__complete_blocks_2') }})
+        )+1
+    {% endset %}
+    {% set next_batch_num = run_query(next_batch_num_query)[0][0] %}
+{% endif %}
+
+WITH blocks AS (
+    SELECT
+        block_id
+    FROM
+        {{ ref("streamline__blocks") }}
+    EXCEPT
+    SELECT
+        block_id
+    FROM
+        {{ source('solana_streamline', 'complete_blocks') }}
+    WHERE
+        block_id <= 295976123
+    EXCEPT
+    SELECT 
+        block_id
+    FROM
+        {{ ref('streamline__complete_blocks_2') }}
+)
+SELECT
+    block_id,
+    replace(current_date::string,'-','_') AS partition_key, -- Issue with streamline handling `-` in partition key so changing to `_`
+    {{ target.database }}.live.udf_api(
+        'POST',
+        '{service}/{Authentication}',
+        OBJECT_CONSTRUCT(
+            'Content-Type',
+            'application/json'
+        ),
+        OBJECT_CONSTRUCT(
+            'id',
+            block_id,
+            'jsonrpc',
+            '2.0',
+            'method',
+            'getBlock',
+            'params',
+            ARRAY_CONSTRUCT(
+                block_id,
+                OBJECT_CONSTRUCT(
+                    'encoding',
+                    'jsonParsed',
+                    'rewards',
+                    False,
+                    'transactionDetails',
+                    'none',
+                    'maxSupportedTransactionVersion',
+                    0
+                )
+            )
+        ),
+        'Vault/prod/solana/quicknode/mainnet'
+    ) AS request
+FROM
+    blocks