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<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.
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<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,
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<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,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<HashSet<u64, FastHash>>,
+    /// [`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<String, u8> = 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<Arc<ShardRoutingContext>>,
     is_raft_leader: Arc<dyn Fn() -> 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 `<db>_<rp>_<measurement>`.
+//! 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<HashSet<u64>>,
+    /// [`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<dyn MetadataPort>,
+        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<dyn MetadataPort>,
+        db: &str,
+        rp: &str,
+        measurement: &str,
+        field: &str,
+        tags: impl Iterator<Item = &str>,
+    ) -> 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<dyn MetadataPort>,
+        db: &str,
+        rp: &str,
+        measurement: &str,
+        names: &BatchNames<'_>,
+    ) -> Result<(), HyperbytedbError> {
+        let mut hashes: Vec<u64> = 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<dyn MetadataPort>,
+    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<String, &str> =
+        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<dyn MetadataPort> {
+        let meta: Arc<dyn MetadataPort> = 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<dyn MetadataPort>,
+        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<String>) {
         _ => {}
     }
 }
+
+/// The fact-table condition that selects the rows a `DELETE` or
+/// `DROP SERIES` removes, for `ALTER TABLE <fact> DELETE WHERE <it>`.
+///
+/// Tags live in the series table, so each comparison on a tag becomes a
+/// series lookup (`series_id IN (SELECT series_id FROM <series> 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<dyn MetadataPort>,
+    db: &str,
+    rp: &str,
+    measurement: &str,
+    cond: Option<&Expr>,
+) -> Result<String, HyperbytedbError> {
+    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<dyn MetadataPort>,
+    sink: Option<&Arc<dyn PointsSinkPort>>,
+    wal: Option<&Arc<dyn WalPort>>,
+    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<crate::application::predicate_sql::LocalRows<'_>>,
     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<Vec<Database>, HyperbytedbError>;
     async fn get_database(&self, name: &str) -> Result<Option<Database>, 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<Option<i64>, 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<usize> {
+    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::<i64>() {
+            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<Token, HyperbytedbError> {
         let rest = &self.input[start..];
-        let mut best: Option<(Option<i64>, 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<usize> {
+        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<char> = "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<TokenKind> {
         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<Field, HyperbytedbError> {
     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<usize> = const { std::cell::Cell::new(0) };
+    static SUBQUERY_DEPTH: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
+}
+
+/// One level of parser recursion, counted while it lives.
+struct ExprDepth(&'static std::thread::LocalKey<std::cell::Cell<usize>>);
+
+impl ExprDepth {
+    fn enter() -> Result<Self, HyperbytedbError> {
+        Self::enter_on(&EXPR_DEPTH, MAX_EXPR_DEPTH, "expression")
+    }
+
+    fn enter_subquery() -> Result<Self, HyperbytedbError> {
+        Self::enter_on(&SUBQUERY_DEPTH, MAX_SUBQUERY_DEPTH, "subquery")
+    }
+
+    fn enter_on(
+        counter: &'static std::thread::LocalKey<std::cell::Cell<usize>>,
+        max: usize,
+        what: &str,
+    ) -> Result<Self, HyperbytedbError> {
+        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<Expr, HyperbytedbError> {
+    let _depth = ExprDepth::enter()?;
+    parse_expr_at_depth(input)
+}
+
+fn parse_expr_at_depth(input: &str) -> Result<Expr, HyperbytedbError> {
     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<Option<Expr>, 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<Expr>, op: &BinaryOp) -> Option<Expr> {
+    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<Option<Expr>, HyperbytedbError> {
     let operators = [
         ("=~", BinaryOp::RegexMatch),
@@ -477,7 +573,7 @@ fn parse_atom(input: &str) -> Result<Expr, HyperbytedbError> {
     }
 
     // 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<Expr, HyperbytedbError> {
     }
 
     // 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<Vec<MeasurementSource>, 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<Measurement, HyperbytedbError> {
     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<FillOption>), 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<OrderBy, HyperbytedbError> {
 
 #[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<f64> {
+    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<TestContext> {
+    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:?}");
+}