fix: ETL run supports not initializing first (#143)

Author: peasee

crates/etl/src/lib.rs

@@ -1195,14 +1195,6 @@ impl ETLPipeline {
     ///
     /// Returns an error if the pipeline is not in the [`Initialized`] state.
     pub async fn run(&mut self, step_count: usize) -> anyhow::Result<()> {
-        let current_state = self.state_rx.borrow().clone();
-        if current_state != PipelineState::Initialized {
-            anyhow::bail!(
-                "Cannot run pipeline: current state is {:?} (must be Initialized)",
-                current_state
-            );
-        }
-
         self.batch_budget = Some(step_count);
         self.build_work_plan().await;
         self.spawn_run_task(Some(step_count));
@@ -1248,9 +1240,17 @@ impl ETLPipeline {
         let tables = dataset.tables();
         let mut steps: BTreeMap<u64, Vec<String>> = BTreeMap::new();
 
+        // Only skip the first batch ID per table if initialize() was called
+        let skip_first = *self.state_rx.borrow() == PipelineState::Initialized;
+
         for name in tables.keys() {
             let ids = dataset.clone().batch_ids(name).await;
-            let initialized_id = ids.front().copied();
+            let initialized_id = if skip_first {
+                ids.front().copied()
+            } else {
+                None
+            };
+
             let mut seen_ids = HashSet::new();
 
             for id in ids {