Update to latest Timely

Point Materialize at latest Timely. We need to incorporate some changes
around reachability logging, which is now typed, and event iterators that
return cow'ed data.

Some of the complexity stems from the fact that event links are
single-writer, so we need separate event links for each reachability log
variant.

Signed-off-by: Moritz Hoffmann <mh@materialize.com>

Author: antiguru

Cargo.lock

@@ -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-types/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/adapter/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/Cargo.toml

@@ -5567,15 +5567,22 @@ 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
+    addr2.address = addr1.address || reachability.source
+    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 +5594,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/catalog/src/builtin.rs

@@ -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/cluster/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/Cargo.toml

@@ -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-client/src/logging.rs

@@ -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-types/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"] }