Merge pull request #26 from pinax-network/feature/port-solana-to-parq…

…uet-files

Beacon Parquet

Cargo.toml

@@ -6,6 +6,7 @@ members = [
   "blocks/cosmos",
   "blocks/beacon",
   "blocks/solana-parquet",
+  "blocks/beacon-parquet",
   "common",
 ]
 resolver = "2"

blocks/beacon-parquet/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "raw_blocks_beacon_parquet"
+edition = { workspace = true }
+version = { workspace = true }
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+common = { path = "../../common" }
+substreams-ethereum = { workspace = true }
+substreams-database-change = { workspace = true }
+substreams = { workspace = true }
+prost = { workspace = true }
+prost-types = { workspace = true }

blocks/beacon-parquet/Makefile

@@ -0,0 +1,30 @@
+.PHONY: all
+all:
+	make build
+	make pack
+	make graph
+	make info
+
+.PHONY: build
+build:
+	cargo build --target wasm32-unknown-unknown --release
+
+.PHONY: pack
+pack:
+	substreams pack
+
+.PHONY: graph
+graph:
+	substreams graph
+
+.PHONY: info
+info:
+	substreams info
+
+.PHONY: protogen
+protogen:
+	substreams protogen --exclude-paths google
+
+.PHONY: parquet
+parquet:
+	substreams-sink-files run eth-cl.substreams.pinax.network:443 substreams.yaml ch_out './out' 8626176:8626676 --encoder parquet --file-block-count 100 --development-mode

blocks/beacon-parquet/proto/beacon.rawblocks.proto

@@ -0,0 +1,192 @@
+syntax = "proto3";
+
+package beacon.rawblocks;
+
+import "google/protobuf/timestamp.proto";
+
+message Events {
+  repeated Block blocks = 1;
+  repeated Blob blobs = 2;
+  repeated Deposit deposits = 3;
+  repeated Withdrawal withdrawals = 4;
+  repeated Attestation attestations = 5;
+  repeated AttesterSlashing attester_slashings = 6;
+  repeated BlsToExecutionChange bls_to_execution_changes = 7;
+  repeated ProposerSlashing proposer_slashings = 8;
+  repeated VoluntaryExit voluntary_exits = 9;
+}
+
+
+message Block {
+  // clock
+  google.protobuf.Timestamp time = 1;
+  uint64 number = 2;
+  string date = 3;
+  string hash = 4;
+
+  // block
+  uint32 version = 5;
+  string spec = 6;
+  uint64 slot = 7;
+  uint64 parent_slot = 8;
+  string root = 9;
+  string parent_root = 10;
+  string state_root = 11;
+  uint64 proposer_index = 12;
+  string body_root = 13;
+  string signature = 14;
+}
+
+message Blob {
+  // clock
+  google.protobuf.Timestamp block_time = 1;
+  uint64 block_number = 2;
+  string block_date = 3;
+  string block_hash = 4;
+
+  // blob
+  uint64 index = 5;
+  string blob = 6;
+  string kzg_commitment = 7;
+  string kzg_proof = 8;
+  repeated string kzg_commitment_inclusion_proof = 9;
+}
+
+message Deposit {
+  // clock
+  google.protobuf.Timestamp block_time = 1;
+  uint64 block_number = 2;
+  string block_date = 3;
+  string block_hash = 4;
+
+  // deposit
+  uint64 index = 5;
+  repeated string proof = 6;
+  string pubkey = 7;
+  string withdrawal_credentials = 8;
+  string signature = 9;
+  uint64 gwei = 10;
+}
+
+message Withdrawal {
+  // clock
+  google.protobuf.Timestamp block_time = 1;
+  uint64 block_number = 2;
+  string block_date = 3;
+  string block_hash = 4;
+
+  // withdrawal
+  uint64 withdrawal_index = 5;
+  uint64 validator_index = 6;
+  string address = 7;
+  uint64 gwei = 8;
+}
+
+message Attestation {
+  // clock
+  google.protobuf.Timestamp block_time = 1;
+  uint64 block_number = 2;
+  string block_date = 3;
+  string block_hash = 4;
+
+  // attestation
+  uint64 index = 5;
+  string aggregation_bits = 6;
+  uint64 slot = 7;
+  uint64 committee_index = 8;
+  string beacon_block_root = 9;
+  uint64 source_epoch = 10;
+  string source_root = 11;
+  uint64 target_epoch = 12;
+  string target_root = 13;
+  string signature = 14;
+}
+
+message AttesterSlashing {
+  // clock
+  google.protobuf.Timestamp block_time = 1;
+  uint64 block_number = 2;
+  string block_date = 3;
+  string block_hash = 4;
+
+  // attester slashing
+  uint64 index = 5;
+
+  // attestation 1
+  string attestation_1_attesting_indices = 6;
+  uint64 attestation_1_slot = 7;
+  uint64 attestation_1_committee_index = 8;
+  string attestation_1_beacon_block_root = 9;
+  uint64 attestation_1_source_epoch = 10;
+  string attestation_1_source_root = 11;
+  uint64 attestation_1_target_epoch = 12;
+  string attestation_1_target_root = 13;
+  string attestation_1_signature = 14;
+
+  // attestation 2
+  string attestation_2_attesting_indices = 15;
+  uint64 attestation_2_slot = 16;
+  uint64 attestation_2_committee_index = 17;
+  string attestation_2_beacon_block_root = 18;
+  uint64 attestation_2_source_epoch = 19;
+  string attestation_2_source_root = 20;
+  uint64 attestation_2_target_epoch = 21;
+  string attestation_2_target_root = 22;
+  string attestation_2_signature = 23;
+}
+
+message BlsToExecutionChange {
+  // clock
+  google.protobuf.Timestamp block_time = 1;
+  uint64 block_number = 2;
+  string block_date = 3;
+  string block_hash = 4;
+
+  // bls to execution change
+  uint64 index = 5;
+  uint64 validator_index = 6;
+  string from_bls_pubkey = 7;
+  string to_execution_address = 8;
+  string signature = 9;
+}
+
+message ProposerSlashing {
+  // clock
+  google.protobuf.Timestamp block_time = 1;
+  uint64 block_number = 2;
+  string block_date = 3;
+  string block_hash = 4;
+
+  // proposer slashing
+  uint64 index = 5;
+
+  // signed header 1
+  uint64 signed_header_1_slot = 6;
+  uint64 signed_header_1_proposer_index = 7;
+  string signed_header_1_parent_root = 8;
+  string signed_header_1_state_root = 9;
+  string signed_header_1_body_root = 10;
+  string signed_header_1_signature = 11;
+
+  // signed header 2
+  uint64 signed_header_2_slot = 12;
+  uint64 signed_header_2_proposer_index = 13;
+  string signed_header_2_parent_root = 14;
+  string signed_header_2_state_root = 15;
+  string signed_header_2_body_root = 16;
+  string signed_header_2_signature = 17;
+}
+
+message VoluntaryExit {
+  // clock
+  google.protobuf.Timestamp block_time = 1;
+  uint64 block_number = 2;
+  string block_date = 3;
+  string block_hash = 4;
+
+  // voluntary exit
+  uint64 index = 5;
+  uint64 epoch = 6;
+  uint64 validator_index = 7;
+  string signature = 8;
+}

