Merge pull request #34609 from teskje/remove-dataflow-shutdown-duration

compute,sql: remove dataflow shutdown duration introspection

Author: teskje

doc/user/content/sql/system-catalog/mz_introspection.md

@@ -275,19 +275,6 @@ The `mz_dataflow_operator_parents` view describes how [dataflow] operators are n
 
 <!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_dataflow_operator_parents_per_worker -->
 
-## `mz_dataflow_shutdown_durations_histogram`
-
-The `mz_dataflow_shutdown_durations_histogram` view describes a histogram of the time in nanoseconds required to fully shut down dropped [dataflows][dataflow].
-
-<!-- RELATION_SPEC mz_introspection.mz_dataflow_shutdown_durations_histogram -->
-| Field          | Type        | Meaning                                                |
-| -------------- |-------------| --------                                               |
-| `duration_ns`  | [`uint8`]   | The upper bound of the bucket in nanoseconds.          |
-| `count`        | [`numeric`] | The (noncumulative) count of dataflows in this bucket. |
-
-<!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_dataflow_shutdown_durations_histogram_per_worker -->
-<!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_dataflow_shutdown_durations_histogram_raw -->
-
 ## `mz_expected_group_size_advice`
 
 The `mz_expected_group_size_advice` view provides advice on opportunities to set [query hints].

src/catalog/src/builtin.rs

@@ -1945,15 +1945,6 @@ pub static MZ_PEEK_DURATIONS_HISTOGRAM_RAW: LazyLock<BuiltinLog> = LazyLock::new
     access: vec![PUBLIC_SELECT],
 });
 
-pub static MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM_RAW: LazyLock<BuiltinLog> =
-    LazyLock::new(|| BuiltinLog {
-        name: "mz_dataflow_shutdown_durations_histogram_raw",
-        schema: MZ_INTROSPECTION_SCHEMA,
-        oid: oid::LOG_MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM_RAW_OID,
-        variant: LogVariant::Compute(ComputeLog::ShutdownDuration),
-        access: vec![PUBLIC_SELECT],
-    });
-
 pub static MZ_ARRANGEMENT_HEAP_SIZE_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
     name: "mz_arrangement_heap_size_raw",
     schema: MZ_INTROSPECTION_SCHEMA,
@@ -7963,62 +7954,6 @@ GROUP BY type, duration_ns",
     access: vec![PUBLIC_SELECT],
 });
 
