working

Author: jayshrivastava

src/bin/fuzz.rs

@@ -21,9 +21,9 @@ struct Cli {
 
 /// Trait for defining workloads that can be fuzzed
 #[async_trait]
-trait Workload: Send + Sync {
+trait Workload {
     /// Setup the workload by registering tables and preparing data
-    async fn setup(&self, ctx: &SessionContext) -> Result<()>;
+    async fn setup(ctx: SessionContext) -> Result<()>;
 
     /// Get queries as an iterator of (query_name, query_sql) pairs  
     async fn queries(&self) -> Result<Vec<(String, String)>>;
@@ -70,10 +70,9 @@ impl TpcdsWorkload {
 
 #[async_trait]
 impl Workload for TpcdsWorkload {
-    async fn setup(&self, ctx: &SessionContext) -> Result<()> {
-        // Register tables
-        info!("🔧 Registering TPC-DS tables...");
-        let (registered_tables, missing_tables) = register_available_tpcds_tables(ctx, None).await?;
+    async fn setup(ctx: SessionContext) -> Result<()> {
+        info!("Registering TPC-DS tables...");
+        let (registered_tables, missing_tables) = register_available_tpcds_tables(&ctx, None).await?;
 
         if !missing_tables.is_empty() {
             debug!("Missing TPCDS tables: {:?}", missing_tables);
@@ -144,21 +143,14 @@ fn validate_tpcds_args(scale_factor: &str) -> std::result::Result<(), String> {
 }
 
 /// Run a workload using the generic workload trait
-async fn run_workload(workload: TpcdsWorkload) -> Result<()> {
+async fn run_workload<W>(workload: W) -> Result<()> 
+where W: Workload {
     info!("📊 Running {} workload...", workload.name());
 
     // Create FuzzDB with randomized session config
     info!("⚙️  Setting up distributed session with randomized configuration...");
 
-    let fuzz_db = match FuzzDB::new({
-        let workload_clone = workload.clone();
-        move |ctx| {
-            let workload_inner = workload_clone.clone();
-            async move {
-                workload_inner.setup(&ctx).await
-            }
-        }
-    }).await {
+    let fuzz_db = match FuzzDB::new(W::setup).await {
         Ok(db) => db,
         Err(e) => {
             return Err(datafusion::error::DataFusionError::Execution(format!(
@@ -183,7 +175,7 @@ async fn run_workload(workload: TpcdsWorkload) -> Result<()> {
     let start_time = Instant::now();
 
     for (query_name, query_sql) in all_queries {
-        debug!("🔄 Executing query: {}", query_name);
+        info!("Executing query: {}", query_name);
         debug!("Executing fuzz query: {}", query_sql.trim());
 
         let query_start = Instant::now();