Update to latest Timely

src/adapter-types/Cargo.toml

@@ -15,7 +15,7 @@ mz-ore = { path = "../ore" }
 mz-repr = { path = "../repr" }
 mz-storage-types = { path = "../storage-types" }
 serde = "1.0.152"
-timely = "0.16.0"
+timely = "0.17.0"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
 
 [package.metadata.cargo-udeps.ignore]

src/adapter/Cargo.toml

@@ -18,7 +18,7 @@ bytesize = "1.1.0"
 chrono = { version = "0.4.35", default-features = false, features = ["std"] }
 dec = "0.4.8"
 derivative = "2.2.0"
-differential-dataflow = "0.13.3"
+differential-dataflow = "0.13.4"
 enum-kinds = "0.5.1"
 fail = { version = "0.5.1", features = ["failpoints"] }
 futures = "0.3.25"
@@ -78,7 +78,7 @@ serde_plain = "1.0.1"
 sha2 = "0.10.6"
 smallvec = { version = "1.10.0", features = ["union"] }
 static_assertions = "1.1"
-timely = "0.16.0"
+timely = "0.17.0"
 tokio = { version = "1.38.0", features = ["rt", "time"] }
 tokio-postgres = { version = "0.7.8" }
 tracing = "0.1.37"

src/catalog/Cargo.toml

@@ -18,7 +18,7 @@ bytesize = "1.1.0"
 chrono = { version = "0.4.35", default-features = false, features = ["std"] }
 clap = { version = "4.5.23", features = ["derive"] }
 derivative = "2.2.0"
-differential-dataflow = "0.13.3"
+differential-dataflow = "0.13.4"
 futures = "0.3.25"
 ipnet = "2.5.0"
 itertools = "0.12.1"
@@ -60,7 +60,7 @@ serde_plain = "1.0.1"
 static_assertions = "1.1"
 sha2 = "0.10.6"
 thiserror = "1.0.37"
-timely = "0.16.0"
+timely = "0.17.0"
 tokio = { version = "1.38.0" }
 tracing = "0.1.37"
 uuid = "1.2.2"

src/catalog/src/builtin.rs

@@ -5567,15 +5567,25 @@ pub static MZ_DATAFLOW_OPERATOR_REACHABILITY_PER_WORKER: LazyLock<BuiltinView> =
         oid: oid::VIEW_MZ_DATAFLOW_OPERATOR_REACHABILITY_PER_WORKER_OID,
         column_defs: None,
         sql: "SELECT
-    address,
+    addr2.id,
+    reachability.worker_id,
     port,
-    worker_id,
     update_type,
     time,
     pg_catalog.count(*) as count
 FROM
-    mz_introspection.mz_dataflow_operator_reachability_raw
-GROUP BY address, port, worker_id, update_type, time",
+    mz_introspection.mz_dataflow_operator_reachability_raw reachability,
+    mz_introspection.mz_dataflow_addresses_per_worker addr1,
+    mz_introspection.mz_dataflow_addresses_per_worker addr2
+WHERE
+    CASE
+        WHEN source = 0 THEN addr2.address = addr1.address
+        ELSE addr2.address = addr1.address || reachability.source
+    END
+    AND addr1.id = reachability.id
+    AND addr1.worker_id = reachability.worker_id
+    AND addr2.worker_id = reachability.worker_id
+GROUP BY addr2.id, reachability.worker_id, port, update_type, time",
         access: vec![PUBLIC_SELECT],
     });
 
@@ -5587,13 +5597,13 @@ pub static MZ_DATAFLOW_OPERATOR_REACHABILITY: LazyLock<BuiltinView> =
         column_defs: None,
         sql: "
 SELECT
-    address,
+    id,
     port,
     update_type,
     time,
     pg_catalog.sum(count) as count
 FROM mz_introspection.mz_dataflow_operator_reachability_per_worker
-GROUP BY address, port, update_type, time",
+GROUP BY id, port, update_type, time",
         access: vec![PUBLIC_SELECT],
     });
 

src/cluster/Cargo.toml

