Introduce columnar and derive extensively (#608)

Author: frankmcsherry

communication/Cargo.toml

@@ -17,6 +17,7 @@ license = "MIT"
 default = ["getopts"]
 
 [dependencies]
+columnar = "0.1"
 getopts = { version = "0.2.21", optional = true }
 byteorder = "1.5"
 serde = { version = "1.0", features = ["derive"] }

communication/src/logging.rs

@@ -1,9 +1,10 @@
 //! Configuration and events for communication logging.
 
+use columnar::Columnar;
 use serde::{Serialize, Deserialize};
 
 /// Configuration information about a communication thread.
-#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize, Columnar)]
 pub struct CommunicationSetup {
     /// `true` when this is a send thread (or the receive thread).
     pub sender: bool,
@@ -14,7 +15,7 @@ pub struct CommunicationSetup {
 }
 
 /// Various communication events.
-#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize, Columnar)]
 pub enum CommunicationEvent {
     /// An observed message.
     Message(MessageEvent),
@@ -23,7 +24,7 @@ pub enum CommunicationEvent {
 }
 
 /// An observed message.
-#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize, Columnar)]
 pub struct MessageEvent {
     /// `true` for send event, `false` for receive event
     pub is_send: bool,
@@ -32,7 +33,7 @@ pub struct MessageEvent {
 }
 
 /// Starting or stopping communication threads.
-#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize, Columnar)]
 pub struct StateEvent {
     /// Is the thread a send (vs a recv) thread.
     pub send: bool,

communication/src/networking.rs

@@ -9,6 +9,7 @@ use std::thread::sleep;
 use std::time::Duration;
 
 use byteorder::{ReadBytesExt, WriteBytesExt};
+use columnar::Columnar;
 use serde::{Deserialize, Serialize};
 
 // This constant is sent along immediately after establishing a TCP stream, so
@@ -22,7 +23,7 @@ type ByteOrder = byteorder::BigEndian;
 /// Framing data for each `Vec<u8>` transmission, indicating a typed channel, the source and
 /// destination workers, and the length in bytes.
 // *Warning*: Adding, removing and altering fields requires to adjust the implementation below!
-#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize, Columnar)]
 pub struct MessageHeader {
     /// index of channel.
     pub channel:    usize,

timely/Cargo.toml

@@ -19,6 +19,7 @@ default = ["getopts"]
 getopts = ["getopts-dep", "timely_communication/getopts"]
 
 [dependencies]
+columnar = "0.1"
 getopts-dep = { package = "getopts", version = "0.2.21", optional = true }
 bincode = { version = "1.0" }
 byteorder = "1.5"
@@ -33,5 +34,3 @@ smallvec = { version = "1.13.2", features = ["serde", "const_generics"] }
 [dev-dependencies]
 bytemuck = "1.18.0"
 rand = { version = "0.8", features = ["small_rng"] }
-columnar = "0.1"
-# columnar = { git = "https://github.com/frankmcsherry/columnar" }

timely/src/dataflow/channels/mod.rs

@@ -12,7 +12,7 @@ pub mod pullers;
 pub mod pact;
 
 /// A serializable representation of timestamped data.
-#[derive(Clone, Serialize, Deserialize)]
+#[derive(Clone)]
 pub struct Message<T, C> {
     /// The timestamp associated with the message.
     pub time: T,

timely/src/dataflow/operators/core/capture/event.rs

@@ -4,10 +4,11 @@
 //! stream. There are two types of events, (i) the receipt of data and (ii) reports of progress
 //! of timestamps.
 
+use columnar::Columnar;
 use serde::{Deserialize, Serialize};
 
 /// Data and progress events of the captured stream.
-#[derive(Debug, Clone, Hash, Ord, PartialOrd, Eq, PartialEq, Deserialize, Serialize)]
+#[derive(Debug, Clone, Hash, Ord, PartialOrd, Eq, PartialEq, Deserialize, Serialize, Columnar)]
 pub enum Event<T, C> {
     /// Progress received via `push_external_progress`.
     Progress(Vec<(T, i64)>),

timely/src/logging.rs

@@ -10,6 +10,7 @@ pub type TimelyLogger = Logger<TimelyEvent>;
 pub type TimelyProgressLogger = Logger<TimelyProgressEvent>;
 
 use std::time::Duration;
+use columnar::Columnar;
 use serde::{Deserialize, Serialize};
 use crate::dataflow::operators::capture::{Event, EventPusher};
 
@@ -49,7 +50,7 @@ impl<T, E, P> Drop for BatchLogger<T, E, P> where P: EventPusher<Duration, Vec<(
     }
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// The creation of an `Operate` implementor.
 pub struct OperatesEvent {
     /// Worker-unique identifier for the operator.
@@ -60,7 +61,7 @@ pub struct OperatesEvent {
     pub name: String,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// The creation of a channel between operators.
 pub struct ChannelsEvent {
     /// Worker-unique identifier for the channel
@@ -146,14 +147,14 @@ pub struct TimelyProgressEvent {
     pub internal: Box<dyn ProgressEventTimestampVec>,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// External progress pushed onto an operator
 pub struct PushProgressEvent {
     /// Worker-unique operator identifier
     pub op_id: usize,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Message send or receive event
 pub struct MessagesEvent {
     /// `true` if send event, `false` if receive event.
@@ -171,15 +172,15 @@ pub struct MessagesEvent {
 }
 
 /// Records the starting and stopping of an operator.
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, PartialEq, Eq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, PartialEq, Eq, Ord, PartialOrd)]
 pub enum StartStop {
     /// Operator starts.
     Start,
     /// Operator stops.
     Stop,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Operator start or stop.
 pub struct ScheduleEvent {
     /// Worker-unique identifier for the operator, linkable to the identifiers in [`OperatesEvent`].
@@ -197,14 +198,14 @@ impl ScheduleEvent {
     pub fn stop(id: usize) -> Self { ScheduleEvent { id, start_stop: StartStop::Stop } }
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Operator shutdown.
 pub struct ShutdownEvent {
     /// Worker-unique identifier for the operator, linkable to the identifiers in [`OperatesEvent`].
     pub id: usize,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Application-defined code start or stop
 pub struct ApplicationEvent {
     /// Unique event type identifier
@@ -213,28 +214,28 @@ pub struct ApplicationEvent {
     pub is_start: bool,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Application-defined code start or stop
 pub struct GuardedMessageEvent {
     /// `true` when activity begins, `false` when it stops
     pub is_start: bool,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Application-defined code start or stop
 pub struct GuardedProgressEvent {
     /// `true` when activity begins, `false` when it stops
     pub is_start: bool,
 }
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Hash, Clone, Copy)]
+#[derive(Serialize, Deserialize, Columnar, Debug, PartialEq, Eq, Hash, Clone, Copy)]
 /// Identifier of the worker that generated a log line
 pub struct TimelySetup {
     /// Worker index
     pub index: usize,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Kind of communication channel
 pub enum CommChannelKind {
     /// Communication channel carrying progress information
@@ -243,7 +244,7 @@ pub enum CommChannelKind {
     Data,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Event on a communication channel
 pub struct CommChannelsEvent {
     /// Communication channel identifier
@@ -252,15 +253,15 @@ pub struct CommChannelsEvent {
     pub kind: CommChannelKind,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// Input logic start/stop
 pub struct InputEvent {
     /// True when activity begins, false when it stops
     pub start_stop: StartStop,
 }
 
 /// Records the starting and stopping of an operator.
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, PartialEq, Eq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, PartialEq, Eq, Ord, PartialOrd)]
 pub enum ParkEvent {
     /// Worker parks.
     Park(Option<Duration>),
@@ -275,7 +276,7 @@ impl ParkEvent {
     pub fn unpark() -> Self { ParkEvent::Unpark }
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// An event in a timely worker
 pub enum TimelyEvent {
     /// Operator creation.

timely/src/order.rs

@@ -61,6 +61,7 @@ pub use product::flatcontainer::ProductRegion as FlatProductRegion;
 /// A pair of timestamps, partially ordered by the product order.
 mod product {
     use std::fmt::{Formatter, Error, Debug};
+    use columnar::Columnar;
     use serde::{Deserialize, Serialize};
 
     use crate::container::columnation::{Columnation, Region};
@@ -73,7 +74,7 @@ mod product {
     ///
     /// We use `Product` rather than `(TOuter, TInner)` so that we can derive our own `PartialOrder`,
     /// because Rust just uses the lexicographic total order.
-    #[derive(Copy, Clone, Hash, Eq, PartialEq, Default, Ord, PartialOrd, Serialize, Deserialize)]
+    #[derive(Copy, Clone, Hash, Eq, PartialEq, Default, Ord, PartialOrd, Serialize, Deserialize, Columnar)]
     pub struct Product<TOuter, TInner> {
         /// Outer timestamp.
         pub outer: TOuter,

timely/src/progress/mod.rs

@@ -1,5 +1,6 @@
 //! Progress tracking mechanisms to support notification in timely dataflow
 
+use columnar::Columnar;
 use serde::{Deserialize, Serialize};
 pub use self::operate::Operate;
 pub use self::subgraph::{Subgraph, SubgraphBuilder};
@@ -16,7 +17,7 @@ pub mod reachability;
 pub mod subgraph;
 
 /// A timely dataflow location.
-#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Serialize, Deserialize)]
+#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Serialize, Deserialize, Columnar)]
 pub struct Location {
     /// A scope-local operator identifier.
     pub node: usize,
@@ -58,7 +59,7 @@ impl From<Source> for Location {
 }
 
 /// An operator port.
-#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Serialize, Deserialize)]
+#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Serialize, Deserialize, Columnar)]
 pub enum Port {
     /// An operator input.
     Target(usize),