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

iceberg_catalog_glue/
catalog.rs
