storage: allow singleton sources on multi-replica clusters

src/adapter/src/coord/caught_up.rs

@@ -245,7 +245,6 @@ impl Coordinator {
             .controller
             .storage
             .active_ingestions(cluster.id)
-            .iter()
             .copied()
             .filter(|id| !id.is_transient() && !exclude_collections.contains(id))
             .map(|id| {

src/adapter/src/coord/sequencer/inner.rs

@@ -311,25 +311,9 @@ impl Coordinator {
                         .in_cluster
                         .expect("ingestion plans must specify cluster");
                     match ingestion.desc.connection {
-                        GenericSourceConnection::Postgres(_) => {
-                            if let Some(cluster) = self.catalog().try_get_cluster(cluster_id) {
-                                if cluster.replica_ids().len() > 1 {
-                                    return Err(AdapterError::Unsupported(
-                                        "Postgres sources in clusters with >1 replicas",
-                                    ));
-                                }
-                            }
-                        }
-                        GenericSourceConnection::MySql(_) => {
-                            if let Some(cluster) = self.catalog().try_get_cluster(cluster_id) {
-                                if cluster.replica_ids().len() > 1 {
-                                    return Err(AdapterError::Unsupported(
-                                        "MySQL sources in clusters with >1 replicas",
-                                    ));
-                                }
-                            }
-                        }
-                        GenericSourceConnection::Kafka(_)
+                        GenericSourceConnection::Postgres(_)
+                        | GenericSourceConnection::MySql(_)
+                        | GenericSourceConnection::Kafka(_)
                         | GenericSourceConnection::LoadGenerator(_) => {
                             if let Some(cluster) = self.catalog().try_get_cluster(cluster_id) {
                                 let enable_multi_replica_sources = ENABLE_MULTI_REPLICA_SOURCES

src/sql/src/plan/statement/ddl.rs

@@ -5319,12 +5319,14 @@ fn contains_single_replica_objects(scx: &StatementContext, cluster: &dyn Catalog
         let item = scx.catalog.get_item(id);
         let single_replica_source = match item.source_desc() {
             Ok(Some(desc)) => match desc.connection {
-                GenericSourceConnection::Kafka(_) | GenericSourceConnection::LoadGenerator(_) => {
+                GenericSourceConnection::Kafka(_)
+                | GenericSourceConnection::LoadGenerator(_)
+                | GenericSourceConnection::MySql(_)
+                | GenericSourceConnection::Postgres(_) => {
                     let enable_multi_replica_sources =
                         ENABLE_MULTI_REPLICA_SOURCES.get(scx.catalog.system_vars().dyncfgs());
                     !enable_multi_replica_sources
                 }
-                GenericSourceConnection::MySql(_) | GenericSourceConnection::Postgres(_) => true,
             },
             _ => false,
         };

src/storage-client/src/controller.rs

@@ -339,7 +339,10 @@ pub trait StorageController: Debug {
     fn active_collection_metadatas(&self) -> Vec<(GlobalId, CollectionMetadata)>;
 
     /// Returns the IDs of all active ingestions for the given storage instance.
-    fn active_ingestions(&self, instance_id: StorageInstanceId) -> &BTreeSet<GlobalId>;
+    fn active_ingestions(
+        &self,
+        instance_id: StorageInstanceId,
+    ) -> Box<dyn Iterator<Item = &GlobalId> + '_>;
 
     /// Checks whether a collection exists under the given `GlobalId`. Returns
     /// an error if the collection does not exist.

src/storage-controller/src/history.rs

@@ -50,7 +50,7 @@ impl<T: timely::progress::Timestamp + TotalOrder> CommandHistory<T> {
     }
 
     /// Returns an iterator over the contained storage commands.
-    pub fn iter(&self) -> impl Iterator<Item = &StorageCommand<T>> {
+    pub fn iter(&self) -> impl DoubleEndedIterator<Item = &StorageCommand<T>> {
         self.commands.iter()
     }