Merge pull request #33729 from teskje/source-exports-running

Author: teskje

src/catalog/src/builtin.rs

@@ -4531,17 +4531,6 @@ pub static MZ_SOURCE_STATUSES: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinV
                     self_events.status <> 'ceased' AND
                     parent_events.status = 'stalled'
                 THEN parent_events.source_id
-                -- TODO: Remove this once subsources eagerly propogate their status
-                -- Subsources move from starting to running lazily once they see
-                -- a record flow through, even though they are online and healthy.
-                -- This has been repeatedly brought up as confusing by users.
-                -- So now, if the parent source is running, and the subsource is
-                -- starting, we override its status to running to relfect its healthy
-                -- status.
-                WHEN
-                    self_events.status = 'starting' AND
-                    parent_events.status = 'running'
-                THEN parent_events.source_id
                 ELSE self_events.source_id
             END AS id_to_use
         FROM

src/storage/src/source/generator.rs

@@ -243,7 +243,8 @@ fn render_simple_generator<G: Scope<Timestamp = MzOffset>>(
     let mut builder = AsyncOperatorBuilder::new(config.name.clone(), scope.clone());
 
     let (data_output, stream) = builder.new_output::<AccountedStackBuilder<_>>();
-    let partition_count = u64::cast_from(config.source_exports.len());
+    let export_ids: Vec<_> = config.source_exports.keys().copied().collect();
+    let partition_count = u64::cast_from(export_ids.len());
     let data_streams: Vec<_> = stream.partition::<CapacityContainerBuilder<_>, _, _>(
         partition_count,
         |((output, data), time, diff): &(
@@ -256,7 +257,7 @@ fn render_simple_generator<G: Scope<Timestamp = MzOffset>>(
         },
     );
     let mut data_collections = BTreeMap::new();
-    for (id, data_stream) in config.source_exports.keys().zip_eq(data_streams) {
+    for (id, data_stream) in export_ids.iter().zip_eq(data_streams) {
         data_collections.insert(*id, data_stream.as_collection());
     }
 
@@ -373,14 +374,17 @@ fn render_simple_generator<G: Scope<Timestamp = MzOffset>>(
                     Event::Progress(Some(offset)) => {
                         if resume_offset <= offset && health_cap.is_some() {
                             let health_cap = health_cap.take().expect("known to exist");
-                            health_output.give(
-                                &health_cap,
-                                HealthStatusMessage {
-                                    id: None,
-                                    namespace: StatusNamespace::Generator,
-                                    update: HealthStatusUpdate::running(),
-                                },
-                            );
+                            let export_ids = export_ids.iter().copied();
+                            for id in export_ids.map(Some).chain(None) {
+                                health_output.give(
+                                    &health_cap,
+                                    HealthStatusMessage {
+                                        id,
+                                        namespace: StatusNamespace::Generator,
+                                        update: HealthStatusUpdate::running(),
+                                    },
+                                );
+                            }
                         }
 
                         // If we've reached the requested maximum offset, cease.

src/storage/src/source/generator/key_value.rs

@@ -56,7 +56,8 @@ pub fn render<G: Scope<Timestamp = MzOffset>>(
     let mut builder = AsyncOperatorBuilder::new(config.name.clone(), scope.clone());
 
     let (data_output, stream) = builder.new_output::<AccountedStackBuilder<_>>();
-    let partition_count = u64::cast_from(config.source_exports.len());
+    let export_ids: Vec<_> = config.source_exports.keys().copied().collect();
+    let partition_count = u64::cast_from(export_ids.len());
     let data_streams: Vec<_> = stream.partition::<CapacityContainerBuilder<_>, _, _>(
         partition_count,
         |((output, data), time, diff): &(
@@ -69,7 +70,7 @@ pub fn render<G: Scope<Timestamp = MzOffset>>(
         },
     );
     let mut data_collections = BTreeMap::new();
-    for (id, data_stream) in config.source_exports.keys().zip_eq(data_streams) {
+    for (id, data_stream) in export_ids.iter().zip_eq(data_streams) {
         data_collections.insert(*id, data_stream.as_collection());
     }
 
@@ -252,12 +253,17 @@ pub fn render<G: Scope<Timestamp = MzOffset>>(
         })
     });
 
-    let status = [HealthStatusMessage {
-        id: None,
-        namespace: StatusNamespace::Generator,
-        update: HealthStatusUpdate::running(),
-    }]
-    .to_stream(scope);
+    let status = export_ids
+        .into_iter()
+        .map(Some)
+        .chain(None)
+        .map(|id| HealthStatusMessage {
+            id,
+            namespace: StatusNamespace::Generator,
+            update: HealthStatusUpdate::running(),
+        })
+        .collect::<Vec<_>>()
+        .to_stream(scope);
     (
         data_collections,
         progress_stream,

src/storage/src/source/mysql.rs

@@ -196,12 +196,17 @@ impl SourceRender for MySqlSourceConnection {
             data_collections.insert(*id, data_stream.as_collection());
         }
 
-        let health_init = std::iter::once(HealthStatusMessage {
-            id: None,
-            namespace: Self::STATUS_NAMESPACE,
-            update: HealthStatusUpdate::Running,
-        })
-        .to_stream(scope);
+        let export_ids = config.source_exports.keys().copied();
+        let health_init = export_ids
+            .map(Some)
+            .chain(None)
+            .map(|id| HealthStatusMessage {
+                id,
+                namespace: Self::STATUS_NAMESPACE,
+                update: HealthStatusUpdate::Running,
+            })
+            .collect::<Vec<_>>()
+            .to_stream(scope);
 
         let health_errs = snapshot_err
             .concat(&repl_err)

src/storage/src/source/postgres.rs

@@ -214,12 +214,17 @@ impl SourceRender for PostgresSourceConnection {
             data_collections.insert(*id, data_stream.as_collection());
         }
 
-        let init = std::iter::once(HealthStatusMessage {
-            id: None,
-            namespace: Self::STATUS_NAMESPACE,
-            update: HealthStatusUpdate::Running,
-        })
-        .to_stream(scope);
+        let export_ids = config.source_exports.keys().copied();
+        let health_init = export_ids
+            .map(Some)
+            .chain(None)
+            .map(|id| HealthStatusMessage {
+                id,
+                namespace: Self::STATUS_NAMESPACE,
+                update: HealthStatusUpdate::Running,
+            })
+            .collect::<Vec<_>>()
+            .to_stream(scope);
 
         // N.B. Note that we don't check ssh tunnel statuses here. We could, but immediately on
         // restart we are going to set the status to an ssh error correctly, so we don't do this
@@ -249,7 +254,7 @@ impl SourceRender for PostgresSourceConnection {
             }
         });
 
-        let health = init.concat(&errs);
+        let health = health_init.concat(&errs);
 
         (
             data_collections,

src/storage/src/source/source_reader_pipeline.rs

@@ -365,30 +365,6 @@ where
 
                 while let Some((cap, data)) = input.next() {
                     for (message, _, _) in data.iter() {
-                        let status = match &message {
-                            Ok(_) => HealthStatusUpdate::running(),
-                            // All errors coming into the data stream are definite.
-                            // Downstream consumers of this data will preserve this
-                            // status.
-                            Err(error) => HealthStatusUpdate::stalled(
-                                error.to_string(),
-                                Some(
-                                    "retracting the errored value may resume the source"
-                                        .to_string(),
-                                ),
-                            ),
-                        };
-
-                        let status = HealthStatusMessage {
-                            id: Some(id),
-                            namespace: C::STATUS_NAMESPACE.clone(),
-                            update: status,
-                        };
-                        if last_status.as_ref() != Some(&status) {
-                            last_status = Some(status.clone());
-                            health_output.session(&health_cap).give(status);
-                        }
-
                         match message {
                             Ok(message) => {
                                 source_statistics.inc_messages_received_by(1);
@@ -397,7 +373,27 @@ where
                                 bytes_read_counter.inc_by(key_len + value_len);
                                 source_statistics.inc_bytes_received_by(key_len + value_len);
                             }
-                            Err(_) => {}
+                            Err(error) => {
+                                // All errors coming into the data stream are definite.
+                                // Downstream consumers of this data will preserve this
+                                // status.
+                                let update = HealthStatusUpdate::stalled(
+                                    error.to_string(),
+                                    Some(
+                                        "retracting the errored value may resume the source"
+                                            .to_string(),
+                                    ),
+                                );
+                                let status = HealthStatusMessage {
+                                    id: Some(id),
+                                    namespace: C::STATUS_NAMESPACE.clone(),
+                                    update,
+                                };
+                                if last_status.as_ref() != Some(&status) {
+                                    last_status = Some(status.clone());
+                                    health_output.session(&health_cap).give(status);
+                                }
+                            }
                         }
                     }
                     let mut output = output.activate();

src/storage/src/source/sql_server.rs

@@ -189,12 +189,18 @@ impl SourceRender for SqlServerSourceConnection {
             data_collections.insert(*id, data_stream.as_collection());
         }
 
-        let health_init = std::iter::once(HealthStatusMessage {
-            id: None,
-            namespace: Self::STATUS_NAMESPACE,
-            update: HealthStatusUpdate::Running,
-        })
-        .to_stream(scope);
+        let export_ids = config.source_exports.keys().copied();
+        let health_init = export_ids
+            .map(Some)
+            .chain(None)
+            .map(|id| HealthStatusMessage {
+                id,
+                namespace: Self::STATUS_NAMESPACE,
+                update: HealthStatusUpdate::Running,
+            })
+            .collect::<Vec<_>>()
+            .to_stream(scope);
+
         let health_errs = repl_errs.concat(&progress_errs).map(move |err| {
             // This update will cause the dataflow to restart
             let err_string = err.display_with_causes().to_string();