Enforce exactly one sink export

Author: ggevay

src/adapter/src/optimize/copy_to.rs

@@ -149,6 +149,11 @@ impl GlobalLirPlan {
         &self.df_desc
     }
 
+    /// Returns the id of the dataflow's sink export.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the dataflow has no sink exports or has more than one.
     pub fn sink_id(&self) -> GlobalId {
         self.df_desc.sink_id()
     }

src/adapter/src/optimize/subscribe.rs

@@ -152,19 +152,27 @@ pub struct GlobalLirPlan {
 }
 
 impl GlobalLirPlan {
+    /// Returns the id of the dataflow's sink export.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the dataflow has no sink exports or has more than one.
     pub fn sink_id(&self) -> GlobalId {
-        let sink_exports = &self.df_desc.sink_exports;
-        let sink_id = sink_exports.keys().next().expect("valid sink");
-        *sink_id
+        self.df_desc.sink_id()
     }
 
     pub fn as_of(&self) -> Option<Timestamp> {
         self.df_desc.as_of.clone().map(|as_of| as_of.into_element())
     }
 
+    /// Returns the description of the dataflow's sink export.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the dataflow has no sink exports or has more than one.
     pub fn sink_desc(&self) -> &ComputeSinkDesc {
         let sink_exports = &self.df_desc.sink_exports;
-        let sink_desc = sink_exports.values().next().expect("valid sink");
+        let sink_desc = sink_exports.values().into_element();
         sink_desc
     }
 }

src/compute-types/src/dataflows.rs

@@ -13,6 +13,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use std::fmt;
 
 use mz_expr::{CollectionPlan, MirRelationExpr, MirScalarExpr, OptimizedMirRelationExpr};
+use mz_ore::collections::CollectionExt;
 use mz_ore::soft_assert_or_log;
 use mz_repr::refresh_schedule::RefreshSchedule;
 use mz_repr::{GlobalId, SqlRelationType};
@@ -397,14 +398,14 @@ impl<P, S, T> DataflowDescription<P, S, T> {
         build
     }
 
-    /// Returns the id of the first sink export.
+    /// Returns the id of the dataflow's sink export.
     ///
     /// # Panics
     ///
-    /// Panics if the dataflow has no sink exports.
+    /// Panics if the dataflow has no sink exports or has more than one.
     pub fn sink_id(&self) -> GlobalId {
         let sink_exports = &self.sink_exports;
-        let sink_id = sink_exports.keys().next().expect("valid sink");
+        let sink_id = sink_exports.keys().into_element();
         *sink_id
     }
 }