etop_core/datasets/
blocks.rs

1use crate::{DataSpec, DataWarehouse, EtopError, InputDataset};
2use etop_format::ColumnFormatShorthand;
3use polars::prelude::*;
4use std::collections::HashMap;
5
6/// blocks
7#[derive(Clone)]
8pub struct Blocks;
9
10impl DataSpec for Blocks {
11    fn name(&self) -> String {
12        "blocks".to_string()
13    }
14
15    fn row_noun(&self) -> String {
16        "blocks".to_string()
17    }
18
19    fn inputs(&self) -> Vec<InputDataset> {
20        vec![InputDataset::Raw("blocks".into()), InputDataset::Raw("transactions".into())]
21    }
22
23    fn transform(
24        &self,
25        warehouse: &DataWarehouse,
26        start_block: Option<u32>,
27        end_block: Option<u32>,
28    ) -> Result<DataFrame, EtopError> {
29        let sort = SortOptions {
30            descending: true,
31            nulls_last: true,
32            multithreaded: true,
33            maintain_order: true,
34        };
35        let join_args = JoinArgs {
36            how: JoinType::Left,
37            validation: JoinValidation::ManyToMany,
38            suffix: None,
39            slice: None,
40        };
41
42        let txs = warehouse
43            .get_dataset("transactions")?
44            .clone()
45            .lazy()
46            .group_by(["block_number"])
47            .agg([count().alias("n_txs")])
48            .collect()?;
49        let blocks = warehouse
50            .get_dataset("blocks")?
51            .clone()
52            .lazy()
53            .with_column(col("base_fee_per_gas") / lit(1e9))
54            .sort("block_number", sort)
55            .collect()
56            .map_err(EtopError::PolarsError)?;
57
58        let blocks = crate::filter_by_block_number(blocks, start_block, end_block)?;
59
60        blocks
61            .clone()
62            .lazy()
63            .join(txs.lazy(), [col("block_number")], [col("block_number")], join_args)
64            .collect()
65            .map_err(EtopError::PolarsError)
66    }
67
68    fn default_columns(&self) -> Option<Vec<String>> {
69        let columns =
70            ["block_number", "timestamp", "n_txs", "gas_used", "base_fee_per_gas", "author"]
71                .iter()
72                .map(|s| s.to_string())
73                .collect();
74        Some(columns)
75    }
76
77    fn default_column_formats(&self) -> Option<HashMap<String, ColumnFormatShorthand>> {
78        let integer_oom = etop_format::NumberFormat::new().integer_oom().precision(1);
79        let float_oom = etop_format::NumberFormat::new().float_oom().precision(1);
80        let timestamp_fmt = etop_format::NumberFormat::new().timestamp();
81
82        let formats = vec![
83            ColumnFormatShorthand::new().name("block_number").newline_underscores(),
84            ColumnFormatShorthand::new().name("timestamp").set_format(timestamp_fmt),
85            ColumnFormatShorthand::new().name("n_txs").set_format(integer_oom.clone()),
86            ColumnFormatShorthand::new()
87                .name("gas_used")
88                .set_format(integer_oom)
89                .min_width(5)
90                .newline_underscores(),
91            ColumnFormatShorthand::new()
92                .name("base_fee_per_gas")
93                .display_name("base_fee")
94                .set_format(float_oom)
95                .min_width(5)
96                .newline_underscores(),
97            ColumnFormatShorthand::new().name("author"),
98        ]
99        .into_iter()
100        .map(|column| (column.name.clone(), column))
101        .collect();
102
103        Some(formats)
104    }
105}
etop_core/datasets/blocks.rs

etop_core/datasets/
blocks.rs
