Merge pull request #1666 from input-output-hk/ensemble/1656/use-sqlite-transactions-insert-ctx

Add benchmark tests for storing Cardano transactions

Author: dlachaume

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-persistence"
-version = "0.1.8"
+version = "0.1.9"
 description = "Common types, interfaces, and utilities to persist data for Mithril nodes."
 authors = { workspace = true }
 edition = { workspace = true }

internal/mithril-persistence/Cargo.toml

@@ -70,20 +70,20 @@ impl WhereCondition {
 
     /// Turn the condition into a SQL string representation.
     pub fn expand(self) -> (String, Vec<Value>) {
-        let mut expression = self.condition.expand();
+        let expression = self.condition.expand();
         let parameters = self.parameters;
-        let mut param_index = 1;
         //
         // Replace parameters placeholders by numerated parameters.
-        loop {
-            if !expression.contains("?*") {
-                break;
+        let mut final_expression = "".to_string();
+        for (param_index, sql_part) in expression.split("?*").enumerate() {
+            if param_index > 0 {
+                final_expression.push('?');
+                final_expression.push_str(&param_index.to_string());
             }
-            expression = expression.replacen("?*", &format!("?{param_index}"), 1);
-            param_index += 1;
+            final_expression.push_str(sql_part);
         }
 
-        (expression, parameters)
+        (final_expression, parameters)
     }
 
     /// Instanciate a condition with a `IN` statement.

internal/mithril-persistence/src/sqlite/condition.rs

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.5.3"
+version = "0.5.4"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }
@@ -9,6 +9,10 @@ homepage = { workspace = true }
 license = { workspace = true }
 repository = { workspace = true }
 
+[[bench]]
+name = "cardano_transactions_import"
+harness = false
+
 [dependencies]
 anyhow = "1.0.79"
 async-trait = "0.1.77"
@@ -50,6 +54,7 @@ zstd = { version = "0.13.0", features = ["zstdmt"] }
 tikv-jemallocator = { version = "0.5.4", optional = true }
 
 [dev-dependencies]
+criterion = { version = "0.5.1", features = ["html_reports", "async_tokio"] }
 httpmock = "0.7.0"
 mithril-common = { path = "../mithril-common", features = [
     "allow_skip_signer_certification",

mithril-aggregator/Cargo.toml

@@ -0,0 +1,63 @@
+use criterion::{criterion_group, criterion_main, Criterion};
+use sqlite::ConnectionThreadSafe;
+use std::sync::Arc;
+
+use mithril_aggregator::{
+    database::repository::CardanoTransactionRepository, services::TransactionStore,
+};
+use mithril_common::{entities::CardanoTransaction, test_utils::TempDir};
+use mithril_persistence::sqlite::ConnectionBuilder;
+
+fn cardano_tx_db_connection() -> ConnectionThreadSafe {
+    let db_path =
+        TempDir::create("aggregator_benches", "bench_store_transactions").join("cardano_tx.db");
+
+    if db_path.exists() {
+        std::fs::remove_file(db_path.clone()).unwrap();
+    }
+
+    ConnectionBuilder::open_file(&db_path)
+        .with_migrations(
+            mithril_aggregator::database::cardano_transaction_migration::get_migrations(),
+        )
+        .build()
+        .unwrap()
+}
+
+fn generate_transactions(nb_transactions: usize) -> Vec<CardanoTransaction> {
+    (0..nb_transactions)
+        .map(|i| {
+            CardanoTransaction::new(
+                format!("tx_hash-{}", i),
+                i as u64,
+                i as u64 + 1,
+                format!("block_hash-{}", i),
+                i as u64 + 2,
+            )
+        })
+        .collect()
+}
+
+fn bench_store_transactions(c: &mut Criterion) {
+    const NB_CARDANO_TRANSACTIONS: usize = 1_000_000;
+    let runtime = tokio::runtime::Runtime::new().unwrap();
+    let transactions = generate_transactions(NB_CARDANO_TRANSACTIONS);
+
+    let mut group = c.benchmark_group("Store transactions");
+    group.bench_function("store_transactions", |bencher| {
+        bencher.to_async(&runtime).iter(|| async {
+            let connection = Arc::new(cardano_tx_db_connection());
+            let repository = CardanoTransactionRepository::new(connection);
+            repository.store_transactions(transactions.clone()).await
+        });
+    });
+
+    group.finish();
+}
+
+criterion_group! {
+    name = benches;
+    config = Criterion::default().sample_size(10);
+    targets = bench_store_transactions
+}
+criterion_main!(benches);

mithril-aggregator/benches/cardano_transactions_import.rs

@@ -242,11 +242,18 @@ impl TransactionStore for CardanoTransactionRepository {
     }
 
     async fn store_transactions(&self, transactions: Vec<CardanoTransaction>) -> StdResult<()> {
-        // Chunk transactions to avoid an error when we exceed sqlite binding limitations
-        for transactions_in_chunk in transactions.chunks(100) {
-            self.create_transactions(transactions_in_chunk.to_vec())
-                .await
-                .with_context(|| "CardanoTransactionRepository can not store transactions")?;
+        const DB_TRANSACTION_SIZE: usize = 100000;
+        for transactions_in_db_transaction_chunk in transactions.chunks(DB_TRANSACTION_SIZE) {
+            self.connection.execute("BEGIN TRANSACTION;")?;
+
+            // Chunk transactions to avoid an error when we exceed sqlite binding limitations
+            for transactions_in_chunk in transactions_in_db_transaction_chunk.chunks(100) {
+                self.create_transactions(transactions_in_chunk.to_vec())
+                    .await
+                    .with_context(|| "CardanoTransactionRepository can not store transactions")?;
+            }
+
+            self.connection.execute("END TRANSACTION;")?;
         }
         Ok(())
     }

mithril-aggregator/src/database/repository/cardano_transaction_repository.rs

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-signer"
-version = "0.2.131"
+version = "0.2.132"
 description = "A Mithril Signer"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-signer/Cargo.toml

@@ -236,11 +236,18 @@ impl TransactionStore for CardanoTransactionRepository {
     }
 
     async fn store_transactions(&self, transactions: Vec<CardanoTransaction>) -> StdResult<()> {
-        // Chunk transactions to avoid an error when we exceed sqlite binding limitations
-        for transactions_in_chunk in transactions.chunks(100) {
-            self.create_transactions(transactions_in_chunk.to_vec())
-                .await
-                .with_context(|| "CardanoTransactionRepository can not store transactions")?;
+        const DB_TRANSACTION_SIZE: usize = 100000;
+        for transactions_in_db_transaction_chunk in transactions.chunks(DB_TRANSACTION_SIZE) {
+            self.connection.execute("BEGIN TRANSACTION;")?;
+
+            // Chunk transactions to avoid an error when we exceed sqlite binding limitations
+            for transactions_in_chunk in transactions_in_db_transaction_chunk.chunks(100) {
+                self.create_transactions(transactions_in_chunk.to_vec())
+                    .await
+                    .with_context(|| "CardanoTransactionRepository can not store transactions")?;
+            }
+
+            self.connection.execute("END TRANSACTION;")?;
         }
         Ok(())
     }