iceberg_catalog_glue/
catalog.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashMap;
19use std::fmt::Debug;
20use std::str::FromStr;
21use std::sync::Arc;
22
23use anyhow::anyhow;
24use async_trait::async_trait;
25use aws_sdk_glue::operation::create_table::CreateTableError;
26use aws_sdk_glue::operation::update_table::UpdateTableError;
27use aws_sdk_glue::types::TableInput;
28use iceberg::encryption::kms::{KeyManagementClient, KmsClientFactory};
29use iceberg::io::{
30    FileIO, FileIOBuilder, S3_ACCESS_KEY_ID, S3_ENDPOINT, S3_REGION, S3_SECRET_ACCESS_KEY,
31    S3_SESSION_TOKEN, StorageFactory,
32};
33use iceberg::spec::{TableMetadata, TableMetadataBuilder};
34use iceberg::table::Table;
35use iceberg::{
36    Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
37    Runtime, TableCommit, TableCreation, TableIdent,
38};
39use iceberg_property_macro::Properties;
40use iceberg_storage_opendal::OpenDalStorageFactory;
41
42use crate::error::{from_aws_build_error, from_aws_sdk_error};
43use crate::utils::{
44    convert_to_database, convert_to_glue_table, convert_to_namespace, create_sdk_config,
45    get_default_table_location, get_metadata_location, is_iceberg_table, validate_namespace,
46};
47use crate::{
48    AWS_ACCESS_KEY_ID, AWS_REGION_NAME, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN, with_catalog_id,
49};
50
51/// Glue catalog URI
52pub const GLUE_CATALOG_PROP_URI: &str = "uri";
53/// Glue catalog id
54pub const GLUE_CATALOG_PROP_CATALOG_ID: &str = "catalog_id";
55/// Glue catalog warehouse location
56pub const GLUE_CATALOG_PROP_WAREHOUSE: &str = "warehouse";
57
58/// Builder for [`GlueCatalog`].
59#[derive(Debug, Default)]
60pub struct GlueCatalogBuilder {
61    storage_factory: Option<Arc<dyn StorageFactory>>,
62    kms_client_factory: Option<Arc<dyn KmsClientFactory>>,
63    runtime: Option<Runtime>,
64}
65
66impl CatalogBuilder for GlueCatalogBuilder {
67    type C = GlueCatalog;
68
69    fn with_storage_factory(mut self, storage_factory: Arc<dyn StorageFactory>) -> Self {
70        self.storage_factory = Some(storage_factory);
71        self
72    }
73
74    fn with_kms_client_factory(mut self, kms_client_factory: Arc<dyn KmsClientFactory>) -> Self {
75        self.kms_client_factory = Some(kms_client_factory);
76        self
77    }
78
79    fn with_runtime(mut self, runtime: Runtime) -> Self {
80        self.runtime = Some(runtime);
81        self
82    }
83
84    fn load(
85        self,
86        name: impl Into<String>,
87        props: HashMap<String, String>,
88    ) -> impl Future<Output = Result<Self::C>> + Send {
89        let name = name.into();
90
91        async move {
92            let catalog_properties = GlueCatalogProperties::from_properties(&props)?;
93            if catalog_properties.warehouse.is_empty() {
94                return Err(Error::new(
95                    ErrorKind::DataInvalid,
96                    "Catalog warehouse is required",
97                ));
98            }
99
100            let runtime = match self.runtime {
101                Some(rt) => rt,
102                None => Runtime::try_current()?,
103            };
104            let kms_client = match self.kms_client_factory {
105                Some(factory) => Some(factory.create_kms_client(&props).await?),
106                None => None,
107            };
108            GlueCatalog::new(
109                name,
110                catalog_properties,
111                props,
112                self.storage_factory,
113                runtime,
114                kms_client,
115            )
116            .await
117        }
118    }
119}
120
121/// Glue catalog properties parsed from the `load` property map.
122#[derive(Debug, Properties)]
123pub(crate) struct GlueCatalogProperties {
124    #[property(key = GLUE_CATALOG_PROP_URI, default = None)]
125    uri: Option<String>,
126    #[property(key = GLUE_CATALOG_PROP_CATALOG_ID, default = None)]
127    catalog_id: Option<String>,
128    #[property(key = GLUE_CATALOG_PROP_WAREHOUSE, default = "")]
129    warehouse: String,
130}
131
132struct GlueClient(aws_sdk_glue::Client);
133
134/// Glue Catalog
135pub struct GlueCatalog {
136    name: String,
137    properties: GlueCatalogProperties,
138    client: GlueClient,
139    file_io: FileIO,
140    runtime: Runtime,
141    kms_client: Option<Arc<dyn KeyManagementClient>>,
142}
143
144impl Debug for GlueCatalog {
145    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
146        f.debug_struct("GlueCatalog")
147            .field("name", &self.name)
148            .field("properties", &self.properties)
149            .finish_non_exhaustive()
150    }
151}
152
153impl GlueCatalog {
154    /// Create a new glue catalog
155    async fn new(
156        name: String,
157        properties: GlueCatalogProperties,
158        props: HashMap<String, String>,
159        storage_factory: Option<Arc<dyn StorageFactory>>,
160        runtime: Runtime,
161        kms_client: Option<Arc<dyn KeyManagementClient>>,
162    ) -> Result<Self> {
163        let sdk_config = create_sdk_config(&props, properties.uri.as_ref()).await;
164        let mut file_io_props = props;
165        if !file_io_props.contains_key(S3_ACCESS_KEY_ID)
166            && let Some(access_key_id) = file_io_props.get(AWS_ACCESS_KEY_ID)
167        {
168            file_io_props.insert(S3_ACCESS_KEY_ID.to_string(), access_key_id.to_string());
169        }
170        if !file_io_props.contains_key(S3_SECRET_ACCESS_KEY)
171            && let Some(secret_access_key) = file_io_props.get(AWS_SECRET_ACCESS_KEY)
172        {
173            file_io_props.insert(
174                S3_SECRET_ACCESS_KEY.to_string(),
175                secret_access_key.to_string(),
176            );
177        }
178        if !file_io_props.contains_key(S3_REGION)
179            && let Some(region) = file_io_props.get(AWS_REGION_NAME)
180        {
181            file_io_props.insert(S3_REGION.to_string(), region.to_string());
182        }
183        if !file_io_props.contains_key(S3_SESSION_TOKEN)
184            && let Some(session_token) = file_io_props.get(AWS_SESSION_TOKEN)
185        {
186            file_io_props.insert(S3_SESSION_TOKEN.to_string(), session_token.to_string());
187        }
188        if !file_io_props.contains_key(S3_ENDPOINT)
189            && let Some(aws_endpoint) = properties.uri.as_ref()
190        {
191            file_io_props.insert(S3_ENDPOINT.to_string(), aws_endpoint.to_string());
192        }
193
194        let client = aws_sdk_glue::Client::new(&sdk_config);
195
196        // Use provided factory or default to OpenDalStorageFactory::S3
197        let factory = storage_factory.unwrap_or_else(|| {
198            Arc::new(OpenDalStorageFactory::S3 {
199                customized_credential_load: None,
200            })
201        });
202        let file_io = FileIOBuilder::new(factory)
203            .with_props(file_io_props)
204            .build();
205
206        Ok(GlueCatalog {
207            name,
208            properties,
209            client: GlueClient(client),
210            file_io,
211            runtime,
212            kms_client,
213        })
214    }
215    /// Get the catalogs `FileIO`
216    pub fn file_io(&self) -> FileIO {
217        self.file_io.clone()
218    }
219
220    /// Loads a table from the Glue Catalog along with its version_id for optimistic locking.
221    ///
222    /// # Returns
223    /// A `Result` wrapping a tuple of (`Table`, `Option<String>`) where the String is the version_id
224    /// from Glue that should be used for optimistic concurrency control when updating the table.
225    ///
226    /// # Errors
227    /// This function may return an error in several scenarios, including:
228    /// - Failure to validate the namespace.
229    /// - Failure to retrieve the table from the Glue Catalog.
230    /// - Absence of metadata location information in the table's properties.
231    /// - Issues reading or deserializing the table's metadata file.
232    async fn load_table_with_version_id(
233        &self,
234        table: &TableIdent,
235    ) -> Result<(Table, Option<String>)> {
236        let db_name = validate_namespace(table.namespace())?;
237        let table_name = table.name();
238
239        let builder = self
240            .client
241            .0
242            .get_table()
243            .database_name(&db_name)
244            .name(table_name);
245        let builder = with_catalog_id!(builder, self.properties);
246
247        let glue_table_output = builder.send().await.map_err(from_aws_sdk_error)?;
248
249        let glue_table = glue_table_output.table().ok_or_else(|| {
250            Error::new(
251                ErrorKind::TableNotFound,
252                format!(
253                    "Table object for database: {db_name} and table: {table_name} does not exist"
254                ),
255            )
256        })?;
257
258        let version_id = glue_table.version_id.clone();
259        let metadata_location = get_metadata_location(&glue_table.parameters)?;
260
261        let metadata = TableMetadata::read_from(&self.file_io, &metadata_location).await?;
262
263        let mut builder = Table::builder()
264            .file_io(self.file_io())
265            .metadata_location(metadata_location)
266            .metadata(metadata)
267            .identifier(TableIdent::new(
268                NamespaceIdent::new(db_name),
269                table_name.to_owned(),
270            ))
271            .runtime(self.runtime.clone());
272        if let Some(kms_client) = self.kms_client.clone() {
273            builder = builder.kms_client(kms_client);
274        }
275        let table = builder.build()?;
276
277        Ok((table, version_id))
278    }
279}
280
281#[async_trait]
282impl Catalog for GlueCatalog {
283    /// List namespaces from glue catalog.
284    ///
285    /// Glue doesn't support nested namespaces.
286    /// We will return an empty list if parent is some.
287    async fn list_namespaces(
288        &self,
289        parent: Option<&NamespaceIdent>,
290    ) -> Result<Vec<NamespaceIdent>> {
291        if parent.is_some() {
292            return Ok(vec![]);
293        }
294
295        let mut database_list: Vec<NamespaceIdent> = Vec::new();
296        let mut next_token: Option<String> = None;
297
298        loop {
299            let builder = match &next_token {
300                Some(token) => self.client.0.get_databases().next_token(token),
301                None => self.client.0.get_databases(),
302            };
303            let builder = with_catalog_id!(builder, self.properties);
304            let resp = builder.send().await.map_err(from_aws_sdk_error)?;
305
306            let dbs: Vec<NamespaceIdent> = resp
307                .database_list()
308                .iter()
309                .map(|db| NamespaceIdent::new(db.name().to_string()))
310                .collect();
311
312            database_list.extend(dbs);
313
314            next_token = resp.next_token().map(ToOwned::to_owned);
315            if next_token.is_none() {
316                break;
317            }
318        }
319
320        Ok(database_list)
321    }
322
323    /// Creates a new namespace with the given identifier and properties.
324    ///
325    /// Attempts to create a namespace defined by the `namespace`
326    /// parameter and configured with the specified `properties`.
327    ///
328    /// This function can return an error in the following situations:
329    ///
330    /// - Errors from `validate_namespace` if the namespace identifier does not
331    /// meet validation criteria.
332    /// - Errors from `convert_to_database` if the properties cannot be
333    /// successfully converted into a database configuration.
334    /// - Errors from the underlying database creation process, converted using
335    /// `from_sdk_error`.
336    async fn create_namespace(
337        &self,
338        namespace: &NamespaceIdent,
339        properties: HashMap<String, String>,
340    ) -> Result<Namespace> {
341        if self.namespace_exists(namespace).await? {
342            return Err(Error::new(
343                ErrorKind::NamespaceAlreadyExists,
344                format!("Namespace {namespace:?} already exists"),
345            ));
346        }
347
348        let db_input = convert_to_database(namespace, &properties)?;
349
350        let builder = self.client.0.create_database().database_input(db_input);
351        let builder = with_catalog_id!(builder, self.properties);
352
353        builder.send().await.map_err(from_aws_sdk_error)?;
354
355        Ok(Namespace::with_properties(namespace.clone(), properties))
356    }
357
358    /// Retrieves a namespace by its identifier.
359    ///
360    /// Validates the given namespace identifier and then queries the
361    /// underlying database client to fetch the corresponding namespace data.
362    /// Constructs a `Namespace` object with the retrieved data and returns it.
363    ///
364    /// This function can return an error in any of the following situations:
365    /// - If the provided namespace identifier fails validation checks
366    /// - If there is an error querying the database, returned by
367    /// `from_sdk_error`.
368    async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
369        let db_name = validate_namespace(namespace)?;
370
371        let builder = self.client.0.get_database().name(&db_name);
372        let builder = with_catalog_id!(builder, self.properties);
373
374        let resp = builder.send().await.map_err(|err| {
375            if err
376                .as_service_error()
377                .map(|e| e.is_entity_not_found_exception())
378                == Some(true)
379            {
380                return Error::new(
381                    ErrorKind::NamespaceNotFound,
382                    format!("Namespace {namespace:?} does not exist"),
383                );
384            }
385            from_aws_sdk_error(err)
386        })?;
387
388        match resp.database() {
389            Some(db) => {
390                let namespace = convert_to_namespace(db);
391                Ok(namespace)
392            }
393            None => Err(Error::new(
394                ErrorKind::NamespaceNotFound,
395                format!("Database with name: {db_name} does not exist"),
396            )),
397        }
398    }
399
400    /// Checks if a namespace exists within the Glue Catalog.
401    ///
402    /// Validates the namespace identifier by querying the Glue Catalog
403    /// to determine if the specified namespace (database) exists.
404    ///
405    /// # Returns
406    /// A `Result<bool>` indicating the outcome of the check:
407    /// - `Ok(true)` if the namespace exists.
408    /// - `Ok(false)` if the namespace does not exist, identified by a specific
409    /// `EntityNotFoundException` variant.
410    /// - `Err(...)` if an error occurs during validation or the Glue Catalog
411    /// query, with the error encapsulating the issue.
412    async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result<bool> {
413        let db_name = validate_namespace(namespace)?;
414
415        let builder = self.client.0.get_database().name(&db_name);
416        let builder = with_catalog_id!(builder, self.properties);
417
418        let resp = builder.send().await;
419
420        match resp {
421            Ok(_) => Ok(true),
422            Err(err) => {
423                if err
424                    .as_service_error()
425                    .map(|e| e.is_entity_not_found_exception())
426                    == Some(true)
427                {
428                    return Ok(false);
429                }
430                Err(from_aws_sdk_error(err))
431            }
432        }
433    }
434
435    /// Asynchronously updates properties of an existing namespace.
436    ///
437    /// Converts the given namespace identifier and properties into a database
438    /// representation and then attempts to update the corresponding namespace
439    /// in the Glue Catalog.
440    ///
441    /// # Returns
442    /// Returns `Ok(())` if the namespace update is successful. If the
443    /// namespace cannot be updated due to missing information or an error
444    /// during the update process, an `Err(...)` is returned.
445    async fn update_namespace(
446        &self,
447        namespace: &NamespaceIdent,
448        properties: HashMap<String, String>,
449    ) -> Result<()> {
450        if !self.namespace_exists(namespace).await? {
451            return Err(Error::new(
452                ErrorKind::NamespaceNotFound,
453                format!("Namespace {namespace:?} does not exist"),
454            ));
455        }
456
457        let db_name = validate_namespace(namespace)?;
458        let db_input = convert_to_database(namespace, &properties)?;
459
460        let builder = self
461            .client
462            .0
463            .update_database()
464            .name(&db_name)
465            .database_input(db_input);
466        let builder = with_catalog_id!(builder, self.properties);
467
468        builder.send().await.map_err(from_aws_sdk_error)?;
469
470        Ok(())
471    }
472
473    /// Asynchronously drops a namespace from the Glue Catalog.
474    ///
475    /// Checks if the namespace is empty. If it still contains tables the
476    /// namespace will not be dropped, but an error is returned instead.
477    ///
478    /// # Returns
479    /// A `Result<()>` indicating the outcome:
480    /// - `Ok(())` signifies successful namespace deletion.
481    /// - `Err(...)` signifies failure to drop the namespace due to validation
482    /// errors, connectivity issues, or Glue Catalog constraints.
483    async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
484        if !self.namespace_exists(namespace).await? {
485            return Err(Error::new(
486                ErrorKind::NamespaceNotFound,
487                format!("Namespace {namespace:?} does not exist"),
488            ));
489        }
490
491        let db_name = validate_namespace(namespace)?;
492
493        // Check for ANY Glue table in the database, not just Iceberg tables.
494        // Glue's `delete_database` will fail if any table (Iceberg or not) is
495        // still present, and `list_tables` only returns Iceberg tables, so we
496        // query Glue directly here.
497        let builder = self
498            .client
499            .0
500            .get_tables()
501            .database_name(&db_name)
502            .max_results(1);
503        let builder = with_catalog_id!(builder, self.properties);
504        let resp = builder.send().await.map_err(from_aws_sdk_error)?;
505
506        if !resp.table_list().is_empty() {
507            return Err(Error::new(
508                ErrorKind::DataInvalid,
509                format!("Database with name: {} is not empty", db_name),
510            ));
511        }
512
513        let builder = self.client.0.delete_database().name(db_name);
514        let builder = with_catalog_id!(builder, self.properties);
515
516        builder.send().await.map_err(from_aws_sdk_error)?;
517
518        Ok(())
519    }
520
521    /// Asynchronously lists all Iceberg tables within a specified namespace.
522    ///
523    /// Glue databases may contain a mix of Iceberg and non-Iceberg tables
524    /// (e.g. plain Hive tables). Only tables whose `table_type` parameter is
525    /// set to `ICEBERG` (case-insensitive) are returned
526    ///
527    /// # Returns
528    /// A `Result<Vec<TableIdent>>`, which is:
529    /// - `Ok(vec![...])` containing a vector of `TableIdent` instances, each
530    /// representing an Iceberg table within the specified namespace.
531    /// - `Err(...)` if an error occurs during namespace validation or while
532    /// querying the database.
533    async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
534        let db_name = validate_namespace(namespace)?;
535
536        let mut table_list: Vec<TableIdent> = Vec::new();
537        let mut next_token: Option<String> = None;
538
539        loop {
540            let builder = match &next_token {
541                Some(token) => self
542                    .client
543                    .0
544                    .get_tables()
545                    .database_name(&db_name)
546                    .next_token(token),
547                None => self.client.0.get_tables().database_name(&db_name),
548            };
549            let builder = with_catalog_id!(builder, self.properties);
550            let resp = builder.send().await.map_err(from_aws_sdk_error)?;
551
552            let tables: Vec<_> = resp
553                .table_list()
554                .iter()
555                .filter(|tbl| is_iceberg_table(&tbl.parameters))
556                .map(|tbl| TableIdent::new(namespace.clone(), tbl.name().to_string()))
557                .collect();
558
559            table_list.extend(tables);
560
561            next_token = resp.next_token().map(ToOwned::to_owned);
562            if next_token.is_none() {
563                break;
564            }
565        }
566
567        Ok(table_list)
568    }
569
570    /// Creates a new table within a specified namespace using the provided
571    /// table creation settings.
572    ///
573    /// # Returns
574    /// A `Result` wrapping a `Table` object representing the newly created
575    /// table.
576    ///
577    /// # Errors
578    /// This function may return an error in several cases, including invalid
579    /// namespace identifiers, failure to determine a default storage location,
580    /// issues generating or writing table metadata, and errors communicating
581    /// with the Glue Catalog.
582    async fn create_table(
583        &self,
584        namespace: &NamespaceIdent,
585        mut creation: TableCreation,
586    ) -> Result<Table> {
587        let db_name = validate_namespace(namespace)?;
588        let table_name = creation.name.clone();
589
590        if creation.location.is_none() {
591            let ns = self.get_namespace(namespace).await?;
592            let location =
593                get_default_table_location(&ns, &db_name, &table_name, &self.properties.warehouse);
594            creation.location = Some(location);
595        }
596        let metadata = TableMetadataBuilder::from_table_creation(creation)?
597            .build()?
598            .metadata;
599        let metadata_location = MetadataLocation::try_new_with_metadata(&metadata)?;
600
601        metadata.write_to(&self.file_io, &metadata_location).await?;
602
603        let metadata_location_str = metadata_location.to_string();
604        let glue_table = convert_to_glue_table(
605            &table_name,
606            metadata_location_str.clone(),
607            &metadata,
608            metadata.properties(),
609            None,
610        )?;
611
612        let builder = self
613            .client
614            .0
615            .create_table()
616            .database_name(&db_name)
617            .table_input(glue_table);
618        let builder = with_catalog_id!(builder, self.properties);
619
620        builder.send().await.map_err(from_aws_sdk_error)?;
621
622        let mut builder = Table::builder()
623            .file_io(self.file_io())
624            .metadata_location(metadata_location_str)
625            .metadata(metadata)
626            .identifier(TableIdent::new(NamespaceIdent::new(db_name), table_name))
627            .runtime(self.runtime.clone());
628        if let Some(kms_client) = self.kms_client.clone() {
629            builder = builder.kms_client(kms_client);
630        }
631        builder.build()
632    }
633
634    /// Loads a table from the Glue Catalog and constructs a `Table` object
635    /// based on its metadata.
636    ///
637    /// # Returns
638    /// A `Result` wrapping a `Table` object that represents the loaded table.
639    ///
640    /// # Errors
641    /// This function may return an error in several scenarios, including:
642    /// - Failure to validate the namespace.
643    /// - Failure to retrieve the table from the Glue Catalog.
644    /// - Absence of metadata location information in the table's properties.
645    /// - Issues reading or deserializing the table's metadata file.
646    async fn load_table(&self, table: &TableIdent) -> Result<Table> {
647        let (table, _) = self.load_table_with_version_id(table).await?;
648        Ok(table)
649    }
650
651    /// Asynchronously drops a table from the database.
652    ///
653    /// # Errors
654    /// Returns an error if:
655    /// - The namespace provided in `table` cannot be validated
656    /// or does not exist.
657    /// - The underlying database client encounters an error while
658    /// attempting to drop the table. This includes scenarios where
659    /// the table does not exist.
660    /// - Any network or communication error occurs with the database backend.
661    async fn drop_table(&self, table: &TableIdent) -> Result<()> {
662        let db_name = validate_namespace(table.namespace())?;
663        let table_name = table.name();
664
665        let builder = self
666            .client
667            .0
668            .delete_table()
669            .database_name(&db_name)
670            .name(table_name);
671        let builder = with_catalog_id!(builder, self.properties);
672
673        builder.send().await.map_err(from_aws_sdk_error)?;
674
675        Ok(())
676    }
677
678    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
679        let table_info = self.load_table(table).await?;
680        self.drop_table(table).await?;
681        iceberg::drop_table_data(&table_info).await
682    }
683
684    /// Asynchronously checks the existence of a specified table
685    /// in the database.
686    ///
687    /// # Returns
688    /// - `Ok(true)` if the table exists in the database.
689    /// - `Ok(false)` if the table does not exist in the database.
690    /// - `Err(...)` if an error occurs during the process
691    async fn table_exists(&self, table: &TableIdent) -> Result<bool> {
692        let db_name = validate_namespace(table.namespace())?;
693        let table_name = table.name();
694
695        let builder = self
696            .client
697            .0
698            .get_table()
699            .database_name(&db_name)
700            .name(table_name);
701        let builder = with_catalog_id!(builder, self.properties);
702
703        let resp = builder.send().await;
704
705        match resp {
706            Ok(_) => Ok(true),
707            Err(err) => {
708                if err
709                    .as_service_error()
710                    .map(|e| e.is_entity_not_found_exception())
711                    == Some(true)
712                {
713                    return Ok(false);
714                }
715                Err(from_aws_sdk_error(err))
716            }
717        }
718    }
719
720    /// Asynchronously renames a table within the database
721    /// or moves it between namespaces (databases).
722    ///
723    /// # Returns
724    /// - `Ok(())` on successful rename or move of the table.
725    /// - `Err(...)` if an error occurs during the process.
726    async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> {
727        let src_db_name = validate_namespace(src.namespace())?;
728        let dest_db_name = validate_namespace(dest.namespace())?;
729
730        let src_table_name = src.name();
731        let dest_table_name = dest.name();
732
733        let builder = self
734            .client
735            .0
736            .get_table()
737            .database_name(&src_db_name)
738            .name(src_table_name);
739        let builder = with_catalog_id!(builder, self.properties);
740
741        let glue_table_output = builder.send().await.map_err(from_aws_sdk_error)?;
742
743        match glue_table_output.table() {
744            None => Err(Error::new(
745                ErrorKind::TableNotFound,
746                format!(
747                    "'Table' object for database: {src_db_name} and table: {src_table_name} does not exist"
748                ),
749            )),
750            Some(table) => {
751                let rename_table_input = TableInput::builder()
752                    .name(dest_table_name)
753                    .set_parameters(table.parameters.clone())
754                    .set_storage_descriptor(table.storage_descriptor.clone())
755                    .set_table_type(table.table_type.clone())
756                    .set_description(table.description.clone())
757                    .build()
758                    .map_err(from_aws_build_error)?;
759
760                let builder = self
761                    .client
762                    .0
763                    .create_table()
764                    .database_name(&dest_db_name)
765                    .table_input(rename_table_input);
766                let builder = with_catalog_id!(builder, self.properties);
767
768                builder.send().await.map_err(from_aws_sdk_error)?;
769
770                let drop_src_table_result = self.drop_table(src).await;
771
772                match drop_src_table_result {
773                    Ok(_) => Ok(()),
774                    Err(_) => {
775                        let err_msg_src_table =
776                            format!("Failed to drop old table {src_db_name}.{src_table_name}.");
777
778                        let drop_dest_table_result = self.drop_table(dest).await;
779
780                        match drop_dest_table_result {
781                            Ok(_) => Err(Error::new(
782                                ErrorKind::Unexpected,
783                                format!(
784                                    "{err_msg_src_table} Rolled back table creation for {dest_db_name}.{dest_table_name}."
785                                ),
786                            )),
787                            Err(_) => Err(Error::new(
788                                ErrorKind::Unexpected,
789                                format!(
790                                    "{err_msg_src_table} Failed to roll back table creation for {dest_db_name}.{dest_table_name}. Please clean up manually."
791                                ),
792                            )),
793                        }
794                    }
795                }
796            }
797        }
798    }
799
800    /// registers an existing table into the Glue Catalog.
801    ///
802    /// Converts the provided table identifier and metadata location into a
803    /// Glue-compatible table representation, and attempts to create the
804    /// corresponding table in the Glue Catalog.
805    ///
806    /// # Returns
807    /// Returns `Ok(Table)` if the table is successfully registered and loaded.
808    /// If the registration fails due to validation issues, existing table conflicts,
809    /// metadata problems, or errors during the registration or loading process,
810    /// an `Err(...)` is returned.
811    async fn register_table(
812        &self,
813        table_ident: &TableIdent,
814        metadata_location: String,
815    ) -> Result<Table> {
816        let db_name = validate_namespace(table_ident.namespace())?;
817        let table_name = table_ident.name();
818        let metadata = TableMetadata::read_from(&self.file_io, &metadata_location).await?;
819
820        let table_input = convert_to_glue_table(
821            table_name,
822            metadata_location.clone(),
823            &metadata,
824            metadata.properties(),
825            None,
826        )?;
827
828        let builder = self
829            .client
830            .0
831            .create_table()
832            .database_name(&db_name)
833            .table_input(table_input);
834        let builder = with_catalog_id!(builder, self.properties);
835
836        builder.send().await.map_err(|e| {
837            let error = e.into_service_error();
838            match error {
839                CreateTableError::EntityNotFoundException(_) => Error::new(
840                    ErrorKind::NamespaceNotFound,
841                    format!("Database {db_name} does not exist"),
842                ),
843                CreateTableError::AlreadyExistsException(_) => Error::new(
844                    ErrorKind::TableAlreadyExists,
845                    format!("Table {table_ident} already exists"),
846                ),
847                _ => Error::new(
848                    ErrorKind::Unexpected,
849                    format!("Failed to register table {table_ident} due to AWS SDK error"),
850                ),
851            }
852            .with_source(anyhow!("aws sdk error: {error:?}"))
853        })?;
854
855        let mut builder = Table::builder()
856            .identifier(table_ident.clone())
857            .metadata_location(metadata_location)
858            .metadata(metadata)
859            .file_io(self.file_io())
860            .runtime(self.runtime.clone());
861        if let Some(kms_client) = self.kms_client.clone() {
862            builder = builder.kms_client(kms_client);
863        }
864        Ok(builder.build()?)
865    }
866
867    async fn update_table(&self, commit: TableCommit) -> Result<Table> {
868        let table_ident = commit.identifier().clone();
869        let table_namespace = validate_namespace(table_ident.namespace())?;
870
871        let (current_table, current_version_id) =
872            self.load_table_with_version_id(&table_ident).await?;
873        let current_metadata_location = current_table.metadata_location_result()?.to_string();
874
875        let staged_table = commit.apply(current_table)?;
876        let staged_metadata_location_str = staged_table.metadata_location_result()?;
877        let staged_metadata_location = MetadataLocation::from_str(staged_metadata_location_str)?;
878
879        // Write new metadata
880        staged_table
881            .metadata()
882            .write_to(staged_table.file_io(), &staged_metadata_location)
883            .await?;
884
885        // Persist staged table to Glue with optimistic locking
886        let mut builder = self
887            .client
888            .0
889            .update_table()
890            .database_name(table_namespace)
891            .set_skip_archive(Some(true)) // todo make this configurable
892            .table_input(convert_to_glue_table(
893                table_ident.name(),
894                staged_metadata_location.to_string(),
895                staged_table.metadata(),
896                staged_table.metadata().properties(),
897                Some(current_metadata_location),
898            )?);
899
900        // Add VersionId for optimistic locking
901        if let Some(version_id) = current_version_id {
902            builder = builder.version_id(version_id);
903        }
904
905        let builder = with_catalog_id!(builder, self.properties);
906        let _ = builder.send().await.map_err(|e| {
907            let error = e.into_service_error();
908            match error {
909                UpdateTableError::EntityNotFoundException(_) => Error::new(
910                    ErrorKind::TableNotFound,
911                    format!("Table {table_ident} is not found"),
912                ),
913                UpdateTableError::ConcurrentModificationException(_) => Error::new(
914                    ErrorKind::CatalogCommitConflicts,
915                    format!("Commit failed for table: {table_ident}"),
916                )
917                .with_retryable(true),
918                _ => Error::new(
919                    ErrorKind::Unexpected,
920                    format!("Operation failed for table: {table_ident} for hitting aws sdk error"),
921                ),
922            }
923            .with_source(anyhow!("aws sdk error: {error:?}"))
924        })?;
925
926        Ok(staged_table)
927    }
928}
929
930#[cfg(test)]
931mod tests {
932    use std::sync::Mutex;
933
934    use super::*;
935
936    #[test]
937    fn test_catalog_properties() {
938        let properties = GlueCatalogProperties::from_properties(&HashMap::from([
939            (
940                GLUE_CATALOG_PROP_URI.to_string(),
941                "http://localhost".to_string(),
942            ),
943            (
944                GLUE_CATALOG_PROP_CATALOG_ID.to_string(),
945                "catalog".to_string(),
946            ),
947            (
948                GLUE_CATALOG_PROP_WAREHOUSE.to_string(),
949                "s3://warehouse".to_string(),
950            ),
951        ]))
952        .unwrap();
953
954        assert_eq!(properties.uri.as_deref(), Some("http://localhost"));
955        assert_eq!(properties.catalog_id.as_deref(), Some("catalog"));
956        assert_eq!(properties.warehouse, "s3://warehouse");
957    }
958
959    #[test]
960    fn test_catalog_properties_defaults() {
961        let properties = GlueCatalogProperties::from_properties(&HashMap::new()).unwrap();
962
963        assert_eq!(properties.uri, None);
964        assert_eq!(properties.catalog_id, None);
965        assert_eq!(properties.warehouse, "");
966    }
967
968    #[tokio::test]
969    async fn test_load_rejects_missing_or_empty_warehouse() {
970        for (case, props) in [
971            ("missing", HashMap::new()),
972            (
973                "empty",
974                HashMap::from([(GLUE_CATALOG_PROP_WAREHOUSE.to_string(), String::new())]),
975            ),
976        ] {
977            let error = GlueCatalogBuilder::default()
978                .load("glue", props)
979                .await
980                .unwrap_err();
981
982            assert_eq!(
983                (error.kind(), error.message()),
984                (ErrorKind::DataInvalid, "Catalog warehouse is required"),
985                "{case}"
986            );
987        }
988    }
989
990    #[tokio::test]
991    async fn test_load_forwards_all_properties_to_kms_factory() {
992        #[derive(Debug, Default)]
993        struct RecordingKmsClientFactory(Mutex<HashMap<String, String>>);
994
995        #[async_trait]
996        impl KmsClientFactory for RecordingKmsClientFactory {
997            async fn create_kms_client(
998                &self,
999                properties: &HashMap<String, String>,
1000            ) -> Result<Arc<dyn KeyManagementClient>> {
1001                *self.0.lock().unwrap() = properties.clone();
1002                Err(Error::new(
1003                    ErrorKind::Unexpected,
1004                    "KMS initialization failed",
1005                ))
1006            }
1007        }
1008
1009        let properties = HashMap::from([
1010            (
1011                GLUE_CATALOG_PROP_URI.to_string(),
1012                "http://localhost".to_string(),
1013            ),
1014            (
1015                GLUE_CATALOG_PROP_CATALOG_ID.to_string(),
1016                "catalog".to_string(),
1017            ),
1018            (
1019                GLUE_CATALOG_PROP_WAREHOUSE.to_string(),
1020                "s3://warehouse".to_string(),
1021            ),
1022            (AWS_REGION_NAME.to_string(), "us-east-1".to_string()),
1023            ("custom.property".to_string(), "value".to_string()),
1024        ]);
1025        let factory = Arc::new(RecordingKmsClientFactory::default());
1026        let error = GlueCatalogBuilder::default()
1027            .with_kms_client_factory(factory.clone())
1028            .load("glue", properties.clone())
1029            .await
1030            .unwrap_err();
1031
1032        assert_eq!(*factory.0.lock().unwrap(), properties);
1033        assert_eq!(
1034            (error.kind(), error.message()),
1035            (ErrorKind::Unexpected, "KMS initialization failed")
1036        );
1037    }
1038}
iceberg_catalog_glue/catalog.rs

iceberg_catalog_glue/
catalog.rs
