Implement drop namespace and namespace_exists

Author: osipovartem

catalogs/iceberg-s3tables-catalog/src/error.rs

@@ -3,6 +3,7 @@ use aws_sdk_s3tables::{
     error::SdkError,
     operation::{
         create_namespace::CreateNamespaceError, create_table::CreateTableError,
+        delete_namespace::DeleteNamespaceError, get_namespace::GetNamespaceError,
         delete_table::DeleteTableError, get_table::GetTableError,
         get_table_metadata_location::GetTableMetadataLocationError,
         list_namespaces::ListNamespacesError, list_tables::ListTablesError,
@@ -21,6 +22,10 @@ pub enum Error {
     #[error(transparent)]
     CreateNamespace(#[from] SdkError<CreateNamespaceError, HttpResponse>),
     #[error(transparent)]
+    DeleteNamespace(#[from] SdkError<DeleteNamespaceError, HttpResponse>),
+    #[error(transparent)]
+    GetNamespace(#[from] SdkError<GetNamespaceError, HttpResponse>),
+    #[error(transparent)]
     ListTables(#[from] SdkError<ListTablesError, HttpResponse>),
     #[error(transparent)]
     ListNamespaces(#[from] SdkError<ListNamespacesError, HttpResponse>),

catalogs/iceberg-s3tables-catalog/src/lib.rs

@@ -93,9 +93,18 @@ impl Catalog for S3TablesCatalog {
         Ok(HashMap::new())
     }
     /// Drop a namespace in the catalog
-    async fn drop_namespace(&self, _namespace: &Namespace) -> Result<(), IcebergError> {
-        todo!()
+    async fn drop_namespace(&self, namespace: &Namespace) -> Result<(), IcebergError> {
+        self.client
+            .delete_namespace()
+            .table_bucket_arn(&self.arn)
+            .namespace(namespace[0].as_str())
+            .send()
+            .await
+            .map_err(Error::from)?;
+
+        Ok(())
     }
+
     /// Load the namespace properties from the catalog
     async fn load_namespace(
         &self,
@@ -113,8 +122,15 @@ impl Catalog for S3TablesCatalog {
         todo!()
     }
     /// Check if a namespace exists
-    async fn namespace_exists(&self, _namespace: &Namespace) -> Result<bool, IcebergError> {
-        todo!()
+    async fn namespace_exists(&self, namespace: &Namespace) -> Result<bool, IcebergError> {
+        Ok(self.client
+            .get_namespace()
+            .table_bucket_arn(&self.arn)
+            .namespace(namespace[0].as_str())
+            .send()
+            .await
+            .map_err(Error::from)
+            .is_ok())
     }
     async fn list_tabulars(&self, namespace: &Namespace) -> Result<Vec<Identifier>, IcebergError> {
         let mut tabulars = Vec::new();