fix(namespace): fence stale table deregistration

protos/table.proto

@@ -603,8 +603,8 @@ message FragmentReuseIndexDetails {
 // ============================================================================
 
 // Lifecycle status of a WAL shard. Drives drop-table two-phase commit:
-// a SEALED shard refuses new writer claims (reversible) until the drop
-// commits (the shard dir is deleted) or rolls back (status -> ACTIVE).
+// a SEALED shard refuses new writer claims while the outcome is unresolved;
+// a rollback transitions back to ACTIVE.
 enum ShardStatus {
   // Normal: the shard accepts writer claims.
   ACTIVE = 0;

rust/lance-namespace-impls/src/dir.rs

@@ -92,6 +92,8 @@ use crate::credentials::{
     CredentialVendor, create_credential_vendor_for_location, has_credential_vendor_config,
 };
 
+const EXPECTED_DEREGISTER_LOCATION_CONTEXT_KEY: &str = "expected_location";
+
 /// Thread-safe metrics tracker for namespace operations.
 ///
 /// Tracks the count of each API operation when `ops_metrics_enabled` is true.
@@ -271,6 +273,7 @@ pub struct DirectoryNamespaceBuilder {
     dir_listing_enabled: bool,
     inline_optimization_enabled: bool,
     table_version_tracking_enabled: bool,
+    async_drop_enabled: bool,
     /// When true, enables migration mode where the namespace checks the manifest first
     /// before falling back to directory listing for root-level tables. When false (default),
     /// root-level tables use directory listing directly without checking the manifest,
@@ -342,6 +345,7 @@ impl DirectoryNamespaceBuilder {
             dir_listing_enabled: true, // Default to enabled for backwards compatibility
             inline_optimization_enabled: true,
             table_version_tracking_enabled: false, // Default to disabled
+            async_drop_enabled: false,
             dir_listing_to_manifest_migration_enabled: false, // Default to disabled
             credential_vendor_properties: HashMap::new(),
             context_provider: None,
@@ -402,6 +406,12 @@ impl DirectoryNamespaceBuilder {
         self
     }
 
+    /// Enable unique physical generations required by expected-location drop fencing.
+    pub fn async_drop_enabled(mut self, enabled: bool) -> Self {
+        self.async_drop_enabled = enabled;
+        self
+    }
+
     /// Create a DirectoryNamespaceBuilder from properties HashMap.
     ///
     /// This method parses a properties map into builder configuration.
@@ -410,6 +420,7 @@ impl DirectoryNamespaceBuilder {
     /// - `manifest_enabled`: Enable manifest-based table tracking (optional, default: true)
     /// - `dir_listing_enabled`: Enable directory listing for table discovery (optional, default: true)
     /// - `inline_optimization_enabled`: Enable replacement indices on __manifest rewrites (optional, default: true)
+    /// - `async_drop_enabled`: Enable unique physical generations for asynchronous drop fencing (optional, default: false)
     /// - `storage.*`: Storage options (optional, prefix will be stripped)
     ///
     /// Credential vendor properties (prefixed with `credential_vendor.`, prefix is stripped):
@@ -519,6 +530,11 @@ impl DirectoryNamespaceBuilder {
             .and_then(|v| v.parse::<bool>().ok())
             .unwrap_or(false);
 
+        let async_drop_enabled = properties
+            .get("async_drop_enabled")
+            .and_then(|v| v.parse::<bool>().ok())
+            .unwrap_or(false);
+
         // Extract dir_listing_to_manifest_migration_enabled (default: false)
         let dir_listing_to_manifest_migration_enabled = properties
             .get("dir_listing_to_manifest_migration_enabled")
@@ -565,6 +581,7 @@ impl DirectoryNamespaceBuilder {
             dir_listing_enabled,
             inline_optimization_enabled,
             table_version_tracking_enabled,
+            async_drop_enabled,
             dir_listing_to_manifest_migration_enabled,
             credential_vendor_properties,
             context_provider: None,
@@ -741,6 +758,13 @@ impl DirectoryNamespaceBuilder {
     /// - Connection to the storage backend fails
     /// - Storage options are invalid
     pub async fn build(self) -> Result<DirectoryNamespace> {
+        if self.async_drop_enabled && self.manifest_enabled && self.dir_listing_enabled {
+            return Err(NamespaceError::InvalidInput {
+                message: "async_drop_enabled requires dir_listing_enabled=false when the manifest is enabled"
+                    .to_string(),
+            }
+            .into());
+        }
         let (object_store, base_path) =
             Self::initialize_object_store(&self.root, &self.storage_options, &self.session).await?;
 
@@ -752,6 +776,7 @@ impl DirectoryNamespaceBuilder {
                 object_store.clone(),
                 base_path.clone(),
                 self.dir_listing_enabled,
+                self.async_drop_enabled,
                 self.inline_optimization_enabled,
                 self.commit_retries,
             )
@@ -809,6 +834,7 @@ impl DirectoryNamespaceBuilder {
             dir_listing_to_manifest_migration_enabled: self
                 .dir_listing_to_manifest_migration_enabled,
             table_version_tracking_enabled: self.table_version_tracking_enabled,
+            async_drop_enabled: self.async_drop_enabled,
             credential_vendor,
             context_provider: self.context_provider,
             vend_input_storage_options: self.vend_input_storage_options,
@@ -895,6 +921,7 @@ pub struct DirectoryNamespace {
     /// When true, `describe_table` returns `managed_versioning: true` to indicate
     /// commits should go through namespace table version APIs.
     table_version_tracking_enabled: bool,
+    async_drop_enabled: bool,
     /// Credential vendor created once during initialization.
     /// Used to vend temporary credentials for table access.
     credential_vendor: Option<Arc<dyn CredentialVendor>>,
@@ -1030,6 +1057,7 @@ impl DirectoryNamespace {
                     self.object_store.clone(),
                     self.base_path.clone(),
                     self.dir_listing_enabled,
+                    self.async_drop_enabled,
                     self.inline_optimization_enabled,
                     self.commit_retries,
                 )
@@ -1040,6 +1068,17 @@ impl DirectoryNamespace {
         Ok(Some(manifest_ns.clone()))
     }
 
+    /// Retire a durable manifest drop intent after physical cleanup completes.
+    pub async fn complete_drop_tombstone(&self, tombstone_id: &str) -> Result<bool> {
+        let manifest_ns =
+            self.manifest_ns_for_write()
+                .await?
+                .ok_or_else(|| NamespaceError::InvalidInput {
+                    message: "drop tombstones require a manifest-enabled namespace".to_string(),
+                })?;
+        manifest_ns.complete_drop_tombstone(tombstone_id).await
+    }
+
     /// Lazily open the `__manifest` dataset (read-only) into the read cell.
     ///
     /// `manifest_ns` is populated at construction only if `__manifest` already
@@ -1068,6 +1107,7 @@ impl DirectoryNamespace {
                     self.object_store.clone(),
                     self.base_path.clone(),
                     self.dir_listing_enabled,
+                    self.async_drop_enabled,
                     self.inline_optimization_enabled,
                     self.commit_retries,
                 )
@@ -3687,6 +3727,18 @@ impl LanceNamespace for DirectoryNamespace {
             return LanceNamespace::deregister_table(manifest_ns.as_ref(), request).await;
         }
 
+        if request
+            .context
+            .as_ref()
+            .is_some_and(|context| context.contains_key(EXPECTED_DEREGISTER_LOCATION_CONTEXT_KEY))
+        {
+            return Err(NamespaceError::Unsupported {
+                message: "Expected-location fencing is unsupported when the manifest is disabled because table locations are deterministic"
+                    .to_string(),
+            }
+            .into());
+        }
+
         // V1 mode: create a .lance-deregistered marker file in the table directory
         let table_name = Self::table_name_from_id(&request.id)?;
         let table_uri = self.table_full_uri(&table_name);
@@ -9307,10 +9359,12 @@ mod tests {
         properties.insert("root".to_string(), temp_dir.to_str().unwrap().to_string());
         properties.insert("manifest_enabled".to_string(), "true".to_string());
         properties.insert("dir_listing_enabled".to_string(), "false".to_string());
+        properties.insert("async_drop_enabled".to_string(), "true".to_string());
 
         let builder = DirectoryNamespaceBuilder::from_properties(properties, None).unwrap();
         assert!(builder.manifest_enabled);
         assert!(!builder.dir_listing_enabled);
+        assert!(builder.async_drop_enabled);
 
         let namespace = builder.build().await.unwrap();
 
@@ -9372,6 +9426,7 @@ mod tests {
         // Both should default to true
         assert!(builder.manifest_enabled);
         assert!(builder.dir_listing_enabled);
+        assert!(!builder.async_drop_enabled);
     }
 
     #[tokio::test]
@@ -10110,7 +10165,9 @@ mod tests {
 
     #[tokio::test]
     async fn test_register_table() {
-        use lance_namespace::models::{RegisterTableRequest, TableExistsRequest};
+        use lance_namespace::models::{
+            DeregisterTableRequest, RegisterTableRequest, TableExistsRequest,
+        };
 
         let temp_dir = TempStdDir::default();
         let temp_path = temp_dir.to_str().unwrap();
@@ -10155,6 +10212,22 @@ mod tests {
         list_req.id = Some(vec![]);
         let tables = namespace.list_tables(list_req).await.unwrap();
         assert!(tables.tables.contains(&"registered_table".to_string()));
+
+        let mut deregister_req = DeregisterTableRequest::new();
+        deregister_req.id = Some(vec!["registered_table".to_string()]);
+        deregister_req.context = Some(HashMap::from([(
+            EXPECTED_DEREGISTER_LOCATION_CONTEXT_KEY.to_string(),
+            table_uri,
+        )]));
+        let error = namespace
+            .deregister_table(deregister_req)
+            .await
+            .unwrap_err();
+        assert!(
+            error
+                .to_string()
+                .contains("was not generated as a unique table incarnation")
+        );
     }
 
     #[tokio::test]
@@ -10257,6 +10330,167 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_deregister_table_expected_location_fence() {
+        use lance_namespace::models::{DeregisterTableRequest, DescribeTableRequest};
+
+        let temp_dir = TempStdDir::default();
+        let namespace = DirectoryNamespaceBuilder::new(temp_dir.to_str().unwrap())
+            .manifest_enabled(true)
+            .dir_listing_enabled(false)
+            .async_drop_enabled(true)
+            .build()
+            .await
+            .unwrap();
+
+        let schema = create_test_schema();
+        let ipc_data = create_test_ipc_data(&schema);
+        let mut create_req = CreateTableRequest::new();
+        create_req.id = Some(vec!["test_table".to_string()]);
+        namespace
+            .create_table(create_req, bytes::Bytes::from(ipc_data))
+            .await
+            .unwrap();
+
+        let table_id = vec!["test_table".to_string()];
+        let location = namespace
+            .describe_table(DescribeTableRequest {
+                id: Some(table_id.clone()),
+                ..Default::default()
+            })
+            .await
+            .unwrap()
+            .location
+            .unwrap();
+
+        let mut stale_req = DeregisterTableRequest::new();
+        stale_req.id = Some(table_id.clone());
+        stale_req.context = Some(HashMap::from([(
+            "expected_location".to_string(),
+            format!("{location}-replacement"),
+        )]));
+        let error = namespace.deregister_table(stale_req).await.unwrap_err();
+        assert!(error.to_string().contains("instead of expected location"));
+
+        let mut matching_req = DeregisterTableRequest::new();
+        matching_req.id = Some(table_id);
+        matching_req.context = Some(HashMap::from([(
+            "expected_location".to_string(),
+            format!("{location}/"),
+        )]));
+        let response = namespace.deregister_table(matching_req).await.unwrap();
+        assert_eq!(response.location.as_deref(), Some(location.as_str()));
+    }
+
+    #[tokio::test]
+    async fn test_deregister_table_rejects_deterministic_location_fence() {
+        use lance_namespace::models::DeregisterTableRequest;
+
+        let temp_dir = TempStdDir::default();
+        let namespace = DirectoryNamespaceBuilder::new(temp_dir.to_str().unwrap())
+            .manifest_enabled(true)
+            .dir_listing_enabled(true)
+            .build()
+            .await
+            .unwrap();
+
+        let schema = create_test_schema();
+        let ipc_data = create_test_ipc_data(&schema);
+        let mut create_req = CreateTableRequest::new();
+        create_req.id = Some(vec!["test_table".to_string()]);
+        let location = namespace
+            .create_table(create_req, bytes::Bytes::from(ipc_data))
+            .await
+            .unwrap()
+            .location
+            .unwrap();
+
+        let mut deregister_req = DeregisterTableRequest::new();
+        deregister_req.id = Some(vec!["test_table".to_string()]);
+        deregister_req.context = Some(HashMap::from([("expected_location".to_string(), location)]));
+        let error = namespace
+            .deregister_table(deregister_req)
+            .await
+            .unwrap_err();
+        assert!(
+            error
+                .to_string()
+                .contains("was not generated as a unique table incarnation")
+        );
+    }
+
+    #[tokio::test]
+    async fn test_deregister_table_rejects_deterministic_location_after_config_change() {
+        use lance_namespace::models::DeregisterTableRequest;
+
+        let temp_dir = TempStdDir::default();
+        let root = temp_dir.to_str().unwrap();
+        let creating_namespace = DirectoryNamespaceBuilder::new(root)
+            .manifest_enabled(true)
+            .dir_listing_enabled(true)
+            .build()
+            .await
+            .unwrap();
+
+        let schema = create_test_schema();
+        let ipc_data = create_test_ipc_data(&schema);
+        let mut create_req = CreateTableRequest::new();
+        create_req.id = Some(vec!["test_table".to_string()]);
+        let location = creating_namespace
+            .create_table(create_req, bytes::Bytes::from(ipc_data))
+            .await
+            .unwrap()
+            .location
+            .unwrap();
+
+        let fencing_namespace = DirectoryNamespaceBuilder::new(root)
+            .manifest_enabled(true)
+            .dir_listing_enabled(false)
+            .build()
+            .await
+            .unwrap();
+        let mut deregister_req = DeregisterTableRequest::new();
+        deregister_req.id = Some(vec!["test_table".to_string()]);
+        deregister_req.context = Some(HashMap::from([(
+            EXPECTED_DEREGISTER_LOCATION_CONTEXT_KEY.to_string(),
+            location,
+        )]));
+
+        let error = fencing_namespace
+            .deregister_table(deregister_req)
+            .await
+            .unwrap_err();
+        assert!(
+            error
+                .to_string()
+                .contains("was not generated as a unique table incarnation")
+        );
+    }
+
+    #[tokio::test]
+    async fn test_deregister_table_rejects_location_fence_without_manifest() {
+        use lance_namespace::models::DeregisterTableRequest;
+
+        let temp_dir = TempStdDir::default();
+        let namespace = DirectoryNamespaceBuilder::new(temp_dir.to_str().unwrap())
+            .manifest_enabled(false)
+            .build()
+            .await
+            .unwrap();
+        let mut deregister_req = DeregisterTableRequest::new();
+        deregister_req.id = Some(vec!["test_table".to_string()]);
+        deregister_req.context = Some(HashMap::from([(
+            EXPECTED_DEREGISTER_LOCATION_CONTEXT_KEY.to_string(),
+            "file:///stale/test_table.lance".to_string(),
+        )]));
+
+        let error = namespace
+            .deregister_table(deregister_req)
+            .await
+            .unwrap_err();
+        assert!(error.to_string().contains("manifest is disabled"));
+    }
+
     #[tokio::test]
     async fn test_deregister_table_in_child_namespace() {
         use lance_namespace::models::{