fix(asap-tools): send ClickHouse ticks at once in the paced feeder

asap-tools/data-sources/paced-feeder/README.md

@@ -53,10 +53,12 @@ credentials.
 
 ## Pacing and stats
 
-Each tick sends its `--samples-per-sec` sub-second slots in turn. A slot's
-rows are split into `--batch-rows` requests and sent with up to
-`--concurrency` in flight, and the next slot starts only after every request
-of the current one returns, so a series' samples never arrive out of order.
+Each tick's rows are split into `--batch-rows` requests and sent with up to
+`--concurrency` in flight. For remote write, a tick sends its
+`--samples-per-sec` sub-second slots in turn, and the next slot starts only
+after every request of the current one returns, so each aggregation group's
+samples reach ASAPQuery in timestamp order. ClickHouse sorts on insert, so it
+gets the whole tick at once.
 `--samples-per-sec` is at most 1000, so timestamps (in ms) never repeat. The
 default batch size is 50,000 rows for remote write, which keeps a compressed
 request under axum's 2 MB body limit in ASAPQuery's ingest, and 100,000 for

asap-tools/data-sources/paced-feeder/src/main.rs

@@ -16,7 +16,7 @@ use serde::Serialize;
 use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tokio::sync::Semaphore;
-use workload::Workload;
+use workload::{Row, Workload};
 
 #[derive(Debug, Clone, Copy, ValueEnum, Serialize)]
 #[serde(rename_all = "kebab-case")]
@@ -263,18 +263,25 @@ async fn main() -> Result<()> {
         let started = Instant::now();
 
         let (mut rows, mut bytes) = (0, 0);
-        // A slot's requests all return before the next slot's are sent, so a
-        // series' samples never arrive out of timestamp order.
-        for slot in 0..args.samples_per_sec {
-            let slot_rows = Arc::new(workload.slot_rows(tick, slot, start_ms));
+        // ASAPQuery can drop samples behind a group's watermark, so remote write
+        // sends one slot at a time: a slot's requests all return before the
+        // next slot's are sent. ClickHouse sorts on insert, so it sends the
+        // whole tick at once.
+        let slots = (0..args.samples_per_sec).map(|slot| workload.slot_rows(tick, slot, start_ms));
+        let waves: Vec<Vec<Row>> = match args.sink {
+            SinkKind::RemoteWrite => slots.collect(),
+            SinkKind::Clickhouse => vec![slots.flatten().collect()],
+        };
+        for wave in waves {
+            let wave_rows = Arc::new(wave);
             let mut tasks = tokio::task::JoinSet::new();
-            for begin in (0..slot_rows.len()).step_by(batch_rows) {
-                let end = (begin + batch_rows).min(slot_rows.len());
-                let (slot_rows, sink) = (Arc::clone(&slot_rows), Arc::clone(&sink));
+            for begin in (0..wave_rows.len()).step_by(batch_rows) {
+                let end = (begin + batch_rows).min(wave_rows.len());
+                let (wave_rows, sink) = (Arc::clone(&wave_rows), Arc::clone(&sink));
                 let permit = Arc::clone(&permits).acquire_owned().await?;
                 tasks.spawn(async move {
                     let _permit = permit;
-                    let body = sink.encoder.encode(&slot_rows[begin..end]);
+                    let body = sink.encoder.encode(&wave_rows[begin..end]);
                     let bytes = body.len();
                     sink.send(body).await.map(|()| (end - begin, bytes))
                 });