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

blocks

Browse Source
This commit is contained in:
Jack Forgash 2025-02-04 11:13:25 -07:00
parent 03d554e7e9
commit 777deb3722
13 changed files with 240 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
macros/api_udf/create_get_nearblocks_fts.sql
View File

@ -1,3 +1,5 @@
-- TODO slated for deprecation and drop
{# Deprecated 9/25/2023 #}
{% macro create_get_nearblocks_fts() %}
{% set create_table %}

2
macros/api_udf/get_nearblocks_fts.sql
View File

@ -1,3 +1,5 @@
-- TODO slated for deprecation and drop
{# Deprecated 9/25/2023 #}
{% macro get_nearblocks_fts() %}

2
macros/create_procedure_get_chainhead.sql
View File

@ -1,3 +1,5 @@
-- TODO slated for deprecation and drop
{% macro create_PROCEDURE_GET_CHAINHEAD() %}
{% set sql %}
CREATE OR REPLACE PROCEDURE {{ target.database }}.STREAMLINE.GET_CHAINHEAD(

1
macros/incremental_utils.sql
View File

@ -1,3 +1,4 @@
-- TODO slated for deprecation and drop
{% macro incremental_load_filter(time_col) -%}
{% if is_incremental() %}

68
macros/streamline/models.sql Normal file
View File

@ -0,0 +1,68 @@
{% macro streamline_external_table_query_v2(
model,
partition_function
) %}
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, SYSDATE()),
table_name => '{{ source( "bronze_streamline", model) }}')
) A
)
SELECT
s.*,
b.file_name,
_inserted_timestamp
FROM
{{ source(
"bronze_streamline",
model
) }}
s
JOIN meta b
ON b.file_name = metadata$filename
AND b.partition_key = s.partition_key
WHERE
b.partition_key = s.partition_key
AND DATA :error IS NULL
{% endmacro %}
{% macro streamline_external_table_FR_query_v2(
model,
partition_function
) %}
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", model) }}'
)
) A
)
SELECT
s.*,
b.file_name,
_inserted_timestamp
FROM
{{ source(
"bronze_streamline",
model
) }}
s
JOIN meta b
ON b.file_name = metadata$filename
AND b.partition_key = s.partition_key
WHERE
b.partition_key = s.partition_key
AND DATA :error IS NULL
{% endmacro %}

27
macros/udfs/create_udf_get_chainhead.sql
View File

@ -7,21 +7,22 @@
SELECT
{{ target.database }}.live.udf_api(
'POST',
'https://rpc.mainnet.near.org',
{
'Content-Type': 'application/json'
},
{
'jsonrpc': '2.0',
'id': 'dontcare',
'method' :'status',
'params':{
'finality': 'final'
}
}
'{Service}',
OBJECT_CONSTRUCT(
'Content-Type', 'application/json'
),
OBJECT_CONSTRUCT(
'jsonrpc', '2.0',
'id', 'Flipside/getChainhead/0.1',
'method', 'status',
'params', OBJECT_CONSTRUCT(
'finality', 'final'
)
),
'Vault/prod/near/quicknode/mainnet'
) :data :result :sync_info :latest_block_height :: INT AS block_id
$$
{% endset %}
{% do run_query(sql) %}
{% endmacro %}
{% endmacro %}

9
models/bronze/bronze__FR_blocks.sql Normal file
View File

@ -0,0 +1,9 @@
{{ config (
materialized = 'view',
tags = ['streamline_helper']
) }}
{{ streamline_external_table_FR_query_v2(
model = "blocks_v2",
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )"
) }}

10
models/bronze/bronze__blocks.sql Normal file
View File

@ -0,0 +1,10 @@
{{ config (
materialized = 'view',
tags = ['streamline_helper']
) }}
{{ streamline_external_table_query_v2(
model = "blocks_v2",
partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )"
) }}

11
models/sources.yml
View File

@ -32,4 +32,13 @@ sources:
schema: near
database: hevo
tables:
- name: flipsidecrypto_near_ft_balances_daily
- name: flipsidecrypto_near_ft_balances_daily
- name: bronze_streamline
database: streamline
schema: |
{{ "NEAR_DEV" if var("STREAMLINE_USE_DEV_FOR_EXTERNAL_TABLES", False) else "NEAR" }}
tables:
- name: blocks_v2
- name: chunks_v2
- name: tx_status_v2

