Protobuf support for LoggingConfig

src/dataflow-types/Cargo.toml

@@ -52,6 +52,8 @@ tokio-util = { version = "0.7.1", features = ["codec"] }
 tracing = "0.1.34"
 url = { version = "2.2.2", features = ["serde"] }
 uuid = { version = "0.8.2", features = ["serde", "v4"] }
+proptest = { git = "https://github.com/MaterializeInc/proptest.git", default-features = false, features = ["std"]}
+proptest-derive = { git = "https://github.com/MaterializeInc/proptest.git"}
 
 [build-dependencies]
 prost-build = { version = "0.10.1", features = ["vendored"] }

src/dataflow-types/build.rs

@@ -9,7 +9,18 @@
 
 fn main() {
     prost_build::Config::new()
-        .type_attribute(".", "#[derive(Eq, serde::Serialize, serde::Deserialize)]")
-        .compile_protos(&["dataflow-types/src/postgres_source.proto"], &[".."])
+        .extern_path(".mz_repr.global_id", "::mz_repr::global_id")
+        .extern_path(".mz_repr.proto", "::mz_repr::proto")
+        .type_attribute(
+            ".mz_dataflow_types.postgres_source",
+            "#[derive(Eq, serde::Serialize, serde::Deserialize)]",
+        )
+        .compile_protos(
+            &[
+                "dataflow-types/src/logging.proto",
+                "dataflow-types/src/postgres_source.proto",
+            ],
+            &[".."],
+        )
         .unwrap();
 }

src/dataflow-types/src/logging.proto

@@ -0,0 +1,66 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+syntax = "proto3";
+
+import "google/protobuf/empty.proto";
+import "repr/src/global_id.proto";
+import "repr/src/proto.proto";
+
+package mz_dataflow_types.logging;
+
+message ProtoActiveLog {
+    ProtoLogVariant key = 1;
+    mz_repr.global_id.ProtoGlobalId value = 2;
+}
+
+message ProtoTimelyLog {
+    oneof kind {
+        google.protobuf.Empty operates = 1;
+        google.protobuf.Empty channels = 2;
+        google.protobuf.Empty elapsed = 3;
+        google.protobuf.Empty histogram = 4;
+        google.protobuf.Empty addresses = 5;
+        google.protobuf.Empty parks = 6;
+        google.protobuf.Empty messages_sent = 7;
+        google.protobuf.Empty messages_received = 8;
+        google.protobuf.Empty reachability = 9;
+    }
+}
+
+message ProtoDifferentialLog {
+    oneof kind {
+        google.protobuf.Empty arrangement_batches = 1;
+        google.protobuf.Empty arrangement_records = 2;
+        google.protobuf.Empty sharing = 3;
+    }
+}
+
+message ProtoMaterializedLog {
+    oneof kind {
+        google.protobuf.Empty dataflow_current = 1;
+        google.protobuf.Empty dataflow_dependency = 2;
+        google.protobuf.Empty frontier_current = 3;
+        google.protobuf.Empty peek_current = 4;
+        google.protobuf.Empty peek_duration = 5;
+    }
+}
+message ProtoLogVariant {
+    oneof kind {
+        ProtoTimelyLog timely = 1;
+        ProtoDifferentialLog differential = 2;
+        ProtoMaterializedLog materialized = 3;
+    }
+}
+
+message ProtoLoggingConfig {
+    mz_repr.proto.ProtoU128 granularity_ns = 1;
+    repeated ProtoActiveLog active_logs = 2;
+    bool log_logging = 3;
+}

src/dataflow-types/src/logging.rs

@@ -9,12 +9,16 @@
 
 use std::collections::HashMap;
 
+use proptest_derive::Arbitrary;
 use serde::{Deserialize, Serialize};
 
+use mz_repr::proto::{FromProtoIfSome, ProtoRepr, TryFromProtoError, TryIntoIfSome};
 use mz_repr::{GlobalId, RelationDesc, ScalarType};
 
+include!(concat!(env!("OUT_DIR"), "/mz_dataflow_types.logging.rs"));
+
 /// Logging configuration.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Arbitrary, Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct LoggingConfig {
     pub granularity_ns: u128,
     pub active_logs: HashMap<LogVariant, GlobalId>,
@@ -29,14 +33,97 @@ impl LoggingConfig {
     }
 }
 
