Takes a snapshot of the upstream table that the specified `table` represents.
(
&'b mut self,
table: &SqlServerTableRaw,
worker_id: usize,
source_id: GlobalId,
)
| 152 | |
| 153 | /// Takes a snapshot of the upstream table that the specified `table` represents. |
| 154 | pub async fn snapshot<'b>( |
| 155 | &'b mut self, |
| 156 | table: &SqlServerTableRaw, |
| 157 | worker_id: usize, |
| 158 | source_id: GlobalId, |
| 159 | ) -> Result< |
| 160 | ( |
| 161 | Lsn, |
| 162 | impl Stream<Item = Result<tiberius::Row, SqlServerError>>, |
| 163 | ), |
| 164 | SqlServerError, |
| 165 | > { |
| 166 | static SAVEPOINT_NAME: &str = "_mz_snap_"; |
| 167 | |
| 168 | // The client that will be used for fencing does not need any special isolation level |
| 169 | // as it will be just be locking the table(s). |
| 170 | let mut fencing_client = self.client.new_connection().await?; |
| 171 | let mut fence_txn = fencing_client.transaction().await?; |
| 172 | let qualified_table_name = format!( |
| 173 | "{schema_name}.{table_name}", |
| 174 | schema_name = &table.schema_name, |
| 175 | table_name = &table.name |
| 176 | ); |
| 177 | self.metrics |
| 178 | .snapshot_table_lock_start(&qualified_table_name); |
| 179 | let result: Result<_, SqlServerError> = async { |
| 180 | fence_txn |
| 181 | .lock_table_shared(&table.schema_name, &table.name) |
| 182 | .await?; |
| 183 | tracing::info!(%source_id, %table.schema_name, %table.name, "timely-{worker_id} locked table"); |
| 184 | |
| 185 | self.client |
| 186 | .set_transaction_isolation(TransactionIsolationLevel::Snapshot) |
| 187 | .await?; |
| 188 | let mut txn = self.client.transaction().await?; |
| 189 | // Creating a savepoint forces a write to the transaction log, which will |
| 190 | // assign an LSN, but it does not force a transaction sequence number to be |
| 191 | // assigned as far as I can tell. I have not observed any entries added to |
| 192 | // `sys.dm_tran_active_snapshot_database_transactions` when creating a savepoint |
| 193 | // or when reading system views to retrieve the LSN. |
| 194 | // |
| 195 | // We choose cdc.change_tables because it is a system table that will exist |
| 196 | // when CDC is enabled, it has a well known schema, and as a CDC client, |
| 197 | // we should be able to read from it already. |
| 198 | let res = txn |
| 199 | .simple_query("SELECT TOP 1 object_id FROM cdc.change_tables") |
| 200 | .await?; |
| 201 | if res.len() != 1 { |
| 202 | Err(SqlServerError::InvariantViolated( |
| 203 | "No objects found in cdc.change_tables".into(), |
| 204 | ))? |
| 205 | } |
| 206 | |
| 207 | // Because the table is locked, any write operation has either |
| 208 | // completed, or is blocked. The LSN and XSN acquired now will represent a |
| 209 | // consistent point-in-time view, such that any committed write will be |
| 210 | // visible to this snapshot and the LSN of such a write will be less than |
| 211 | // or equal to the LSN captured here. Creating the savepoint sets the LSN, |
no test coverage detected