fix: Add some ETL debug logs for sink progress (#150)

* fix: Switch run spicebench to spiceai dev runners

* fix

* fix

* chore: auto-fix cargo fmt + clippy

* Update run_spicebench.yml

---------

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

Author: peasee

.github/workflows/run_spicebench.yml

@@ -55,6 +55,11 @@ on:
         required: false
         default: false
         type: boolean
+      enable_module_debug_logging:
+        description: 'Enable debug logs only for SpiceBench modules (etl, spicebench, data_generation)'
+        required: false
+        default: false
+        type: boolean
 
 jobs:
   run-spicebench:
@@ -243,6 +248,7 @@ jobs:
           ETL_REGION: ${{ github.event.inputs.etl_region || 'us-east-1' }}
           ETL_SINK: ${{ github.event.inputs.etl_sink || 'hive' }}
           VALIDATE_CHECKPOINT_RESULTS: ${{ github.event.inputs.validate_checkpoint_results || 'false' }}
+          ENABLE_MODULE_DEBUG_LOGGING: ${{ github.event.inputs.enable_module_debug_logging || 'false' }}
           SPICEAI_BENCHMARK_METRICS_KEY: ${{ secrets.SPICEAI_BENCHMARK_METRICS_KEY }}
           AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
@@ -253,9 +259,14 @@ jobs:
           LAKEBASE_DATABASE_INSTANCE: ${{ secrets.LAKEBASE_DATABASE_INSTANCE }}
           SPIDAPTER_ICEBERG_REGION: us-west-1
           SPIDAPTER_ICEBERG_CATALOG_FROM: iceberg:https://glue.us-west-1.amazonaws.com/iceberg/v1/catalogs/211125479522/namespaces
-          RUST_LOG: 'info'
         run: |
           set -euo pipefail
+          if [ "${ENABLE_MODULE_DEBUG_LOGGING}" = "true" ]; then
+            export RUST_LOG='info,etl=debug,spicebench=debug,data_generation=debug'
+          else
+            export RUST_LOG='info'
+          fi
+
           TABLE_FORMAT="parquet"
           EXECUTOR_INSTANCE_TYPE="github-hosted-ubuntu-latest"
           ETL_ENDPOINT=""

crates/etl/src/lib.rs

@@ -15,7 +15,7 @@ limitations under the License.
 */
 
 use std::sync::Arc;
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use arrow::array::{Array, RecordBatch, StringArray, TimestampMicrosecondArray};
 use arrow::compute;
@@ -34,7 +34,7 @@ use system_adapter_protocol::DatasetConfig as ProtocolDatasetConfig;
 use tokio::sync::watch;
 use tokio::task::{JoinHandle, JoinSet};
 use tokio_util::sync::CancellationToken;
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, trace, warn};
 
 use crate::sink::{InsertOp, Sink};
 
@@ -55,7 +55,7 @@ const TARGET_BATCH_ROWS: usize = 8_192 * 4;
 
 /// Maximum number of in-flight sink writes allowed per table task when the
 /// current segment set is insert-only.
-const MAX_IN_FLIGHT_TABLE_WRITES: usize = 2;
+const MAX_IN_FLIGHT_TABLE_WRITES: usize = 1;
 
 /// Maximum number of concurrent source logical-batch reads per ETL table task.
 const MAX_IN_FLIGHT_SOURCE_BATCH_READS: usize = 2;
@@ -113,6 +113,33 @@ fn build_partition_columns(dataset_columns: Vec<String>) -> Vec<String> {
     columns
 }
 
+fn record_timing_sample_ms(samples: &StdArc<StdMutex<Vec<u64>>>, elapsed: Duration) {
+    let elapsed_ms = elapsed.as_millis().min(u128::from(u64::MAX)) as u64;
+    let mut values = samples.lock().expect("timing samples lock poisoned");
+    values.push(elapsed_ms);
+}
+
+fn take_median_sample_ms(samples: &StdArc<StdMutex<Vec<u64>>>) -> Option<(f64, usize)> {
+    let mut values = {
+        let mut guard = samples.lock().expect("timing samples lock poisoned");
+        if guard.is_empty() {
+            return None;
+        }
+        std::mem::take(&mut *guard)
+    };
+
+    values.sort_unstable();
+    let count = values.len();
+    let mid = count / 2;
+    let median_ms = if count % 2 == 1 {
+        values[mid] as f64
+    } else {
+        (values[mid - 1] as f64 + values[mid] as f64) / 2.0
+    };
+
+    Some((median_ms, count))
+}
+
 /// Concatenates small input batches and splits large input batches so each
 /// resulting batch has at most [`TARGET_BATCH_ROWS`] rows.
 ///
@@ -1367,11 +1394,13 @@ async fn run_pipeline(
         "ETL pipeline run started"
     );
 
-    // Shared progress counters for periodic logging.
+    // Shared progress counters and timing samples for periodic logging.
     let logical_steps_consumed = StdArc::new(AtomicU64::new(0));
     let batches_processed = StdArc::new(AtomicU64::new(0));
     let rows_processed = StdArc::new(AtomicU64::new(0));
     let tables_finished_counter = StdArc::new(AtomicU64::new(0));
+    let batch_retrieval_samples_ms = StdArc::new(StdMutex::new(Vec::<u64>::new()));
+    let sink_write_samples_ms = StdArc::new(StdMutex::new(Vec::<u64>::new()));
     let pipeline_start = Instant::now();
 
     // Spawn periodic progress logger (every 5 seconds).
