feat: ClickHouse batch size config (#406)

Author: ExoMonk

core/src/database/clickhouse/batch_operations/dynamic.rs

@@ -22,7 +22,7 @@ pub async fn execute_dynamic_batch_operation(
         return Ok(());
     }
 
-    for batch in rows.chunks(1000) {
+    for batch in rows.chunks(database.batch_size()) {
         execute_batch(database, table_name, op_type, batch).await.map_err(|e| {
             tracing::error!("{} - Batch operation failed: {}", event_name, e);
             e

core/src/database/clickhouse/batch_operations/macros.rs

@@ -211,7 +211,7 @@ macro_rules! create_batch_clickhouse_operation {
                 Ok(())
             }
 
-            for batch in filtered_results.chunks(1000) {
+            for batch in filtered_results.chunks(database.batch_size()) {
                 if let Err(e) = execute_batch(database, batch).await {
                     rindexer_error!("{} - Batch operation failed: {}", $event_name, e);
                     return Err(e);

core/src/database/clickhouse/client.rs

@@ -8,6 +8,9 @@ use tracing::info;
 use crate::metrics::database::{self as db_metrics, ops};
 use crate::EthereumSqlTypeWrapper;
 
+const DEFAULT_CLICKHOUSE_BATCH_SIZE: usize = 1000;
+const CLICKHOUSE_BATCH_SIZE_ENV: &str = "RINDEXER_CLICKHOUSE_BATCH_SIZE";
+
 pub struct ClickhouseConnection {
     url: String,
     user: String,
@@ -45,11 +48,35 @@ pub enum ClickhouseError {
 
 pub struct ClickhouseClient {
     pub(crate) conn: Client,
+    batch_size: usize,
+}
+
+fn parse_clickhouse_batch_size() -> usize {
+    parse_clickhouse_batch_size_value(env::var(CLICKHOUSE_BATCH_SIZE_ENV).ok())
+}
+
+fn parse_clickhouse_batch_size_value(value: Option<String>) -> usize {
+    match value {
+        Some(raw) => match raw.parse::<usize>() {
+            Ok(parsed) if parsed > 0 => parsed,
+            _ => {
+                tracing::warn!(
+                    "{} is invalid (value: {:?}); using default {}",
+                    CLICKHOUSE_BATCH_SIZE_ENV,
+                    raw,
+                    DEFAULT_CLICKHOUSE_BATCH_SIZE
+                );
+                DEFAULT_CLICKHOUSE_BATCH_SIZE
+            }
+        },
+        None => DEFAULT_CLICKHOUSE_BATCH_SIZE,
+    }
 }
 
 impl ClickhouseClient {
     pub async fn new() -> Result<Self, ClickhouseConnectionError> {
         let connection = clickhouse_connection()?;
+        let batch_size = parse_clickhouse_batch_size();
 
         let client = Client::default()
             .with_url(connection.url)
@@ -58,9 +85,13 @@ impl ClickhouseClient {
             .with_password(connection.password);
 
         client.query("select 1").execute().await?;
-        info!("Clickhouse client connected successfully!");
+        info!("Clickhouse client connected successfully! dynamic batch size={}", batch_size);
+
+        Ok(ClickhouseClient { conn: client, batch_size })
+    }
 
-        Ok(ClickhouseClient { conn: client })
+    pub fn batch_size(&self) -> usize {
+        self.batch_size
     }
 
     pub async fn query_one<T>(&self, sql: &str) -> Result<T, ClickhouseError>
@@ -177,3 +208,30 @@ impl ClickhouseClient {
         self.bulk_insert_via_query(table_name, column_names, bulk_data).await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{parse_clickhouse_batch_size_value, DEFAULT_CLICKHOUSE_BATCH_SIZE};
+
+    #[test]
+    fn clickhouse_batch_size_defaults_when_env_missing() {
+        assert_eq!(parse_clickhouse_batch_size_value(None), DEFAULT_CLICKHOUSE_BATCH_SIZE);
+    }
+
+    #[test]
+    fn clickhouse_batch_size_reads_positive_env_override() {
+        assert_eq!(parse_clickhouse_batch_size_value(Some("5000".to_string())), 5000);
+    }
+
+    #[test]
+    fn clickhouse_batch_size_rejects_zero_or_invalid_values() {
+        assert_eq!(
+            parse_clickhouse_batch_size_value(Some("0".to_string())),
+            DEFAULT_CLICKHOUSE_BATCH_SIZE
+        );
+        assert_eq!(
+            parse_clickhouse_batch_size_value(Some("invalid".to_string())),
+            DEFAULT_CLICKHOUSE_BATCH_SIZE
+        );
+    }
+}

documentation/docs/pages/docs/changelog.mdx

@@ -9,6 +9,7 @@
 ### Features
 -------------------------------------------------
 - feat: Add Twilio SMS alerts support — send SMS notifications for on-chain events via the Twilio API
+- feat: **`RINDEXER_CLICKHOUSE_BATCH_SIZE` env var** — configure the ClickHouse batch chunk size used for no-code/custom table writes. Defaults to `1000`. Increasing it reduces the number of sequential `INSERT` statements for high-volume streams, at the cost of larger per-request payloads.
 
 ## Releases
 -------------------------------------------------
@@ -36,7 +37,6 @@ github branch - https://github.com/joshstevens19/rindexer/tree/release/0.38.0
 
 - feat: **`database` field on custom tables** — optional YAML field that directs a custom table to a specific ClickHouse database (or PostgreSQL schema) instead of the default `{project}_{contract}` naming. Enables multiple contracts to write to a shared table (e.g., `database: indexer` → `indexer.events`).
 
-
 # 0.37.2-beta - 1st April 2026
 
 github branch - https://github.com/joshstevens19/rindexer/tree/release/0.37.2

documentation/docs/pages/docs/start-building/yaml-config/storage.mdx

@@ -1010,8 +1010,13 @@ CLICKHOUSE_URL="http://[host]:[port]"
 CLICKHOUSE_DB="default"
 CLICKHOUSE_USER="default"
 CLICKHOUSE_PASSWORD="default"
+RINDEXER_CLICKHOUSE_BATCH_SIZE="1000"
 ```
 
+`RINDEXER_CLICKHOUSE_BATCH_SIZE` controls the chunk size used when rindexer writes dynamic/no-code ClickHouse batches. The default is `1000`.
+
+For high-volume streams, increasing this value reduces the number of sequential ClickHouse `INSERT` requests. The tradeoff is that each request becomes larger, so values should be increased carefully based on the workload and ClickHouse capacity.
+
 ### enabled
 
 If clickhouse is enabled or not, if you do not wish to use clickhouse you can set this to false or remove clickhouse from the storage completely.