-pub static MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM_PER_WORKER: LazyLock<BuiltinView> =
-    LazyLock::new(|| BuiltinView {
-        name: "mz_dataflow_shutdown_durations_histogram_per_worker",
-        schema: MZ_INTROSPECTION_SCHEMA,
-        oid: oid::VIEW_MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM_PER_WORKER_OID,
-        desc: RelationDesc::builder()
-            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
-            .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
-            .with_column("count", SqlScalarType::Int64.nullable(false))
-            .with_key(vec![0, 1])
-            .finish(),
-        column_comments: BTreeMap::new(),
-        sql: "SELECT
-    worker_id, duration_ns, pg_catalog.count(*) AS count
-FROM
-    mz_introspection.mz_dataflow_shutdown_durations_histogram_raw
-GROUP BY
-    worker_id, duration_ns",
-        access: vec![PUBLIC_SELECT],
-    });
-
-pub static MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM: LazyLock<BuiltinView> =
-    LazyLock::new(|| BuiltinView {
-        name: "mz_dataflow_shutdown_durations_histogram",
-        schema: MZ_INTROSPECTION_SCHEMA,
-        oid: oid::VIEW_MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM_OID,
-        desc: RelationDesc::builder()
-            .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
-            .with_column(
-                "count",
-                SqlScalarType::Numeric {
-                    max_scale: Some(NumericMaxScale::ZERO),
-                }
-                .nullable(false),
-            )
-            .with_key(vec![0])
-            .finish(),
-        column_comments: BTreeMap::from_iter([
-            (
-                "duration_ns",
-                "The upper bound of the bucket in nanoseconds.",
-            ),
-            (
-                "count",
-                "The (noncumulative) count of dataflows in this bucket.",
-            ),
-        ]),
-        sql: "
-SELECT
-    duration_ns,
-    pg_catalog.sum(count) AS count
-FROM mz_introspection.mz_dataflow_shutdown_durations_histogram_per_worker
-GROUP BY duration_ns",
-        access: vec![PUBLIC_SELECT],
-    });
-
 pub static MZ_SCHEDULING_ELAPSED_PER_WORKER: LazyLock<BuiltinView> =
     LazyLock::new(|| BuiltinView {
         name: "mz_scheduling_elapsed_per_worker",
@@ -13855,7 +13790,6 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::Log(&MZ_MESSAGE_BATCH_COUNTS_SENT_RAW),
         Builtin::Log(&MZ_ACTIVE_PEEKS_PER_WORKER),
         Builtin::Log(&MZ_PEEK_DURATIONS_HISTOGRAM_RAW),
-        Builtin::Log(&MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM_RAW),
         Builtin::Log(&MZ_ARRANGEMENT_HEAP_CAPACITY_RAW),
         Builtin::Log(&MZ_ARRANGEMENT_HEAP_ALLOCATIONS_RAW),
         Builtin::Log(&MZ_ARRANGEMENT_HEAP_SIZE_RAW),
@@ -13980,8 +13914,6 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::View(&MZ_RECORDS_PER_DATAFLOW),
         Builtin::View(&MZ_PEEK_DURATIONS_HISTOGRAM_PER_WORKER),
         Builtin::View(&MZ_PEEK_DURATIONS_HISTOGRAM),
-        Builtin::View(&MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM_PER_WORKER),
-        Builtin::View(&MZ_DATAFLOW_SHUTDOWN_DURATIONS_HISTOGRAM),
         Builtin::View(&MZ_SCHEDULING_ELAPSED_PER_WORKER),
         Builtin::View(&MZ_SCHEDULING_ELAPSED),
         Builtin::View(&MZ_SCHEDULING_PARKS_HISTOGRAM_PER_WORKER),

src/catalog/src/durable/transaction.rs

@@ -859,7 +859,6 @@ impl<'a> Transaction<'a> {
             LogVariant::Compute(ComputeLog::ArrangementHeapSize) => 24,
             LogVariant::Compute(ComputeLog::ArrangementHeapCapacity) => 25,
             LogVariant::Compute(ComputeLog::ArrangementHeapAllocations) => 26,
-            LogVariant::Compute(ComputeLog::ShutdownDuration) => 27,
             LogVariant::Compute(ComputeLog::ErrorCount) => 28,
             LogVariant::Compute(ComputeLog::HydrationTime) => 29,
             LogVariant::Compute(ComputeLog::LirMapping) => 30,

src/compute-client/src/logging.rs

@@ -128,8 +128,6 @@ pub enum ComputeLog {
     ArrangementHeapCapacity,
     /// Arrangement heap allocations.
     ArrangementHeapAllocations,
-    /// A histogram over dataflow shutdown durations.
-    ShutdownDuration,
     /// Counts of errors in exported collections.
     ErrorCount,
     /// Hydration times of exported collections.
@@ -300,11 +298,6 @@ impl LogVariant {
                 .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
                 .finish(),
 
-            LogVariant::Compute(ComputeLog::ShutdownDuration) => RelationDesc::builder()
-                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
-                .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
-                .finish(),
-
             LogVariant::Compute(ComputeLog::ErrorCount) => RelationDesc::builder()
                 .with_column("export_id", SqlScalarType::String.nullable(false))
                 .with_column("worker_id", SqlScalarType::UInt64.nullable(false))

src/compute/src/logging/compute.rs

@@ -338,8 +338,6 @@ pub(super) fn construct<S: Scheduler + 'static, G: Scope<Timestamp = Timestamp>>
         let mut peek_out = OutputBuilder::from(peek_out);
         let (peek_duration_out, peek_duration) = demux.new_output();
         let mut peek_duration_out = OutputBuilder::from(peek_duration_out);
-        let (shutdown_duration_out, shutdown_duration) = demux.new_output();
-        let mut shutdown_duration_out = OutputBuilder::from(shutdown_duration_out);
         let (arrangement_heap_size_out, arrangement_heap_size) = demux.new_output();
         let mut arrangement_heap_size_out = OutputBuilder::from(arrangement_heap_size_out);
         let (arrangement_heap_capacity_out, arrangement_heap_capacity) = demux.new_output();
@@ -366,7 +364,6 @@ pub(super) fn construct<S: Scheduler + 'static, G: Scope<Timestamp = Timestamp>>
                 let mut import_frontier = import_frontier_out.activate();
                 let mut peek = peek_out.activate();
                 let mut peek_duration = peek_duration_out.activate();
-                let mut shutdown_duration = shutdown_duration_out.activate();
                 let mut arrangement_heap_size = arrangement_heap_size_out.activate();
                 let mut arrangement_heap_capacity = arrangement_heap_capacity_out.activate();
                 let mut arrangement_heap_allocations = arrangement_heap_allocations_out.activate();
@@ -383,7 +380,6 @@ pub(super) fn construct<S: Scheduler + 'static, G: Scope<Timestamp = Timestamp>>
                         import_frontier: import_frontier.session_with_builder(&cap),
                         peek: peek.session_with_builder(&cap),
                         peek_duration: peek_duration.session_with_builder(&cap),
-                        shutdown_duration: shutdown_duration.session_with_builder(&cap),
                         arrangement_heap_allocations: arrangement_heap_allocations.session_with_builder(&cap),
                         arrangement_heap_capacity: arrangement_heap_capacity.session_with_builder(&cap),
                         arrangement_heap_size: arrangement_heap_size.session_with_builder(&cap),
@@ -424,7 +420,6 @@ pub(super) fn construct<S: Scheduler + 'static, G: Scope<Timestamp = Timestamp>>
             (OperatorHydrationStatus, operator_hydration_status),
             (PeekCurrent, peek),
             (PeekDuration, peek_duration),
-            (ShutdownDuration, shutdown_duration),
         ];
 
         // Build the output arrangements.
@@ -476,12 +471,6 @@ struct DemuxState<A> {
     scratch_string_b: String,
     /// State tracked per dataflow export.
     exports: BTreeMap<GlobalId, ExportState>,
-    /// Maps live dataflows to counts of their exports.
-    dataflow_export_counts: BTreeMap<usize, u32>,
-    /// Maps dropped dataflows to their drop time.
-    dataflow_drop_times: BTreeMap<usize, Duration>,
-    /// Contains dataflows that have shut down but not yet been dropped.
-    shutdown_dataflows: BTreeSet<usize>,
     /// Maps pending peeks to their installation time.
     peek_stash: BTreeMap<Uuid, Duration>,
     /// Arrangement size stash.
@@ -514,8 +503,6 @@ struct DemuxState<A> {
     peek_duration_packer: PermutedRowPacker,
     /// A row packer for the peek output.
     peek_packer: PermutedRowPacker,
-    /// A row packer for the shutdown duration output.
-    shutdown_duration_packer: PermutedRowPacker,
     /// A row packer for the hydration time output.
     hydration_time_packer: PermutedRowPacker,
 }
@@ -528,9 +515,6 @@ impl<A: Scheduler> DemuxState<A> {
             scratch_string_a: String::new(),
             scratch_string_b: String::new(),
             exports: Default::default(),
-            dataflow_export_counts: Default::default(),
-            dataflow_drop_times: Default::default(),
-            shutdown_dataflows: Default::default(),
             peek_stash: Default::default(),
             arrangement_size: Default::default(),
             lir_mapping: Default::default(),
@@ -554,7 +538,6 @@ impl<A: Scheduler> DemuxState<A> {
             ),
             peek_duration_packer: PermutedRowPacker::new(ComputeLog::PeekDuration),
             peek_packer: PermutedRowPacker::new(ComputeLog::PeekCurrent),
-            shutdown_duration_packer: PermutedRowPacker::new(ComputeLog::ShutdownDuration),
         }
     }
 
@@ -727,14 +710,6 @@ impl<A: Scheduler> DemuxState<A> {
             Datum::MzTimestamp(time),
         ])
     }
-
-    /// Pack a shutdown duration update key-value for the given time bucket.
-    fn pack_shutdown_duration_update(&mut self, bucket: u128) -> (&RowRef, &RowRef) {
-        self.shutdown_duration_packer.pack_slice(&[
-            Datum::UInt64(u64::cast_from(self.worker_id)),
-            Datum::UInt64(bucket.try_into().expect("bucket too big")),
-        ])
-    }
 }
 
 /// State tracked for each dataflow export.
@@ -781,7 +756,6 @@ struct DemuxOutput<'a, 'b> {
     import_frontier: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>,
     peek: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>,
     peek_duration: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>,
-    shutdown_duration: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>,
     arrangement_heap_allocations: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>,
     arrangement_heap_capacity: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>,
     arrangement_heap_size: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>,
@@ -862,12 +836,6 @@ impl<A: Scheduler> DemuxHandler<'_, '_, '_, A> {
             error!(%export_id, "export already registered");
         }
 
-        *self
-            .state
-            .dataflow_export_counts
-            .entry(dataflow_index)
-            .or_default() += 1;
-
         // Insert hydration time logging for this export.
         let datum = self.state.pack_hydration_time_update(export_id, None);
         self.output.hydration_time.give((datum, ts, Diff::ONE));
@@ -889,18 +857,6 @@ impl<A: Scheduler> DemuxHandler<'_, '_, '_, A> {
         let datum = self.state.pack_export_update(export_id, dataflow_index);
         self.output.export.give((datum, ts, Diff::MINUS_ONE));
 
-        match self.state.dataflow_export_counts.get_mut(&dataflow_index) {
-            entry @ Some(0) | entry @ None => {
-                error!(
-                    %export_id,
-                    %dataflow_index,
-                    "invalid dataflow_export_counts entry at time of export drop: {entry:?}",
-                );
-            }
-            Some(1) => self.handle_dataflow_dropped(dataflow_index),
-            Some(count) => *count -= 1,
-        }
-
         // Remove error count logging for this export.
         if export.error_count != Diff::ZERO {
             let datum = self
@@ -928,46 +884,12 @@ impl<A: Scheduler> DemuxHandler<'_, '_, '_, A> {
         }
     }
 
