From 3c527f7d10ebd1c7a4eed627924ed7b3021e91af Mon Sep 17 00:00:00 2001 From: "austin.barrington" Date: Wed, 15 Jul 2026 14:57:31 +0100 Subject: [PATCH] feat: proxy backend exclusion, materialized view backfill, config defaults - Add operator-driven backend exclusion to the proxy (exclude/ready endpoints, per-backend AtomicBool, reconcile-time GC, pool_status) - Add WITH BACKFILL opt-in for CREATE MATERIALIZED VIEW (parser, AST, digest normalization, service gating, tests) - Bump Raft heartbeat default from 300ms to 1000ms - Rename API domain from hyperbytedb.hyperbytedb.io to hyperbytedb.hyperbyte.cloud in CRD manifests and RBAC --- config.toml.example | 2 +- deploy/examples/three-node.yaml | 2 +- deploy/kind/kind-config.yaml | 6 +- deploy/kind/manifests/hyperbytedb-cr.yaml | 3 +- deploy/kind/manifests/operator.yaml | 6 +- docs/deep-dive/deep-dive-clustering.md | 4 +- docs/user-guide/advanced-features.md | 15 +- docs/user-guide/configuration.md | 2 +- docs/user-guide/operator/cluster.md | 4 +- hyperbytedb-proxy/Dockerfile | 1 + hyperbytedb-proxy/src/admin.rs | 75 ++- hyperbytedb-proxy/src/backend.rs | 16 +- hyperbytedb-proxy/src/lib.rs | 5 +- hyperbytedb-proxy/src/pool.rs | 103 +++- .../src/application/cluster/bootstrap.rs | 2 +- .../application/materialized_view_service.rs | 26 +- hyperbytedb/src/config.rs | 2 +- hyperbytedb/src/domain/materialized_view.rs | 3 + hyperbytedb/src/timeseriesql/ast.rs | 2 + hyperbytedb/src/timeseriesql/ddl_parser.rs | 31 ++ hyperbytedb/src/timeseriesql/digest.rs | 26 + hyperbytedb/src/timeseriesql/lexer.rs | 1 + hyperbytedb/src/timeseriesql/parser.rs | 50 ++ hyperbytedb/tests/compat/ddl_tests.rs | 454 +++++++++++++++++- 24 files changed, 797 insertions(+), 44 deletions(-) diff --git a/config.toml.example b/config.toml.example index 16148ef..2023ec5 100644 --- a/config.toml.example +++ b/config.toml.example @@ -69,7 +69,7 @@ replicate_receiver_queue_depth = 1024 replicate_receiver_workers = 1 replication_truncate_stale_peer_multiplier = 2 # Raft consensus settings (used when cluster is enabled) -# raft_heartbeat_interval_ms = 300 +# raft_heartbeat_interval_ms = 1000 # raft_election_timeout_ms = 1000 # raft_snapshot_threshold = 1000 diff --git a/deploy/examples/three-node.yaml b/deploy/examples/three-node.yaml index 265b5d6..e942c9a 100644 --- a/deploy/examples/three-node.yaml +++ b/deploy/examples/three-node.yaml @@ -37,7 +37,7 @@ spec: heartbeatIntervalSecs: 2 heartbeatMissThreshold: 5 replicationMaxRetries: 5 - raftHeartbeatIntervalMs: 300 + raftHeartbeatIntervalMs: 1000 raftElectionTimeoutMs: 1000 replication: mode: async diff --git a/deploy/kind/kind-config.yaml b/deploy/kind/kind-config.yaml index f5ac6b9..e1affb2 100644 --- a/deploy/kind/kind-config.yaml +++ b/deploy/kind/kind-config.yaml @@ -7,13 +7,13 @@ nodes: # hyperbytedb-proxy NodePort → 30086 → localhost:8086 extraPortMappings: - containerPort: 30086 - hostPort: 8086 + hostPort: 18086 protocol: TCP - containerPort: 30090 - hostPort: 9090 + hostPort: 19090 protocol: TCP - containerPort: 30000 - hostPort: 3000 + hostPort: 13000 protocol: TCP - role: worker extraMounts: diff --git a/deploy/kind/manifests/hyperbytedb-cr.yaml b/deploy/kind/manifests/hyperbytedb-cr.yaml index 97e533b..ba7fac2 100644 --- a/deploy/kind/manifests/hyperbytedb-cr.yaml +++ b/deploy/kind/manifests/hyperbytedb-cr.yaml @@ -1,6 +1,6 @@ # HyperbyteDB cluster for kind — tuned for a 4-CPU / high-memory node (16 GiB pod limit). # Matches docker-compose tuning: chDB pool = cores, larger flush batches, faster WAL drain. -apiVersion: hyperbytedb.hyperbytedb.io/v1alpha1 +apiVersion: hyperbytedb.hyperbyte.cloud/v1alpha1 kind: HyperbytedbCluster metadata: name: hyperbytedb @@ -48,6 +48,7 @@ spec: raftHeartbeatIntervalMs: 1000 raftElectionTimeoutMs: 5000 raftSnapshotThreshold: 1000 + drainWaitSecs: 30 replication: mode: async ackTimeoutMs: 5000 diff --git a/deploy/kind/manifests/operator.yaml b/deploy/kind/manifests/operator.yaml index ab13a04..f97c70c 100644 --- a/deploy/kind/manifests/operator.yaml +++ b/deploy/kind/manifests/operator.yaml @@ -28,13 +28,13 @@ rules: - apiGroups: [policy] resources: [poddisruptionbudgets] verbs: [get, list, watch, create, update, patch, delete] -- apiGroups: [hyperbytedb.hyperbytedb.io] +- apiGroups: [hyperbytedb.hyperbyte.cloud] resources: [hyperbytedbclusters, hyperbytedbbackups, hyperbytedbrestores] verbs: [get, list, watch, create, update, patch, delete] -- apiGroups: [hyperbytedb.hyperbytedb.io] +- apiGroups: [hyperbytedb.hyperbyte.cloud] resources: [hyperbytedbclusters/status, hyperbytedbbackups/status, hyperbytedbrestores/status] verbs: [get, update, patch] -- apiGroups: [hyperbytedb.hyperbytedb.io] +- apiGroups: [hyperbytedb.hyperbyte.cloud] resources: [hyperbytedbclusters/finalizers, hyperbytedbbackups/finalizers, hyperbytedbrestores/finalizers] verbs: [update] - apiGroups: [coordination.k8s.io] diff --git a/docs/deep-dive/deep-dive-clustering.md b/docs/deep-dive/deep-dive-clustering.md index 885f8ea..a25bf1f 100644 --- a/docs/deep-dive/deep-dive-clustering.md +++ b/docs/deep-dive/deep-dive-clustering.md @@ -296,7 +296,7 @@ Raft messages are exchanged via HTTP: | Parameter | Default | Description | |-----------|---------|-------------| -| `raft_heartbeat_interval_ms` | 300 | Raft heartbeat interval | +| `raft_heartbeat_interval_ms` | 1000 | Raft heartbeat interval | | `raft_election_timeout_ms` | 1000 | Raft election timeout | | `raft_snapshot_threshold` | 1000 | Log entries before snapshot | @@ -564,7 +564,7 @@ Step 5: Set node state to Leaving | `raft_dir` | `"./raft"` | RocksDB directory for Raft state | | `sync_max_concurrent_files` | `4` | Max concurrent file downloads during sync | | `replication_max_retries` | `5` | Max retries for failed replications | -| `raft_heartbeat_interval_ms` | `300` | Raft heartbeat interval (milliseconds) | +| `raft_heartbeat_interval_ms` | `1000` | Raft heartbeat interval (milliseconds) | | `raft_election_timeout_ms` | `1000` | Raft election timeout (milliseconds) | | `raft_snapshot_threshold` | `1000` | Log entries before Raft snapshot | | `replication.mode` | `"async"` | Coordinator replication mode: `"async"` (default) or `"sync_quorum"` | diff --git a/docs/user-guide/advanced-features.md b/docs/user-guide/advanced-features.md index be9ca89..acfb49d 100644 --- a/docs/user-guide/advanced-features.md +++ b/docs/user-guide/advanced-features.md @@ -105,6 +105,17 @@ CREATE MATERIALIZED VIEW "mv_cpu_1h" ON "mydb" AS SELECT mean("usage_idle") INTO "cpu_1h" FROM "cpu" GROUP BY time(1h), * ``` +By default, `CREATE` installs the destination tables and ClickHouse materialized views only — rollups begin with the next write to the source. Historical backfill is **disabled by default** because it can scan the entire source measurement and take minutes on large datasets. + +To also aggregate existing source history on create, add `WITH BACKFILL`: + +```sql +CREATE MATERIALIZED VIEW "mv_cpu_1h" ON "mydb" WITH BACKFILL +AS SELECT mean("usage_idle") INTO "cpu_1h" FROM "cpu" GROUP BY time(1h), * +``` + +A `WHERE time > ...` clause in the SELECT limits the backfill scan when `WITH BACKFILL` is used. + Requirements (same as `SELECT INTO` / continuous queries): - `INTO` destination measurement is required @@ -116,7 +127,7 @@ On `CREATE`, HyperbyteDB: 1. Registers the destination measurement in metadata 2. Creates destination MergeTree tables in chDB 3. Installs fact and series ClickHouse materialized views -4. Backfills historical data from the source +4. Backfills historical data from the source **only when `WITH BACKFILL` is specified** ### Manage materialized views @@ -133,7 +144,7 @@ DROP MATERIALIZED VIEW "mv_cpu_1h" ON "mydb" |--|------------------|-------------------| | Trigger | 10s scheduler | Each flush to source | | Latency | Up to resample interval | Near real-time | -| Backfill | Re-scans window each run | One-time on CREATE; then incremental | +| Backfill | Re-scans window each run | Opt-in on CREATE (`WITH BACKFILL`); then incremental | | Engine | WAL writeback | ClickHouse MV | --- diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index 624a69b..5f7c974 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -130,7 +130,7 @@ Master-master peer-to-peer clustering with Raft consensus for schema mutations. | `replicate_receiver_queue_depth` | integer | `1024` | Bounded apply queue on the replicate receiver | | `replicate_receiver_workers` | integer | `1` | **Ignored.** Receiver uses a single ordered worker | | `replication_truncate_stale_peer_multiplier` | integer | `2` | When >0, peers with ack 0 and stale heartbeats are omitted from truncate barrier (× heartbeat interval) | -| `raft_heartbeat_interval_ms` | — | *unset* | Optional Raft heartbeat (ms); uses internal default if omitted | +| `raft_heartbeat_interval_ms` | — | *unset* | Optional Raft heartbeat (ms); defaults to 1000 if omitted | | `raft_election_timeout_ms` | — | *unset* | Optional Raft election timeout (ms) | | `raft_snapshot_threshold` | — | *unset* | Optional log entries before Raft snapshot | diff --git a/docs/user-guide/operator/cluster.md b/docs/user-guide/operator/cluster.md index 7f37e91..f15a359 100644 --- a/docs/user-guide/operator/cluster.md +++ b/docs/user-guide/operator/cluster.md @@ -86,7 +86,7 @@ spec: heartbeatIntervalSecs: 1 heartbeatMissThreshold: 3 replicationMaxRetries: 10 - raftHeartbeatIntervalMs: 200 + raftHeartbeatIntervalMs: 1000 raftElectionTimeoutMs: 800 raftSnapshotThreshold: 500 replication: @@ -306,7 +306,7 @@ Tuning parameters for multi-replica cluster behavior. These only take effect whe | `replicationMaxCoalesceBodyBytes` | int64 | `8388608` | Max bytes for coalescing consecutive WAL batches | | `replicateReceiverQueueDepth` | int32 | `1024` | Bounded apply queue on the replicate receiver | | `replicationTruncateStalePeerMultiplier` | int64 | `2` | Omit stale peers from WAL truncate barrier | -| `raftHeartbeatIntervalMs` | int32 | `300` | Raft leader heartbeat interval | +| `raftHeartbeatIntervalMs` | int32 | `1000` | Raft leader heartbeat interval | | `raftElectionTimeoutMs` | int32 | `1000` | Raft election timeout | | `raftSnapshotThreshold` | int32 | `1000` | Log entries before Raft snapshot | | `replication.mode` | string | `async` | `async` or `sync_quorum` | diff --git a/hyperbytedb-proxy/Dockerfile b/hyperbytedb-proxy/Dockerfile index 6e1374d..8366224 100644 --- a/hyperbytedb-proxy/Dockerfile +++ b/hyperbytedb-proxy/Dockerfile @@ -43,6 +43,7 @@ RUN mkdir -p hyperbytedb/benches hyperbytedb/benches/support \ && echo "fn main(){}" > hyperbytedb/benches/ingestion_line_protocol.rs \ && echo "fn main(){}" > hyperbytedb/benches/query_fixed_dataset.rs \ && echo "fn main(){}" > hyperbytedb/benches/flush_service.rs \ + && echo "fn main(){}" > hyperbytedb/benches/ingestion_prepared.rs \ && echo "" > hyperbytedb/benches/support/mod.rs \ && echo "fn main(){}" > hyperbytedb-proxy/benches/routing.rs \ && echo "" > hyperbytedb-proxy/benches/support/mod.rs diff --git a/hyperbytedb-proxy/src/admin.rs b/hyperbytedb-proxy/src/admin.rs index cb9beaa..c63a966 100644 --- a/hyperbytedb-proxy/src/admin.rs +++ b/hyperbytedb-proxy/src/admin.rs @@ -2,17 +2,18 @@ //! itself. //! //! - `GET /healthz` — liveness, always 200 once the process is up. -//! - `GET /readyz` — readiness, 200 only when ≥1 backend is `Active`. +//! - `GET /readyz` — readiness, 200 only when ≥1 backend is routable (Active and not excluded). //! - `GET /metrics` — Prometheus exposition. //! - `GET /admin/backends` — JSON dump of the current pool, for debugging. //! //! Routes are chosen so they can be allowlisted before the catch-all proxy //! handler, with no risk of colliding with a hyperbytedb path. +use std::net::IpAddr; use std::sync::Arc; use axum::Json; -use axum::extract::State; +use axum::extract::{Path, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use serde::Serialize; @@ -32,7 +33,10 @@ pub async fn healthz() -> Response { pub async fn readyz(State(state): State) -> Response { let snap = state.pool.snapshot().await; - let active = snap.iter().filter(|b| b.health() == Health::Active).count(); + let active = snap + .iter() + .filter(|b| b.health() == Health::Active && !b.is_excluded()) + .count(); if active > 0 { ( StatusCode::OK, @@ -87,3 +91,68 @@ pub async fn list_backends(State(state): State) -> Response { .collect(); Json(body).into_response() } + +// --------------------------------------------------------------------------- +// Operator-driven backend exclusion endpoints +// --------------------------------------------------------------------------- + +#[derive(Serialize)] +struct ExcludeResponse { + status: &'static str, + ip: String, +} + +/// `POST /admin/backends/{ip}/exclude` — Tell the proxy to stop routing to +/// this backend. Called by the operator before killing a pod during rolling +/// upgrades. +pub async fn exclude_backend(State(state): State, Path(ip): Path) -> Response { + match state.pool.exclude_backend(ip).await { + Ok(true) => ( + StatusCode::OK, + Json(ExcludeResponse { + status: "excluded", + ip: ip.to_string(), + }), + ) + .into_response(), + Ok(false) => ( + StatusCode::CONFLICT, + Json(ExcludeResponse { + status: "already_excluded", + ip: ip.to_string(), + }), + ) + .into_response(), + Err(e) => ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({"status": "error", "message": e.to_string()})), + ) + .into_response(), + } +} + +/// `POST /admin/backends/{ip}/include` — Clear the exclusion flag so the +/// proxy may route to this backend again. Called by the operator after the +/// replacement pod is healthy. +pub async fn include_backend(State(state): State, Path(ip): Path) -> Response { + let was_excluded = state.pool.include_backend(ip).await; + let status = if was_excluded { + "included" + } else { + "not_excluded" + }; + ( + StatusCode::OK, + Json(ExcludeResponse { + status, + ip: ip.to_string(), + }), + ) + .into_response() +} + +/// `GET /admin/pool` — Full pool status including exclusion flags. +pub async fn pool_status(State(state): State) -> Response { + let statuses = state.pool.pool_status().await; + Json(statuses).into_response() +} diff --git a/hyperbytedb-proxy/src/backend.rs b/hyperbytedb-proxy/src/backend.rs index 08f41c8..e66578b 100644 --- a/hyperbytedb-proxy/src/backend.rs +++ b/hyperbytedb-proxy/src/backend.rs @@ -2,7 +2,7 @@ //! counters are atomic so the routing hot path is lock-free. use std::net::IpAddr; -use std::sync::atomic::{AtomicI64, AtomicU8, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicUsize, Ordering}; use std::time::SystemTime; /// Health classification observed by the most recent probe. @@ -62,6 +62,10 @@ pub struct Backend { /// Consecutive probe failures. Used to log transitions cleanly and (in /// future) to back-off probes for backends that are deeply broken. consecutive_failures: AtomicUsize, + /// Operator-driven exclusion flag. When true, `pick_active` skips this + /// backend entirely — set by the operator before killing a pod during + /// rolling upgrades. + excluded: AtomicBool, } impl Backend { @@ -75,6 +79,7 @@ impl Backend { inflight: AtomicUsize::new(0), last_probe_unix: AtomicI64::new(0), consecutive_failures: AtomicUsize::new(0), + excluded: AtomicBool::new(false), } } @@ -110,6 +115,15 @@ impl Backend { self.inflight.load(Ordering::Relaxed) } + pub fn is_excluded(&self) -> bool { + self.excluded.load(Ordering::Acquire) + } + + /// Set the operator-driven exclusion flag. Returns the previous value. + pub fn set_excluded(&self, val: bool) -> bool { + self.excluded.swap(val, Ordering::AcqRel) + } + /// RAII guard that increments `inflight` on construction and decrements /// on drop, even on panic. Use for the lifetime of one proxied request. pub fn enter(self: &std::sync::Arc) -> InflightGuard { diff --git a/hyperbytedb-proxy/src/lib.rs b/hyperbytedb-proxy/src/lib.rs index 6c6fd4e..530e7a9 100644 --- a/hyperbytedb-proxy/src/lib.rs +++ b/hyperbytedb-proxy/src/lib.rs @@ -14,7 +14,7 @@ use std::sync::Arc; use anyhow::{Context, Result}; use axum::Router; -use axum::routing::{any, get}; +use axum::routing::{any, get, post}; use metrics_exporter_prometheus::PrometheusBuilder; use tokio::net::TcpListener; use tokio::signal::unix::{SignalKind, signal}; @@ -60,6 +60,9 @@ pub async fn run() -> Result<()> { .route("/readyz", get(admin::readyz)) .route("/metrics", get(admin::metrics_endpoint)) .route("/admin/backends", get(admin::list_backends)) + .route("/admin/backends/{ip}/exclude", post(admin::exclude_backend)) + .route("/admin/backends/{ip}/include", post(admin::include_backend)) + .route("/admin/pool", get(admin::pool_status)) .with_state(admin_state); // Order matters: admin routes first, then the catch-all proxy fallback. diff --git a/hyperbytedb-proxy/src/pool.rs b/hyperbytedb-proxy/src/pool.rs index c963ad6..d93b13e 100644 --- a/hyperbytedb-proxy/src/pool.rs +++ b/hyperbytedb-proxy/src/pool.rs @@ -6,19 +6,32 @@ //! immediately, and walks the snapshot. Discovery and health probing are //! background tasks that mutate the pool atomically (publish-via-replace). -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::net::IpAddr; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Duration, Instant}; use anyhow::{Context, Result}; +use serde::Serialize; use tokio::sync::{Notify, RwLock}; use tokio::time::{MissedTickBehavior, interval, timeout}; use crate::backend::{Backend, Health}; use crate::config::ProxyConfig; +/// JSON-serializable backend status for the admin pool endpoint. +#[derive(Serialize)] +pub struct BackendStatus { + pub addr: String, + pub port: u16, + pub health: String, + pub excluded: bool, + pub inflight: usize, + pub consecutive_failures: usize, + pub last_probe_unix: i64, +} + pub struct BackendPool { cfg: ProxyConfig, @@ -26,6 +39,13 @@ pub struct BackendPool { /// IPs; mutated in place (atomically per-backend) when health changes. backends: RwLock>>>, + /// Operator-driven exclusion set. IPs in this set are never routed to + /// even when their health is `Active`. The operator populates this via + /// `POST /admin/backends/{ip}/exclude` before killing a pod during + /// rolling upgrades, and clears it via `POST /admin/backends/{ip}/include` + /// once the replacement pod is healthy. + excluded: RwLock>, + /// Round-robin cursor. Wraps via modulo at pick-time. cursor: AtomicUsize, @@ -49,6 +69,7 @@ impl BackendPool { Ok(Arc::new(Self { cfg, backends: RwLock::new(Arc::new(Vec::new())), + excluded: RwLock::new(HashSet::new()), cursor: AtomicUsize::new(0), probe_client, on_active: Notify::new(), @@ -72,18 +93,66 @@ impl BackendPool { Arc::clone(&*self.backends.read().await) } - /// Round-robin pick over backends in `Active` state. Returns `None` if - /// the pool currently has zero active backends — the caller decides - /// whether to hold-and-retry or fail fast. + /// Operator-driven exclusion: mark a backend IP so `pick_active` never + /// routes to it. Returns `true` if newly excluded, or `Err` if the IP + /// is not in the pool. + pub async fn exclude_backend(&self, ip: IpAddr) -> anyhow::Result { + let snap = self.snapshot().await; + let backend = snap + .iter() + .find(|b| b.addr == ip) + .ok_or_else(|| anyhow::anyhow!("backend {ip} not found in pool"))?; + backend.set_excluded(true); + let mut guard = self.excluded.write().await; + Ok(guard.insert(ip)) + } + + /// Operator-driven inclusion: clear the exclusion flag so `pick_active` + /// may route to this backend again. Returns `true` if it was previously + /// excluded. + pub async fn include_backend(&self, ip: IpAddr) -> bool { + let snap = self.snapshot().await; + if let Some(backend) = snap.iter().find(|b| b.addr == ip) { + backend.set_excluded(false); + } + let mut guard = self.excluded.write().await; + guard.remove(&ip) + } + + /// Returns true if the given IP is currently excluded. + pub async fn is_excluded(&self, ip: &IpAddr) -> bool { + self.excluded.read().await.contains(ip) + } + + /// JSON-serializable snapshot of the pool for `GET /admin/pool`. + pub async fn pool_status(&self) -> Vec { + let snap = self.snapshot().await; + let excluded = self.excluded.read().await; + snap.iter() + .map(|b| BackendStatus { + addr: b.addr.to_string(), + port: b.port, + health: b.health().as_str().to_string(), + excluded: excluded.contains(&b.addr), + inflight: b.inflight(), + consecutive_failures: b.consecutive_failures(), + last_probe_unix: b.last_probe_unix(), + }) + .collect() + } + + /// Round-robin pick over backends in `Active` state that are not + /// excluded. Returns `None` if the pool currently has zero routable + /// backends — the caller decides whether to hold-and-retry or fail fast. pub async fn pick_active(&self) -> Option> { let snap = self.snapshot().await; if snap.is_empty() { return None; } - // Filter by health into a tight Vec; small (cluster size <= dozens). + // Filter by health and exclusion into a tight Vec; small (cluster size <= dozens). let active: Vec<&Arc> = snap .iter() - .filter(|b| b.health() == Health::Active) + .filter(|b| b.health() == Health::Active && !b.is_excluded()) .collect(); if active.is_empty() { return None; @@ -92,13 +161,16 @@ impl BackendPool { Some(Arc::clone(active[idx])) } - /// Pick an active backend that isn't `exclude`. Used by the retry loop so - /// we don't re-try the same broken backend twice in a row. + /// Pick an active backend that isn't `exclude` and isn't excluded by the + /// operator. Used by the retry loop so we don't re-try the same broken + /// backend twice in a row. pub async fn pick_active_excluding(&self, exclude: &Arc) -> Option> { let snap = self.snapshot().await; let active: Vec<&Arc> = snap .iter() - .filter(|b| b.health() == Health::Active && !Arc::ptr_eq(b, exclude)) + .filter(|b| { + b.health() == Health::Active && !Arc::ptr_eq(b, exclude) && !b.is_excluded() + }) .collect(); if active.is_empty() { return None; @@ -222,8 +294,17 @@ impl BackendPool { // Atomic publish. let new_snap = Arc::new(next); - let mut guard = self.backends.write().await; - *guard = new_snap; + { + let mut guard = self.backends.write().await; + *guard = new_snap; + } + + // Garbage-collect exclusion entries for IPs that are no longer in the pool. + if removed > 0 { + let mut excl = self.excluded.write().await; + let fresh_set: HashSet = fresh_ips.into_iter().collect(); + excl.retain(|ip| fresh_set.contains(ip)); + } } async fn probe_one(&self, backend: &Arc) { diff --git a/hyperbytedb/src/application/cluster/bootstrap.rs b/hyperbytedb/src/application/cluster/bootstrap.rs index 3f97871..2cb1ed6 100644 --- a/hyperbytedb/src/application/cluster/bootstrap.rs +++ b/hyperbytedb/src/application/cluster/bootstrap.rs @@ -263,7 +263,7 @@ impl ClusterBootstrap { let network = Network::new(); let raft_config = match (RaftConfig { - heartbeat_interval: config.raft_heartbeat_interval_ms.unwrap_or(300), + heartbeat_interval: config.raft_heartbeat_interval_ms.unwrap_or(1000), election_timeout_min: config.raft_election_timeout_ms.unwrap_or(1000), election_timeout_max: config.raft_election_timeout_ms.unwrap_or(1000) * 2, snapshot_policy: openraft::SnapshotPolicy::LogsSinceLast( diff --git a/hyperbytedb/src/application/materialized_view_service.rs b/hyperbytedb/src/application/materialized_view_service.rs index 0a0b5da..6daacc0 100644 --- a/hyperbytedb/src/application/materialized_view_service.rs +++ b/hyperbytedb/src/application/materialized_view_service.rs @@ -52,7 +52,8 @@ impl MaterializedViewService { ))); } - self.materialize_ddl(mv, true).await?; + self.materialize_ddl(mv, true, mv.backfill_on_create) + .await?; let (source_db, source_rp_opt, source_measurement) = extract_source(&mv.query, &mv.database)?; @@ -87,6 +88,7 @@ impl MaterializedViewService { ch_fact_mv_name: unquoted_fact_mv_name(&mv.database, &dest_rp, &mv.name), ch_series_mv_name: unquoted_series_mv_name(&mv.database, &dest_rp, &mv.name), created_at: chrono::Utc::now().to_rfc3339(), + backfill_on_create: mv.backfill_on_create, }; self.metadata @@ -98,6 +100,7 @@ impl MaterializedViewService { db = %mv.database, source = %def.source_measurement, dest = %def.dest_measurement, + backfill = mv.backfill_on_create, "materialized view created" ); @@ -221,6 +224,7 @@ impl MaterializedViewService { database: definition.database.clone(), query: parse_mv_select(&definition.query_text)?, raw_query: definition.query_text.clone(), + backfill_on_create: definition.backfill_on_create, }; self.create(&stmt).await?; Ok(()) @@ -269,8 +273,9 @@ impl MaterializedViewService { database: def.database.clone(), query: parse_mv_select(&def.query_text)?, raw_query: def.query_text.clone(), + backfill_on_create: false, }; - self.materialize_ddl(&stmt, false).await?; + self.materialize_ddl(&stmt, false, false).await?; tracing::info!( mv = %def.name, db = %def.database, @@ -283,6 +288,7 @@ impl MaterializedViewService { &self, mv: &CreateMaterializedViewStatement, reset_destination: bool, + backfill_on_create: bool, ) -> Result<(), HyperbytedbError> { let (source_db, source_rp_opt, source_measurement) = extract_source(&mv.query, &mv.database)?; @@ -406,10 +412,19 @@ impl MaterializedViewService { self.query_port.execute_sql(&create_fact_mv).await?; self.query_port.execute_sql(&create_series_mv).await?; - self.query_port.execute_sql(&backfill_fact).await?; - let backfill_series = format!("INSERT INTO {dest_series}\n{series_select}"); - self.query_port.execute_sql(&backfill_series).await?; + if backfill_on_create { + self.query_port.execute_sql(&backfill_fact).await?; + + let backfill_series = format!("INSERT INTO {dest_series}\n{series_select}"); + self.query_port.execute_sql(&backfill_series).await?; + } else { + tracing::info!( + mv = %mv.name, + db = %mv.database, + "skipping materialized view historical backfill (use WITH BACKFILL to enable)" + ); + } self.metadata .register_measurement(&dest_db, &dest_rp, &dest_meta) @@ -470,6 +485,7 @@ pub fn def_from_statement( ch_fact_mv_name: unquoted_fact_mv_name(&mv.database, dest_rp, &mv.name), ch_series_mv_name: unquoted_series_mv_name(&mv.database, dest_rp, &mv.name), created_at: chrono::Utc::now().to_rfc3339(), + backfill_on_create: mv.backfill_on_create, }) } diff --git a/hyperbytedb/src/config.rs b/hyperbytedb/src/config.rs index c448124..8778ff4 100644 --- a/hyperbytedb/src/config.rs +++ b/hyperbytedb/src/config.rs @@ -193,7 +193,7 @@ pub struct ClusterConfig { /// When >0, peers with ack 0 and stale heartbeats are omitted from truncate barrier. #[serde(default = "default_replication_truncate_stale_peer_multiplier")] pub replication_truncate_stale_peer_multiplier: u64, - /// Raft heartbeat interval in milliseconds (default: 300). + /// Raft heartbeat interval in milliseconds (default: 1000). pub raft_heartbeat_interval_ms: Option, /// Raft election timeout in milliseconds (default: 1000). pub raft_election_timeout_ms: Option, diff --git a/hyperbytedb/src/domain/materialized_view.rs b/hyperbytedb/src/domain/materialized_view.rs index 8e39912..e698508 100644 --- a/hyperbytedb/src/domain/materialized_view.rs +++ b/hyperbytedb/src/domain/materialized_view.rs @@ -14,4 +14,7 @@ pub struct MaterializedViewDef { pub ch_fact_mv_name: String, pub ch_series_mv_name: String, pub created_at: String, + /// Whether CREATE ran a one-time historical backfill (`WITH BACKFILL`). + #[serde(default)] + pub backfill_on_create: bool, } diff --git a/hyperbytedb/src/timeseriesql/ast.rs b/hyperbytedb/src/timeseriesql/ast.rs index cac8297..fedcf81 100644 --- a/hyperbytedb/src/timeseriesql/ast.rs +++ b/hyperbytedb/src/timeseriesql/ast.rs @@ -448,6 +448,8 @@ pub struct CreateMaterializedViewStatement { pub database: String, pub query: SelectStatement, pub raw_query: String, + /// When true, run a one-time historical backfill on CREATE (`WITH BACKFILL`). + pub backfill_on_create: bool, } #[cfg(test)] diff --git a/hyperbytedb/src/timeseriesql/ddl_parser.rs b/hyperbytedb/src/timeseriesql/ddl_parser.rs index 958cd2b..596fe29 100644 --- a/hyperbytedb/src/timeseriesql/ddl_parser.rs +++ b/hyperbytedb/src/timeseriesql/ddl_parser.rs @@ -677,6 +677,12 @@ fn parse_create_materialized_view( ) -> Result { let name = cur.take_ident()?; let database = parse_on_db(cur)?; + let backfill_on_create = if cur.match_keyword("WITH") { + cur.expect_keyword("BACKFILL")?; + true + } else { + false + }; let (raw_query, select_stmt) = if cur.match_keyword("AS") { let start = cur.peek().map(|t| t.start).unwrap_or(0); let inner = cur.input[start..].trim(); @@ -700,6 +706,7 @@ fn parse_create_materialized_view( database, query: select_stmt, raw_query, + backfill_on_create, }, )) } @@ -1252,4 +1259,28 @@ mod tests { ); assert!(stmt.is_ok(), "AS-form MV must parse: {stmt:?}"); } + + #[test] + fn create_materialized_view_with_backfill_parses() { + let stmt = parse_ddl_statement( + r#"CREATE MATERIALIZED VIEW mv ON db WITH BACKFILL AS SELECT mean("v") FROM m GROUP BY time(5m)"#, + ) + .unwrap(); + match stmt { + Statement::CreateMaterializedView(mv) => assert!(mv.backfill_on_create), + other => panic!("expected CreateMaterializedView, got {other:?}"), + } + } + + #[test] + fn create_materialized_view_without_backfill_defaults_false() { + let stmt = parse_ddl_statement( + r#"CREATE MATERIALIZED VIEW mv ON db AS SELECT mean("v") FROM m GROUP BY time(5m)"#, + ) + .unwrap(); + match stmt { + Statement::CreateMaterializedView(mv) => assert!(!mv.backfill_on_create), + other => panic!("expected CreateMaterializedView, got {other:?}"), + } + } } diff --git a/hyperbytedb/src/timeseriesql/digest.rs b/hyperbytedb/src/timeseriesql/digest.rs index c09078c..875e7b4 100644 --- a/hyperbytedb/src/timeseriesql/digest.rs +++ b/hyperbytedb/src/timeseriesql/digest.rs @@ -199,6 +199,9 @@ fn normalize_statement(stmt: &Statement) -> String { mv.name, mv.database ) .ok(); + if mv.backfill_on_create { + out.push_str(" with backfill"); + } } Statement::DropMaterializedView { name, db } => { write!(out, "drop materialized view {} on {}", name, db).ok(); @@ -574,4 +577,27 @@ mod tests { assert!(!redacted.contains("s3cret")); assert!(redacted.contains("****")); } + + #[test] + fn normalize_create_mv_without_backfill() { + let stmt = crate::timeseriesql::parse( + r#"CREATE MATERIALIZED VIEW "mv" ON "db" AS SELECT mean("v") FROM "m" GROUP BY time(1m)"#, + ) + .unwrap() + .remove(0); + let (_, norm) = fingerprint(&stmt); + assert_eq!(norm, "create materialized view mv on db"); + assert!(!norm.contains("with backfill")); + } + + #[test] + fn normalize_create_mv_with_backfill() { + let stmt = crate::timeseriesql::parse( + r#"CREATE MATERIALIZED VIEW "mv" ON "db" WITH BACKFILL AS SELECT mean("v") FROM "m" GROUP BY time(1m)"#, + ) + .unwrap() + .remove(0); + let (_, norm) = fingerprint(&stmt); + assert_eq!(norm, "create materialized view mv on db with backfill"); + } } diff --git a/hyperbytedb/src/timeseriesql/lexer.rs b/hyperbytedb/src/timeseriesql/lexer.rs index deed52a..41f53bd 100644 --- a/hyperbytedb/src/timeseriesql/lexer.rs +++ b/hyperbytedb/src/timeseriesql/lexer.rs @@ -861,6 +861,7 @@ fn is_keyword(word: &str) -> bool { | "BEGIN" | "END" | "RESAMPLE" + | "BACKFILL" | "EVERY" | "FOR" | "KEY" diff --git a/hyperbytedb/src/timeseriesql/parser.rs b/hyperbytedb/src/timeseriesql/parser.rs index 902c611..21daf35 100644 --- a/hyperbytedb/src/timeseriesql/parser.rs +++ b/hyperbytedb/src/timeseriesql/parser.rs @@ -1349,6 +1349,7 @@ mod tests { Statement::CreateMaterializedView(mv) => { assert_eq!(mv.name, "mv_5m"); assert_eq!(mv.database, "mydb"); + assert!(!mv.backfill_on_create); assert!(mv.query.into.is_some()); assert!(mv.query.group_by.is_some()); } @@ -1356,6 +1357,55 @@ mod tests { } } + #[test] + fn test_parse_create_materialized_view_with_backfill() { + let q = r#"CREATE MATERIALIZED VIEW "mv_5m" ON "mydb" WITH BACKFILL AS SELECT mean("value") INTO "cpu_5m" FROM "cpu" GROUP BY time(5m), *"#; + let stmts = parse_query(q).unwrap(); + match &stmts[0] { + Statement::CreateMaterializedView(mv) => { + assert_eq!(mv.name, "mv_5m"); + assert_eq!(mv.database, "mydb"); + assert!(mv.backfill_on_create); + } + _ => panic!("expected CREATE MATERIALIZED VIEW"), + } + } + + #[test] + fn test_parse_create_materialized_view_with_without_backfill_keyword_fails() { + let q = r#"CREATE MATERIALIZED VIEW "mv" ON "mydb" WITH AS SELECT mean("value") INTO "cpu_5m" FROM "cpu" GROUP BY time(5m), *"#; + assert!( + parse_query(q).is_err(), + "WITH without BACKFILL should be a parse error" + ); + } + + #[test] + fn test_parse_create_materialized_view_with_backfill_begin_syntax() { + let q = r#"CREATE MATERIALIZED VIEW "mv_1h" ON "mydb" WITH BACKFILL BEGIN SELECT mean("value") INTO "cpu_1h" FROM "cpu" GROUP BY time(1h), * END"#; + let stmts = parse_query(q).unwrap(); + match &stmts[0] { + Statement::CreateMaterializedView(mv) => { + assert_eq!(mv.name, "mv_1h"); + assert!(mv.backfill_on_create); + assert!(mv.query.group_by.is_some()); + } + _ => panic!("expected CREATE MATERIALIZED VIEW"), + } + } + + #[test] + fn test_parse_create_materialized_view_begin_syntax_without_backfill() { + let q = r#"CREATE MATERIALIZED VIEW "mv_1h" ON "mydb" BEGIN SELECT mean("value") INTO "cpu_1h" FROM "cpu" GROUP BY time(1h), * END"#; + let stmts = parse_query(q).unwrap(); + match &stmts[0] { + Statement::CreateMaterializedView(mv) => { + assert!(!mv.backfill_on_create); + } + _ => panic!("expected CREATE MATERIALIZED VIEW"), + } + } + #[test] fn test_parse_create_materialized_view_requires_group_by_time() { let q = r#"CREATE MATERIALIZED VIEW "mv" ON "mydb" AS SELECT mean("value") INTO "cpu_5m" FROM "cpu""#; diff --git a/hyperbytedb/tests/compat/ddl_tests.rs b/hyperbytedb/tests/compat/ddl_tests.rs index dd58e31..e67de9e 100644 --- a/hyperbytedb/tests/compat/ddl_tests.rs +++ b/hyperbytedb/tests/compat/ddl_tests.rs @@ -13,6 +13,20 @@ use super::TestContext; /// Epoch nanoseconds on a 1-minute boundary (matches MV `toStartOfInterval` bucket keys). const MV_MINUTE_ALIGNED_NS: i64 = 1_700_000_040_000_000_000; +async fn mv_fact_row_count(ctx: &TestContext, table: &str) -> u64 { + ctx.query_port + .execute_sql(&format!( + "SELECT count() AS c FROM `{table}` FORMAT JSONEachRow" + )) + .await + .unwrap() + .lines() + .next() + .and_then(|l| serde_json::from_str::(l).ok()) + .and_then(|v| v.get("c").and_then(|c| c.as_u64())) + .unwrap_or(0) +} + // --------------------------------------------------------------------------- // CREATE / DROP DATABASE // --------------------------------------------------------------------------- @@ -504,7 +518,7 @@ async fn materialized_view_sum_survives_multiple_flushes() { let create_resp = ctx .query( "mvdb", - r#"CREATE MATERIALIZED VIEW "mv_metrics_sum" ON "mvdb" AS SELECT sum("value") AS "value" INTO "metrics_1m" FROM "metrics" GROUP BY time(1m), "host""#, + r#"CREATE MATERIALIZED VIEW "mv_metrics_sum" ON "mvdb" WITH BACKFILL AS SELECT sum("value") AS "value" INTO "metrics_1m" FROM "metrics" GROUP BY time(1m), "host""#, ) .await .unwrap(); @@ -606,7 +620,7 @@ async fn materialized_view_dest_uses_summing_merge_tree() { let create_resp = ctx .query( "mvdb", - r#"CREATE MATERIALIZED VIEW "mv_metrics_sum" ON "mvdb" AS SELECT sum("value") AS "value" INTO "metrics_1m" FROM "metrics" GROUP BY time(1m), "host""#, + r#"CREATE MATERIALIZED VIEW "mv_metrics_sum" ON "mvdb" WITH BACKFILL AS SELECT sum("value") AS "value" INTO "metrics_1m" FROM "metrics" GROUP BY time(1m), "host""#, ) .await .unwrap(); @@ -654,7 +668,7 @@ async fn materialized_view_sum_matches_raw_after_many_single_point_flushes() { let create_resp = ctx .query( "mvdb", - r#"CREATE MATERIALIZED VIEW "mv_metrics_many" ON "mvdb" AS SELECT sum("value") AS "value" INTO "metrics_many" FROM "metrics" GROUP BY time(1m), "host""#, + r#"CREATE MATERIALIZED VIEW "mv_metrics_many" ON "mvdb" WITH BACKFILL AS SELECT sum("value") AS "value" INTO "metrics_many" FROM "metrics" GROUP BY time(1m), "host""#, ) .await .unwrap(); @@ -745,7 +759,7 @@ async fn materialized_view_mean_survives_multiple_flushes() { let create_resp = ctx .query( "mvdb", - r#"CREATE MATERIALIZED VIEW "mv_cpu_mean" ON "mvdb" AS SELECT mean("value") INTO "cpu_1m" FROM "cpu" GROUP BY time(1m), "host""#, + r#"CREATE MATERIALIZED VIEW "mv_cpu_mean" ON "mvdb" WITH BACKFILL AS SELECT mean("value") INTO "cpu_1m" FROM "cpu" GROUP BY time(1m), "host""#, ) .await .unwrap(); @@ -846,7 +860,7 @@ async fn materialized_view_sum_dedupes_duplicate_source_rows() { let create_resp = ctx .query( "mvdb", - r#"CREATE MATERIALIZED VIEW "mv_metrics_dedup" ON "mvdb" AS SELECT sum("value") AS "value" INTO "metrics_dedup" FROM "metrics" GROUP BY time(1m), "host""#, + r#"CREATE MATERIALIZED VIEW "mv_metrics_dedup" ON "mvdb" WITH BACKFILL AS SELECT sum("value") AS "value" INTO "metrics_dedup" FROM "metrics" GROUP BY time(1m), "host""#, ) .await .unwrap(); @@ -959,6 +973,435 @@ async fn materialized_view_dest_in_different_rp_poisons_same_name_in_autogen() { ); } +#[tokio::test] +#[serial(chdb)] +async fn create_materialized_view_without_backfill_leaves_dest_empty_until_new_writes() { + let ctx = match TestContext::new() { + Ok(c) => c, + Err(_) => { + eprintln!("skipping MV no-backfill test: chDB not available"); + return; + } + }; + + ctx.metadata.create_database("mvdb").await.unwrap(); + + let t = MV_MINUTE_ALIGNED_NS; + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=10 {t}")) + .await + .unwrap(); + + let create_resp = ctx + .query( + "mvdb", + r#"CREATE MATERIALIZED VIEW "mv_no_backfill" ON "mvdb" AS SELECT sum("value") AS "value" INTO "metrics_nobf" FROM "metrics" GROUP BY time(1m), "host""#, + ) + .await + .unwrap(); + assert!( + create_resp.results[0].error.is_none(), + "create MV failed: {:?}", + create_resp.results[0].error + ); + + let dest_count: u64 = mv_fact_row_count(&ctx, "mvdb_autogen_metrics_nobf").await; + assert_eq!( + dest_count, 0, + "dest fact table should be empty before any post-create writes" + ); + + let t2 = t + 1_000_000_000; + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=5 {t2}")) + .await + .unwrap(); + + let dest_resp = ctx + .query( + "mvdb", + &format!(r#"SELECT sum("value") FROM "metrics_nobf" WHERE time = {t}"#), + ) + .await + .unwrap(); + assert!( + dest_resp.results[0].error.is_none(), + "query dest failed: {:?}", + dest_resp.results[0].error + ); + let dest_sum: f64 = dest_resp.results[0] + .series + .as_ref() + .unwrap() + .first() + .and_then(|s| s.values.first()) + .and_then(|row| row.last()) + .and_then(|v| v.as_f64()) + .unwrap_or(0.0); + assert!( + (dest_sum - 5.0).abs() < 0.01, + "only post-create writes should appear in dest, got {dest_sum}" + ); +} + +#[tokio::test] +#[serial(chdb)] +async fn create_materialized_view_with_backfill_populates_history() { + let ctx = match TestContext::new() { + Ok(c) => c, + Err(_) => { + eprintln!("skipping MV with-backfill test: chDB not available"); + return; + } + }; + + ctx.metadata.create_database("mvdb").await.unwrap(); + + let t = MV_MINUTE_ALIGNED_NS; + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=10 {t}")) + .await + .unwrap(); + + let create_resp = ctx + .query( + "mvdb", + r#"CREATE MATERIALIZED VIEW "mv_with_backfill" ON "mvdb" WITH BACKFILL AS SELECT sum("value") AS "value" INTO "metrics_bf" FROM "metrics" GROUP BY time(1m), "host""#, + ) + .await + .unwrap(); + assert!( + create_resp.results[0].error.is_none(), + "create MV failed: {:?}", + create_resp.results[0].error + ); + + let mvs = ctx.metadata.list_materialized_views("mvdb").await.unwrap(); + let mv_def = mvs.iter().find(|m| m.name == "mv_with_backfill").unwrap(); + assert!(mv_def.backfill_on_create); + + let dest_resp = ctx + .query( + "mvdb", + &format!(r#"SELECT sum("value") FROM "metrics_bf" WHERE time = {t}"#), + ) + .await + .unwrap(); + assert!( + dest_resp.results[0].error.is_none(), + "query dest failed: {:?}", + dest_resp.results[0].error + ); + let dest_sum: f64 = dest_resp.results[0] + .series + .as_ref() + .unwrap() + .first() + .and_then(|s| s.values.first()) + .and_then(|row| row.last()) + .and_then(|v| v.as_f64()) + .unwrap_or(0.0); + assert!( + (dest_sum - 10.0).abs() < 0.01, + "WITH BACKFILL should populate historical data, got {dest_sum}" + ); +} + +#[tokio::test] +#[serial(chdb)] +async fn create_materialized_view_without_backfill_records_metadata_flag() { + let ctx = match TestContext::new() { + Ok(c) => c, + Err(_) => { + eprintln!("skipping MV metadata flag test: chDB not available"); + return; + } + }; + + ctx.metadata.create_database("mvdb").await.unwrap(); + + let t = MV_MINUTE_ALIGNED_NS; + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=1 {t}")) + .await + .unwrap(); + + let create_resp = ctx + .query( + "mvdb", + r#"CREATE MATERIALIZED VIEW "mv_flag_off" ON "mvdb" AS SELECT sum("value") AS "value" INTO "metrics_flag" FROM "metrics" GROUP BY time(1m), "host""#, + ) + .await + .unwrap(); + assert!( + create_resp.results[0].error.is_none(), + "create MV failed: {:?}", + create_resp.results[0].error + ); + + let mvs = ctx.metadata.list_materialized_views("mvdb").await.unwrap(); + let mv_def = mvs.iter().find(|m| m.name == "mv_flag_off").unwrap(); + assert!( + !mv_def.backfill_on_create, + "default CREATE should persist backfill_on_create=false" + ); +} + +#[tokio::test] +#[serial(chdb)] +async fn create_materialized_view_without_backfill_on_empty_source() { + let ctx = match TestContext::new() { + Ok(c) => c, + Err(_) => { + eprintln!("skipping MV empty-source no-backfill test: chDB not available"); + return; + } + }; + + ctx.metadata.create_database("mvdb").await.unwrap(); + + let t = MV_MINUTE_ALIGNED_NS; + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=1 {t}")) + .await + .unwrap(); + let delete_resp = ctx + .query("mvdb", &format!("DELETE FROM metrics WHERE time <= {t}")) + .await + .unwrap(); + assert!( + delete_resp.results[0].error.is_none(), + "delete failed: {:?}", + delete_resp.results[0].error + ); + + let create_resp = ctx + .query( + "mvdb", + r#"CREATE MATERIALIZED VIEW "mv_empty_src" ON "mvdb" AS SELECT sum("value") AS "value" INTO "metrics_empty" FROM "metrics" GROUP BY time(1m), "host""#, + ) + .await + .unwrap(); + assert!( + create_resp.results[0].error.is_none(), + "create MV failed: {:?}", + create_resp.results[0].error + ); + + assert_eq!( + mv_fact_row_count(&ctx, "mvdb_autogen_metrics_empty").await, + 0, + "dest should stay empty when source has no data and backfill is off" + ); +} + +#[tokio::test] +#[serial(chdb)] +async fn create_materialized_view_without_vs_with_backfill_on_same_source() { + let ctx = match TestContext::new() { + Ok(c) => c, + Err(_) => { + eprintln!("skipping MV without-vs-with test: chDB not available"); + return; + } + }; + + ctx.metadata.create_database("mvdb").await.unwrap(); + + let t = MV_MINUTE_ALIGNED_NS; + ctx.write_and_flush( + "mvdb", + &format!("metrics,host=h1 value=10 {t}\nmetrics,host=h2 value=20 {t}"), + ) + .await + .unwrap(); + + let no_bf = ctx + .query( + "mvdb", + r#"CREATE MATERIALIZED VIEW "mv_cmp_no_bf" ON "mvdb" AS SELECT sum("value") AS "value" INTO "metrics_cmp_nobf" FROM "metrics" GROUP BY time(1m), "host""#, + ) + .await + .unwrap(); + assert!( + no_bf.results[0].error.is_none(), + "create without backfill failed: {:?}", + no_bf.results[0].error + ); + + let with_bf = ctx + .query( + "mvdb", + r#"CREATE MATERIALIZED VIEW "mv_cmp_with_bf" ON "mvdb" WITH BACKFILL AS SELECT sum("value") AS "value" INTO "metrics_cmp_bf" FROM "metrics" GROUP BY time(1m), "host""#, + ) + .await + .unwrap(); + assert!( + with_bf.results[0].error.is_none(), + "create with backfill failed: {:?}", + with_bf.results[0].error + ); + + assert_eq!( + mv_fact_row_count(&ctx, "mvdb_autogen_metrics_cmp_nobf").await, + 0, + "without-backfill dest should be empty before new writes" + ); + assert!( + mv_fact_row_count(&ctx, "mvdb_autogen_metrics_cmp_bf").await >= 2, + "with-backfill dest should contain rolled-up rows for each host" + ); + + let bf_total_resp = ctx + .query( + "mvdb", + &format!(r#"SELECT sum("value") FROM "metrics_cmp_bf" WHERE time = {t}"#), + ) + .await + .unwrap(); + assert!(bf_total_resp.results[0].error.is_none()); + let bf_total: f64 = bf_total_resp.results[0] + .series + .as_ref() + .unwrap() + .iter() + .filter_map(|s| s.values.first()?.last()?.as_f64()) + .sum(); + assert!( + (bf_total - 30.0).abs() < 0.01, + "with-backfill dest should include all pre-create source data, got {bf_total}" + ); +} + +#[tokio::test] +#[serial(chdb)] +async fn create_materialized_view_with_backfill_respects_where_time() { + let ctx = match TestContext::new() { + Ok(c) => c, + Err(_) => { + eprintln!("skipping MV bounded backfill test: chDB not available"); + return; + } + }; + + ctx.metadata.create_database("mvdb").await.unwrap(); + + let t_old = MV_MINUTE_ALIGNED_NS; + let t_new = t_old + 3_600_000_000_000; // +1h, distinct minute bucket + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=100 {t_old}")) + .await + .unwrap(); + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=5 {t_new}")) + .await + .unwrap(); + + let create_resp = ctx + .query( + "mvdb", + &format!( + r#"CREATE MATERIALIZED VIEW "mv_bounded_bf" ON "mvdb" WITH BACKFILL AS SELECT sum("value") AS "value" INTO "metrics_bounded" FROM "metrics" WHERE time >= {t_new} GROUP BY time(1m), "host""# + ), + ) + .await + .unwrap(); + assert!( + create_resp.results[0].error.is_none(), + "create MV failed: {:?}", + create_resp.results[0].error + ); + + let old_count: u64 = ctx + .query_port + .execute_sql("SELECT count() AS c FROM `mvdb_autogen_metrics_bounded` FORMAT JSONEachRow") + .await + .unwrap() + .lines() + .next() + .and_then(|l| serde_json::from_str::(l).ok()) + .and_then(|v| v.get("c").and_then(|c| c.as_u64())) + .unwrap_or(0); + assert_eq!( + old_count, 1, + "bounded backfill should only materialize the recent bucket" + ); + + let new_resp = ctx + .query( + "mvdb", + &format!(r#"SELECT sum("value") FROM "metrics_bounded" WHERE time = {t_new}"#), + ) + .await + .unwrap(); + assert!(new_resp.results[0].error.is_none()); + let new_sum: f64 = new_resp.results[0] + .series + .as_ref() + .unwrap() + .first() + .and_then(|s| s.values.first()) + .and_then(|row| row.last()) + .and_then(|v| v.as_f64()) + .unwrap_or(0.0); + assert!( + (new_sum - 5.0).abs() < 0.01, + "bounded backfill should include matching recent bucket, got {new_sum}" + ); +} + +#[tokio::test] +#[serial(chdb)] +async fn create_materialized_view_with_backfill_then_incremental_write() { + let ctx = match TestContext::new() { + Ok(c) => c, + Err(_) => { + eprintln!("skipping MV backfill plus incremental test: chDB not available"); + return; + } + }; + + ctx.metadata.create_database("mvdb").await.unwrap(); + + let t = MV_MINUTE_ALIGNED_NS; + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=10 {t}")) + .await + .unwrap(); + + let create_resp = ctx + .query( + "mvdb", + r#"CREATE MATERIALIZED VIEW "mv_bf_incr" ON "mvdb" WITH BACKFILL AS SELECT sum("value") AS "value" INTO "metrics_bf_incr" FROM "metrics" GROUP BY time(1m), "host""#, + ) + .await + .unwrap(); + assert!( + create_resp.results[0].error.is_none(), + "create MV failed: {:?}", + create_resp.results[0].error + ); + + let t2 = t + 1_000_000_000; + ctx.write_and_flush("mvdb", &format!("metrics,host=h1 value=5 {t2}")) + .await + .unwrap(); + + let dest_resp = ctx + .query( + "mvdb", + &format!(r#"SELECT sum("value") FROM "metrics_bf_incr" WHERE time = {t}"#), + ) + .await + .unwrap(); + assert!(dest_resp.results[0].error.is_none()); + let dest_sum: f64 = dest_resp.results[0] + .series + .as_ref() + .unwrap() + .first() + .and_then(|s| s.values.first()) + .and_then(|row| row.last()) + .and_then(|v| v.as_f64()) + .unwrap_or(0.0); + assert!( + (dest_sum - 15.0).abs() < 0.01, + "backfill plus incremental flush should accumulate in the same bucket, got {dest_sum}" + ); +} + #[tokio::test] async fn drop_materialized_view() { let ctx = TestContext::new_no_chdb().unwrap(); @@ -981,6 +1424,7 @@ async fn drop_materialized_view() { ch_fact_mv_name: "testdb_autogen_mv_drop_mv".to_string(), ch_series_mv_name: "testdb_autogen_mv_drop_series_mv".to_string(), created_at: chrono::Utc::now().to_rfc3339(), + backfill_on_create: false, }, ) .await