storage: allow singleton source on multi-replica clusters, schedule on one replica only

Implements MaterializeInc/database-issues#9079

Per the design in https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/20250127_multi_replica_scheduling_singleton_sources.md

Author: aljoscha

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-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()
     }