remove nullable types

blocks/evm/Makefile

@@ -35,7 +35,7 @@ sql-setup:
 
 .PHONY: sql-run
 sql-run:
-	substreams-sink-sql run clickhouse://default:default@localhost:9000/default substreams.yaml -e eth.substreams.pinax.network:443 1:10000 --final-blocks-only --undo-buffer-size 100 --on-module-hash-mistmatch=warn
+	substreams-sink-sql run clickhouse://default:default@localhost:9000/default substreams.yaml -e eth.substreams.pinax.network:443 1:10000 --final-blocks-only --undo-buffer-size 100 --on-module-hash-mistmatch=warn --batch-block-flush-interval 10
 
 .PHONY: deploy
 deploy:

blocks/evm/schema.sql

@@ -26,19 +26,19 @@ CREATE TABLE IF NOT EXISTS blocks
     state_root              String,
     receipts_root           String,
     miner                   String,
-    difficulty              Int64,
-    total_difficulty        Int128,
+    difficulty              Int64, -- Nullable
+    total_difficulty        Int64, -- Nullable
     size                    String,
     mix_hash                String,
     extra_data              String,
     gas_limit               UInt64,
     gas_used                UInt64,
-    blob_gas_used           UInt64,
+    blob_gas_used           UInt64, -- Nullable
     total_transactions      UInt64,
     successful_transactions UInt64,
     failed_transactions     UInt64,
-    base_fee_per_gas        String,
-    parent_beacon_root      String
+    base_fee_per_gas        String, -- Nullable
+    parent_beacon_root      String  -- Nullable
 )
     ENGINE = ReplacingMergeTree()
         PRIMARY KEY (date, time, number, hash)

blocks/evm/src/blocks.rs

@@ -5,6 +5,7 @@ use substreams::pb::substreams::Clock;
 use substreams_database_change::pb::database::{table_change, DatabaseChanges};
 use substreams_ethereum::pb::eth::v2::Block;
 
+// https://github.com/streamingfast/firehose-ethereum/blob/develop/proto/sf/ethereum/type/v2/type.proto
 pub fn insert_blocks(tables: &mut DatabaseChanges, clock: &Clock, block: &Block) {
     let header = block.clone().header.unwrap();
     let parent_hash = bytes_to_hex(header.parent_hash);
@@ -58,22 +59,24 @@ pub fn insert_blocks(tables: &mut DatabaseChanges, clock: &Clock, block: &Block)
     // optional fields
     match header.difficulty {
         Some(difficulty) => row.change("difficulty", ("", difficulty.with_decimal(0).to_string().as_str())),
-        None => row,
+        None => row.change("difficulty", ("", "0")), // Nullable
     };
+
     match header.total_difficulty {
         Some(total_difficulty) => row.change("total_difficulty", ("", total_difficulty.with_decimal(0).to_string().as_str())),
-        None => row,
+        None => row.change("total_difficulty", ("", "0")), // Nullable
     };
+
     match header.blob_gas_used {
         Some(blob_gas_used) => row.change("blob_gas_used", ("", blob_gas_used.to_string().as_str())),
-        None => row,
+        None => row.change("blob_gas_used", ("", "0")), // Nullable
     };
     match header.base_fee_per_gas {
         Some(base_fee_per_gas) => row.change("base_fee_per_gas", ("", base_fee_per_gas.with_decimal(0).to_string().as_str())),
-        None => row,
+        None => row.change("base_fee_per_gas", ("", "0")), // Nullable
     };
     match header.parent_beacon_root.len() {
-        0 => row,
+        0 => row.change("parent_beacon_root", ("", "0")), // Nullable,
         _ => row.change("parent_beacon_root", ("", bytes_to_hex(header.parent_beacon_root).as_str())),
     };
 }

blocks/evm/src/logs.rs

@@ -6,6 +6,7 @@ use substreams::pb::substreams::Clock;
 use substreams_database_change::pb::database::{table_change, DatabaseChanges};
 use substreams_ethereum::pb::eth::v2::Block;
 
+// https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L512
 pub fn insert_logs(tables: &mut DatabaseChanges, clock: &Clock, block: &Block) {
     // logs
     for log in block.logs() {

common/src/keys.rs

@@ -6,10 +6,10 @@ use crate::utils::block_time_to_date;
 
 pub fn block_keys(clock: &Clock) -> HashMap<String, String> {
     let timestamp = clock.clone().timestamp.unwrap();
-    let block_time = timestamp.to_string();
+    let block_date = block_time_to_date(&timestamp.to_string()).to_string();
+    let block_time = timestamp.seconds.to_string();
     let block_number = clock.number.to_string();
     let block_hash = format!("0x{}", clock.id);
-    let block_date = block_time_to_date(block_time.as_str()).to_string();
 
     HashMap::from([
         ("block_date".to_string(), block_date),

common/src/sinks.rs

@@ -5,15 +5,15 @@ use crate::utils::block_time_to_date;
 
 pub fn insert_timestamp(row: &mut TableChange, clock: &Clock, is_block: bool) {
     let timestamp = clock.clone().timestamp.unwrap();
+    let block_date = block_time_to_date(timestamp.to_string().as_str());
     let block_time = timestamp.seconds.to_string();
     let block_number = clock.number.to_string();
     let block_hash = format!("0x{}", clock.id);
-    let block_date = block_time_to_date(block_time.as_str());
     let prefix = if is_block { "" } else { "block_" };
 
     row
+        .change(format!("{}date", prefix).as_str(), ("", block_date.as_str()))
         .change(format!("{}time", prefix).as_str(), ("", block_time.as_str()))
         .change(format!("{}number", prefix).as_str(), ("", block_number.as_str()))
-        .change(format!("{}date", prefix).as_str(), ("", block_date.as_str()))
         .change(format!("{}hash", prefix).as_str(), ("", block_hash.as_str()));
 }