Get all available source references from the upstream system and return a `RetrievedSourceReferences` object that can be used to resolve user-specified source references and create `SourceReferences` for storage in the catalog.
(
mut self,
)
| 209 | /// to resolve user-specified source references and create `SourceReferences` |
| 210 | /// for storage in the catalog. |
| 211 | pub(super) async fn get_source_references( |
| 212 | mut self, |
| 213 | ) -> Result<RetrievedSourceReferences, PlanError> { |
| 214 | let references = match self { |
| 215 | SourceReferenceClient::Postgres { |
| 216 | client, |
| 217 | publication, |
| 218 | database, |
| 219 | } => { |
| 220 | let tables = mz_postgres_util::publication_info(client, publication, None).await?; |
| 221 | |
| 222 | if tables.is_empty() { |
| 223 | Err(PgSourcePurificationError::EmptyPublication( |
| 224 | publication.to_string(), |
| 225 | ))?; |
| 226 | } |
| 227 | |
| 228 | tables |
| 229 | .into_iter() |
| 230 | .map(|(_oid, desc)| ReferenceMetadata::Postgres { |
| 231 | table: desc, |
| 232 | database: database.to_string(), |
| 233 | }) |
| 234 | .collect() |
| 235 | } |
| 236 | SourceReferenceClient::MySql { |
| 237 | ref mut conn, |
| 238 | include_system_schemas, |
| 239 | } => { |
| 240 | let request = if include_system_schemas { |
| 241 | mz_mysql_util::SchemaRequest::AllWithSystemSchemas |
| 242 | } else { |
| 243 | mz_mysql_util::SchemaRequest::All |
| 244 | }; |
| 245 | // NOTE: mysql will only expose the schemas of tables we have at least one privilege on |
| 246 | // and we can't tell if a table exists without a privilege |
| 247 | let tables = mz_mysql_util::schema_info((*conn).deref_mut(), &request).await?; |
| 248 | |
| 249 | tables.into_iter().map(ReferenceMetadata::MySql).collect() |
| 250 | } |
| 251 | SourceReferenceClient::SqlServer { |
| 252 | ref mut client, |
| 253 | ref database, |
| 254 | } => { |
| 255 | let tables = mz_sql_server_util::inspect::get_tables(client).await?; |
| 256 | |
| 257 | let mut unique_tables: BTreeMap<(Arc<str>, Arc<str>), SqlServerTableRaw> = |
| 258 | BTreeMap::default(); |
| 259 | for table in tables { |
| 260 | let key = (Arc::clone(&table.schema_name), Arc::clone(&table.name)); |
| 261 | |
| 262 | unique_tables |
| 263 | .entry(key) |
| 264 | .and_modify(|chosen_table: &mut SqlServerTableRaw| { |
| 265 | // When multiple capture instances exist for the same table, |
| 266 | // we select deterministically based on: |
| 267 | // 1. Most recent create_date (newer capture instance) |
| 268 | // 2. If dates are equal, lexicographically greatest capture_instance name |
no test coverage detected