1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
//! ClickHouse sink for the Spate framework.
//!
//! Writes **directly to shard-local tables** through the official
//! `clickhouse` crate: rows are encoded to RowBinary on the pipeline
//! threads (this crate's own [serializer](rowbinary)), shipped as
//! pre-formatted frames, and inserted one `INSERT ... FORMAT RowBinary`
//! per sealed batch with a deterministic `insert_deduplication_token`.
//! Direct-to-shard writes beat `Distributed`-table inserts for an ETL
//! writer: bigger blocks, less merge pressure, and — crucially for
//! checkpointing — a synchronous server acknowledgement
//! (`wait_end_of_query=1`; `write_batch` returning `Ok` covers
//! materialized views too).
//!
//! # ⚠ Deduplication requires a window — silently off on plain MergeTree
//!
//! Retries reuse the batch's deduplication token, making them idempotent —
//! **but only if the server keeps a deduplication window**:
//!
//! - `Replicated*MergeTree`: deduplication is on by default
//! (`replicated_deduplication_window = 10000`).
//! - **Plain `MergeTree`: the window defaults to `0` and token
//! deduplication silently does nothing.** Set it explicitly:
//!
//! ```sql
//! CREATE TABLE orders (...) ENGINE = MergeTree ORDER BY id
//! SETTINGS non_replicated_deduplication_window = 100;
//! ```
//!
//! # At-least-once, honestly
//!
//! Tokens cover *same-batch retries* (including on another replica after a
//! timeout). They do **not** cover crash replay: after a restart, data is
//! re-batched with different boundaries and different tokens, and replayed
//! rows will land again. Design target tables to tolerate duplicates —
//! `ReplacingMergeTree` with a version column is the sanctioned pattern.
//!
//! Re-homing is a different hazard entirely: a key that moves to another
//! shard — changed shard weights, a changed sharding key, a changed shard
//! count — leaves its already-written rows behind on the old shard, and
//! nothing collapses them there. Dedup tokens are shard-scoped, and
//! `ReplacingMergeTree` merges within one shard's local table only, so
//! the old copies persist indefinitely: unpruned reads through a
//! `Distributed` table return both homes' rows, while pruned reads return
//! only the new home's. Treat a re-homing change as a planned migration —
//! backfill or delete the moved keys' old-shard rows — not as replay
//! noise that settles. A plain restart with unchanged topology is the
//! benign case: replayed rows land on the *same* shard, where the
//! patterns above apply.
//!
//! # Shard-affinity routing (Distributed parity)
//!
//! [`DistributedRouter`] (built via [`ClickHouseSink::router`]) places each
//! record on the same shard a ClickHouse `Distributed` table with sharding
//! expression `xxHash64(<key column>)` would — so SELECTs through the
//! Distributed table can prune shards (`optimize_skip_unused_shards=1`)
//! while inserts keep going direct-to-local. The ordering contract is the
//! operator's: `shards[i]` in the sink config must be the cluster's
//! `shard_num = i + 1` (the `remote_servers` order), with matching
//! `weight`s. The opt-in `distributed_check` config block verifies shard
//! count, weights, and the DDL's sharding expression at startup
//! ([`ClickHouseSink::validate_distributed`]); the ordering itself it can
//! only warn about. See [`router`] for the algorithm.
//!
//! # Column order is the wire contract
//!
//! RowBinary carries no column names. The configured `columns` list and
//! the row struct's **field declaration order** must match; reordering
//! either is a breaking change to the pipeline. See [`rowbinary`] for the
//! full type mapping.
//!
//! # Wiring
//!
//! ```yaml
//! sink:
//! clickhouse:
//! table: orders_local
//! columns: [id, name, amount]
//! shards:
//! - replicas: ["http://ch-0-0:8123", "http://ch-0-1:8123"]
//! - replicas: ["http://ch-1-0:8123", "http://ch-1-1:8123"]
//! # Only when inserting into JSON columns (as String fields of JSON
//! # text — see the rowbinary type table):
//! settings: { input_format_binary_read_json_as_string: "1" }
//! ```
//!
//! [`config::from_component_config`] turns that section into a
//! [`ClickHouseWriter`] (the framework's `ShardWriter`), per-shard
//! [`ClickHouseEndpoint`]s, and the sink-pool configuration;
//! [`ClickHouseEncoder`] is the matching `RowEncoder`, parameterized by a
//! record family: owned rows use `Owned<Row>`, and the encode path itself
//! needs only `Row: serde::Serialize`.
//!
//! # Wire format
//!
//! The default is row-wise **RowBinary** ([`rowbinary`]). An opt-in columnar
//! **Native** encoder ([`native`], `format: native`) transposes each chunk
//! into one self-describing block: it costs more client CPU but cuts server
//! parse CPU and compressed wire size substantially — build it from a fetched
//! schema via [`ClickHouseSink::native_schema`] and [`NativeEncoder`].
pub use ;
pub use DistributedCheckError;
pub use ;
pub use ;
pub use ;
pub use ;
pub use ;
pub use *;
pub use ;