feat: DynamicInsert -- runtime RowBinary from serde_json::Value

Cargo.toml

@@ -92,12 +92,17 @@ default = ["lz4"]
 
 test-util = ["hyper/server"]
 inserter = ["dep:quanta"]
+async-inserter = ["inserter", "tokio/time", "tokio/sync"]
+batcher = ["async-inserter"]
 uuid = ["dep:uuid"]
 time = ["dep:time"]
 lz4 = ["dep:lz4_flex", "dep:cityhash-rs"]
 chrono = ["dep:chrono"]
 futures03 = []
 
+## Native TCP protocol transport
+native-transport = ["dep:cityhash-rs", "dep:lz4_flex", "dep:zstd", "dep:socket2", "dep:deadpool", "tokio/net", "tokio/io-util"]
+
 ##  TLS
 native-tls = ["dep:hyper-tls"]
 # ext: native-tls-alpn
@@ -141,18 +146,20 @@ hyper-rustls = { version = "0.27.3", default-features = false, features = [
 url = "2.1.1"
 futures-util = { version = "0.3.5", default-features = false, features = ["sink", "io"] }
 futures-channel = { version = "0.3.30", features = ["sink"] }
-lz4_flex = { version = "0.11.3", default-features = false, features = [
+lz4_flex = { version = "0.11.6", default-features = false, features = [
     "std",
 ], optional = true }
 cityhash-rs = { version = "=1.0.1", optional = true } # exact version for safety, this package has been stable for years
+zstd = { version = "0.13", optional = true }
+socket2 = { version = "0.6", features = ["all"], optional = true }
 uuid = { version = "1", optional = true }
 time = { version = "0.3", optional = true }
 chrono = { version = "0.4", optional = true, features = ["serde"] }
 bstr = { version = "1.11.0", default-features = false }
 quanta = { version = "0.12", optional = true }
-polonius-the-crab = "0.5.0"
-
 bnum = "0.13.0"
+deadpool = { version = "0.12", features = ["managed", "rt_tokio_1"], optional = true }
+serde_json = "1"
 
 [dev-dependencies]
 clickhouse-macros = { version = "0.3.0", path = "macros" }
@@ -161,8 +168,7 @@ serde = { version = "1.0.106", features = ["derive"] }
 tokio = { version = "1.0.1", features = ["full", "test-util", "io-util"] }
 hyper = { version = "1.1", features = ["server"] }
 indexmap = { version = "2.10.0", features = ["serde"] }
-linked-hash-map = { version = "0.5.6", features = ["serde_impl"] }
-fxhash = { version = "0.2.1" }
+rustc-hash = "2"
 serde_bytes = "0.11.4"
 serde_json = "1"
 serde_repr = "0.1.7"

README.md

@@ -602,6 +602,42 @@ The functionality can be enabled with the `test-util` feature. Use it **only** i
 
 See [the example](https://github.com/ClickHouse/clickhouse-rs/tree/main/examples/mock.rs).
 
+## Native TCP Transport (HyperI Fork)
+
+This fork adds a native TCP protocol client (`feature = "native-transport"`)
+that connects on port 9000 — the same binary protocol used by `clickhouse-client`
+and the Go client.
+
+```rust
+use clickhouse::native::NativeClient;
+
+let client = NativeClient::default()
+    .with_addr("localhost:9000")
+    .with_database("default")
+    .with_lz4();
+```
+
+The same `#[derive(Row)]` structs work with both HTTP and native transports.
+
+### Additional feature flags (HyperI)
+
+| Feature | Description |
+|---|---|
+| `native-transport` | Native TCP client with connection pooling, SELECT + INSERT |
+| `async-inserter` | `AsyncInserter<T>` — concurrent MPSC-based inserter (HTTP + native) |
+| `batcher` | `TableBatcher<T>` — Go-style `append`/`flush`/`send` wrapper |
+
+### Extended documentation
+
+| Guide | Description |
+|---|---|
+| [Native Transport](docs/native-transport.md) | Connect, query, and insert over TCP |
+| [Connection Pooling](docs/connection-pooling.md) | Deadpool pool, health checks, recycling |
+| [Batching](docs/batching.md) | AsyncInserter, AsyncNativeInserter, TableBatcher |
+| [Types](docs/types.md) | Full type coverage matrix (HTTP + native) |
+| [Wire Format](docs/wire-format.md) | LowCardinality, Dynamic, Variant encoding internals |
+| [Migration](docs/migration.md) | HTTP vs native trade-offs, switching guide |
+
 ## Support Policies
 
 ### Minimum Supported Rust Version (MSRV)

docs/batching.md

@@ -0,0 +1,166 @@
+# Batching and Concurrent Inserters
+
+This crate provides a hierarchy of inserter types for different concurrency and
+transport needs. All share the same three-threshold flush policy: **row count**,
+**byte size**, and **time period**.
+
+## Inserter hierarchy
+
+```mermaid
+graph TD
+    subgraph HTTP Transport
+        TB["TableBatcher&lt;T&gt;<br/><i>feature = batcher</i><br/>Go-style append/flush/send"]
+        AI["AsyncInserter&lt;T&gt;<br/><i>feature = async-inserter</i><br/>MPSC channel + background task"]
+        I["Inserter&lt;T&gt;<br/><i>feature = inserter</i><br/>Single-owner &amp;mut self"]
+        TB -- delegates to --> AI
+        AI -- wraps --> I
+    end
+
+    subgraph Native TCP Transport
+        ANI["AsyncNativeInserter&lt;T&gt;<br/><i>feature = native-transport</i><br/>MPSC channel + background task"]
+        NI["NativeInserter&lt;T&gt;<br/><i>feature = native-transport</i><br/>Single-owner &amp;mut self"]
+        ANI -- wraps --> NI
+    end
+```
+
+## Choosing an inserter
+
+| Type | Transport | Concurrency | Use case |
+|---|---|---|---|
+| `Insert<T>` / `NativeInsert<T>` | HTTP / Native | Single owner | One-shot batch, manual control |
+| `Inserter<T>` / `NativeInserter<T>` | HTTP / Native | Single owner (`&mut self`) | Long-running pipeline, single task |
+| `AsyncInserter<T>` | HTTP | Multi-task (`&self` + handles) | Fan-in from many producers |
+| `AsyncNativeInserter<T>` | Native | Multi-task (`&self` + handles) | Fan-in from many producers |
+| `TableBatcher<T>` | HTTP | Multi-task (Go-style API) | Drop-in replacement for Go `Batch` |
+
+## AsyncInserter / AsyncNativeInserter
+
+Both share identical architecture — an MPSC channel feeding a background tokio
+task that owns the underlying `Inserter` or `NativeInserter`:
+
+```mermaid
+graph TD
+    A["Task A<br/>tx.send()"] --> CH{{"bounded mpsc channel<br/>(default: 8192 slots)"}}
+    B["Task B<br/>tx.send()"] --> CH
+    C["Task C<br/>tx.send()"] --> CH
+    CH --> BG["Background Task<br/><br/>select! {<br/>&nbsp;&nbsp;cmd = rx.recv() &nbsp; ← biased<br/>&nbsp;&nbsp;_ = interval.tick() ← periodic flush<br/>}<br/><br/>serialize → buffer<br/>check limits → flush"]
+    BG --> CK[("ClickHouse server")]
+```
+
+### Key properties
+
+- **`&self` on `write()` and `flush()`** — safe to call from multiple tasks
+  without external synchronization.
+- **Backpressure** — the bounded channel blocks producers when the background
+  task can't keep up. Tune with `with_channel_capacity()`.
+- **Error propagation** — each `write()` returns a `Result<()>` via a oneshot
+  channel. Serialization or network errors are surfaced to the caller.
+- **`RowOwned` requirement** — rows cross a channel boundary, so `T` must own
+  its data (no borrowed `&str` fields). This is automatic for `#[derive(Row)]`
+  structs with owned fields.
+
+### Configuration
+
+```rust
+use clickhouse::async_inserter::{AsyncInserter, AsyncInserterConfig};
+
+let config = AsyncInserterConfig::default()
+    .with_max_rows(100_000)        // flush at 100K rows
+    .with_max_bytes(10_485_760)    // flush at 10 MiB
+    .with_max_period(Duration::from_secs(5))  // flush every 5s
+    .with_channel_capacity(4096);  // backpressure at 4K pending
+
+let inserter = AsyncInserter::<MyRow>::new(&client, "my_table", config);
+```
+
+### Handles
+
+Handles are cheap clones of the channel sender. Use them to fan out writes
+across tasks:
+
+```rust
+let inserter = AsyncInserter::<MyRow>::new(&client, "my_table", config);
+
+for i in 0..10 {
+    let h = inserter.handle();
+    tokio::spawn(async move {
+        h.write(MyRow { id: i, data: format!("task-{i}") }).await.unwrap();
+    });
+}
+
+let stats = inserter.end().await?;  // graceful shutdown
+```
+
+### Lifecycle
+
+1. **`new()`** — spawns background task immediately.
+2. **`write(row)`** — sends row over channel; blocks if full.
+3. **`flush()`** — forces immediate flush of buffered rows.
+4. **`end()`** — sends shutdown command, waits for final flush, joins task.
+
+Dropping without `end()` causes the background task to flush and exit when all
+senders are dropped (including handles).
+
+## TableBatcher
+
+`TableBatcher<T>` is a thin wrapper over `AsyncInserter<T>` with Go client
+naming conventions:
+
+| TableBatcher | AsyncInserter | Go `Batch` |
+|---|---|---|
+| `append(row)` | `write(row)` | `Append(args...)` |
+| `flush()` | `flush()` | `Flush()` |
+| `send()` | `end()` | `Send()` |
+
+```rust
+use clickhouse::batcher::{TableBatcher, BatchConfig};
+
+let config = BatchConfig {
+    max_rows: 100_000,
+    max_bytes: 10 * 1024 * 1024,
+    max_period: Some(Duration::from_secs(5)),
+};
+
+let batcher = TableBatcher::<MyRow>::new(&client, "my_table", config);
+batcher.append(MyRow { id: 1, data: "foo".into() }).await?;
+batcher.append(MyRow { id: 2, data: "bar".into() }).await?;
+
+let stats = batcher.send().await?;  // final flush + shutdown
+```
+
+## Default thresholds
+
+All inserter configs share these defaults, aligned with ClickHouse server
+settings and the Go client:
+
+| Setting | Default | Rationale |
+|---|---|---|
+| `max_rows` | 100,000 | Upper end of recommended per-INSERT batch size |
+| `max_bytes` | 10 MiB | Matches `async_insert_max_data_size` server default |
+| `max_period` | 5 seconds | Balances latency vs throughput |
+| `channel_capacity` | 8,192 | Backpressure threshold for concurrent inserters |
+
+### Why client-side batching?
+
+ClickHouse's server-side `async_insert` is convenient for distributed agents but
+has drawbacks for high-throughput pipelines:
+
+- **OOM risk** — server buffers data in memory until thresholds are met
+- **Delayed visibility** — data is not queryable until the server flushes
+- **Error opacity** — insert errors may be lost or delayed
+- **No client control** — can't tune batch size per table or per source
+
+Client-side batching (as implemented here) gives you immediate query visibility,
+per-table tuning, and explicit error handling at the cost of managing batch
+state in the client.
+
+### MergeTree part fragmentation
+
+Each INSERT creates one part per partition in MergeTree. Too many small INSERTs
+cause part fragmentation:
+
+- `parts_to_delay_insert` (default: 150) — slows down inserts
+- `parts_to_throw_insert` (default: 300) — hard "Too many parts" error
+
+The defaults here (100K rows, 10 MiB) are designed to produce reasonably-sized
+parts. For tables with multiple partitions, adjust downward.

docs/connection-pooling.md

@@ -0,0 +1,95 @@
+# Connection Pooling
+
+The native transport uses [deadpool](https://docs.rs/deadpool) for connection
+pooling. Each `NativeClient` owns a pool; clones share it (the pool is
+`Arc`-backed internally).
+
+## Configuration
+
+```rust
+use clickhouse::native::NativeClient;
+
+let client = NativeClient::default()
+    .with_pool_size(20);   // max 20 connections (default: 10)
+```
+
+The pool is bounded — when all connections are in use, `acquire()` waits until
+one is returned. There is no idle timeout; connections persist until the client
+is dropped or they fail a health check.
+
+## Health checks (recycle)
+
+When a connection is returned to the pool, deadpool calls `recycle()` which
+runs `check_alive()`:
+
+1. **Poisoned flag** — if `discard()` was called, the connection is dropped
+   unconditionally.
+2. **Buffered data** — if the `BufReader` has leftover bytes from an incomplete
+   read, the connection is dropped (stale protocol state).
+3. **Non-blocking poll** — a non-blocking `poll_read` detects EOF or unexpected
+   server data. If either is found, the connection is dropped.
+
+Connections that pass all three checks are returned to the idle queue for reuse.
+
+## The discard pattern
+
+When an I/O error or incomplete protocol exchange leaves a connection in an
+unrecoverable state, call `discard()` on the `PooledConnection`. This sets a
+`poisoned` flag that causes `recycle()` to drop the connection rather than
+returning it to the pool.
+
+This is used internally by:
+- `NativeRowCursor::Drop` — if a cursor is dropped mid-stream (e.g. after
+  `fetch_one` without draining), the connection is discarded.
+- Error paths in `NativeInsert` — if an INSERT fails mid-stream, the
+  connection is discarded rather than risk protocol desync.
+
+## Cursor drain
+
+`NativeRowCursor` provides a `drain()` method that reads and discards all
+remaining server packets until `EndOfStream`. This is called automatically by
+`fetch_one` and `fetch_optional` to clean up the connection before returning it
+to the pool:
+
+```mermaid
+graph LR
+    F["fetch_one()"] --> N["next().await?<br/>get first row"]
+    N --> D["drain().await?<br/>consume remaining packets"]
+    D --> R["return row<br/>connection returned to pool"]
+```
+
+If `drain()` is not called (e.g. cursor dropped early), the `Drop` impl
+discards the connection as a safety net.
+
+## Pool rebuild on config change
+
+Builder methods that affect connection parameters trigger `rebuild_pool()`,
+which creates a new pool instance. Existing connections from the old pool are
+not immediately closed — they drain naturally as they're returned and not
+recycled into the new pool.
+
+Affected methods:
+- `with_addr()`
+- `with_database()`
+- `with_user()` / `with_password()`
+- `with_lz4()`
+- `with_pool_size()`
+- `with_setting()`
+
+## Architecture
+
+```mermaid
+graph TD
+    NC["NativeClient"] --> POOL["pool: NativePool<br/>(deadpool::managed::Pool)"]
+    NC --> SC["schema_cache: Arc&lt;NativeSchemaCache&gt;<br/>HashMap&lt;table, (columns, expires_at)&gt;"]
+    NC --> SET["settings: Arc&lt;Vec&lt;(key, value)&gt;&gt;"]
+
+    POOL --> MGR["NativeConnectionManager"]
+    MGR -->|"create()"| OPEN["NativeConnection::open()"]
+    MGR -->|"recycle()"| CHECK["check_alive()"]
+
+    POOL --> IDLE["Idle queue<br/>(bounded semaphore)"]
+    IDLE --> C1["conn 1"]
+    IDLE --> C2["conn 2"]
+    IDLE --> C3["..."]
+```