@@ -13,15 +13,15 @@ workspace = true
 anyhow = "1.0.66"
 async-trait = "0.1.83"
 crossbeam-channel = "0.5.8"
-differential-dataflow = "0.13.3"
+differential-dataflow = "0.13.4"
 futures = "0.3.25"
 mz-cluster-client = { path = "../cluster-client" }
 mz-ore = { path = "../ore", features = ["async", "process", "tracing_"] }
 mz-persist-client = { path = "../persist-client" }
 mz-service = { path = "../service" }
 mz-txn-wal = { path = "../txn-wal" }
 regex = "1.7.0"
-timely = "0.16.0"
+timely = "0.17.0"
 tokio = { version = "1.38.0", features = ["fs", "rt", "sync", "net"] }
 tracing = "0.1.37"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }

src/compute-client/Cargo.toml

@@ -15,7 +15,7 @@ async-trait = "0.1.83"
 bytesize = "1.1.0"
 crossbeam-channel = "0.5.8"
 derivative = "2.2.0"
-differential-dataflow = "0.13.3"
+differential-dataflow = "0.13.4"
 futures = "0.3.25"
 http = "1.1.0"
 mz-build-info = { path = "../build-info" }
@@ -43,7 +43,7 @@ prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = "1.0.125"
 thiserror = "1.0.37"
-timely = "0.16.0"
+timely = "0.17.0"
 tokio = "1.38.0"
 tokio-stream = "0.1.11"
 tonic = "0.12.1"

src/compute-client/src/logging.rs

