flip/base-models
1
0
Fork 0
mirror of https://github.com/FlipsideCrypto/base-models.git synced 2026-02-06 17:36:52 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

initial core streamline model updates for sl2

Browse Source
This commit is contained in:
drethereum 2024-12-06 14:05:46 +12:00
parent 6974a14ddd
commit e83ca9ae41
57 changed files with 2110 additions and 666 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
models/streamline/bronze/core/bronze__blocks.sql Normal file
View File

@ -0,0 +1,39 @@
{# Set variables #}
{% set source_name = 'BLOCKS' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) or var('GLOBAL_USES_BLOCKS_TRANSACTIONS_PATH', false) else '' %}
{% set model_type = '' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
{# Main query starts here #}
{{ streamline_external_table_query(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

41
models/streamline/bronze/core/bronze__blocks_fr.sql Normal file
View File

@ -0,0 +1,41 @@
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
SELECT
partition_key,
block_number,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__blocks_fr_v2') }}
{% if var('GLOBAL_USES_STREAMLINE_V1', false) %}
UNION ALL
SELECT
_partition_by_block_id AS partition_key,
block_number,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__blocks_fr_v1') }}
{% endif %}
{% if var('GLOBAL_USES_BLOCKS_TRANSACTIONS_PATH', false) %}
UNION ALL
SELECT
partition_key,
block_number,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__blocks_fr_v2_1') }}
{% endif %}

40
models/streamline/bronze/core/bronze__blocks_fr_v1.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'BLOCKS' %}
{% set source_version = '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER)" %}
{% set partition_join_key = '_partition_by_block_id' %}
{% set balances = default_vars['balances'] %}
{% set block_number = false %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core_streamline_v1']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

40
models/streamline/bronze/core/bronze__blocks_fr_v2.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'BLOCKS' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) else '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

39
models/streamline/bronze/core/bronze__confirm_blocks.sql Normal file
View File

@ -0,0 +1,39 @@
{# Set variables #}
{% set source_name = 'CONFIRM_BLOCKS' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) else '' %}
{% set model_type = '' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
{# Main query starts here #}
{{ streamline_external_table_query(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

28
models/streamline/bronze/core/bronze__confirm_blocks_fr.sql Normal file
View File

@ -0,0 +1,28 @@
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
SELECT
partition_key,
block_number,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__confirm_blocks_fr_v2') }}
{% if var('GLOBAL_USES_STREAMLINE_V1', false) %}
UNION ALL
SELECT
_partition_by_block_id AS partition_key,
block_number,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__confirm_blocks_fr_v1') }}
{% endif %}

40
models/streamline/bronze/core/bronze__confirm_blocks_fr_v1.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'CONFIRM_BLOCKS' %}
{% set source_version = '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER)" %}
{% set partition_join_key = '_partition_by_block_id' %}
{% set balances = default_vars['balances'] %}
{% set block_number = false %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core_streamline_v1']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

40
models/streamline/bronze/core/bronze__confirm_blocks_fr_v2.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'CONFIRM_BLOCKS' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) else '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

39
models/streamline/bronze/core/bronze__receipts.sql Normal file
View File

@ -0,0 +1,39 @@
{# Set variables #}
{% set source_name = 'RECEIPTS' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) else '' %}
{% set model_type = '' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_receipts']
) }}
{# Main query starts here #}
{{ streamline_external_table_query(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

30
models/streamline/bronze/core/bronze__receipts_fr.sql Normal file
View File

@ -0,0 +1,30 @@
{{ config (
materialized = 'view',
tags = ['bronze_receipts']
) }}
SELECT
partition_key,
block_number,
array_index,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__receipts_fr_v2') }}
{% if var('GLOBAL_USES_STREAMLINE_V1', false) %}
UNION ALL
SELECT
_partition_by_block_id AS partition_key,
block_number,
VALUE :"array_index" :: INT AS array_index,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__receipts_fr_v1') }}
{% endif %}

40
models/streamline/bronze/core/bronze__receipts_fr_v1.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'RECEIPTS' %}
{% set source_version = '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER)" %}
{% set partition_join_key = '_partition_by_block_id' %}
{% set balances = default_vars['balances'] %}
{% set block_number = false %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core_streamline_v1','bronze_receipts']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

40
models/streamline/bronze/core/bronze__receipts_fr_v2.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'RECEIPTS' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) else '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_receipts']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

11
models/streamline/bronze/core/bronze__streamline_blocks.sql
View File

@ -1,11 +0,0 @@
{{ config (
materialized = 'view'
) }}
{% set model = this.identifier.split("_") [-1] %}
{{ streamline_external_table_query(
model,
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER)",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

9
models/streamline/bronze/core/bronze__streamline_confirm_blocks.sql
View File

@ -1,9 +0,0 @@
{{ config (
materialized = 'view'
) }}
{{ streamline_external_table_query(
model = "confirm_blocks",
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 4), '_', 1) AS INTEGER )",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

11
models/streamline/bronze/core/bronze__streamline_receipts.sql
View File

@ -1,11 +0,0 @@
{{ config (
materialized = 'view'
) }}
{% set model = this.identifier.split("_") [-1] %}
{{ streamline_external_table_query(
model,
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

10
models/streamline/bronze/core/bronze__streamline_traces.sql
View File

@ -1,10 +0,0 @@
{{ config (
materialized = 'view'
) }}
{{ streamline_external_table_query(
model = "debug_traceBlockByNumber",
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

11
models/streamline/bronze/core/bronze__streamline_transactions.sql
View File

@ -1,11 +0,0 @@
{{ config (
materialized = 'view'
) }}
{% set model = this.identifier.split("_") [-1] %}
{{ streamline_external_table_query(
model,
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

39
models/streamline/bronze/core/bronze__traces.sql Normal file
View File

@ -0,0 +1,39 @@
{# Set variables #}
{% set source_name = 'TRACES' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) else '' %}
{% set model_type = '' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
{# Main query starts here #}
{{ streamline_external_table_query(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

30
models/streamline/bronze/core/bronze__traces_fr.sql Normal file
View File

@ -0,0 +1,30 @@
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
SELECT
partition_key,
block_number,
array_index,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__traces_fr_v2') }}
{% if var('GLOBAL_USES_STREAMLINE_V1', false) %}
UNION ALL
SELECT
_partition_by_block_id AS partition_key,
block_number,
VALUE :"array_index" :: INT AS array_index,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__traces_fr_v1') }}
{% endif %}

40
models/streamline/bronze/core/bronze__traces_fr_v1.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'TRACES' %}
{% set source_version = '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER)" %}
{% set partition_join_key = '_partition_by_block_id' %}
{% set balances = default_vars['balances'] %}
{% set block_number = false %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core_streamline_v1']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

40
models/streamline/bronze/core/bronze__traces_fr_v2.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'TRACES' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) else '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

39
models/streamline/bronze/core/bronze__transactions.sql Normal file
View File

@ -0,0 +1,39 @@
{# Set variables #}
{% set source_name = 'TRANSACTIONS' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) or var('GLOBAL_USES_BLOCKS_TRANSACTIONS_PATH', false) else '' %}
{% set model_type = '' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
{# Main query starts here #}
{{ streamline_external_table_query(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

41
models/streamline/bronze/core/bronze__transactions_fr.sql Normal file
View File

@ -0,0 +1,41 @@
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
SELECT
partition_key,
block_number,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__transactions_fr_v2') }}
{% if var('GLOBAL_USES_STREAMLINE_V1', false) %}
UNION ALL
SELECT
_partition_by_block_id AS partition_key,
block_number,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__transactions_fr_v1') }}
{% endif %}
{% if var('GLOBAL_USES_BLOCKS_TRANSACTIONS_PATH', false) %}
UNION ALL
SELECT
partition_key,
block_number,
VALUE,
DATA,
metadata,
file_name,
_inserted_timestamp
FROM
{{ ref('bronze__transactions_fr_v2_1') }}
{% endif %}

40
models/streamline/bronze/core/bronze__transactions_fr_v1.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'TRANSACTIONS' %}
{% set source_version = '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER)" %}
{% set partition_join_key = '_partition_by_block_id' %}
{% set balances = default_vars['balances'] %}
{% set block_number = false %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core_streamline_v1']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

40
models/streamline/bronze/core/bronze__transactions_fr_v2.sql Normal file
View File

@ -0,0 +1,40 @@
{# Set variables #}
{% set source_name = 'TRANSACTIONS' %}
{% set source_version = 'V2' if var('GLOBAL_USES_STREAMLINE_V1', false) else '' %}
{% set model_type = 'FR' %}
{%- set default_vars = set_default_variables_bronze(source_name, model_type) -%}
{% set partition_function = default_vars['partition_function'] %}
{% set partition_join_key = default_vars['partition_join_key'] %}
{% set balances = default_vars['balances'] %}
{% set block_number = default_vars['block_number'] %}
{% set uses_receipts_by_hash = default_vars['uses_receipts_by_hash'] %}
{# Log configuration details #}
{{ log_bronze_details(
source_name = source_name,
source_version = source_version,
model_type = model_type,
partition_function = partition_function,
partition_join_key = partition_join_key,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}
{# Set up dbt configuration #}
{{ config (
materialized = 'view',
tags = ['bronze_core']
) }}
{# Main query starts here #}
{{ streamline_external_table_query_fr(
source_name = source_name.lower(),
source_version = source_version.lower(),
partition_function = partition_function,
partition_join_key = partition_join_key,
balances = balances,
block_number = block_number,
uses_receipts_by_hash = uses_receipts_by_hash
) }}

11
models/streamline/bronze/core/fr/bronze__streamline_fr_blocks.sql
View File

@ -1,11 +0,0 @@
{{ config (
materialized = 'view'
) }}
{% set model = this.identifier.split("_") [-1] %}
{{ streamline_external_table_fr_query(
model,
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER)",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

9
models/streamline/bronze/core/fr/bronze__streamline_fr_confirm_blocks.sql
View File

@ -1,9 +0,0 @@
{{ config (
materialized = 'view'
) }}
{{ streamline_external_table_fr_query(
model = "confirm_blocks",
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 4), '_', 1) AS INTEGER )",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

11
models/streamline/bronze/core/fr/bronze__streamline_fr_receipts.sql
View File

@ -1,11 +0,0 @@
{{ config (
materialized = 'view'
) }}
{% set model = this.identifier.split("_") [-1] %}
{{ streamline_external_table_fr_query(
model,
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

10
models/streamline/bronze/core/fr/bronze__streamline_fr_traces.sql
View File

@ -1,10 +0,0 @@
{{ config (
materialized = 'view'
) }}
{{ streamline_external_table_fr_query(
model = "debug_traceBlockByNumber",
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

11
models/streamline/bronze/core/fr/bronze__streamline_fr_transactions.sql
View File

@ -1,11 +0,0 @@
{{ config (
materialized = 'view'
) }}
{% set model = this.identifier.split("_") [-1] %}
{{ streamline_external_table_fr_query(
model,
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )",
partition_name = "_partition_by_block_id",
unique_key = "block_number"
) }}

23
models/streamline/silver/_24_hour_lookback.sql
View File

@ -1,18 +1,17 @@
{{ config (
materialized = "ephemeral"
{{ config(
materialized = 'ephemeral'
) }}
WITH max_time AS (
SELECT
MAX(block_timestamp) AS max_timestamp
FROM
{{ ref("silver__blocks") }}
)
WITH max_time AS (
SELECT
MAX(block_timestamp) AS max_timestamp
FROM
{{ ref("core__fact_blocks") }}
)
SELECT
MIN(block_number) AS block_number
COALESCE(MIN(block_number), 0) AS block_number
FROM
{{ ref("silver__blocks") }}
{{ ref("core__fact_blocks") }}
JOIN max_time
ON block_timestamp BETWEEN DATEADD(
'hour',
@ -23,4 +22,4 @@ FROM
'hour',
-24,
max_timestamp
)
)

17
models/streamline/silver/_block_lookback.sql
View File

@ -1,11 +1,20 @@
{{ config (
materialized = "ephemeral"
{{ config(
materialized = 'ephemeral'
) }}
{% set uses_lookback = var('GLOBAL_USES_LOOKBACK', True) %}
{% if not uses_lookback %}
SELECT
0 AS block_number
{% else %}
SELECT
MIN(block_number) AS block_number
COALESCE(MIN(block_number), 0) AS block_number
FROM
{{ ref("silver__blocks") }}
{{ ref("core__fact_blocks") }}
WHERE
block_timestamp >= DATEADD('hour', -72, TRUNCATE(SYSDATE(), 'HOUR'))
AND block_timestamp < DATEADD('hour', -71, TRUNCATE(SYSDATE(), 'HOUR'))
{% endif %}

11
models/streamline/silver/_max_block_by_date.sql
View File

@ -1,15 +1,14 @@
{{ config (
materialized = "ephemeral",
unique_key = "block_number",
{{ config(
materialized = 'ephemeral',
unique_key = 'block_number'
) }}
WITH base AS (
SELECT
block_timestamp :: DATE AS block_date,
MAX(block_number) block_number
MAX(block_number) AS block_number
FROM
{{ ref("silver__blocks") }}
{{ ref("core__fact_blocks") }}
GROUP BY
block_timestamp :: DATE
)

11
models/streamline/silver/_max_block_by_hour.sql
View File

@ -1,17 +1,16 @@
{{ config (
materialized = "ephemeral"
{{ config(
materialized = 'ephemeral'
) }}
WITH base AS (
SELECT
DATE_TRUNC(
'hour',
block_timestamp
) AS block_hour,
MAX(block_number) block_number
MAX(block_number) AS block_number
FROM
{{ ref("silver__blocks") }}
{{ ref("core__fact_blocks") }}
WHERE
block_timestamp > DATEADD(
'day',
@ -34,4 +33,4 @@ WHERE
)
FROM
base
)
)

50
models/streamline/silver/core/complete/streamline__blocks_complete.sql Normal file
View File

@ -0,0 +1,50 @@
{# Set variables #}
{%- set source_name = 'BLOCKS' -%}
{%- set model_type = 'COMPLETE' -%}
{%- set full_refresh_type = var((source_name ~ '_complete_full_refresh').upper(), false) -%}
{% set post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(block_number)" %}
{# Log configuration details #}
{{ log_complete_details(
post_hook = post_hook,
full_refresh_type = full_refresh_type
) }}
{# Set up dbt configuration #}
-- depends_on: {{ ref('bronze__' ~ source_name.lower()) }}
{{ config (
materialized = "incremental",
unique_key = "block_number",
cluster_by = "ROUND(block_number, -3)",
post_hook = post_hook,
full_refresh = full_refresh_type,
tags = ['streamline_core_complete']
) }}
{# Main query starts here #}
SELECT
block_number,
file_name,
{{ dbt_utils.generate_surrogate_key(['block_number']) }} AS complete_{{ source_name.lower() }}_id,
SYSDATE() AS inserted_timestamp,
SYSDATE() AS modified_timestamp,
_inserted_timestamp,
'{{ invocation_id }}' AS _invocation_id
FROM
{% if is_incremental() %}
{{ ref('bronze__' ~ source_name.lower()) }}
WHERE
_inserted_timestamp >= (
SELECT
COALESCE(MAX(_inserted_timestamp), '1970-01-01'::TIMESTAMP) AS _inserted_timestamp
FROM
{{ this }}
)
{% else %}
{{ ref('bronze__' ~ source_name.lower() ~ '_fr') }}
{% endif %}
QUALIFY (ROW_NUMBER() OVER (PARTITION BY block_number ORDER BY _inserted_timestamp DESC)) = 1

29
models/streamline/silver/core/complete/streamline__complete_confirmed_blocks.sql
View File

@ -1,29 +0,0 @@
-- depends_on: {{ ref('bronze__streamline_confirm_blocks') }}
{{ config (
materialized = "incremental",
unique_key = "id",
cluster_by = "ROUND(block_number, -3)",
tags = ['streamline_core_complete']
) }}
SELECT
id,
block_number,
_inserted_timestamp
FROM
{% if is_incremental() %}
{{ ref('bronze__streamline_confirm_blocks') }}
WHERE
_inserted_timestamp >= (
SELECT
COALESCE(MAX(_inserted_timestamp), '1970-01-01' :: TIMESTAMP) _inserted_timestamp
FROM
{{ this }})
{% else %}
{{ ref('bronze__streamline_fr_confirm_blocks') }}
{% endif %}
qualify(ROW_NUMBER() over (PARTITION BY id
ORDER BY
_inserted_timestamp DESC)) = 1

33
models/streamline/silver/core/complete/streamline__complete_debug_traceBlockByNumber.sql
View File

@ -1,33 +0,0 @@
-- depends_on: {{ ref('bronze__streamline_traces') }}
{{ config (
materialized = "incremental",
unique_key = "id",
cluster_by = "ROUND(block_number, -3)",
merge_update_columns = ["id"],
post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(id)",
tags = ['streamline_core_complete']
) }}
SELECT
id,
block_number,
file_name,
_inserted_timestamp
FROM
{% if is_incremental() %}
{{ ref('bronze__streamline_traces') }}
WHERE
_inserted_timestamp >= (
SELECT
MAX(_inserted_timestamp) _inserted_timestamp
FROM
{{ this }}
)
{% else %}
{{ ref('bronze__streamline_fr_traces') }}
{% endif %}
qualify(ROW_NUMBER() over (PARTITION BY id
ORDER BY
_inserted_timestamp DESC)) = 1

33
models/streamline/silver/core/complete/streamline__complete_qn_getBlockWithReceipts.sql
View File

@ -1,33 +0,0 @@
-- depends_on: {{ ref('bronze__streamline_blocks') }}
{{ config (
materialized = "incremental",
unique_key = "id",
cluster_by = "ROUND(block_number, -3)",
merge_update_columns = ["id"],
post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(id)",
tags = ['streamline_core_complete']
) }}
SELECT
id,
block_number,
file_name,
_inserted_timestamp
FROM
{% if is_incremental() %}
{{ ref('bronze__streamline_blocks') }}
WHERE
_inserted_timestamp >= (
SELECT
MAX(_inserted_timestamp) _inserted_timestamp
FROM
{{ this }}
)
{% else %}
{{ ref('bronze__streamline_fr_blocks') }}
{% endif %}
qualify(ROW_NUMBER() over (PARTITION BY id
ORDER BY
_inserted_timestamp DESC)) = 1

50
models/streamline/silver/core/complete/streamline__confirm_blocks_complete.sql Normal file
View File

@ -0,0 +1,50 @@
{# Set variables #}
{%- set source_name = 'CONFIRM_BLOCKS' -%}
{%- set model_type = 'COMPLETE' -%}
{%- set full_refresh_type = var((source_name ~ '_complete_full_refresh').upper(), false) -%}
{% set post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(block_number)" %}
{# Log configuration details #}
{{ log_complete_details(
post_hook = post_hook,
full_refresh_type = full_refresh_type
) }}
{# Set up dbt configuration #}
-- depends_on: {{ ref('bronze__' ~ source_name.lower()) }}
{{ config (
materialized = "incremental",
unique_key = "block_number",
cluster_by = "ROUND(block_number, -3)",
post_hook = post_hook,
full_refresh = full_refresh_type,
tags = ['streamline_core_complete_confirm_blocks']
) }}
{# Main query starts here #}
SELECT
block_number,
file_name,
{{ dbt_utils.generate_surrogate_key(['block_number']) }} AS complete_{{ source_name.lower() }}_id,
SYSDATE() AS inserted_timestamp,
SYSDATE() AS modified_timestamp,
_inserted_timestamp,
'{{ invocation_id }}' AS _invocation_id
FROM
{% if is_incremental() %}
{{ ref('bronze__' ~ source_name.lower()) }}
WHERE
_inserted_timestamp >= (
SELECT
COALESCE(MAX(_inserted_timestamp), '1970-01-01'::TIMESTAMP) AS _inserted_timestamp
FROM
{{ this }}
)
{% else %}
{{ ref('bronze__' ~ source_name.lower() ~ '_fr') }}
{% endif %}
QUALIFY (ROW_NUMBER() OVER (PARTITION BY block_number ORDER BY _inserted_timestamp DESC)) = 1

50
models/streamline/silver/core/complete/streamline__receipts_complete.sql Normal file
View File

@ -0,0 +1,50 @@
{# Set variables #}
{%- set source_name = 'RECEIPTS' -%}
{%- set model_type = 'COMPLETE' -%}
{%- set full_refresh_type = var((source_name ~ '_complete_full_refresh').upper(), false) -%}
{% set post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(block_number)" %}
{# Log configuration details #}
{{ log_complete_details(
post_hook = post_hook,
full_refresh_type = full_refresh_type
) }}
{# Set up dbt configuration #}
-- depends_on: {{ ref('bronze__' ~ source_name.lower()) }}
{{ config (
materialized = "incremental",
unique_key = "block_number",
cluster_by = "ROUND(block_number, -3)",
post_hook = post_hook,
full_refresh = full_refresh_type,
tags = ['streamline_core_complete_receipts']
) }}
{# Main query starts here #}
SELECT
block_number,
file_name,
{{ dbt_utils.generate_surrogate_key(['block_number']) }} AS complete_{{ source_name.lower() }}_id,
SYSDATE() AS inserted_timestamp,
SYSDATE() AS modified_timestamp,
_inserted_timestamp,
'{{ invocation_id }}' AS _invocation_id
FROM
{% if is_incremental() %}
{{ ref('bronze__' ~ source_name.lower()) }}
WHERE
_inserted_timestamp >= (
SELECT
COALESCE(MAX(_inserted_timestamp), '1970-01-01'::TIMESTAMP) AS _inserted_timestamp
FROM
{{ this }}
)
{% else %}
{{ ref('bronze__' ~ source_name.lower() ~ '_fr') }}
{% endif %}
QUALIFY (ROW_NUMBER() OVER (PARTITION BY block_number ORDER BY _inserted_timestamp DESC)) = 1

50
models/streamline/silver/core/complete/streamline__traces_complete.sql Normal file
View File

@ -0,0 +1,50 @@
{# Set variables #}
{%- set source_name = 'TRACES' -%}
{%- set model_type = 'COMPLETE' -%}
{%- set full_refresh_type = var((source_name ~ '_complete_full_refresh').upper(), false) -%}
{% set post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(block_number)" %}
{# Log configuration details #}
{{ log_complete_details(
post_hook = post_hook,
full_refresh_type = full_refresh_type
) }}
{# Set up dbt configuration #}
-- depends_on: {{ ref('bronze__' ~ source_name.lower()) }}
{{ config (
materialized = "incremental",
unique_key = "block_number",
cluster_by = "ROUND(block_number, -3)",
post_hook = post_hook,
full_refresh = full_refresh_type,
tags = ['streamline_core_complete']
) }}
{# Main query starts here #}
SELECT
block_number,
file_name,
{{ dbt_utils.generate_surrogate_key(['block_number']) }} AS complete_{{ source_name.lower() }}_id,
SYSDATE() AS inserted_timestamp,
SYSDATE() AS modified_timestamp,
_inserted_timestamp,
'{{ invocation_id }}' AS _invocation_id
FROM
{% if is_incremental() %}
{{ ref('bronze__' ~ source_name.lower()) }}
WHERE
_inserted_timestamp >= (
SELECT
COALESCE(MAX(_inserted_timestamp), '1970-01-01'::TIMESTAMP) AS _inserted_timestamp
FROM
{{ this }}
)
{% else %}
{{ ref('bronze__' ~ source_name.lower() ~ '_fr') }}
{% endif %}
QUALIFY (ROW_NUMBER() OVER (PARTITION BY block_number ORDER BY _inserted_timestamp DESC)) = 1

50
models/streamline/silver/core/complete/streamline__transactions_complete.sql Normal file
View File

@ -0,0 +1,50 @@
{# Set variables #}
{%- set source_name = 'TRANSACTIONS' -%}
{%- set model_type = 'COMPLETE' -%}
{%- set full_refresh_type = var((source_name ~ '_complete_full_refresh').upper(), false) -%}
{% set post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(block_number)"%}
{# Log configuration details #}
{{ log_complete_details(
post_hook = post_hook,
full_refresh_type = full_refresh_type
) }}
{# Set up dbt configuration #}
-- depends_on: {{ ref('bronze__' ~ source_name.lower()) }}
{{ config (
materialized = "incremental",
unique_key = "block_number",
cluster_by = "ROUND(block_number, -3)",
post_hook = post_hook,
full_refresh = full_refresh_type,
tags = ['streamline_core_complete']
) }}
{# Main query starts here #}
SELECT
block_number,
file_name,
{{ dbt_utils.generate_surrogate_key(['block_number']) }} AS complete_{{ source_name.lower() }}_id,
SYSDATE() AS inserted_timestamp,
SYSDATE() AS modified_timestamp,
_inserted_timestamp,
'{{ invocation_id }}' AS _invocation_id
FROM
{% if is_incremental() %}
{{ ref('bronze__' ~ source_name.lower()) }}
WHERE
_inserted_timestamp >= (
SELECT
COALESCE(MAX(_inserted_timestamp), '1970-01-01'::TIMESTAMP) AS _inserted_timestamp
FROM
{{ this }}
)
{% else %}
{{ ref('bronze__' ~ source_name.lower() ~ '_fr') }}
{% endif %}
QUALIFY (ROW_NUMBER() OVER (PARTITION BY block_number ORDER BY _inserted_timestamp DESC)) = 1

112
models/streamline/silver/core/history/streamline__blocks_transactions_history.sql Normal file
View File

@ -0,0 +1,112 @@
{# Set variables #}
{%- set model_name = 'BLOCKS_TRANSACTIONS' -%}
{%- set model_type = 'HISTORY' -%}
{%- set default_vars = set_default_variables_streamline(model_name, model_type) -%}
{# Set up parameters for the streamline process. These will come from the vars set in dbt_project.yml #}
{%- set streamline_params = set_streamline_parameters(
model_name=model_name,
model_type=model_type
) -%}
{%- set node_url = default_vars['node_url'] -%}
{%- set node_secret_path = default_vars['node_secret_path'] -%}
{%- set model_quantum_state = default_vars['model_quantum_state'] -%}
{%- set sql_limit = streamline_params['sql_limit'] -%}
{%- set testing_limit = default_vars['testing_limit'] -%}
{%- set order_by_clause = default_vars['order_by_clause'] -%}
{%- set new_build = default_vars['new_build'] -%}
{%- set method_params = streamline_params['method_params'] -%}
{%- set method = streamline_params['method'] -%}
{# Log configuration details #}
{{ log_streamline_details(
model_name=model_name,
model_type=model_type,
node_url=node_url,
model_quantum_state=model_quantum_state,
sql_limit=sql_limit,
testing_limit=testing_limit,
order_by_clause=order_by_clause,
new_build=new_build,
streamline_params=streamline_params,
method_params=method_params,
method=method
) }}
{# Set up dbt configuration #}
{{ 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 = streamline_params
),
tags = ['streamline_core_history']
) }}
{# Main query starts here #}
WITH
{% if not new_build %}
last_3_days AS (
SELECT block_number
FROM {{ ref("_block_lookback") }}
),
{% endif %}
{# Identify blocks that need processing #}
to_do AS (
SELECT block_number
FROM {{ ref("streamline__blocks") }}
WHERE
block_number IS NOT NULL
{% if not new_build %}
AND block_number <= (SELECT block_number FROM last_3_days)
{% endif %}
EXCEPT
SELECT block_number
FROM {{ ref("streamline__blocks_complete") }} b
INNER JOIN {{ ref("streamline__transactions_complete") }} t USING(block_number)
WHERE 1=1
{% if not new_build %}
AND block_number <= (SELECT block_number FROM last_3_days)
{% endif %}
),
ready_blocks AS (
SELECT block_number
FROM to_do
{% if testing_limit is not none %}
LIMIT {{ testing_limit }}
{% endif %}
)
{# Generate API requests for each block #}
SELECT
block_number,
ROUND(block_number, -3) AS partition_key,
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id', block_number,
'jsonrpc', '2.0',
'method', '{{ method }}',
'params', {{ method_params }}
),
'{{ node_secret_path }}'
) AS request
FROM
ready_blocks
{{ order_by_clause }}
LIMIT {{ sql_limit }}

133
models/streamline/silver/core/history/streamline__confirm_blocks_history.sql Normal file
View File

@ -0,0 +1,133 @@
{# Set variables #}
{%- set model_name = 'CONFIRM_BLOCKS' -%}
{%- set model_type = 'HISTORY' -%}
{%- set default_vars = set_default_variables_streamline(model_name, model_type) -%}
{# Set up parameters for the streamline process. These will come from the vars set in dbt_project.yml #}
{%- set streamline_params = set_streamline_parameters(
model_name=model_name,
model_type=model_type
) -%}
{%- set node_url = default_vars['node_url'] -%}
{%- set node_secret_path = default_vars['node_secret_path'] -%}
{%- set model_quantum_state = default_vars['model_quantum_state'] -%}
{%- set sql_limit = streamline_params['sql_limit'] -%}
{%- set testing_limit = default_vars['testing_limit'] -%}
{%- set order_by_clause = default_vars['order_by_clause'] -%}
{%- set new_build = default_vars['new_build'] -%}
{%- set method_params = streamline_params['method_params'] -%}
{%- set method = streamline_params['method'] -%}
{# Log configuration details #}
{{ log_streamline_details(
model_name=model_name,
model_type=model_type,
node_url=node_url,
model_quantum_state=model_quantum_state,
sql_limit=sql_limit,
testing_limit=testing_limit,
order_by_clause=order_by_clause,
new_build=new_build,
streamline_params=streamline_params,
method_params=method_params,
method=method
) }}
{# Set up dbt configuration #}
{{ 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 = streamline_params
),
tags = ['streamline_core_history_confirm_blocks']
) }}
{# Main query starts here #}
WITH
{% if not new_build %}
last_3_days AS (
SELECT block_number
FROM {{ ref("_block_lookback") }}
),
{% endif %}
{# Delay blocks #}
look_back AS (
SELECT
block_number
FROM
{{ ref("_max_block_by_hour") }}
qualify ROW_NUMBER() over (
ORDER BY
block_number DESC
) = 6
),
{# Identify blocks that need processing #}
to_do AS (
SELECT block_number
FROM {{ ref("streamline__blocks") }}
WHERE
block_number IS NOT NULL
AND block_number <= (SELECT block_number FROM look_back)
{% if not new_build %}
AND block_number <= (SELECT block_number FROM last_3_days)
{% endif %}
EXCEPT
{# Exclude blocks that have already been processed #}
SELECT block_number
FROM {{ ref('streamline__' ~ model_name.lower() ~ '_complete') }}
WHERE 1=1
AND block_number IS NOT NULL
AND block_number <= (SELECT block_number FROM look_back)
AND _inserted_timestamp >= DATEADD(
'day',
-4,
SYSDATE()
)
{% if not new_build %}
AND block_number <= (SELECT block_number FROM last_3_days)
{% endif %}
)
{# Prepare the final list of blocks to process #}
,ready_blocks AS (
SELECT block_number
FROM to_do
{% if testing_limit is not none %}
LIMIT {{ testing_limit }}
{% endif %}
)
{# Generate API requests for each block #}
SELECT
block_number,
ROUND(block_number, -3) AS partition_key,
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id', block_number,
'jsonrpc', '2.0',
'method', '{{ method }}',
'params', {{ method_params }}
),
'{{ node_secret_path }}'
) AS request
FROM
ready_blocks
{{ order_by_clause }}
LIMIT {{ sql_limit }}

72
models/streamline/silver/core/history/streamline__debug_traceBlockByNumber_history.sql
View File

@ -1,72 +0,0 @@
{{ config (
materialized = "view",
post_hook = if_data_call_function(
func = "{{this.schema}}.udf_bulk_get_traces(object_construct('sql_source', '{{this.identifier}}','exploded_key','[\"result\"]', 'method', 'debug_traceBlockByNumber', 'external_table', 'debug_traceBlockByNumber', 'sql_limit', {{var('sql_limit','200000')}}, 'producer_batch_size', {{var('producer_batch_size','200000')}}, 'worker_batch_size', {{var('worker_batch_size','100000')}}, 'batch_call_limit', {{var('batch_call_limit','1')}}))",
target = "{{this.schema}}.{{this.identifier}}"
),
tags = ['streamline_core_history']
) }}
WITH last_3_days AS (
SELECT
block_number
FROM
{{ ref("_max_block_by_date") }}
qualify ROW_NUMBER() over (
ORDER BY
block_number DESC
) = 3
),
blocks AS (
SELECT
block_number
FROM
{{ ref("streamline__blocks") }}
WHERE
block_number <= (
SELECT
block_number
FROM
last_3_days
)
EXCEPT
SELECT
block_number
FROM
{{ ref("streamline__complete_debug_traceBlockByNumber") }}
WHERE
block_number <= (
SELECT
block_number
FROM
last_3_days
)
)
SELECT
PARSE_JSON(
CONCAT(
'{"jsonrpc": "2.0",',
'"method": "debug_traceBlockByNumber", "params":["',
REPLACE(
concat_ws(
'',
'0x',
to_char(
block_number :: INTEGER,
'XXXXXXXX'
)
),
' ',
''
),
'",{"tracer": "callTracer","timeout": "30s"}',
'],"id":"',
block_number :: INTEGER,
'"}'
)
) AS request
FROM
blocks
ORDER BY
block_number ASC

71
models/streamline/silver/core/history/streamline__qn_getBlockWithReceipts_history.sql
View File

@ -1,71 +0,0 @@
{{ config (
materialized = "view",
post_hook = if_data_call_function(
func = "{{this.schema}}.udf_bulk_json_rpc(object_construct('sql_source', '{{this.identifier}}', 'external_table', 'qn_getBlockWithReceipts', 'sql_limit', {{var('sql_limit','200000')}}, 'producer_batch_size', {{var('producer_batch_size','200000')}}, 'worker_batch_size', {{var('worker_batch_size','100000')}}, 'batch_call_limit', {{var('batch_call_limit','1')}}))",
target = "{{this.schema}}.{{this.identifier}}"
),
tags = ['streamline_core_history']
) }}
WITH last_3_days AS (
SELECT
block_number
FROM
{{ ref("_max_block_by_date") }}
qualify ROW_NUMBER() over (
ORDER BY
block_number DESC
) = 3
),
blocks AS (
SELECT
block_number :: STRING AS block_number
FROM
{{ ref("streamline__blocks") }}
WHERE
block_number <= (
SELECT
block_number
FROM
last_3_days
)
EXCEPT
SELECT
block_number :: STRING
FROM
{{ ref("streamline__complete_qn_getBlockWithReceipts") }}
WHERE
block_number <= (
SELECT
block_number
FROM
last_3_days
)
)
SELECT
PARSE_JSON(
CONCAT(
'{"jsonrpc": "2.0",',
'"method": "qn_getBlockWithReceipts", "params":["',
REPLACE(
concat_ws(
'',
'0x',
to_char(
block_number :: INTEGER,
'XXXXXXXX'
)
),
' ',
''
),
'"],"id":"',
block_number :: INTEGER,
'"}'
)
) AS request
FROM
blocks
ORDER BY
block_number ASC

113
models/streamline/silver/core/history/streamline__receipts_history.sql Normal file
View File

@ -0,0 +1,113 @@
{# Set variables #}
{%- set model_name = 'RECEIPTS' -%}
{%- set model_type = 'HISTORY' -%}
{%- set default_vars = set_default_variables_streamline(model_name, model_type) -%}
{# Set up parameters for the streamline process. These will come from the vars set in dbt_project.yml #}
{%- set streamline_params = set_streamline_parameters(
model_name=model_name,
model_type=model_type
) -%}
{%- set node_url = default_vars['node_url'] -%}
{%- set node_secret_path = default_vars['node_secret_path'] -%}
{%- set model_quantum_state = default_vars['model_quantum_state'] -%}
{%- set sql_limit = streamline_params['sql_limit'] -%}
{%- set testing_limit = default_vars['testing_limit'] -%}
{%- set order_by_clause = default_vars['order_by_clause'] -%}
{%- set new_build = default_vars['new_build'] -%}
{%- set method_params = streamline_params['method_params'] -%}
{%- set method = streamline_params['method'] -%}
{# Log configuration details #}
{{ log_streamline_details(
model_name=model_name,
model_type=model_type,
node_url=node_url,
model_quantum_state=model_quantum_state,
sql_limit=sql_limit,
testing_limit=testing_limit,
order_by_clause=order_by_clause,
new_build=new_build,
streamline_params=streamline_params,
method_params=method_params,
method=method
) }}
{# Set up dbt configuration #}
{{ 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 = streamline_params
),
tags = ['streamline_core_history_receipts']
) }}
{# Main query starts here #}
WITH
{% if not new_build %}
last_3_days AS (
SELECT block_number
FROM {{ ref("_block_lookback") }}
),
{% endif %}
{# Identify blocks that need processing #}
to_do AS (
SELECT block_number
FROM {{ ref("streamline__blocks") }}
WHERE
block_number IS NOT NULL
{% if not new_build %}
AND block_number <= (SELECT block_number FROM last_3_days)
{% endif %}
EXCEPT
{# Exclude blocks that have already been processed #}
SELECT block_number
FROM {{ ref('streamline__' ~ model_name.lower() ~ '_complete') }}
WHERE 1=1
{% if not new_build %}
AND block_number <= (SELECT block_number FROM last_3_days)
{% endif %}
)
{# Prepare the final list of blocks to process #}
,ready_blocks AS (
SELECT block_number
FROM to_do
{% if testing_limit is not none %}
LIMIT {{ testing_limit }}
{% endif %}
)
{# Generate API requests for each block #}
SELECT
block_number,
ROUND(block_number, -3) AS partition_key,
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id', block_number,
'jsonrpc', '2.0',
'method', '{{ method }}',
'params', {{ method_params }}
),
'{{ node_secret_path }}'
) AS request
FROM
ready_blocks
{{ order_by_clause }}
LIMIT {{ sql_limit }}

113
models/streamline/silver/core/history/streamline__traces_history.sql Normal file
View File

@ -0,0 +1,113 @@
{# Set variables #}
{%- set model_name = 'TRACES' -%}
{%- set model_type = 'HISTORY' -%}
{%- set default_vars = set_default_variables_streamline(model_name, model_type) -%}
{# Set up parameters for the streamline process. These will come from the vars set in dbt_project.yml #}
{%- set streamline_params = set_streamline_parameters(
model_name=model_name,
model_type=model_type
) -%}
{%- set node_url = default_vars['node_url'] -%}
{%- set node_secret_path = default_vars['node_secret_path'] -%}
{%- set model_quantum_state = default_vars['model_quantum_state'] -%}
{%- set sql_limit = streamline_params['sql_limit'] -%}
{%- set testing_limit = default_vars['testing_limit'] -%}
{%- set order_by_clause = default_vars['order_by_clause'] -%}
{%- set new_build = default_vars['new_build'] -%}
{%- set method_params = streamline_params['method_params'] -%}
{%- set method = streamline_params['method'] -%}
{# Log configuration details #}
{{ log_streamline_details(
model_name=model_name,
model_type=model_type,
node_url=node_url,
model_quantum_state=model_quantum_state,
sql_limit=sql_limit,
testing_limit=testing_limit,
order_by_clause=order_by_clause,
new_build=new_build,
streamline_params=streamline_params,
method_params=method_params,
method=method
) }}
{# Set up dbt configuration #}
{{ 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 = streamline_params
),
tags = ['streamline_core_history']
) }}
{# Main query starts here #}
WITH
{% if not new_build %}
last_3_days AS (
SELECT block_number
FROM {{ ref("_block_lookback") }}
),
{% endif %}
{# Identify blocks that need processing #}
to_do AS (
SELECT block_number
FROM {{ ref("streamline__blocks") }}
WHERE
block_number IS NOT NULL
{% if not new_build %}
AND block_number <= (SELECT block_number FROM last_3_days)
{% endif %}
EXCEPT
{# Exclude blocks that have already been processed #}
SELECT block_number
FROM {{ ref('streamline__' ~ model_name.lower() ~ '_complete') }}
WHERE 1=1
{% if not new_build %}
AND block_number <= (SELECT block_number FROM last_3_days)
{% endif %}
)
{# Prepare the final list of blocks to process #}
,ready_blocks AS (
SELECT block_number
FROM to_do
{% if testing_limit is not none %}
LIMIT {{ testing_limit }}
{% endif %}
)
{# Generate API requests for each block #}
SELECT
block_number,
ROUND(block_number, -3) AS partition_key,
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id', block_number,
'jsonrpc', '2.0',
'method', '{{ method }}',
'params', {{ method_params }}
),
'{{ node_secret_path }}'
) AS request
FROM
ready_blocks
{{ order_by_clause }}
LIMIT {{ sql_limit }}

126
models/streamline/silver/core/realtime/streamline__blocks_transactions_realtime.sql Normal file
View File

@ -0,0 +1,126 @@
{# Set variables #}
{%- set model_name = 'BLOCKS_TRANSACTIONS' -%}
{%- set model_type = 'REALTIME' -%}
{%- set min_block = var('GLOBAL_START_UP_BLOCK', none) -%}
{%- set default_vars = set_default_variables_streamline(model_name, model_type) -%}
{# Set up parameters for the streamline process. These will come from the vars set in dbt_project.yml #}
{%- set streamline_params = set_streamline_parameters(
model_name=model_name,
model_type=model_type
) -%}
{%- set node_url = default_vars['node_url'] -%}
{%- set node_secret_path = default_vars['node_secret_path'] -%}
{%- set model_quantum_state = default_vars['model_quantum_state'] -%}
{%- set sql_limit = streamline_params['sql_limit'] -%}
{%- set testing_limit = default_vars['testing_limit'] -%}
{%- set order_by_clause = default_vars['order_by_clause'] -%}
{%- set new_build = default_vars['new_build'] -%}
{%- set method_params = streamline_params['method_params'] -%}
{%- set method = streamline_params['method'] -%}
{# Log configuration details #}
{{ log_streamline_details(
model_name=model_name,
model_type=model_type,
node_url=node_url,
model_quantum_state=model_quantum_state,
sql_limit=sql_limit,
testing_limit=testing_limit,
order_by_clause=order_by_clause,
new_build=new_build,
streamline_params=streamline_params,
method_params=method_params,
method=method,
min_block=min_block
) }}
{# Set up dbt configuration #}
{{ 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 = streamline_params
),
tags = ['streamline_core_realtime']
) }}
{# Main query starts here #}
WITH
{% if not new_build %}
last_3_days AS (
SELECT block_number
FROM {{ ref("_block_lookback") }}
),
{% endif %}
{# Identify blocks that need processing #}
to_do AS (
SELECT block_number
FROM {{ ref("streamline__blocks") }}
WHERE
block_number IS NOT NULL
{% if not new_build %}
AND block_number >= (SELECT block_number FROM last_3_days)
{% endif %}
{% if min_block is not none %}
AND block_number >= {{ min_block }}
{% endif %}
EXCEPT
SELECT block_number
FROM {{ ref("streamline__blocks_complete") }} b
INNER JOIN {{ ref("streamline__transactions_complete") }} t USING(block_number)
WHERE 1=1
{% if not new_build %}
AND block_number >= (SELECT block_number FROM last_3_days)
{% endif %}
),
ready_blocks AS (
SELECT block_number
FROM to_do
{% if not new_build %}
UNION
SELECT block_number
FROM {{ ref("_unconfirmed_blocks") }}
UNION
SELECT block_number
FROM {{ ref("_missing_txs") }}
{% endif %}
{% if testing_limit is not none %}
LIMIT {{ testing_limit }}
{% endif %}
)
{# Generate API requests for each block #}
SELECT
block_number,
ROUND(block_number, -3) AS partition_key,
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id', block_number,
'jsonrpc', '2.0',
'method', '{{ method }}',
'params', {{ method_params }}
),
'{{ node_secret_path }}'
) AS request
FROM
ready_blocks
{{ order_by_clause }}
LIMIT {{ sql_limit }}

200
models/streamline/silver/core/realtime/streamline__confirm_blocks_realtime.sql
View File

@ -1,100 +1,138 @@
{{ config (
materialized = "view",
post_hook = if_data_call_function(
func = "{{this.schema}}.udf_bulk_json_rpc(object_construct('sql_source', '{{this.identifier}}', 'external_table', 'confirm_blocks', 'sql_limit', {{var('sql_limit','2000')}}, 'producer_batch_size', {{var('producer_batch_size','2000')}}, 'worker_batch_size', {{var('worker_batch_size','1000')}}, 'batch_call_limit', {{var('batch_call_limit','3')}}))",
target = "{{this.schema}}.{{this.identifier}}"
),
tags = ['streamline_core_realtime']
{# Set variables #}
{%- set model_name = 'CONFIRM_BLOCKS' -%}
{%- set model_type = 'REALTIME' -%}
{%- set min_block = var('GLOBAL_START_UP_BLOCK', none) -%}
{%- set default_vars = set_default_variables_streamline(model_name, model_type) -%}
{# Set up parameters for the streamline process. These will come from the vars set in dbt_project.yml #}
{%- set streamline_params = set_streamline_parameters(
model_name=model_name,
model_type=model_type
) -%}
{%- set node_url = default_vars['node_url'] -%}
{%- set node_secret_path = default_vars['node_secret_path'] -%}
{%- set model_quantum_state = default_vars['model_quantum_state'] -%}
{%- set sql_limit = streamline_params['sql_limit'] -%}
{%- set testing_limit = default_vars['testing_limit'] -%}
{%- set order_by_clause = default_vars['order_by_clause'] -%}
{%- set new_build = default_vars['new_build'] -%}
{%- set method_params = streamline_params['method_params'] -%}
{%- set method = streamline_params['method'] -%}
{# Log configuration details #}
{{ log_streamline_details(
model_name=model_name,
model_type=model_type,
node_url=node_url,
model_quantum_state=model_quantum_state,
sql_limit=sql_limit,
testing_limit=testing_limit,
order_by_clause=order_by_clause,
new_build=new_build,
streamline_params=streamline_params,
method_params=method_params,
method=method,
min_block=min_block
) }}
WITH last_3_days AS (
{# Set up dbt configuration #}
{{ 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 = streamline_params
),
tags = ['streamline_core_realtime_confirm_blocks']
) }}
{# Main query starts here #}
WITH
{% if not new_build %}
last_3_days AS (
SELECT block_number
FROM {{ ref("_block_lookback") }}
),
{% endif %}
{# Delay blocks #}
look_back AS (
SELECT
block_number
FROM
{{ ref("_block_lookback") }}
),
last_3_days AS (
SELECT
block_number
FROM
{{ ref("_max_block_by_date") }}
{{ ref("_max_block_by_hour") }}
qualify ROW_NUMBER() over (
ORDER BY
block_number DESC
) = 3
),
tbl AS (
SELECT
block_number
FROM
{{ ref("streamline__blocks") }}
WHERE
(
block_number >= (
SELECT
block_number
FROM
last_3_days
)
)
AND block_number >= (
SELECT
block_number
FROM
last_3_days
)
) = 6
),
{# Identify blocks that need processing #}
to_do AS (
SELECT block_number
FROM {{ ref("streamline__blocks") }}
WHERE
block_number IS NOT NULL
AND block_number <= (SELECT block_number FROM look_back)
{% if not new_build %}
AND block_number >= (SELECT block_number FROM last_3_days)
{% endif %}
{% if min_block is not none %}
AND block_number >= {{ min_block }}
{% endif %}
EXCEPT
SELECT
block_number
FROM
{{ ref("streamline__complete_confirmed_blocks") }}
WHERE
(
block_number >= (
SELECT
block_number
FROM
last_3_days
)
)
{# Exclude blocks that have already been processed #}
SELECT block_number
FROM {{ ref('streamline__' ~ model_name.lower() ~ '_complete') }}
WHERE 1=1
AND block_number IS NOT NULL
AND block_number <= (SELECT block_number FROM look_back)
AND _inserted_timestamp >= DATEADD(
'day',
-4,
SYSDATE()
)
AND block_number >= (
SELECT
block_number
FROM
last_3_days
)
{% if not new_build %}
AND block_number >= (SELECT block_number FROM last_3_days)
{% endif %}
)
{# Prepare the final list of blocks to process #}
,ready_blocks AS (
SELECT block_number
FROM to_do
{% if testing_limit is not none %}
LIMIT {{ testing_limit }}
{% endif %}
)
{# Generate API requests for each block #}
SELECT
PARSE_JSON(
CONCAT(
'{"jsonrpc": "2.0",',
'"method": "eth_getBlockByNumber", "params":["',
REPLACE(
concat_ws(
'',
'0x',
to_char(
block_number :: INTEGER,
'XXXXXXXX'
)
),
' ',
''
),
'", false],"id":"',
block_number :: INTEGER,
'"}'
)
block_number,
ROUND(block_number, -3) AS partition_key,
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id', block_number,
'jsonrpc', '2.0',
'method', '{{ method }}',
'params', {{ method_params }}
),
'{{ node_secret_path }}'
) AS request
FROM
tbl
ORDER BY
block_number ASC
LIMIT
20000
ready_blocks
{{ order_by_clause }}
LIMIT {{ sql_limit }}

100
models/streamline/silver/core/realtime/streamline__debug_traceBlockByNumber_realtime.sql
View File

@ -1,100 +0,0 @@
{{ config (
materialized = "view",
post_hook = if_data_call_function(
func = "{{this.schema}}.udf_bulk_get_traces(object_construct('sql_source', '{{this.identifier}}', 'external_table', 'debug_traceBlockByNumber', 'sql_limit', {{var('sql_limit','4000')}}, 'producer_batch_size', {{var('producer_batch_size','4000')}}, 'worker_batch_size', {{var('worker_batch_size','1000')}}, 'batch_call_limit', {{var('batch_call_limit','1')}}))",
target = "{{this.schema}}.{{this.identifier}}"
),
tags = ['streamline_core_realtime']
) }}
WITH last_3_days AS (
SELECT
block_number
FROM
{{ ref("_block_lookback") }}
),
blocks AS (
SELECT
block_number
FROM
{{ ref("streamline__blocks") }}
WHERE
(
block_number >= (
SELECT
block_number
FROM
last_3_days
)
)
EXCEPT
SELECT
block_number
FROM
{{ ref("streamline__complete_debug_traceBlockByNumber") }}
WHERE
(
block_number >= (
SELECT
block_number
FROM
last_3_days
)
)
AND _inserted_timestamp >= DATEADD(
'day',
-4,
SYSDATE()
)
),
all_blocks AS (
SELECT
block_number
FROM
blocks
UNION
SELECT
block_number
FROM
(
SELECT
block_number
FROM
{{ ref("_missing_traces") }}
UNION
SELECT
block_number
FROM
{{ ref("_unconfirmed_blocks") }}
)
)
SELECT
PARSE_JSON(
CONCAT(
'{"jsonrpc": "2.0",',
'"method": "debug_traceBlockByNumber", "params":["',
REPLACE(
concat_ws(
'',
'0x',
to_char(
block_number :: INTEGER,
'XXXXXXXX'
)
),
' ',
''
),
'",{"tracer": "callTracer", "timeout": "30s"}',
'],"id":"',
block_number :: INTEGER,
'"}'
)
) AS request
FROM
all_blocks
ORDER BY
block_number ASC
LIMIT
10000

104
models/streamline/silver/core/realtime/streamline__qn_getBlockWithReceipts_realtime.sql
View File

@ -1,104 +0,0 @@
{{ config (
materialized = "view",
post_hook = if_data_call_function(
func = "{{this.schema}}.udf_bulk_json_rpc(object_construct('sql_source', '{{this.identifier}}', 'external_table', 'qn_getBlockWithReceipts', 'sql_limit', {{var('sql_limit','2000')}}, 'producer_batch_size', {{var('producer_batch_size','2000')}}, 'worker_batch_size', {{var('worker_batch_size','1000')}}, 'batch_call_limit', {{var('batch_call_limit','1')}}))",
target = "{{this.schema}}.{{this.identifier}}"
),
tags = ['streamline_core_realtime']
) }}
WITH last_3_days AS (
SELECT
block_number
FROM
{{ ref("_block_lookback") }}
),
blocks AS (
SELECT
block_number :: STRING AS block_number
FROM
{{ ref("streamline__blocks") }}
WHERE
(
block_number >= (
SELECT
block_number
FROM
last_3_days
)
)
EXCEPT
SELECT
block_number :: STRING
FROM
{{ ref("streamline__complete_qn_getBlockWithReceipts") }}
WHERE
(
block_number >= (
SELECT
block_number
FROM
last_3_days
)
)
AND _inserted_timestamp >= DATEADD(
'day',
-4,
SYSDATE()
)
),
all_blocks AS (
SELECT
block_number
FROM
blocks
UNION
SELECT
block_number
FROM
(
SELECT
block_number
FROM
{{ ref("_missing_receipts") }}
UNION
SELECT
block_number
FROM
{{ ref("_missing_txs") }}
UNION
SELECT
block_number
FROM
{{ ref("_unconfirmed_blocks") }}
)
)
SELECT
PARSE_JSON(
CONCAT(
'{"jsonrpc": "2.0",',
'"method": "qn_getBlockWithReceipts", "params":["',
REPLACE(
concat_ws(
'',
'0x',
to_char(
block_number :: INTEGER,
'XXXXXXXX'
)
),
' ',
''
),
'"],"id":"',
block_number :: INTEGER,
'"}'
)
) AS request
FROM
all_blocks
ORDER BY
block_number ASC
LIMIT
10000

130
models/streamline/silver/core/realtime/streamline__receipts_realtime.sql Normal file
View File

@ -0,0 +1,130 @@
{# Set variables #}
{%- set model_name = 'RECEIPTS' -%}
{%- set model_type = 'REALTIME' -%}
{%- set min_block = var('GLOBAL_START_UP_BLOCK', none) -%}
{%- set default_vars = set_default_variables_streamline(model_name, model_type) -%}
{# Set up parameters for the streamline process. These will come from the vars set in dbt_project.yml #}
{%- set streamline_params = set_streamline_parameters(
model_name=model_name,
model_type=model_type
) -%}
{%- set node_url = default_vars['node_url'] -%}
{%- set node_secret_path = default_vars['node_secret_path'] -%}
{%- set model_quantum_state = default_vars['model_quantum_state'] -%}
{%- set sql_limit = streamline_params['sql_limit'] -%}
{%- set testing_limit = default_vars['testing_limit'] -%}
{%- set order_by_clause = default_vars['order_by_clause'] -%}
{%- set new_build = default_vars['new_build'] -%}
{%- set method_params = streamline_params['method_params'] -%}
{%- set method = streamline_params['method'] -%}
{# Log configuration details #}
{{ log_streamline_details(
model_name=model_name,
model_type=model_type,
node_url=node_url,
model_quantum_state=model_quantum_state,
sql_limit=sql_limit,
testing_limit=testing_limit,
order_by_clause=order_by_clause,
new_build=new_build,
streamline_params=streamline_params,
method_params=method_params,
method=method,
min_block=min_block
) }}
{# Set up dbt configuration #}
{{ 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 = streamline_params
),
tags = ['streamline_core_realtime_receipts']
) }}
{# Main query starts here #}
WITH
{% if not new_build %}
last_3_days AS (
SELECT block_number
FROM {{ ref("_block_lookback") }}
),
{% endif %}
{# Identify blocks that need processing #}
to_do AS (
SELECT block_number
FROM {{ ref("streamline__blocks") }}
WHERE
block_number IS NOT NULL
{% if not new_build %}
AND block_number >= (SELECT block_number FROM last_3_days)
{% endif %}
{% if min_block is not none %}
AND block_number >= {{ min_block }}
{% endif %}
EXCEPT
{# Exclude blocks that have already been processed #}
SELECT block_number
FROM {{ ref('streamline__' ~ model_name.lower() ~ '_complete') }}
WHERE 1=1
{% if not new_build %}
AND block_number >= (SELECT block_number FROM last_3_days)
{% endif %}
)
{# Prepare the final list of blocks to process #}
,ready_blocks AS (
SELECT block_number
FROM to_do
{% if not new_build %}
UNION
SELECT block_number
FROM {{ ref("_unconfirmed_blocks") }}
UNION
SELECT block_number
FROM {{ ref("_missing_txs") }}
UNION
SELECT block_number
FROM {{ ref("_missing_receipts") }}
{% endif %}
{% if testing_limit is not none %}
LIMIT {{ testing_limit }}
{% endif %}
)
{# Generate API requests for each block #}
SELECT
block_number,
ROUND(block_number, -3) AS partition_key,
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id', block_number,
'jsonrpc', '2.0',
'method', '{{ method }}',
'params', {{ method_params }}
),
'{{ node_secret_path }}'
) AS request
FROM
ready_blocks
{{ order_by_clause }}
LIMIT {{ sql_limit }}

127
models/streamline/silver/core/realtime/streamline__traces_realtime.sql Normal file
View File

@ -0,0 +1,127 @@
{# Set variables #}
{%- set model_name = 'TRACES' -%}
{%- set model_type = 'REALTIME' -%}
{%- set min_block = var('GLOBAL_START_UP_BLOCK', none) -%}
{%- set default_vars = set_default_variables_streamline(model_name, model_type) -%}
{# Set up parameters for the streamline process. These will come from the vars set in dbt_project.yml #}
{%- set streamline_params = set_streamline_parameters(
model_name=model_name,
model_type=model_type
) -%}
{%- set node_url = default_vars['node_url'] -%}
{%- set node_secret_path = default_vars['node_secret_path'] -%}
{%- set model_quantum_state = default_vars['model_quantum_state'] -%}
{%- set sql_limit = streamline_params['sql_limit'] -%}
{%- set testing_limit = default_vars['testing_limit'] -%}
{%- set order_by_clause = default_vars['order_by_clause'] -%}
{%- set new_build = default_vars['new_build'] -%}
{%- set method_params = streamline_params['method_params'] -%}
{%- set method = streamline_params['method'] -%}
{# Log configuration details #}
{{ log_streamline_details(
model_name=model_name,
model_type=model_type,
node_url=node_url,
model_quantum_state=model_quantum_state,
sql_limit=sql_limit,
testing_limit=testing_limit,
order_by_clause=order_by_clause,
new_build=new_build,
streamline_params=streamline_params,
method_params=method_params,
method=method,
min_block=min_block
) }}
{# Set up dbt configuration #}
{{ 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 = streamline_params
),
tags = ['streamline_core_realtime']
) }}
{# Main query starts here #}
WITH
{% if not new_build %}
last_3_days AS (
SELECT block_number
FROM {{ ref("_block_lookback") }}
),
{% endif %}
{# Identify blocks that need processing #}
to_do AS (
SELECT block_number
FROM {{ ref("streamline__blocks") }}
WHERE
block_number IS NOT NULL
{% if not new_build %}
AND block_number >= (SELECT block_number FROM last_3_days)
{% endif %}
{% if min_block is not none %}
AND block_number >= {{ min_block }}
{% endif %}
EXCEPT
{# Exclude blocks that have already been processed #}
SELECT block_number
FROM {{ ref('streamline__' ~ model_name.lower() ~ '_complete') }}
WHERE 1=1
{% if not new_build %}
AND block_number >= (SELECT block_number FROM last_3_days)
{% endif %}
)
{# Prepare the final list of blocks to process #}
,ready_blocks AS (
SELECT block_number
FROM to_do
{% if not new_build %}
UNION
SELECT block_number
FROM {{ ref("_unconfirmed_blocks") }}
UNION
SELECT block_number
FROM {{ ref("_missing_traces") }}
{% endif %}
{% if testing_limit is not none %}
LIMIT {{ testing_limit }}
{% endif %}
)
{# Generate API requests for each block #}
SELECT
block_number,
ROUND(block_number, -3) AS partition_key,
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id', block_number,
'jsonrpc', '2.0',
'method', '{{ method }}',
'params', {{ method_params }}
),
'{{ node_secret_path }}'
) AS request
FROM
ready_blocks
{{ order_by_clause }}
LIMIT {{ sql_limit }}

36
models/streamline/silver/core/streamline__blocks.sql
View File

@ -1,20 +1,34 @@
{%- if flags.WHICH == 'compile' and execute -%}
{% set config_log = '\n' %}
{% set config_log = config_log ~ '\n=== DBT Model Config ===\n'%}
{% set config_log = config_log ~ '\n{{ config (\n' %}
{% set config_log = config_log ~ ' materialized = "' ~ config.get('materialized') ~ '",\n' %}
{% set config_log = config_log ~ ' tags = ' ~ config.get('tags') | tojson ~ '\n' %}
{% set config_log = config_log ~ ') }}\n' %}
{{ log(config_log, info=True) }}
{{ log("", info=True) }}
{%- endif -%}
{{ config (
materialized = "view",
tags = ['streamline_core_complete']
) }}
{% if execute %}
{% set height = run_query('SELECT streamline.udf_get_chainhead()') %}
{% set block_height = height.columns [0].values() [0] %}
{% else %}
{% set block_height = 0 %}
{% endif %}
SELECT
_id AS block_number
_id AS block_number,
utils.udf_int_to_hex(_id) AS block_number_hex
FROM
{{ ref("silver__number_sequence") }}
{{ ref('silver__number_sequence') }}
WHERE
_id <= {{ block_height }}
_id <= (
SELECT
COALESCE(
block_number,
0
)
FROM
{{ ref("streamline__get_chainhead") }}
)
ORDER BY
_id ASC
_id ASC

54
models/streamline/silver/core/streamline__get_chainhead.sql Normal file
View File

@ -0,0 +1,54 @@
{%- set model_quantum_state = var('CHAINHEAD_QUANTUM_STATE', 'livequery') -%}
{%- set node_url = var('GLOBAL_NODE_URL', '{Service}/{Authentication}') -%}
{%- if flags.WHICH == 'compile' and execute -%}
{{ log("=== Current Variable Settings ===", info=True) }}
{{ log("CHAINHEAD_QUANTUM_STATE: " ~ model_quantum_state, info=True) }}
{{ log("", info=True) }}
{{ log("=== API Details ===", info=True) }}
{{ log("NODE_URL: " ~ node_url, info=True) }}
{{ log("NODE_SECRET_PATH: " ~ var('GLOBAL_NODE_SECRET_PATH'), info=True) }}
{{ log("", info=True) }}
{% set config_log = '\n' %}
{% set config_log = config_log ~ '\n=== DBT Model Config ===\n'%}
{% set config_log = config_log ~ '\n{{ config (\n' %}
{% set config_log = config_log ~ ' materialized = "' ~ config.get('materialized') ~ '",\n' %}
{% set config_log = config_log ~ ' tags = ' ~ config.get('tags') | tojson ~ '\n' %}
{% set config_log = config_log ~ ') }}\n' %}
{{ log(config_log, info=True) }}
{{ log("", info=True) }}
{%- endif -%}
{{ config (
materialized = 'table',
tags = ['streamline_core_complete','chainhead']
) }}
SELECT
live.udf_api(
'POST',
'{{ node_url }}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json',
'fsc-quantum-state', '{{ model_quantum_state }}'
),
OBJECT_CONSTRUCT(
'id',
0,
'jsonrpc',
'2.0',
'method',
'eth_blockNumber',
'params',
[]
),
'{{ var('GLOBAL_NODE_SECRET_PATH') }}'
) AS resp,
utils.udf_hex_to_int(
resp :data :result :: STRING
) AS block_number

9
models/streamline/silver/core/streamline__get_chainhead.yml Normal file
View File

@ -0,0 +1,9 @@
version: 2
models:
- name: streamline__get_chainhead
description: "This model is used to get the chainhead from the blockchain."
columns:
- name: BLOCK_NUMBER
tests:
- not_null