diff --git a/config.toml.example b/config.toml.example index 6942b1c..2a39cf1 100644 --- a/config.toml.example +++ b/config.toml.example @@ -19,8 +19,6 @@ wal_format = "bincode" [flush] interval_secs = 10 -wal_size_threshold_mb = 64 -time_bucket_duration = "1h" max_points_per_batch = 50000 # WAL group-commit: max entries to coalesce per write batch (0 = disabled) wal_batch_size = 64 @@ -33,8 +31,15 @@ arrow_wal_enabled = true [chdb] session_data_path = "./chdb_data" -pool_size = 4 # same-path chDB connections for parallel flush inserts and queries +# Legacy: sets both pools when query_pool_size / write_pool_size are unset. +# pool_size = 4 +query_pool_size = 4 # connections reserved for queries +write_pool_size = 4 # connections reserved for ingest/flush (isolated from queries) schema_cache_max_entries = 10000 +# tag_low_cardinality_max = 100000 # unset = use [cardinality].max_tag_values_per_measurement +insert_max_threads = 4 +# insert_min_insert_block_size_rows = 100000 # match max_points_per_batch to reduce small parts +# insert_max_insert_block_size = 0 # 0 = ClickHouse engine default [auth] enabled = false diff --git a/deploy/examples/three-node.yaml b/deploy/examples/three-node.yaml index e942c9a..73f5578 100644 --- a/deploy/examples/three-node.yaml +++ b/deploy/examples/three-node.yaml @@ -16,8 +16,6 @@ spec: size: 10Gi flush: intervalSecs: 10 - walSizeThresholdMb: 64 - timeBucketDuration: "1h" chdb: sessionDataPath: /var/lib/hyperbytedb/chdb retention: diff --git a/deploy/kind/manifests/hyperbytedb-cr.yaml b/deploy/kind/manifests/hyperbytedb-cr.yaml index ba7fac2..4c60810 100644 --- a/deploy/kind/manifests/hyperbytedb-cr.yaml +++ b/deploy/kind/manifests/hyperbytedb-cr.yaml @@ -24,8 +24,6 @@ spec: poolSize: 4 flush: intervalSecs: 5 - walSizeThresholdMb: 256 - timeBucketDuration: "1h" maxPointsPerBatch: 100000 walBatchSize: 128 walBatchDelayUs: 0 diff --git a/docs/deep-dive/deep-dive-read-path.md b/docs/deep-dive/deep-dive-read-path.md index 7a3da11..1c8cd3e 100644 --- a/docs/deep-dive/deep-dive-read-path.md +++ b/docs/deep-dive/deep-dive-read-path.md @@ -177,7 +177,7 @@ On SELECT, `inject_tombstone_predicates()` loads all tombstones for the measurem ### Session model -chDB runs inside `spawn_blocking`. HyperbyteDB opens `chdb.pool_size` connections to the same `session_data_path`; each connection has its own `ChdbClient` mutex, so flush inserts and queries can overlap when the pool has more than one connection. Tune `server.max_concurrent_queries` (≥ `pool_size`) to cap in-flight query tasks. +chDB runs inside `spawn_blocking`. HyperbyteDB opens separate query and write connection pools to the same `session_data_path` (`chdb.query_pool_size` and `chdb.write_pool_size`, default 4 each). The read path uses the query pool only; each connection has its own `ChdbClient` mutex, so concurrent query tasks overlap when `query_pool_size > 1`. Ingest/flush uses the write pool, so heavy queries do not block inserts. Tune `server.max_concurrent_queries` (≥ `query_pool_size`) to cap in-flight query tasks. ### Output format diff --git a/docs/developer-guide/system-architecture.md b/docs/developer-guide/system-architecture.md index af4d43b..3935857 100644 --- a/docs/developer-guide/system-architecture.md +++ b/docs/developer-guide/system-architecture.md @@ -415,13 +415,20 @@ HyperbyteDB uses **chDB** (embedded ClickHouse) as its query engine and storage ### Session management -Each chDB `Connection` is `Send` but not `Sync`. HyperbyteDB keeps a pool of connections to the same `session_data_path` (see `ChdbConnectionPool` in `adapters/chdb/connection_pool.rs`). Queries and inserts run in `spawn_blocking`, checking out one connection per task. +Each chDB `Connection` is `Send` but not `Sync`. HyperbyteDB opens **two** connection pools to the same `session_data_path` (see `ChdbConnectionPool` in `adapters/chdb/connection_pool.rs`): -### Single connection (`pool_size = 1`) +- **Query pool** (`chdb.query_pool_size`, default 4) — used by `ChdbQueryAdapter` for reads. +- **Write pool** (`chdb.write_pool_size`, default 4) — used by `ChdbNativeAdapter` for ingest/flush. -One connection: all chDB work serializes on that client's mutex (legacy / minimal footprint). +Legacy `chdb.pool_size` (when non-zero) sets both pools to the same size when the explicit keys are unset. Each pool is clamped to 1–128 connections. -### Connection pool (`pool_size > 1`) +Queries and inserts run in `spawn_blocking`, checking out one connection from the appropriate pool per task. Separate pools isolate flush inserts from concurrent queries. + +### Single connection (`query_pool_size = 1` or `write_pool_size = 1`) + +One connection in a pool: all work on that pool serializes on that client's mutex (minimal footprint). + +### Connection pool (size > 1) ```rust struct ChdbConnectionPool { @@ -430,7 +437,7 @@ struct ChdbConnectionPool { } ``` -Round-robin checkout with `try_lock` on busy slots. Multiple connections share one process-global `EmbeddedServer` for the data path; each connection gets an independent `ChdbClient` mutex, so concurrent flush inserts and queries can overlap. +Round-robin checkout with `try_lock` on busy slots. Multiple connections share one process-global `EmbeddedServer` for the data path; each connection gets an independent `ChdbClient` mutex, so concurrent tasks within a pool can overlap. ### Output format diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index d64e117..21ad2fb 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -61,8 +61,6 @@ Controls the background WAL-to-chDB flush pipeline. | Key | Type | Default | Description | |-----|------|---------|-------------| | `interval_secs` | integer | `10` | How often the flush service runs (seconds) | -| `wal_size_threshold_mb` | integer | `64` | WAL size that triggers an immediate flush (MB) | -| `time_bucket_duration` | string | `"1h"` | Time bucket granularity used when grouping WAL entries for flush | | `max_points_per_batch` | integer | `50000` | Max points per chDB insert batch (server clamps to 10k–500k; `0` uses the same default) | | `wal_batch_size` | integer | `64` | WAL group-commit: max entries to coalesce per write batch; `0` = disabled | | `wal_batch_delay_us` | integer | `200` | WAL group-commit: max microseconds to wait for more entries before flushing | @@ -77,7 +75,14 @@ Embedded ClickHouse (chDB) query engine settings. | Key | Type | Default | Description | |-----|------|---------|-------------| | `session_data_path` | string | `"./chdb_data"` | chDB session state directory | -| `pool_size` | integer | `4` | Number of chDB connections to the same `session_data_path`. Each connection has its own client mutex, so flush inserts and concurrent queries overlap when `pool_size > 1`. Clamped to 1–32. For best overlap, set `server.max_concurrent_queries` ≥ `pool_size`. | +| `query_pool_size` | integer | `4` | chDB connections reserved for queries (`ChdbQueryAdapter`). Each connection has its own client mutex, so concurrent `spawn_blocking` query tasks overlap when > 1. Clamped to 1–128. For best overlap, set `server.max_concurrent_queries` ≥ `query_pool_size`. | +| `write_pool_size` | integer | `4` | chDB connections reserved for ingest/flush (`ChdbNativeAdapter`), isolated from the query pool so heavy queries do not block inserts. Clamped to 1–128. | +| `pool_size` | integer | `0` (unused) | **Legacy.** When non-zero and `query_pool_size` / `write_pool_size` are unset, applies the same size to both pools. Prefer explicit `query_pool_size` and `write_pool_size`. | +| `schema_cache_max_entries` | integer | `10000` | Max `(db, rp, measurement)` entries in the chDB native adapter schema and series caches. Oldest entries are evicted (LRU). | +| `insert_max_threads` | integer | `4` | ClickHouse `max_threads` for Arrow bulk inserts. Match CPU cores on the node. | +| `insert_min_insert_block_size_rows` | integer | `0` (unset) | ClickHouse `min_insert_block_size_rows` for Arrow bulk inserts. Set to ~`max_points_per_batch` to avoid many small parts. `0` = engine default. | +| `insert_max_insert_block_size` | integer | `0` (unset) | ClickHouse `max_insert_block_size` for Arrow bulk inserts (bytes). Caps part size for wide measurements. `0` = engine default. | +| `tag_low_cardinality_max` | integer | *(linked)* | Max distinct tag values per key before DDL uses plain `String` instead of `LowCardinality(String)`. When unset, uses `[cardinality].max_tag_values_per_measurement`. High-cardinality tags (trace IDs, request IDs) should stay as plain `String`. | --- @@ -239,7 +244,8 @@ interval_secs = 10 [chdb] session_data_path = "./chdb_data" -pool_size = 4 +query_pool_size = 4 +write_pool_size = 4 [logging] level = "info" @@ -266,6 +272,9 @@ interval_secs = 10 [chdb] session_data_path = "/var/lib/hyperbytedb/chdb" +query_pool_size = 32 +write_pool_size = 4 +schema_cache_max_entries = 10000 [cluster] enabled = true diff --git a/docs/user-guide/operator/cluster.md b/docs/user-guide/operator/cluster.md index f15a359..967eec7 100644 --- a/docs/user-guide/operator/cluster.md +++ b/docs/user-guide/operator/cluster.md @@ -49,8 +49,6 @@ spec: storageClassName: fast-ssd flush: intervalSecs: 5 - walSizeThresholdMb: 128 - timeBucketDuration: "1h" chdb: sessionDataPath: /var/lib/hyperbytedb/chdb auth: @@ -226,8 +224,6 @@ HyperbyteDB stores WAL, metadata, Raft state, and chDB session data on the per-r | Field | Type | Default | Description | |-------|------|---------|-------------| | `intervalSecs` | int32 | `10` | How often the WAL is flushed to chDB | -| `walSizeThresholdMb` | int32 | `64` | WAL size threshold that triggers an early flush | -| `timeBucketDuration` | string | `1h` | Parquet time-bucket width (`1h` or `1d`) | | `maxPointsPerBatch` | int32 | `50000` | Max points per chDB insert batch (written to ConfigMap as `max_points_per_batch`) | | `walBatchSize` | int32 | `64` | WAL group-commit batch size (`0` disables) | | `walBatchDelayUs` | int64 | `200` | WAL group-commit delay in microseconds | diff --git a/docs/user-guide/resource-sizing.md b/docs/user-guide/resource-sizing.md index bf02a4e..9cc3668 100644 --- a/docs/user-guide/resource-sizing.md +++ b/docs/user-guide/resource-sizing.md @@ -49,9 +49,10 @@ Size CPU and RAM from how many queries run at once and how heavy they are — no max_concurrent_queries = 16 [chdb] - pool_size = 4 + query_pool_size = 4 + write_pool_size = 4 ``` - Set `max_concurrent_queries` ≥ `pool_size` so flush and queries can overlap. See [Configuration](configuration.md). + Set `max_concurrent_queries` ≥ `query_pool_size` so concurrent queries can use the query pool. Ingest/flush uses a separate `write_pool_size` pool. See [Configuration](configuration.md). - If queries are slow but CPU is idle, you may be I/O-bound on disk — check storage type and free space before adding cores. - If CPU is saturated while ingest stays healthy, reduce concurrency or simplify queries before scaling write throughput assumptions. @@ -176,7 +177,7 @@ Example starting points (per node, before replication overhead): | Symptom | Likely cause | What to try | |---------|-------------|-------------| -| High CPU | Many concurrent or heavy queries | Lower `max_concurrent_queries`; narrow time ranges in queries; reduce `pool_size` if threads oversubscribe | +| High CPU | Many concurrent or heavy queries | Lower `max_concurrent_queries`; narrow time ranges in queries; reduce `query_pool_size` if threads oversubscribe | | Slow queries | Wide scans, missing time filter | Add `WHERE time > ...`; reduce concurrent query load | | High memory | chDB working set or Arrow cache | Cap concurrent queries; set `arrow_wal_enabled = false` if flush cache is the issue | | Disk filling up | Retention too long or underestimated volume | Shorten retention policies; verify `[retention]` is enabled | @@ -205,7 +206,7 @@ Start with defaults, deploy with realistic query patterns (same dashboards and a ## See Also -- [Configuration](configuration.md) — Tuning parameters (`max_concurrent_queries`, `pool_size`, flush settings) +- [Configuration](configuration.md) — Tuning parameters (`max_concurrent_queries`, `query_pool_size`, `write_pool_size`, flush settings) - [Administration](administration.md) — Metrics and monitoring - [Troubleshooting](troubleshooting.md) — Query timeouts, memory, and flush issues - [V1 Stable Scope](v1-stable-scope.md) — Supported topologies and availability model diff --git a/docs/user-guide/troubleshooting.md b/docs/user-guide/troubleshooting.md index 7b78782..4033d6c 100644 --- a/docs/user-guide/troubleshooting.md +++ b/docs/user-guide/troubleshooting.md @@ -135,7 +135,7 @@ Data must be flushed from the WAL into chDB MergeTree tables before it becomes q 2. **Add a time range to your query.** Queries without `WHERE time > ...` scan all data. -3. **Cap concurrent queries.** Tune `server.max_concurrent_queries` so heavy queries do not oversubscribe the Tokio blocking pool. For overlapping chDB work, also set `chdb.pool_size` > 1 (same data path, multiple connections) and keep `max_concurrent_queries` ≥ `pool_size`. +3. **Cap concurrent queries.** Tune `server.max_concurrent_queries` so heavy queries do not oversubscribe the Tokio blocking pool. For overlapping chDB query work, set `chdb.query_pool_size` > 1 (same data path, multiple connections) and keep `max_concurrent_queries` ≥ `query_pool_size`. Ingest/flush uses a separate `write_pool_size` pool, so heavy queries do not block inserts. 4. **Narrow the time range** in your query to reduce scanned data volume. diff --git a/hyperbytedb/src/adapters/chdb/connection_pool.rs b/hyperbytedb/src/adapters/chdb/connection_pool.rs index 140279f..6863572 100644 --- a/hyperbytedb/src/adapters/chdb/connection_pool.rs +++ b/hyperbytedb/src/adapters/chdb/connection_pool.rs @@ -13,8 +13,10 @@ use parking_lot::Mutex; use crate::error::HyperbytedbError; pub const MIN_POOL_SIZE: usize = 1; -pub const MAX_POOL_SIZE: usize = 32; +pub const MAX_POOL_SIZE: usize = 128; pub const DEFAULT_POOL_SIZE: usize = 4; +pub const DEFAULT_QUERY_POOL_SIZE: usize = 4; +pub const DEFAULT_WRITE_POOL_SIZE: usize = 4; /// Clamp configured pool size to a safe range. pub fn clamp_pool_size(size: usize) -> usize { @@ -116,7 +118,8 @@ mod tests { assert_eq!(clamp_pool_size(0), 1); assert_eq!(clamp_pool_size(1), 1); assert_eq!(clamp_pool_size(4), 4); - assert_eq!(clamp_pool_size(100), 32); + assert_eq!(clamp_pool_size(100), 100); + assert_eq!(clamp_pool_size(200), 128); } #[test] diff --git a/hyperbytedb/src/adapters/chdb/native_adapter.rs b/hyperbytedb/src/adapters/chdb/native_adapter.rs index aef42d0..e4f8381 100644 --- a/hyperbytedb/src/adapters/chdb/native_adapter.rs +++ b/hyperbytedb/src/adapters/chdb/native_adapter.rs @@ -66,12 +66,12 @@ use crate::error::HyperbytedbError; use crate::ports::metadata::MetadataPort; use crate::ports::points_sink::{PointsSinkPort, WriteAck}; -/// Above this many distinct values per tag key, ClickHouse -/// `LowCardinality(String)` hurts more than it helps; use plain `String`. -pub const TAG_LOW_CARDINALITY_MAX: usize = 100_000; +/// Default threshold when no `[chdb].tag_low_cardinality_max` or cardinality +/// config is available (e.g. unit tests constructing the adapter directly). +pub const DEFAULT_TAG_LOW_CARDINALITY_MAX: usize = 100_000; /// Column kind for tags and fields. Tags use `LowCardinality(String)` only -/// while distinct value count stays at or below [`TAG_LOW_CARDINALITY_MAX`]. +/// while distinct value count stays at or below the configured threshold. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] enum ColumnKind { TagLowCardinality, @@ -133,8 +133,8 @@ fn table_schema_from_measurement_meta( ) } -fn tag_column_kind(distinct_values: usize) -> ColumnKind { - if distinct_values > TAG_LOW_CARDINALITY_MAX { +fn tag_column_kind(distinct_values: usize, threshold: usize) -> ColumnKind { + if distinct_values > threshold { ColumnKind::TagString } else { ColumnKind::TagLowCardinality @@ -233,6 +233,10 @@ pub struct ChdbNativeAdapter { /// serialization + re-parsing and is several times faster. Set /// `HYPERBYTEDB_DISABLE_ARROW_INSERT=1` to fall back to the SQL path. use_arrow: bool, + /// ClickHouse insert SETTINGS for Arrow bulk paths (from `[chdb]` config). + insert_options: InsertOptions, + /// Distinct tag values per key above which DDL uses plain `String`. + tag_low_cardinality_max: usize, } impl ChdbNativeAdapter { @@ -245,6 +249,8 @@ impl ChdbNativeAdapter { session, metadata, crate::config::default_schema_cache_max_entries(), + InsertOptions::default_bulk(), + crate::config::default_tag_low_cardinality_max(), ) } @@ -252,6 +258,8 @@ impl ChdbNativeAdapter { session: SharedSession, metadata: Option>, schema_cache_max_entries: usize, + insert_options: InsertOptions, + tag_low_cardinality_max: usize, ) -> Self { let cache_capacity = table_cache_capacity(schema_cache_max_entries); let use_arrow = std::env::var("HYPERBYTEDB_DISABLE_ARROW_INSERT").is_err(); @@ -264,6 +272,8 @@ impl ChdbNativeAdapter { ddl_locks: Arc::new(tokio::sync::Mutex::new(LruCache::new(cache_capacity))), cache_capacity, use_arrow, + insert_options, + tag_low_cardinality_max, } } @@ -781,7 +791,7 @@ impl ChdbNativeAdapter { }; if !series_cached.materialized && !series_cached.columns.is_empty() { // Warmed-from-metadata series table may still carry LowCardinality - // tags that have since crossed TAG_LOW_CARDINALITY_MAX. + // tags that have since crossed the configured threshold. series_alters.extend(build_alter_reconcile_tag_strings(&series_table, &tag_phys)); } @@ -866,15 +876,12 @@ impl ChdbNativeAdapter { let batch = build_series_record_batch(ensured, &new_series)?; let pool = self.session.pool()?; let series_table = ensured.series_table.clone(); + let insert_options = self.insert_options.clone(); tokio::task::spawn_blocking(move || { pool.with_connection(|conn| { - insert_record_batch_direct( - conn, - &series_table, - batch, - InsertOptions::default_bulk(), + insert_record_batch_direct(conn, &series_table, batch, insert_options).map_err( + |e| HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e)), ) - .map_err(|e| HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e))) }) }) .await @@ -911,7 +918,10 @@ impl ChdbNativeAdapter { let count = meta .count_tag_values(db, rp, tag_key, Some(measurement)) .await?; - tag_kinds.insert(tag_key.clone(), tag_column_kind(count)); + tag_kinds.insert( + tag_key.clone(), + tag_column_kind(count, self.tag_low_cardinality_max), + ); } Ok(table_schema_from_measurement_meta(meas_meta, &tag_kinds)) } @@ -942,7 +952,7 @@ impl ChdbNativeAdapter { // tag and did a metadata lookup per distinct value // (O(rows) — the dominant flush cost for high-cardinality // tags like `machine_id`). Auto LC->String promotion above - // TAG_LOW_CARDINALITY_MAX still happens at first + // the configured threshold still happens at first // materialization and whenever a new tag *key* appears; an // oversized LowCardinality column degrades gracefully rather // than being written incorrectly. @@ -954,7 +964,7 @@ impl ChdbNativeAdapter { let count = self .distinct_tag_value_count(db, rp, measurement, tag_key, points) .await?; - Ok(tag_column_kind(count)) + Ok(tag_column_kind(count, self.tag_low_cardinality_max)) } /// Distinct tag values for `(db, measurement, tag_key)` from metadata plus @@ -1323,15 +1333,11 @@ impl ChdbNativeAdapter { let pool = self.session.pool()?; let series_table = ensured.series_table.clone(); let batch = series_batch.clone(); + let insert_options = self.insert_options.clone(); tokio::task::spawn_blocking(move || { pool.with_connection(|conn| { - insert_record_batch_direct( - conn, - &series_table, - batch, - InsertOptions::default_bulk(), - ) - .map_err(|e| HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e))) + insert_record_batch_direct(conn, &series_table, batch, insert_options) + .map_err(|e| HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e))) }) }) .await @@ -1426,12 +1432,12 @@ impl PointsSinkPort for ChdbNativeAdapter { let pool = self.session.pool()?; let table = ensured.table.clone(); let insert_start = std::time::Instant::now(); + let insert_options = self.insert_options.clone(); tokio::task::spawn_blocking(move || { pool.with_connection(|conn| { - insert_record_batch_direct(conn, &table, batch, InsertOptions::default_bulk()) - .map_err(|e| { - HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e)) - }) + insert_record_batch_direct(conn, &table, batch, insert_options).map_err(|e| { + HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e)) + }) }) }) .await @@ -1505,10 +1511,11 @@ impl PointsSinkPort for ChdbNativeAdapter { let table = batch.table_name.clone(); let fact = Arc::new(padded); let insert_start = std::time::Instant::now(); + let insert_options = self.insert_options.clone(); tokio::task::spawn_blocking(move || { let batch = (*fact).clone(); pool.with_connection(|conn| { - insert_record_batch_direct(conn, &table, batch, InsertOptions::default_bulk()) + insert_record_batch_direct(conn, &table, batch, insert_options) .map_err(|e| HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e))) }) }) @@ -1789,8 +1796,8 @@ fn build_alter_add_series_columns( } /// After a metadata-only cache warm, existing MergeTree tables may still use -/// `LowCardinality(String)` for tags that have since crossed -/// [`TAG_LOW_CARDINALITY_MAX`]. `MODIFY` to `String` is safe when the column +/// `LowCardinality(String)` for tags that have since crossed the configured +/// low-cardinality threshold. `MODIFY` to `String` is safe when the column /// is already plain `String`. fn build_alter_reconcile_tag_strings( table: &str, @@ -2460,13 +2467,14 @@ mod tests { } #[test] - fn tag_column_kind_switches_at_100k() { + fn tag_column_kind_switches_at_threshold() { + let threshold = DEFAULT_TAG_LOW_CARDINALITY_MAX; assert_eq!( - tag_column_kind(TAG_LOW_CARDINALITY_MAX), + tag_column_kind(threshold, threshold), ColumnKind::TagLowCardinality ); assert_eq!( - tag_column_kind(TAG_LOW_CARDINALITY_MAX + 1), + tag_column_kind(threshold + 1, threshold), ColumnKind::TagString ); } diff --git a/hyperbytedb/src/bootstrap.rs b/hyperbytedb/src/bootstrap.rs index 5cf9baf..d014636 100644 --- a/hyperbytedb/src/bootstrap.rs +++ b/hyperbytedb/src/bootstrap.rs @@ -158,14 +158,17 @@ pub async fn build_services(config: &HyperbytedbConfig) -> anyhow::Result tracing::info!( attached, "attached restored chDB tables from on-disk metadata" @@ -176,13 +179,17 @@ pub async fn build_services(config: &HyperbytedbConfig) -> anyhow::Result = Arc::new(chdb_adapter); let native_sink = ChdbNativeAdapter::with_metadata_and_cache_limit( - shared_chdb.clone(), + write_chdb, Some(metadata.clone()), config.chdb.schema_cache_max_entries, + config.chdb.insert_options(), + config + .chdb + .resolved_tag_low_cardinality_max(&config.cardinality), ); match native_sink.warm_schemas_from_metadata().await { Ok(tables) => tracing::info!( diff --git a/hyperbytedb/src/config.rs b/hyperbytedb/src/config.rs index d5133dd..f53f41a 100644 --- a/hyperbytedb/src/config.rs +++ b/hyperbytedb/src/config.rs @@ -68,8 +68,6 @@ fn default_wal_format() -> String { #[serde(deny_unknown_fields)] pub struct FlushConfig { pub interval_secs: u64, - pub wal_size_threshold_mb: u64, - pub time_bucket_duration: String, /// Max points per chDB insert batch. `0` uses [`default_max_points_per_batch`]. #[serde(default = "default_max_points_per_batch")] pub max_points_per_batch: usize, @@ -110,6 +108,10 @@ pub fn default_schema_cache_max_entries() -> usize { 10_000 } +pub fn default_tag_low_cardinality_max() -> usize { + 100_000 +} + fn default_wal_batch_size() -> usize { 64 } @@ -118,21 +120,74 @@ fn default_wal_batch_delay_us() -> u64 { 200 } +fn default_insert_max_threads() -> u32 { + 4 +} + #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(deny_unknown_fields)] pub struct ChdbConfig { pub session_data_path: String, - /// Number of chDB connections opened to the same `session_data_path`. - /// Each connection has its own `ChdbClient` mutex, so flush inserts and - /// concurrent queries overlap when `pool_size > 1`. A second connection - /// to a *different* path still fails (process-global singleton per path). - /// Clamped to 1..=32. For best overlap, set `server.max_concurrent_queries` - /// ≥ `pool_size`. + /// Legacy: when non-zero and `query_pool_size` / `write_pool_size` are unset, + /// applies the same size to both pools. + #[serde(default)] pub pool_size: usize, + /// chDB connections for queries (`ChdbQueryAdapter`). When unset, uses + /// `pool_size` if non-zero, else 4. Clamped to 1..=128. + #[serde(default)] + pub query_pool_size: Option, + /// chDB connections for ingest/flush (`ChdbNativeAdapter`). When unset, uses + /// `pool_size` if non-zero, else 4. Clamped to 1..=128. + #[serde(default)] + pub write_pool_size: Option, /// Max `(db, rp, measurement)` entries in the chDB native adapter in-memory /// schema and series caches. Oldest entries are evicted (LRU). #[serde(default = "default_schema_cache_max_entries")] pub schema_cache_max_entries: usize, + /// ClickHouse `max_threads` for Arrow bulk inserts. + #[serde(default = "default_insert_max_threads")] + pub insert_max_threads: u32, + /// ClickHouse `min_insert_block_size_rows` for Arrow bulk inserts. `0` = engine default. + #[serde(default)] + pub insert_min_insert_block_size_rows: u64, + /// ClickHouse `max_insert_block_size` for Arrow bulk inserts. `0` = engine default. + #[serde(default)] + pub insert_max_insert_block_size: u64, + /// Max distinct tag values per key before DDL uses plain `String` instead of + /// `LowCardinality(String)`. When unset, uses + /// `[cardinality].max_tag_values_per_measurement`. + #[serde(default)] + pub tag_low_cardinality_max: Option, +} + +impl ChdbConfig { + /// Maps insert tuning keys to chDB [`InsertOptions`] for the native adapter. + pub fn insert_options(&self) -> chdb_rust::InsertOptions { + chdb_rust::InsertOptions { + max_threads: Some(self.insert_max_threads), + max_insert_block_size: (self.insert_max_insert_block_size > 0) + .then_some(self.insert_max_insert_block_size), + min_insert_block_size_rows: (self.insert_min_insert_block_size_rows > 0) + .then_some(self.insert_min_insert_block_size_rows), + } + } + + pub fn resolved_query_pool_size(&self) -> usize { + self.query_pool_size + .or((self.pool_size > 0).then_some(self.pool_size)) + .unwrap_or(crate::adapters::chdb::connection_pool::DEFAULT_QUERY_POOL_SIZE) + } + + pub fn resolved_write_pool_size(&self) -> usize { + self.write_pool_size + .or((self.pool_size > 0).then_some(self.pool_size)) + .unwrap_or(crate::adapters::chdb::connection_pool::DEFAULT_WRITE_POOL_SIZE) + } + + pub fn resolved_tag_low_cardinality_max(&self, cardinality: &CardinalityConfig) -> usize { + self.tag_low_cardinality_max + .unwrap_or(cardinality.max_tag_values_per_measurement) + } } #[derive(Debug, Clone, Deserialize, Serialize)] @@ -513,10 +568,6 @@ impl RetentionConfig { } } -fn default_chdb_pool_size() -> usize { - crate::adapters::chdb::connection_pool::DEFAULT_POOL_SIZE -} - impl HyperbytedbConfig { pub fn load(config_path: Option<&str>) -> anyhow::Result { let mut figment = Figment::new().merge(Serialized::defaults(Self::defaults())); @@ -552,8 +603,6 @@ impl HyperbytedbConfig { }, flush: FlushConfig { interval_secs: 10, - wal_size_threshold_mb: 64, - time_bucket_duration: "1h".to_string(), max_points_per_batch: default_max_points_per_batch(), wal_batch_size: default_wal_batch_size(), wal_batch_delay_us: default_wal_batch_delay_us(), @@ -562,8 +611,14 @@ impl HyperbytedbConfig { }, chdb: ChdbConfig { session_data_path: "./chdb_data".to_string(), - pool_size: default_chdb_pool_size(), + pool_size: 0, + query_pool_size: None, + write_pool_size: None, schema_cache_max_entries: default_schema_cache_max_entries(), + insert_max_threads: default_insert_max_threads(), + insert_min_insert_block_size_rows: 0, + insert_max_insert_block_size: 0, + tag_low_cardinality_max: None, }, auth: AuthConfig { enabled: false, @@ -772,6 +827,191 @@ mod retention_config_tests { } } +#[cfg(test)] +mod chdb_pool_config_tests { + use super::{ + CardinalityConfig, ChdbConfig, default_insert_max_threads, default_tag_low_cardinality_max, + }; + use chdb_rust::InsertOptions; + + #[test] + fn split_pool_defaults_when_unset() { + let c = ChdbConfig { + session_data_path: "./chdb".into(), + pool_size: 0, + query_pool_size: None, + write_pool_size: None, + schema_cache_max_entries: 10_000, + insert_max_threads: default_insert_max_threads(), + insert_min_insert_block_size_rows: 0, + insert_max_insert_block_size: 0, + tag_low_cardinality_max: None, + }; + assert_eq!(c.resolved_query_pool_size(), 4); + assert_eq!(c.resolved_write_pool_size(), 4); + } + + #[test] + fn legacy_pool_size_applies_to_both() { + let c = ChdbConfig { + session_data_path: "./chdb".into(), + pool_size: 6, + query_pool_size: None, + write_pool_size: None, + schema_cache_max_entries: 10_000, + insert_max_threads: default_insert_max_threads(), + insert_min_insert_block_size_rows: 0, + insert_max_insert_block_size: 0, + tag_low_cardinality_max: None, + }; + assert_eq!(c.resolved_query_pool_size(), 6); + assert_eq!(c.resolved_write_pool_size(), 6); + } + + #[test] + fn explicit_split_overrides_pool_size() { + let c = ChdbConfig { + session_data_path: "./chdb".into(), + pool_size: 6, + query_pool_size: Some(64), + write_pool_size: Some(8), + schema_cache_max_entries: 10_000, + insert_max_threads: default_insert_max_threads(), + insert_min_insert_block_size_rows: 0, + insert_max_insert_block_size: 0, + tag_low_cardinality_max: None, + }; + assert_eq!(c.resolved_query_pool_size(), 64); + assert_eq!(c.resolved_write_pool_size(), 8); + } + + #[test] + fn deserializes_split_pools_from_toml() { + use figment::Figment; + use figment::providers::{Format, Toml}; + + let c: ChdbConfig = Figment::new() + .merge(Toml::string( + r#" + session_data_path = "./chdb" + query_pool_size = 32 + write_pool_size = 4 + "#, + )) + .extract() + .expect("parse"); + assert_eq!(c.resolved_query_pool_size(), 32); + assert_eq!(c.resolved_write_pool_size(), 4); + } + + #[test] + fn insert_options_maps_config_keys() { + let c = ChdbConfig { + session_data_path: "./chdb".into(), + pool_size: 0, + query_pool_size: None, + write_pool_size: None, + schema_cache_max_entries: 10_000, + insert_max_threads: 8, + insert_min_insert_block_size_rows: 100_000, + insert_max_insert_block_size: 0, + tag_low_cardinality_max: None, + }; + assert_eq!( + c.insert_options(), + InsertOptions { + max_threads: Some(8), + min_insert_block_size_rows: Some(100_000), + max_insert_block_size: None, + } + ); + } + + #[test] + fn deserializes_insert_settings_from_toml() { + use figment::Figment; + use figment::providers::{Format, Toml}; + + let c: ChdbConfig = Figment::new() + .merge(Toml::string( + r#" + session_data_path = "./chdb" + insert_max_threads = 4 + insert_min_insert_block_size_rows = 100000 + insert_max_insert_block_size = 1048576 + "#, + )) + .extract() + .expect("parse"); + assert_eq!( + c.insert_options(), + InsertOptions { + max_threads: Some(4), + min_insert_block_size_rows: Some(100_000), + max_insert_block_size: Some(1_048_576), + } + ); + } + + #[test] + fn tag_low_cardinality_max_defaults_to_cardinality_limit() { + let c = ChdbConfig { + session_data_path: "./chdb".into(), + pool_size: 0, + query_pool_size: None, + write_pool_size: None, + schema_cache_max_entries: 10_000, + insert_max_threads: default_insert_max_threads(), + insert_min_insert_block_size_rows: 0, + insert_max_insert_block_size: 0, + tag_low_cardinality_max: None, + }; + let cardinality = CardinalityConfig { + max_tag_values_per_measurement: 50_000, + max_measurements_per_database: 10_000, + }; + assert_eq!(c.resolved_tag_low_cardinality_max(&cardinality), 50_000); + } + + #[test] + fn tag_low_cardinality_max_explicit_override() { + let c = ChdbConfig { + session_data_path: "./chdb".into(), + pool_size: 0, + query_pool_size: None, + write_pool_size: None, + schema_cache_max_entries: 10_000, + insert_max_threads: default_insert_max_threads(), + insert_min_insert_block_size_rows: 0, + insert_max_insert_block_size: 0, + tag_low_cardinality_max: Some(25_000), + }; + let cardinality = CardinalityConfig { + max_tag_values_per_measurement: 100_000, + max_measurements_per_database: 10_000, + }; + assert_eq!(c.resolved_tag_low_cardinality_max(&cardinality), 25_000); + } + + #[test] + fn deserializes_tag_low_cardinality_max_from_toml() { + use figment::Figment; + use figment::providers::{Format, Toml}; + + let c: ChdbConfig = Figment::new() + .merge(Toml::string( + r#" + session_data_path = "./chdb" + tag_low_cardinality_max = 75000 + "#, + )) + .extract() + .expect("parse"); + assert_eq!(c.tag_low_cardinality_max, Some(75_000)); + assert_eq!(default_tag_low_cardinality_max(), 100_000); + } +} + #[cfg(test)] mod replicate_body_limit_tests { use super::ClusterConfig;