feat!(catalog): adding support for purge_table (#2232)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #2133 

## What changes are included in this PR?
- Add catalog/utils.rs to provide helpers to delete table data using
file_io and table_metadata
- Add new API `purge_table` to `Catalog` trait and add default
implementation
- Implement purge_table for S3TableCatalog and RestCatalog

<!--
Provide a summary of the modifications in this PR. List the main changes
such as new features, bug fixes, refactoring, or any other updates.
-->

## Are these changes tested?
Added new tests in table_suite
<!--
Specify what test covers (unit test, integration test, etc.).

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

Author: CTTY

crates/catalog/glue/src/catalog.rs

@@ -659,6 +659,17 @@ impl Catalog for GlueCatalog {
         Ok(())
     }
 
+    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
+        let table_info = self.load_table(table).await?;
+        self.drop_table(table).await?;
+        iceberg::drop_table_data(
+            table_info.file_io(),
+            table_info.metadata(),
+            table_info.metadata_location(),
+        )
+        .await
+    }
+
     /// Asynchronously checks the existence of a specified table
     /// in the database.
     ///

crates/catalog/hms/src/catalog.rs

@@ -604,6 +604,17 @@ impl Catalog for HmsCatalog {
         Ok(())
     }
 
+    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
+        let table_info = self.load_table(table).await?;
+        self.drop_table(table).await?;
+        iceberg::drop_table_data(
+            table_info.file_io(),
+            table_info.metadata(),
+            table_info.metadata_location(),
+        )
+        .await
+    }
+
     /// Asynchronously checks the existence of a specified table
     /// in the database.
     ///

crates/catalog/loader/tests/common/mod.rs

@@ -335,7 +335,7 @@ pub fn assert_map_contains(expected: &HashMap<String, String>, actual: &HashMap<
 pub async fn cleanup_namespace_dyn(catalog: &dyn Catalog, namespace: &NamespaceIdent) {
     if let Ok(tables) = catalog.list_tables(namespace).await {
         for table in tables {
-            let _ = catalog.drop_table(&table).await;
+            let _ = catalog.purge_table(&table).await;
         }
     }
     let _ = catalog.drop_namespace(namespace).await;

crates/catalog/loader/tests/table_suite.rs

@@ -274,3 +274,80 @@ async fn test_catalog_drop_table_missing_errors(#[case] kind: CatalogKind) -> Re
     assert!(catalog.drop_table(&table_ident).await.is_err());
     Ok(())
 }
+
+// Common behavior: purge_table removes the table from the catalog.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_purge_table(#[case] kind: CatalogKind) -> Result<()> {
+    let Some(harness) = load_catalog(kind).await else {
+        return Ok(());
+    };
+    let catalog = harness.catalog;
+    let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+        "catalog_purge_table",
+        harness.label
+    ));
+
+    cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+    catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+    let table_name = normalize_test_name_with_parts!("catalog_purge_table", harness.label, "table");
+    let table = catalog
+        .create_table(&namespace, table_creation(table_name))
+        .await?;
+    let ident = table.identifier().clone();
+
+    assert!(catalog.table_exists(&ident).await?);
+
+    // Capture metadata location and file_io before purge so we can verify
+    // that the underlying files are actually deleted.
+    let metadata_location = table.metadata_location().map(|s| s.to_string());
+    let file_io = table.file_io().clone();
+
+    catalog.purge_table(&ident).await?;
+    assert!(!catalog.table_exists(&ident).await?);
+
+    if let Some(location) = &metadata_location {
+        assert!(
+            !file_io.exists(location).await?,
+            "Metadata file should have been deleted after purge"
+        );
+    }
+
+    catalog.drop_namespace(&namespace).await?;
+
+    Ok(())
+}
+
+// Common behavior: purging a missing table should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_purge_table_missing_errors(#[case] kind: CatalogKind) -> Result<()> {
+    let Some(harness) = load_catalog(kind).await else {
+        return Ok(());
+    };
+    let catalog = harness.catalog;
+    let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+        "catalog_purge_table_missing_errors",
+        harness.label
+    ));
+
+    cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+    catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+    let table_ident = TableIdent::new(namespace.clone(), "missing".to_string());
+    assert!(catalog.purge_table(&table_ident).await.is_err());
+    Ok(())
+}

crates/catalog/rest/src/catalog.rs

@@ -363,6 +363,35 @@ impl RestCatalog {
         }
     }
 