-#[derive(Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
+impl From<&LoggingConfig> for ProtoLoggingConfig {
+    fn from(x: &LoggingConfig) -> Self {
+        ProtoLoggingConfig {
+            granularity_ns: Some(x.granularity_ns.into_proto()),
+            active_logs: x.active_logs.iter().map(Into::into).collect(),
+            log_logging: x.log_logging,
+        }
+    }
+}
+
+impl TryFrom<ProtoLoggingConfig> for LoggingConfig {
+    type Error = TryFromProtoError;
+
+    fn try_from(x: ProtoLoggingConfig) -> Result<Self, Self::Error> {
+        Ok(LoggingConfig {
+            granularity_ns: x
+                .granularity_ns
+                .from_proto_if_some("ProtoLoggingConfig::granularity_ns")?,
+            active_logs: x
+                .active_logs
+                .into_iter()
+                .map(TryInto::try_into)
+                .collect::<Result<HashMap<_, _>, _>>()?,
+            log_logging: x.log_logging,
+        })
+    }
+}
+
+impl From<(&LogVariant, &GlobalId)> for ProtoActiveLog {
+    fn from(x: (&LogVariant, &GlobalId)) -> Self {
+        ProtoActiveLog {
+            key: Some(x.0.into()),
+            value: Some(x.1.into()),
+        }
+    }
+}
+
+impl TryFrom<ProtoActiveLog> for (LogVariant, GlobalId) {
+    type Error = TryFromProtoError;
+
+    fn try_from(x: ProtoActiveLog) -> Result<Self, Self::Error> {
+        Ok((
+            x.key.try_into_if_some("ProtoActiveLog::key")?,
+            x.value.try_into_if_some("ProtoActiveLog::value")?,
+        ))
+    }
+}
+
+#[derive(Arbitrary, Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
 pub enum LogVariant {
     Timely(TimelyLog),
     Differential(DifferentialLog),
     Materialized(MaterializedLog),
 }
 
