Fix query-engine crash, name collisions, DELETE semantics and parser panics

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);
+    }
+}

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)

hyperbytedb/src/adapters/chdb/session.rs

@@ -23,7 +23,11 @@ pub(crate) fn execute_connection(
     query: &str,
     format: OutputFormat,
 ) -> Result<QueryResult, chdb_rust::error::Error> {
-    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.

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,

hyperbytedb/src/adapters/http/ping.rs

@@ -233,6 +233,17 @@ pub async fn health(State(state): State<Arc<AppState>>) -> 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<Arc<AppState>>) -> 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,

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,

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<HashMap<String, Database>>,
     meas_cache: RwLock<HashMap<String, MeasurementMeta>>,
     /// 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<usize, HyperbytedbError> {
+        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,17 +2098,41 @@ 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())
             })?;
             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(())
     }