Fix ETL step counting, ADBC initialization, checkpoint validation, and convergence timing (#165)

- Fix ETL coalescing race condition: use outer_steps_processed instead of
  logical_steps_consumed for step limit check. Coalescing internally consumes
  steps from future batches, inflating logical_steps_consumed and causing the
  pipeline to stop one step early.

- Add missing pipeline.initialize() call in ADBC sink path, ensuring batch 0
  is loaded for all tables before the pipeline run starts.

- Widen schema equivalence in checkpoint validation: allow Decimal128 with
  same scale but different precision (DuckDB vs DataFusion), and add reverse
  Decimal128 to Int64/Float64 matching.

- Measure checkpoint convergence latency from the first passing query rather
  than from when the full validation sweep completes. The first pass
  indicates when the data was correctly ingested; the remaining queries
  confirm the result. Replace fixed 5s polling with watch channel
  notification.

Checkpoint expected data in S3 regenerated for versions 1, 2, and 3.

Author: phillipleblanc

crates/etl/src/lib.rs

@@ -1463,12 +1463,19 @@ async fn run_pipeline(
     let mut outer_steps_processed: usize = 0;
 
     loop {
-        // Check step budget.
+        // Check step budget using outer_steps_processed rather than
+        // logical_steps_consumed so that steps consumed internally by
+        // batch coalescing (read_batches_until_min_rows) do not count
+        // against the budget.  logical_steps_consumed can be incremented
+        // when a coalescing reservation removes the last table from a
+        // future step, effectively "consuming" that step without the outer
+        // loop ever processing it for the remaining tables.
         if let Some(limit) = step_limit
-            && logical_steps_consumed.load(Ordering::Relaxed) >= limit as u64
+            && outer_steps_processed >= limit
         {
             info!(
-                steps_processed = logical_steps_consumed.load(Ordering::Relaxed),
+                outer_steps_processed,
+                logical_steps_consumed = logical_steps_consumed.load(Ordering::Relaxed),
                 "Step limit reached, pausing pipeline"
             );
             progress_logger.abort();

crates/test-framework/src/queries/validation/mod.rs

@@ -145,6 +145,10 @@ fn datatype_equivalent(expected_type: &DataType, actual_type: &DataType) -> bool
                 (None, Some("UTC" | "+00:00")) | (Some("UTC" | "+00:00"), None)
             )
         }
+        // Decimal128 with same scale but different precision (e.g. DuckDB returns
+        // Decimal128(38,2) while DataFusion returns Decimal128(25,2)).  The stringified
+        // values are identical because scale controls the fractional digits.
+        (DataType::Decimal128(_, s1), DataType::Decimal128(_, s2)) => s1 == s2,
         // Existing numeric and string type equivalences
         _ => matches!(
             (expected_type, actual_type),

crates/test-framework/src/spicetest/datasets/checkpoint_validation.rs

@@ -28,6 +28,7 @@ limitations under the License.
 
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Instant;
 
 use arrow::array::RecordBatch;
 
@@ -85,6 +86,12 @@ pub enum ValidationStatus {
         /// every validated query passing. Once set, stays `true` for the
         /// remainder of this validation window.
         converged: bool,
+        /// The instant at which the first query passed in the iteration
+        /// that ultimately converged.  `None` until a pass is recorded in
+        /// the current (non-converged) iteration.  Reset at the start of
+        /// each new iteration so that a failing iteration's timestamp is
+        /// discarded.
+        first_pass_instant: Option<Instant>,
     },
 }
 
@@ -131,6 +138,18 @@ impl ValidationStatus {
             } => *completed_iterations,
         }
     }
+
+    /// Returns the [`Instant`] at which the first query passed in the
+    /// converging iteration, or `None` if not yet available.
+    #[must_use]
+    pub fn first_pass_instant(&self) -> Option<Instant> {
+        match self {
+            ValidationStatus::Inactive => None,
+            ValidationStatus::Active {
+                first_pass_instant, ..
+            } => *first_pass_instant,
+        }
+    }
 }
 
 /// Handles for the load runner to interact with checkpoint validation.

crates/test-framework/src/spicetest/datasets/worker.rs

@@ -97,6 +97,9 @@ struct CheckpointValidationState {
     /// Set to `true` once a complete iteration finishes with every query
     /// passing. Sticky — once set, stays `true` for the validation window.
     converged: bool,
+    /// The instant at which the first query passed in the current
+    /// (non-converged) iteration.  Reset at each iteration boundary.
+    first_pass_instant: Option<Instant>,
 }
 
 impl CheckpointValidationState {
@@ -111,6 +114,7 @@ impl CheckpointValidationState {
             command_rx: handles.command_rx,
             current_iteration_results: HashMap::new(),
             converged: false,
+            first_pass_instant: None,
         }
     }
 
