(isolation_level: &str)
| 1828 | |
| 1829 | #[allow(clippy::disallowed_methods)] |
| 1830 | async fn test_session_linearizability(isolation_level: &str) { |
| 1831 | // Set the timestamp to zero for deterministic initial timestamps. |
| 1832 | let now = Arc::new(Mutex::new(0)); |
| 1833 | let now_fn = { |
| 1834 | let now = Arc::clone(&now); |
| 1835 | NowFn::from(move || *now.lock().unwrap()) |
| 1836 | }; |
| 1837 | let server = test_util::TestHarness::default() |
| 1838 | .with_now(now_fn) |
| 1839 | .unsafe_mode() |
| 1840 | .start() |
| 1841 | .await; |
| 1842 | server |
| 1843 | .enable_feature_flags(&["enable_session_timelines"]) |
| 1844 | .await; |
| 1845 | let mz_client = server.connect().await.unwrap(); |
| 1846 | |
| 1847 | let pg_table_name = "v_lin"; |
| 1848 | let pg_source_name = "source_lin"; |
| 1849 | let (pg_client, cleanup_fn) = test_util::create_postgres_source_with_table( |
| 1850 | &server, |
| 1851 | &mz_client, |
| 1852 | pg_table_name, |
| 1853 | "(a INT)", |
| 1854 | pg_source_name, |
| 1855 | ) |
| 1856 | .await; |
| 1857 | // Insert value into postgres table. |
| 1858 | let _ = pg_client |
| 1859 | .execute(&format!("INSERT INTO {pg_table_name} VALUES (42);"), &[]) |
| 1860 | .await |
| 1861 | .unwrap(); |
| 1862 | |
| 1863 | test_util::wait_for_pg_table_population(&mz_client, pg_table_name, 1).await; |
| 1864 | |
| 1865 | // The user table's write frontier will be close to zero because we use a |
| 1866 | // deterministic now function in this test. It may be slightly higher than |
| 1867 | // zero because bootstrapping and background tasks push the global timestamp |
| 1868 | // forward. |
| 1869 | // |
| 1870 | // The source's write frontier will be close to the system time because it |
| 1871 | // uses the system clock to close timestamps. |
| 1872 | // |
| 1873 | // Therefore queries that only involve the Postgres table (from the source) |
| 1874 | // will normally happen at a higher timestamp than queries that involve the |
| 1875 | // Materialze user table. However, we prevent this when in strict |
| 1876 | // serializable mode. |
| 1877 | |
| 1878 | mz_client |
| 1879 | .batch_execute("SET transaction_isolation = serializable") |
| 1880 | .await |
| 1881 | .unwrap(); |
| 1882 | |
| 1883 | // Wait until the Coordinator has learned about at least one upper |
| 1884 | // advancement in the source, which results in picking a higher timestamp |
| 1885 | // for queries. |
| 1886 | let mut source_ts; |
| 1887 | loop { |
no test coverage detected