From 7ea964373bf2c1ead30ca5b5548e1be95bb32c5f Mon Sep 17 00:00:00 2001 From: Austin Barrington Date: Tue, 3 Nov 2026 21:57:32 +0000 Subject: [PATCH] Fix four bugs from the 0.9.4 review: query-engine crash, name collisions, DELETE, parser panics - A WHERE clause with a few thousand OR terms overflowed the stack in the InfluxQL parser (one recursion level per term); chDB's signal handler took the fault and shut the query engine down while /health stayed 200. AND/OR chains now parse into balanced trees, expression and subquery nesting are bounded, and /health fails once chDB reports a fatal error. - Names that sanitise to the same table or column (cpu.load / cpu_load, db a + rp b_c / db a_b + rp c, non-ASCII names, field a.b / a_b) shared storage. The client write path now refuses a new colliding name with a 400 before the WAL; existing tables keep their names. - DELETE and DROP SERIES stored a tombstone every later query applied, so points written afterwards were hidden too, and one naming an unknown column broke the measurement for good. They now delete the rows (ALTER TABLE ... DELETE, tags as series lookups) after the WAL flushes, and refuse fields and unknown names. DROP MEASUREMENT clears legacy tombstones, and the ingest schema cache starts over after a drop, so a measurement written again keeps its fields. - Lone ', " and / no longer panic the parser, and a number followed by a multi-byte character no longer panics the lexer (now a linear scan). --- hyperbytedb/src/adapters/chdb/mod.rs | 41 ++ .../src/adapters/chdb/native_adapter.rs | 56 +- hyperbytedb/src/adapters/chdb/session.rs | 6 +- hyperbytedb/src/adapters/http/error.rs | 3 +- hyperbytedb/src/adapters/http/ping.rs | 21 + .../src/adapters/http/shard_handlers.rs | 28 +- .../src/adapters/metadata/rocksdb_meta.rs | 44 ++ .../src/application/ingest_metadata.rs | 22 + .../src/application/ingestion_service.rs | 16 + hyperbytedb/src/application/mod.rs | 1 + .../src/application/peer_ingestion_service.rs | 16 + .../src/application/peer_query_service.rs | 66 ++- hyperbytedb/src/application/physical_names.rs | 484 ++++++++++++++++++ hyperbytedb/src/application/predicate_sql.rs | 158 ++++++ hyperbytedb/src/application/query_service.rs | 115 +++-- .../src/application/schema_mutation_apply.rs | 83 ++- hyperbytedb/src/application/shard_routing.rs | 27 +- hyperbytedb/src/error.rs | 5 + hyperbytedb/src/ports/metadata.rs | 9 + hyperbytedb/src/ports/points_sink.rs | 20 + hyperbytedb/src/timeseriesql/lexer.rs | 127 ++++- hyperbytedb/src/timeseriesql/parser.rs | 213 +++++++- hyperbytedb/tests/compat/ddl_tests.rs | 236 +++++++-- hyperbytedb/tests/compat/query_tests.rs | 29 ++ hyperbytedb/tests/compat/write_tests.rs | 53 ++ 25 files changed, 1732 insertions(+), 147 deletions(-) create mode 100644 hyperbytedb/src/application/physical_names.rs diff --git a/hyperbytedb/src/adapters/chdb/mod.rs b/hyperbytedb/src/adapters/chdb/mod.rs index 9fe7e13..407869b 100644 --- a/hyperbytedb/src/adapters/chdb/mod.rs +++ b/hyperbytedb/src/adapters/chdb/mod.rs @@ -3,3 +3,44 @@ pub mod connection_pool; pub mod native_adapter; pub mod query_adapter; pub mod session; + +use std::sync::atomic::{AtomicBool, Ordering}; + +/// Set once chDB reports that its engine shut down after a fatal error. +static ENGINE_FAILED: AtomicBool = AtomicBool::new(false); + +/// Note a chDB error. After a fatal error (a crash its signal handler caught, +/// in its threads or ours) the embedded engine refuses every query and +/// insert until the process restarts, so `/health` must fail: it kept +/// answering 200 while the WAL grew and no query could run. +pub fn note_error(message: &str) { + if message.contains("shutting down due to a fatal error") && !engine_failed() { + ENGINE_FAILED.store(true, Ordering::Release); + tracing::error!( + error = message, + "chDB engine failed; failing /health so the node is restarted" + ); + } +} + +/// Whether chDB's engine has failed (see [`note_error`]). +#[must_use] +pub fn engine_failed() -> bool { + ENGINE_FAILED.load(Ordering::Acquire) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_the_fatal_shutdown_error_marks_the_engine_failed() { + note_error("Code: 62. DB::Exception: Syntax error"); + assert!(!engine_failed()); + note_error( + "Code: 236. DB::Exception: The server is shutting down due to a fatal error. (ABORTED)", + ); + assert!(engine_failed()); + ENGINE_FAILED.store(false, Ordering::Release); + } +} diff --git a/hyperbytedb/src/adapters/chdb/native_adapter.rs b/hyperbytedb/src/adapters/chdb/native_adapter.rs index fdbdb80..e7e746c 100644 --- a/hyperbytedb/src/adapters/chdb/native_adapter.rs +++ b/hyperbytedb/src/adapters/chdb/native_adapter.rs @@ -139,6 +139,11 @@ fn table_schema_from_measurement_meta( ) } +/// A chDB error for a table that does not exist. +fn is_unknown_table(msg: &str) -> bool { + msg.contains("UNKNOWN_TABLE") || msg.contains("Code: 60") || msg.contains("doesn't exist") +} + fn tag_column_kind(distinct_values: usize, threshold: usize) -> ColumnKind { if distinct_values > threshold { ColumnKind::TagString @@ -962,9 +967,9 @@ impl ChdbNativeAdapter { let insert_options = self.insert_options.clone(); tokio::task::spawn_blocking(move || { pool.with_connection(|conn| { - insert_record_batch_direct(conn, &series_table, batch, insert_options).map_err( - |e| HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e)), - ) + checked_insert_direct(conn, &series_table, batch, insert_options).map_err(|e| { + HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e)) + }) }) }) .await @@ -1442,7 +1447,7 @@ impl ChdbNativeAdapter { let insert_options = self.insert_options.clone(); tokio::task::spawn_blocking(move || { pool.with_connection(|conn| { - insert_record_batch_direct(conn, &series_table, batch, insert_options) + checked_insert_direct(conn, &series_table, batch, insert_options) .map_err(|e| HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e))) }) }) @@ -1549,6 +1554,31 @@ impl PointsSinkPort for ChdbNativeAdapter { Ok(()) } + fn deletes_rows(&self) -> bool { + true + } + + async fn delete_rows( + &self, + db: &str, + rp: &str, + measurement: &str, + predicate: &str, + ) -> Result<(), HyperbytedbError> { + let fact = quoted_table_name(db, rp, measurement); + // `mutations_sync = 2` returns once the rows are gone, so a query + // after the DELETE does not see them. A mutation rewrites only the + // parts that exist when it starts: rows inserted after it are kept. + let sql = format!( + "ALTER TABLE {fact} DELETE WHERE {predicate} \ + SETTINGS mutations_sync = 2, allow_nondeterministic_mutations = 1" + ); + match self.execute(sql).await { + Err(e) if is_unknown_table(&e.to_string()) => Ok(()), + other => other, + } + } + async fn delete_series_id_range( &self, db: &str, @@ -1678,7 +1708,7 @@ impl ChdbNativeAdapter { let insert_options = self.insert_options.clone(); tokio::task::spawn_blocking(move || { pool.with_connection(|conn| { - insert_record_batch_direct(conn, &table, batch, insert_options).map_err(|e| { + checked_insert_direct(conn, &table, batch, insert_options).map_err(|e| { HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e)) }) }) @@ -1758,7 +1788,7 @@ impl ChdbNativeAdapter { tokio::task::spawn_blocking(move || { let batch = (*fact).clone(); pool.with_connection(|conn| { - insert_record_batch_direct(conn, &table, batch, insert_options) + checked_insert_direct(conn, &table, batch, insert_options) .map_err(|e| HyperbytedbError::Chdb(crate::error::ChainedError::from_error(e))) }) }) @@ -1809,6 +1839,20 @@ impl ChdbNativeAdapter { } } +/// [`insert_record_batch_direct`], noting a fatal engine error for `/health`. +fn checked_insert_direct( + conn: &chdb_rust::connection::Connection, + table: &str, + batch: arrow::record_batch::RecordBatch, + options: InsertOptions, +) -> Result<(), chdb_rust::error::Error> { + let result = insert_record_batch_direct(conn, table, batch, options); + if let Err(e) = &result { + crate::adapters::chdb::note_error(&e.to_string()); + } + result +} + fn run_sync(pool: &ChdbConnectionPool, sql: &str) -> Result<(), HyperbytedbError> { pool.with_connection(|conn| { execute_connection(conn, sql, OutputFormat::JSONEachRow) diff --git a/hyperbytedb/src/adapters/chdb/session.rs b/hyperbytedb/src/adapters/chdb/session.rs index 47b74ff..919337d 100644 --- a/hyperbytedb/src/adapters/chdb/session.rs +++ b/hyperbytedb/src/adapters/chdb/session.rs @@ -23,7 +23,11 @@ pub(crate) fn execute_connection( query: &str, format: OutputFormat, ) -> Result { - conn.query(query, format) + let result = conn.query(query, format); + if let Err(e) = &result { + crate::adapters::chdb::note_error(&e.to_string()); + } + result } /// Shared handle to a same-path chDB connection pool. diff --git a/hyperbytedb/src/adapters/http/error.rs b/hyperbytedb/src/adapters/http/error.rs index 01d9c2a..391c9fa 100644 --- a/hyperbytedb/src/adapters/http/error.rs +++ b/hyperbytedb/src/adapters/http/error.rs @@ -31,7 +31,8 @@ fn error_to_status_and_message(err: &HyperbytedbError) -> (StatusCode, String) { HyperbytedbError::DatabaseRequired | HyperbytedbError::MissingParameter(_) | HyperbytedbError::QueryParse(_) - | HyperbytedbError::WriteToMaterializedDestination(_) => StatusCode::BAD_REQUEST, + | HyperbytedbError::WriteToMaterializedDestination(_) + | HyperbytedbError::NameCollision(_) => StatusCode::BAD_REQUEST, HyperbytedbError::CardinalityExceeded { .. } => StatusCode::UNPROCESSABLE_ENTITY, HyperbytedbError::RequestPointLimitExceeded { .. } | HyperbytedbError::PayloadTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE, diff --git a/hyperbytedb/src/adapters/http/ping.rs b/hyperbytedb/src/adapters/http/ping.rs index 43650ba..66eeab8 100644 --- a/hyperbytedb/src/adapters/http/ping.rs +++ b/hyperbytedb/src/adapters/http/ping.rs @@ -233,6 +233,17 @@ pub async fn health(State(state): State>) -> Response { ); } + if crate::adapters::chdb::engine_failed() { + return json_response( + StatusCode::SERVICE_UNAVAILABLE, + health_body( + "fail", + "query engine failed after a fatal error; restart the node", + readiness, + ), + ); + } + json_response( StatusCode::OK, health_body("pass", "ready for queries and writes", readiness), @@ -278,6 +289,16 @@ pub async fn health_ready(State(state): State>) -> Response { .into_response(); } + if crate::adapters::chdb::engine_failed() { + return ( + StatusCode::SERVICE_UNAVAILABLE, + [("Content-Type", "application/json")], + r#"{"status":"fail","message":"query engine failed after a fatal error; restart the node"}"# + .to_string(), + ) + .into_response(); + } + match state.query_port.ping().await { Ok(()) => ( StatusCode::OK, diff --git a/hyperbytedb/src/adapters/http/shard_handlers.rs b/hyperbytedb/src/adapters/http/shard_handlers.rs index fcd1470..b6e288c 100644 --- a/hyperbytedb/src/adapters/http/shard_handlers.rs +++ b/hyperbytedb/src/adapters/http/shard_handlers.rs @@ -1723,11 +1723,31 @@ pub async fn handle_shard_delete( } let predicate = req.predicate.as_deref().unwrap_or(""); + // Delete the rows; points written after the delete stay. A coordinator + // older than this node sends tag names, which the fact table lacks: that + // condition is kept as a tombstone, as before. if !predicate.is_empty() - && let Err(e) = state - .metadata - .store_tombstone(&req.db, &req.rp, &req.measurement, predicate) - .await + && let Err(e) = match crate::application::predicate_sql::delete_rows( + Some(state.wal.as_ref()), + state.points_sink.as_ref(), + &req.db, + &req.rp, + &req.measurement, + predicate, + ) + .await + { + Err(e) + if crate::application::schema_mutation_apply::is_unknown_column(&e.to_string()) => + { + state + .metadata + .store_tombstone(&req.db, &req.rp, &req.measurement, predicate) + .await + .map(|_| ()) + } + other => other, + } { return ( StatusCode::INTERNAL_SERVER_ERROR, diff --git a/hyperbytedb/src/adapters/metadata/rocksdb_meta.rs b/hyperbytedb/src/adapters/metadata/rocksdb_meta.rs index 9ff42bd..b95e81a 100644 --- a/hyperbytedb/src/adapters/metadata/rocksdb_meta.rs +++ b/hyperbytedb/src/adapters/metadata/rocksdb_meta.rs @@ -280,6 +280,9 @@ pub struct RocksDbMetadata { /// (measurement create/delete, db create/drop). Used by /// `meas_list_cache` to detect a stale snapshot. generation: AtomicU64, + /// Bumped whenever measurements, series or whole databases / retention + /// policies are dropped; see [`MetadataPort::drop_generation`]. + drop_generation: AtomicU64, db_cache: RwLock>, meas_cache: RwLock>, /// Storage keys of tag values already persisted. Warmed with the counts @@ -383,6 +386,7 @@ impl RocksDbMetadata { Ok(Self { db, generation: AtomicU64::new(0), + drop_generation: AtomicU64::new(0), db_cache: RwLock::new(HashMap::new()), meas_cache: RwLock::new(HashMap::new()), tag_known: Arc::new(RwLock::new(HashSet::new())), @@ -573,6 +577,10 @@ where #[async_trait] impl MetadataPort for RocksDbMetadata { + fn drop_generation(&self) -> u64 { + self.drop_generation.load(Ordering::Acquire) + } + async fn create_database(&self, name: &str) -> Result<(), HyperbytedbError> { let rdb = self.db.clone(); let name_owned = name.to_string(); @@ -737,6 +745,7 @@ impl MetadataPort for RocksDbMetadata { measurement: Option<&str>, _predicate_sql: &str, ) -> Result { + self.drop_generation.fetch_add(1, Ordering::AcqRel); let rdb = self.db.clone(); let db_s = db.to_string(); let rp_s = rp.to_string(); @@ -792,6 +801,8 @@ impl MetadataPort for RocksDbMetadata { } }); } + // Again once done: a cache refilled while the drop ran is stale. + self.drop_generation.fetch_add(1, Ordering::AcqRel); Ok(removed) } @@ -872,6 +883,7 @@ impl MetadataPort for RocksDbMetadata { } async fn drop_database(&self, name: &str) -> Result<(), HyperbytedbError> { + self.drop_generation.fetch_add(1, Ordering::AcqRel); let rdb = self.db.clone(); let name_owned = name.to_string(); meta_blocking(move || { @@ -928,6 +940,8 @@ impl MetadataPort for RocksDbMetadata { // Structural change — invalidate the `meas_list_cache` // snapshot for this database. self.generation.fetch_add(1, Ordering::Relaxed); + // Again once done: a cache refilled while the drop ran is stale. + self.drop_generation.fetch_add(1, Ordering::AcqRel); Ok(()) } @@ -1047,6 +1061,7 @@ impl MetadataPort for RocksDbMetadata { } async fn drop_retention_policy(&self, db: &str, name: &str) -> Result<(), HyperbytedbError> { + self.drop_generation.fetch_add(1, Ordering::AcqRel); let mut db_obj = self .get_database(db) .await? @@ -1080,6 +1095,8 @@ impl MetadataPort for RocksDbMetadata { }) .await?; self.db_cache.write().insert(db_name, db_obj); + // Again once done: a cache refilled while the drop ran is stale. + self.drop_generation.fetch_add(1, Ordering::AcqRel); Ok(()) } @@ -2081,10 +2098,15 @@ impl MetadataPort for RocksDbMetadata { rp: &str, name: &str, ) -> Result<(), HyperbytedbError> { + self.drop_generation.fetch_add(1, Ordering::AcqRel); let rdb = self.db.clone(); let db_owned = db.to_string(); let name_owned = name.to_string(); let key = meas_key(db, rp, name); + // A dropped measurement's tombstones go with it: kept, they hid the + // points of a measurement later written under the same name, and one + // naming a column it lacked failed every query on it. + let tombstone_prefix = format!("tombstone:{db}:{rp}:{name}:"); let series_to_delete = meta_blocking(move || { let cf = rdb.cf_handle(META_CF).ok_or_else(|| { HyperbytedbError::Metadata("metadata column family not found".into()) @@ -2092,6 +2114,25 @@ impl MetadataPort for RocksDbMetadata { rdb.delete_cf(&cf, key).map_err(|e| { HyperbytedbError::Metadata(crate::error::ChainedError::from_error(e)) })?; + let mut tombstones = Vec::new(); + for item in rdb.iterator_cf_opt( + &cf, + rocksdb::ReadOptions::default(), + IteratorMode::From(tombstone_prefix.as_bytes(), rocksdb::Direction::Forward), + ) { + let (key, _) = item.map_err(|e| { + HyperbytedbError::Metadata(crate::error::ChainedError::from_error(e)) + })?; + if !key.starts_with(tombstone_prefix.as_bytes()) { + break; + } + tombstones.push(key); + } + for k in tombstones { + rdb.delete_cf(&cf, k).map_err(|e| { + HyperbytedbError::Metadata(crate::error::ChainedError::from_error(e)) + })?; + } // Sweep series rows for this measurement across all retention policies. // The series key is rp-scoped (`series:{db}:{rp}:{meas}:{id}`) so we @@ -2128,6 +2169,7 @@ impl MetadataPort for RocksDbMetadata { .await?; let cache_key = format!("{}:{}:{}", db, rp, name); + self.tombstone_cache.write().remove(&cache_key); self.meas_cache.write().remove(&cache_key); self.meas_list_cache.write().remove(db); { @@ -2147,6 +2189,8 @@ impl MetadataPort for RocksDbMetadata { // Bump the structural generation so the `meas_list_cache` // snapshot for this database invalidates on next read. self.generation.fetch_add(1, Ordering::Relaxed); + // Again once done: a cache refilled while the drop ran is stale. + self.drop_generation.fetch_add(1, Ordering::AcqRel); Ok(()) } diff --git a/hyperbytedb/src/application/ingest_metadata.rs b/hyperbytedb/src/application/ingest_metadata.rs index 4ea8078..cdbc91a 100644 --- a/hyperbytedb/src/application/ingest_metadata.rs +++ b/hyperbytedb/src/application/ingest_metadata.rs @@ -115,6 +115,8 @@ pub struct IngestSchemaCache { /// Set of hashed (db, rp, measurement, series_id) tuples already registered, /// letting steady-state batches skip `register_series_batch` entirely. series: RwLock>, + /// [`MetadataPort::drop_generation`] the sets above were filled at. + drop_generation: std::sync::atomic::AtomicU64, } /// The cache's keys and sets live only in memory, so they use foldhash rather @@ -140,9 +142,23 @@ impl IngestSchemaCache { schema: RwLock::new(HashMap::with_capacity_and_hasher(256, FastHash::default())), tags: RwLock::new(HashSet::with_capacity_and_hasher(4096, FastHash::default())), series: RwLock::new(HashSet::with_capacity_and_hasher(8192, FastHash::default())), + drop_generation: std::sync::atomic::AtomicU64::new(0), } } + /// Forget everything when something was dropped since the cache was + /// filled: a dropped measurement written again must be registered again. + pub fn sync_drop_generation(&self, generation: u64) { + use std::sync::atomic::Ordering; + if self.drop_generation.load(Ordering::Acquire) == generation { + return; + } + self.schema.write().clear(); + self.tags.write().clear(); + self.series.write().clear(); + self.drop_generation.store(generation, Ordering::Release); + } + fn hash_key(db: &str, rp: &str, meas: &str) -> u64 { let mut h = fast_hasher(); db.hash(&mut h); @@ -452,6 +468,9 @@ pub async fn prepare_columnar_metadata( limits: &IngestCardinalityLimits, schema_cache: Option<&IngestSchemaCache>, ) -> Result<(), HyperbytedbError> { + if let Some(sc) = schema_cache { + sc.sync_drop_generation(metadata.drop_generation()); + } let mut field_types: HashMap = HashMap::with_capacity(1); field_types.insert(batch.field.clone(), 0); // Float64 @@ -716,6 +735,9 @@ pub async fn prepare_batch_metadata( // seen only after that), so only fresh points can carry an unseen tag // value. Checking every tag of every point instead was ~5% of a node's // CPU on n3, nearly all of it on series that were long known. + if let Some(sc) = schema_cache { + sc.sync_drop_generation(metadata.drop_generation()); + } let mut fresh: Vec<&Point> = Vec::new(); match schema_cache { Some(sc) => { diff --git a/hyperbytedb/src/application/ingestion_service.rs b/hyperbytedb/src/application/ingestion_service.rs index ccf21ba..103b0e5 100644 --- a/hyperbytedb/src/application/ingestion_service.rs +++ b/hyperbytedb/src/application/ingestion_service.rs @@ -9,6 +9,7 @@ use crate::application::ingest_metadata::{ }; use crate::application::line_protocol::parse_line_body_to_points_limited; use crate::application::msgpack_ingest::parse_msgpack_body_to_points_limited; +use crate::application::physical_names::PhysicalNameGuard; use crate::application::wal_append::{ ColumnarWalAppend, append_columnar_with_prepared, append_points_with_prepared, }; @@ -25,6 +26,7 @@ pub struct IngestionServiceImpl { limits: IngestCardinalityLimits, max_points_per_request: usize, schema_cache: IngestSchemaCache, + name_guard: PhysicalNameGuard, } impl IngestionServiceImpl { @@ -58,6 +60,7 @@ impl IngestionServiceImpl { limits, max_points_per_request, schema_cache: IngestSchemaCache::new(), + name_guard: PhysicalNameGuard::new(), } } @@ -120,6 +123,16 @@ impl IngestionPort for IngestionServiceImpl { let t2 = std::time::Instant::now(); histogram!("hyperbytedb_ingest_parse_seconds").record((t2 - t1).as_secs_f64()); + self.name_guard + .check_columnar( + &self.metadata, + db, + &retention_policy, + &wire.measurement, + &wire.field, + wire.tags.keys().map(String::as_str), + ) + .await?; prepare_columnar_metadata( &self.metadata, db, @@ -213,6 +226,9 @@ impl IngestionPort for IngestionServiceImpl { let t2 = std::time::Instant::now(); histogram!("hyperbytedb_ingest_parse_seconds").record((t2 - t1).as_secs_f64()); + self.name_guard + .check_points(&self.metadata, db, &retention_policy, &points) + .await?; prepare_batch_metadata( &self.metadata, db, diff --git a/hyperbytedb/src/application/mod.rs b/hyperbytedb/src/application/mod.rs index 7b51d12..4bf7f34 100644 --- a/hyperbytedb/src/application/mod.rs +++ b/hyperbytedb/src/application/mod.rs @@ -16,6 +16,7 @@ pub mod msgpack_limits; pub mod mv_handoff; pub mod peer_ingestion_service; pub mod peer_query_service; +pub mod physical_names; pub mod predicate_sql; pub mod query_arrow_series; pub mod query_service; diff --git a/hyperbytedb/src/application/peer_ingestion_service.rs b/hyperbytedb/src/application/peer_ingestion_service.rs index 7a50c24..f01c5f6 100644 --- a/hyperbytedb/src/application/peer_ingestion_service.rs +++ b/hyperbytedb/src/application/peer_ingestion_service.rs @@ -51,6 +51,7 @@ pub struct PeerIngestionService { limits: IngestCardinalityLimits, max_points_per_request: usize, schema_cache: IngestSchemaCache, + name_guard: crate::application::physical_names::PhysicalNameGuard, replication: ReplicationConfig, shard_routing: Option>, is_raft_leader: Arc bool + Send + Sync>, @@ -170,6 +171,7 @@ impl PeerIngestionService { limits, max_points_per_request, schema_cache: IngestSchemaCache::new(), + name_guard: crate::application::physical_names::PhysicalNameGuard::new(), replication, shard_routing: None, is_raft_leader: Arc::new(|| false), @@ -865,6 +867,16 @@ impl IngestionPort for PeerIngestionService { let t2 = std::time::Instant::now(); histogram!("hyperbytedb_ingest_parse_seconds").record((t2 - t1).as_secs_f64()); + self.name_guard + .check_columnar( + &self.metadata, + db, + &retention_policy, + &wire.measurement, + &wire.field, + wire.tags.keys().map(String::as_str), + ) + .await?; prepare_columnar_metadata( &self.metadata, db, @@ -982,6 +994,10 @@ impl IngestionPort for PeerIngestionService { return partial.map_or(Ok(()), Err); } + self.name_guard + .check_points(&self.metadata, db, &retention_policy, &points) + .await?; + let precision_val = Precision::from_str_opt(precision); let replication_body = match format { // The request as sent, unless points were dropped from it. diff --git a/hyperbytedb/src/application/peer_query_service.rs b/hyperbytedb/src/application/peer_query_service.rs index 0c43787..f03f816 100644 --- a/hyperbytedb/src/application/peer_query_service.rs +++ b/hyperbytedb/src/application/peer_query_service.rs @@ -413,11 +413,14 @@ async fn mutation_request_from_statement( Statement::DropDatabase(name) => Ok(MutationRequest::DropDatabase(name.clone())), Statement::Delete(del) => { let del_rp = metadata.get_default_rp(db).await?; - let predicate_sql = if let Some(ref cond) = del.condition { - build_predicate_sql(metadata, db, &del_rp, &del.from, cond).await? - } else { - String::new() - }; + let predicate_sql = crate::application::predicate_sql::build_delete_predicate( + metadata, + db, + &del_rp, + &del.from, + del.condition.as_ref(), + ) + .await?; Ok(MutationRequest::Delete { database: db.to_string(), rp: del_rp, @@ -541,11 +544,18 @@ async fn mutation_request_from_statement( MeasurementName::Regex(_) => None, }); let drop_rp = metadata.get_default_rp(target_db).await?; - let predicate_sql = if let Some(ref cond) = s.condition { - let meas = measurement.as_deref().unwrap_or(""); - build_predicate_sql(metadata, target_db, &drop_rp, meas, cond).await? - } else { - String::new() + let predicate_sql = match (measurement.as_deref(), s.condition.as_ref()) { + // Replicas delete the rows this selects. + (Some(meas), cond) => { + crate::application::predicate_sql::build_delete_predicate( + metadata, target_db, &drop_rp, meas, cond, + ) + .await? + } + (None, Some(cond)) => { + build_predicate_sql(metadata, target_db, &drop_rp, "", cond).await? + } + (None, None) => String::new(), }; Ok(MutationRequest::DropSeries { database: target_db.to_string(), @@ -709,12 +719,15 @@ impl QueryService for PeerQueryService { } else { self.metadata.get_default_rp(db).await? }; - let predicate_sql = if let Some(ref cond) = del.condition { - build_predicate_sql(&self.metadata, db, &del_rp, &del.from, cond) - .await? - } else { - String::new() - }; + let predicate_sql = + crate::application::predicate_sql::build_delete_predicate( + &self.metadata, + db, + &del_rp, + &del.from, + del.condition.as_ref(), + ) + .await?; self.replicate_mutation(MutationRequest::Delete { database: db.to_string(), rp: del_rp, @@ -1045,12 +1058,23 @@ impl QueryService for PeerQueryService { } else { self.metadata.get_default_rp(target_db).await? }; - let predicate_sql = if let Some(ref cond) = s.condition { - let meas = measurement.as_deref().unwrap_or(""); - build_predicate_sql(&self.metadata, target_db, &ds_rp, meas, cond) + let predicate_sql = match (measurement.as_deref(), s.condition.as_ref()) { + // Replicas delete the rows this selects. + (Some(meas), cond) => { + crate::application::predicate_sql::build_delete_predicate( + &self.metadata, + target_db, + &ds_rp, + meas, + cond, + ) .await? - } else { - String::new() + } + (None, Some(cond)) => { + build_predicate_sql(&self.metadata, target_db, &ds_rp, "", cond) + .await? + } + (None, None) => String::new(), }; self.replicate_mutation(MutationRequest::DropSeries { database: target_db.to_string(), diff --git a/hyperbytedb/src/application/physical_names.rs b/hyperbytedb/src/application/physical_names.rs new file mode 100644 index 0000000..784c0bc --- /dev/null +++ b/hyperbytedb/src/application/physical_names.rs @@ -0,0 +1,484 @@ +//! Refuse writes whose names would share storage with other names. +//! +//! Tables and columns are named from sanitised identifiers +//! ([`crate::domain::chdb_naming`]): every character outside +//! `[A-Za-z0-9_]` becomes `_`, and a table is `__`. +//! So `cpu.load` and `cpu_load` are one table, as are database `a` with +//! retention policy `b_c` and database `a_b` with policy `c`, and the +//! fields `a.b` and `a_b` are one column. Writing the second name of such a +//! pair overwrote and read back the first one's data. +//! +//! The names stay as they are, since tables already stored under them +//! would otherwise be lost. Instead the client write path checks each +//! measurement, field and tag key it has not checked before, and refuses +//! the write (400) when a new name maps onto storage another name already +//! owns. The check runs before the WAL append, so a refused write is never +//! flushed. Replication and shard transfers do not check: they carry data +//! an entry node already accepted. + +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; +use std::hash::{Hash, Hasher}; +use std::sync::Arc; + +use parking_lot::RwLock; + +use crate::domain::chdb_naming::{ + field_column_name, tag_column_name, unquoted_series_table_name, unquoted_table_name, +}; +use crate::domain::point::Point; +use crate::error::HyperbytedbError; +use crate::ports::metadata::MetadataPort; + +/// Upper bound on remembered names; past it the set is cleared and names +/// are checked again as they recur. +const MAX_CHECKED: usize = 1 << 20; + +/// Names (measurement, field and tag keys) already checked by this node. +#[derive(Default)] +pub struct PhysicalNameGuard { + checked: RwLock>, + /// [`MetadataPort::drop_generation`] `checked` was filled at: a dropped + /// name no longer owns its table, and one that took it since must not + /// be shadowed by the first being remembered as checked. + drop_generation: std::sync::atomic::AtomicU64, +} + +/// The names one batch writes to one measurement. +#[derive(Default)] +struct BatchNames<'a> { + fields: BTreeSet<&'a str>, + tags: BTreeSet<&'a str>, +} + +fn name_hash(db: &str, rp: &str, measurement: &str, kind: u8, key: &str) -> u64 { + let mut h = std::hash::DefaultHasher::new(); + (db, rp, measurement, kind, key).hash(&mut h); + h.finish() +} + +impl PhysicalNameGuard { + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Refuse `points` if any of their names maps onto storage that another + /// name already owns. + /// + /// # Errors + /// [`HyperbytedbError::NameCollision`] for a colliding name, or a + /// metadata error. + pub async fn check_points( + &self, + metadata: &Arc, + db: &str, + rp: &str, + points: &[Point], + ) -> Result<(), HyperbytedbError> { + let mut by_meas: BTreeMap<&str, BatchNames<'_>> = BTreeMap::new(); + let mut last: Option<&str> = None; + for p in points { + // Batches are mostly runs of one measurement. + if last != Some(p.measurement.as_str()) { + by_meas.entry(p.measurement.as_str()).or_default(); + last = Some(p.measurement.as_str()); + } + let Some(names) = by_meas.get_mut(p.measurement.as_str()) else { + continue; + }; + for k in p.fields.keys() { + if !names.fields.contains(k.as_str()) { + names.fields.insert(k); + } + } + for k in p.tags.keys() { + if !names.tags.contains(k.as_str()) { + names.tags.insert(k); + } + } + } + for (measurement, names) in &by_meas { + self.check_measurement(metadata, db, rp, measurement, names) + .await?; + } + Ok(()) + } + + /// [`Self::check_points`] for a columnar batch: one measurement, one + /// field, and its tags. + /// + /// # Errors + /// As [`Self::check_points`]. + pub async fn check_columnar( + &self, + metadata: &Arc, + db: &str, + rp: &str, + measurement: &str, + field: &str, + tags: impl Iterator, + ) -> Result<(), HyperbytedbError> { + let names = BatchNames { + fields: BTreeSet::from([field]), + tags: tags.collect(), + }; + self.check_measurement(metadata, db, rp, measurement, &names) + .await + } + + async fn check_measurement( + &self, + metadata: &Arc, + db: &str, + rp: &str, + measurement: &str, + names: &BatchNames<'_>, + ) -> Result<(), HyperbytedbError> { + let mut hashes: Vec = Vec::with_capacity(1 + names.fields.len() + names.tags.len()); + hashes.push(name_hash(db, rp, measurement, 0, "")); + hashes.extend( + names + .fields + .iter() + .map(|k| name_hash(db, rp, measurement, 1, k)), + ); + hashes.extend( + names + .tags + .iter() + .map(|k| name_hash(db, rp, measurement, 2, k)), + ); + let generation = metadata.drop_generation(); + if self + .drop_generation + .swap(generation, std::sync::atomic::Ordering::AcqRel) + != generation + { + self.checked.write().clear(); + } + if hashes.iter().all(|h| self.checked.read().contains(h)) { + return Ok(()); + } + + let existing = metadata.get_measurement(db, rp, measurement).await?; + if existing.is_none() { + check_table_names(metadata, db, rp, measurement).await?; + } + let (stored_fields, stored_tags) = existing + .map(|m| (m.field_types.into_keys().collect(), m.tag_keys)) + .unwrap_or_else(|| (Vec::new(), Vec::new())); + check_column_names(measurement, &stored_fields, &stored_tags, names)?; + + let mut checked = self.checked.write(); + if checked.len() + hashes.len() > MAX_CHECKED { + checked.clear(); + } + checked.extend(hashes); + Ok(()) + } +} + +/// A new measurement's fact and series tables must not be another +/// measurement's, in any database or retention policy. +async fn check_table_names( + metadata: &Arc, + db: &str, + rp: &str, + measurement: &str, +) -> Result<(), HyperbytedbError> { + let mine = [ + unquoted_table_name(db, rp, measurement).to_string(), + unquoted_series_table_name(db, rp, measurement).to_string(), + ]; + for database in metadata.list_databases().await? { + let rps: BTreeSet<&str> = database + .retention_policies + .iter() + .map(|r| r.name.as_str()) + .collect(); + for other_rp in rps { + for other in metadata + .list_measurements_for_rp(&database.name, other_rp) + .await? + { + if database.name == db && other_rp == rp && other == measurement { + continue; + } + let theirs = [ + unquoted_table_name(&database.name, other_rp, &other).to_string(), + unquoted_series_table_name(&database.name, other_rp, &other).to_string(), + ]; + if mine.iter().any(|n| theirs.contains(n)) { + return Err(HyperbytedbError::NameCollision(format!( + "measurement \"{measurement}\" (database \"{db}\", retention policy \ + \"{rp}\") would be stored with measurement \"{other}\" (database \ + \"{}\", retention policy \"{other_rp}\"); rename it", + database.name + ))); + } + } + } + } + Ok(()) +} + +/// A new field key, or tag key, of one measurement must not share a column +/// with another. Stored keys that already share one (written before this +/// check) are left alone, so their measurement can still be written. +fn check_column_names( + measurement: &str, + stored_fields: &[String], + stored_tags: &[String], + names: &BatchNames<'_>, +) -> Result<(), HyperbytedbError> { + let field_set: HashSet<&str> = stored_fields + .iter() + .map(String::as_str) + .chain(names.fields.iter().copied()) + .collect(); + first_new_collision(stored_fields, &names.fields, field_column_name).map_or( + Ok(()), + |(new, other)| { + Err(HyperbytedbError::NameCollision(format!( + "field \"{new}\" of measurement \"{measurement}\" would be stored with field \ + \"{other}\"; rename it" + ))) + }, + )?; + first_new_collision(stored_tags, &names.tags, |t| tag_column_name(t, &field_set)).map_or( + Ok(()), + |(new, other)| { + Err(HyperbytedbError::NameCollision(format!( + "tag key \"{new}\" of measurement \"{measurement}\" would be stored with tag key \ + \"{other}\"; rename it" + ))) + }, + ) +} + +/// The first key of `batch` not in `stored` whose column is another key's. +fn first_new_collision<'a>( + stored: &'a [String], + batch: &BTreeSet<&'a str>, + column: impl Fn(&str) -> String, +) -> Option<(&'a str, &'a str)> { + let stored_set: HashSet<&str> = stored.iter().map(String::as_str).collect(); + let mut columns: HashMap = + stored.iter().map(|k| (column(k), k.as_str())).collect(); + for &k in batch { + if stored_set.contains(k) { + continue; + } + match columns.get(&column(k)) { + Some(&other) if other != k => return Some((k, other)), + Some(_) => {} + None => { + columns.insert(column(k), k); + } + } + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::adapters::metadata::rocksdb_meta::RocksDbMetadata; + use crate::domain::database::RetentionPolicy; + use crate::domain::measurement::MeasurementMeta; + use crate::domain::point::FieldValue; + + fn point(meas: &str, field: &str, tag: &str) -> Point { + Point { + measurement: meas.to_string(), + tags: BTreeMap::from([(tag.to_string(), "x".to_string())]), + fields: BTreeMap::from([(field.to_string(), FieldValue::Float(1.0))]), + timestamp: 1, + series_id_cache: Default::default(), + } + } + + async fn meta_with( + dir: &tempfile::TempDir, + db: &str, + rp: &str, + meas: &str, + ) -> Arc { + let meta: Arc = Arc::new(RocksDbMetadata::open(dir.path()).unwrap()); + meta.create_database(db).await.unwrap(); + if rp != "autogen" { + let mut policy = RetentionPolicy::autogen(); + policy.name = rp.to_string(); + policy.is_default = false; + meta.create_retention_policy(db, policy).await.unwrap(); + } + register(&meta, db, rp, meas, "v", "host").await; + meta + } + + async fn register( + meta: &Arc, + db: &str, + rp: &str, + meas: &str, + field: &str, + tag: &str, + ) { + meta.register_measurement( + db, + rp, + &MeasurementMeta { + name: meas.into(), + field_types: HashMap::from([(field.to_string(), 0u8)]), + tag_keys: vec![tag.to_string()], + ..Default::default() + }, + ) + .await + .unwrap(); + } + + fn is_collision(r: Result<(), HyperbytedbError>) -> bool { + matches!(r, Err(HyperbytedbError::NameCollision(_))) + } + + #[tokio::test] + async fn measurement_sharing_a_table_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let meta = meta_with(&dir, "db", "autogen", "cpu.load").await; + let guard = PhysicalNameGuard::new(); + for m in ["cpu_load", "cpu load", "cpu-load"] { + let r = guard + .check_points(&meta, "db", "autogen", &[point(m, "v", "host")]) + .await; + assert!(is_collision(r), "{m}"); + } + // The owner itself, and unrelated names, still write. + guard + .check_points(&meta, "db", "autogen", &[point("cpu.load", "v", "host")]) + .await + .unwrap(); + guard + .check_points(&meta, "db", "autogen", &[point("cpu", "v", "host")]) + .await + .unwrap(); + } + + #[tokio::test] + async fn non_ascii_names_sharing_a_table_are_refused() { + let dir = tempfile::tempdir().unwrap(); + let meta = meta_with(&dir, "db", "autogen", "温度").await; + let r = PhysicalNameGuard::new() + .check_points(&meta, "db", "autogen", &[point("湿度", "v", "host")]) + .await; + assert!(is_collision(r)); + } + + #[tokio::test] + async fn another_database_sharing_a_table_is_refused() { + // db `a_b` rp `c` and db `a` rp `b_c` are both table `a_b_c_m`. + let dir = tempfile::tempdir().unwrap(); + let meta = meta_with(&dir, "a_b", "c", "m").await; + meta.create_database("a").await.unwrap(); + let r = PhysicalNameGuard::new() + .check_points(&meta, "a", "b_c", &[point("m", "v", "host")]) + .await; + assert!(is_collision(r)); + } + + #[tokio::test] + async fn measurement_named_like_a_series_table_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let meta = meta_with(&dir, "db", "autogen", "cpu").await; + let r = PhysicalNameGuard::new() + .check_points(&meta, "db", "autogen", &[point("cpu_series", "v", "host")]) + .await; + assert!(is_collision(r)); + } + + #[tokio::test] + async fn fields_or_tags_sharing_a_column_are_refused() { + let dir = tempfile::tempdir().unwrap(); + let meta = meta_with(&dir, "db", "autogen", "cpu").await; + let guard = PhysicalNameGuard::new(); + let r = guard + .check_points(&meta, "db", "autogen", &[point("cpu", "v.", "host")]) + .await; + assert!(r.is_ok(), "`v_` is not `v`"); + let r = guard + .check_points(&meta, "db", "autogen", &[point("mem", "a.b", "t")]) + .await; + assert!(r.is_ok()); + // Within one batch, and against the stored schema. + let r = guard + .check_points( + &meta, + "db", + "autogen", + &[point("disk", "a.b", "t"), point("disk", "a_b", "t")], + ) + .await; + assert!(is_collision(r)); + register(&meta, "db", "autogen", "net", "a.b", "host.name").await; + let r = guard + .check_points(&meta, "db", "autogen", &[point("net", "a_b", "host.name")]) + .await; + assert!(is_collision(r)); + let r = guard + .check_points(&meta, "db", "autogen", &[point("net", "a.b", "host_name")]) + .await; + assert!(is_collision(r)); + } + + #[tokio::test] + async fn stored_keys_that_already_collide_still_write() { + let dir = tempfile::tempdir().unwrap(); + let meta = meta_with(&dir, "db", "autogen", "cpu").await; + meta.register_measurement( + "db", + "autogen", + &MeasurementMeta { + name: "old".into(), + field_types: HashMap::from([("a.b".to_string(), 0u8), ("a_b".to_string(), 0u8)]), + tag_keys: vec!["host".to_string()], + ..Default::default() + }, + ) + .await + .unwrap(); + PhysicalNameGuard::new() + .check_points(&meta, "db", "autogen", &[point("old", "a_b", "host")]) + .await + .unwrap(); + } + + #[tokio::test] + async fn columnar_batches_are_checked() { + let dir = tempfile::tempdir().unwrap(); + let meta = meta_with(&dir, "db", "autogen", "cpu.load").await; + let r = PhysicalNameGuard::new() + .check_columnar( + &meta, + "db", + "autogen", + "cpu_load", + "v", + ["host"].into_iter(), + ) + .await; + assert!(is_collision(r)); + } + + #[tokio::test] + async fn a_refused_name_stays_refused() { + let dir = tempfile::tempdir().unwrap(); + let meta = meta_with(&dir, "db", "autogen", "cpu.load").await; + let guard = PhysicalNameGuard::new(); + for _ in 0..2 { + let r = guard + .check_points(&meta, "db", "autogen", &[point("cpu_load", "v", "host")]) + .await; + assert!(is_collision(r)); + } + } +} diff --git a/hyperbytedb/src/application/predicate_sql.rs b/hyperbytedb/src/application/predicate_sql.rs index 79554d7..e144c76 100644 --- a/hyperbytedb/src/application/predicate_sql.rs +++ b/hyperbytedb/src/application/predicate_sql.rs @@ -54,3 +54,161 @@ pub(crate) fn collect_expr_identifiers(expr: &Expr, out: &mut Vec) { _ => {} } } + +/// The fact-table condition that selects the rows a `DELETE` or +/// `DROP SERIES` removes, for `ALTER TABLE DELETE WHERE `. +/// +/// Tags live in the series table, so each comparison on a tag becomes a +/// series lookup (`series_id IN (SELECT series_id FROM WHERE ..)`); +/// comparisons on `time` stay as they are. No condition selects every row. +/// +/// As in InfluxDB 1.x, only tags and `time` may be used: a field is refused, +/// and so is a name the measurement does not have (it was stored and then +/// failed every later query on the measurement with "Unknown identifier"). +/// A coordinator with no catalog entry for the measurement (a sharded node +/// that holds none of its regions) cannot tell fields from tags and takes +/// every name but `time` to be a tag. +/// +/// # Errors +/// [`HyperbytedbError::QueryParse`] for a field or an unknown name. +pub async fn build_delete_predicate( + metadata: &Arc, + db: &str, + rp: &str, + measurement: &str, + cond: Option<&Expr>, +) -> Result { + let Some(cond) = cond else { + return Ok("1".to_string()); + }; + let mut ids = Vec::new(); + collect_expr_identifiers(cond, &mut ids); + let mapping = match metadata.get_measurement(db, rp, measurement).await? { + Some(meta) => { + let mapping = ColumnMapping::from_measurement_meta(&meta); + for id in &ids { + if id == "time" || mapping.tag_keys.contains(id) { + continue; + } + return Err(HyperbytedbError::QueryParse( + if mapping.field_names.contains(id) { + format!( + "fields not supported in WHERE clause during deletion: \"{id}\" \ + is a field of measurement \"{measurement}\"" + ) + } else { + format!("measurement \"{measurement}\" has no tag key \"{id}\"") + }, + )); + } + mapping + } + None => ColumnMapping::from_identifiers( + ids.iter().filter(|id| *id != "time").cloned(), + std::iter::empty(), + ), + }; + let series = crate::domain::chdb_naming::quoted_series_table_name(db, rp, measurement); + let mut sql = String::new(); + delete_term(cond, &mapping, series.as_str(), &mut sql)?; + Ok(sql) +} + +fn delete_term( + expr: &Expr, + mapping: &ColumnMapping, + series_table: &str, + out: &mut String, +) -> Result<(), HyperbytedbError> { + use crate::timeseriesql::ast::{BinaryOp, UnaryOp}; + match expr { + Expr::BinaryExpr(b) if matches!(b.op, BinaryOp::And | BinaryOp::Or) => { + out.push('('); + delete_term(&b.left, mapping, series_table, out)?; + out.push_str(if b.op == BinaryOp::And { + " AND " + } else { + " OR " + }); + delete_term(&b.right, mapping, series_table, out)?; + out.push(')'); + } + Expr::UnaryExpr(UnaryOp::Not, inner) => { + out.push_str("NOT ("); + delete_term(inner, mapping, series_table, out)?; + out.push(')'); + } + other => { + let mut ids = Vec::new(); + collect_expr_identifiers(other, &mut ids); + if ids.iter().any(|id| mapping.tag_keys.contains(id)) { + out.push_str("`series_id` IN (SELECT `series_id` FROM "); + out.push_str(series_table); + out.push_str(" WHERE "); + to_clickhouse::translate_condition(other, mapping, out)?; + out.push(')'); + } else { + out.push('('); + to_clickhouse::translate_condition(other, mapping, out)?; + out.push(')'); + } + } + } + Ok(()) +} + +/// What a node deletes its own rows with: its WAL (flushed first) and sink. +#[derive(Clone, Copy)] +pub struct LocalRows<'a> { + pub wal: Option<&'a dyn crate::ports::wal::WalPort>, + pub sink: &'a dyn crate::ports::points_sink::PointsSinkPort, +} + +/// Delete the rows `predicate` (from [`build_delete_predicate`]) selects +/// from this node's copy of the measurement. +/// +/// Rows still in the WAL are flushed first, up to the WAL's last entry when +/// the delete arrived, so a point written before the delete is removed with +/// the rest; one written after it is not touched. An empty `predicate` +/// deletes nothing. +/// +/// # Errors +/// A chDB error from the delete. +pub async fn delete_rows( + wal: Option<&dyn crate::ports::wal::WalPort>, + sink: &dyn crate::ports::points_sink::PointsSinkPort, + db: &str, + rp: &str, + measurement: &str, + predicate: &str, +) -> Result<(), HyperbytedbError> { + if predicate.is_empty() || !sink.deletes_rows() { + return Ok(()); + } + if let Some(wal) = wal { + wait_for_flush(wal).await?; + } + sink.delete_rows(db, rp, measurement, predicate).await +} + +/// Longest wait for the flush service to write the WAL's current entries: +/// past one flush interval (10 s by default), well inside the query +/// timeout (30 s by default). +const DELETE_FLUSH_WAIT: std::time::Duration = std::time::Duration::from_secs(15); + +async fn wait_for_flush(wal: &dyn crate::ports::wal::WalPort) -> Result<(), HyperbytedbError> { + let target = wal.last_sequence().await?; + let deadline = std::time::Instant::now() + DELETE_FLUSH_WAIT; + while wal.flushed_through() < target { + if std::time::Instant::now() >= deadline { + tracing::warn!( + target, + flushed_through = wal.flushed_through(), + "delete: WAL not flushed in time; points written just before the delete may remain" + ); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + Ok(()) +} diff --git a/hyperbytedb/src/application/query_service.rs b/hyperbytedb/src/application/query_service.rs index 4cedc71..efdcb24 100644 --- a/hyperbytedb/src/application/query_service.rs +++ b/hyperbytedb/src/application/query_service.rs @@ -1294,7 +1294,8 @@ async fn execute_statement( }; svc.metadata.delete_measurement(db, &rp, name).await?; if let Some(ctx) = svc.shard_routing.as_ref() { - scatter_delete_to_regions(ctx, svc.metadata.as_ref(), db, &rp, name, "").await?; + scatter_delete_to_regions(ctx, svc.metadata.as_ref(), None, db, &rp, name, "") + .await?; } if let Err(e) = svc.points_sink.drop_measurement(db, &rp, name).await { tracing::warn!( @@ -1320,27 +1321,36 @@ async fn execute_statement( .await .unwrap_or_else(|_| "autogen".to_string()) }; - let predicate_sql = if let Some(ref cond) = del.condition { - crate::application::predicate_sql::build_predicate_sql( - &svc.metadata, - db, - &del_rp, - &del.from, - cond, - ) - .await? - } else { - String::new() - }; - - svc.metadata - .store_tombstone(db, &del_rp, &del.from, &predicate_sql) - .await?; + // The rows are deleted, as in InfluxDB: points written after the + // DELETE stay. It used to store a tombstone that every later query + // applied, hiding later points too. + let predicate_sql = crate::application::predicate_sql::build_delete_predicate( + &svc.metadata, + db, + &del_rp, + &del.from, + del.condition.as_ref(), + ) + .await?; if let Some(ctx) = svc.shard_routing.as_ref() { scatter_delete_to_regions( ctx, svc.metadata.as_ref(), + Some(crate::application::predicate_sql::LocalRows { + wal: Some(svc.wal.as_ref()), + sink: svc.points_sink.as_ref(), + }), + db, + &del_rp, + &del.from, + &predicate_sql, + ) + .await?; + } else { + crate::application::predicate_sql::delete_rows( + Some(svc.wal.as_ref()), + svc.points_sink.as_ref(), db, &del_rp, &del.from, @@ -1353,7 +1363,7 @@ async fn execute_statement( db = db, measurement = %del.from, predicate = %predicate_sql, - "DELETE tombstone stored" + "DELETE applied" ); Ok(StatementResult { @@ -1786,40 +1796,63 @@ async fn execute_statement( String::new() }; if measurement.is_some() || !predicate_sql.is_empty() { - if let Some(ref meas) = measurement - && !predicate_sql.is_empty() - { - svc.metadata - .store_tombstone(target_db, &rp, meas, &predicate_sql) - .await?; - } svc.metadata .delete_series_matching(target_db, &rp, measurement.as_deref(), &predicate_sql) .await?; - if let Some(ctx) = svc.shard_routing.as_ref() { - if let Some(ref meas) = measurement { + // The series' rows are deleted; points written to them later + // stay. A tombstone used to hide those too. + let mut targets: Vec<(String, String)> = Vec::new(); + if let Some(ref meas) = measurement { + let pred = crate::application::predicate_sql::build_delete_predicate( + &svc.metadata, + target_db, + &rp, + meas, + s.condition.as_ref(), + ) + .await?; + targets.push((meas.clone(), pred)); + } else if svc.shard_routing.is_some() { + // Every measurement that has the condition's tags. + for meas in svc.metadata.list_measurements(target_db).await? { + if let Ok(pred) = crate::application::predicate_sql::build_delete_predicate( + &svc.metadata, + target_db, + &rp, + &meas, + s.condition.as_ref(), + ) + .await + { + targets.push((meas, pred)); + } + } + } + for (meas, pred) in targets { + if let Some(ctx) = svc.shard_routing.as_ref() { scatter_delete_to_regions( ctx, svc.metadata.as_ref(), + Some(crate::application::predicate_sql::LocalRows { + wal: Some(svc.wal.as_ref()), + sink: svc.points_sink.as_ref(), + }), target_db, &rp, - meas, - &predicate_sql, + &meas, + &pred, ) .await?; } else { - let measurements = svc.metadata.list_measurements(target_db).await?; - for meas in measurements { - scatter_delete_to_regions( - ctx, - svc.metadata.as_ref(), - target_db, - &rp, - &meas, - &predicate_sql, - ) - .await?; - } + crate::application::predicate_sql::delete_rows( + Some(svc.wal.as_ref()), + svc.points_sink.as_ref(), + target_db, + &rp, + &meas, + &pred, + ) + .await?; } } } diff --git a/hyperbytedb/src/application/schema_mutation_apply.rs b/hyperbytedb/src/application/schema_mutation_apply.rs index c420372..5019033 100644 --- a/hyperbytedb/src/application/schema_mutation_apply.rs +++ b/hyperbytedb/src/application/schema_mutation_apply.rs @@ -99,6 +99,60 @@ pub fn schema_mutation_op_label(mutation: &MutationRequest) -> &'static str { } } +/// Apply a replicated `DELETE` / `DROP SERIES` predicate to this node's rows +/// ([`crate::application::predicate_sql::delete_rows`]). +/// +/// A coordinator older than this node sends the condition with tag names +/// (it stored tombstones), which the fact table does not have; that one is +/// stored as a tombstone, as the coordinator did. +async fn delete_replicated_rows( + metadata: &Arc, + sink: Option<&Arc>, + wal: Option<&Arc>, + db: &str, + rp: &str, + measurement: &str, + predicate: &str, +) -> Result<(), HyperbytedbError> { + if predicate.is_empty() { + return Ok(()); + } + if let Some(sink) = sink { + match crate::application::predicate_sql::delete_rows( + wal.map(|w| w.as_ref()), + sink.as_ref(), + db, + rp, + measurement, + predicate, + ) + .await + { + Err(e) if is_unknown_column(&e.to_string()) => { + tracing::warn!( + db, + measurement, + error = %e, + "replicated delete names columns the table lacks; storing it as a tombstone" + ); + } + other => return other, + } + } + metadata + .store_tombstone(db, rp, measurement, predicate) + .await + .map(|_| ()) +} + +/// A chDB error for a column the table does not have. +pub(crate) fn is_unknown_column(msg: &str) -> bool { + msg.contains("UNKNOWN_IDENTIFIER") + || msg.contains("Unknown identifier") + || msg.contains("Missing columns") + || msg.contains("Code: 47") +} + async fn apply_schema_mutation_inner( deps: SchemaMutationDeps<'_>, mutation: MutationRequest, @@ -145,13 +199,15 @@ async fn apply_schema_mutation_inner( measurement, predicate_sql, } => { - metadata - .store_tombstone(&database, &rp, &measurement, &predicate_sql) - .await?; + let sink = points_sink.or_else(|| mv_service.map(|mv| mv.points_sink())); if let Some(ctx) = shard_routing { shard_routing::scatter_delete_to_regions( ctx, metadata.as_ref(), + sink.map(|s| crate::application::predicate_sql::LocalRows { + wal: wal.map(|w| w.as_ref()), + sink: s.as_ref(), + }), &database, &rp, &measurement, @@ -159,6 +215,16 @@ async fn apply_schema_mutation_inner( ) .await?; } else { + delete_replicated_rows( + metadata, + sink, + wal, + &database, + &rp, + &measurement, + &predicate_sql, + ) + .await?; metadata .delete_series_matching(&database, &rp, Some(&measurement), &predicate_sql) .await?; @@ -208,14 +274,9 @@ async fn apply_schema_mutation_inner( measurement, predicate_sql, } => { - if !predicate_sql.is_empty() && measurement.is_some() { - metadata - .store_tombstone( - &database, - &rp, - measurement.as_deref().unwrap_or(""), - &predicate_sql, - ) + if let Some(meas) = measurement.as_deref() { + let sink = points_sink.or_else(|| mv_service.map(|mv| mv.points_sink())); + delete_replicated_rows(metadata, sink, wal, &database, &rp, meas, &predicate_sql) .await?; } metadata diff --git a/hyperbytedb/src/application/shard_routing.rs b/hyperbytedb/src/application/shard_routing.rs index 5fcb0fb..4d0e0ed 100644 --- a/hyperbytedb/src/application/shard_routing.rs +++ b/hyperbytedb/src/application/shard_routing.rs @@ -1001,6 +1001,7 @@ impl ShardDeleteOutcome { pub async fn scatter_delete_to_regions( ctx: &ShardRoutingContext, metadata: &dyn MetadataPort, + local: Option>, db: &str, rp: &str, measurement: &str, @@ -1014,13 +1015,31 @@ pub async fn scatter_delete_to_regions( }; let mut outcome = ShardDeleteOutcome::default(); + // One delete removes the rows of every local region. + let mut deleted_locally = false; for region in &space.regions { if region.primary == ctx.node_id { - if !predicate.is_empty() { - metadata - .store_tombstone(db, rp, measurement, predicate) - .await?; + if !predicate.is_empty() && !deleted_locally { + match local { + Some(local) => { + crate::application::predicate_sql::delete_rows( + local.wal, + local.sink, + db, + rp, + measurement, + predicate, + ) + .await?; + } + None => { + metadata + .store_tombstone(db, rp, measurement, predicate) + .await?; + } + } + deleted_locally = true; } metadata .delete_series_matching(db, rp, Some(measurement), predicate) diff --git a/hyperbytedb/src/error.rs b/hyperbytedb/src/error.rs index 327a18e..9075f6a 100644 --- a/hyperbytedb/src/error.rs +++ b/hyperbytedb/src/error.rs @@ -104,6 +104,11 @@ pub enum HyperbytedbError { #[error("error parsing query: cannot write directly to materialized view destination \"{0}\"")] WriteToMaterializedDestination(String), + /// A new measurement, field or tag key whose sanitised name is another + /// one's, so both would be stored in one table or column. + #[error("name collision: {0}")] + NameCollision(String), + #[error("authorization failed")] AuthFailed, diff --git a/hyperbytedb/src/ports/metadata.rs b/hyperbytedb/src/ports/metadata.rs index 7bde56c..5ee2505 100644 --- a/hyperbytedb/src/ports/metadata.rs +++ b/hyperbytedb/src/ports/metadata.rs @@ -45,6 +45,15 @@ pub trait MetadataPort: Send + Sync { Ok(0) } async fn drop_database(&self, name: &str) -> Result<(), HyperbytedbError>; + + /// A counter that grows each time a database, retention policy, + /// measurement or series is dropped. In-memory caches of what has been + /// registered (the ingest schema cache) start over when it changes: + /// otherwise a measurement written again after `DROP MEASUREMENT` was + /// taken to be registered still, and lost its fields. + fn drop_generation(&self) -> u64 { + 0 + } async fn list_databases(&self) -> Result, HyperbytedbError>; async fn get_database(&self, name: &str) -> Result, HyperbytedbError>; diff --git a/hyperbytedb/src/ports/points_sink.rs b/hyperbytedb/src/ports/points_sink.rs index 39bc5e7..bf979c3 100644 --- a/hyperbytedb/src/ports/points_sink.rs +++ b/hyperbytedb/src/ports/points_sink.rs @@ -93,6 +93,26 @@ pub trait PointsSinkPort: Send + Sync { Ok(()) } + /// Delete the flushed fact rows that `predicate` selects (see + /// [`crate::application::predicate_sql::build_delete_predicate`]), and + /// only those that exist now: rows inserted later are kept. + /// Whether [`Self::delete_rows`] deletes anything: the default sink + /// stores no rows, so a delete need not wait for the WAL to flush to it. + fn deletes_rows(&self) -> bool { + false + } + + async fn delete_rows( + &self, + db: &str, + rp: &str, + measurement: &str, + predicate: &str, + ) -> Result<(), HyperbytedbError> { + let _ = (db, rp, measurement, predicate); + Ok(()) + } + /// Delete flushed fact and series rows whose `series_id` lies in `[start, end)`. async fn delete_series_id_range( &self, diff --git a/hyperbytedb/src/timeseriesql/lexer.rs b/hyperbytedb/src/timeseriesql/lexer.rs index adee631..41325f3 100644 --- a/hyperbytedb/src/timeseriesql/lexer.rs +++ b/hyperbytedb/src/timeseriesql/lexer.rs @@ -213,6 +213,51 @@ pub fn parse_duration_text(input: &str) -> Result, HyperbytedbError> Ok(Some(total)) } +/// Byte length of the longest prefix of `rest` that [`parse_duration_text`] +/// accepts and that has a unit, with the whitespace after it; `None` when +/// there is none. +/// +/// One pass over the terms (`[+-]digits unit`, whitespace between). Trying +/// each prefix in turn instead re-parsed the input once per byte, quadratic +/// in its length, and sliced it inside multi-byte characters (`5温` +/// panicked). +fn longest_duration_prefix(rest: &str) -> Option { + const UNITS: [&str; 10] = ["ns", "us", "µs", "u", "ms", "s", "m", "h", "d", "w"]; + let bytes = rest.as_bytes(); + let mut i = 0; + let mut best = None; + loop { + while i < bytes.len() && bytes[i].is_ascii_whitespace() { + i += 1; + } + let term = i; + if i < bytes.len() && (bytes[i] == b'-' || bytes[i] == b'+') { + i += 1; + } + let digits = i; + while i < bytes.len() && bytes[i].is_ascii_digit() { + i += 1; + } + if i == digits { + break; + } + match rest[term..i].parse::() { + Ok(v) if v >= 0 => {} + _ => break, + } + let Some(unit) = UNITS.iter().find(|u| rest[i..].starts_with(*u)) else { + break; + }; + i += unit.len(); + let mut end = i; + while end < bytes.len() && bytes[end].is_ascii_whitespace() { + end += 1; + } + best = Some(end); + } + best +} + /// Convert nanoseconds to a single-term AST Duration (largest exact unit). pub fn nanos_to_ast_duration(nanos: i64) -> Duration { const H: i64 = 3_600 * 1_000_000_000; @@ -682,21 +727,13 @@ impl<'a> Lexer<'a> { fn read_number_or_duration(&mut self, start: usize) -> Result { let rest = &self.input[start..]; - let mut best: Option<(Option, usize)> = None; - for end in 1..=rest.len() { - if let Ok(nanos) = parse_duration_text(&rest[..end]) { - let slice = rest[..end].trim(); - // Only a unit-bearing literal (e.g. `0s`, `1h30m`, `5µs`) is a - // duration here. A bare integer — including `0` — is a Number; - // the "0 == infinite" rule is applied where a duration is - // grammatically expected (see ddl_parser::parse_duration_token), - // so `LIMIT 0`/`OFFSET 0` still get a real number token. - let is_duration = slice.chars().any(|c| c.is_ascii_alphabetic() || c == 'µ'); - if is_duration { - best = Some((nanos, end)); - } - } - } + // Only a unit-bearing literal (e.g. `0s`, `1h30m`, `5µs`) is a + // duration here. A bare integer — including `0` — is a Number; the + // "0 == infinite" rule is applied where a duration is grammatically + // expected (see ddl_parser::parse_duration_token), so `LIMIT 0` / + // `OFFSET 0` still get a real number token. + let best = longest_duration_prefix(rest) + .and_then(|end| parse_duration_text(&rest[..end]).ok().map(|n| (n, end))); if let Some((nanos, consumed)) = best { let byte_end = start + consumed; while self.pos < self.chars.len() && self.chars[self.pos].0 < byte_end { @@ -848,6 +885,66 @@ fn is_keyword(word: &str) -> bool { mod tests { use super::*; + /// The scan `longest_duration_prefix` replaced: every prefix in turn, + /// on character boundaries. + fn longest_duration_prefix_slow(rest: &str) -> Option { + let mut best = None; + for (end, _) in rest.char_indices().skip(1).chain([(rest.len(), ' ')]) { + if parse_duration_text(&rest[..end]).is_ok() + && rest[..end] + .chars() + .any(|c| c.is_ascii_alphabetic() || c == 'µ') + { + best = Some(end); + } + } + best + } + + #[test] + fn duration_prefix_matches_the_per_prefix_scan() { + let alphabet: Vec = "0123456789 nsuµmhdwx+-.温".chars().collect(); + let mut state: u64 = 0x2545_f491_4f6c_dd1d; + for _ in 0..50_000 { + let mut s = String::from("1"); + let len = (state >> 59) as usize; + for _ in 0..len { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + s.push(alphabet[(state % alphabet.len() as u64) as usize]); + } + assert_eq!( + longest_duration_prefix(&s), + longest_duration_prefix_slow(&s), + "{s:?}" + ); + } + } + + #[test] + fn number_before_a_multibyte_char_does_not_panic() { + for q in ["SELECT 5温 FROM cpu", "SELECT 5µ FROM cpu", "5温"] { + assert!(std::panic::catch_unwind(|| tokenize(q)).is_ok(), "{q:?}"); + } + assert_eq!( + kinds("5µs"), + vec![TokenKind::Duration { nanos: Some(5_000) }] + ); + } + + #[test] + fn long_number_runs_lex_in_linear_time() { + let q = format!("SELECT {} FROM cpu", "1h".repeat(200_000)); + let t = std::time::Instant::now(); + let _ = tokenize(&q); + assert!( + t.elapsed() < std::time::Duration::from_secs(2), + "{:?}", + t.elapsed() + ); + } + fn kinds(input: &str) -> Vec { tokenize(input) .unwrap() diff --git a/hyperbytedb/src/timeseriesql/parser.rs b/hyperbytedb/src/timeseriesql/parser.rs index 35f45b9..ce805e1 100644 --- a/hyperbytedb/src/timeseriesql/parser.rs +++ b/hyperbytedb/src/timeseriesql/parser.rs @@ -278,7 +278,69 @@ fn parse_field_expr(input: &str) -> Result { Ok(Field { expr, alias }) } +/// Deepest expression nesting [`parse_expr`] accepts: parentheses, +/// arithmetic chains and the like. `AND` / `OR` chains of any length are +/// built as balanced trees and stay far below it. +/// +/// The parser, and every pass over the tree after it, recurses once per +/// level. Without a bound, a query with a few thousand `OR` terms (when +/// they nested one per term) overflowed the thread's stack; chDB's signal +/// handler took the fault for its own and shut the query engine down for +/// good, while `/health` kept answering 200. +pub const MAX_EXPR_DEPTH: usize = 100; + +/// Deepest nesting of subqueries (`FROM (SELECT ...)`) accepted. Each level +/// costs far more stack than an expression level, here and when the query +/// is planned and run. +pub const MAX_SUBQUERY_DEPTH: usize = 16; + +thread_local! { + static EXPR_DEPTH: std::cell::Cell = const { std::cell::Cell::new(0) }; + static SUBQUERY_DEPTH: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +/// One level of parser recursion, counted while it lives. +struct ExprDepth(&'static std::thread::LocalKey>); + +impl ExprDepth { + fn enter() -> Result { + Self::enter_on(&EXPR_DEPTH, MAX_EXPR_DEPTH, "expression") + } + + fn enter_subquery() -> Result { + Self::enter_on(&SUBQUERY_DEPTH, MAX_SUBQUERY_DEPTH, "subquery") + } + + fn enter_on( + counter: &'static std::thread::LocalKey>, + max: usize, + what: &str, + ) -> Result { + counter.with(|d| { + let depth = d.get() + 1; + if depth > max { + return Err(HyperbytedbError::QueryParse(format!( + "{what} nested too deeply (more than {max} levels)" + ))); + } + d.set(depth); + Ok(Self(counter)) + }) + } +} + +impl Drop for ExprDepth { + fn drop(&mut self) { + self.0.with(|d| d.set(d.get().saturating_sub(1))); + } +} + pub fn parse_expr(input: &str) -> Result { + let _depth = ExprDepth::enter()?; + parse_expr_at_depth(input) +} + +fn parse_expr_at_depth(input: &str) -> Result { let input = input.trim(); // Regex literal must be checked early to avoid treating / as division @@ -328,21 +390,55 @@ fn try_parse_logical_expr(input: &str) -> Result, HyperbytedbError> let scan = scan_chars(input)?; // OR has the lowest precedence in InfluxQL, so split at OR first: the // operator split earliest ends up at the root of the tree and binds - // loosest. Which OR occurrence is split at is semantically neutral. + // loosest. + // + // A chain is split at every top-level occurrence at once and its operands + // joined as a balanced tree (both operators are associative). Splitting + // at the first one and recursing on the rest nested one level per term: + // quadratic time, and a stack overflow at a few thousand terms. for (kw, op) in [("OR", BinaryOp::Or), ("AND", BinaryOp::And)] { - if let Some((pos, end)) = find_keyword_position(input, &scan, kw) { - let left = parse_expr(&input[..pos])?; - let right = parse_expr(&input[end..])?; - return Ok(Some(Expr::BinaryExpr(Box::new(BinaryExpr { - left, - op, - right, - })))); + let mut cuts: Vec<(usize, usize)> = Vec::new(); + let mut i = 0; + while i < scan.len() { + match match_keyword_at(input, &scan, i, kw) { + Some((pos, end)) => { + cuts.push((pos, end)); + while i < scan.len() && scan[i].idx < end { + i += 1; + } + } + None => i += 1, + } } + if cuts.is_empty() { + continue; + } + let mut operands = Vec::with_capacity(cuts.len() + 1); + let mut start = 0; + for (pos, end) in cuts { + operands.push(parse_expr(&input[start..pos])?); + start = end; + } + operands.push(parse_expr(&input[start..])?); + return Ok(balanced_chain(operands, &op)); } Ok(None) } +/// `a op b op c ...` as a tree of depth log2(n). Three operands give +/// `a op (b op c)`, as splitting at the first operator did. +fn balanced_chain(mut operands: Vec, op: &BinaryOp) -> Option { + if operands.len() <= 1 { + return operands.pop(); + } + let right = operands.split_off(operands.len() / 2); + Some(Expr::BinaryExpr(Box::new(BinaryExpr { + left: balanced_chain(operands, op)?, + op: op.clone(), + right: balanced_chain(right, op)?, + }))) +} + fn try_parse_comparison_expr(input: &str) -> Result, HyperbytedbError> { let operators = [ ("=~", BinaryOp::RegexMatch), @@ -477,7 +573,7 @@ fn parse_atom(input: &str) -> Result { } // String literal 'value' — both `\'` and `''` escape a quote - if input.starts_with('\'') && input.ends_with('\'') { + if input.len() >= 2 && input.starts_with('\'') && input.ends_with('\'') { let s = input[1..input.len() - 1] .replace("\\'", "'") .replace("''", "'"); @@ -522,7 +618,7 @@ fn parse_atom(input: &str) -> Result { } // Quoted identifier "name" - if input.starts_with('"') && input.ends_with('"') { + if input.len() >= 2 && input.starts_with('"') && input.ends_with('"') { let name = input[1..input.len() - 1].to_string(); return Ok(Expr::Identifier(name)); } @@ -603,6 +699,7 @@ fn parse_from_sources(input: &str) -> Result, Hyperbytedb if input.starts_with('(') && input.ends_with(')') { let inner = input[1..input.len() - 1].trim(); if starts_with_keyword(inner, "SELECT") { + let _depth = ExprDepth::enter_subquery()?; let stmt = parse_select(inner)?; if let Statement::Select(sub) = stmt { return Ok(vec![MeasurementSource::Subquery(Box::new(sub))]); @@ -625,7 +722,7 @@ fn parse_measurement(input: &str) -> Result { let input = input.trim(); // Regex measurement /pattern/ - if input.starts_with('/') && input.ends_with('/') { + if input.len() >= 2 && input.starts_with('/') && input.ends_with('/') { return Ok(Measurement { database: None, retention_policy: None, @@ -728,7 +825,7 @@ fn parse_group_by_clause(input: &str) -> Result<(GroupBy, Option), H dimensions.push(Dimension::Time { interval, offset }); } else if part == "*" { dimensions.push(Dimension::AllTags); - } else if part.starts_with('/') && part.ends_with('/') { + } else if part.len() >= 2 && part.starts_with('/') && part.ends_with('/') { dimensions.push(Dimension::Regex(part[1..part.len() - 1].to_string())); } else { dimensions.push(Dimension::Tag(unquote(part))); @@ -788,6 +885,96 @@ fn parse_order_by(input: &str) -> Result { #[cfg(test)] mod tests { + /// A few thousand `OR` terms nested one level each and overflowed the + /// stack (taking chDB down with it); they now parse into a shallow tree. + #[test] + fn long_or_and_chains_parse_shallow() { + fn depth(e: &Expr) -> usize { + match e { + Expr::BinaryExpr(b) => 1 + depth(&b.left).max(depth(&b.right)), + _ => 1, + } + } + for (kw, n) in [("OR", 4000), ("AND", 4000), ("OR", 50_000)] { + let q = format!( + "SELECT usage FROM cpu WHERE usage > 1{}", + format!(" {kw} usage > 1").repeat(n) + ); + let t = std::time::Instant::now(); + let stmts = std::thread::Builder::new() + .stack_size(256 * 1024) + .spawn(move || crate::timeseriesql::parse(&q)) + .unwrap() + .join() + .unwrap() + .unwrap(); + assert!( + t.elapsed() < std::time::Duration::from_secs(10), + "{kw} x{n}" + ); + let Statement::Select(sel) = &stmts[0] else { + panic!("not a select") + }; + let d = depth(sel.condition.as_ref().unwrap()); + assert!(d <= 20, "{kw} x{n}: depth {d}"); + } + } + + #[test] + fn three_term_chain_keeps_its_shape() { + let stmts = + crate::timeseriesql::parse("SELECT v FROM m WHERE a = 1 OR b = 2 OR c = 3").unwrap(); + let Statement::Select(sel) = &stmts[0] else { + panic!("not a select") + }; + let Some(Expr::BinaryExpr(root)) = &sel.condition else { + panic!("no condition") + }; + assert!(matches!(&root.right, Expr::BinaryExpr(r) if r.op == BinaryOp::Or)); + } + + #[test] + fn deep_nesting_is_an_error_not_a_crash() { + let parens = format!( + "SELECT v FROM m WHERE {}a = 1{}", + "(".repeat(5000), + ")".repeat(5000) + ); + let sum = format!("SELECT v{} FROM m", " + v".repeat(5000)); + let subq = format!( + "SELECT v FROM {}m{}", + "(SELECT v FROM ".repeat(500), + ")".repeat(500) + ); + // Debug-build frames; a tokio worker has 2 MiB. + for q in [parens, sum, subq] { + let r = std::thread::Builder::new() + .stack_size(1024 * 1024) + .spawn(move || crate::timeseriesql::parse(&q).is_err()) + .unwrap() + .join() + .unwrap(); + assert!(r); + } + // The depth counter unwinds: an ordinary query still parses after. + assert!(crate::timeseriesql::parse("SELECT v FROM m WHERE (a = 1)").is_ok()); + } + + #[test] + fn lone_quotes_and_slashes_are_errors_not_panics() { + for q in [ + "SELECT usage FROM cpu WHERE host = '", + "SELECT \"", + "SELECT '", + "SELECT usage FROM /", + "SELECT usage FROM cpu GROUP BY /", + "SELECT usage FROM cpu WHERE host = \"", + ] { + let r = std::panic::catch_unwind(|| crate::timeseriesql::parse(q)); + assert!(r.is_ok(), "panicked on {q:?}"); + } + } + use super::*; #[test] diff --git a/hyperbytedb/tests/compat/ddl_tests.rs b/hyperbytedb/tests/compat/ddl_tests.rs index 412021b..4ca90e6 100644 --- a/hyperbytedb/tests/compat/ddl_tests.rs +++ b/hyperbytedb/tests/compat/ddl_tests.rs @@ -297,62 +297,215 @@ async fn drop_measurement_removes_associated_metadata() { // DELETE // --------------------------------------------------------------------------- +/// The `value`s a query returns, in time order; empty for no series. +async fn values(ctx: &TestContext, q: &str) -> Vec { + let resp = ctx.query("testdb", q).await.unwrap(); + let stmt = &resp.results[0]; + assert!(stmt.error.is_none(), "{q}: {:?}", stmt.error); + let Some(series) = stmt.series.as_ref() else { + return Vec::new(); + }; + let mut out = Vec::new(); + for s in series { + let col = s.columns.iter().position(|c| c == "value").unwrap(); + for row in &s.values { + out.push(row[col].as_f64().unwrap()); + } + } + out +} + +fn chdb_context() -> Option { + match TestContext::new() { + Ok(c) => Some(c), + Err(_) => { + eprintln!("skipping DELETE test: chDB not available"); + None + } + } +} + #[tokio::test] -async fn delete_from_measurement_where_time() { - let ctx = TestContext::new_no_chdb().unwrap(); +#[serial(chdb)] +async fn delete_where_time_removes_only_older_points() { + let Some(ctx) = chdb_context() else { return }; ctx.metadata.create_database("testdb").await.unwrap(); - ctx.ingestion - .ingest( - "testdb", - None, - None, - b"cpu value=1.0 1000000000\ncpu value=2.0 2000000000", - WritePayloadFormat::LineProtocol, - ) + ctx.write_and_flush( + "testdb", + "cpu value=1.0 1000000000\ncpu value=2.0 2000000000", + ) + .await + .unwrap(); + + let resp = ctx + .query("testdb", "DELETE FROM cpu WHERE time < 1500000000") .await .unwrap(); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![2.0]); + assert!( + ctx.metadata + .list_tombstones("testdb", "autogen", "cpu") + .await + .unwrap() + .is_empty(), + "DELETE removes rows; it must not leave a tombstone for every later query" + ); +} + +#[tokio::test] +#[serial(chdb)] +async fn delete_where_tag_keeps_points_written_after_it() { + let Some(ctx) = chdb_context() else { return }; + ctx.metadata.create_database("testdb").await.unwrap(); + ctx.write_and_flush( + "testdb", + "cpu,host=a value=1.0 1000000000\ncpu,host=b value=2.0 2000000000", + ) + .await + .unwrap(); let resp = ctx - .query("testdb", "DELETE FROM cpu WHERE time < 1500000000") + .query("testdb", r#"DELETE FROM cpu WHERE "host" = 'a'"#) .await .unwrap(); - assert!(resp.results[0].error.is_none()); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![2.0]); - let tombstones = ctx - .metadata - .list_tombstones("testdb", "autogen", "cpu") + // InfluxDB deletes what exists; a point written afterwards is kept. + ctx.write_and_flush("testdb", "cpu,host=a value=3.0 3000000000") .await .unwrap(); - assert!(!tombstones.is_empty(), "DELETE should create a tombstone"); + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![2.0, 3.0]); } #[tokio::test] -async fn delete_from_measurement_where_tag() { - let ctx = TestContext::new_no_chdb().unwrap(); +#[serial(chdb)] +async fn delete_with_tag_and_time_or_regex() { + let Some(ctx) = chdb_context() else { return }; ctx.metadata.create_database("testdb").await.unwrap(); - ctx.ingestion - .ingest( + ctx.write_and_flush( + "testdb", + "cpu,host=a value=1.0 1000000000\n\ + cpu,host=a value=2.0 2000000000\n\ + cpu,host=b value=3.0 3000000000\n\ + cpu,host=c value=4.0 4000000000", + ) + .await + .unwrap(); + + let resp = ctx + .query( "testdb", - None, - None, - b"cpu,host=a value=1.0 1000000000\ncpu,host=b value=2.0 2000000000", - WritePayloadFormat::LineProtocol, + r#"DELETE FROM cpu WHERE ("host" = 'a' AND time < 1500000000) OR "host" !~ /[ab]/"#, ) .await .unwrap(); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![2.0, 3.0]); +} + +#[tokio::test] +#[serial(chdb)] +async fn delete_without_where_removes_every_point() { + let Some(ctx) = chdb_context() else { return }; + ctx.metadata.create_database("testdb").await.unwrap(); + ctx.write_and_flush("testdb", "cpu,host=a value=1.0 1000000000") + .await + .unwrap(); + let resp = ctx.query("testdb", "DELETE FROM cpu").await.unwrap(); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + assert!(values(&ctx, "SELECT value FROM cpu").await.is_empty()); +} +#[tokio::test] +#[serial(chdb)] +async fn delete_naming_a_field_or_unknown_key_is_refused() { + let Some(ctx) = chdb_context() else { return }; + ctx.metadata.create_database("testdb").await.unwrap(); + ctx.write_and_flush("testdb", "cpu,host=a value=1.0 1000000000") + .await + .unwrap(); + for q in [ + "DELETE FROM cpu WHERE value > 0", + "DELETE FROM cpu WHERE tim0 < 0", + ] { + let failed = match ctx.query("testdb", q).await { + Err(_) => true, + Ok(resp) => resp.results[0].error.is_some(), + }; + assert!(failed, "{q} must be refused"); + } + // Nothing was stored that breaks later queries. + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![1.0]); +} + +#[tokio::test] +#[serial(chdb)] +async fn drop_series_keeps_points_written_after_it() { + let Some(ctx) = chdb_context() else { return }; + ctx.metadata.create_database("testdb").await.unwrap(); + ctx.write_and_flush( + "testdb", + "cpu,host=a value=1.0 1000000000\ncpu,host=b value=2.0 2000000000", + ) + .await + .unwrap(); let resp = ctx - .query( - "testdb", - r#"DELETE FROM cpu WHERE "host" = 'a' AND time < 2000000000"#, - ) + .query("testdb", r#"DROP SERIES FROM cpu WHERE "host" = 'a'"#) + .await + .unwrap(); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![2.0]); + ctx.write_and_flush("testdb", "cpu,host=a value=3.0 3000000000") .await - .expect("DELETE with tag+time predicate should parse and execute"); + .unwrap(); + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![2.0, 3.0]); +} + +#[tokio::test] +#[serial(chdb)] +async fn drop_measurement_clears_its_tombstones() { + let Some(ctx) = chdb_context() else { return }; + ctx.metadata.create_database("testdb").await.unwrap(); + ctx.write_and_flush("testdb", "cpu,host=a value=1.0 1000000000") + .await + .unwrap(); + // A tombstone stored by an earlier version, naming a column cpu lacks. + ctx.metadata + .store_tombstone("testdb", "autogen", "cpu", "tim0 < 0") + .await + .unwrap(); + let resp = ctx.query("testdb", "DROP MEASUREMENT cpu").await.unwrap(); assert!( resp.results[0].error.is_none(), - "DELETE should succeed: {:?}", + "{:?}", resp.results[0].error ); + ctx.write_and_flush("testdb", "cpu,host=a value=2.0 2000000000") + .await + .unwrap(); + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![2.0]); } // --------------------------------------------------------------------------- @@ -1634,3 +1787,26 @@ async fn drop_database_removes_its_series_metadata() { "series keys of a dropped database" ); } + +/// The ingest schema cache took a measurement written again after DROP +/// MEASUREMENT to be registered still: its fields were never registered +/// again, and every query on it failed with "Unknown expression identifier". +#[tokio::test] +#[serial(chdb)] +async fn measurement_written_again_after_drop_keeps_its_fields() { + let Some(ctx) = chdb_context() else { return }; + ctx.metadata.create_database("testdb").await.unwrap(); + ctx.write_and_flush("testdb", "cpu,host=a value=1.0 1000000000") + .await + .unwrap(); + let resp = ctx.query("testdb", "DROP MEASUREMENT cpu").await.unwrap(); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + ctx.write_and_flush("testdb", "cpu,host=a value=2.0 2000000000") + .await + .unwrap(); + assert_eq!(values(&ctx, "SELECT value FROM cpu").await, vec![2.0]); +} diff --git a/hyperbytedb/tests/compat/query_tests.rs b/hyperbytedb/tests/compat/query_tests.rs index 0835cfa..90be06c 100644 --- a/hyperbytedb/tests/compat/query_tests.rs +++ b/hyperbytedb/tests/compat/query_tests.rs @@ -1434,3 +1434,32 @@ async fn top_and_subqueries() { .await; assert_eq!(series[0].values[0][1], serde_json::json!(122.5)); } + +/// A WHERE clause of 4000 `OR` terms nested one level per term and overflowed +/// the stack while parsing; chDB's signal handler took the fault and shut the +/// query engine down for good. It must run, and later queries with it. +#[tokio::test] +#[serial(chdb)] +async fn where_clause_with_thousands_of_or_terms() { + let ctx = TestContext::new().unwrap(); + ctx.metadata.create_database("testdb").await.unwrap(); + ctx.write_and_flush("testdb", "cpu,host=a usage=5.0 1000000000") + .await + .unwrap(); + let q = format!( + "SELECT usage FROM cpu WHERE usage > 100{} OR usage > 1", + " OR usage > 100".repeat(4000) + ); + let resp = ctx.query("testdb", &q).await.unwrap(); + let stmt = &resp.results[0]; + assert!(stmt.error.is_none(), "{:?}", stmt.error); + assert_eq!(stmt.series.as_ref().unwrap()[0].values.len(), 1); + + let resp = ctx.query("testdb", "SELECT usage FROM cpu").await.unwrap(); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + assert!(!hyperbytedb::adapters::chdb::engine_failed()); +} diff --git a/hyperbytedb/tests/compat/write_tests.rs b/hyperbytedb/tests/compat/write_tests.rs index f2e6be5..12f55d2 100644 --- a/hyperbytedb/tests/compat/write_tests.rs +++ b/hyperbytedb/tests/compat/write_tests.rs @@ -693,3 +693,56 @@ async fn field_type_conflicts_keep_the_other_points() { )); assert_eq!(ctx.wal.read_from(1).await.unwrap().len(), 2); } + +/// `cpu.load` and `cpu_load` are one table (`.` becomes `_`), and database +/// `a` with retention policy `b_c` is database `a_b` with policy `c`: the +/// second name of each pair read and overwrote the first one's points. The +/// write path refuses it before the WAL. +#[tokio::test] +async fn names_that_share_storage_are_refused() { + use hyperbytedb::error::HyperbytedbError; + let ctx = TestContext::new_no_chdb().unwrap(); + ctx.metadata.create_database("db").await.unwrap(); + let write = |db: &'static str, rp: Option<&'static str>, body: &'static str| { + ctx.ingestion.ingest( + db, + rp, + None, + body.as_bytes(), + WritePayloadFormat::LineProtocol, + ) + }; + write("db", None, "cpu.load,host=a v=1 1").await.unwrap(); + let err = write("db", None, "cpu_load,host=a v=2 2") + .await + .unwrap_err(); + assert!(matches!(err, HyperbytedbError::NameCollision(_)), "{err:?}"); + // The owner still writes; a field sharing a column is refused too. + write("db", None, "cpu.load,host=a v=3 3").await.unwrap(); + let err = write("db", None, "cpu.load,host=a v.=4,v_=5 4") + .await + .unwrap_err(); + assert!(matches!(err, HyperbytedbError::NameCollision(_)), "{err:?}"); + + ctx.metadata.create_database("a_b").await.unwrap(); + ctx.metadata.create_database("a").await.unwrap(); + ctx.query( + "a_b", + "CREATE RETENTION POLICY c ON a_b DURATION 1d REPLICATION 1", + ) + .await + .unwrap(); + ctx.query( + "a", + "CREATE RETENTION POLICY b_c ON a DURATION 1d REPLICATION 1", + ) + .await + .unwrap(); + write("a_b", Some("c"), "m,host=x secret=42 1") + .await + .unwrap(); + let err = write("a", Some("b_c"), "m,host=x secret=999 1") + .await + .unwrap_err(); + assert!(matches!(err, HyperbytedbError::NameCollision(_)), "{err:?}"); +}