Merge remote-tracking branch 'origin/main' into rbwnat-insert

Author: slvrtrn

.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,18 +8,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased] - ReleaseDate
 
-### Breaking Changes
+### Removed
 
-- query: `RowBinaryWithNamesAndTypes` is now used by default for query results. This may cause panics if the row struct
-  definition does not match the database schema. Use `Client::with_validation(false)` to revert to the previous behavior
-  which uses plain `RowBinary` format for fetching rows. ([#221])
+- **BREAKING** watch: `Client::watch()` API is removed ([#245]).
+- **BREAKING** mock: `watch()` and `watch_only_events()` are removed ([#245]).
+
+### Changed
+
+- **BREAKING** query: `RowBinaryWithNamesAndTypes` is now used by default for query results. This may cause panics if
+  the row struct definition does not match the database schema. Use `Client::with_validation(false)` to revert to the
+  previous behavior which uses plain `RowBinary` format for fetching rows. ([#221])
+- **BREAKING** mock: when using `test-util` feature, it is now required to use `Client::with_mock(&mock)` to set up the
+  mock server, so it properly handles the response format and automatically disables parsing
+  `RowBinaryWithNamesAndTypes` header parsing and validation. Additionally, it is not required to call `with_url`
+  explicitly. See the [updated example](./examples/mock.rs).
 - query: due to `RowBinaryWithNamesAndTypes` format usage, there might be an impact on fetch performance, which largely
-  depends on how the dataset is defined. If you experience performance issues, consider disabling validation by using
+  depends on how the dataset is defined. If you notice decreased performance, consider disabling validation by using
   `Client::with_validation(false)`.
-- mock: when using `test-util` feature, it is now required to use `Client::with_mock(&mock)` to set up the mock server,
-  so it properly handles the response format and automatically disables parsing `RowBinaryWithNamesAndTypes` header
-  parsing and validation. Additionally, it is not required to call `with_url` explicitly.
-  See the [updated example](./examples/mock.rs).
+- serde: it is now possible to deserialize Map ClickHouse type into `HashMap<K, V>` (or `BTreeMap`, `IndexMap`, 
+  `DashMap`, etc.).
 
 ### Added
 
@@ -31,6 +38,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   functions and utilities to parse the types out of the ClickHouse server response. ([#221]).
 
 [#221]: https://github.com/ClickHouse/clickhouse-rs/pull/221
+[#245]: https://github.com/ClickHouse/clickhouse-rs/pull/245
 
 ## [0.13.3] - 2025-05-29
 ### Added

Cargo.toml

@@ -82,7 +82,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"]
@@ -134,8 +133,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

@@ -1,8 +1,10 @@
 name: clickhouse-rs
 services:
   clickhouse:
-    image: 'clickhouse/clickhouse-server:${CLICKHOUSE_VERSION-24.10-alpine}'
-    container_name: clickhouse-rs-clickhouse-server
+    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'

docker-compose.yml

@@ -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/README.md

@@ -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();
@@ -71,23 +54,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/mock.rs

@@ -29,7 +29,7 @@ async fn ddl(client: &Client) -> Result<()> {
 }
 
 async fn insert(client: &Client) -> Result<()> {
-    let mut insert = client.insert("some").await?;
+    let mut insert = client.insert("some")?;
     for i in 0..1000 {
         insert.write(&MyRow { no: i, name: "foo" }).await?;
     }
@@ -42,12 +42,12 @@ async fn insert(client: &Client) -> Result<()> {
 #[cfg(feature = "inserter")]
 async fn inserter(client: &Client) -> Result<()> {
     let mut inserter = client
-        .inserter("some")
+        .inserter("some")?
         .with_max_rows(100_000)
         .with_period(Some(std::time::Duration::from_secs(15)));
 
     for i in 0..1000 {
-        inserter.write(&MyRow { no: i, name: "foo" }).await?;
+        inserter.write(&MyRow { no: i, name: "foo" })?;
         inserter.commit().await?;
     }
 
@@ -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").await?;
-    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(())
 }

examples/usage.rs

@@ -1,6 +1,2 @@
 edition = "2021"
 merge_derives = false
-imports_granularity = "Crate"
-normalize_comments = true
-reorder_impl_items = true
-wrap_comments = true

rustfmt.toml

@@ -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());