Function test_subscribe_negative_diffs

src/environmentd/tests/sql.rs:396–444

()

#[mz_ore::test]
#[allow(clippy::disallowed_methods)]
fn test_subscribe_negative_diffs() {
    let server = test_util::TestHarness::default().start_blocking();
    let mut client_writes = server.connect(postgres::NoTls).unwrap();
    let mut client_reads = server.connect(postgres::NoTls).unwrap();

    client_writes
        .batch_execute("CREATE TABLE t (data text)")
        .unwrap();
    client_writes.batch_execute(
        "CREATE MATERIALIZED VIEW counts AS SELECT data AS key, COUNT(data) AS count FROM t GROUP BY data",
    ).unwrap();
    client_reads
        .batch_execute(
            "BEGIN;
         DECLARE c CURSOR FOR SUBSCRIBE counts;",
        )
        .unwrap();

    let data = format!("line {}", 42);
    client_writes
        .execute("INSERT INTO t VALUES ($1)", &[&data])
        .unwrap();
    let row = client_reads.query_one("FETCH ALL c", &[]).unwrap();

    assert_eq!(row.get::<_, i64>("mz_diff"), 1);
    assert_eq!(row.get::<_, String>("key"), data);
    assert_eq!(row.get::<_, i64>("count"), 1);

    // send another row with the same key, this will retract the previous
    // count and emit an updated count

    let data = format!("line {}", 42);
    client_writes
        .execute("INSERT INTO t VALUES ($1)", &[&data])
        .unwrap();

    let rows = client_reads.query("FETCH ALL c", &[]).unwrap();
    let mut rows = rows.iter();

    let row = rows.next().expect("missing result");
    assert_eq!(row.get::<_, i64>("mz_diff"), -1);
    assert_eq!(row.get::<_, String>("key"), data);
    assert_eq!(row.get::<_, i64>("count"), 1);

    let row = rows.next().expect("missing result");
    assert_eq!(row.get::<_, i64>("mz_diff"), 1);
    assert_eq!(row.get::<_, String>("key"), data);
    assert_eq!(row.get::<_, i64>("count"), 2);
}

#[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 1))]
#[allow(clippy::disallowed_methods)]