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        .as_any()
47        .downcast_ref::<super::provider::EthGetLogs>()
48        .map(|scan| scan.filter().clone());
49
50    // Traverse all the children too to make sure there is only one scan in this query (e.g. no UNIONs)
51    for child in &plan.children() {
52        let child = sql_to_pushdown_filter_rec(child.as_ref());
53        if child.is_some() {
54            if found.is_some() {
55                unimplemented!("Multiple table scans in one query are not yet supported");
56            }
57            found = child;
58        }
59    }
60
61    found
62}
datafusion_ethers/convert/mod.rs

datafusion_ethers/convert/
mod.rs
