better queue batch size limits

Author: tonyalaribe

src/batch_queue.rs

@@ -3,8 +3,8 @@ use delta_kernel::arrow::record_batch::RecordBatch;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
-use tokio_stream::StreamExt;
 use tokio_stream::wrappers::ReceiverStream;
+use tokio_stream::StreamExt;
 use tracing::{error, info};
 
 #[derive(Debug)]
@@ -16,7 +16,10 @@ pub struct BatchQueue {
 impl BatchQueue {
     pub fn new(db: Arc<crate::database::Database>, interval_ms: u64, max_rows: usize) -> Self {
         // Make channel capacity configurable via environment variable
-        let channel_capacity = std::env::var("TIMEFUSION_BATCH_QUEUE_CAPACITY").unwrap_or_else(|_| "1000".to_string()).parse::<usize>().unwrap_or(1000);
+        let channel_capacity = std::env::var("TIMEFUSION_BATCH_QUEUE_CAPACITY")
+            .unwrap_or_else(|_| "100000000".to_string())
+            .parse::<usize>()
+            .unwrap_or(100_000_000);
 
         let (tx, rx) = mpsc::channel(channel_capacity);
         let shutdown = tokio_util::sync::CancellationToken::new();

src/main.rs

@@ -6,7 +6,7 @@ use dotenv::dotenv;
 use std::{env, sync::Arc};
 use timefusion::batch_queue::BatchQueue;
 use timefusion::database::Database;
-use tokio::time::{Duration, sleep};
+use tokio::time::{sleep, Duration};
 use tracing::{error, info};
 use tracing_subscriber::EnvFilter;
 
@@ -24,8 +24,8 @@ async fn main() -> anyhow::Result<()> {
 
     // Setup batch processing with configurable params
     let interval_ms = env::var("BATCH_INTERVAL_MS").ok().and_then(|v| v.parse().ok()).unwrap_or(1000);
-    let max_size = env::var("MAX_BATCH_SIZE").ok().and_then(|v| v.parse().ok()).unwrap_or(1000);
-    let enable_queue = env::var("ENABLE_BATCH_QUEUE").unwrap_or_else(|_| "false".to_string()) == "true";
+    let max_size = env::var("MAX_BATCH_SIZE").ok().and_then(|v| v.parse().ok()).unwrap_or(100_000);
+    let enable_queue = env::var("ENABLE_BATCH_QUEUE").unwrap_or_else(|_| "true".to_string()) == "true";
 
     // Create batch queue
     let batch_queue = Arc::new(BatchQueue::new(Arc::new(db.clone()), interval_ms, max_size));