storage-{client,controller}: stop being generic over the timestamp type (MaterializeInc#36043)

The storage client code is generic over a `T: Timestamp`, though it's
only ever used with `mz_repr::Timestamp`. There are no plans to support
multiple timestamp types, and if we want to make changes to the
timestamp type, we can always make them to `mz_repr::Timestamp`. Thus
the genericity doesn't give us anything except code noise.

Change the storage client and storage controller code to not be generic
over the timestamp type anymore.

Author: teskje

src/adapter/src/catalog/open.rs

@@ -638,9 +638,7 @@ impl Catalog {
     /// collections created between versions.
     async fn initialize_storage_state(
         &mut self,
-        storage_collections: &Arc<
-            dyn StorageCollections<Timestamp = mz_repr::Timestamp> + Send + Sync,
-        >,
+        storage_collections: &Arc<dyn StorageCollections + Send + Sync>,
     ) -> Result<(), mz_catalog::durable::CatalogError> {
         let collections = self
             .entries()

src/adapter/src/catalog/transact.rs

@@ -421,9 +421,7 @@ impl Catalog {
         &mut self,
         // n.b. this is an option to prevent us from needing to build out a
         // dummy impl of `StorageController` for tests.
-        storage_collections: Option<
-            &mut Arc<dyn StorageCollections<Timestamp = mz_repr::Timestamp> + Send + Sync>,
-        >,
+        storage_collections: Option<&mut Arc<dyn StorageCollections + Send + Sync>>,
         oracle_write_ts: mz_repr::Timestamp,
         session: Option<&ConnMeta>,
         ops: Vec<Op>,
@@ -654,9 +652,7 @@ impl Catalog {
     #[instrument(name = "catalog::transact_inner")]
     async fn transact_inner(
         mode: TransactInnerMode,
-        storage_collections: Option<
-            &mut Arc<dyn StorageCollections<Timestamp = mz_repr::Timestamp> + Send + Sync>,
-        >,
+        storage_collections: Option<&mut Arc<dyn StorageCollections + Send + Sync>>,
         oracle_write_ts: mz_repr::Timestamp,
         session: Option<&ConnMeta>,
         ops: Vec<Op>,

src/adapter/src/command.rs

@@ -456,12 +456,8 @@ pub struct StartupResponse {
     /// Map of (name, VarInput::Flat) tuples of session default variables that should be set.
     pub session_defaults: BTreeMap<String, OwnedVarInput>,
     pub catalog: Arc<Catalog>,
-    pub storage_collections: Arc<
-        dyn mz_storage_client::storage_collections::StorageCollections<
-                Timestamp = mz_repr::Timestamp,
-            > + Send
-            + Sync,
-    >,
+    pub storage_collections:
+        Arc<dyn mz_storage_client::storage_collections::StorageCollections + Send + Sync>,
     pub transient_id_gen: Arc<TransientIdGen>,
     pub optimizer_metrics: OptimizerMetrics,
     pub persist_client: PersistClient,

src/adapter/src/coord/catalog_implications.rs

@@ -49,7 +49,7 @@ use mz_ore::instrument;
 use mz_ore::retry::Retry;
 use mz_ore::str::StrExt;
 use mz_ore::task;
-use mz_repr::{CatalogItemId, GlobalId, RelationVersion, RelationVersionSelector, Timestamp};
+use mz_repr::{CatalogItemId, GlobalId, RelationVersion, RelationVersionSelector};
 use mz_sql::plan::ConnectionDetails;
 use mz_storage_client::controller::{CollectionDescription, DataSource};
 use mz_storage_types::connections::PostgresConnection;
@@ -1056,7 +1056,7 @@ impl Coordinator {
     #[instrument(level = "debug")]
     async fn create_table_collections(
         &mut self,
-        table_collections_to_create: BTreeMap<GlobalId, CollectionDescription<Timestamp>>,
+        table_collections_to_create: BTreeMap<GlobalId, CollectionDescription>,
         execution_timestamps_to_set: BTreeSet<StatementLoggingId>,
     ) -> Result<(), AdapterError> {
         // If we have tables, determine the initial validity for the table.
@@ -1101,7 +1101,7 @@ impl Coordinator {
     #[instrument(level = "debug")]
     async fn create_source_collections(
         &mut self,
-        source_collections_to_create: BTreeMap<GlobalId, CollectionDescription<Timestamp>>,
+        source_collections_to_create: BTreeMap<GlobalId, CollectionDescription>,
     ) -> Result<(), AdapterError> {
         let storage_metadata = self.catalog.state().storage_metadata();
 
@@ -1141,7 +1141,7 @@ impl Coordinator {
     async fn handle_create_table(
         &self,
         ctx: &Option<&mut ExecuteContext>,
-        storage_collections_to_create: &mut BTreeMap<GlobalId, CollectionDescription<Timestamp>>,
+        storage_collections_to_create: &mut BTreeMap<GlobalId, CollectionDescription>,
         storage_policies_to_initialize: &mut BTreeMap<CompactionWindow, BTreeSet<GlobalId>>,
         execution_timestamps_to_set: &mut BTreeSet<StatementLoggingId>,
         table_id: CatalogItemId,
@@ -1192,7 +1192,7 @@ impl Coordinator {
                         let global_ingestion_id =
                             self.catalog().get_entry(ingestion_id).latest_global_id();
 
-                        let collection_desc = CollectionDescription::<Timestamp> {
+                        let collection_desc = CollectionDescription {
                             desc: table.desc.latest(),
                             data_source: DataSource::IngestionExport {
                                 ingestion_id: global_ingestion_id,
@@ -1243,7 +1243,7 @@ impl Coordinator {
                         );
                         let desc = table.desc.latest();
 
-                        let collection_desc = CollectionDescription::<Timestamp> {
+                        let collection_desc = CollectionDescription {
                             desc,
                             data_source: DataSource::Webhook,
                             since: None,
@@ -1368,7 +1368,7 @@ impl Coordinator {
     #[instrument(level = "debug")]
     async fn handle_create_source(
         &self,
-        storage_collections_to_create: &mut BTreeMap<GlobalId, CollectionDescription<Timestamp>>,
+        storage_collections_to_create: &mut BTreeMap<GlobalId, CollectionDescription>,
         storage_policies_to_initialize: &mut BTreeMap<CompactionWindow, BTreeSet<GlobalId>>,
         item_id: CatalogItemId,
         source: Source,
@@ -1447,7 +1447,7 @@ impl Coordinator {
 
         storage_collections_to_create.insert(
             source.global_id,
-            CollectionDescription::<Timestamp> {
+            CollectionDescription {
                 desc: source.desc.clone(),
                 data_source,
                 timeline: Some(source.timeline),

src/adapter/src/coord/sequencer.rs

@@ -52,7 +52,6 @@ use mz_transform::dataflow::DataflowMetainfo;
 use mz_transform::notice::{OptimizerNoticeApi, OptimizerNoticeKind, RawOptimizerNotice};
 use mz_transform::{EmptyStatisticsOracle, StatisticsOracle};
 use timely::progress::Antichain;
-use timely::progress::Timestamp as TimelyTimestamp;
 use tokio::sync::oneshot;
 use tracing::{Instrument, Level, Span, event, warn};
 
@@ -1074,7 +1073,7 @@ pub(crate) async fn explain_pushdown_future_inner<
 >(
     session: &Session,
     catalog: &Catalog,
-    storage_collections: &Arc<dyn StorageCollections<Timestamp = Timestamp> + Send + Sync>,
+    storage_collections: &Arc<dyn StorageCollections + Send + Sync>,
     as_of: Antichain<Timestamp>,
     mz_now: ResultSpec<'static>,
     imports: I,
@@ -1232,10 +1231,11 @@ pub(crate) async fn statistics_oracle(
     query_as_of: &Antichain<Timestamp>,
     is_oneshot: bool,
     system_config: &vars::SystemVars,
-    storage_collections: &dyn StorageCollections<Timestamp = Timestamp>,
+    storage_collections: &dyn StorageCollections,
 ) -> Result<Box<dyn StatisticsOracle>, AdapterError> {
     if !session.vars().enable_session_cardinality_estimates() {
-        return Ok(Box::new(EmptyStatisticsOracle));
+        let stats: Box<dyn StatisticsOracle> = Box::new(EmptyStatisticsOracle);
+        return Ok(stats);
     }
 
     let timeout = if is_oneshot {
@@ -1272,11 +1272,11 @@ struct CachedStatisticsOracle {
 }
 
 impl CachedStatisticsOracle {
-    pub async fn new<T: TimelyTimestamp>(
+    pub async fn new(
         ids: &BTreeSet<GlobalId>,
-        as_of: &Antichain<T>,
-        storage_collections: &dyn StorageCollections<Timestamp = T>,
-    ) -> Result<Self, StorageError<T>> {
+        as_of: &Antichain<Timestamp>,
+        storage_collections: &dyn StorageCollections,
+    ) -> Result<Self, StorageError> {
         let mut cache = BTreeMap::new();
 
         for id in ids {

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

@@ -118,7 +118,7 @@ use crate::util::{ClientTransmitter, ResultExt, viewable_variables};
 use crate::{PeekResponseUnary, ReadHolds};
 
 /// A future that resolves to a real-time recency timestamp.
-type RtrTimestampFuture = BoxFuture<'static, Result<Timestamp, StorageError<Timestamp>>>;
+type RtrTimestampFuture = BoxFuture<'static, Result<Timestamp, StorageError>>;
 
 mod cluster;
 mod copy_from;

src/adapter/src/error.rs

@@ -218,7 +218,7 @@ pub enum AdapterError {
     /// Another session modified the Catalog while this transaction was open.
     DDLTransactionRace,
     /// An error occurred in the storage layer
-    Storage(mz_storage_types::controller::StorageError<mz_repr::Timestamp>),
+    Storage(mz_storage_types::controller::StorageError),
     /// An error occurred in the compute layer
     Compute(anyhow::Error),
     /// An error in the orchestrator layer
@@ -1227,8 +1227,8 @@ impl From<oneshot::error::RecvError> for AdapterError {
     }
 }
 
-impl From<StorageError<mz_repr::Timestamp>> for AdapterError {
-    fn from(e: StorageError<mz_repr::Timestamp>) -> Self {
+impl From<StorageError> for AdapterError {
+    fn from(e: StorageError) -> Self {
         AdapterError::Storage(e)
     }
 }

src/adapter/src/peek_client.rs

@@ -44,11 +44,8 @@ use crate::statement_logging::{
 use crate::{AdapterError, Client, CollectionIdBundle, ReadHolds, statement_logging};
 
 /// Storage collections trait alias we need to consult for since/frontiers.
-pub type StorageCollectionsHandle = Arc<
-    dyn mz_storage_client::storage_collections::StorageCollections<Timestamp = Timestamp>
-        + Send
-        + Sync,
->;
+pub type StorageCollectionsHandle =
+    Arc<dyn mz_storage_client::storage_collections::StorageCollections + Send + Sync>;
 
 /// Clients needed for peek sequencing in the Adapter Frontend.
 #[derive(Debug)]

src/adapter/src/util.rs

@@ -336,7 +336,7 @@ impl ShouldTerminateGracefully for FenceError {
     }
 }
 
-impl<T> ShouldTerminateGracefully for StorageError<T> {
+impl ShouldTerminateGracefully for StorageError {
     fn should_terminate_gracefully(&self) -> bool {
         match self {
             StorageError::ResourceExhausted(_)

src/adapter/src/webhook.rs

@@ -15,7 +15,7 @@ use anyhow::Context;
 use chrono::{DateTime, Utc};
 use derivative::Derivative;
 use mz_ore::cast::CastFrom;
-use mz_repr::{Datum, Diff, Row, RowArena, Timestamp};
+use mz_repr::{Datum, Diff, Row, RowArena};
 use mz_secrets::SecretsReader;
 use mz_secrets::cache::CachingSecretsReader;
 use mz_sql::plan::{WebhookBodyFormat, WebhookHeaders, WebhookValidation, WebhookValidationSecret};
@@ -55,7 +55,7 @@ pub enum AppendWebhookError {
     #[error("internal error: {0:?}")]
     InternalError(#[from] anyhow::Error),
     #[error("internal storage failure! {0:?}")]
-    StorageError(#[from] StorageError<mz_repr::Timestamp>),
+    StorageError(#[from] StorageError),
 }
 
 /// Contains all of the components necessary for running webhook validation.
@@ -245,7 +245,7 @@ pub struct AppendWebhookResponse {
 /// gets modified.
 #[derive(Clone, Debug)]
 pub struct WebhookAppender {
-    tx: MonotonicAppender<Timestamp>,
+    tx: MonotonicAppender,
     guard: WebhookAppenderGuard,
     // Shared statistics related to this webhook.
     stats: Arc<WebhookStatistics>,
@@ -292,7 +292,7 @@ impl WebhookAppender {
     }
 
     pub(crate) fn new(
-        tx: MonotonicAppender<Timestamp>,
+        tx: MonotonicAppender,
         guard: WebhookAppenderGuard,
         stats: Arc<WebhookStatistics>,
     ) -> Self {