@@ -1380,6 +1409,8 @@ async fn run_pipeline(
         let batches_processed = StdArc::clone(&batches_processed);
         let rows_processed = StdArc::clone(&rows_processed);
         let tables_finished_counter = StdArc::clone(&tables_finished_counter);
+        let batch_retrieval_samples_ms = StdArc::clone(&batch_retrieval_samples_ms);
+        let sink_write_samples_ms = StdArc::clone(&sink_write_samples_ms);
         let cancel = cancel.clone();
         tokio::spawn(async move {
             let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
@@ -1405,6 +1436,23 @@ async fn run_pipeline(
                             rows_per_sec = format!("{:.1}", rows_done as f64 / secs),
                             "ETL progress"
                         );
+
+                        let retrieval_summary = take_median_sample_ms(&batch_retrieval_samples_ms);
+                        let sink_write_summary = take_median_sample_ms(&sink_write_samples_ms);
+                        if retrieval_summary.is_some() || sink_write_summary.is_some() {
+                            let (retrieval_median_ms, retrieval_samples) =
+                                retrieval_summary.unwrap_or((0.0, 0));
+                            let (sink_write_median_ms, sink_write_samples) =
+                                sink_write_summary.unwrap_or((0.0, 0));
+
+                            debug!(
+                                retrieval_samples,
+                                retrieval_median_ms = format!("{retrieval_median_ms:.1}"),
+                                sink_write_samples,
+                                sink_write_median_ms = format!("{sink_write_median_ms:.1}"),
+                                "ETL batch timing medians (last 5s)"
+                            );
+                        }
                     }
                     () = cancel.cancelled() => break,
                 }
@@ -1490,12 +1538,16 @@ async fn run_pipeline(
             let work_state = Arc::clone(&work_state);
             let logical_steps_consumed = StdArc::clone(&logical_steps_consumed);
             let last_created_at = Arc::clone(&last_created_at_us);
+            let batch_retrieval_samples_ms = StdArc::clone(&batch_retrieval_samples_ms);
+            let sink_write_samples_ms = StdArc::clone(&sink_write_samples_ms);
             let partition_columns = table_partition_columns
                 .get(&table_name)
                 .cloned()
                 .unwrap_or_else(|| vec![CREATED_AT_COLUMN.to_string()]);
 
             join_set.spawn(async move {
+                let retrieval_started_at = Instant::now();
+
                 // 1. Read from source; keep reading subsequent table batches
                 // until we accumulate enough rows for efficient downstream work.
                 let (source_batches, key_columns, table_finished, consumed_work_units, rows_read) =
@@ -1545,6 +1597,12 @@ async fn run_pipeline(
                         ));
                     }
                 };
+
+                record_timing_sample_ms(
+                    &batch_retrieval_samples_ms,
+                    retrieval_started_at.elapsed(),
+                );
+
                 for batch in &coalesced {
                     let segments = match split_batch_by_op(batch, &key_columns) {
                         Ok(s) => s,
@@ -1561,6 +1619,8 @@ async fn run_pipeline(
                         }
                     };
 
+                    let write_started_at = Instant::now();
+
                     if let Err(err_msg) = write_segments_for_batch(
                         Arc::clone(&data_sink),
                         &table_name,
@@ -1579,13 +1639,16 @@ async fn run_pipeline(
                         );
                         return Err(err_msg);
                     }
+
+                    record_timing_sample_ms(&sink_write_samples_ms, write_started_at.elapsed());
+
                     let tracker = last_created_at
                         .get(&table_name)
                         .expect("table missing from last_created_at map");
                     tracker.store(batch_ts, Ordering::Relaxed);
                 }
 
-                debug!(
+                trace!(
                     table = %table_name,
                     batch_id,
                     "Table batch processed"
@@ -1626,7 +1689,7 @@ async fn run_pipeline(
         batches_processed.fetch_add(step_batch_count, Ordering::Relaxed);
         rows_processed.fetch_add(step_rows_count, Ordering::Relaxed);
 
-        debug!(
+        trace!(
             batch_id,
             outer_steps_processed,
             steps_processed = logical_steps_consumed.load(Ordering::Relaxed),

crates/etl/src/sink/adbc.rs

@@ -267,13 +267,41 @@ impl AdbcSink {
             .as_deref()
             .and_then(|schema| (!schema.is_empty()).then_some(schema));
 
-        match conn.bulk_ingest(
-            table_name,
-            target_db_catalog,
-            target_db_schema,
-            IngestMode::CreateAppend,
-            batch.clone(),
-        ) {
+        let ingest_result = if target_db_catalog.is_some() || target_db_schema.is_some() {
+            match conn.bulk_ingest(
+                &ingest_table_name,
+                None,
+                None,
+                IngestMode::CreateAppend,
+                batch.clone(),
+            ) {
+                Ok(result) => Ok(result),
+                Err(qualified_err) => {
+                    let qualified_message = qualified_err.to_string();
+                    if Self::is_message_too_large_error(&qualified_message) {
+                        Err(qualified_err)
+                    } else {
+                        conn.bulk_ingest(
+                            table_name,
+                            target_db_catalog,
+                            target_db_schema,
+                            IngestMode::CreateAppend,
+                            batch.clone(),
+                        )
+                    }
+                }
+            }
+        } else {
+            conn.bulk_ingest(
+                table_name,
+                target_db_catalog,
+                target_db_schema,
+                IngestMode::CreateAppend,
+                batch.clone(),
+            )
+        };
+
+        match ingest_result {
             Ok(_) => Ok(()),
             Err(e) => {
                 let message = e.to_string();