@@ -448,16 +448,10 @@ impl LogVariant {
                 .finish(),
 
             LogVariant::Timely(TimelyLog::Reachability) => RelationDesc::builder()
-                .with_column(
-                    "address",
-                    ScalarType::List {
-                        element_type: Box::new(ScalarType::UInt64),
-                        custom_id: None,
-                    }
-                    .nullable(false),
-                )
-                .with_column("port", ScalarType::UInt64.nullable(false))
+                .with_column("id", ScalarType::UInt64.nullable(false))
                 .with_column("worker_id", ScalarType::UInt64.nullable(false))
+                .with_column("source", ScalarType::UInt64.nullable(false))
+                .with_column("port", ScalarType::UInt64.nullable(false))
                 .with_column("update_type", ScalarType::String.nullable(false))
                 .with_column("time", ScalarType::MzTimestamp.nullable(true))
                 .finish(),

src/compute-types/Cargo.toml

@@ -12,7 +12,7 @@ workspace = true
 [dependencies]
 columnar = "0.2.2"
 columnation = "0.1.0"
-differential-dataflow = "0.13.3"
+differential-dataflow = "0.13.4"
 itertools = "0.12.1"
 mz-dyncfg = { path = "../dyncfg" }
 mz-expr = { path = "../expr" }
@@ -24,7 +24,7 @@ proptest = { version = "1.6.0", default-features = false, features = ["std"] }
 proptest-derive = { version = "0.5.1", features = ["boxed_union"] }
 prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 serde = { version = "1.0.152", features = ["derive"] }
-timely = "0.16.0"
+timely = "0.17.0"
 tracing = "0.1.37"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
 

src/compute/Cargo.toml

@@ -16,8 +16,8 @@ bytesize = "1.1.0"
 columnar = "0.2.2"
 crossbeam-channel = "0.5.8"
 dec = { version = "0.4.8", features = ["serde"] }
-differential-dataflow = "0.13.3"
-differential-dogs3 = "0.1.3"
+differential-dataflow = "0.13.4"
+differential-dogs3 = "0.1.4"
 futures = "0.3.25"
 itertools = "0.12.1"
 lgalloc = "0.4"
@@ -39,7 +39,7 @@ prometheus = { version = "0.13.3", default-features = false }
 scopeguard = "1.1.0"
 serde = { version = "1.0.152", features = ["derive"] }
 smallvec = { version = "1.10.0", features = ["serde", "union"] }
-timely = "0.16.0"
+timely = "0.17.0"
 tokio = { version = "1.38.0", features = ["fs", "rt", "sync", "net"] }
 tracing = "0.1.37"
 uuid = { version = "1.7.0", features = ["serde", "v4"] }

src/compute/src/logging.rs

@@ -115,18 +115,23 @@ where
 ///
 /// This is just a bundle-type intended to make passing around its contents in the logging
 /// initialization code more convenient.
+///
+/// The `N` type parameter specifies the number of links to create for the event queue. We need
+/// separate links for queues that feed from multiple loggers because the `EventLink` type is not
+/// multi-producer safe (it is a linked-list, and multiple writers would blindly append, replacing
+/// existing new data, and cutting off other writers).
 #[derive(Clone)]
-struct EventQueue<C> {
-    link: Rc<EventLink<Timestamp, C>>,
+struct EventQueue<C, const N: usize = 1> {
+    links: [Rc<EventLink<Timestamp, C>>; N],
     activator: RcActivator,
 }
 
-impl<C> EventQueue<C> {
+impl<C, const N: usize> EventQueue<C, N> {
     fn new(name: &str) -> Self {
         let activator_name = format!("{name}_activator");
         let activate_after = 128;
         Self {
-            link: Rc::new(EventLink::new()),
+            links: [(); N].map(|_| Rc::new(EventLink::new())),
             activator: RcActivator::new(activator_name, activate_after),
         }
     }

src/compute/src/logging/compute.rs

@@ -308,16 +308,16 @@ pub(super) fn construct<A: Allocate + 'static>(
 
     worker.dataflow_named("Dataflow: compute logging", move |scope| {
         let enable_logging = config.enable_logging;
-        let (logs, token) = Some(event_queue.link).mz_replay::<_, ProvidedBuilder<_>, _>(
+        let (logs, token) = event_queue.links.mz_replay::<_, ProvidedBuilder<_>, _>(
             scope,
             "compute logs",
             config.interval,
             event_queue.activator,
-            move |mut session, data| {
+            move |mut session, mut data| {
                 // If logging is disabled, we still need to install the indexes, but we can leave them
                 // empty. We do so by immediately filtering all logs events.
                 if enable_logging {
-                    session.give_container(&mut data.clone())
+                    session.give_container(data.to_mut())
                 }
             },
         );

src/compute/src/logging/differential.rs

@@ -20,15 +20,15 @@ use differential_dataflow::logging::{
 };
 use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, Timestamp};
-use mz_timely_util::containers::{columnar_exchange, Col2ValBatcher, ColumnBuilder};
+use mz_timely_util::containers::{
+    columnar_exchange, Col2ValBatcher, ColumnBuilder, ProvidedBuilder,
+};
 use mz_timely_util::replay::MzReplay;
 use timely::communication::Allocate;
-use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
 use timely::dataflow::channels::pushers::buffer::Session;
 use timely::dataflow::channels::pushers::{Counter, Tee};
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
-use timely::dataflow::operators::Filter;
 use timely::dataflow::Stream;
 
 use crate::extensions::arrange::MzArrangeCore;
@@ -57,21 +57,22 @@ pub(super) fn construct<A: Allocate>(
     let dataflow_index = worker.next_dataflow_index();
 
     worker.dataflow_named("Dataflow: differential logging", move |scope| {
-        let (mut logs, token) = Some(event_queue.link)
-            .mz_replay::<_, CapacityContainerBuilder<_>, _>(
+        let enable_logging = config.enable_logging;
+        let (logs, token) = event_queue.links
+            .mz_replay::<_, ProvidedBuilder<_>, _>(
                 scope,
                 "differential logs",
                 config.interval,
                 event_queue.activator,
-                |mut session, data| session.give_iterator(data.iter()),
+                move |mut session, mut data|{
+                    // If logging is disabled, we still need to install the indexes, but we can leave them
+                    // empty. We do so by immediately filtering all logs events.
+                    if enable_logging {
+                        session.give_container(data.to_mut())
+                    }
+                }
             );
 
-        // If logging is disabled, we still need to install the indexes, but we can leave them
-        // empty. We do so by immediately filtering all logs events.
-        if !config.enable_logging {
-            logs = logs.filter(|_| false);
-        }
-
         // Build a demux operator that splits the replayed event stream up into the separate
         // logging streams.
         let mut demux =