WIP columnar in logging

src/compute/Cargo.toml

@@ -12,19 +12,16 @@ workspace = true
 [dependencies]
 anyhow = "1.0.66"
 async-stream = "0.3.3"
-async-trait = "0.1.68"
 bytesize = "1.1.0"
-clap = { version = "3.2.24", features = ["derive", "env"] }
+columnar = "0.2.0"
 crossbeam-channel = "0.5.8"
 dec = { version = "0.4.8", features = ["serde"] }
 differential-dataflow = "0.13.1"
 differential-dogs3 = "0.1.1"
 futures = "0.3.25"
 itertools = "0.10.5"
 lgalloc = "0.4"
-mz-build-info = { path = "../build-info" }
 mz-cluster = { path = "../cluster" }
-mz-cluster-client = { path = "../cluster-client" }
 mz-compute-client = { path = "../compute-client" }
 mz-compute-types = { path = "../compute-types" }
 mz-dyncfg = { path = "../dyncfg" }
@@ -34,7 +31,6 @@ mz-ore = { path = "../ore", features = ["async", "flatcontainer", "process", "tr
 mz-persist-client = { path = "../persist-client" }
 mz-persist-types = { path = "../persist-types" }
 mz-repr = { path = "../repr" }
-mz-service = { path = "../service" }
 mz-storage-operators = { path = "../storage-operators" }
 mz-storage-types = { path = "../storage-types" }
 mz-timely-util = { path = "../timely-util" }

src/compute/src/logging/timely.rs

@@ -14,12 +14,12 @@ use std::collections::BTreeMap;
 use std::rc::Rc;
 use std::time::Duration;
 
+use columnar::Columnar;
 use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use mz_compute_client::logging::LoggingConfig;
 use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, Timestamp};
 use mz_timely_util::replay::MzReplay;
-use serde::{Deserialize, Serialize};
 use timely::communication::Allocate;
 use timely::container::columnation::{Columnation, CopyRegion};
 use timely::container::CapacityContainerBuilder;
@@ -325,6 +325,7 @@ struct DemuxState {
     schedules_data: BTreeMap<usize, Vec<(isize, i64)>>,
 }
 
+#[derive(Columnar)]
 struct Park {
     /// Time when the park occurred.
     time: Duration,
@@ -333,7 +334,7 @@ struct Park {
 }
 
 /// Organize message counts into number of batches and records.
-#[derive(Default, Copy, Clone, Debug)]
+#[derive(Default, Copy, Clone, Debug, Columnar)]
 struct MessageCount {
     /// The number of batches sent across a channel.
     batches: i64,
@@ -364,7 +365,7 @@ struct DemuxOutput<'a> {
     schedules_histogram: OutputSession<'a, (ScheduleHistogramDatum, ())>,
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Columnar)]
 struct ChannelDatum {
     id: usize,
     source: (usize, usize),
@@ -375,7 +376,7 @@ impl Columnation for ChannelDatum {
     type InnerRegion = CopyRegion<Self>;
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Columnar)]
 struct ParkDatum {
     duration_pow: u128,
     requested_pow: Option<u128>,
@@ -385,7 +386,7 @@ impl Columnation for ParkDatum {
     type InnerRegion = CopyRegion<Self>;
 }
 
-#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Columnar)]
 struct MessageDatum {
     channel: usize,
     worker: usize,
@@ -395,7 +396,7 @@ impl Columnation for MessageDatum {
     type InnerRegion = CopyRegion<Self>;
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Columnar)]
 struct ScheduleHistogramDatum {
     operator: usize,
     duration_pow: u128,

src/timely-util/src/containers.rs

@@ -9,5 +9,99 @@
 
 //! Reusable containers.
 
+use std::collections::VecDeque;
+use std::marker::PhantomData;
+use timely::container::{ContainerBuilder, PushInto, SizableContainer};
+use timely::Container;
+
 pub mod array;
 pub mod stack;
+
+pub struct ConvertingContainerBuilder<CI, CO, I> {
+    current: CO,
+    outbound: VecDeque<CO>,
+    empty: Vec<CO>,
+    inner: I,
+    _phantom: PhantomData<(CI, CO)>,
+}
+
+impl<CI, CO: Default, I> ConvertingContainerBuilder<CI, CO, I> {
+    #[inline]
+    pub fn new(inner: I) -> Self {
+        Self {
+            current: CO::default(),
+            outbound: VecDeque::new(),
+            empty: Vec::default(),
+            inner,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<CI, CO: Default, I: Default> Default for ConvertingContainerBuilder<CI, CO, I> {
+    #[inline]
+    fn default() -> Self {
+        Self::new(I::default())
+    }
+}
+
+impl<CI, CO, I, D> PushInto<D> for ConvertingContainerBuilder<CI, CO, I>
+where
+    I: PushInto<D>,
+{
+    fn push_into(&mut self, item: D) {
+        self.inner.push_into(item);
+    }
+}
+
+impl<CI, CO, I> ConvertingContainerBuilder<CI, CO, I>
+where
+    CI: Container + 'static,
+    CO: SizableContainer + Clone + for<'a> PushInto<CI::Item<'a>> + 'static,
+    I: ContainerBuilder<Container = CI>,
+{
+    fn process(&mut self, mut logic: impl FnMut(&mut I) -> Option<&mut CI>) {
+        while let Some(container) = logic(&mut self.inner) {
+            for item in container.drain() {
+                if self.current.at_capacity() {
+                    if !self.current.is_empty() {
+                        self.outbound.push_back(std::mem::take(&mut self.current));
+                    }
+                    self.current = CO::default();
+                    self.current.ensure_capacity(&mut self.empty.pop());
+                }
+                self.current.push_into(item);
+            }
+        }
+    }
+}
+
+impl<CI, CO, I> ContainerBuilder for ConvertingContainerBuilder<CI, CO, I>
+where
+    CI: Container + 'static,
+    CO: SizableContainer + Clone + for<'a> PushInto<CI::Item<'a>> + 'static,
+    I: ContainerBuilder<Container = CI>,
+{
+    type Container = CO;
+
+    #[inline]
+    fn extract(&mut self) -> Option<&mut CO> {
+        self.process(|inner| inner.extract());
+        if let Some(container) = self.outbound.pop_front() {
+            self.empty.push(container);
+            self.empty.last_mut()
+        } else {
+            None
+        }
+    }
+
+    #[inline]
+    fn finish(&mut self) -> Option<&mut CO> {
+        self.process(|inner| inner.finish());
+        if !self.current.is_empty() {
+            self.outbound.push_back(std::mem::take(&mut self.current));
+            self.empty.truncate(2);
+        }
+        self.extract()
+    }
+}