fix: harden storage consistency and blob admission

crates/lance-context-core/src/datagen_store.rs

@@ -195,6 +195,12 @@ impl DatagenStore {
         self.base.close().await
     }
 
+    /// Stop this handle's writer and delete the dataset through its backend.
+    /// Remote writers must be quiesced by the caller before deletion.
+    pub async fn delete_data(&mut self) -> LanceResult<()> {
+        self.base.delete_data().await
+    }
+
     /// Read one item's event history without materializing blob bytes.
     pub async fn events_for_item(&self, item_id: &str) -> LanceResult<Vec<DatagenEvent>> {
         self.filtered_events(&format!("item_id = '{}'", escape_sql_literal(item_id)))

crates/lance-context-core/src/generic_store.rs

@@ -362,6 +362,12 @@ impl GenericStore {
         self.base.close().await
     }
 
+    /// Stop this handle's writer and delete the dataset through its backend.
+    /// Remote writers must be quiesced by the caller before deletion.
+    pub async fn delete_data(&mut self) -> LanceResult<()> {
+        self.base.delete_data().await
+    }
+
     /// Merge flushed generations into the base table once the count trigger is
     /// met. Returns how many were reclaimed.
     pub async fn maybe_merge_wal(&mut self) -> LanceResult<usize> {

crates/lance-context-core/src/lib.rs

@@ -67,7 +67,9 @@ pub use rollout_store::{
 };
 // Schema declaration lives in the API crate: it is part of the wire contract.
 pub use lance_context_api::{ColumnSpec, ColumnType, SchemaSpec, ID_COLUMN};
-pub use storage::{create_local_dir_if_needed, join_uri, validate_store_name, MAX_STORE_NAME_LEN};
+pub use storage::{
+    create_local_dir_if_needed, join_uri, remove_dataset, validate_store_name, MAX_STORE_NAME_LEN,
+};
 pub use store::{
     CompactionConfig, CompactionStats, ContextStore, ContextStoreOptions, DistanceMetric,
     IdIndexType, ReadProjection,