@@ -134,6 +138,7 @@ impl CheckpointValidationState {
                 self.completed_iterations = 0;
                 self.current_iteration_results.clear();
                 self.converged = false;
+                self.first_pass_instant = None;
                 eprintln!(
                     "Checkpoint validation enabled for checkpoint {}",
                     checkpoint_idx
@@ -147,6 +152,7 @@ impl CheckpointValidationState {
                 self.outcomes.clear();
                 self.current_iteration_results.clear();
                 self.converged = false;
+                self.first_pass_instant = None;
                 eprintln!("Checkpoint validation disabled");
                 let _ = self.status_tx.send(ValidationStatus::Inactive);
                 true
@@ -182,6 +188,9 @@ impl CheckpointValidationState {
                 if !self.converged {
                     self.current_iteration_results
                         .insert(Arc::clone(query_name), true);
+                    if self.first_pass_instant.is_none() {
+                        self.first_pass_instant = Some(Instant::now());
+                    }
                 }
             }
             Ok(QueryValidationResult::Fail(reason)) => {
@@ -261,11 +270,17 @@ impl CheckpointValidationState {
             let all_passed = self.current_iteration_results.values().all(|&v| v);
             if all_passed {
                 self.converged = true;
+                // first_pass_instant is preserved — it holds the instant
+                // the first query passed (possibly in an earlier iteration)
+                // indicating when the data first became correct.
                 eprintln!(
                     "Checkpoint {} validation converged after {} iterations",
                     self.checkpoint_idx, self.completed_iterations
                 );
             }
+            // Do NOT reset first_pass_instant on failure — it marks
+            // when the data first became queryable, even if a mixed
+            // iteration didn't fully converge yet.
         }
         self.current_iteration_results.clear();
 
@@ -278,6 +293,7 @@ impl CheckpointValidationState {
             outcomes: self.outcomes.values().cloned().collect(),
             completed_iterations: self.completed_iterations,
             converged: self.converged,
+            first_pass_instant: self.first_pass_instant,
         };
         let _ = self.status_tx.send(status);
     }

src/commands/load/mod.rs

@@ -492,7 +492,7 @@ pub(crate) async fn run(
     // Always create validation channels so we can track query-set iteration
     // completions. When --validate-results is enabled with checkpoint data,
     // these channels are also used for checkpoint-based results validation.
-    let (validation_controller, validation_worker_handles) = create_validation_channels();
+    let (mut validation_controller, validation_worker_handles) = create_validation_channels();
     test_builder = test_builder.with_checkpoint_validation(validation_worker_handles);
 
     let has_checkpoint_validation =
@@ -581,6 +581,10 @@ pub(crate) async fn run(
                                     );
 
                                     let etl_pause_time = tokio::time::Instant::now();
+                                    // Capture std::time::Instant at the same point so
+                                    // we can compare against the worker's
+                                    // first_pass_instant (which uses std::time::Instant).
+                                    let etl_pause_time_std = std::time::Instant::now();
 
                                     // Tell worker 0 to start validating.
                                     let _ = validation_controller.command_tx.send(Some(
@@ -593,16 +597,28 @@ pub(crate) async fn run(
                                     // Poll the validation status until convergence
                                     // (a complete iteration where every query passes)
                                     // or until the timeout is reached.
-                                    const POLL_INTERVAL: Duration = Duration::from_secs(5);
                                     const MAX_WAIT: Duration = Duration::from_secs(600);
+                                    let deadline =
+                                        tokio::time::Instant::now() + MAX_WAIT;
                                     let mut timed_out = false;
-                                    let mut interrupted = false;
+                                    let interrupted = false;
                                     loop {
                                         let status =
                                             validation_controller.status_rx.borrow().clone();
                                         if status.converged() {
-                                            let latency_ms =
-                                                etl_pause_time.elapsed().as_secs_f64() * 1000.0;
+                                            // Use the instant the first query passed
+                                            // as the latency reference point.  This
+                                            // measures when the data was fully ingested
+                                            // (first correct answer), not when the full
+                                            // validation sweep finished.
+                                            let latency_ms = status
+                                                .first_pass_instant()
+                                                .map_or_else(
+                                                    || etl_pause_time_std.elapsed(),
+                                                    |fpi| fpi.duration_since(etl_pause_time_std),
+                                                )
+                                                .as_secs_f64()
+                                                * 1000.0;
                                             println!(
                                                 "Checkpoint {} converged in {:.1}s ({} iterations)",
                                                 checkpoint_idx,
@@ -617,12 +633,17 @@ pub(crate) async fn run(
                                             timed_out = true;
                                             break;
                                         }
-                                        tokio::select! {
-                                            _ = tokio::time::sleep(POLL_INTERVAL) => {}
-                                            _ = signal::ctrl_c() => {
-                                                interrupted = true;
-                                                break;
-                                            }
+                                        // Wait for the worker to publish a new status
+                                        // update rather than polling on a fixed interval.
+                                        if tokio::time::timeout_at(
+                                            deadline,
+                                            validation_controller.status_rx.changed(),
+                                        )
+                                        .await
+                                        .is_err()
+                                        {
+                                            timed_out = true;
+                                            break;
                                         }
                                     }
 
@@ -633,6 +654,7 @@ pub(crate) async fn run(
                                         outcomes,
                                         completed_iterations: iters,
                                         converged,
+                                        ..
                                     } = &status
                                     {
                                         let total_pass: usize =

src/main.rs

@@ -231,14 +231,16 @@ async fn run_benchmark(
                 target_db_schema,
             )?);
 
-            let pipeline = ETLPipeline::new(
+            let mut pipeline = ETLPipeline::new(
                 dataset_source,
                 &generation_config,
                 Arc::clone(&data_source),
                 target_sink,
                 &mutations,
             )?;
 
+            pipeline.initialize().await?;
+
             (setup_response, pipeline)
         }
     };