Merge pull request #30713 from petrosagg/delete-legacy-reclocking

storage: delete old reclocking implementation

Author: petrosagg

misc/python/materialize/mzcompose/__init__.py

@@ -161,7 +161,6 @@ def get_default_system_parameters(
         "storage_statistics_collection_interval": "1000",
         "storage_statistics_interval": "2000",
         "storage_use_continual_feedback_upsert": "true",
-        "storage_use_reclock_v2": "true",
         "timestamp_oracle": "postgres",
         "wait_catalog_consolidation_on_startup": "true",
         "with_0dt_deployment_max_wait": "900s",

src/storage-types/src/dyncfgs.rs

@@ -205,13 +205,6 @@ pub const STORAGE_SUSPEND_AND_RESTART_DELAY: Config<Duration> = Config::new(
     "Delay interval when reconnecting to a source / sink after halt.",
 );
 
-/// Whether to use the new reclock implementation.
-pub const STORAGE_USE_RECLOCK_V2: Config<bool> = Config::new(
-    "storage_use_reclock_v2",
-    true,
-    "Whether to use the new reclock implementation.",
-);
-
 /// Whether to mint reclock bindings based on the latest probed frontier or the currently ingested
 /// frontier.
 pub const STORAGE_RECLOCK_TO_LATEST: Config<bool> = Config::new(
@@ -249,7 +242,6 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&STORAGE_UPSERT_MAX_SNAPSHOT_BATCH_BUFFERING)
         .add(&STORAGE_ROCKSDB_CLEANUP_TRIES)
         .add(&STORAGE_SUSPEND_AND_RESTART_DELAY)
-        .add(&STORAGE_USE_RECLOCK_V2)
         .add(&STORAGE_RECLOCK_TO_LATEST)
         .add(&STORAGE_USE_CONTINUAL_FEEDBACK_UPSERT)
 }

src/storage/src/metrics.rs

@@ -30,18 +30,12 @@
 
 use std::sync::Arc;
 
-use mz_ore::channel::{
-    instrumented_unbounded_channel, InstrumentedUnboundedReceiver, InstrumentedUnboundedSender,
-};
 use mz_ore::metrics::MetricsRegistry;
-use mz_ore::metrics::{DeleteOnDropCounter, MetricVecExt};
 use mz_repr::GlobalId;
-use prometheus::core::AtomicU64;
 
 use crate::statistics::{SinkStatisticsMetricDefs, SourceStatisticsMetricDefs};
 use mz_storage_operators::metrics::BackpressureMetrics;
 
-pub mod channel;
 pub mod decode;
 pub mod sink;
 pub mod source;
@@ -190,39 +184,4 @@ impl StorageMetrics {
     ) -> sink::kafka::KafkaSinkMetrics {
         sink::kafka::KafkaSinkMetrics::new(&self.sink_defs.kafka_defs, sink_id)
     }
-
-    /// Produce an instrumented channel for use in the source pipeline.
-    pub(crate) fn get_instrumented_source_channel<T>(
-        &self,
-        id: GlobalId,
-        worker_id: usize,
-        worker_count: usize,
-        location: &str,
-    ) -> (
-        InstrumentedUnboundedSender<T, DeleteOnDropCounter<'static, AtomicU64, Vec<String>>>,
-        InstrumentedUnboundedReceiver<T, DeleteOnDropCounter<'static, AtomicU64, Vec<String>>>,
-    ) {
-        let sender_metric = self
-            .source_defs
-            .channel_metric_defs
-            .sends
-            .get_delete_on_drop_metric(vec![
-                id.to_string(),
-                worker_id.to_string(),
-                worker_count.to_string(),
-                location.to_string(),
-            ]);
-        let recv_metric = self
-            .source_defs
-            .channel_metric_defs
-            .recvs
-            .get_delete_on_drop_metric(vec![
-                id.to_string(),
-                worker_id.to_string(),
-                worker_count.to_string(),
-                location.to_string(),
-            ]);
-
-        instrumented_unbounded_channel(sender_metric, recv_metric)
-    }
 }

src/storage/src/metrics/channel.rs

@@ -35,7 +35,6 @@ pub(crate) struct GeneralSourceMetricDefs {
     // Source metrics
     pub(crate) capability: UIntGaugeVec,
     pub(crate) resume_upper: IntGaugeVec,
-    pub(crate) inmemory_remap_bindings: UIntGaugeVec,
     pub(crate) commit_upper_ready_times: UIntGaugeVec,
     pub(crate) commit_upper_accepted_times: UIntGaugeVec,
 
@@ -69,11 +68,6 @@ impl GeneralSourceMetricDefs {
                 help: "The timestamp-domain resumption frontier chosen for a source's ingestion",
                 var_labels: ["source_id"],
             )),
-            inmemory_remap_bindings: registry.register(metric!(
-                name: "mz_source_inmemory_remap_bindings",
-                help: "The number of in-memory remap bindings that reclocking a time needs to iterate over.",
-                var_labels: ["source_id", "worker_id"],
-            )),
             commit_upper_ready_times: registry.register(metric!(
                 name: "mz_source_commit_upper_ready_times",
                 help: "The number of ready remap bindings that are held in the reclock commit upper operator.",
@@ -130,8 +124,6 @@ pub(crate) struct SourceMetrics {
     pub(crate) capability: DeleteOnDropGauge<'static, AtomicU64, Vec<String>>,
     /// The resume_upper for a source.
     pub(crate) resume_upper: DeleteOnDropGauge<'static, AtomicI64, Vec<String>>,
-    /// The number of in-memory remap bindings that reclocking a time needs to iterate over.
-    pub(crate) inmemory_remap_bindings: DeleteOnDropGauge<'static, AtomicU64, Vec<String>>,
     /// The number of ready remap bindings that are held in the reclock commit upper operator.
     pub(crate) commit_upper_ready_times: DeleteOnDropGauge<'static, AtomicU64, Vec<String>>,
     /// The number of accepted remap bindings that are held in the reclock commit upper operator.
@@ -156,9 +148,6 @@ impl SourceMetrics {
             resume_upper: defs
                 .resume_upper
                 .get_delete_on_drop_metric(vec![source_id.to_string()]),
-            inmemory_remap_bindings: defs
-                .inmemory_remap_bindings
-                .get_delete_on_drop_metric(vec![source_id.to_string(), worker_id.to_string()]),
             commit_upper_ready_times: defs
                 .commit_upper_ready_times
                 .get_delete_on_drop_metric(vec![source_id.to_string(), worker_id.to_string()]),
@@ -260,9 +249,6 @@ pub(crate) struct SourceMetricDefs {
     pub(crate) kafka_source_defs: kafka::KafkaSourceMetricDefs,
     /// A cluster-wide counter shared across all sources.
     pub(crate) bytes_read: IntCounter,
-
-    // Additional metrics definitions;
-    pub(crate) channel_metric_defs: super::channel::ChannelMetricDefs,
 }
 
 impl SourceMetricDefs {
@@ -276,7 +262,6 @@ impl SourceMetricDefs {
                 name: "mz_bytes_read_total",
                 help: "Count of bytes read from sources",
             )),
-            channel_metric_defs: super::channel::ChannelMetricDefs::register_with(registry),
         }
     }
 }