(
scope: Scope<'scope, Lsn>,
config: RawSourceCreationConfig,
outputs: BTreeMap<GlobalId, SourceOutputInfo>,
source: SqlServerSourceConnection,
metrics: SqlServerSourceMetrics,
)
| 56 | static REPL_READER: &str = "reader"; |
| 57 | |
| 58 | pub(crate) fn render<'scope>( |
| 59 | scope: Scope<'scope, Lsn>, |
| 60 | config: RawSourceCreationConfig, |
| 61 | outputs: BTreeMap<GlobalId, SourceOutputInfo>, |
| 62 | source: SqlServerSourceConnection, |
| 63 | metrics: SqlServerSourceMetrics, |
| 64 | ) -> ( |
| 65 | StackedCollection<'scope, Lsn, (u64, Result<SourceMessage, DataflowError>)>, |
| 66 | StreamVec<'scope, Lsn, ReplicationError>, |
| 67 | PressOnDropButton, |
| 68 | ) { |
| 69 | let op_name = format!("SqlServerReplicationReader({})", config.id); |
| 70 | let mut builder = AsyncOperatorBuilder::new(op_name, scope); |
| 71 | |
| 72 | let (data_output, data_stream) = builder.new_output::<FueledBuilder<_>>(); |
| 73 | |
| 74 | // Captures DefiniteErrors that affect the entire source, including all outputs |
| 75 | let (definite_error_handle, definite_errors) = |
| 76 | builder.new_output::<CapacityContainerBuilder<_>>(); |
| 77 | |
| 78 | let (button, transient_errors) = builder.build_fallible(move |caps| { |
| 79 | let busy_signal = Arc::clone(&config.busy_signal); |
| 80 | Box::pin(SignaledFuture::new(busy_signal, async move { |
| 81 | let [ |
| 82 | data_cap_set, |
| 83 | definite_error_cap_set, |
| 84 | ]: &mut [_; 2] = caps.try_into().unwrap(); |
| 85 | |
| 86 | let connection_config = source |
| 87 | .connection |
| 88 | .resolve_config( |
| 89 | &config.config.connection_context.secrets_reader, |
| 90 | &config.config, |
| 91 | InTask::Yes, |
| 92 | ) |
| 93 | .await?; |
| 94 | let mut client = mz_sql_server_util::Client::connect(connection_config).await?; |
| 95 | |
| 96 | let worker_id = config.worker_id; |
| 97 | |
| 98 | // The decoder is specific to the export, and each export pulls data from a specific capture instance. |
| 99 | let mut decoder_map: BTreeMap<_, _> = BTreeMap::new(); |
| 100 | // Maps the 'capture instance' to the output index for only those outputs that this worker will snapshot |
| 101 | let mut capture_instance_to_snapshot: BTreeMap<Arc<str>, Vec<_>> = BTreeMap::new(); |
| 102 | // Maps the 'capture instance' to the output index for all outputs of this worker |
| 103 | let mut capture_instances: BTreeMap<Arc<str>, Vec<_>> = BTreeMap::new(); |
| 104 | // Export statistics for a given capture instance |
| 105 | let mut export_statistics: BTreeMap<_, Vec<_>> = BTreeMap::new(); |
| 106 | // Maps the included columns for each output index so we can check |
| 107 | // whether schema updates are valid on a per-output basis |
| 108 | let mut included_columns: HashMap<u64, Vec<Arc<str>>> = HashMap::new(); |
| 109 | |
| 110 | for (export_id, output) in outputs.iter() { |
| 111 | let key = output.partition_index; |
| 112 | if decoder_map.insert(key, Arc::clone(&output.decoder)).is_some() { |
| 113 | panic!("Multiple decoders for output index {}", output.partition_index); |
| 114 | } |
| 115 | // Collect the included columns from decoder for schema |
nothing calls this directly
no test coverage detected