feat!(watch): remove support for WATCH (#245)

Author: loyd

.docker/clickhouse/single_node/config.xml

@@ -14,6 +14,7 @@
   <tmp_path>/var/lib/clickhouse/tmp/</tmp_path>
   <user_files_path>/var/lib/clickhouse/user_files/</user_files_path>
   <access_control_path>/var/lib/clickhouse/access/</access_control_path>
+  <keep_alive_timeout>3</keep_alive_timeout>
 
   <logger>
     <level>debug</level>
@@ -30,4 +31,15 @@
     <partition_by>toYYYYMM(event_date)</partition_by>
     <flush_interval_milliseconds>1000</flush_interval_milliseconds>
   </query_log>
+
+  <!-- required after 25.1+ -->
+  <format_schema_path>/var/lib/clickhouse/format_schemas/</format_schema_path>
+  <user_directories>
+    <users_xml>
+      <path>users.xml</path>
+    </users_xml>
+  </user_directories>
+
+  <!-- Avoid SERVER_OVERLOADED running many parallel tests after 25.5+ -->
+  <os_cpu_busy_time_threshold>1000000000000000000</os_cpu_busy_time_threshold>
 </clickhouse>

CHANGELOG.md

@@ -8,6 +8,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased] - ReleaseDate
 
