Remove the default set of schemas from the Compactor

These are effectively never used in prod - we always have at least one
set of schemas available in practice - and once we stop asserting the
schemas on handle creation, this would let us write batches that don't
match the shard schema.

Author: bkirwi

src/persist-client/src/cli/admin.rs

@@ -754,12 +754,7 @@ pub async fn dangerous_force_compaction_and_break_pushdown<K, V, T, D>(
             );
             machine.applier.metrics.compaction.requested.inc();
             let start = Instant::now();
-            let res = Compactor::<K, V, T, D>::compact_and_apply(
-                &machine,
-                req,
-                write.write_schemas.clone(),
-            )
-            .await;
+            let res = Compactor::<K, V, T, D>::compact_and_apply(&machine, req).await;
             let apply_maintenance = match res {
                 Ok(x) => x,
                 Err(err) => {

src/persist-client/src/internal/compact.rs

@@ -180,7 +180,6 @@ where
     pub fn new(
         cfg: PersistConfig,
         metrics: Arc<Metrics>,
-        write_schemas: Schemas<K, V>,
         gc: GarbageCollector<K, V, T, D>,
     ) -> Self {
         let (compact_req_sender, mut compact_req_receiver) = mpsc::channel::<(
@@ -242,14 +241,12 @@ where
                     .queued_seconds
                     .inc_by(enqueued.elapsed().as_secs_f64());
 
-                let write_schemas = write_schemas.clone();
-
                 let compact_span =
                     debug_span!(parent: None, "compact::apply", shard_id=%machine.shard_id());
                 compact_span.follows_from(&Span::current());
                 let gc = gc.clone();
                 mz_ore::task::spawn(|| "PersistCompactionWorker", async move {
-                    let res = Self::compact_and_apply(&machine, req, write_schemas)
+                    let res = Self::compact_and_apply(&machine, req)
                         .instrument(compact_span)
                         .await;
                     if let Ok(maintenance) = res {
@@ -328,7 +325,6 @@ where
     pub(crate) async fn compact_and_apply(
         machine: &Machine<K, V, T, D>,
         req: CompactReq<T>,
-        write_schemas: Schemas<K, V>,
     ) -> Result<RoutineMaintenance, anyhow::Error> {
         let metrics = Arc::clone(&machine.applier.metrics);
         metrics.compaction.started.inc();
@@ -349,38 +345,38 @@ where
         );
         // always use most recent schema from all the Runs we're compacting to prevent Compactors
         // created before the schema was evolved, from trying to "de-evolve" a Part.
-        let compaction_schema_id = req
+        let Some(compaction_schema_id) = req
             .inputs
             .iter()
             .flat_map(|batch| batch.batch.run_meta.iter())
             .filter_map(|run_meta| run_meta.schema)
             // It's an invariant that SchemaIds are ordered.
-            .max();
-        let maybe_compaction_schema = match compaction_schema_id {
-            Some(id) => machine
-                .get_schema(id)
-                .map(|(key_schema, val_schema)| (id, key_schema, val_schema)),
-            None => None,
+            .max()
+        else {
+            metrics.compaction.schema_selection.no_schema.inc();
+            metrics.compaction.failed.inc();
+            return Err(anyhow!(
+                "compacting {shard_id} and spine ids {spine_ids}: could not determine schema id from inputs",
+                shard_id = req.shard_id,
+                spine_ids = mz_ore::str::separated(", ", req.inputs.iter().map(|i| i.id))
+            ));
         };
-        let use_most_recent_schema = COMPACTION_USE_MOST_RECENT_SCHEMA.get(&machine.applier.cfg);
-
-        let compaction_schema = match maybe_compaction_schema {
-            Some((id, key_schema, val_schema)) if use_most_recent_schema => {
-                metrics.compaction.schema_selection.recent_schema.inc();
-                Schemas {
-                    id: Some(id),
-                    key: Arc::new(key_schema),
-                    val: Arc::new(val_schema),
-                }
-            }
-            Some(_) => {
-                metrics.compaction.schema_selection.disabled.inc();
-                write_schemas
-            }
-            None => {
-                metrics.compaction.schema_selection.no_schema.inc();
-                write_schemas
-            }
+        let Some((key_schema, val_schema)) = machine.get_schema(compaction_schema_id) else {
+            metrics.compaction.schema_selection.no_schema.inc();
+            metrics.compaction.failed.inc();
+            return Err(anyhow!(
+                "compacting {shard_id} and spine ids {spine_ids}: schema id {compaction_schema_id} not present in machine state",
+                shard_id = req.shard_id,
+                spine_ids = mz_ore::str::separated(", ", req.inputs.iter().map(|i| i.id))
+            ));
+        };
+
+        metrics.compaction.schema_selection.recent_schema.inc();
+
+        let compaction_schema = Schemas {
+            id: Some(compaction_schema_id),
+            key: Arc::new(key_schema),
+            val: Arc::new(val_schema),
         };
 
         trace!(

src/persist-client/src/internal/metrics.rs

@@ -955,15 +955,13 @@ impl CompactionStepTimings {
 pub struct CompactionSchemaSelection {
     pub(crate) recent_schema: Counter,
     pub(crate) no_schema: Counter,
-    pub(crate) disabled: Counter,
 }
 
 impl CompactionSchemaSelection {
     fn new(schema_selection: CounterVec) -> CompactionSchemaSelection {
         CompactionSchemaSelection {
             recent_schema: schema_selection.with_label_values(&["recent"]),
             no_schema: schema_selection.with_label_values(&["none"]),
-            disabled: schema_selection.with_label_values(&["disabled"]),
         }
     }
 }

src/persist-client/src/internal/trace.rs

@@ -49,7 +49,7 @@
 
 use std::cmp::Ordering;
 use std::collections::{BTreeMap, BTreeSet};
-use std::fmt::Debug;
+use std::fmt::{Debug, Display};
 use std::mem;
 use std::ops::Range;
 use std::sync::Arc;
@@ -726,6 +726,12 @@ pub enum CompactionInput {
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct SpineId(pub usize, pub usize);
 
+impl Display for SpineId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "[{}, {})", self.0, self.1)
+    }
+}
+
 impl Serialize for SpineId {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where

src/persist-client/src/write.rs

@@ -160,14 +160,9 @@ where
         write_schemas: Schemas<K, V>,
     ) -> Self {
         let isolated_runtime = Arc::clone(&machine.isolated_runtime);
-        let compact = cfg.compaction_enabled.then(|| {
-            Compactor::new(
-                cfg.clone(),
-                Arc::clone(&metrics),
-                write_schemas.clone(),
-                gc.clone(),
-            )
-        });
+        let compact = cfg
+            .compaction_enabled
+            .then(|| Compactor::new(cfg.clone(), Arc::clone(&metrics), gc.clone()));
         let debug_state = HandleDebugState {
             hostname: cfg.hostname.to_owned(),
             purpose: purpose.to_owned(),