blocks/beacon-parquet/src/attestations.rs

@@ -0,0 +1,31 @@
+use common::utils::bytes_to_hex;
+
+use crate::{
+    pb::{beacon::rawblocks::Attestation as RawAttestation, sf::beacon::r#type::v1::Attestation},
+    structs::BlockTimestamp,
+};
+
+pub fn collect_attestations(attestations: &Vec<Attestation>, timestamp: &BlockTimestamp) -> Vec<RawAttestation> {
+    let mut vec = Vec::<RawAttestation>::new();
+
+    for (index, attestation) in attestations.iter().enumerate() {
+        vec.push(RawAttestation {
+            block_time: Some(timestamp.time),
+            block_number: timestamp.number,
+            block_date: timestamp.date.clone(),
+            block_hash: timestamp.hash.clone(),
+            index: index as u64,
+            aggregation_bits: bytes_to_hex(&attestation.aggregation_bits),
+            slot: attestation.data.as_ref().unwrap().slot,
+            committee_index: attestation.data.as_ref().unwrap().committee_index,
+            beacon_block_root: bytes_to_hex(&attestation.data.as_ref().unwrap().beacon_block_root),
+            source_epoch: attestation.data.as_ref().unwrap().source.as_ref().unwrap().epoch,
+            source_root: bytes_to_hex(&attestation.data.as_ref().unwrap().source.as_ref().unwrap().root),
+            target_epoch: attestation.data.as_ref().unwrap().target.as_ref().unwrap().epoch,
+            target_root: bytes_to_hex(&attestation.data.as_ref().unwrap().target.as_ref().unwrap().root),
+            signature: bytes_to_hex(&attestation.signature),
+        });
+    }
+
+    vec
+}

blocks/beacon-parquet/src/attester_slashings.rs

@@ -0,0 +1,39 @@
+use common::utils::{bytes_to_hex, number_array_to_string};
+
+use crate::{
+    pb::{beacon::rawblocks::AttesterSlashing as RawAttesterSlashing, sf::beacon::r#type::v1::AttesterSlashing},
+    structs::BlockTimestamp,
+};
+
+pub fn collect_attester_slashings(attester_slashings: &Vec<AttesterSlashing>, timestamp: &BlockTimestamp) -> Vec<RawAttesterSlashing> {
+    let mut vec = Vec::<RawAttesterSlashing>::new();
+
+    for (index, attester_slashing) in attester_slashings.iter().enumerate() {
+        vec.push(RawAttesterSlashing {
+            block_time: Some(timestamp.time),
+            block_number: timestamp.number,
+            block_date: timestamp.date.clone(),
+            block_hash: timestamp.hash.clone(),
+            index: index as u64,
+            attestation_1_attesting_indices: number_array_to_string(&attester_slashing.attestation_1.as_ref().unwrap().attesting_indices),
+            attestation_1_slot: attester_slashing.attestation_1.as_ref().unwrap().data.as_ref().unwrap().slot,
+            attestation_1_committee_index: attester_slashing.attestation_1.as_ref().unwrap().data.as_ref().unwrap().committee_index,
+            attestation_1_beacon_block_root: bytes_to_hex(&attester_slashing.attestation_1.as_ref().unwrap().data.as_ref().unwrap().beacon_block_root),
+            attestation_1_source_epoch: attester_slashing.attestation_1.as_ref().unwrap().data.as_ref().unwrap().source.as_ref().unwrap().epoch,
+            attestation_1_source_root: bytes_to_hex(&attester_slashing.attestation_1.as_ref().unwrap().data.as_ref().unwrap().source.as_ref().unwrap().root),
+            attestation_1_target_epoch: attester_slashing.attestation_1.as_ref().unwrap().data.as_ref().unwrap().target.as_ref().unwrap().epoch,
+            attestation_1_target_root: bytes_to_hex(&attester_slashing.attestation_1.as_ref().unwrap().data.as_ref().unwrap().target.as_ref().unwrap().root),
+            attestation_1_signature: bytes_to_hex(&attester_slashing.attestation_1.as_ref().unwrap().signature),
+            attestation_2_attesting_indices: number_array_to_string(&attester_slashing.attestation_2.as_ref().unwrap().attesting_indices),
+            attestation_2_slot: attester_slashing.attestation_2.as_ref().unwrap().data.as_ref().unwrap().slot,
+            attestation_2_committee_index: attester_slashing.attestation_2.as_ref().unwrap().data.as_ref().unwrap().committee_index,
+            attestation_2_beacon_block_root: bytes_to_hex(&attester_slashing.attestation_2.as_ref().unwrap().data.as_ref().unwrap().beacon_block_root),
+            attestation_2_source_epoch: attester_slashing.attestation_2.as_ref().unwrap().data.as_ref().unwrap().source.as_ref().unwrap().epoch,
+            attestation_2_source_root: bytes_to_hex(&attester_slashing.attestation_2.as_ref().unwrap().data.as_ref().unwrap().source.as_ref().unwrap().root),
+            attestation_2_target_epoch: attester_slashing.attestation_2.as_ref().unwrap().data.as_ref().unwrap().target.as_ref().unwrap().epoch,
+            attestation_2_target_root: bytes_to_hex(&attester_slashing.attestation_2.as_ref().unwrap().data.as_ref().unwrap().target.as_ref().unwrap().root),
+            attestation_2_signature: bytes_to_hex(&attester_slashing.attestation_2.as_ref().unwrap().signature),
+        });
+    }
+    vec
+}

blocks/beacon-parquet/src/blobs.rs

@@ -0,0 +1,27 @@
+use common::utils::bytes_to_hex;
+
+use crate::{
+    pb::{beacon::rawblocks::Blob as RawBlob, sf::beacon::r#type::v1::Blob},
+    structs::BlockTimestamp,
+    utils::encode_hex_2d_array,
+};
+
+pub fn collect_blobs(blobs: &Vec<Blob>, timestamp: &BlockTimestamp) -> Vec<RawBlob> {
+    let mut vec = Vec::<RawBlob>::new();
+
+    for b in blobs {
+        vec.push(RawBlob {
+            block_time: Some(timestamp.time),
+            block_number: timestamp.number,
+            block_date: timestamp.date.clone(),
+            block_hash: timestamp.hash.clone(),
+            index: b.index,
+            blob: bytes_to_hex(&b.blob),
+            kzg_commitment: bytes_to_hex(&b.kzg_commitment),
+            kzg_proof: bytes_to_hex(&b.kzg_proof),
+            kzg_commitment_inclusion_proof: encode_hex_2d_array(&b.kzg_commitment_inclusion_proof),
+        });
+    }
+
+    vec
+}

blocks/beacon-parquet/src/blocks.rs

@@ -0,0 +1,29 @@
+use common::utils::bytes_to_hex;
+
+use crate::{
+    pb::{beacon::rawblocks::Block, sf::beacon::r#type::v1::Block as BeaconBlock},
+    structs::BlockTimestamp,
+};
+
+pub fn collect_blocks(block: &BeaconBlock, spec: &str, timestamp: &BlockTimestamp) -> Vec<Block> {
+    let mut blocks = Vec::new();
+
+    blocks.push(Block {
+        time: Some(timestamp.time),
+        number: timestamp.number,
+        date: timestamp.date.clone(),
+        hash: timestamp.hash.clone(),
+        version: block.version,
+        spec: spec.to_string(),
+        slot: block.slot,
+        parent_slot: block.parent_slot,
+        root: bytes_to_hex(&block.root),
+        parent_root: bytes_to_hex(&block.parent_root),
+        state_root: bytes_to_hex(&block.state_root),
+        proposer_index: block.proposer_index,
+        body_root: bytes_to_hex(&block.body_root),
+        signature: bytes_to_hex(&block.signature),
+    });
+
+    blocks
+}