11
models/streamline/_block_lookback.sql Normal file
View File

@ -0,0 +1,11 @@
{{ config (
materialized = "ephemeral"
) }}
SELECT
MIN(block_id) AS block_number
FROM
{{ ref("core__fact_blocks") }}
WHERE
block_timestamp >= DATEADD('hour', -72, SYSDATE())
AND block_timestamp < DATEADD('hour', -71, SYSDATE())

41
models/streamline/core/complete/streamline__blocks_complete.sql
View File

@ -0,0 +1,41 @@
-- depends_on: {{ ref('bronze__blocks') }}
-- depends_on: {{ ref('bronze__FR_blocks') }}
{{ config (
materialized = "incremental",
unique_key = "block_number",
cluster_by = "ROUND(block_number, -3)",
merge_exclude_columns = ["inserted_timestamp"],
post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(block_number)",
tags = ['streamline_complete']
) }}
SELECT
VALUE :block_number :: INT AS block_number,
DATA :header :hash :: STRING AS block_hash,
partition_key,
_inserted_timestamp,
DATA :header :hash :: STRING AS complete_blocks_id,
SYSDATE() AS inserted_timestamp,
SYSDATE() AS modified_timestamp,
'{{ invocation_id }}' AS _invocation_id
FROM
{% if is_incremental() %}
{{ ref('bronze__blocks') }}
WHERE
_inserted_timestamp >= COALESCE(
(
SELECT
MAX(_inserted_timestamp) _inserted_timestamp
FROM
{{ this }}
),
'1900-01-01' :: timestamp_ntz
)
{% else %}
{{ ref('bronze__FR_blocks') }}
{% endif %}
qualify(ROW_NUMBER() over (PARTITION BY block_number
ORDER BY
_inserted_timestamp DESC)) = 1

55
models/streamline/core/realtime/streamline__blocks_realtime.sql
View File

@ -1,7 +1,7 @@
{{ config (
materialized = "view",
post_hook = fsc_utils.if_data_call_function_v2(
func = '{{this.schema}}.udf_bulk_rest_api_v2',
func = 'streamline.udf_bulk_rest_api_v2',
target = "{{this.schema}}.{{this.identifier}}",
params = {
"external_table": "blocks_v2",
@ -10,12 +10,53 @@
"worker_batch_size": "100",
"sql_source": "{{this.identifier}}"
}
)
),
tags = ['streamline_realtime']
) }}
-- single block for testing
WITH last_3_days AS (
SELECT
ZEROIFNULL(block_number) AS block_number
FROM
{{ ref("_block_lookback") }}
),
tbl AS (
SELECT
block_number
FROM
{{ ref('streamline__blocks') }}
WHERE
(
block_number >= (
SELECT
block_number
FROM
last_3_days
)
)
AND block_number IS NOT NULL
EXCEPT
SELECT
block_number
FROM
{{ ref('streamline__blocks_complete') }}
WHERE
block_number >= (
SELECT
block_number
FROM
last_3_days
)
AND _inserted_timestamp >= DATEADD(
'day',
-4,
SYSDATE()
)
-- AND {} IS NOT NULL -- TODO, determine identifier for bad response
)
SELECT
138515000 AS block_number,
block_number,
DATE_PART('EPOCH', SYSDATE()) :: INTEGER AS partition_key,
{{ target.database }}.live.udf_api(
'POST',
@ -26,11 +67,13 @@ SELECT
OBJECT_CONSTRUCT(
'jsonrpc', '2.0',
'method', 'block',
'id', 'dontcare',
'id', 'Flipside/getBlock/0.1',
'params', OBJECT_CONSTRUCT(
'block_id', block_number
)
),
'Vault/prod/near/quicknode/mainnet'
) AS request
FROM tbl
ORDER BY
block_number DESC

21
models/streamline/streamline__blocks.sql Normal file
View File

@ -0,0 +1,21 @@
{{ config(
materialized = "view",
tags = ['streamline_realtime', 'streamline_history', 'streamline_helper']
) }}
{% if execute %}
{% set height = run_query("SELECT streamline.udf_get_chainhead()") %}
{% set block_number = height.columns [0].values() [0] %}
{% else %}
{% set block_number = 0 %}
{% endif %}
SELECT
_id AS block_number
FROM
{{ source(
'crosschain_silver',
'number_sequence'
) }}
WHERE
_id <= {{ block_number }}