iceberg_catalog_s3tables/
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::future::Future;
20use std::str::FromStr;
21use std::sync::Arc;
22
23use async_trait::async_trait;
24use aws_sdk_s3tables::operation::create_table::CreateTableOutput;
25use aws_sdk_s3tables::operation::get_namespace::GetNamespaceOutput;
26use aws_sdk_s3tables::operation::get_table::{GetTableError, GetTableOutput};
27use aws_sdk_s3tables::operation::list_tables::ListTablesOutput;
28use aws_sdk_s3tables::operation::update_table_metadata_location::UpdateTableMetadataLocationError;
29use aws_sdk_s3tables::types::OpenTableFormat;
30use iceberg::encryption::kms::{KeyManagementClient, KmsClientFactory};
31use iceberg::io::{FileIO, FileIOBuilder, StorageFactory};
32use iceberg::spec::{TableMetadata, TableMetadataBuilder};
33use iceberg::table::Table;
34use iceberg::{
35    Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
36    Runtime, TableCommit, TableCreation, TableIdent,
37};
38use iceberg_storage_opendal::OpenDalStorageFactory;
39
40use crate::utils::create_sdk_config;
41
42/// S3Tables table bucket ARN property
43pub const S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN: &str = "table_bucket_arn";
44/// S3Tables endpoint URL property
45pub const S3TABLES_CATALOG_PROP_ENDPOINT_URL: &str = "endpoint_url";
46
47/// S3Tables catalog configuration.
48#[derive(Debug)]
49struct S3TablesCatalogConfig {
50    /// Catalog name.
51    name: Option<String>,
52    /// Unlike other buckets, S3Tables bucket is not a physical bucket, but a virtual bucket
53    /// that is managed by s3tables. We can't directly access the bucket with path like
54    /// s3://{bucket_name}/{file_path}, all the operations are done with respect of the bucket
55    /// ARN.
56    table_bucket_arn: String,
57    /// Endpoint URL for the catalog.
58    endpoint_url: Option<String>,
59    /// Optional pre-configured AWS SDK client for S3Tables.
60    client: Option<aws_sdk_s3tables::Client>,
61    /// Properties for the catalog. The available properties are:
62    /// - `profile_name`: The name of the AWS profile to use.
63    /// - `region_name`: The AWS region to use.
64    /// - `aws_access_key_id`: The AWS access key ID to use.
65    /// - `aws_secret_access_key`: The AWS secret access key to use.
66    /// - `aws_session_token`: The AWS session token to use.
67    props: HashMap<String, String>,
68}
69
70/// Builder for [`S3TablesCatalog`].
71#[derive(Debug)]
72pub struct S3TablesCatalogBuilder {
73    config: S3TablesCatalogConfig,
74    storage_factory: Option<Arc<dyn StorageFactory>>,
75    kms_client_factory: Option<Arc<dyn KmsClientFactory>>,
76    runtime: Option<Runtime>,
77}
78
79/// Default builder for [`S3TablesCatalog`].
80impl Default for S3TablesCatalogBuilder {
81    fn default() -> Self {
82        Self {
83            config: S3TablesCatalogConfig {
84                name: None,
85                table_bucket_arn: "".to_string(),
86                endpoint_url: None,
87                client: None,
88                props: HashMap::new(),
89            },
90            storage_factory: None,
91            kms_client_factory: None,
92            runtime: None,
93        }
94    }
95}
96
97/// Builder methods for [`S3TablesCatalog`].
98impl S3TablesCatalogBuilder {
99    /// Configure the catalog with a custom endpoint URL (useful for local testing/mocking).
100    ///
101    /// # Behavior with Properties
102    ///
103    /// If both this method and the `endpoint_url` property are provided during catalog loading,
104    /// the property value will take precedence and overwrite the value set by this method.
105    /// This follows the general pattern where properties specified in the `load()` method
106    /// have higher priority than builder method configurations.
107    pub fn with_endpoint_url(mut self, endpoint_url: impl Into<String>) -> Self {
108        self.config.endpoint_url = Some(endpoint_url.into());
109        self
110    }
111
112    /// Configure the catalog with a pre-built AWS SDK client.
113    pub fn with_client(mut self, client: aws_sdk_s3tables::Client) -> Self {
114        self.config.client = Some(client);
115        self
116    }
117
118    /// Configure the catalog with a table bucket ARN.
119    ///
120    /// # Behavior with Properties
121    ///
122    /// If both this method and the `table_bucket_arn` property are provided during catalog loading,
123    /// the property value will take precedence and overwrite the value set by this method.
124    /// This follows the general pattern where properties specified in the `load()` method
125    /// have higher priority than builder method configurations.
126    pub fn with_table_bucket_arn(mut self, table_bucket_arn: impl Into<String>) -> Self {
127        self.config.table_bucket_arn = table_bucket_arn.into();
128        self
129    }
130}
131
132impl CatalogBuilder for S3TablesCatalogBuilder {
133    type C = S3TablesCatalog;
134
135    fn with_storage_factory(mut self, storage_factory: Arc<dyn StorageFactory>) -> Self {
136        self.storage_factory = Some(storage_factory);
137        self
138    }
139
140    fn with_kms_client_factory(mut self, kms_client_factory: Arc<dyn KmsClientFactory>) -> Self {
141        self.kms_client_factory = Some(kms_client_factory);
142        self
143    }
144
145    fn with_runtime(mut self, runtime: Runtime) -> Self {
146        self.runtime = Some(runtime);
147        self
148    }
149
150    fn load(
151        mut self,
152        name: impl Into<String>,
153        props: HashMap<String, String>,
154    ) -> impl Future<Output = Result<Self::C>> + Send {
155        let catalog_name = name.into();
156        self.config.name = Some(catalog_name.clone());
157
158        if props.contains_key(S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN) {
159            self.config.table_bucket_arn = props
160                .get(S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN)
161                .cloned()
162                .unwrap_or_default();
163        }
164
165        if props.contains_key(S3TABLES_CATALOG_PROP_ENDPOINT_URL) {
166            self.config.endpoint_url = props.get(S3TABLES_CATALOG_PROP_ENDPOINT_URL).cloned();
167        }
168
169        // Collect other remaining properties
170        self.config.props = props
171            .into_iter()
172            .filter(|(k, _)| {
173                k != S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN
174                    && k != S3TABLES_CATALOG_PROP_ENDPOINT_URL
175            })
176            .collect();
177
178        async move {
179            if catalog_name.trim().is_empty() {
180                Err(Error::new(
181                    ErrorKind::DataInvalid,
182                    "Catalog name cannot be empty",
183                ))
184            } else if self.config.table_bucket_arn.is_empty() {
185                Err(Error::new(
186                    ErrorKind::DataInvalid,
187                    "Table bucket ARN is required",
188                ))
189            } else {
190                let runtime = match self.runtime {
191                    Some(rt) => rt,
192                    None => Runtime::try_current()?,
193                };
194                let kms_client = match self.kms_client_factory {
195                    Some(factory) => Some(factory.create_kms_client(&self.config.props).await?),
196                    None => None,
197                };
198                S3TablesCatalog::new(self.config, self.storage_factory, runtime, kms_client).await
199            }
200        }
201    }
202}
203
204/// S3Tables catalog implementation.
205#[derive(Debug)]
206pub struct S3TablesCatalog {
207    config: S3TablesCatalogConfig,
208    s3tables_client: aws_sdk_s3tables::Client,
209    file_io: FileIO,
210    runtime: Runtime,
211    kms_client: Option<Arc<dyn KeyManagementClient>>,
212}
213
214impl S3TablesCatalog {
215    /// Creates a new S3Tables catalog.
216    async fn new(
217        config: S3TablesCatalogConfig,
218        storage_factory: Option<Arc<dyn StorageFactory>>,
219        runtime: Runtime,
220        kms_client: Option<Arc<dyn KeyManagementClient>>,
221    ) -> Result<Self> {
222        let s3tables_client = if let Some(client) = config.client.clone() {
223            client
224        } else {
225            let aws_config = create_sdk_config(&config.props, config.endpoint_url.clone()).await;
226            aws_sdk_s3tables::Client::new(&aws_config)
227        };
228
229        // Use provided factory or default to OpenDalStorageFactory::S3
230        let factory = storage_factory.unwrap_or_else(|| {
231            Arc::new(OpenDalStorageFactory::S3 {
232                customized_credential_load: None,
233            })
234        });
235        let file_io = FileIOBuilder::new(factory)
236            .with_props(&config.props)
237            .build();
238
239        Ok(Self {
240            config,
241            s3tables_client,
242            file_io,
243            runtime,
244            kms_client,
245        })
246    }
247
248    async fn load_table_with_version_token(
249        &self,
250        table_ident: &TableIdent,
251    ) -> Result<(Table, String)> {
252        let req = self
253            .s3tables_client
254            .get_table()
255            .table_bucket_arn(self.config.table_bucket_arn.clone())
256            .namespace(table_ident.namespace().to_url_string())
257            .name(table_ident.name());
258
259        let resp = req.send().await.map_err(|err| {
260            if err
261                .as_service_error()
262                .is_some_and(GetTableError::is_not_found_exception)
263            {
264                // S3 Tables GetTable API only reports that the resource was not found, and does not distinguish between namespaces and tables.
265                // https://docs.aws.amazon.com/AmazonS3/latest/API/API_s3Buckets_GetTable.html#API_s3Buckets_GetTable_Errors
266                Error::new(
267                    ErrorKind::TableNotFound,
268                    format!("Table {table_ident} is not found, either because the namespace or table did not exist"),
269                )
270                .with_source(err)
271            } else {
272                from_aws_sdk_error(err)
273            }
274        })?;
275
276        // when a table is created, it's possible that the metadata location is not set.
277        let metadata_location = resp.metadata_location().ok_or_else(|| {
278            Error::new(
279                ErrorKind::Unexpected,
280                format!(
281                    "Table {} does not have metadata location",
282                    table_ident.name()
283                ),
284            )
285        })?;
286        let metadata = TableMetadata::read_from(&self.file_io, metadata_location).await?;
287
288        let mut builder = Table::builder()
289            .identifier(table_ident.clone())
290            .metadata(metadata)
291            .metadata_location(metadata_location)
292            .file_io(self.file_io.clone())
293            .runtime(self.runtime.clone());
294        if let Some(kms_client) = self.kms_client.clone() {
295            builder = builder.kms_client(kms_client);
296        }
297        let table = builder.build()?;
298        Ok((table, resp.version_token))
299    }
300}
301
302#[async_trait]
303impl Catalog for S3TablesCatalog {
304    /// List namespaces from s3tables catalog.
305    ///
306    /// S3Tables doesn't support nested namespaces. If parent is provided, it will
307    /// return an empty list.
308    async fn list_namespaces(
309        &self,
310        parent: Option<&NamespaceIdent>,
311    ) -> Result<Vec<NamespaceIdent>> {
312        if parent.is_some() {
313            return Ok(vec![]);
314        }
315
316        let mut result = Vec::new();
317        let mut continuation_token = None;
318        loop {
319            let mut req = self
320                .s3tables_client
321                .list_namespaces()
322                .table_bucket_arn(self.config.table_bucket_arn.clone());
323            if let Some(token) = continuation_token {
324                req = req.continuation_token(token);
325            }
326            let resp = req.send().await.map_err(from_aws_sdk_error)?;
327            for ns in resp.namespaces() {
328                result.push(NamespaceIdent::from_vec(ns.namespace().to_vec())?);
329            }
330            continuation_token = resp.continuation_token().map(|s| s.to_string());
331            if continuation_token.is_none() {
332                break;
333            }
334        }
335        Ok(result)
336    }
337
338    /// Creates a new namespace with the given identifier and properties.
339    ///
340    /// Attempts to create a namespace defined by the `namespace`. The `properties`
341    /// parameter is ignored.
342    ///
343    /// The following naming rules apply to namespaces:
344    ///
345    /// - Names must be between 3 (min) and 63 (max) characters long.
346    /// - Names can consist only of lowercase letters, numbers, and underscores (_).
347    /// - Names must begin and end with a letter or number.
348    /// - Names must not contain hyphens (-) or periods (.).
349    ///
350    /// This function can return an error in the following situations:
351    ///
352    /// - Errors from the underlying database creation process, converted using
353    /// `from_aws_sdk_error`.
354    async fn create_namespace(
355        &self,
356        namespace: &NamespaceIdent,
357        _properties: HashMap<String, String>,
358    ) -> Result<Namespace> {
359        if self.namespace_exists(namespace).await? {
360            return Err(Error::new(
361                ErrorKind::NamespaceAlreadyExists,
362                format!("Namespace {namespace:?} already exists"),
363            ));
364        }
365
366        let req = self
367            .s3tables_client
368            .create_namespace()
369            .table_bucket_arn(self.config.table_bucket_arn.clone())
370            .namespace(namespace.to_url_string());
371        req.send().await.map_err(from_aws_sdk_error)?;
372        Ok(Namespace::with_properties(
373            namespace.clone(),
374            HashMap::new(),
375        ))
376    }
377
378    /// Retrieves a namespace by its identifier.
379    ///
380    /// Validates the given namespace identifier and then queries the
381    /// underlying database client to fetch the corresponding namespace data.
382    /// Constructs a `Namespace` object with the retrieved data and returns it.
383    ///
384    /// This function can return an error in any of the following situations:
385    /// - If there is an error querying the database, returned by
386    /// `from_aws_sdk_error`.
387    async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
388        if !self.namespace_exists(namespace).await? {
389            return Err(Error::new(
390                ErrorKind::NamespaceNotFound,
391                format!("Namespace {namespace:?} does not exist"),
392            ));
393        }
394
395        let req = self
396            .s3tables_client
397            .get_namespace()
398            .table_bucket_arn(self.config.table_bucket_arn.clone())
399            .namespace(namespace.to_url_string());
400        let resp: GetNamespaceOutput = req.send().await.map_err(from_aws_sdk_error)?;
401        let properties = HashMap::new();
402        Ok(Namespace::with_properties(
403            NamespaceIdent::from_vec(resp.namespace().to_vec())?,
404            properties,
405        ))
406    }
407
408    /// Checks if a namespace exists within the s3tables catalog.
409    ///
410    /// Validates the namespace identifier by querying the s3tables catalog
411    /// to determine if the specified namespace exists.
412    ///
413    /// # Returns
414    /// A `Result<bool>` indicating the outcome of the check:
415    /// - `Ok(true)` if the namespace exists.
416    /// - `Ok(false)` if the namespace does not exist, identified by a specific
417    /// `IsNotFoundException` variant.
418    /// - `Err(...)` if an error occurs during validation or the s3tables catalog
419    /// query, with the error encapsulating the issue.
420    async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result<bool> {
421        let req = self
422            .s3tables_client
423            .get_namespace()
424            .table_bucket_arn(self.config.table_bucket_arn.clone())
425            .namespace(namespace.to_url_string());
426        match req.send().await {
427            Ok(_) => Ok(true),
428            Err(err) => {
429                if err.as_service_error().map(|e| e.is_not_found_exception()) == Some(true) {
430                    Ok(false)
431                } else {
432                    Err(from_aws_sdk_error(err))
433                }
434            }
435        }
436    }
437
438    /// Updates the properties of an existing namespace.
439    ///
440    /// S3Tables doesn't support updating namespace properties, so this function
441    /// will always return an error.
442    async fn update_namespace(
443        &self,
444        _namespace: &NamespaceIdent,
445        _properties: HashMap<String, String>,
446    ) -> Result<()> {
447        Err(Error::new(
448            ErrorKind::FeatureUnsupported,
449            "Update namespace is not supported for s3tables catalog",
450        ))
451    }
452
453    /// Drops an existing namespace from the s3tables catalog.
454    ///
455    /// Validates the namespace identifier and then deletes the corresponding
456    /// namespace from the s3tables catalog.
457    ///
458    /// This function can return an error in the following situations:
459    /// - Errors from the underlying database deletion process, converted using
460    /// `from_aws_sdk_error`.
461    async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
462        if !self.namespace_exists(namespace).await? {
463            return Err(Error::new(
464                ErrorKind::NamespaceNotFound,
465                format!("Namespace {namespace:?} does not exist"),
466            ));
467        }
468
469        let req = self
470            .s3tables_client
471            .delete_namespace()
472            .table_bucket_arn(self.config.table_bucket_arn.clone())
473            .namespace(namespace.to_url_string());
474        req.send().await.map_err(from_aws_sdk_error)?;
475        Ok(())
476    }
477
478    /// Lists all tables within a given namespace.
479    ///
480    /// Retrieves all tables associated with the specified namespace and returns
481    /// their identifiers.
482    ///
483    /// This function can return an error in the following situations:
484    /// - Errors from the underlying database query process, converted using
485    /// `from_aws_sdk_error`.
486    async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
487        let mut result = Vec::new();
488        let mut continuation_token = None;
489        loop {
490            let mut req = self
491                .s3tables_client
492                .list_tables()
493                .table_bucket_arn(self.config.table_bucket_arn.clone())
494                .namespace(namespace.to_url_string());
495            if let Some(token) = continuation_token {
496                req = req.continuation_token(token);
497            }
498            let resp: ListTablesOutput = req.send().await.map_err(from_aws_sdk_error)?;
499            for table in resp.tables() {
500                result.push(TableIdent::new(
501                    NamespaceIdent::from_vec(table.namespace().to_vec())?,
502                    table.name().to_string(),
503                ));
504            }
505            continuation_token = resp.continuation_token().map(|s| s.to_string());
506            if continuation_token.is_none() {
507                break;
508            }
509        }
510        Ok(result)
511    }
512
513    /// Creates a new table within a specified namespace.
514    ///
515    /// Attempts to create a table defined by the `creation` parameter. The metadata
516    /// location is generated by the s3tables catalog, looks like:
517    ///
518    /// s3://{RANDOM WAREHOUSE LOCATION}/metadata/{VERSION}-{UUID}.metadata.json
519    ///
520    /// We have to get this random warehouse location after the table is created.
521    ///
522    /// This function can return an error in the following situations:
523    /// - If the location of the table is set by user, identified by a specific
524    /// `DataInvalid` variant.
525    /// - Errors from the underlying database creation process, converted using
526    /// `from_aws_sdk_error`.
527    async fn create_table(
528        &self,
529        namespace: &NamespaceIdent,
530        mut creation: TableCreation,
531    ) -> Result<Table> {
532        let table_ident = TableIdent::new(namespace.clone(), creation.name.clone());
533
534        // create table
535        let create_resp: CreateTableOutput = self
536            .s3tables_client
537            .create_table()
538            .table_bucket_arn(self.config.table_bucket_arn.clone())
539            .namespace(namespace.to_url_string())
540            .format(OpenTableFormat::Iceberg)
541            .name(table_ident.name())
542            .send()
543            .await
544            .map_err(from_aws_sdk_error)?;
545
546        // prepare table location. the warehouse location is generated by s3tables catalog,
547        // which looks like: s3://e6c9bf20-991a-46fb-kni5xs1q2yxi3xxdyxzjzigdeop1quse2b--table-s3
548        let table_location = match &creation.location {
549            Some(_) => {
550                return Err(Error::new(
551                    ErrorKind::DataInvalid,
552                    "The location of the table is generated by s3tables catalog, can't be set by user.",
553                ));
554            }
555            None => {
556                let get_resp: GetTableOutput = self
557                    .s3tables_client
558                    .get_table()
559                    .table_bucket_arn(self.config.table_bucket_arn.clone())
560                    .namespace(namespace.to_url_string())
561                    .name(table_ident.name())
562                    .send()
563                    .await
564                    .map_err(from_aws_sdk_error)?;
565                get_resp.warehouse_location().to_string()
566            }
567        };
568
569        // write metadata to file
570        creation.location = Some(table_location.clone());
571        let metadata = TableMetadataBuilder::from_table_creation(creation)?
572            .build()?
573            .metadata;
574        let metadata_location = MetadataLocation::try_new_with_metadata(&metadata)?;
575        metadata.write_to(&self.file_io, &metadata_location).await?;
576
577        // update metadata location
578        let metadata_location_str = metadata_location.to_string();
579        self.s3tables_client
580            .update_table_metadata_location()
581            .table_bucket_arn(self.config.table_bucket_arn.clone())
582            .namespace(namespace.to_url_string())
583            .name(table_ident.name())
584            .metadata_location(metadata_location_str.clone())
585            .version_token(create_resp.version_token())
586            .send()
587            .await
588            .map_err(from_aws_sdk_error)?;
589
590        let mut builder = Table::builder()
591            .identifier(table_ident)
592            .metadata_location(metadata_location_str)
593            .metadata(metadata)
594            .file_io(self.file_io.clone())
595            .runtime(self.runtime.clone());
596        if let Some(kms_client) = self.kms_client.clone() {
597            builder = builder.kms_client(kms_client);
598        }
599        let table = builder.build()?;
600        Ok(table)
601    }
602
603    /// Loads an existing table from the s3tables catalog.
604    ///
605    /// Retrieves the metadata location of the specified table and constructs a
606    /// `Table` object with the retrieved metadata.
607    ///
608    /// This function can return an error in the following situations:
609    /// - If the table does not have a metadata location, identified by a specific
610    /// `Unexpected` variant.
611    /// - Errors from the underlying database query process, converted using
612    /// `from_aws_sdk_error`.
613    async fn load_table(&self, table_ident: &TableIdent) -> Result<Table> {
614        Ok(self.load_table_with_version_token(table_ident).await?.0)
615    }
616
617    /// Not supported for S3Tables. Use `purge_table` instead.
618    ///
619    /// S3 Tables doesn't support soft delete, so dropping a table will permanently remove it from the catalog.
620    async fn drop_table(&self, _table: &TableIdent) -> Result<()> {
621        Err(Error::new(
622            ErrorKind::FeatureUnsupported,
623            "drop_table is not supported for S3Tables; use purge_table instead",
624        ))
625    }
626
627    /// Purge a table from the S3 Tables catalog.
628    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
629        let req = self
630            .s3tables_client
631            .delete_table()
632            .table_bucket_arn(self.config.table_bucket_arn.clone())
633            .namespace(table.namespace().to_url_string())
634            .name(table.name());
635        req.send().await.map_err(from_aws_sdk_error)?;
636        Ok(())
637    }
638
639    /// Checks if a table exists within the s3tables catalog.
640    ///
641    /// Validates the table identifier by querying the s3tables catalog
642    /// to determine if the specified table exists.
643    ///
644    /// # Returns
645    /// A `Result<bool>` indicating the outcome of the check:
646    /// - `Ok(true)` if the table exists.
647    /// - `Ok(false)` if the table does not exist, identified by a specific
648    /// `IsNotFoundException` variant.
649    /// - `Err(...)` if an error occurs during validation or the s3tables catalog
650    /// query, with the error encapsulating the issue.
651    async fn table_exists(&self, table_ident: &TableIdent) -> Result<bool> {
652        let req = self
653            .s3tables_client
654            .get_table()
655            .table_bucket_arn(self.config.table_bucket_arn.clone())
656            .namespace(table_ident.namespace().to_url_string())
657            .name(table_ident.name());
658        match req.send().await {
659            Ok(_) => Ok(true),
660            Err(err) => {
661                if err.as_service_error().map(|e| e.is_not_found_exception()) == Some(true) {
662                    Ok(false)
663                } else {
664                    Err(from_aws_sdk_error(err))
665                }
666            }
667        }
668    }
669
670    /// Renames an existing table within the s3tables catalog.
671    ///
672    /// Validates the source and destination table identifiers and then renames
673    /// the source table to the destination table.
674    ///
675    /// This function can return an error in the following situations:
676    /// - Errors from the underlying database renaming process, converted using
677    /// `from_aws_sdk_error`.
678    async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> {
679        let req = self
680            .s3tables_client
681            .rename_table()
682            .table_bucket_arn(self.config.table_bucket_arn.clone())
683            .namespace(src.namespace().to_url_string())
684            .name(src.name())
685            .new_namespace_name(dest.namespace().to_url_string())
686            .new_name(dest.name());
687        req.send().await.map_err(from_aws_sdk_error)?;
688        Ok(())
689    }
690
691    async fn register_table(
692        &self,
693        _table_ident: &TableIdent,
694        _metadata_location: String,
695    ) -> Result<Table> {
696        Err(Error::new(
697            ErrorKind::FeatureUnsupported,
698            "Registering a table is not supported yet",
699        ))
700    }
701
702    /// Updates an existing table within the s3tables catalog.
703    async fn update_table(&self, commit: TableCommit) -> Result<Table> {
704        let table_ident = commit.identifier().clone();
705        let table_namespace = table_ident.namespace();
706        let (current_table, version_token) =
707            self.load_table_with_version_token(&table_ident).await?;
708
709        let staged_table = commit.apply(current_table)?;
710        let staged_metadata_location_str = staged_table.metadata_location_result()?;
711        let staged_metadata_location = MetadataLocation::from_str(staged_metadata_location_str)?;
712
713        staged_table
714            .metadata()
715            .write_to(staged_table.file_io(), &staged_metadata_location)
716            .await?;
717
718        let builder = self
719            .s3tables_client
720            .update_table_metadata_location()
721            .table_bucket_arn(&self.config.table_bucket_arn)
722            .namespace(table_namespace.to_url_string())
723            .name(table_ident.name())
724            .version_token(version_token)
725            .metadata_location(staged_metadata_location_str);
726
727        let _ = builder
728            .send()
729            .await
730            .map_err(|e| match e.as_service_error() {
731                Some(UpdateTableMetadataLocationError::ConflictException(_)) => Error::new(
732                    ErrorKind::CatalogCommitConflicts,
733                    format!("Commit conflicted for table: {table_ident}"),
734                )
735                .with_retryable(true)
736                .with_source(e),
737                Some(UpdateTableMetadataLocationError::NotFoundException(_)) => Error::new(
738                    ErrorKind::TableNotFound,
739                    format!("Table {table_ident} is not found"),
740                )
741                .with_source(e),
742                _ => from_aws_sdk_error(e),
743            })?;
744
745        Ok(staged_table)
746    }
747}
748
749/// Format AWS SDK error into iceberg error
750pub(crate) fn from_aws_sdk_error<T>(error: aws_sdk_s3tables::error::SdkError<T>) -> Error
751where T: std::error::Error + Send + Sync + 'static {
752    Error::new(
753        ErrorKind::Unexpected,
754        format!("Operation failed for hitting aws sdk error: {error:?}"),
755    )
756    .with_source(error)
757}
758
759#[cfg(test)]
760mod tests {
761    use std::error::Error as _;
762
763    use futures::TryStreamExt;
764    use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
765    use iceberg::transaction::{ApplyTransactionAction, Transaction};
766
767    use super::*;
768
769    #[test]
770    fn test_from_aws_sdk_error_keeps_sdk_error_as_source() {
771        let err = from_aws_sdk_error(
772            aws_sdk_s3tables::error::SdkError::<GetTableError>::construction_failure("boom"),
773        );
774
775        assert_eq!(err.kind(), ErrorKind::Unexpected);
776        assert!(
777            err.source()
778                .and_then(|source| source
779                    .downcast_ref::<aws_sdk_s3tables::error::SdkError<GetTableError>>())
780                .is_some()
781        );
782    }
783
784    async fn load_s3tables_catalog_from_env() -> Result<Option<S3TablesCatalog>> {
785        let table_bucket_arn = match std::env::var("TABLE_BUCKET_ARN").ok() {
786            Some(table_bucket_arn) => table_bucket_arn,
787            None => return Ok(None),
788        };
789
790        let config = S3TablesCatalogConfig {
791            name: None,
792            table_bucket_arn,
793            endpoint_url: None,
794            client: None,
795            props: HashMap::new(),
796        };
797
798        Ok(Some(
799            S3TablesCatalog::new(config, None, Runtime::current(), None).await?,
800        ))
801    }
802
803    #[tokio::test]
804    async fn test_s3tables_list_namespace() {
805        let catalog = match load_s3tables_catalog_from_env().await {
806            Ok(Some(catalog)) => catalog,
807            Ok(None) => return,
808            Err(e) => panic!("Error loading catalog: {e}"),
809        };
810
811        let namespaces = catalog.list_namespaces(None).await.unwrap();
812        assert!(!namespaces.is_empty());
813    }
814
815    #[tokio::test]
816    async fn test_s3tables_list_tables() {
817        let catalog = match load_s3tables_catalog_from_env().await {
818            Ok(Some(catalog)) => catalog,
819            Ok(None) => return,
820            Err(e) => panic!("Error loading catalog: {e}"),
821        };
822
823        let tables = catalog
824            .list_tables(&NamespaceIdent::new("aws_s3_metadata".to_string()))
825            .await
826            .unwrap();
827        assert!(!tables.is_empty());
828    }
829
830    #[tokio::test]
831    async fn test_s3tables_load_table() {
832        let catalog = match load_s3tables_catalog_from_env().await {
833            Ok(Some(catalog)) => catalog,
834            Ok(None) => return,
835            Err(e) => panic!("Error loading catalog: {e}"),
836        };
837
838        let _table = catalog
839            .load_table(&TableIdent::new(
840                NamespaceIdent::new("aws_s3_metadata".to_string()),
841                "query_storage_metadata".to_string(),
842            ))
843            .await
844            .expect("table that exists should be loaded");
845
846        let load_table_err = catalog
847            .load_table(&TableIdent::new(
848                NamespaceIdent::new("not_a_namespace".to_string()),
849                "not_a_table_name".to_string(),
850            ))
851            .await
852            .expect_err("loading a table that does not exist should fail");
853        assert_eq!(
854            load_table_err.kind(),
855            ErrorKind::TableNotFound,
856            "must return table not found error for non-existent table"
857        );
858    }
859
860    #[tokio::test]
861    async fn test_s3tables_create_delete_namespace() {
862        let catalog = match load_s3tables_catalog_from_env().await {
863            Ok(Some(catalog)) => catalog,
864            Ok(None) => return,
865            Err(e) => panic!("Error loading catalog: {e}"),
866        };
867
868        let namespace = NamespaceIdent::new("test_s3tables_create_delete_namespace".to_string());
869        catalog
870            .create_namespace(&namespace, HashMap::new())
871            .await
872            .unwrap();
873        assert!(catalog.namespace_exists(&namespace).await.unwrap());
874        catalog.drop_namespace(&namespace).await.unwrap();
875        assert!(!catalog.namespace_exists(&namespace).await.unwrap());
876    }
877
878    #[tokio::test]
879    async fn test_s3tables_create_delete_table() {
880        let catalog = match load_s3tables_catalog_from_env().await {
881            Ok(Some(catalog)) => catalog,
882            Ok(None) => return,
883            Err(e) => panic!("Error loading catalog: {e}"),
884        };
885
886        let creation = {
887            let schema = Schema::builder()
888                .with_schema_id(0)
889                .with_fields(vec![
890                    NestedField::required(1, "foo", Type::Primitive(PrimitiveType::Int)).into(),
891                    NestedField::required(2, "bar", Type::Primitive(PrimitiveType::String)).into(),
892                ])
893                .build()
894                .unwrap();
895            TableCreation::builder()
896                .name("test_s3tables_create_delete_table".to_string())
897                .properties(HashMap::new())
898                .schema(schema)
899                .build()
900        };
901
902        let namespace = NamespaceIdent::new("test_s3tables_create_delete_table".to_string());
903        let table_ident = TableIdent::new(
904            namespace.clone(),
905            "test_s3tables_create_delete_table".to_string(),
906        );
907        catalog.drop_namespace(&namespace).await.ok();
908        catalog.drop_table(&table_ident).await.ok();
909
910        catalog
911            .create_namespace(&namespace, HashMap::new())
912            .await
913            .unwrap();
914        catalog.create_table(&namespace, creation).await.unwrap();
915        assert!(catalog.table_exists(&table_ident).await.unwrap());
916        catalog.drop_table(&table_ident).await.unwrap();
917        assert!(!catalog.table_exists(&table_ident).await.unwrap());
918        catalog.drop_namespace(&namespace).await.unwrap();
919    }
920
921    #[tokio::test]
922    async fn test_s3tables_update_table() {
923        let catalog = match load_s3tables_catalog_from_env().await {
924            Ok(Some(catalog)) => catalog,
925            Ok(None) => return,
926            Err(e) => panic!("Error loading catalog: {e}"),
927        };
928
929        // Create a test namespace and table
930        let namespace = NamespaceIdent::new("test_s3tables_update_table".to_string());
931        let table_ident =
932            TableIdent::new(namespace.clone(), "test_s3tables_update_table".to_string());
933
934        // Clean up any existing resources from previous test runs
935        catalog.drop_table(&table_ident).await.ok();
936        catalog.drop_namespace(&namespace).await.ok();
937
938        // Create namespace and table
939        catalog
940            .create_namespace(&namespace, HashMap::new())
941            .await
942            .unwrap();
943
944        let creation = {
945            let schema = Schema::builder()
946                .with_schema_id(0)
947                .with_fields(vec![
948                    NestedField::required(1, "foo", Type::Primitive(PrimitiveType::Int)).into(),
949                    NestedField::required(2, "bar", Type::Primitive(PrimitiveType::String)).into(),
950                ])
951                .build()
952                .unwrap();
953            TableCreation::builder()
954                .name(table_ident.name().to_string())
955                .properties(HashMap::new())
956                .schema(schema)
957                .build()
958        };
959
960        let table = catalog.create_table(&namespace, creation).await.unwrap();
961
962        // Create a transaction to update the table
963        let tx = Transaction::new(&table);
964
965        // Store the original metadata location for comparison
966        let original_metadata_location = table.metadata_location();
967
968        // Update table properties using the transaction
969        let tx = tx
970            .update_table_properties()
971            .set("test_property".to_string(), "test_value".to_string())
972            .apply(tx)
973            .unwrap();
974
975        // Commit the transaction to the catalog
976        let updated_table = tx.commit(&catalog).await.unwrap();
977
978        // Verify the update was successful
979        assert_eq!(
980            updated_table.metadata().properties().get("test_property"),
981            Some(&"test_value".to_string())
982        );
983
984        // Verify the metadata location has been updated
985        assert_ne!(
986            updated_table.metadata_location(),
987            original_metadata_location,
988            "Metadata location should be updated after commit"
989        );
990
991        // Load the table again from the catalog to verify changes were persisted
992        let reloaded_table = catalog.load_table(&table_ident).await.unwrap();
993
994        // Verify the reloaded table matches the updated table
995        assert_eq!(
996            reloaded_table.metadata().properties().get("test_property"),
997            Some(&"test_value".to_string())
998        );
999        assert_eq!(
1000            reloaded_table.metadata_location(),
1001            updated_table.metadata_location(),
1002            "Reloaded table should have the same metadata location as the updated table"
1003        );
1004    }
1005
1006    #[tokio::test]
1007    async fn test_builder_load_missing_bucket_arn() {
1008        let builder = S3TablesCatalogBuilder::default();
1009        let result = builder.load("s3tables", HashMap::new()).await;
1010
1011        assert!(result.is_err());
1012        if let Err(err) = result {
1013            assert_eq!(err.kind(), ErrorKind::DataInvalid);
1014            assert_eq!(err.message(), "Table bucket ARN is required");
1015        }
1016    }
1017
1018    #[tokio::test]
1019    async fn test_builder_with_endpoint_url_ok() {
1020        let builder = S3TablesCatalogBuilder::default().with_endpoint_url("http://localhost:4566");
1021
1022        let result = builder
1023            .load(
1024                "s3tables",
1025                HashMap::from([
1026                    (
1027                        S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(),
1028                        "arn:aws:s3tables:us-east-1:123456789012:bucket/test".to_string(),
1029                    ),
1030                    ("some_prop".to_string(), "some_value".to_string()),
1031                ]),
1032            )
1033            .await;
1034
1035        assert!(result.is_ok());
1036    }
1037
1038    #[tokio::test]
1039    async fn test_builder_with_client_ok() {
1040        use aws_config::BehaviorVersion;
1041
1042        let sdk_config = aws_config::defaults(BehaviorVersion::latest()).load().await;
1043        let client = aws_sdk_s3tables::Client::new(&sdk_config);
1044
1045        let builder = S3TablesCatalogBuilder::default().with_client(client);
1046        let result = builder
1047            .load(
1048                "s3tables",
1049                HashMap::from([(
1050                    S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(),
1051                    "arn:aws:s3tables:us-east-1:123456789012:bucket/test".to_string(),
1052                )]),
1053            )
1054            .await;
1055
1056        assert!(result.is_ok());
1057    }
1058
1059    #[tokio::test]
1060    async fn test_builder_with_table_bucket_arn() {
1061        let test_arn = "arn:aws:s3tables:us-west-2:123456789012:bucket/test-bucket";
1062        let builder = S3TablesCatalogBuilder::default().with_table_bucket_arn(test_arn);
1063
1064        let result = builder.load("s3tables", HashMap::new()).await;
1065
1066        assert!(result.is_ok());
1067        let catalog = result.unwrap();
1068        assert_eq!(catalog.config.table_bucket_arn, test_arn);
1069    }
1070
1071    #[tokio::test]
1072    async fn test_builder_empty_table_bucket_arn_edge_cases() {
1073        let mut props = HashMap::new();
1074        props.insert(
1075            S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(),
1076            "".to_string(),
1077        );
1078
1079        let builder = S3TablesCatalogBuilder::default();
1080        let result = builder.load("s3tables", props).await;
1081
1082        assert!(result.is_err());
1083        if let Err(err) = result {
1084            assert_eq!(err.kind(), ErrorKind::DataInvalid);
1085            assert_eq!(err.message(), "Table bucket ARN is required");
1086        }
1087    }
1088
1089    #[tokio::test]
1090    async fn test_endpoint_url_property_overrides_builder_method() {
1091        let test_arn = "arn:aws:s3tables:us-west-2:123456789012:bucket/test-bucket";
1092        let builder_endpoint = "http://localhost:4566";
1093        let property_endpoint = "http://localhost:8080";
1094
1095        let builder = S3TablesCatalogBuilder::default()
1096            .with_table_bucket_arn(test_arn)
1097            .with_endpoint_url(builder_endpoint);
1098
1099        let mut props = HashMap::new();
1100        props.insert(
1101            S3TABLES_CATALOG_PROP_ENDPOINT_URL.to_string(),
1102            property_endpoint.to_string(),
1103        );
1104
1105        let result = builder.load("s3tables", props).await;
1106
1107        assert!(result.is_ok());
1108        let catalog = result.unwrap();
1109
1110        // Property value should override builder method value
1111        assert_eq!(
1112            catalog.config.endpoint_url,
1113            Some(property_endpoint.to_string())
1114        );
1115        assert_ne!(
1116            catalog.config.endpoint_url,
1117            Some(builder_endpoint.to_string())
1118        );
1119    }
1120
1121    #[tokio::test]
1122    async fn test_endpoint_url_builder_method_only() {
1123        let test_arn = "arn:aws:s3tables:us-west-2:123456789012:bucket/test-bucket";
1124        let builder_endpoint = "http://localhost:4566";
1125
1126        let builder = S3TablesCatalogBuilder::default()
1127            .with_table_bucket_arn(test_arn)
1128            .with_endpoint_url(builder_endpoint);
1129
1130        let result = builder.load("s3tables", HashMap::new()).await;
1131
1132        assert!(result.is_ok());
1133        let catalog = result.unwrap();
1134
1135        assert_eq!(
1136            catalog.config.endpoint_url,
1137            Some(builder_endpoint.to_string())
1138        );
1139    }
1140
1141    #[tokio::test]
1142    async fn test_endpoint_url_property_only() {
1143        let test_arn = "arn:aws:s3tables:us-west-2:123456789012:bucket/test-bucket";
1144        let property_endpoint = "http://localhost:8080";
1145
1146        let builder = S3TablesCatalogBuilder::default().with_table_bucket_arn(test_arn);
1147
1148        let mut props = HashMap::new();
1149        props.insert(
1150            S3TABLES_CATALOG_PROP_ENDPOINT_URL.to_string(),
1151            property_endpoint.to_string(),
1152        );
1153
1154        let result = builder.load("s3tables", props).await;
1155
1156        assert!(result.is_ok());
1157        let catalog = result.unwrap();
1158
1159        assert_eq!(
1160            catalog.config.endpoint_url,
1161            Some(property_endpoint.to_string())
1162        );
1163    }
1164
1165    #[tokio::test]
1166    async fn test_table_bucket_arn_property_overrides_builder_method() {
1167        let builder_arn = "arn:aws:s3tables:us-west-2:123456789012:bucket/builder-bucket";
1168        let property_arn = "arn:aws:s3tables:us-east-1:987654321098:bucket/property-bucket";
1169
1170        let builder = S3TablesCatalogBuilder::default().with_table_bucket_arn(builder_arn);
1171
1172        let mut props = HashMap::new();
1173        props.insert(
1174            S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(),
1175            property_arn.to_string(),
1176        );
1177
1178        let result = builder.load("s3tables", props).await;
1179
1180        assert!(result.is_ok());
1181        let catalog = result.unwrap();
1182
1183        assert_eq!(catalog.config.table_bucket_arn, property_arn);
1184        assert_ne!(catalog.config.table_bucket_arn, builder_arn);
1185    }
1186
1187    #[tokio::test]
1188    async fn test_table_bucket_arn_builder_method_only() {
1189        let builder_arn = "arn:aws:s3tables:us-west-2:123456789012:bucket/builder-bucket";
1190
1191        let builder = S3TablesCatalogBuilder::default().with_table_bucket_arn(builder_arn);
1192
1193        let result = builder.load("s3tables", HashMap::new()).await;
1194
1195        assert!(result.is_ok());
1196        let catalog = result.unwrap();
1197
1198        assert_eq!(catalog.config.table_bucket_arn, builder_arn);
1199    }
1200
1201    #[tokio::test]
1202    async fn test_table_bucket_arn_property_only() {
1203        let property_arn = "arn:aws:s3tables:us-east-1:987654321098:bucket/property-bucket";
1204
1205        let builder = S3TablesCatalogBuilder::default();
1206
1207        let mut props = HashMap::new();
1208        props.insert(
1209            S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(),
1210            property_arn.to_string(),
1211        );
1212
1213        let result = builder.load("s3tables", props).await;
1214
1215        assert!(result.is_ok());
1216        let catalog = result.unwrap();
1217
1218        assert_eq!(catalog.config.table_bucket_arn, property_arn);
1219    }
1220
1221    #[tokio::test]
1222    async fn test_builder_empty_name_validation() {
1223        let test_arn = "arn:aws:s3tables:us-west-2:123456789012:bucket/test-bucket";
1224        let builder = S3TablesCatalogBuilder::default().with_table_bucket_arn(test_arn);
1225
1226        let result = builder.load("", HashMap::new()).await;
1227
1228        assert!(result.is_err());
1229        if let Err(err) = result {
1230            assert_eq!(err.kind(), ErrorKind::DataInvalid);
1231            assert_eq!(err.message(), "Catalog name cannot be empty");
1232        }
1233    }
1234
1235    #[tokio::test]
1236    async fn test_builder_whitespace_only_name_validation() {
1237        let test_arn = "arn:aws:s3tables:us-west-2:123456789012:bucket/test-bucket";
1238        let builder = S3TablesCatalogBuilder::default().with_table_bucket_arn(test_arn);
1239
1240        let result = builder.load("   \t\n  ", HashMap::new()).await;
1241
1242        assert!(result.is_err());
1243        if let Err(err) = result {
1244            assert_eq!(err.kind(), ErrorKind::DataInvalid);
1245            assert_eq!(err.message(), "Catalog name cannot be empty");
1246        }
1247    }
1248
1249    #[tokio::test]
1250    async fn test_builder_name_validation_with_missing_arn() {
1251        let builder = S3TablesCatalogBuilder::default();
1252
1253        let result = builder.load("", HashMap::new()).await;
1254
1255        assert!(result.is_err());
1256        if let Err(err) = result {
1257            assert_eq!(err.kind(), ErrorKind::DataInvalid);
1258            assert_eq!(err.message(), "Catalog name cannot be empty");
1259        }
1260    }
1261
1262    /// Verify that an S3 Table catalog can create a table, write data, load the same table, and read from it.
1263    #[tokio::test]
1264    async fn test_s3tables_create_table_write_load_table_read() {
1265        use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
1266        use iceberg::writer::file_writer::ParquetWriterBuilder;
1267        use iceberg::writer::file_writer::location_generator::{
1268            DefaultFileNameGenerator, DefaultLocationGenerator,
1269        };
1270        use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder;
1271        use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
1272
1273        let catalog = match load_s3tables_catalog_from_env().await {
1274            Ok(Some(c)) => c,
1275            Ok(None) => return,
1276            Err(e) => panic!("Error loading catalog: {e}"),
1277        };
1278
1279        let ns = NamespaceIdent::new(format!("test_rw_{}", uuid::Uuid::new_v4().simple()));
1280        catalog.create_namespace(&ns, HashMap::new()).await.unwrap();
1281
1282        let table_name = String::from("table");
1283
1284        let schema = Schema::builder()
1285            .with_fields(vec![
1286                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
1287            ])
1288            .build()
1289            .unwrap();
1290        let creation = TableCreation::builder()
1291            .name(table_name.clone())
1292            .schema(schema)
1293            .build();
1294
1295        let table = catalog.create_table(&ns, creation).await.unwrap();
1296
1297        // Write one row.
1298        let arrow_schema: Arc<arrow_schema::Schema> = Arc::new(
1299            table
1300                .metadata()
1301                .current_schema()
1302                .as_ref()
1303                .try_into()
1304                .unwrap(),
1305        );
1306        let batch = arrow_array::RecordBatch::try_new(arrow_schema, vec![Arc::new(
1307            arrow_array::Int32Array::from(vec![42]),
1308        )])
1309        .unwrap();
1310
1311        // Locations will be generated based on the table metadata, which will be using `s3://` for Amazon S3 Tables.
1312        let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap();
1313        let file_name_generator = DefaultFileNameGenerator::new(
1314            "test".to_string(),
1315            None,
1316            iceberg::spec::DataFileFormat::Parquet,
1317        );
1318        let parquet_writer_builder = ParquetWriterBuilder::new(
1319            parquet::file::properties::WriterProperties::default(),
1320            table.metadata().current_schema().clone(),
1321        );
1322        let rw = RollingFileWriterBuilder::new_with_default_file_size(
1323            parquet_writer_builder,
1324            table.file_io().clone(),
1325            location_generator,
1326            file_name_generator,
1327        );
1328        let mut writer = DataFileWriterBuilder::new(rw).build(None).await.unwrap();
1329        writer.write(batch.clone()).await.unwrap();
1330        let data_files = writer.close().await.unwrap();
1331
1332        let tx = Transaction::new(&table);
1333        let tx = tx
1334            .fast_append()
1335            .add_data_files(data_files)
1336            .apply(tx)
1337            .unwrap();
1338        tx.commit(&catalog).await.unwrap();
1339
1340        // Reload from catalog and read back.
1341        let table_ident = TableIdent::new(ns.clone(), table_name.clone());
1342        let reloaded = catalog.load_table(&table_ident).await.unwrap();
1343        let batches: Vec<arrow_array::RecordBatch> = reloaded
1344            .scan()
1345            .select_all()
1346            .build()
1347            .expect("scan to be valid (snapshot exists, schema is OK)")
1348            .to_arrow()
1349            .await
1350            .expect("scan tasks should be OK")
1351            .try_collect()
1352            .await
1353            .expect("scan should complete successfully");
1354
1355        assert_eq!(batches.len(), 1);
1356        assert_eq!(
1357            batches[0], batch,
1358            "read records should match records written earlier"
1359        );
1360
1361        // Clean up.
1362        catalog.purge_table(&table_ident).await.ok();
1363        catalog.drop_namespace(&ns).await.ok();
1364    }
1365}
iceberg_catalog_s3tables/catalog.rs

iceberg_catalog_s3tables/
catalog.rs
