datafusion_ethers/convert/
hybrid.rs

1use alloy::json_abi::Event;
2use alloy::rpc::types::eth::Log;
3use datafusion::arrow::array::RecordBatch;
4use datafusion::arrow::datatypes::{DataType, Field, SchemaBuilder, SchemaRef};
5use std::sync::Arc;
6
7use super::{AppendError, Transcoder};
8
9///////////////////////////////////////////////////////////////////////////////////////////////////
10
11/// Transcodes raw Ethereum log events into Arrow record batches while also creating a nested
12/// struct with fields decoded according to the provided log signature
13pub struct EthRawAndDecodedLogsToArrow {
14    schema: SchemaRef,
15    raw: super::EthRawLogsToArrow,
16    decoded: super::EthDecodedLogsToArrow,
17}
18
19impl EthRawAndDecodedLogsToArrow {
20    pub fn new(event_type: &Event) -> Self {
21        let raw = super::EthRawLogsToArrow::new();
22        let decoded = super::EthDecodedLogsToArrow::new(event_type);
23
24        let mut builder = SchemaBuilder::from(&raw.schema().fields);
25        builder.push(Field::new(
26            "event",
27            DataType::Struct(decoded.schema().fields.clone()),
28            false,
29        ));
30        Self {
31            schema: Arc::new(builder.finish()),
32            raw,
33            decoded,
34        }
35    }
36
37    pub fn new_from_signature(signature: &str) -> Result<Self, alloy::dyn_abi::parser::Error> {
38        let event_type = alloy::json_abi::Event::parse(signature)?;
39        Ok(Self::new(&event_type))
40    }
41}
42
43impl Transcoder for EthRawAndDecodedLogsToArrow {
44    fn schema(&self) -> SchemaRef {
45        self.schema.clone()
46    }
47
48    #[allow(clippy::get_first)]
49    fn append(&mut self, logs: &[Log]) -> Result<(), AppendError> {
50        self.raw.append(logs)?;
51        self.decoded.append(logs)?;
52        Ok(())
53    }
54
55    fn len(&self) -> usize {
56        self.raw.len()
57    }
58
59    fn finish(&mut self) -> RecordBatch {
60        let raw = self.raw.finish();
61        let decoded = self.decoded.finish();
62
63        let event_col = datafusion::arrow::array::StructArray::new(
64            decoded.schema().fields().clone(),
65            decoded.columns().to_vec(),
66            None,
67        );
68
69        let mut columns = raw.columns().to_vec();
70        columns.push(Arc::new(event_col));
71
72        RecordBatch::try_new(self.schema.clone(), columns).unwrap()
73    }
74}
75
76///////////////////////////////////////////////////////////////////////////////////////////////////
datafusion_ethers/convert/hybrid.rs

datafusion_ethers/convert/
hybrid.rs
