Function test_subscribe_basic

src/environmentd/tests/sql.rs:514–677

()

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

    server.enable_feature_flags(&["enable_index_options", "enable_logical_compaction_window"]);

    // Create a table with disabled compaction.
    client_writes
        .batch_execute("CREATE TABLE t (data text) WITH (RETAIN HISTORY FOR '1000 hours')")
        .unwrap();
    client_writes.batch_execute("SELECT * FROM t").unwrap();
    client_reads
        .batch_execute(
            "BEGIN;
         DECLARE c CURSOR FOR SUBSCRIBE t;",
        )
        .unwrap();
    // Locks the timestamp of the SUBSCRIBE to before any of the following INSERTs, which is required
    // for mz_timestamp column to be accurate
    let _ = client_reads.query_one("FETCH 0 c", &[]);

    let mut events = vec![];

    for i in 1..=3 {
        let data = format!("line {}", i);
        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>("data"), data);
        events.push((row.get::<_, MzTimestamp>("mz_timestamp").0, data));

        if i > 1 {
            // write timestamps should all increase
            assert!(events[i - 1].0 > events[i - 2].0);
        }
    }

    // Now subscribe without a snapshot as of each timestamp, verifying that when we do
    // so we only see events that occur as of or later than that timestamp.
    for (ts, _) in &events {
        client_reads
            .batch_execute(&*format!(
                "COMMIT; BEGIN;
            DECLARE c CURSOR FOR SUBSCRIBE t WITH (SNAPSHOT = false) AS OF {}",
                ts - 1
            ))
            .unwrap();

        // Skip by the things we won't be able to see.
        for (_, expected) in events.iter().skip_while(|(inner_ts, _)| inner_ts < ts) {
            let actual = client_reads.query_one("FETCH c", &[]).unwrap();
            assert_eq!(actual.get::<_, String>("data"), *expected);
        }