storage: infrastructure for independent source output streams

This PR puts in the basis for allowing a source implementation to
produce an independent DD collection per output.

The situation before this PR was that the `SourceRender` trait required
a single, multiplexed DD collection to be produced, of type `(usize,
D)`, where the `usize` designated the output. Since all outputs were
multiplexed, a single frontier had to describe their overall progress,
which described the "slowest" one. This is generally fine when all
outputs more or less march forwards together but it's not ok when a new
subsource is added to a source that has otherwise been running for a
while. In this situation the upper frontier of the multiplexed
collection would necessarily have to stay stuck until the new subsource
finished its snapshot and caught up with the other ones, making the
previously healthy sources unavailable for all this time.

This PR fixes this by requiring a `BTreeMap<GlobalId, Collection>`
output type from source implementations. This way each subsource can be
driven independently and a new subsource can be added without imposing a
frontier stall to the previously ingested subsources.

This PR does not change any of the source implementations to take
advantage of this new interface, since that would create a giant PR.
Instead, this only changes the interface and the fallout of all the
changes in the various generic parts of the pipeline. Follow up PRs will
be done that target individual source implementation and change them to
be directly produce multiple collections.

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

Author: petrosagg

src/storage-types/src/sources.rs

@@ -141,43 +141,6 @@ impl<S> IngestionDescription<S> {
     }
 }
 
-impl<S: Clone> IngestionDescription<S> {
-    pub fn indexed_source_exports(
-        &self,
-        primary_source_id: &GlobalId,
-    ) -> BTreeMap<GlobalId, IndexedSourceExport<S>> {
-        let mut source_exports = BTreeMap::new();
-        // `self.source_exports` contains all source-exports (e.g. subsources & tables) as well as
-        // the primary source relation. It's not guaranteed that the primary source relation is
-        // the first element in the map, however it much be set to output 0 to align with
-        // assumptions in source implementations. This is the case even if the primary
-        // export will not have any data output for it, since output 0 is the convention
-        // used for errors that should halt the entire source dataflow.
-        // TODO: See if we can simplify this to avoid needing to include the primary output
-        // if no data will be exported to it. This requires refactoring all error output handling.
-        let mut next_output = 1;
-        for (id, export) in self.source_exports.iter() {
-            let ingestion_output = if id == primary_source_id {
-                0
-            } else {
-                let idx = next_output;
-                next_output += 1;
-                idx
-            };
-
-            source_exports.insert(
-                *id,
-                IndexedSourceExport {
-                    ingestion_output,
-                    export: export.clone(),
-                },
-            );
-        }
-
-        source_exports
-    }
-}
-
 impl<S: Debug + Eq + PartialEq + AlterCompatible> AlterCompatible for IngestionDescription<S> {
     fn alter_compatible(
         &self,
@@ -278,14 +241,6 @@ impl<R: ConnectionResolver> IntoInlineConnection<IngestionDescription, R>
     }
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Arbitrary)]
-pub struct IndexedSourceExport<S = ()> {
-    /// Which output index from the ingestion this export refers to.
-    pub ingestion_output: usize,
-    /// The SourceExport
-    pub export: SourceExport<S>,
-}
-
 #[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Arbitrary)]
 pub struct SourceExport<S = (), C: ConnectionAccess = InlinedConnection> {
     /// The collection metadata needed to write the exported data

src/storage/src/decode.rs

@@ -571,7 +571,7 @@ pub fn render_decode_delimited<G: Scope, FromTime: Timestamp>(
     let health = transient_errors.map(|err: Rc<CsrConnectError>| {
         let halt_status = HealthStatusUpdate::halting(err.display_with_causes().to_string(), None);
         HealthStatusMessage {
-            index: 0,
+            id: None,
             namespace: if matches!(&*err, CsrConnectError::Ssh(_)) {
                 StatusNamespace::Ssh
             } else {