Function test_concurrent_writes

src/environmentd/tests/sql.rs:2083–2137

()

#[mz_ore::test]
#[allow(clippy::disallowed_methods)]
fn test_concurrent_writes() {
    let server = test_util::TestHarness::default().start_blocking();

    let num_tables = 10;

    {
        let mut client = server.connect(postgres::NoTls).unwrap();
        for i in 0..num_tables {
            client
                .batch_execute(&format!("CREATE TABLE t_{i} (a INT, b text, c text)"))
                .unwrap();
        }
    }

    let num_threads = 3;
    let num_loops = 10;

    let mut clients = Vec::new();
    for _ in 0..num_threads {
        clients.push(server.connect(postgres::NoTls).unwrap());
    }

    let handles: Vec<_> = clients
        .into_iter()
        .map(|mut client| {
            std::thread::spawn(move || {
                for j in 0..num_loops {
                    for i in 0..num_tables {
                        let string_a = "A";
                        let string_b = "B";
                        client
                            .batch_execute(&format!(
                                "INSERT INTO t_{i} VALUES ({j}, '{string_a}', '{string_b}')"
                            ))
                            .unwrap();
                    }
                }
            })
        })
        .collect();

    for handle in handles {
        handle.join().unwrap();
    }

    let mut client = server.connect(postgres::NoTls).unwrap();

    for i in 0..num_tables {
        let count = client
            .query_one(&format!("SELECT count(*) FROM t_{i}"), &[])
            .unwrap()
            .get::<_, i64>(0);
        assert_eq!(num_loops * num_threads, count);
    }
}

#[mz_ore::test]
#[allow(clippy::disallowed_methods)]