-#[derive(Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
+impl From<&LogVariant> for ProtoLogVariant {
+    fn from(x: &LogVariant) -> Self {
+        ProtoLogVariant {
+            kind: Some(match x {
+                LogVariant::Timely(timely) => proto_log_variant::Kind::Timely(timely.into()),
+                LogVariant::Differential(differential) => {
+                    proto_log_variant::Kind::Differential(differential.into())
+                }
+                LogVariant::Materialized(materialize) => {
+                    proto_log_variant::Kind::Materialized(materialize.into())
+                }
+            }),
+        }
+    }
+}
+
+impl TryFrom<ProtoLogVariant> for LogVariant {
+    type Error = TryFromProtoError;
+
+    fn try_from(x: ProtoLogVariant) -> Result<Self, Self::Error> {
+        match x.kind {
+            Some(proto_log_variant::Kind::Timely(timely)) => {
+                Ok(LogVariant::Timely(timely.try_into()?))
+            }
+            Some(proto_log_variant::Kind::Differential(differential)) => {
+                Ok(LogVariant::Differential(differential.try_into()?))
+            }
+            Some(proto_log_variant::Kind::Materialized(materialized)) => {
+                Ok(LogVariant::Materialized(materialized.try_into()?))
+            }
+            None => Err(TryFromProtoError::missing_field("ProtoLogVariant::kind")),
+        }
+    }
+}
+
+#[derive(Arbitrary, Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
 pub enum TimelyLog {
     Operates,
     Channels,
@@ -49,14 +136,82 @@ pub enum TimelyLog {
     Reachability,
 }
 
-#[derive(Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
+impl From<&TimelyLog> for ProtoTimelyLog {
+    fn from(x: &TimelyLog) -> Self {
+        use proto_timely_log::Kind::*;
+        ProtoTimelyLog {
+            kind: Some(match x {
+                TimelyLog::Operates => Operates(()),
+                TimelyLog::Channels => Channels(()),
+                TimelyLog::Elapsed => Elapsed(()),
+                TimelyLog::Histogram => Histogram(()),
+                TimelyLog::Addresses => Addresses(()),
+                TimelyLog::Parks => Parks(()),
+                TimelyLog::MessagesSent => MessagesSent(()),
+                TimelyLog::MessagesReceived => MessagesReceived(()),
+                TimelyLog::Reachability => Reachability(()),
+            }),
+        }
+    }
+}
+
+impl TryFrom<ProtoTimelyLog> for TimelyLog {
+    type Error = TryFromProtoError;
+
+    fn try_from(x: ProtoTimelyLog) -> Result<Self, Self::Error> {
+        use proto_timely_log::Kind::*;
+        match x.kind {
+            Some(Operates(())) => Ok(TimelyLog::Operates),
+            Some(Channels(())) => Ok(TimelyLog::Channels),
+            Some(Elapsed(())) => Ok(TimelyLog::Elapsed),
+            Some(Histogram(())) => Ok(TimelyLog::Histogram),
+            Some(Addresses(())) => Ok(TimelyLog::Addresses),
+            Some(Parks(())) => Ok(TimelyLog::Parks),
+            Some(MessagesSent(())) => Ok(TimelyLog::MessagesSent),
+            Some(MessagesReceived(())) => Ok(TimelyLog::MessagesReceived),
+            Some(Reachability(())) => Ok(TimelyLog::Reachability),
+            None => Err(TryFromProtoError::missing_field("ProtoTimelyLog::kind")),
+        }
+    }
+}
+
+#[derive(Arbitrary, Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
 pub enum DifferentialLog {
     ArrangementBatches,
     ArrangementRecords,
     Sharing,
 }
 
-#[derive(Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
+impl From<&DifferentialLog> for ProtoDifferentialLog {
+    fn from(x: &DifferentialLog) -> Self {
+        use proto_differential_log::Kind::*;
+        ProtoDifferentialLog {
+            kind: Some(match x {
+                DifferentialLog::ArrangementBatches => ArrangementBatches(()),
+                DifferentialLog::ArrangementRecords => ArrangementRecords(()),
+                DifferentialLog::Sharing => Sharing(()),
+            }),
+        }
+    }
+}
+
+impl TryFrom<ProtoDifferentialLog> for DifferentialLog {
+    type Error = TryFromProtoError;
+
+    fn try_from(x: ProtoDifferentialLog) -> Result<Self, Self::Error> {
+        use proto_differential_log::Kind::*;
+        match x.kind {
+            Some(ArrangementBatches(())) => Ok(DifferentialLog::ArrangementBatches),
+            Some(ArrangementRecords(())) => Ok(DifferentialLog::ArrangementRecords),
+            Some(Sharing(())) => Ok(DifferentialLog::Sharing),
+            None => Err(TryFromProtoError::missing_field(
+                "ProtoDifferentialLog::kind",
+            )),
+        }
+    }
+}
+
+#[derive(Arbitrary, Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
 pub enum MaterializedLog {
     DataflowCurrent,
     DataflowDependency,
@@ -65,6 +220,39 @@ pub enum MaterializedLog {
     PeekDuration,
 }
 
+impl From<&MaterializedLog> for ProtoMaterializedLog {
+    fn from(x: &MaterializedLog) -> Self {
+        use proto_materialized_log::Kind::*;
+        ProtoMaterializedLog {
+            kind: Some(match x {
+                MaterializedLog::DataflowCurrent => DataflowCurrent(()),
+                MaterializedLog::DataflowDependency => DataflowDependency(()),
+                MaterializedLog::FrontierCurrent => FrontierCurrent(()),
+                MaterializedLog::PeekCurrent => PeekCurrent(()),
+                MaterializedLog::PeekDuration => PeekDuration(()),
+            }),
+        }
+    }
+}
+
+impl TryFrom<ProtoMaterializedLog> for MaterializedLog {
+    type Error = TryFromProtoError;
+
+    fn try_from(x: ProtoMaterializedLog) -> Result<Self, Self::Error> {
+        use proto_materialized_log::Kind::*;
+        match x.kind {
+            Some(DataflowCurrent(())) => Ok(MaterializedLog::DataflowCurrent),
+            Some(DataflowDependency(())) => Ok(MaterializedLog::DataflowDependency),
+            Some(FrontierCurrent(())) => Ok(MaterializedLog::FrontierCurrent),
+            Some(PeekCurrent(())) => Ok(MaterializedLog::PeekCurrent),
+            Some(PeekDuration(())) => Ok(MaterializedLog::PeekDuration),
+            None => Err(TryFromProtoError::missing_field(
+                "ProtoMaterializedLog::kind",
+            )),
+        }
+    }
+}
+
 impl LogVariant {
     /// By which columns should the logs be indexed.
     ///
@@ -232,3 +420,19 @@ impl LogVariant {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mz_repr::proto::protobuf_roundtrip;
+    use proptest::prelude::*;
+
+    proptest! {
+        #[test]
+        fn logging_config_protobuf_roundtrip(expect in any::<LoggingConfig>()) {
+            let actual = protobuf_roundtrip::<_, ProtoLoggingConfig>(&expect);
+            assert!(actual.is_ok());
+            assert_eq!(actual.unwrap(), expect);
+        }
+    }
+}

src/repr/build.rs

@@ -13,6 +13,7 @@ fn main() {
             &[
                 "repr/src/chrono.proto",
                 "repr/src/global_id.proto",
+                "repr/src/proto.proto",
                 "repr/src/row.proto",
                 "repr/src/strconv.proto",
                 "repr/src/relation_and_scalar.proto",

src/repr/src/proto.proto

@@ -0,0 +1,17 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+syntax = "proto3";
+
+package mz_repr.proto;
+
+message ProtoU128{
+    uint64 hi = 1;
+    uint64 lo = 2;
+}