-    fn handle_dataflow_dropped(&mut self, dataflow_index: usize) {
-        let ts = self.ts();
-        self.state.dataflow_export_counts.remove(&dataflow_index);
-
-        if self.state.shutdown_dataflows.remove(&dataflow_index) {
-            // Dataflow has already shut down before it was dropped.
-            let datum = self.state.pack_shutdown_duration_update(0);
-            self.output.shutdown_duration.give((datum, ts, Diff::ONE));
-        } else {
-            // Dataflow has not yet shut down.
-            let existing = self
-                .state
-                .dataflow_drop_times
-                .insert(dataflow_index, self.time);
-            if existing.is_some() {
-                error!(%dataflow_index, "dataflow already dropped");
-            }
-        }
-    }
-
     fn handle_dataflow_shutdown(
         &mut self,
         DataflowShutdownReference { dataflow_index }: Ref<'_, DataflowShutdown>,
     ) {
         let ts = self.ts();
 
-        if let Some(start) = self.state.dataflow_drop_times.remove(&dataflow_index) {
-            // Dataflow has already been dropped.
-            let elapsed_ns = self.time.saturating_sub(start).as_nanos();
-            let elapsed_pow = elapsed_ns.next_power_of_two();
-            let datum = self.state.pack_shutdown_duration_update(elapsed_pow);
-            self.output.shutdown_duration.give((datum, ts, Diff::ONE));
-        } else {
-            // Dataflow has not yet been dropped.
-            let was_new = self.state.shutdown_dataflows.insert(dataflow_index);
-            if !was_new {
-                error!(%dataflow_index, "dataflow already shutdown");
-            }
-        }
-
         // We deal with any `GlobalId` based mappings in this event.
         if let Some(global_ids) = self.state.dataflow_global_ids.remove(&dataflow_index) {
             for global_id in global_ids {