Merge pull request #76 from enviodev/dz/improve-get-events-join-logic

Improve Get Event join logic

Author: DZakh

examples/call_decode_output/src/main.rs

@@ -71,15 +71,15 @@ async fn main() -> anyhow::Result<()> {
                             {
                                 if !results.is_empty() {
                                     let (balance, _) = results[0].as_uint().unwrap();
-                                    println!("ADDRESS {} : {} DAI", address, balance);
+                                    println!("ADDRESS {address} : {balance} DAI");
                                 }
                             }
                         }
                     }
                 }
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
             }
         }
     }

examples/watch/src/main.rs

@@ -15,7 +15,7 @@ async fn main() {
 
     let height = client.get_height().await.unwrap();
 
-    println!("server height is {}", height);
+    println!("server height is {height}");
 
     // The query to run
     let mut query: Query = serde_json::from_value(serde_json::json!( {

hypersync-client/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hypersync-client"
-version = "0.18.5"
+version = "0.19.0"
 edition = "2021"
 description = "client library for hypersync"
 license = "MPL-2.0"

hypersync-client/src/column_mapping.rs

@@ -204,7 +204,7 @@ fn binary_to_int_str_array(arr: &BinaryArray<i32>) -> Result<Utf8Array<i32>> {
 
 fn binary_to_int_str(binary: &[u8]) -> Result<String> {
     let big_num = I256::try_from_be_slice(binary).context("failed to parse number into I256")?;
-    Ok(format!("{}", big_num))
+    Ok(format!("{big_num}"))
 }
 
 fn map_to_f64(col: &dyn Array) -> Result<Float64Array> {

hypersync-client/src/lib.rs

@@ -29,7 +29,6 @@ pub use hypersync_net_types as net_types;
 pub use hypersync_schema as schema;
 
 use parse_response::parse_query_response;
-use simple_types::Event;
 use tokio::sync::mpsc;
 use types::{EventResponse, ResponseData};
 use url::Url;
@@ -41,6 +40,8 @@ pub use decode::Decoder;
 pub use decode_call::CallDecoder;
 pub use types::{ArrowBatch, ArrowResponse, ArrowResponseData, QueryResponse};
 
+use crate::simple_types::InternalEventJoinStrategy;
+
 type ArrowChunk = Chunk<Box<dyn Array>>;
 
 /// Internal client to handle http requests and retries.
@@ -151,7 +152,8 @@ impl Client {
     ) -> Result<EventResponse> {
         check_simple_stream_params(&config)?;
 
-        add_event_join_fields_to_selection(&mut query);
+        let event_join_strategy = InternalEventJoinStrategy::from(&query.field_selection);
+        event_join_strategy.add_join_fields_to_selection(&mut query.field_selection);
 
         let mut recv = stream::stream_arrow(self, query, config)
             .await
@@ -165,9 +167,9 @@ impl Client {
         while let Some(res) = recv.recv().await {
             let res = res.context("get response")?;
             let res: QueryResponse = QueryResponse::from(&res);
-            let events: Vec<Event> = res.data.into();
+            let events = event_join_strategy.join_from_response_data(res.data);
 
-            data.push(events);
+            data.extend(events);
 
             archive_height = res.archive_height;
             next_block = res.next_block;
@@ -306,10 +308,9 @@ impl Client {
                 Ok(res) => return Ok(res),
                 Err(e) => {
                     log::error!(
-                        "failed to get chain_id from server, retrying... The error was: {:?}",
-                        e
+                        "failed to get chain_id from server, retrying... The error was: {e:?}"
                     );
-                    err = err.context(format!("{:?}", e));
+                    err = err.context(format!("{e:?}"));
                 }
             }
 
@@ -338,10 +339,9 @@ impl Client {
                 Ok(res) => return Ok(res),
                 Err(e) => {
                     log::error!(
-                        "failed to get height from server, retrying... The error was: {:?}",
-                        e
+                        "failed to get height from server, retrying... The error was: {e:?}"
                     );
-                    err = err.context(format!("{:?}", e));
+                    err = err.context(format!("{e:?}"));
                 }
             }
 
@@ -374,9 +374,13 @@ impl Client {
     /// Add block, transaction and log fields selection to the query, executes it with retries
     /// and returns the response.
     pub async fn get_events(&self, mut query: Query) -> Result<EventResponse> {
-        add_event_join_fields_to_selection(&mut query);
+        let event_join_strategy = InternalEventJoinStrategy::from(&query.field_selection);
+        event_join_strategy.add_join_fields_to_selection(&mut query.field_selection);
         let arrow_response = self.get_arrow(&query).await.context("get data")?;
-        Ok(EventResponse::from(&arrow_response))
+        Ok(EventResponse::from_arrow_response(
+            &arrow_response,
+            &event_join_strategy,
+        ))
     }
 
     /// Executes query once and returns the result in (Arrow, size) format.
@@ -430,10 +434,9 @@ impl Client {
                 Ok(res) => return Ok(res),
                 Err(e) => {
                     log::error!(
-                        "failed to get arrow data from server, retrying... The error was: {:?}",
-                        e
+                        "failed to get arrow data from server, retrying... The error was: {e:?}"
                     );
-                    err = err.context(format!("{:?}", e));
+                    err = err.context(format!("{e:?}"));
                 }
             }
 
@@ -493,7 +496,9 @@ impl Client {
     ) -> Result<mpsc::Receiver<Result<EventResponse>>> {
         check_simple_stream_params(&config)?;
 
-        add_event_join_fields_to_selection(&mut query);
+        let event_join_strategy = InternalEventJoinStrategy::from(&query.field_selection);
+
+        event_join_strategy.add_join_fields_to_selection(&mut query.field_selection);
 
         let (tx, rx): (_, mpsc::Receiver<Result<EventResponse>>) =
             mpsc::channel(config.concurrency.unwrap_or(10));
@@ -507,7 +512,9 @@ impl Client {
             while let Some(resp) = inner_rx.recv().await {
                 let is_err = resp.is_err();
                 if tx
-                    .send(resp.map(|r| EventResponse::from(&r)))
+                    .send(
+                        resp.map(|r| EventResponse::from_arrow_response(&r, &event_join_strategy)),
+                    )
                     .await
                     .is_err()
                     || is_err
@@ -551,29 +558,3 @@ fn check_simple_stream_params(config: &StreamConfig) -> Result<()> {
 
     Ok(())
 }
-
-fn add_event_join_fields_to_selection(query: &mut Query) {
-    // Field lists for implementing event based API, these fields are used for joining
-    // so they should always be added to the field selection.
-    const BLOCK_JOIN_FIELDS: &[&str] = &["number"];
-    const TX_JOIN_FIELDS: &[&str] = &["hash"];
-    const LOG_JOIN_FIELDS: &[&str] = &["transaction_hash", "block_number"];
-
-    if !query.field_selection.block.is_empty() {
-        for field in BLOCK_JOIN_FIELDS.iter() {
-            query.field_selection.block.insert(field.to_string());
-        }
-    }
-
-    if !query.field_selection.transaction.is_empty() {
-        for field in TX_JOIN_FIELDS.iter() {
-            query.field_selection.transaction.insert(field.to_string());
-        }
-    }
-
-    if !query.field_selection.log.is_empty() {
-        for field in LOG_JOIN_FIELDS.iter() {
-            query.field_selection.log.insert(field.to_string());
-        }
-    }
-}

hypersync-client/src/parquet_out.rs

@@ -173,7 +173,7 @@ fn spawn_writer(path: PathBuf) -> Result<(mpsc::Sender<ArrowBatch>, JoinHandle<R
         match run_writer(rx, path).await {
             Ok(v) => Ok(v),
             Err(e) => {
-                log::error!("failed to run parquet writer: {:?}", e);
+                log::error!("failed to run parquet writer: {e:?}");
                 Err(e)
             }
         }

hypersync-client/src/simple_types.rs

@@ -6,6 +6,7 @@ use hypersync_format::{
     AccessList, Address, Authorization, BlockNumber, BloomFilter, Data, Hash, LogArgument,
     LogIndex, Nonce, Quantity, TransactionIndex, TransactionStatus, TransactionType, Withdrawal,
 };
+use hypersync_net_types::FieldSelection;
 use nohash_hasher::IntMap;
 use serde::{Deserialize, Serialize};
 use xxhash_rust::xxh3::Xxh3Builder;
@@ -23,8 +24,91 @@ pub struct Event {
     pub log: Log,
 }
 
-impl From<ResponseData> for Vec<Event> {
-    fn from(data: ResponseData) -> Self {
+// Field lists for implementing event based API, these fields are used for joining
+// so they should always be added to the field selection.
+const BLOCK_JOIN_FIELD: &str = "number";
+const TX_JOIN_FIELD: &str = "hash";
+const LOG_JOIN_FIELD_WITH_TX: &str = "transaction_hash";
+const LOG_JOIN_FIELD_WITH_BLOCK: &str = "block_number";
+
+enum InternalJoinStrategy {
+    NotSelected,
+    OnlyLogJoinField,
+    FullJoin,
+}
+
+/// Internal event join strategy for determining how to join blocks and transactions with logs
+pub(crate) struct InternalEventJoinStrategy {
+    block: InternalJoinStrategy,
+    transaction: InternalJoinStrategy,
+}
+
+impl From<&FieldSelection> for InternalEventJoinStrategy {
+    fn from(field_selection: &FieldSelection) -> Self {
+        let block_fields_num = field_selection.block.len();
+        let transaction_fields_num = field_selection.transaction.len();
+
+        Self {
+            block: if block_fields_num == 0 {
+                InternalJoinStrategy::NotSelected
+            } else if block_fields_num == 1 && field_selection.block.contains(BLOCK_JOIN_FIELD) {
+                InternalJoinStrategy::OnlyLogJoinField
+            } else {
+                InternalJoinStrategy::FullJoin
+            },
+            transaction: if transaction_fields_num == 0 {
+                InternalJoinStrategy::NotSelected
+            } else if transaction_fields_num == 1
+                && field_selection.transaction.contains(TX_JOIN_FIELD)
+            {
+                InternalJoinStrategy::OnlyLogJoinField
+            } else {
+                InternalJoinStrategy::FullJoin
+            },
+        }
+    }
+}
+
+impl InternalEventJoinStrategy {
+    /// Add join fields to field selection based on the event join strategy
+    pub(crate) fn add_join_fields_to_selection(&self, field_selection: &mut FieldSelection) {
+        match self.block {
+            InternalJoinStrategy::NotSelected => (),
+            InternalJoinStrategy::OnlyLogJoinField => {
+                field_selection
+                    .log
+                    .insert(LOG_JOIN_FIELD_WITH_BLOCK.to_string());
+                field_selection.block.remove(BLOCK_JOIN_FIELD);
+            }
+            InternalJoinStrategy::FullJoin => {
+                field_selection
+                    .log
+                    .insert(LOG_JOIN_FIELD_WITH_BLOCK.to_string());
+                field_selection.block.insert(BLOCK_JOIN_FIELD.to_string());
+            }
+        }
+
+        match self.transaction {
+            InternalJoinStrategy::NotSelected => (),
+            InternalJoinStrategy::OnlyLogJoinField => {
+                field_selection
+                    .log
+                    .insert(LOG_JOIN_FIELD_WITH_TX.to_string());
+                field_selection.transaction.remove(TX_JOIN_FIELD);
+            }
+            InternalJoinStrategy::FullJoin => {
+                field_selection
+                    .log
+                    .insert(LOG_JOIN_FIELD_WITH_TX.to_string());
+                field_selection
+                    .transaction
+                    .insert(TX_JOIN_FIELD.to_string());
+            }
+        }
+    }
+
+    /// Join response data into events based on the event join strategy
+    pub(crate) fn join_from_response_data(&self, data: ResponseData) -> Vec<Event> {
         let blocks = data
             .blocks
             .into_iter()
@@ -48,10 +132,26 @@ impl From<ResponseData> for Vec<Event> {
             .into_iter()
             .flat_map(|logs| {
                 logs.into_iter().map(|log| {
-                    let block = blocks.get(&log.block_number.unwrap().into()).cloned();
-                    let transaction = transactions
-                        .get(log.transaction_hash.as_ref().unwrap())
-                        .cloned();
+                    let block = match self.block {
+                        InternalJoinStrategy::NotSelected => None,
+                        InternalJoinStrategy::OnlyLogJoinField => Some(Arc::new(Block {
+                            number: Some(log.block_number.unwrap().into()),
+                            ..Block::default()
+                        })),
+                        InternalJoinStrategy::FullJoin => {
+                            blocks.get(&log.block_number.unwrap().into()).cloned()
+                        }
+                    };
+                    let transaction = match self.transaction {
+                        InternalJoinStrategy::NotSelected => None,
+                        InternalJoinStrategy::OnlyLogJoinField => Some(Arc::new(Transaction {
+                            hash: log.transaction_hash.clone(),
+                            ..Transaction::default()
+                        })),
+                        InternalJoinStrategy::FullJoin => transactions
+                            .get(log.transaction_hash.as_ref().unwrap())
+                            .cloned(),
+                    };
 
                     Event {
                         transaction,

hypersync-client/src/types.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use crate::{
-    simple_types::{Block, Event, Log, Trace, Transaction},
+    simple_types::{Block, Event, InternalEventJoinStrategy, Log, Trace, Transaction},
     ArrowChunk, FromArrow,
 };
 use anyhow::{anyhow, Context, Result};
@@ -38,14 +38,18 @@ pub struct ResponseData {
     pub traces: Vec<Vec<Trace>>,
 }
 
-impl From<&'_ ArrowResponse> for EventResponse {
-    fn from(arrow_response: &'_ ArrowResponse) -> Self {
+impl EventResponse {
+    /// Create EventResponse from ArrowResponse with the specified event join strategy
+    pub(crate) fn from_arrow_response(
+        arrow_response: &ArrowResponse,
+        event_join_strategy: &InternalEventJoinStrategy,
+    ) -> Self {
         let r: QueryResponse = arrow_response.into();
         Self {
             archive_height: r.archive_height,
             next_block: r.next_block,
             total_execution_time: r.total_execution_time,
-            data: vec![r.data.into()],
+            data: event_join_strategy.join_from_response_data(r.data),
             rollback_guard: r.rollback_guard,
         }
     }
@@ -114,7 +118,7 @@ pub struct QueryResponse<T = ResponseData> {
 /// Alias for Arrow Query response
 pub type ArrowResponse = QueryResponse<ArrowResponseData>;
 /// Alias for Event oriented, vectorized QueryResponse
-pub type EventResponse = QueryResponse<Vec<Vec<Event>>>;
+pub type EventResponse = QueryResponse<Vec<Event>>;
 
 /// Arrow chunk with schema
 #[derive(Debug, Clone)]

hypersync-client/src/util.rs

@@ -134,8 +134,7 @@ pub fn decode_logs_batch(sig: &str, batch: &ArrowBatch) -> Result<ArrowBatch> {
                     Err(e) => {
                         log::trace!(
                             "failed to decode body of a log, will write null instead. Error was: \
-                             {:?}",
-                            e
+                             {e:?}"
                         );
                         None
                     }