+### Removed
+- **BREAKING** watch: `Client::watch()` API is removed ([#245]).
+- **BREAKING** mock: `watch()` and `watch_only_events()` are removed ([#245]).
+
+[#245]: https://github.com/ClickHouse/clickhouse-rs/pull/245
+
 ## [0.13.3] - 2025-05-29
 ### Added
 - client: added `Client::with_access_token` to support JWT authentication ClickHouse Cloud feature ([#215]).

Cargo.toml

@@ -66,7 +66,6 @@ default = ["lz4"]
 
 test-util = ["hyper/server"]
 inserter = ["dep:quanta"]
-watch = ["dep:sha-1", "dep:serde_json", "serde/derive"]
 uuid = ["dep:uuid"]
 time = ["dep:time"]
 lz4 = ["dep:lz4_flex", "dep:cityhash-rs"]
@@ -117,8 +116,6 @@ futures = "0.3.5"
 futures-channel = "0.3.30"
 static_assertions = "1.1"
 sealed = "0.6"
-sha-1 = { version = "0.10", optional = true }
-serde_json = { version = "1.0.68", optional = true }
 lz4_flex = { version = "0.11.3", default-features = false, features = [
     "std",
 ], optional = true }

README.md

@@ -25,7 +25,6 @@ Official pure Rust typed client for ClickHouse DB.
 * Provides API for selecting.
 * Provides API for inserting.
 * Provides API for infinite transactional (see below) inserting.
-* Provides API for watching live views.
 * Provides mocks for unit testing.
 
 Note: [ch2rs](https://github.com/ClickHouse/ch2rs) is useful to generate a row type from ClickHouse.
@@ -185,43 +184,11 @@ client.query("DROP TABLE IF EXISTS some").execute().await?;
 ```
 
 </details>
-<details>
-<summary>
-
-### Live views
-
-</summary>
-
-Requires the `watch` feature.
-
-```rust,ignore
-let mut cursor = client
-    .watch("SELECT max(no), argMax(name, no) FROM some")
-    .fetch::<Row<'_>>()?;
-
-let (version, row) = cursor.next().await?.unwrap();
-println!("live view updated: version={}, row={:?}", version, row);
-
-// Use `only_events()` to iterate over versions only.
-let mut cursor = client.watch("some_live_view").limit(20).only_events().fetch()?;
-println!("live view updated: version={:?}", cursor.next().await?);
-```
-
-* Use [carefully](https://github.com/ClickHouse/ClickHouse/issues/28309#issuecomment-908666042).
-* This code uses or creates if not exists a temporary live view named `lv_{sha1(query)}` to reuse the same live view by parallel watchers.
-* You can specify a name instead of a query.
-* This API uses `JSONEachRowWithProgress` under the hood because of [the issue](https://github.com/ClickHouse/ClickHouse/issues/22996).
-* Only struct rows can be used. Avoid `fetch::<u64>()` and other without specified names.
-
-</details>
-
-See [examples](https://github.com/ClickHouse/clickhouse-rs/tree/main/examples).
 
 ## Feature Flags
-* `lz4` (enabled by default) — enables `Compression::Lz4`. If enabled, `Compression::Lz4` is used by default for all queries except for `WATCH`.
+* `lz4` (enabled by default) — enables `Compression::Lz4`. If enabled, `Compression::Lz4` is used by default for all queries.
 * `inserter` — enables `client.inserter()`.
 * `test-util` — adds mocks. See [the example](https://github.com/ClickHouse/clickhouse-rs/tree/main/examples/mock.rs). Use it only in `dev-dependencies`.
-* `watch` — enables `client.watch` functionality. See the corresponding section for details.
 * `uuid` — adds `serde::uuid` to work with [uuid](https://docs.rs/uuid) crate.
 * `time` — adds `serde::time` to work with [time](https://docs.rs/time) crate.
 * `chrono` — adds `serde::chrono` to work with [chrono](https://docs.rs/chrono) crate.
@@ -504,7 +471,7 @@ See also the additional examples:
 * [Variant data type](examples/data_types_variant.rs)
 
 ## Mocking
-The crate provides utils for mocking CH server and testing DDL, `SELECT`, `INSERT` and `WATCH` queries.
+The crate provides utils for mocking CH server and testing DDL, `SELECT` and `INSERT` queries.
 
 The functionality can be enabled with the `test-util` feature. Use it **only** in dev-dependencies.
 

docker-compose.yml

@@ -1,7 +1,9 @@
 services:
   clickhouse:
-    image: 'clickhouse/clickhouse-server:${CLICKHOUSE_VERSION-24.10-alpine}'
+    image: 'clickhouse/clickhouse-server:${CLICKHOUSE_VERSION-latest-alpine}'
     container_name: 'clickhouse-rs-clickhouse-server'
+    environment:
+      CLICKHOUSE_SKIP_USER_SETUP: 1
     ports:
       - '8123:8123'
       - '9000:9000'

examples/README.md

@@ -8,7 +8,7 @@ If something is missing, or you found a mistake in one of these examples, please
 
 ### General usage
 
-- [usage.rs](usage.rs) - creating tables, executing other DDLs, inserting the data, and selecting it back. Additionally, it covers `WATCH` queries. Optional cargo features: `inserter`, `watch`.
+- [usage.rs](usage.rs) - creating tables, executing other DDLs, inserting the data, and selecting it back. Optional cargo features: `inserter`.
 - [mock.rs](mock.rs) - writing tests with `mock` feature. Cargo features: requires `test-util`.
 - [inserter.rs](inserter.rs) - using the client-side batching via the `inserter` feature. Cargo features: requires `inserter`.
 - [async_insert.rs](async_insert.rs) - using the server-side batching via the [asynchronous inserts](https://clickhouse.com/docs/en/optimize/asynchronous-inserts) ClickHouse feature
@@ -57,7 +57,7 @@ cargo run --package clickhouse --example async_insert
 If a particular example requires a cargo feature, you could run it as follows:
 
 ```sh
-cargo run --package clickhouse --example usage --features inserter watch
+cargo run --package clickhouse --example usage --features inserter
 ```
 
 Additionally, the individual examples should be runnable via the IDE such as CLion or RustRover.

examples/mock.rs

@@ -26,23 +26,6 @@ async fn make_insert(client: &Client, data: &[SomeRow]) -> Result<()> {
     insert.end().await
 }
 
-#[cfg(feature = "watch")]
-async fn make_watch(client: &Client) -> Result<(u64, SomeRow)> {
-    client
-        .watch("SELECT max(no) no FROM test")
-        .fetch_one::<SomeRow>()
-        .await
-}
-
-#[cfg(feature = "watch")]
-async fn make_watch_only_events(client: &Client) -> Result<u64> {
-    client
-        .watch("SELECT max(no) no FROM test")
-        .only_events()
-        .fetch_one()
-        .await
-}
-
 #[tokio::main]
 async fn main() {
     let mock = test::Mock::new();
@@ -69,23 +52,4 @@ async fn main() {
     make_insert(&client, &list).await.unwrap();
     let rows: Vec<SomeRow> = recording.collect().await;
     assert_eq!(rows, list);
-
-    // How to test WATCH.
-    #[cfg(feature = "watch")]
-    {
-        // Check `CREATE LIVE VIEW` (for `watch(query)` case only).
-        let recording = mock.add(test::handlers::record_ddl());
-        mock.add(test::handlers::watch(list.into_iter().map(|row| (42, row))));
-        let (version, row) = make_watch(&client).await.unwrap();
-        assert!(recording.query().await.contains("CREATE LIVE VIEW"));
-        assert_eq!(version, 42);
-        assert_eq!(row, SomeRow { no: 1 });
-
-        // `EVENTS`.
-        let recording = mock.add(test::handlers::record_ddl());
-        mock.add(test::handlers::watch_only_events(3..5));
-        let version = make_watch_only_events(&client).await.unwrap();
-        assert!(recording.query().await.contains("CREATE LIVE VIEW"));
-        assert_eq!(version, 3);
-    }
 }

examples/usage.rs

@@ -107,39 +107,6 @@ async fn select_count(client: &Client) -> Result<()> {
     Ok(())
 }
 
-#[cfg(feature = "watch")]
-async fn watch(client: &Client) -> Result<()> {
-    let mut cursor = client
-        .watch("SELECT max(no) no, argMax(name, some.no) name FROM some")
-        .fetch::<MyRow<'_>>()?;
-
-    let (version, row) = cursor.next().await?.unwrap();
-    println!("version={version}, row={row:?}");
-
-    let mut insert = client.insert("some")?;
-    let row = MyRow {
-        no: row.no + 1,
-        name: "bar",
-    };
-    insert.write(&row).await?;
-    insert.end().await?;
-
-    let (version, row) = cursor.next().await?.unwrap();
-    println!("version={version}, row={row:?}");
-
-    // Or you can request only events without data.
-    let mut cursor = client
-        // It's possible to specify a view name.
-        .watch("lv_f2ac5347c013c5b9a6c1aab7192dd97c2748daa0")
-        .limit(10)
-        .only_events()
-        .fetch()?;
-
-    println!("{:?}", cursor.next().await);
-
-    Ok(())
-}
-
 #[tokio::main]
 async fn main() -> Result<()> {
     let client = Client::default().with_url("http://localhost:8123");
@@ -153,8 +120,6 @@ async fn main() -> Result<()> {
     fetch_all(&client).await?;
     delete(&client).await?;
     select_count(&client).await?;
-    #[cfg(feature = "watch")]
-    watch(&client).await?;
 
     Ok(())
 }

src/bytes_ext.rs

@@ -29,7 +29,7 @@ impl BytesExt {
         self.cursor = self.bytes.len() - n;
     }
 
-    #[cfg(any(test, feature = "lz4", feature = "watch"))]
+    #[cfg(any(test, feature = "lz4"))]
     #[inline(always)]
     pub(crate) fn advance(&mut self, n: usize) {
         debug_assert!(n <= self.remaining());