Function write_rows_single_ts

src/clusterd-test-driver/src/data.rs:147–177

Writes `rows` to `shard` at `ts`, advancing `upper` to `ts+1`. All rows are inserted with diff `+1`. Returns once the append succeeds.

(
    client: &PersistClient,
    shard: ShardId,
    desc: &RelationDesc,
    rows: &[Row],
    ts: Timestamp,
)

/// Writes `rows` to `shard` at `ts`, advancing `upper` to `ts+1`. All rows are
/// inserted with diff `+1`. Returns once the append succeeds.
pub async fn write_rows_single_ts(
    client: &PersistClient,
    shard: ShardId,
    desc: &RelationDesc,
    rows: &[Row],
    ts: Timestamp,
) -> anyhow::Result<()> {
    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 write".to_string(),
            },
        )
        .await?;

    let updates: Vec<_> = rows
        .iter()
        .map(|r| ((SourceData(Ok(r.clone())), ()), ts, 1i64))
        .collect();
    let lower = Antichain::from_elem(ts);
    let upper = Antichain::from_elem(ts.step_forward());
    writer
        .compare_and_append(&updates, lower, upper)
        .await?
        .map_err(|e| anyhow::anyhow!("{e}"))?;
    Ok(())
}

/// 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