dbt project vars, core bronze views and fsc-evm temp macros

2026-02-06 17:36:52 +00:00 · 2024-12-06 18:49:32 +12:00 · 2024-12-06 18:49:32 +12:00 · ea79847066
commit ea79847066
parent e83ca9ae41
8 changed files with 457 additions and 2 deletions
--- a/dbt_project.yml
+++ b/dbt_project.yml
@ -74,4 +74,94 @@ vars:
  API_INTEGRATION: '{{ var("config")[target.name]["API_INTEGRATION"] if var("config")[target.name] else var("config")["dev"]["API_INTEGRATION"] }}' 
  EXTERNAL_FUNCTION_URI: '{{ var("config")[target.name]["EXTERNAL_FUNCTION_URI"] if var("config")[target.name] else var("config")["dev"]["EXTERNAL_FUNCTION_URI"] }}'
  ROLES: |
-      ["INTERNAL_DEV"]
+      ["INTERNAL_DEV"]
+
+#### STREAMLINE 2.0 BEGIN ####
+
+  API_INTEGRATION: '{{ var("config")[target.name]["API_INTEGRATION"] if var("config")[target.name] else var("config")["dev"]["API_INTEGRATION"] }}'
+  EXTERNAL_FUNCTION_URI: '{{ var("config")[target.name]["EXTERNAL_FUNCTION_URI"] if var("config")[target.name] else var("config")["dev"]["EXTERNAL_FUNCTION_URI"] }}'
+  ROLES: |
+    ["INTERNAL_DEV"]
+
+  config:
+  # The keys correspond to dbt profiles and are case sensitive
+    dev:
+      API_INTEGRATION: AWS_BASE_API_DEV
+      EXTERNAL_FUNCTION_URI: 
+      ROLES:
+        - AWS_LAMBDA_BASE_API 
+        - INTERNAL_DEV
+
+    # prod:
+    #   API_INTEGRATION: AWS_BASE_API
+    #   EXTERNAL_FUNCTION_URI: 
+    #   ROLES:
+    #     - AWS_LAMBDA_BASE_API 
+    #     - INTERNAL_DEV
+    #     - DBT_CLOUD_BASE
+
+#### STREAMLINE 2.0 END ####
+
+#### FSC_EVM BEGIN ####
+# Visit https://github.com/FlipsideCrypto/fsc-evm/wiki for more information on required and optional variables
+
+  ### GLOBAL VARIABLES BEGIN ###
+  ## REQUIRED
+  GLOBAL_PROD_DB_NAME: 'base' 
+  GLOBAL_NODE_SECRET_PATH: 'Vault/prod/base/quicknode/mainnet'
+  GLOBAL_BLOCKS_PER_HOUR: 1800 
+  GLOBAL_USES_STREAMLINE_V1: True
+  GLOBAL_USES_SINGLE_FLIGHT_METHOD: True
+
+  ### GLOBAL VARIABLES END ###
+
+  ### MAIN_PACKAGE VARIABLES BEGIN ###
+
+  ### CORE ###
+  ## REQUIRED
+
+  ## OPTIONAL
+  # BLOCKS_TRANSACTIONS_REALTIME_NEW_BUILD: True
+  # BLOCKS_TRANSACTIONS_REALTIME_TESTING_LIMIT: 3
+
+  # BLOCKS_TRANSACTIONS_HISTORY_NEW_BUILD: True
+  # BLOCKS_TRANSACTIONS_HISTORY_SQL_LIMIT: 300000
+  # BLOCKS_TRANSACTIONS_HISTORY_PRODUCER_BATCH_SIZE: 60000
+  # BLOCKS_TRANSACTIONS_HISTORY_WORKER_BATCH_SIZE: 12000
+  # BLOCKS_TRANSACTIONS_HISTORY_TESTING_LIMIT: 4
+  
+  # RECEIPTS_REALTIME_NEW_BUILD: True
+  # RECEIPTS_REALTIME_TESTING_LIMIT: 3
+
+  # RECEIPTS_HISTORY_NEW_BUILD: True
+  # RECEIPTS_HISTORY_SQL_LIMIT: 300000
+  # RECEIPTS_HISTORY_PRODUCER_BATCH_SIZE: 60000
+  # RECEIPTS_HISTORY_WORKER_BATCH_SIZE: 12000
+  # RECEIPTS_HISTORY_TESTING_LIMIT: 4
+
+  # TRACES_REALTIME_NEW_BUILD: True
+  # TRACES_REALTIME_TESTING_LIMIT: 3
+
+  # TRACES_HISTORY_NEW_BUILD: True
+  # TRACES_HISTORY_SQL_LIMIT: 1500000
+  # TRACES_HISTORY_PRODUCER_BATCH_SIZE: 20000
+  # TRACES_HISTORY_WORKER_BATCH_SIZE: 2000
+  # TRACES_HISTORY_ORDER_BY_CLAUSE: 'ORDER BY partition_key DESC'
+  # TRACES_HISTORY_TESTING_LIMIT: 4
+
+  # CONFIRM_BLOCKS_REALTIME_NEW_BUILD: True
+  # CONFIRM_BLOCKS_REALTIME_TESTING_LIMIT: 3
+
+  # CONFIRM_BLOCKS_HISTORY_NEW_BUILD: True
+  # CONFIRM_BLOCKS_HISTORY_SQL_LIMIT: 1500000
+  # CONFIRM_BLOCKS_HISTORY_PRODUCER_BATCH_SIZE: 360000
+  # CONFIRM_BLOCKS_HISTORY_WORKER_BATCH_SIZE: 36000
+  # CONFIRM_BLOCKS_HISTORY_ORDER_BY_CLAUSE: 'ORDER BY partition_key DESC'
+  # CONFIRM_BLOCKS_HISTORY_TESTING_LIMIT: 4
+
+  # GOLD_FULL_REFRESH: True
+  # SILVER_FULL_REFRESH: True
+
+  ### MAIN_PACKAGE VARIABLES END ###
+
+#### FSC_EVM END ####
--- a/macros/fsc_evm_temp/main_package/bronze/streamline_external_table_queries.sql
+++ b/macros/fsc_evm_temp/main_package/bronze/streamline_external_table_queries.sql
@ -0,0 +1,141 @@
+{% macro streamline_external_table_query(
+        source_name,
+        source_version,
+        partition_function,
+        balances,
+        block_number,
+        uses_receipts_by_hash
+    ) %}
+
+    {% if source_version != '' %}
+        {% set source_version = '_' ~ source_version.lower() %}
+    {% endif %}
+
+    WITH meta AS (
+        SELECT
+            job_created_time AS _inserted_timestamp,
+            file_name,
+            {{ partition_function }} AS partition_key
+        FROM
+            TABLE(
+                information_schema.external_table_file_registration_history(
+                    start_time => DATEADD('day', -3, CURRENT_TIMESTAMP()),
+                    table_name => '{{ source( "bronze_streamline", source_name ~ source_version) }}')
+                ) A
+            )
+        SELECT
+            s.*,
+            b.file_name,
+            b._inserted_timestamp
+
+            {% if balances %},
+            r.block_timestamp :: TIMESTAMP AS block_timestamp
+        {% endif %}
+
+        {% if block_number %},
+            COALESCE(
+                s.value :"BLOCK_NUMBER" :: STRING,
+                s.metadata :request :"data" :id :: STRING,
+                PARSE_JSON(
+                    s.metadata :request :"data"
+                ) :id :: STRING
+            ) :: INT AS block_number
+        {% endif %}
+        {% if uses_receipts_by_hash %},
+            s.value :"TX_HASH" :: STRING AS tx_hash
+        {% endif %}
+        FROM
+            {{ source(
+                "bronze_streamline",
+                source_name ~ source_version
+            ) }}
+            s
+            JOIN meta b
+            ON b.file_name = metadata$filename
+            AND b.partition_key = s.partition_key
+
+            {% if balances %}
+            JOIN {{ ref('_block_ranges') }}
+            r
+            ON r.block_number = COALESCE(
+                s.value :"BLOCK_NUMBER" :: INT,
+                s.value :"block_number" :: INT
+            )
+        {% endif %}
+        WHERE
+            b.partition_key = s.partition_key
+            AND DATA :error IS NULL
+            AND DATA IS NOT NULL
+{% endmacro %}
+
+{% macro streamline_external_table_query_fr(
+        source_name,
+        source_version,
+        partition_function,
+        partition_join_key,
+        balances,
+        block_number,
+        uses_receipts_by_hash
+    ) %}
+
+    {% if source_version != '' %}
+        {% set source_version = '_' ~ source_version.lower() %}
+    {% endif %}
+    
+    WITH meta AS (
+        SELECT
+            registered_on AS _inserted_timestamp,
+            file_name,
+            {{ partition_function }} AS partition_key
+        FROM
+            TABLE(
+                information_schema.external_table_files(
+                    table_name => '{{ source( "bronze_streamline", source_name ~ source_version) }}'
+                )
+            ) A
+    )
+SELECT
+    s.*,
+    b.file_name,
+    b._inserted_timestamp
+
+    {% if balances %},
+    r.block_timestamp :: TIMESTAMP AS block_timestamp
+{% endif %}
+
+{% if block_number %},
+    COALESCE(
+        s.value :"BLOCK_NUMBER" :: STRING,
+        s.value :"block_number" :: STRING,
+        s.metadata :request :"data" :id :: STRING,
+        PARSE_JSON(
+            s.metadata :request :"data"
+        ) :id :: STRING
+    ) :: INT AS block_number
+{% endif %}
+{% if uses_receipts_by_hash %},
+    s.value :"TX_HASH" :: STRING AS tx_hash
+{% endif %}
+FROM
+    {{ source(
+        "bronze_streamline",
+        source_name ~ source_version
+    ) }}
+    s
+    JOIN meta b
+    ON b.file_name = metadata$filename
+    AND b.partition_key = s.{{ partition_join_key }}
+
+    {% if balances %}
+        JOIN {{ ref('_block_ranges') }}
+        r
+        ON r.block_number = COALESCE(
+            s.value :"BLOCK_NUMBER" :: INT,
+            s.value :"block_number" :: INT
+        )
+    {% endif %}
+WHERE
+    b.partition_key = s.{{ partition_join_key }}
+    AND DATA :error IS NULL
+    AND DATA IS NOT NULL
+{% endmacro %}
--- a/macros/fsc_evm_temp/main_package/logging/bronze.sql
+++ b/macros/fsc_evm_temp/main_package/logging/bronze.sql
@ -0,0 +1,36 @@
+{% macro log_bronze_details(source_name, source_version, model_type, partition_function, partition_join_key, block_number, uses_receipts_by_hash) %}
+
+{% if source_version != '' %}
+    {% set source_version = '_' ~ source_version.lower() %}
+{% endif %}
+{% if model_type != '' %}
+    {% set model_type = '_' ~ model_type %}
+{% endif %}
+
+{%- if flags.WHICH == 'compile' and execute -%}
+
+    {{ log("=== Current Variable Settings ===", info=True) }}
+    {{ log(source_name ~ model_type ~ '_PARTITION_FUNCTION: ' ~ partition_function, info=True) }}
+    {{ log(source_name ~ model_type ~ '_PARTITION_JOIN_KEY: ' ~ partition_join_key, info=True) }}
+    {{ log(source_name ~ model_type ~ '_BLOCK_NUMBER: ' ~ block_number, info=True) }}
+    {% if uses_receipts_by_hash %}
+        {{ log("USES_RECEIPTS_BY_HASH: " ~ uses_receipts_by_hash, info=True) }}
+    {% endif %}
+
+    {{ log("", info=True) }}
+    {{ log("=== Source Details ===", info=True) }}
+    {{ log("Source: " ~ source('bronze_streamline', source_name.lower() ~ source_version.lower()), 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 -%}
+
+{% endmacro %}
--- a/macros/fsc_evm_temp/main_package/logging/complete.sql
+++ b/macros/fsc_evm_temp/main_package/logging/complete.sql
@ -0,0 +1,29 @@
+{% macro log_complete_details(post_hook, full_refresh_type, uses_receipts_by_hash) %}
+
+{%- if flags.WHICH == 'compile' and execute -%}
+
+    {% if uses_receipts_by_hash %}
+
+        {{ log("=== Current Variable Settings ===", info=True) }}
+        {{ log("USES_RECEIPTS_BY_HASH: " ~ uses_receipts_by_hash, info=True) }}
+    
+    {% endif %}
+
+    {% 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 ~ '    unique_key = "' ~ config.get('unique_key') ~ '",\n' %}
+    {% set config_log = config_log ~ '    cluster_by = "' ~ config.get('cluster_by') ~ '",\n' %}
+    {% set config_log = config_log ~ '    merge_update_columns = ' ~ config.get('merge_update_columns') | tojson ~ ',\n' %}
+    {% set config_log = config_log ~ '    post_hook = "' ~ post_hook ~ '",\n' %}
+    {% set config_log = config_log ~ '    incremental_predicates = ' ~ config.get('incremental_predicates') | tojson ~ ',\n' %}
+    {% set config_log = config_log ~ '    full_refresh = ' ~ full_refresh_type ~ ',\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 -%}
+
+{% endmacro %}
--- a/macros/fsc_evm_temp/main_package/logging/requests.sql
+++ b/macros/fsc_evm_temp/main_package/logging/requests.sql
@ -0,0 +1,55 @@
+{% macro log_streamline_details(model_name, model_type, node_url, model_quantum_state, sql_limit, testing_limit, order_by_clause, new_build, streamline_params, uses_receipts_by_hash, method, method_params, min_block=0) %}
+
+{%- if flags.WHICH == 'compile' and execute -%}
+
+    {{ log("=== Current Variable Settings ===", info=True) }}
+    {{ log("START_UP_BLOCK: " ~ min_block, 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) }}
+
+    {{ log("=== Current Variable Settings ===", info=True) }}
+
+    {{ log((model_name ~ '_' ~ model_type ~ '_model_quantum_state').upper() ~ ': ' ~ model_quantum_state, info=True) }}
+    {{ log((model_name ~ '_' ~ model_type ~ '_sql_limit').upper() ~ ': ' ~ sql_limit, info=True) }}
+    {{ log((model_name ~ '_' ~ model_type ~ '_testing_limit').upper() ~ ': ' ~ testing_limit, info=True) }}
+    {{ log((model_name ~ '_' ~ model_type ~ '_order_by_clause').upper() ~ ': ' ~ order_by_clause, info=True) }}
+    {{ log((model_name ~ '_' ~ model_type ~ '_new_build').upper() ~ ': ' ~ new_build, info=True) }}
+    {{ log('USES_RECEIPTS_BY_HASH' ~ ': ' ~ uses_receipts_by_hash, info=True) }}
+    {{ log("", info=True) }}
+
+    {{ log("=== RPC Details ===", info=True) }}
+
+    {{ log(model_name ~ ": {", info=True) }}
+    {{ log("    method: '" ~ method ~ "',", info=True) }}
+    {{ log("    method_params: " ~ method_params, info=True) }}
+    {{ log("}", info=True) }}
+    {{ log("", info=True) }}
+
+    {% set params_str = streamline_params | tojson %}
+    {% set params_formatted = params_str | replace('{', '{\n            ') | replace('}', '\n        }') | replace(', ', ',\n            ') %}
+    
+    {# Clean up the method_params formatting #}
+    {% set params_formatted = params_formatted | replace('"method_params": "', '"method_params": "') | replace('\\n', ' ') | replace('\\u0027', "'") %}
+
+    {% 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 ~ '    post_hook = fsc_utils.if_data_call_function_v2(\n' %}
+    {% set config_log = config_log ~ '        func = "streamline.udf_bulk_rest_api_v2",\n' %}
+    {% set config_log = config_log ~ '        target = "' ~ this.schema ~ '.' ~ this.identifier ~ '",\n' %}
+    {% set config_log = config_log ~ '        params = ' ~ params_formatted ~ '\n' %}
+    {% set config_log = config_log ~ '    ),\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 -%}
+
+{% endmacro %}
--- a/macros/fsc_evm_temp/main_package/streamline/set_default_variables.sql
+++ b/macros/fsc_evm_temp/main_package/streamline/set_default_variables.sql
@ -0,0 +1,47 @@
+{% macro set_default_variables_streamline(model_name, model_type) %}
+
+{%- set node_url = var('GLOBAL_NODE_URL', '{Service}/{Authentication}') -%}
+{%- set node_secret_path = var('GLOBAL_NODE_SECRET_PATH', '') -%}
+{%- set model_quantum_state = var((model_name ~ '_' ~ model_type ~ '_quantum_state').upper(), 'streamline') -%}
+{%- set testing_limit = var((model_name ~ '_' ~ model_type ~ '_testing_limit').upper(), none) -%}
+{%- set new_build = var((model_name ~ '_' ~ model_type ~ '_new_build').upper(), false) -%}
+{%- set default_order = 'ORDER BY partition_key DESC, block_number DESC' if model_type.lower() == 'realtime' 
+    else 'ORDER BY partition_key ASC, block_number ASC' -%}
+{%- set order_by_clause = var((model_name ~ '_' ~ model_type ~ '_order_by_clause').upper(), default_order) -%}
+{%- set uses_receipts_by_hash = var('GLOBAL_USES_RECEIPTS_BY_HASH', false) -%}
+
+{%- set variables = {
+    'node_url': node_url,
+    'node_secret_path': node_secret_path,
+    'model_quantum_state': model_quantum_state,
+    'testing_limit': testing_limit,
+    'new_build': new_build,
+    'order_by_clause': order_by_clause,
+    'uses_receipts_by_hash': uses_receipts_by_hash
+} -%}
+
+{{ return(variables) }}
+
+{% endmacro %}  
+
+{% macro set_default_variables_bronze(source_name, model_type) %}
+
+{%- set partition_function = var(source_name ~ model_type ~ '_PARTITION_FUNCTION', 
+ "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 4), '_', 1) AS INTEGER)") 
+-%}
+{%- set partition_join_key = var(source_name ~ model_type ~ '_PARTITION_JOIN_KEY', 'partition_key') -%}
+{%- set block_number = var(source_name ~ model_type ~ '_BLOCK_NUMBER', true) -%}
+{%- set balances = var(source_name ~ model_type ~ '_BALANCES', false) -%}
+{%- set uses_receipts_by_hash = var('GLOBAL_USES_RECEIPTS_BY_HASH', false) -%}
+
+{%- set variables = {
+    'partition_function': partition_function,
+    'partition_join_key': partition_join_key,
+    'block_number': block_number,
+    'balances': balances,
+    'uses_receipts_by_hash': uses_receipts_by_hash
+} -%}
+
+{{ return(variables) }}
+
+{% endmacro %} 
--- a/macros/fsc_evm_temp/main_package/streamline/set_streamline_parameters.sql
+++ b/macros/fsc_evm_temp/main_package/streamline/set_streamline_parameters.sql
@ -0,0 +1,57 @@
+{% macro set_streamline_parameters(model_name, model_type, multiplier=1) %}
+
+{%- set rpc_config_details = {
+    "blocks_transactions": {
+        "method": 'eth_getBlockByNumber',
+        "method_params": 'ARRAY_CONSTRUCT(utils.udf_int_to_hex(block_number), TRUE)',
+        "exploded_key": ['result', 'result.transactions']
+    },
+    "receipts_by_hash": {
+        "method": 'eth_getTransactionReceipt',
+        "method_params": 'ARRAY_CONSTRUCT(tx_hash)'
+    },
+    "receipts": {
+        "method": 'eth_getBlockReceipts',
+        "method_params": 'ARRAY_CONSTRUCT(utils.udf_int_to_hex(block_number))',
+        "exploded_key": ['result'],
+        "lambdas": 2
+
+    },
+    "traces": {
+        "method": 'debug_traceBlockByNumber',
+        "method_params": "ARRAY_CONSTRUCT(utils.udf_int_to_hex(block_number), OBJECT_CONSTRUCT('tracer', 'callTracer', 'timeout', '120s'))",
+        "exploded_key": ['result'],
+        "lambdas": 2
+    },
+    "confirm_blocks": {
+        "method": 'eth_getBlockByNumber',
+        "method_params": 'ARRAY_CONSTRUCT(utils.udf_int_to_hex(block_number), FALSE)'
+    }
+} -%}
+
+{%- set rpc_config = rpc_config_details[model_name.lower()] -%}
+
+{%- set params = {
+    "external_table": var((model_name ~ '_' ~ model_type ~ '_external_table').upper(), model_name.lower()),
+    "sql_limit": var((model_name ~ '_' ~ model_type ~ '_sql_limit').upper(), 2 * var('GLOBAL_BLOCKS_PER_HOUR',0) * multiplier),
+    "producer_batch_size": var((model_name ~ '_' ~ model_type ~ '_producer_batch_size').upper(), 2 * var('GLOBAL_BLOCKS_PER_HOUR',0) * multiplier),
+    "worker_batch_size": var(
+        (model_name ~ '_' ~ model_type ~ '_worker_batch_size').upper(), 
+        (2 * var('GLOBAL_BLOCKS_PER_HOUR',0) * multiplier) // (rpc_config.get('lambdas', 1))
+    ),
+    "sql_source": (model_name ~ '_' ~ model_type).lower(),
+    "method": rpc_config['method'],
+    "method_params": rpc_config['method_params']
+} -%}
+
+{%- if rpc_config.get('exploded_key') is not none -%}
+    {%- do params.update({"exploded_key": tojson(rpc_config['exploded_key'])}) -%}
+{%- endif -%}
+
+{%- if rpc_config.get('lambdas') is not none -%}
+    {%- do params.update({"lambdas": rpc_config['lambdas']}) -%}
+{%- endif -%}
+
+{{ return(params) }}
+
+{% endmacro %}
--- a/models/streamline/bronze/core/bronze__traces_fr_v1.sql
+++ b/models/streamline/bronze/core/bronze__traces_fr_v1.sql
@ -1,5 +1,5 @@
 {# Set variables #}
-{% set source_name = 'TRACES' %}
+{% set source_name = 'DEBUG_TRACEBLOCKBYNUMBER' if var('GLOBAL_USES_SINGLE_FLIGHT_METHOD',false) else 'TRACES' %}
 {% set source_version = '' %}
 {% set model_type = 'FR' %}