+    /// Sends a DELETE request for the given table, optionally requesting purge.
+    async fn delete_table(&self, table: &TableIdent, purge: bool) -> Result<()> {
+        let context = self.context().await?;
+
+        let mut request_builder = context
+            .client
+            .request(Method::DELETE, context.config.table_endpoint(table));
+
+        if purge {
+            request_builder = request_builder.query(&[("purgeRequested", "true")]);
+        }
+
+        let request = request_builder.build()?;
+        let http_response = context.client.query_catalog(request).await?;
+
+        match http_response.status() {
+            StatusCode::NO_CONTENT | StatusCode::OK => Ok(()),
+            StatusCode::NOT_FOUND => Err(Error::new(
+                ErrorKind::TableNotFound,
+                "Tried to drop a table that does not exist",
+            )),
+            _ => Err(deserialize_unexpected_catalog_error(
+                http_response,
+                context.client.disable_header_redaction(),
+            )
+            .await),
+        }
+    }
+
     /// Gets the [`RestContext`] from the catalog.
     async fn context(&self) -> Result<&RestContext> {
         self.ctx
@@ -828,27 +857,13 @@ impl Catalog for RestCatalog {
 
     /// Drop a table from the catalog.
     async fn drop_table(&self, table: &TableIdent) -> Result<()> {
-        let context = self.context().await?;
-
-        let request = context
-            .client
-            .request(Method::DELETE, context.config.table_endpoint(table))
-            .build()?;
-
-        let http_response = context.client.query_catalog(request).await?;
+        self.delete_table(table, false).await
+    }
 
-        match http_response.status() {
-            StatusCode::NO_CONTENT | StatusCode::OK => Ok(()),
-            StatusCode::NOT_FOUND => Err(Error::new(
-                ErrorKind::TableNotFound,
-                "Tried to drop a table that does not exist",
-            )),
-            _ => Err(deserialize_unexpected_catalog_error(
-                http_response,
-                context.client.disable_header_redaction(),
-            )
-            .await),
-        }
+    /// Drop a table from the catalog and purge its data by sending
+    /// `purgeRequested=true` to the REST server.
+    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
+        self.delete_table(table, true).await
     }
 
     /// Check if a table exists in the catalog.

crates/catalog/s3tables/src/catalog.rs

@@ -562,15 +562,18 @@ impl Catalog for S3TablesCatalog {
         Ok(self.load_table_with_version_token(table_ident).await?.0)
     }
 
-    /// Drops an existing table from the s3tables catalog.
+    /// Not supported for S3Tables. Use `purge_table` instead.
     ///
-    /// Validates the table identifier and then deletes the corresponding
-    /// table from the s3tables catalog.
-    ///
-    /// This function can return an error in the following situations:
-    /// - Errors from the underlying database deletion process, converted using
-    /// `from_aws_sdk_error`.
-    async fn drop_table(&self, table: &TableIdent) -> Result<()> {
+    /// S3 Tables doesn't support soft delete, so dropping a table will permanently remove it from the catalog.
+    async fn drop_table(&self, _table: &TableIdent) -> Result<()> {
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "drop_table is not supported for S3Tables; use purge_table instead",
+        ))
+    }
+
+    /// Purge a table from the S3 Tables catalog.
+    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
         let req = self
             .s3tables_client
             .delete_table()

crates/catalog/sql/src/catalog.rs

@@ -757,6 +757,17 @@ impl Catalog for SqlCatalog {
         Ok(())
     }
 
+    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
+        let table_info = self.load_table(table).await?;
+        self.drop_table(table).await?;
+        iceberg::drop_table_data(
+            table_info.file_io(),
+            table_info.metadata(),
+            table_info.metadata_location(),
+        )
+        .await
+    }
+
     async fn load_table(&self, identifier: &TableIdent) -> Result<Table> {
         if !self.table_exists(identifier).await? {
             return no_such_table_err(identifier);

crates/iceberg/src/catalog/memory/catalog.rs

@@ -326,6 +326,17 @@ impl Catalog for MemoryCatalog {
         Ok(())
     }
 
+    async fn purge_table(&self, table_ident: &TableIdent) -> Result<()> {
+        let table_info = self.load_table(table_ident).await?;
+        self.drop_table(table_ident).await?;
+        crate::catalog::utils::drop_table_data(
+            table_info.file_io(),
+            table_info.metadata(),
+            table_info.metadata_location(),
+        )
+        .await
+    }
+
     /// Check if a table exists in the catalog.
     async fn table_exists(&self, table_ident: &TableIdent) -> Result<bool> {
         let root_namespace_state = self.root_namespace_state.lock().await;

crates/iceberg/src/catalog/mod.rs

@@ -19,6 +19,7 @@
 
 pub mod memory;
 mod metadata_location;
+pub(crate) mod utils;
 
 use std::collections::HashMap;
 use std::fmt::{Debug, Display};
@@ -98,6 +99,14 @@ pub trait Catalog: Debug + Sync + Send {
     /// Drop a table from the catalog, or returns error if it doesn't exist.
     async fn drop_table(&self, table: &TableIdent) -> Result<()>;
 
+    /// Drop a table from the catalog and delete the underlying table data.
+    ///
+    /// Implementations should load the table metadata, drop the table
+    /// from the catalog, then delete all associated data and metadata files.
+    /// The [`drop_table_data`](utils::drop_table_data) utility function can
+    /// be used for the file cleanup step.
+    async fn purge_table(&self, table: &TableIdent) -> Result<()>;
+
     /// Check if a table exists in the catalog.
     async fn table_exists(&self, table: &TableIdent) -> Result<bool>;