fix: Remove unused ingestor code, rename to DataGenerator (#59)

* fix: Remove unused ingestor code

* fix: Rename ingestor to generator

* chore: auto-fix cargo fmt + clippy

* Update generator.rs

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: peasee

crates/data-generation/src/ingestor.rs crates/data-generation/src/generator.rscrates/data-generation/src/ingestor.rs renamed to crates/data-generation/src/generator.rs

@@ -26,14 +26,14 @@ use super::dataset::Dataset;
 use super::metrics::{IngestResult, Metrics};
 use super::target::Target;
 
-pub struct Ingestor {
+pub struct DataGenerator {
     dataset: Arc<dyn Dataset>,
     target: Arc<dyn Target>,
     metrics: Metrics,
     semaphore: Arc<Semaphore>,
 }
 
-impl Ingestor {
+impl DataGenerator {
     pub fn new(
         dataset: Arc<dyn Dataset>,
         target: Arc<dyn Target>,
@@ -52,38 +52,7 @@ impl Ingestor {
     ///
     /// Pulls one batch per table from the dataset using `next_batches()`, then writes
     /// them sequentially so the data is guaranteed to be present when this returns.
-    ///
-    /// If `table_location_fn` is provided, prints a JSON object mapping each table to
-    /// its connector and location, e.g.:
-    /// `{"customer": {"connector": "s3", "location": "s3://bucket/prefix/customer/"}, ...}`
-    pub async fn initialize(
-        &self,
-        table_location_fn: Option<&dyn Fn(&str) -> String>,
-    ) -> anyhow::Result<IngestResult> {
-        // Print table locations as JSON
-        if let Some(loc_fn) = table_location_fn {
-            let mut map = serde_json::Map::new();
-            for (name, table) in self.dataset.tables() {
-                let mut entry = serde_json::Map::new();
-                entry.insert(
-                    "connector".to_string(),
-                    serde_json::Value::String("s3".to_string()),
-                );
-                entry.insert(
-                    "location".to_string(),
-                    serde_json::Value::String(loc_fn(&name)),
-                );
-                if let Some(ref time_col) = table.time_column {
-                    entry.insert(
-                        "time_column".to_string(),
-                        serde_json::Value::String(time_col.clone()),
-                    );
-                }
-                map.insert(name, serde_json::Value::Object(entry));
-            }
-            println!("{}", serde_json::Value::Object(map));
-        }
-
+    pub async fn initialize(&self) -> anyhow::Result<IngestResult> {
         let table_count = self.dataset.tables().len();
 
         tracing::info!(

crates/data-generation/src/lib.rs

@@ -16,7 +16,7 @@ limitations under the License.
 
 pub mod config;
 pub mod dataset;
-pub mod ingestor;
+pub mod generator;
 pub mod metrics;
 pub mod source;
 pub mod target;

crates/data-generation/src/main.rs

@@ -23,7 +23,7 @@ use std::sync::Arc;
 use data_generation::config::{Cli, Command, CommonArgs};
 use data_generation::dataset;
 use data_generation::dataset::tpch::TpchDataset;
-use data_generation::ingestor::Ingestor;
+use data_generation::generator::DataGenerator;
 use data_generation::metrics::{IngestResult, Metrics};
 use data_generation::target::s3::S3Target;
 
@@ -50,7 +50,7 @@ fn print_summary(result: &IngestResult) {
     println!("  Avg write latency: {:?}", result.avg_write_latency);
 }
 
-fn build(args: &CommonArgs) -> anyhow::Result<(Ingestor, Arc<S3Target>)> {
+fn build(args: &CommonArgs) -> anyhow::Result<DataGenerator> {
     let dataset_config = args.dataset_config();
     let target_config = args.target_config();
     let ingestor_config = args.ingestor_config();
@@ -72,13 +72,13 @@ fn build(args: &CommonArgs) -> anyhow::Result<(Ingestor, Arc<S3Target>)> {
     let target = Arc::new(S3Target::new(&target_config)?);
     let metrics = Metrics::new();
 
-    let ingestor = Ingestor::new(
+    let ingestor = DataGenerator::new(
         dataset,
-        Arc::clone(&target) as Arc<dyn Target>,
+        target as Arc<dyn Target>,
         &ingestor_config,
         metrics,
     );
-    Ok((ingestor, target))
+    Ok(ingestor)
 }
 
 #[tokio::main]
@@ -91,16 +91,15 @@ async fn main() -> anyhow::Result<()> {
 
     match cli.command {
         Command::Initialize(args) => {
-            let (ingestor, target) = build(&args)?;
-            let loc_fn = |table: &str| target.table_s3_path(table);
-            let result = ingestor.initialize(Some(&loc_fn)).await?;
+            let ingestor = build(&args)?;
+            let result = ingestor.initialize().await?;
 
             if result.write_errors > 0 {
                 anyhow::bail!("Initialization failed with {} errors", result.write_errors);
             }
         }
         Command::Run(run_args) => {
-            let (ingestor, _target) = build(&run_args.common)?;
+            let ingestor = build(&run_args.common)?;
             if run_args.skip_initial {
                 ingestor.skip_initial_batches().await?;
             }