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        if creation.location.is_none() {
624            let ns = self.get_namespace(namespace).await?;
625            let location =
626                get_default_table_location(&ns, &db_name, &table_name, &self.config.warehouse);
627            creation.location = Some(location);
628        }
629        let metadata = TableMetadataBuilder::from_table_creation(creation)?
630            .build()?
631            .metadata;
632        let metadata_location = MetadataLocation::try_new_with_metadata(&metadata)?;
633
634        metadata.write_to(&self.file_io, &metadata_location).await?;
635
636        let metadata_location_str = metadata_location.to_string();
637        let glue_table = convert_to_glue_table(
638            &table_name,
639            metadata_location_str.clone(),
640            &metadata,
641            metadata.properties(),
642            None,
643        )?;
644
645        let builder = self
646            .client
647            .0
648            .create_table()
649            .database_name(&db_name)
650            .table_input(glue_table);
651        let builder = with_catalog_id!(builder, self.config);
652
653        builder.send().await.map_err(from_aws_sdk_error)?;
654
655        let mut builder = Table::builder()
656            .file_io(self.file_io())
657            .metadata_location(metadata_location_str)
658            .metadata(metadata)
659            .identifier(TableIdent::new(NamespaceIdent::new(db_name), table_name))
660            .runtime(self.runtime.clone());
661        if let Some(kms_client) = self.kms_client.clone() {
662            builder = builder.kms_client(kms_client);
663        }
664        builder.build()
665    }
666
667    /// Loads a table from the Glue Catalog and constructs a `Table` object
668    /// based on its metadata.
669    ///
670    /// # Returns
671    /// A `Result` wrapping a `Table` object that represents the loaded table.
672    ///
673    /// # Errors
674    /// This function may return an error in several scenarios, including:
675    /// - Failure to validate the namespace.
676    /// - Failure to retrieve the table from the Glue Catalog.
677    /// - Absence of metadata location information in the table's properties.
678    /// - Issues reading or deserializing the table's metadata file.
679    async fn load_table(&self, table: &TableIdent) -> Result<Table> {
680        let (table, _) = self.load_table_with_version_id(table).await?;
681        Ok(table)
682    }
683
684    /// Asynchronously drops a table from the database.
685    ///
686    /// # Errors
687    /// Returns an error if:
688    /// - The namespace provided in `table` cannot be validated
689    /// or does not exist.
690    /// - The underlying database client encounters an error while
691    /// attempting to drop the table. This includes scenarios where
692    /// the table does not exist.
693    /// - Any network or communication error occurs with the database backend.
694    async fn drop_table(&self, table: &TableIdent) -> Result<()> {
695        let db_name = validate_namespace(table.namespace())?;
696        let table_name = table.name();
697
698        let builder = self
699            .client
700            .0
701            .delete_table()
702            .database_name(&db_name)
703            .name(table_name);
704        let builder = with_catalog_id!(builder, self.config);
705
706        builder.send().await.map_err(from_aws_sdk_error)?;
707
708        Ok(())
709    }
710
711    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
712        let table_info = self.load_table(table).await?;
713        self.drop_table(table).await?;
714        iceberg::drop_table_data(&table_info).await
715    }
716
717    /// Asynchronously checks the existence of a specified table
718    /// in the database.
719    ///
720    /// # Returns
721    /// - `Ok(true)` if the table exists in the database.
722    /// - `Ok(false)` if the table does not exist in the database.
723    /// - `Err(...)` if an error occurs during the process
724    async fn table_exists(&self, table: &TableIdent) -> Result<bool> {
725        let db_name = validate_namespace(table.namespace())?;
726        let table_name = table.name();
727
728        let builder = self
729            .client
730            .0
731            .get_table()
732            .database_name(&db_name)
733            .name(table_name);
734        let builder = with_catalog_id!(builder, self.config);
735
736        let resp = builder.send().await;
737
738        match resp {
739            Ok(_) => Ok(true),
740            Err(err) => {
741                if err
742                    .as_service_error()
743                    .map(|e| e.is_entity_not_found_exception())
744                    == Some(true)
745                {
746                    return Ok(false);
747                }
748                Err(from_aws_sdk_error(err))
749            }
750        }
751    }
752
753    /// Asynchronously renames a table within the database
754    /// or moves it between namespaces (databases).
755    ///
756    /// # Returns
757    /// - `Ok(())` on successful rename or move of the table.
758    /// - `Err(...)` if an error occurs during the process.
759    async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> {
760        let src_db_name = validate_namespace(src.namespace())?;
761        let dest_db_name = validate_namespace(dest.namespace())?;
762
763        let src_table_name = src.name();
764        let dest_table_name = dest.name();
765
766        let builder = self
767            .client
768            .0
769            .get_table()
770            .database_name(&src_db_name)
771            .name(src_table_name);
772        let builder = with_catalog_id!(builder, self.config);
773
774        let glue_table_output = builder.send().await.map_err(from_aws_sdk_error)?;
775
776        match glue_table_output.table() {
777            None => Err(Error::new(
778                ErrorKind::TableNotFound,
779                format!(
780                    "'Table' object for database: {src_db_name} and table: {src_table_name} does not exist"
781                ),
782            )),
783            Some(table) => {
784                let rename_table_input = TableInput::builder()
785                    .name(dest_table_name)
786                    .set_parameters(table.parameters.clone())
787                    .set_storage_descriptor(table.storage_descriptor.clone())
788                    .set_table_type(table.table_type.clone())
789                    .set_description(table.description.clone())
790                    .build()
791                    .map_err(from_aws_build_error)?;
792
793                let builder = self
794                    .client
795                    .0
796                    .create_table()
797                    .database_name(&dest_db_name)
798                    .table_input(rename_table_input);
799                let builder = with_catalog_id!(builder, self.config);
800
801                builder.send().await.map_err(from_aws_sdk_error)?;
802
803                let drop_src_table_result = self.drop_table(src).await;
804
805                match drop_src_table_result {
806                    Ok(_) => Ok(()),
807                    Err(_) => {
808                        let err_msg_src_table =
809                            format!("Failed to drop old table {src_db_name}.{src_table_name}.");
810
811                        let drop_dest_table_result = self.drop_table(dest).await;
812
813                        match drop_dest_table_result {
814                            Ok(_) => Err(Error::new(
815                                ErrorKind::Unexpected,
816                                format!(
817                                    "{err_msg_src_table} Rolled back table creation for {dest_db_name}.{dest_table_name}."
818                                ),
819                            )),
820                            Err(_) => Err(Error::new(
821                                ErrorKind::Unexpected,
822                                format!(
823                                    "{err_msg_src_table} Failed to roll back table creation for {dest_db_name}.{dest_table_name}. Please clean up manually."
824                                ),
825                            )),
826                        }
827                    }
828                }
829            }
830        }
831    }
832
833    /// registers an existing table into the Glue Catalog.
834    ///
835    /// Converts the provided table identifier and metadata location into a
836    /// Glue-compatible table representation, and attempts to create the
837    /// corresponding table in the Glue Catalog.
838    ///
839    /// # Returns
840    /// Returns `Ok(Table)` if the table is successfully registered and loaded.
841    /// If the registration fails due to validation issues, existing table conflicts,
842    /// metadata problems, or errors during the registration or loading process,
843    /// an `Err(...)` is returned.
844    async fn register_table(
845        &self,
846        table_ident: &TableIdent,
847        metadata_location: String,
848    ) -> Result<Table> {
849        let db_name = validate_namespace(table_ident.namespace())?;
850        let table_name = table_ident.name();
851        let metadata = TableMetadata::read_from(&self.file_io, &metadata_location).await?;
852
853        let table_input = convert_to_glue_table(
854            table_name,
855            metadata_location.clone(),
856            &metadata,
857            metadata.properties(),
858            None,
859        )?;
860
861        let builder = self
862            .client
863            .0
864            .create_table()
865            .database_name(&db_name)
866            .table_input(table_input);
867        let builder = with_catalog_id!(builder, self.config);
868
869        builder.send().await.map_err(|e| {
870            let error = e.into_service_error();
871            match error {
872                CreateTableError::EntityNotFoundException(_) => Error::new(
873                    ErrorKind::NamespaceNotFound,
874                    format!("Database {db_name} does not exist"),
875                ),
876                CreateTableError::AlreadyExistsException(_) => Error::new(
877                    ErrorKind::TableAlreadyExists,
878                    format!("Table {table_ident} already exists"),
879                ),
880                _ => Error::new(
881                    ErrorKind::Unexpected,
882                    format!("Failed to register table {table_ident} due to AWS SDK error"),
883                ),
884            }
885            .with_source(anyhow!("aws sdk error: {error:?}"))
886        })?;
887
888        let mut builder = Table::builder()
889            .identifier(table_ident.clone())
890            .metadata_location(metadata_location)
891            .metadata(metadata)
892            .file_io(self.file_io())
893            .runtime(self.runtime.clone());
894        if let Some(kms_client) = self.kms_client.clone() {
895            builder = builder.kms_client(kms_client);
896        }
897        Ok(builder.build()?)
898    }
899
900    async fn update_table(&self, commit: TableCommit) -> Result<Table> {
901        let table_ident = commit.identifier().clone();
902        let table_namespace = validate_namespace(table_ident.namespace())?;
903
904        let (current_table, current_version_id) =
905            self.load_table_with_version_id(&table_ident).await?;
906        let current_metadata_location = current_table.metadata_location_result()?.to_string();
907
908        let staged_table = commit.apply(current_table)?;
909        let staged_metadata_location_str = staged_table.metadata_location_result()?;
910        let staged_metadata_location = MetadataLocation::from_str(staged_metadata_location_str)?;
911
912        // Write new metadata
913        staged_table
914            .metadata()
915            .write_to(staged_table.file_io(), &staged_metadata_location)
916            .await?;
917
918        // Persist staged table to Glue with optimistic locking
919        let mut builder = self
920            .client
921            .0
922            .update_table()
923            .database_name(table_namespace)
924            .set_skip_archive(Some(true)) // todo make this configurable
925            .table_input(convert_to_glue_table(
926                table_ident.name(),
927                staged_metadata_location.to_string(),
928                staged_table.metadata(),
929                staged_table.metadata().properties(),
930                Some(current_metadata_location),
931            )?);
932
933        // Add VersionId for optimistic locking
934        if let Some(version_id) = current_version_id {
935            builder = builder.version_id(version_id);
936        }
937
938        let builder = with_catalog_id!(builder, self.config);
939        let _ = builder.send().await.map_err(|e| {
940            let error = e.into_service_error();
941            match error {
942                UpdateTableError::EntityNotFoundException(_) => Error::new(
943                    ErrorKind::TableNotFound,
944                    format!("Table {table_ident} is not found"),
945                ),
946                UpdateTableError::ConcurrentModificationException(_) => Error::new(
947                    ErrorKind::CatalogCommitConflicts,
948                    format!("Commit failed for table: {table_ident}"),
949                )
950                .with_retryable(true),
951                _ => Error::new(
952                    ErrorKind::Unexpected,
953                    format!("Operation failed for table: {table_ident} for hitting aws sdk error"),
954                ),
955            }
956            .with_source(anyhow!("aws sdk error: {error:?}"))
957        })?;
958
959        Ok(staged_table)
960    }
961}
iceberg_catalog_glue/catalog.rs

iceberg_catalog_glue/
catalog.rs
