fix: Flush ADBC sink before resuming ETL after checkpoint pause

When the ETL pipeline pauses at a checkpoint boundary for validation,
the reusable ADBC bulk ingest streams (long-lived DoPut connections)
sit idle for the duration of the validation window (often 10+ minutes).
These connections can go stale or be closed by the server during this
period.

When the pipeline resumes via continue_pipeline(), it reuses the same
AdbcSink with its stale streams. The next write attempt hangs
indefinitely on the dead connection, causing the ETL to stall at
step 0/9 with zero progress.

Fix: call data_sink.flush() in continue_pipeline() before spawning the
new run task. This closes all reusable bulk ingest streams so they are
recreated fresh when the first write of the new phase arrives.

This requires making continue_pipeline() async since flush() is async.

Author: phillipleblanc

.github/workflows/run_spicebench_debug_spice_cloud.yml

@@ -177,6 +177,7 @@ jobs:
           EXECUTOR_REPLICAS: ${{ github.event.inputs.executor_replicas || '4' }}
           SPIDAPTER_APP_MEMORY_LIMIT: ${{ github.event.inputs.app_memory_limit || '16Gi' }}
           SPIDAPTER_EXECUTOR_MEMORY_LIMIT: ${{ github.event.inputs.executor_memory_limit || '16Gi' }}
+          SPIDAPTER_ORGANIZATION_TAG: 'spicehq'
           CUSTOM_IMAGE: ${{ github.event.inputs.custom_image || '' }}
         run: |
           set -euo pipefail
@@ -225,6 +226,7 @@ jobs:
 
           ADAPTER_DOCKER_OPTS="${ADAPTER_DOCKER_OPTS} -e SPIDAPTER_APP_MEMORY_LIMIT=${SPIDAPTER_APP_MEMORY_LIMIT}"
           ADAPTER_DOCKER_OPTS="${ADAPTER_DOCKER_OPTS} -e SPIDAPTER_EXECUTOR_MEMORY_LIMIT=${SPIDAPTER_EXECUTOR_MEMORY_LIMIT}"
+          ADAPTER_DOCKER_OPTS="${ADAPTER_DOCKER_OPTS} -e SPIDAPTER_ORGANIZATION_TAG"
 
           # Parse custom image into registry, name, and tag for spidapter.
           # Expected format: registry/image:tag (e.g. ghcr.io/spiceai/spiceai-dev:spicebench-sf10)

crates/etl/src/lib.rs

@@ -1457,7 +1457,7 @@ impl ETLPipeline {
     /// [`PipelineState::Paused`].
     ///
     /// Returns an error if the pipeline is not in the [`Paused`] state.
-    pub fn continue_pipeline(&mut self) -> anyhow::Result<()> {
+    pub async fn continue_pipeline(&mut self) -> anyhow::Result<()> {
         let current_state = self.state_rx.borrow().clone();
         if current_state != PipelineState::Paused {
             anyhow::bail!(
@@ -1476,6 +1476,12 @@ impl ETLPipeline {
             handle.abort();
         }
 
+        // Flush the sink to close any stale reusable connections (e.g. ADBC
+        // bulk ingest streams) that may have gone idle during the checkpoint
+        // validation window.  Without this, resumed writes can hang on dead
+        // connections that were held open across the pause.
+        self.data_sink.flush().await?;
+
         self.spawn_run_task(self.batch_budget);
         Ok(())
     }

src/commands/checkpoint.rs

@@ -282,7 +282,7 @@ async fn execute_duckdb(args: &CheckpointArgs) -> anyhow::Result<()> {
                 .await?;
                 checkpoint_idx += 1;
 
-                pipeline.continue_pipeline()?;
+                pipeline.continue_pipeline().await?;
             }
             PipelineState::Stopped(StopReason::Completed) => {
                 tracing::info!(

src/commands/load/mod.rs

@@ -656,9 +656,7 @@ async fn run_checkpoint_validation(
     // Phase 0: validate table row counts first as a fast correctness probe.
     // Row count queries are cheap and immediately surface data loss/duplication
     // without waiting for expensive analytical queries to converge.
-    println!(
-        "Checkpoint {checkpoint_idx}: validating table row counts before probing queries",
-    );
+    println!("Checkpoint {checkpoint_idx}: validating table row counts before probing queries",);
     {
         let mut row_count_ticker = tokio::time::interval(probe_period);
         let mut row_count_attempt = 0u64;
@@ -1007,7 +1005,7 @@ pub(crate) async fn run(
                             }
                         }
 
-                        if let Err(e) = etl_pipeline.continue_pipeline() {
+                        if let Err(e) = etl_pipeline.continue_pipeline().await {
                             eprintln!("Failed to continue ETL pipeline after pause: {e}");
                             shutdown_token.cancel();
                             break Some(RunOutcome::PipelineFailure(format!("Failed to continue ETL pipeline: {e}")));