Add source table support for SQL server #32987
Conversation
| ) | ||
| .await?; | ||
|
|
||
| // There should be exactly one source_export returned for this statement |
There was a problem hiding this comment.
You can use source_exports.into_element() to extract the first element and additionally assert that there isn't a second one
There was a problem hiding this comment.
Just saw the idiom you used was prior art in the other cases. We could fix those too since we're here
| external_reference: _, | ||
| } = sql_server::generate_source_export_statement_values(&scx, purified_export)?; | ||
|
|
||
| if let Some(text_cols_option) = with_options |
There was a problem hiding this comment.
nit: if you have if let Some(..) = opt_value and you also include an else statement then it's more idiomatic to write a match statement with Some(..) and None branches
There was a problem hiding this comment.
This was also carrying forward existing patterns. I updated the lot.
| let mut cdc_handle = client | ||
| .cdc(capture_instances.keys().cloned()) | ||
| .max_lsn_wait(SNAPSHOT_MAX_LSN_WAIT.get(config.config.config_set())); | ||
|
|
||
| // Snapshot any instances that require it. | ||
| // Snapshot any instances that requires it. | ||
| // The LSN returned here will be the max LSN for any capture instance on the SQL server. | ||
| let snapshot_lsn = { |
There was a problem hiding this comment.
If I'm reading the code correctly here we are computing a snapshot LSN that is common among all exports, and that snapshot LSN depends on the max LSN that we get from each capture instance, and capture instances are created on demand as the user requests more and more exports from the available ones.
If we put all these together I think we can end up in a situation where one replica that only knows about export foo computes some snapshot_lsn1 but another replica, that say hears about two exports, foo and bar computes a higher snapshot_lsn2 > snapshot_lsn1 due to the capture instance of bar.
Is this reasoning correct? If yes I think there is a possibility for double ingesting the snapshot
There was a problem hiding this comment.
Great catch! I'll have to capture the lsn earlier on. I think MySQL does this as well, should be able to follow the same pattern.
There was a problem hiding this comment.
Now captures LSN at purification and passes it down for resume_lsn computation
| tracing::warn!(%config.id, "timely-{} no resume_lsn, waiting.", config.worker_id); | ||
| std::future::pending::<()>().await; | ||
| } | ||
| let resume_lsn = resume_lsn.unwrap(); |
There was a problem hiding this comment.
nit: having this be a let resume_lsn = match resume_lsn {...} is more idiomatic than checking is_none() and then unwraping
There was a problem hiding this comment.
Since we should never continue beyond that point, I opted for let .. else { ...; unreachable!() }; - but am happy to change it to match with None => std::future::pending::<Lsn>().await if you prefer.
| @@ -221,22 +222,44 @@ pub(crate) fn render<G: Scope<Timestamp = Lsn>>( | |||
|
|
|||
| // Start replicating from the LSN __after__ we took the snapshot. | |||
| let replication_start_lsn = snapshot_lsn.increment(); | |||
| let mut rewinds:BTreeMap<_,_> = export_ids_to_snapshot | |||
There was a problem hiding this comment.
| let mut rewinds:BTreeMap<_,_> = export_ids_to_snapshot | |
| let mut rewinds: BTreeMap<_, _> = export_ids_to_snapshot |
rustfmt gives up in this file :(
| @@ -349,7 +396,7 @@ type StackedAsyncOutputHandle<T, D> = AsyncOutputHandle< | |||
| /// Helper method to return a "definite" error upstream. | |||
| async fn return_definite_error( | |||
| err: DefiniteError, | |||
| outputs: &[usize], | |||
| outputs: impl Iterator<Item = &u64>, | |||
There was a problem hiding this comment.
This function probably prefers just u64 since it unconditionally dereferences. You can add a .copied() in the .iter() calls in the flatmap above to get a u64 from a &u64 iterator
Co-authored-by: Petros Angelatos <[email protected]>
Thanks for the review @petrosagg! I've split review updates into separate commits. You can ignore the last couple commits, which were me rushing to fix typos and then trying to get CI to work. The most meaningful change is in the I ran into errors where I was setting things up too quickly and there was no max LSN yet, and tests consistently required sleeps to avoid this. So I opted to include a short retry window in purification to get the LSN for a better experience, and that added a little bit of refactoring to centralize retries. |
| // If CDC is enabled for a table, there is a period where the max LSN will not be | ||
| // available. Rather than return an error to the user, we retry for 5 seconds to | ||
| // allow the CDC job a chance to run. By default, the job runs every 5 seconds. | ||
| let initial_lsn = | ||
| mz_sql_server_util::inspect::get_max_lsn_retry(&mut client, Duration::from_secs(5)) | ||
| .await?; |
There was a problem hiding this comment.
nit: can we make the retry configurable? I guess it's pretty unlikely that someone will reconfigure the timing on the capture instance.
| // If CDC is enabled for a table, there is a period where the max LSN will not be | ||
| // available. Rather than return an error to the user, we retry for 5 seconds to | ||
| // allow the CDC job a chance to run. By default, the job runs every 5 seconds. | ||
| let initial_lsn = | ||
| mz_sql_server_util::inspect::get_max_lsn_retry(&mut client, Duration::from_secs(5)) | ||
| .await?; |
There was a problem hiding this comment.
nit: at a minimum would be nice to get the value we retry for into a shared constant so it's less of a magic number.
Adds support for
CREATE TABLE .. FROM SOURCEfor SQL Server.Motivation
implements https://github.com/MaterializeInc/database-issues/issues/9202
Checklist
$T ⇔ Proto$Tmapping (possibly in a backwards-incompatible way), then it is tagged with aT-protolabel.