datafusion_ethers/convert/
mod.rs

1mod decoded;
2mod hybrid;
3mod raw;
4
5use alloy_core::dyn_abi::Error as DynAbiError;
6use alloy_rpc_types_eth::{Filter, Log};
7use datafusion::arrow::array::RecordBatch;
8use datafusion::arrow::datatypes::SchemaRef;
9pub use decoded::*;
10pub use hybrid::*;
11pub use raw::*;
12
13use datafusion::error::Result as DfResult;
14use datafusion::execution::context::SessionContext;
15use datafusion::physical_plan::ExecutionPlan;
16
17///////////////////////////////////////////////////////////////////////////////////////////////////
18
19pub trait Transcoder {
20    fn schema(&self) -> SchemaRef;
21    fn append(&mut self, logs: &[Log]) -> Result<(), AppendError>;
22    fn len(&self) -> usize;
23    fn finish(&mut self) -> RecordBatch;
24    fn is_empty(&self) -> bool {
25        self.len() == 0
26    }
27}
28
29#[derive(Debug, thiserror::Error)]
30pub enum AppendError {
31    #[error(transparent)]
32    EventDecodingError(#[from] DynAbiError),
33}
34
35///////////////////////////////////////////////////////////////////////////////////////////////////
36
37/// Analyzes the SQL and returns a pushed-down filter that will be used when querying logs from the ETH node
38pub async fn sql_to_pushdown_filter(ctx: &SessionContext, sql: &str) -> DfResult<Option<Filter>> {
39    let df = ctx.sql(sql).await?;
40    let plan = df.create_physical_plan().await?;
41    Ok(sql_to_pushdown_filter_rec(plan.as_ref()))
42}
43
44fn sql_to_pushdown_filter_rec(plan: &dyn ExecutionPlan) -> Option<Filter> {
45    let mut found = plan
46        .downcast_ref::<super::provider::EthGetLogs>()
47        .map(|scan| scan.filter().clone());
48
49    // Traverse all the children too to make sure there is only one scan in this query (e.g. no UNIONs)
50    for child in &plan.children() {
51        let child = sql_to_pushdown_filter_rec(child.as_ref());
52        if child.is_some() {
53            if found.is_some() {
54                unimplemented!("Multiple table scans in one query are not yet supported");
55            }
56            found = child;
57        }
58    }
59
60    found
61}
datafusion_ethers/convert/mod.rs

datafusion_ethers/convert/
mod.rs
