Function write_rows_spread

src/clusterd-test-driver/src/data.rs:185–219

Writes `rows` spread across timestamps `0..n_ts` (row `i` at time `i % n_ts`) in a single append that seals `[0, n_ts)`. All rows are inserted with diff `+1`. This is one `compare_and_append` regardless of `n_ts` — persist accepts updates at any timestamp within the sealed range — so it stays fast even for very large `n_ts` (a per-timestamp append would be `n_ts` consensus round-trips).

(
    client: &PersistClient,
    shard: ShardId,
    desc: &RelationDesc,
    rows: &[Row],
    n_ts: u64,
)

/// very large `n_ts` (a per-timestamp append would be `n_ts` consensus
/// round-trips).
pub async fn write_rows_spread(
    client: &PersistClient,
    shard: ShardId,
    desc: &RelationDesc,
    rows: &[Row],
    n_ts: u64,
) -> anyhow::Result<()> {
    assert!(n_ts > 0, "n_ts must be positive");
    let mut writer = client
        .open_writer::<SourceData, (), Timestamp, StorageDiff>(
            shard,
            Arc::new(desc.clone()),
            Arc::new(UnitSchema),
            Diagnostics {
                shard_name: "driver-data".to_string(),
                handle_purpose: "headless driver spread write".to_string(),
            },
        )
        .await?;
    let updates: Vec<_> = rows
        .iter()
        .enumerate()
        .map(|(i, r)| {
            let t = u64::cast_from(i) % n_ts;
            ((SourceData(Ok(r.clone())), ()), Timestamp::from(t), 1i64)
        })
        .collect();
    let lower = Antichain::from_elem(Timestamp::from(0));
    let upper = Antichain::from_elem(Timestamp::from(n_ts));
    writer
        .compare_and_append(&updates, lower, upper)
        .await?
        .map_err(|e| anyhow::anyhow!("{e}"))?;
    Ok(())
}

/// Number of rows needed to roughly hit `target_bytes` given `pad`-wide
/// payloads. Overhead per row is approximate; coarse sizing, not exact.