Skip to main content

iceberg_catalog_sql/
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, HashSet};
19use std::str::FromStr;
20use std::sync::Arc;
21use std::time::Duration;
22
23use async_trait::async_trait;
24use iceberg::encryption::kms::{KeyManagementClient, KmsClientFactory};
25use iceberg::io::{FileIO, FileIOBuilder, StorageFactory};
26use iceberg::spec::{TableMetadata, TableMetadataBuilder};
27use iceberg::table::Table;
28use iceberg::{
29    Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
30    Runtime, TableCommit, TableCreation, TableIdent,
31};
32use sqlx::any::{AnyPoolOptions, AnyQueryResult, AnyRow, install_default_drivers};
33use sqlx::{Any, AnyPool, Row, Transaction};
34
35use crate::error::{
36    from_sqlx_error, no_such_namespace_err, no_such_table_err, table_already_exists_err,
37};
38
39/// catalog URI
40pub const SQL_CATALOG_PROP_URI: &str = "uri";
41/// catalog warehouse location
42pub const SQL_CATALOG_PROP_WAREHOUSE: &str = "warehouse";
43/// catalog sql bind style
44pub const SQL_CATALOG_PROP_BIND_STYLE: &str = "sql_bind_style";
45
46static CATALOG_TABLE_NAME: &str = "iceberg_tables";
47static CATALOG_FIELD_CATALOG_NAME: &str = "catalog_name";
48static CATALOG_FIELD_TABLE_NAME: &str = "table_name";
49static CATALOG_FIELD_TABLE_NAMESPACE: &str = "table_namespace";
50static CATALOG_FIELD_METADATA_LOCATION_PROP: &str = "metadata_location";
51static CATALOG_FIELD_PREVIOUS_METADATA_LOCATION_PROP: &str = "previous_metadata_location";
52static CATALOG_FIELD_RECORD_TYPE: &str = "iceberg_type";
53static CATALOG_FIELD_TABLE_RECORD_TYPE: &str = "TABLE";
54
55static NAMESPACE_TABLE_NAME: &str = "iceberg_namespace_properties";
56static NAMESPACE_FIELD_NAME: &str = "namespace";
57static NAMESPACE_FIELD_PROPERTY_KEY: &str = "property_key";
58static NAMESPACE_FIELD_PROPERTY_VALUE: &str = "property_value";
59
60static NAMESPACE_LOCATION_PROPERTY_KEY: &str = "location";
61
62static MAX_CONNECTIONS: u32 = 10; // Default the SQL pool to 10 connections if not provided
63static IDLE_TIMEOUT: u64 = 10; // Default the maximum idle timeout per connection to 10s before it is closed
64static TEST_BEFORE_ACQUIRE: bool = true; // Default the health-check of each connection to enabled prior to returning
65
66/// Builder for [`SqlCatalog`]
67#[derive(Debug)]
68pub struct SqlCatalogBuilder {
69    config: SqlCatalogConfig,
70    storage_factory: Option<Arc<dyn StorageFactory>>,
71    kms_client_factory: Option<Arc<dyn KmsClientFactory>>,
72    runtime: Option<Runtime>,
73}
74
75impl Default for SqlCatalogBuilder {
76    fn default() -> Self {
77        Self {
78            config: SqlCatalogConfig {
79                uri: "".to_string(),
80                name: "".to_string(),
81                warehouse_location: "".to_string(),
82                sql_bind_style: SqlBindStyle::DollarNumeric,
83                props: HashMap::new(),
84            },
85            storage_factory: None,
86            kms_client_factory: None,
87            runtime: None,
88        }
89    }
90}
91
92impl SqlCatalogBuilder {
93    /// Configure the database URI
94    ///
95    /// If `SQL_CATALOG_PROP_URI` has a value set in `props` during `SqlCatalogBuilder::load`,
96    /// that value takes precedence, and the value specified by this method will not be used.
97    pub fn uri(mut self, uri: impl Into<String>) -> Self {
98        self.config.uri = uri.into();
99        self
100    }
101
102    /// Configure the warehouse location
103    ///
104    /// If `SQL_CATALOG_PROP_WAREHOUSE` has a value set in `props` during `SqlCatalogBuilder::load`,
105    /// that value takes precedence, and the value specified by this method will not be used.
106    pub fn warehouse_location(mut self, location: impl Into<String>) -> Self {
107        self.config.warehouse_location = location.into();
108        self
109    }
110
111    /// Configure the bound SQL Statement
112    ///
113    /// If `SQL_CATALOG_PROP_BIND_STYLE` has a value set in `props` during `SqlCatalogBuilder::load`,
114    /// that value takes precedence, and the value specified by this method will not be used.
115    pub fn sql_bind_style(mut self, sql_bind_style: SqlBindStyle) -> Self {
116        self.config.sql_bind_style = sql_bind_style;
117        self
118    }
119
120    /// Configure the any properties
121    ///
122    /// If the same key has values set in `props` during `SqlCatalogBuilder::load`,
123    /// those values will take precedence.
124    pub fn props(mut self, props: HashMap<String, String>) -> Self {
125        for (k, v) in props {
126            self.config.props.insert(k, v);
127        }
128        self
129    }
130
131    /// Set a new property on the property to be configured.
132    /// When multiple methods are executed with the same key,
133    /// the later-set value takes precedence.
134    ///
135    /// If the same key has values set in `props` during `SqlCatalogBuilder::load`,
136    /// those values will take precedence.
137    pub fn prop(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
138        self.config.props.insert(key.into(), value.into());
139        self
140    }
141}
142
143impl CatalogBuilder for SqlCatalogBuilder {
144    type C = SqlCatalog;
145
146    fn with_storage_factory(mut self, storage_factory: Arc<dyn StorageFactory>) -> Self {
147        self.storage_factory = Some(storage_factory);
148        self
149    }
150
151    fn with_kms_client_factory(mut self, kms_client_factory: Arc<dyn KmsClientFactory>) -> Self {
152        self.kms_client_factory = Some(kms_client_factory);
153        self
154    }
155
156    fn with_runtime(mut self, runtime: Runtime) -> Self {
157        self.runtime = Some(runtime);
158        self
159    }
160
161    fn load(
162        mut self,
163        name: impl Into<String>,
164        props: HashMap<String, String>,
165    ) -> impl Future<Output = Result<Self::C>> + Send {
166        for (k, v) in props {
167            self.config.props.insert(k, v);
168        }
169
170        if let Some(uri) = self.config.props.remove(SQL_CATALOG_PROP_URI) {
171            self.config.uri = uri;
172        }
173        if let Some(warehouse_location) = self.config.props.remove(SQL_CATALOG_PROP_WAREHOUSE) {
174            self.config.warehouse_location = warehouse_location;
175        }
176
177        let name = name.into();
178
179        let mut valid_sql_bind_style = true;
180        if let Some(sql_bind_style) = self.config.props.remove(SQL_CATALOG_PROP_BIND_STYLE) {
181            if let Ok(sql_bind_style) = SqlBindStyle::from_str(&sql_bind_style) {
182                self.config.sql_bind_style = sql_bind_style;
183            } else {
184                valid_sql_bind_style = false;
185            }
186        }
187
188        let valid_name = !name.trim().is_empty();
189
190        async move {
191            if !valid_name {
192                Err(Error::new(
193                    ErrorKind::DataInvalid,
194                    "Catalog name cannot be empty",
195                ))
196            } else if !valid_sql_bind_style {
197                Err(Error::new(
198                    ErrorKind::DataInvalid,
199                    format!(
200                        "`{}` values are valid only if they're `{}` or `{}`",
201                        SQL_CATALOG_PROP_BIND_STYLE,
202                        SqlBindStyle::DollarNumeric,
203                        SqlBindStyle::QMark
204                    ),
205                ))
206            } else {
207                self.config.name = name;
208                let runtime = match self.runtime {
209                    Some(rt) => rt,
210                    None => Runtime::try_current()?,
211                };
212                let kms_client = match self.kms_client_factory {
213                    Some(factory) => Some(factory.create_kms_client(&self.config.props).await?),
214                    None => None,
215                };
216                SqlCatalog::new(self.config, self.storage_factory, runtime, kms_client).await
217            }
218        }
219    }
220}
221
222/// A struct representing the SQL catalog configuration.
223///
224/// This struct contains various parameters that are used to configure a SQL catalog,
225/// such as the database URI, warehouse location, and file I/O settings.
226/// You are required to provide a `SqlBindStyle`, which determines how SQL statements will be bound to values in the catalog.
227/// The options available for this parameter include:
228/// - `SqlBindStyle::DollarNumeric`: Binds SQL statements using `$1`, `$2`, etc., as placeholders. This is for PostgreSQL databases.
229/// - `SqlBindStyle::QuestionMark`: Binds SQL statements using `?` as a placeholder. This is for MySQL and SQLite databases.
230#[derive(Debug)]
231struct SqlCatalogConfig {
232    uri: String,
233    name: String,
234    warehouse_location: String,
235    sql_bind_style: SqlBindStyle,
236    props: HashMap<String, String>,
237}
238
239#[derive(Debug)]
240/// Sql catalog implementation.
241pub struct SqlCatalog {
242    name: String,
243    connection: AnyPool,
244    warehouse_location: String,
245    fileio: FileIO,
246    sql_bind_style: SqlBindStyle,
247    runtime: Runtime,
248    kms_client: Option<Arc<dyn KeyManagementClient>>,
249}
250
251#[derive(Debug, PartialEq, strum::EnumString, strum::Display)]
252/// Set the SQL parameter bind style to either $1..$N (Postgres style) or ? (SQLite/MySQL/MariaDB)
253pub enum SqlBindStyle {
254    /// DollarNumeric uses parameters of the form `$1..$N``, which is the Postgres style
255    DollarNumeric,
256    /// QMark uses parameters of the form `?` which is the style for other dialects (SQLite/MySQL/MariaDB)
257    QMark,
258}
259
260impl SqlCatalog {
261    /// Create new sql catalog instance
262    async fn new(
263        config: SqlCatalogConfig,
264        storage_factory: Option<Arc<dyn StorageFactory>>,
265        runtime: Runtime,
266        kms_client: Option<Arc<dyn KeyManagementClient>>,
267    ) -> Result<Self> {
268        let factory = storage_factory.ok_or_else(|| {
269            Error::new(
270                ErrorKind::Unexpected,
271                "StorageFactory must be provided for SqlCatalog. Use `with_storage_factory` to configure it.",
272            )
273        })?;
274        // Forward catalog props so storage-backend keys reach the FileIO.
275        // Unrecognized keys are ignored by backends.
276        let fileio = FileIOBuilder::new(factory)
277            .with_props(config.props.clone())
278            .build();
279
280        install_default_drivers();
281        let max_connections: u32 = config
282            .props
283            .get("pool.max-connections")
284            .map(|v| v.parse().unwrap())
285            .unwrap_or(MAX_CONNECTIONS);
286        let idle_timeout: u64 = config
287            .props
288            .get("pool.idle-timeout")
289            .map(|v| v.parse().unwrap())
290            .unwrap_or(IDLE_TIMEOUT);
291        let test_before_acquire: bool = config
292            .props
293            .get("pool.test-before-acquire")
294            .map(|v| v.parse().unwrap())
295            .unwrap_or(TEST_BEFORE_ACQUIRE);
296
297        let pool = AnyPoolOptions::new()
298            .max_connections(max_connections)
299            .idle_timeout(Duration::from_secs(idle_timeout))
300            .test_before_acquire(test_before_acquire)
301            .connect(&config.uri)
302            .await
303            .map_err(from_sqlx_error)?;
304
305        sqlx::query(&format!(
306            "CREATE TABLE IF NOT EXISTS {CATALOG_TABLE_NAME} (
307                {CATALOG_FIELD_CATALOG_NAME} VARCHAR(255) NOT NULL,
308                {CATALOG_FIELD_TABLE_NAMESPACE} VARCHAR(255) NOT NULL,
309                {CATALOG_FIELD_TABLE_NAME} VARCHAR(255) NOT NULL,
310                {CATALOG_FIELD_METADATA_LOCATION_PROP} VARCHAR(1000),
311                {CATALOG_FIELD_PREVIOUS_METADATA_LOCATION_PROP} VARCHAR(1000),
312                {CATALOG_FIELD_RECORD_TYPE} VARCHAR(5),
313                PRIMARY KEY ({CATALOG_FIELD_CATALOG_NAME}, {CATALOG_FIELD_TABLE_NAMESPACE}, {CATALOG_FIELD_TABLE_NAME}))"
314        ))
315        .execute(&pool)
316        .await
317        .map_err(from_sqlx_error)?;
318
319        sqlx::query(&format!(
320            "CREATE TABLE IF NOT EXISTS {NAMESPACE_TABLE_NAME} (
321                {CATALOG_FIELD_CATALOG_NAME} VARCHAR(255) NOT NULL,
322                {NAMESPACE_FIELD_NAME} VARCHAR(255) NOT NULL,
323                {NAMESPACE_FIELD_PROPERTY_KEY} VARCHAR(255),
324                {NAMESPACE_FIELD_PROPERTY_VALUE} VARCHAR(1000),
325                PRIMARY KEY ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}))"
326        ))
327        .execute(&pool)
328        .await
329        .map_err(from_sqlx_error)?;
330
331        Ok(SqlCatalog {
332            name: config.name.to_owned(),
333            connection: pool,
334            warehouse_location: config.warehouse_location,
335            fileio,
336            sql_bind_style: config.sql_bind_style,
337            runtime,
338            kms_client,
339        })
340    }
341
342    /// SQLX Any does not implement PostgresSQL bindings, so we have to do this.
343    fn replace_placeholders(&self, query: &str) -> String {
344        match self.sql_bind_style {
345            SqlBindStyle::DollarNumeric => {
346                let mut count = 1;
347                query
348                    .chars()
349                    .fold(String::with_capacity(query.len()), |mut acc, c| {
350                        if c == '?' {
351                            acc.push('$');
352                            acc.push_str(&count.to_string());
353                            count += 1;
354                        } else {
355                            acc.push(c);
356                        }
357                        acc
358                    })
359            }
360            _ => query.to_owned(),
361        }
362    }
363
364    /// Fetch a vec of AnyRows from a given query
365    async fn fetch_rows(&self, query: &str, args: Vec<Option<&str>>) -> Result<Vec<AnyRow>> {
366        let query_with_placeholders = self.replace_placeholders(query);
367
368        let mut sqlx_query = sqlx::query(&query_with_placeholders);
369        for arg in args {
370            sqlx_query = sqlx_query.bind(arg);
371        }
372
373        sqlx_query
374            .fetch_all(&self.connection)
375            .await
376            .map_err(from_sqlx_error)
377    }
378
379    /// Execute statements in a transaction, provided or not
380    async fn execute(
381        &self,
382        query: &str,
383        args: Vec<Option<&str>>,
384        transaction: Option<&mut Transaction<'_, Any>>,
385    ) -> Result<AnyQueryResult> {
386        let query_with_placeholders = self.replace_placeholders(query);
387
388        let mut sqlx_query = sqlx::query(&query_with_placeholders);
389        for arg in args {
390            sqlx_query = sqlx_query.bind(arg);
391        }
392
393        match transaction {
394            Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
395            None => {
396                let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
397                let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
398                let _ = tx.commit().await.map_err(from_sqlx_error);
399                result
400            }
401        }
402    }
403}
404
405#[async_trait]
406impl Catalog for SqlCatalog {
407    async fn list_namespaces(
408        &self,
409        parent: Option<&NamespaceIdent>,
410    ) -> Result<Vec<NamespaceIdent>> {
411        // UNION will remove duplicates.
412        let all_namespaces_stmt = format!(
413            "SELECT {CATALOG_FIELD_TABLE_NAMESPACE}
414             FROM {CATALOG_TABLE_NAME}
415             WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
416             UNION
417             SELECT {NAMESPACE_FIELD_NAME}
418             FROM {NAMESPACE_TABLE_NAME}
419             WHERE {CATALOG_FIELD_CATALOG_NAME} = ?"
420        );
421
422        let namespace_rows = self
423            .fetch_rows(&all_namespaces_stmt, vec![
424                Some(&self.name),
425                Some(&self.name),
426            ])
427            .await?;
428
429        let mut namespaces = HashSet::<NamespaceIdent>::with_capacity(namespace_rows.len());
430
431        if let Some(parent) = parent {
432            if self.namespace_exists(parent).await? {
433                let parent_str = parent.join(".");
434
435                for row in namespace_rows.iter() {
436                    let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
437                    // if parent = a, then we only want to see a.b, a.c returned.
438                    if nsp != parent_str && nsp.starts_with(&parent_str) {
439                        namespaces.insert(NamespaceIdent::from_strs(nsp.split("."))?);
440                    }
441                }
442
443                Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
444            } else {
445                no_such_namespace_err(parent)
446            }
447        } else {
448            for row in namespace_rows.iter() {
449                let nsp = row.try_get::<String, _>(0).map_err(from_sqlx_error)?;
450                let mut levels = nsp.split(".").collect::<Vec<&str>>();
451                if !levels.is_empty() {
452                    let first_level = levels.drain(..1).collect::<Vec<&str>>();
453                    namespaces.insert(NamespaceIdent::from_strs(first_level)?);
454                }
455            }
456
457            Ok(namespaces.into_iter().collect::<Vec<NamespaceIdent>>())
458        }
459    }
460
461    async fn create_namespace(
462        &self,
463        namespace: &NamespaceIdent,
464        properties: HashMap<String, String>,
465    ) -> Result<Namespace> {
466        let exists = self.namespace_exists(namespace).await?;
467
468        if exists {
469            return Err(Error::new(
470                ErrorKind::NamespaceAlreadyExists,
471                format!("Namespace {namespace:?} already exists"),
472            ));
473        }
474
475        let namespace_str = namespace.join(".");
476        let insert = format!(
477            "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
478             VALUES (?, ?, ?, ?)");
479        if !properties.is_empty() {
480            let mut insert_properties = properties.clone();
481            insert_properties.insert("exists".to_string(), "true".to_string());
482
483            let mut query_args = Vec::with_capacity(insert_properties.len() * 4);
484            let mut insert_stmt = insert.clone();
485            for (index, (key, value)) in insert_properties.iter().enumerate() {
486                query_args.extend_from_slice(&[
487                    Some(self.name.as_str()),
488                    Some(namespace_str.as_str()),
489                    Some(key.as_str()),
490                    Some(value.as_str()),
491                ]);
492                if index > 0 {
493                    insert_stmt.push_str(", (?, ?, ?, ?)");
494                }
495            }
496
497            self.execute(&insert_stmt, query_args, None).await?;
498
499            Ok(Namespace::with_properties(
500                namespace.clone(),
501                insert_properties,
502            ))
503        } else {
504            // set a default property of exists = true
505            self.execute(
506                &insert,
507                vec![
508                    Some(&self.name),
509                    Some(&namespace_str),
510                    Some("exists"),
511                    Some("true"),
512                ],
513                None,
514            )
515            .await?;
516            Ok(Namespace::with_properties(namespace.clone(), properties))
517        }
518    }
519
520    async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
521        let exists = self.namespace_exists(namespace).await?;
522        if exists {
523            let namespace_props = self
524                .fetch_rows(
525                    &format!(
526                        "SELECT
527                            {NAMESPACE_FIELD_NAME},
528                            {NAMESPACE_FIELD_PROPERTY_KEY},
529                            {NAMESPACE_FIELD_PROPERTY_VALUE}
530                            FROM {NAMESPACE_TABLE_NAME}
531                            WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
532                            AND {NAMESPACE_FIELD_NAME} = ?"
533                    ),
534                    vec![Some(&self.name), Some(&namespace.join("."))],
535                )
536                .await?;
537
538            let mut properties = HashMap::with_capacity(namespace_props.len());
539
540            for row in namespace_props {
541                let key = row
542                    .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_KEY)
543                    .map_err(from_sqlx_error)?;
544                let value = row
545                    .try_get::<String, _>(NAMESPACE_FIELD_PROPERTY_VALUE)
546                    .map_err(from_sqlx_error)?;
547
548                properties.insert(key, value);
549            }
550
551            Ok(Namespace::with_properties(namespace.clone(), properties))
552        } else {
553            no_such_namespace_err(namespace)
554        }
555    }
556
557    async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result<bool> {
558        let namespace_str = namespace.join(".");
559
560        let table_namespaces = self
561            .fetch_rows(
562                &format!(
563                    "SELECT 1 FROM {CATALOG_TABLE_NAME}
564                     WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
565                      AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
566                     LIMIT 1"
567                ),
568                vec![Some(&self.name), Some(&namespace_str)],
569            )
570            .await?;
571
572        if !table_namespaces.is_empty() {
573            Ok(true)
574        } else {
575            let namespaces = self
576                .fetch_rows(
577                    &format!(
578                        "SELECT 1 FROM {NAMESPACE_TABLE_NAME}
579                         WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
580                          AND {NAMESPACE_FIELD_NAME} = ?
581                         LIMIT 1"
582                    ),
583                    vec![Some(&self.name), Some(&namespace_str)],
584                )
585                .await?;
586            if !namespaces.is_empty() {
587                Ok(true)
588            } else {
589                Ok(false)
590            }
591        }
592    }
593
594    async fn update_namespace(
595        &self,
596        namespace: &NamespaceIdent,
597        properties: HashMap<String, String>,
598    ) -> Result<()> {
599        let exists = self.namespace_exists(namespace).await?;
600        if exists {
601            let existing_properties = self.get_namespace(namespace).await?.properties().clone();
602            let namespace_str = namespace.join(".");
603
604            let mut updates = vec![];
605            let mut inserts = vec![];
606
607            for (key, value) in properties.iter() {
608                if existing_properties.contains_key(key) {
609                    if existing_properties.get(key) != Some(value) {
610                        updates.push((key, value));
611                    }
612                } else {
613                    inserts.push((key, value));
614                }
615            }
616
617            let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
618            let update_stmt = format!(
619                "UPDATE {NAMESPACE_TABLE_NAME} SET {NAMESPACE_FIELD_PROPERTY_VALUE} = ?
620                 WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
621                 AND {NAMESPACE_FIELD_NAME} = ?
622                 AND {NAMESPACE_FIELD_PROPERTY_KEY} = ?"
623            );
624
625            let insert_stmt = format!(
626                "INSERT INTO {NAMESPACE_TABLE_NAME} ({CATALOG_FIELD_CATALOG_NAME}, {NAMESPACE_FIELD_NAME}, {NAMESPACE_FIELD_PROPERTY_KEY}, {NAMESPACE_FIELD_PROPERTY_VALUE})
627                 VALUES (?, ?, ?, ?)"
628            );
629
630            for (key, value) in updates {
631                self.execute(
632                    &update_stmt,
633                    vec![
634                        Some(value),
635                        Some(&self.name),
636                        Some(&namespace_str),
637                        Some(key),
638                    ],
639                    Some(&mut tx),
640                )
641                .await?;
642            }
643
644            for (key, value) in inserts {
645                self.execute(
646                    &insert_stmt,
647                    vec![
648                        Some(&self.name),
649                        Some(&namespace_str),
650                        Some(key),
651                        Some(value),
652                    ],
653                    Some(&mut tx),
654                )
655                .await?;
656            }
657
658            let _ = tx.commit().await.map_err(from_sqlx_error)?;
659
660            Ok(())
661        } else {
662            no_such_namespace_err(namespace)
663        }
664    }
665
666    async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
667        let exists = self.namespace_exists(namespace).await?;
668        if exists {
669            // if there are tables in the namespace, don't allow drop.
670            let tables = self.list_tables(namespace).await?;
671            if !tables.is_empty() {
672                return Err(Error::new(
673                    ErrorKind::Unexpected,
674                    format!(
675                        "Namespace {:?} is not empty. {} tables exist.",
676                        namespace,
677                        tables.len()
678                    ),
679                ));
680            }
681
682            self.execute(
683                &format!(
684                    "DELETE FROM {NAMESPACE_TABLE_NAME}
685                     WHERE {NAMESPACE_FIELD_NAME} = ?
686                      AND {CATALOG_FIELD_CATALOG_NAME} = ?"
687                ),
688                vec![Some(&namespace.join(".")), Some(&self.name)],
689                None,
690            )
691            .await?;
692
693            Ok(())
694        } else {
695            no_such_namespace_err(namespace)
696        }
697    }
698
699    async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
700        let exists = self.namespace_exists(namespace).await?;
701        if exists {
702            let rows = self
703                .fetch_rows(
704                    &format!(
705                        "SELECT {CATALOG_FIELD_TABLE_NAME},
706                                {CATALOG_FIELD_TABLE_NAMESPACE}
707                         FROM {CATALOG_TABLE_NAME}
708                         WHERE {CATALOG_FIELD_TABLE_NAMESPACE} = ?
709                          AND {CATALOG_FIELD_CATALOG_NAME} = ?
710                          AND (
711                                {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
712                                OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
713                          )",
714                    ),
715                    vec![Some(&namespace.join(".")), Some(&self.name)],
716                )
717                .await?;
718
719            let mut tables = HashSet::<TableIdent>::with_capacity(rows.len());
720
721            for row in rows.iter() {
722                let tbl = row
723                    .try_get::<String, _>(CATALOG_FIELD_TABLE_NAME)
724                    .map_err(from_sqlx_error)?;
725                let ns_strs = row
726                    .try_get::<String, _>(CATALOG_FIELD_TABLE_NAMESPACE)
727                    .map_err(from_sqlx_error)?;
728                let ns = NamespaceIdent::from_strs(ns_strs.split("."))?;
729                tables.insert(TableIdent::new(ns, tbl));
730            }
731
732            Ok(tables.into_iter().collect::<Vec<TableIdent>>())
733        } else {
734            no_such_namespace_err(namespace)
735        }
736    }
737
738    async fn table_exists(&self, identifier: &TableIdent) -> Result<bool> {
739        let namespace = identifier.namespace().join(".");
740        let table_name = identifier.name();
741        let table_counts = self
742            .fetch_rows(
743                &format!(
744                    "SELECT 1
745                     FROM {CATALOG_TABLE_NAME}
746                     WHERE {CATALOG_FIELD_TABLE_NAMESPACE} = ?
747                      AND {CATALOG_FIELD_CATALOG_NAME} = ?
748                      AND {CATALOG_FIELD_TABLE_NAME} = ?
749                      AND (
750                        {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
751                        OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
752                      )"
753                ),
754                vec![Some(&namespace), Some(&self.name), Some(table_name)],
755            )
756            .await?;
757
758        if !table_counts.is_empty() {
759            Ok(true)
760        } else {
761            Ok(false)
762        }
763    }
764
765    async fn drop_table(&self, identifier: &TableIdent) -> Result<()> {
766        if !self.table_exists(identifier).await? {
767            return no_such_table_err(identifier);
768        }
769
770        self.execute(
771            &format!(
772                "DELETE FROM {CATALOG_TABLE_NAME}
773                 WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
774                  AND {CATALOG_FIELD_TABLE_NAME} = ?
775                  AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
776                  AND (
777                    {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
778                    OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
779                  )"
780            ),
781            vec![
782                Some(&self.name),
783                Some(identifier.name()),
784                Some(&identifier.namespace().join(".")),
785            ],
786            None,
787        )
788        .await?;
789
790        Ok(())
791    }
792
793    async fn purge_table(&self, table: &TableIdent) -> Result<()> {
794        let table_info = self.load_table(table).await?;
795        self.drop_table(table).await?;
796        iceberg::drop_table_data(&table_info).await
797    }
798
799    async fn load_table(&self, identifier: &TableIdent) -> Result<Table> {
800        if !self.table_exists(identifier).await? {
801            return no_such_table_err(identifier);
802        }
803
804        let rows = self
805            .fetch_rows(
806                &format!(
807                    "SELECT {CATALOG_FIELD_METADATA_LOCATION_PROP}
808                     FROM {CATALOG_TABLE_NAME}
809                     WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
810                      AND {CATALOG_FIELD_TABLE_NAME} = ?
811                      AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
812                      AND (
813                        {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
814                        OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
815                      )"
816                ),
817                vec![
818                    Some(&self.name),
819                    Some(identifier.name()),
820                    Some(&identifier.namespace().join(".")),
821                ],
822            )
823            .await?;
824
825        if rows.is_empty() {
826            return no_such_table_err(identifier);
827        }
828
829        let row = &rows[0];
830        let tbl_metadata_location = row
831            .try_get::<String, _>(CATALOG_FIELD_METADATA_LOCATION_PROP)
832            .map_err(from_sqlx_error)?;
833
834        let metadata = TableMetadata::read_from(&self.fileio, &tbl_metadata_location).await?;
835
836        let mut builder = Table::builder()
837            .file_io(self.fileio.clone())
838            .identifier(identifier.clone())
839            .metadata_location(tbl_metadata_location)
840            .metadata(metadata)
841            .runtime(self.runtime.clone());
842        if let Some(kms_client) = self.kms_client.clone() {
843            builder = builder.kms_client(kms_client);
844        }
845        Ok(builder.build()?)
846    }
847
848    async fn create_table(
849        &self,
850        namespace: &NamespaceIdent,
851        creation: TableCreation,
852    ) -> Result<Table> {
853        if !self.namespace_exists(namespace).await? {
854            return no_such_namespace_err(namespace);
855        }
856
857        let tbl_name = creation.name.clone();
858        let tbl_ident = TableIdent::new(namespace.clone(), tbl_name.clone());
859
860        if self.table_exists(&tbl_ident).await? {
861            return table_already_exists_err(&tbl_ident);
862        }
863
864        let (tbl_creation, location) = match creation.location.clone() {
865            Some(location) => (creation, location),
866            None => {
867                // fall back to namespace-specific location
868                // and then to warehouse location
869                let nsp_properties = self.get_namespace(namespace).await?.properties().clone();
870                let nsp_location = match nsp_properties.get(NAMESPACE_LOCATION_PROPERTY_KEY) {
871                    Some(location) => location.clone(),
872                    None => {
873                        format!(
874                            "{}/{}",
875                            self.warehouse_location.clone(),
876                            namespace.join("/")
877                        )
878                    }
879                };
880
881                let tbl_location = format!("{}/{}", nsp_location, tbl_ident.name());
882
883                (
884                    TableCreation {
885                        location: Some(tbl_location.clone()),
886                        ..creation
887                    },
888                    tbl_location,
889                )
890            }
891        };
892
893        let tbl_metadata = TableMetadataBuilder::from_table_creation(tbl_creation)?
894            .build()?
895            .metadata;
896        let tbl_metadata_location =
897            MetadataLocation::new_with_metadata(location.clone(), &tbl_metadata);
898
899        tbl_metadata
900            .write_to(&self.fileio, &tbl_metadata_location)
901            .await?;
902
903        let tbl_metadata_location_str = tbl_metadata_location.to_string();
904        self.execute(&format!(
905            "INSERT INTO {CATALOG_TABLE_NAME}
906             ({CATALOG_FIELD_CATALOG_NAME}, {CATALOG_FIELD_TABLE_NAMESPACE}, {CATALOG_FIELD_TABLE_NAME}, {CATALOG_FIELD_METADATA_LOCATION_PROP}, {CATALOG_FIELD_RECORD_TYPE})
907             VALUES (?, ?, ?, ?, ?)
908            "), vec![Some(&self.name), Some(&namespace.join(".")), Some(&tbl_name.clone()), Some(&tbl_metadata_location_str), Some(CATALOG_FIELD_TABLE_RECORD_TYPE)], None).await?;
909
910        let mut builder = Table::builder()
911            .file_io(self.fileio.clone())
912            .metadata_location(tbl_metadata_location_str)
913            .identifier(tbl_ident)
914            .metadata(tbl_metadata)
915            .runtime(self.runtime.clone());
916        if let Some(kms_client) = self.kms_client.clone() {
917            builder = builder.kms_client(kms_client);
918        }
919        Ok(builder.build()?)
920    }
921
922    async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> {
923        if src == dest {
924            return Ok(());
925        }
926
927        if !self.table_exists(src).await? {
928            return no_such_table_err(src);
929        }
930
931        if !self.namespace_exists(dest.namespace()).await? {
932            return no_such_namespace_err(dest.namespace());
933        }
934
935        if self.table_exists(dest).await? {
936            return table_already_exists_err(dest);
937        }
938
939        self.execute(
940            &format!(
941                "UPDATE {CATALOG_TABLE_NAME}
942                 SET {CATALOG_FIELD_TABLE_NAME} = ?, {CATALOG_FIELD_TABLE_NAMESPACE} = ?
943                 WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
944                  AND {CATALOG_FIELD_TABLE_NAME} = ?
945                  AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
946                  AND (
947                    {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
948                    OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
949                )"
950            ),
951            vec![
952                Some(dest.name()),
953                Some(&dest.namespace().join(".")),
954                Some(&self.name),
955                Some(src.name()),
956                Some(&src.namespace().join(".")),
957            ],
958            None,
959        )
960        .await?;
961
962        Ok(())
963    }
964
965    async fn register_table(
966        &self,
967        table_ident: &TableIdent,
968        metadata_location: String,
969    ) -> Result<Table> {
970        if self.table_exists(table_ident).await? {
971            return table_already_exists_err(table_ident);
972        }
973
974        let metadata = TableMetadata::read_from(&self.fileio, &metadata_location).await?;
975
976        let namespace = table_ident.namespace();
977        let tbl_name = table_ident.name().to_string();
978
979        self.execute(&format!(
980            "INSERT INTO {CATALOG_TABLE_NAME}
981             ({CATALOG_FIELD_CATALOG_NAME}, {CATALOG_FIELD_TABLE_NAMESPACE}, {CATALOG_FIELD_TABLE_NAME}, {CATALOG_FIELD_METADATA_LOCATION_PROP}, {CATALOG_FIELD_RECORD_TYPE})
982             VALUES (?, ?, ?, ?, ?)
983            "), vec![Some(&self.name), Some(&namespace.join(".")), Some(&tbl_name), Some(&metadata_location), Some(CATALOG_FIELD_TABLE_RECORD_TYPE)], None).await?;
984
985        let mut builder = Table::builder()
986            .identifier(table_ident.clone())
987            .metadata_location(metadata_location)
988            .metadata(metadata)
989            .file_io(self.fileio.clone())
990            .runtime(self.runtime.clone());
991        if let Some(kms_client) = self.kms_client.clone() {
992            builder = builder.kms_client(kms_client);
993        }
994        Ok(builder.build()?)
995    }
996
997    /// Updates an existing table within the SQL catalog.
998    async fn update_table(&self, commit: TableCommit) -> Result<Table> {
999        let table_ident = commit.identifier().clone();
1000        let current_table = self.load_table(&table_ident).await?;
1001        let current_metadata_location = current_table.metadata_location_result()?.to_string();
1002
1003        let staged_table = commit.apply(current_table)?;
1004        let staged_metadata_location_str = staged_table.metadata_location_result()?;
1005        let staged_metadata_location = MetadataLocation::from_str(staged_metadata_location_str)?;
1006
1007        staged_table
1008            .metadata()
1009            .write_to(staged_table.file_io(), &staged_metadata_location)
1010            .await?;
1011
1012        let staged_metadata_location_str = staged_metadata_location.to_string();
1013        let update_result = self
1014            .execute(
1015                &format!(
1016                    "UPDATE {CATALOG_TABLE_NAME}
1017                     SET {CATALOG_FIELD_METADATA_LOCATION_PROP} = ?, {CATALOG_FIELD_PREVIOUS_METADATA_LOCATION_PROP} = ?
1018                     WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
1019                      AND {CATALOG_FIELD_TABLE_NAME} = ?
1020                      AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
1021                      AND (
1022                        {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
1023                        OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
1024                      )
1025                      AND {CATALOG_FIELD_METADATA_LOCATION_PROP} = ?"
1026                ),
1027                vec![
1028                    Some(&staged_metadata_location_str),
1029                    Some(current_metadata_location.as_str()),
1030                    Some(&self.name),
1031                    Some(table_ident.name()),
1032                    Some(&table_ident.namespace().join(".")),
1033                    Some(current_metadata_location.as_str()),
1034                ],
1035                None,
1036            )
1037            .await?;
1038
1039        if update_result.rows_affected() == 0 {
1040            return Err(Error::new(
1041                ErrorKind::CatalogCommitConflicts,
1042                format!("Commit conflicted for table: {table_ident}"),
1043            )
1044            .with_retryable(true));
1045        }
1046
1047        Ok(staged_table)
1048    }
1049}
1050
1051#[cfg(test)]
1052mod tests {
1053    use std::collections::{HashMap, HashSet};
1054    use std::hash::Hash;
1055    use std::sync::Arc;
1056
1057    use iceberg::io::LocalFsStorageFactory;
1058    use iceberg::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, SortOrder, Type};
1059    use iceberg::table::Table;
1060    use iceberg::{Catalog, CatalogBuilder, Namespace, NamespaceIdent, TableCreation, TableIdent};
1061    use itertools::Itertools;
1062    use regex::Regex;
1063    use sqlx::migrate::MigrateDatabase;
1064    use tempfile::TempDir;
1065
1066    use crate::catalog::{
1067        NAMESPACE_LOCATION_PROPERTY_KEY, SQL_CATALOG_PROP_BIND_STYLE, SQL_CATALOG_PROP_URI,
1068        SQL_CATALOG_PROP_WAREHOUSE,
1069    };
1070    use crate::{SqlBindStyle, SqlCatalogBuilder};
1071
1072    const UUID_REGEX_STR: &str = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
1073
1074    fn temp_path() -> String {
1075        let temp_dir = TempDir::new().unwrap();
1076        temp_dir.path().to_str().unwrap().to_string()
1077    }
1078
1079    fn to_set<T: Eq + Hash>(vec: Vec<T>) -> HashSet<T> {
1080        HashSet::from_iter(vec)
1081    }
1082
1083    fn default_properties() -> HashMap<String, String> {
1084        HashMap::from([("exists".to_string(), "true".to_string())])
1085    }
1086
1087    /// Create a new SQLite catalog for testing. If name is not specified it defaults to "iceberg".
1088    async fn new_sql_catalog(
1089        warehouse_location: String,
1090        name: Option<impl ToString>,
1091    ) -> impl Catalog {
1092        let name = if let Some(name) = name {
1093            name.to_string()
1094        } else {
1095            "iceberg".to_string()
1096        };
1097        let sql_lite_uri = format!("sqlite:{}", temp_path());
1098        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1099
1100        let props = HashMap::from_iter([
1101            (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri.to_string()),
1102            (SQL_CATALOG_PROP_WAREHOUSE.to_string(), warehouse_location),
1103            (
1104                SQL_CATALOG_PROP_BIND_STYLE.to_string(),
1105                SqlBindStyle::DollarNumeric.to_string(),
1106            ),
1107        ]);
1108        SqlCatalogBuilder::default()
1109            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1110            .load(&name, props)
1111            .await
1112            .unwrap()
1113    }
1114
1115    async fn create_namespace<C: Catalog>(catalog: &C, namespace_ident: &NamespaceIdent) {
1116        let _ = catalog
1117            .create_namespace(namespace_ident, HashMap::new())
1118            .await
1119            .unwrap();
1120    }
1121
1122    async fn create_namespaces<C: Catalog>(catalog: &C, namespace_idents: &Vec<&NamespaceIdent>) {
1123        for namespace_ident in namespace_idents {
1124            let _ = create_namespace(catalog, namespace_ident).await;
1125        }
1126    }
1127
1128    fn simple_table_schema() -> Schema {
1129        Schema::builder()
1130            .with_fields(vec![
1131                NestedField::required(1, "foo", Type::Primitive(PrimitiveType::Int)).into(),
1132            ])
1133            .build()
1134            .unwrap()
1135    }
1136
1137    async fn create_table<C: Catalog>(catalog: &C, table_ident: &TableIdent) {
1138        let _ = catalog
1139            .create_table(
1140                &table_ident.namespace,
1141                TableCreation::builder()
1142                    .name(table_ident.name().into())
1143                    .schema(simple_table_schema())
1144                    .location(temp_path())
1145                    .build(),
1146            )
1147            .await
1148            .unwrap();
1149    }
1150
1151    async fn create_tables<C: Catalog>(catalog: &C, table_idents: Vec<&TableIdent>) {
1152        for table_ident in table_idents {
1153            create_table(catalog, table_ident).await;
1154        }
1155    }
1156
1157    fn assert_table_eq(table: &Table, expected_table_ident: &TableIdent, expected_schema: &Schema) {
1158        assert_eq!(table.identifier(), expected_table_ident);
1159
1160        let metadata = table.metadata();
1161
1162        assert_eq!(metadata.current_schema().as_ref(), expected_schema);
1163
1164        let expected_partition_spec = PartitionSpec::builder(expected_schema.clone())
1165            .with_spec_id(0)
1166            .build()
1167            .unwrap();
1168
1169        assert_eq!(
1170            metadata
1171                .partition_specs_iter()
1172                .map(|p| p.as_ref())
1173                .collect_vec(),
1174            vec![&expected_partition_spec]
1175        );
1176
1177        let expected_sorted_order = SortOrder::builder()
1178            .with_order_id(0)
1179            .with_fields(vec![])
1180            .build(expected_schema)
1181            .unwrap();
1182
1183        assert_eq!(
1184            metadata
1185                .sort_orders_iter()
1186                .map(|s| s.as_ref())
1187                .collect_vec(),
1188            vec![&expected_sorted_order]
1189        );
1190
1191        assert_eq!(metadata.properties(), &HashMap::new());
1192
1193        assert!(!table.readonly());
1194    }
1195
1196    fn assert_table_metadata_location_matches(table: &Table, regex_str: &str) {
1197        let actual = table.metadata_location().unwrap().to_string();
1198        let regex = Regex::new(regex_str).unwrap();
1199        assert!(regex.is_match(&actual))
1200    }
1201
1202    #[tokio::test]
1203    async fn test_initialized() {
1204        let warehouse_loc = temp_path();
1205        new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
1206        // catalog instantiation should not fail even if tables exist
1207        new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
1208        new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
1209    }
1210
1211    // Regression test: storage-backend props set on the catalog must reach
1212    // the FileIO; otherwise authenticated backends fail with 401s on writes.
1213    #[tokio::test]
1214    async fn test_storage_props_propagate_to_file_io() {
1215        let sql_lite_uri = format!("sqlite:{}", temp_path());
1216        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1217        let warehouse_location = temp_path();
1218
1219        let catalog = SqlCatalogBuilder::default()
1220            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1221            .load(
1222                "iceberg",
1223                HashMap::from_iter([
1224                    (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri),
1225                    (SQL_CATALOG_PROP_WAREHOUSE.to_string(), warehouse_location),
1226                    ("s3.region".to_string(), "us-east-1".to_string()),
1227                    ("hf.token".to_string(), "hf_test_token".to_string()),
1228                ]),
1229            )
1230            .await
1231            .unwrap();
1232
1233        let props = catalog.fileio.config().props();
1234        assert_eq!(props.get("s3.region"), Some(&"us-east-1".to_string()));
1235        assert_eq!(props.get("hf.token"), Some(&"hf_test_token".to_string()));
1236    }
1237
1238    #[tokio::test]
1239    async fn test_builder_method() {
1240        let sql_lite_uri = format!("sqlite:{}", temp_path());
1241        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1242        let warehouse_location = temp_path();
1243
1244        let catalog = SqlCatalogBuilder::default()
1245            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1246            .uri(sql_lite_uri.to_string())
1247            .warehouse_location(warehouse_location.clone())
1248            .sql_bind_style(SqlBindStyle::QMark)
1249            .load("iceberg", HashMap::default())
1250            .await;
1251        assert!(catalog.is_ok());
1252
1253        let catalog = catalog.unwrap();
1254        assert!(catalog.warehouse_location == warehouse_location);
1255        assert!(catalog.sql_bind_style == SqlBindStyle::QMark);
1256    }
1257
1258    /// Overwriting an sqlite database with a non-existent path causes
1259    /// catalog generation to fail
1260    #[tokio::test]
1261    async fn test_builder_props_non_existent_path_fails() {
1262        let sql_lite_uri = format!("sqlite:{}", temp_path());
1263        let sql_lite_uri2 = format!("sqlite:{}", temp_path());
1264        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1265        let warehouse_location = temp_path();
1266
1267        let catalog = SqlCatalogBuilder::default()
1268            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1269            .uri(sql_lite_uri)
1270            .warehouse_location(warehouse_location)
1271            .load(
1272                "iceberg",
1273                HashMap::from_iter([(SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri2)]),
1274            )
1275            .await;
1276        assert!(catalog.is_err());
1277    }
1278
1279    /// Even when an invalid URI is specified in a builder method,
1280    /// it can be successfully overridden with a valid URI in props
1281    /// for catalog generation to succeed.
1282    #[tokio::test]
1283    async fn test_builder_props_set_valid_uri() {
1284        let sql_lite_uri = format!("sqlite:{}", temp_path());
1285        let sql_lite_uri2 = format!("sqlite:{}", temp_path());
1286        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1287        let warehouse_location = temp_path();
1288
1289        let catalog = SqlCatalogBuilder::default()
1290            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1291            .uri(sql_lite_uri2)
1292            .warehouse_location(warehouse_location)
1293            .load(
1294                "iceberg",
1295                HashMap::from_iter([(SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri.clone())]),
1296            )
1297            .await;
1298        assert!(catalog.is_ok());
1299    }
1300
1301    /// values assigned via props take precedence
1302    #[tokio::test]
1303    async fn test_builder_props_take_precedence() {
1304        let sql_lite_uri = format!("sqlite:{}", temp_path());
1305        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1306        let warehouse_location = temp_path();
1307        let warehouse_location2 = temp_path();
1308
1309        let catalog = SqlCatalogBuilder::default()
1310            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1311            .warehouse_location(warehouse_location2)
1312            .sql_bind_style(SqlBindStyle::DollarNumeric)
1313            .load(
1314                "iceberg",
1315                HashMap::from_iter([
1316                    (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri),
1317                    (
1318                        SQL_CATALOG_PROP_WAREHOUSE.to_string(),
1319                        warehouse_location.clone(),
1320                    ),
1321                    (
1322                        SQL_CATALOG_PROP_BIND_STYLE.to_string(),
1323                        SqlBindStyle::QMark.to_string(),
1324                    ),
1325                ]),
1326            )
1327            .await;
1328
1329        assert!(catalog.is_ok());
1330
1331        let catalog = catalog.unwrap();
1332        assert!(catalog.warehouse_location == warehouse_location);
1333        assert!(catalog.sql_bind_style == SqlBindStyle::QMark);
1334    }
1335
1336    /// values assigned via props take precedence
1337    #[tokio::test]
1338    async fn test_builder_props_take_precedence_props() {
1339        let sql_lite_uri = format!("sqlite:{}", temp_path());
1340        let sql_lite_uri2 = format!("sqlite:{}", temp_path());
1341        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1342        let warehouse_location = temp_path();
1343        let warehouse_location2 = temp_path();
1344
1345        let props = HashMap::from_iter([
1346            (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri.clone()),
1347            (
1348                SQL_CATALOG_PROP_WAREHOUSE.to_string(),
1349                warehouse_location.clone(),
1350            ),
1351            (
1352                SQL_CATALOG_PROP_BIND_STYLE.to_string(),
1353                SqlBindStyle::QMark.to_string(),
1354            ),
1355        ]);
1356        let props2 = HashMap::from_iter([
1357            (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri2.clone()),
1358            (
1359                SQL_CATALOG_PROP_WAREHOUSE.to_string(),
1360                warehouse_location2.clone(),
1361            ),
1362            (
1363                SQL_CATALOG_PROP_BIND_STYLE.to_string(),
1364                SqlBindStyle::DollarNumeric.to_string(),
1365            ),
1366        ]);
1367
1368        let catalog = SqlCatalogBuilder::default()
1369            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1370            .props(props2)
1371            .load("iceberg", props)
1372            .await;
1373
1374        assert!(catalog.is_ok());
1375
1376        let catalog = catalog.unwrap();
1377        assert!(catalog.warehouse_location == warehouse_location);
1378        assert!(catalog.sql_bind_style == SqlBindStyle::QMark);
1379    }
1380
1381    /// values assigned via props take precedence
1382    #[tokio::test]
1383    async fn test_builder_props_take_precedence_prop() {
1384        let sql_lite_uri = format!("sqlite:{}", temp_path());
1385        let sql_lite_uri2 = format!("sqlite:{}", temp_path());
1386        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1387        let warehouse_location = temp_path();
1388        let warehouse_location2 = temp_path();
1389
1390        let props = HashMap::from_iter([
1391            (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri.clone()),
1392            (
1393                SQL_CATALOG_PROP_WAREHOUSE.to_string(),
1394                warehouse_location.clone(),
1395            ),
1396            (
1397                SQL_CATALOG_PROP_BIND_STYLE.to_string(),
1398                SqlBindStyle::QMark.to_string(),
1399            ),
1400        ]);
1401
1402        let catalog = SqlCatalogBuilder::default()
1403            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1404            .prop(SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri2)
1405            .prop(SQL_CATALOG_PROP_WAREHOUSE.to_string(), warehouse_location2)
1406            .prop(
1407                SQL_CATALOG_PROP_BIND_STYLE.to_string(),
1408                SqlBindStyle::DollarNumeric.to_string(),
1409            )
1410            .load("iceberg", props)
1411            .await;
1412
1413        assert!(catalog.is_ok());
1414
1415        let catalog = catalog.unwrap();
1416        assert!(catalog.warehouse_location == warehouse_location);
1417        assert!(catalog.sql_bind_style == SqlBindStyle::QMark);
1418    }
1419
1420    /// invalid value for `SqlBindStyle` causes catalog creation to fail
1421    #[tokio::test]
1422    async fn test_builder_props_invalid_bind_style_fails() {
1423        let sql_lite_uri = format!("sqlite:{}", temp_path());
1424        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
1425        let warehouse_location = temp_path();
1426
1427        let catalog = SqlCatalogBuilder::default()
1428            .with_storage_factory(Arc::new(LocalFsStorageFactory))
1429            .load(
1430                "iceberg",
1431                HashMap::from_iter([
1432                    (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri),
1433                    (SQL_CATALOG_PROP_WAREHOUSE.to_string(), warehouse_location),
1434                    (SQL_CATALOG_PROP_BIND_STYLE.to_string(), "AAA".to_string()),
1435                ]),
1436            )
1437            .await;
1438
1439        assert!(catalog.is_err());
1440    }
1441
1442    #[tokio::test]
1443    async fn test_list_namespaces_returns_empty_vector() {
1444        let warehouse_loc = temp_path();
1445        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1446
1447        assert_eq!(catalog.list_namespaces(None).await.unwrap(), vec![]);
1448    }
1449
1450    #[tokio::test]
1451    async fn test_list_namespaces_returns_empty_different_name() {
1452        let warehouse_loc = temp_path();
1453        let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
1454        let namespace_ident_1 = NamespaceIdent::new("a".into());
1455        let namespace_ident_2 = NamespaceIdent::new("b".into());
1456        create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
1457        assert_eq!(
1458            to_set(catalog.list_namespaces(None).await.unwrap()),
1459            to_set(vec![namespace_ident_1, namespace_ident_2])
1460        );
1461
1462        let catalog2 = new_sql_catalog(warehouse_loc, Some("test")).await;
1463        assert_eq!(catalog2.list_namespaces(None).await.unwrap(), vec![]);
1464    }
1465
1466    #[tokio::test]
1467    async fn test_list_namespaces_returns_only_top_level_namespaces() {
1468        let warehouse_loc = temp_path();
1469        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1470        let namespace_ident_1 = NamespaceIdent::new("a".into());
1471        let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1472        let namespace_ident_3 = NamespaceIdent::new("b".into());
1473        create_namespaces(&catalog, &vec![
1474            &namespace_ident_1,
1475            &namespace_ident_2,
1476            &namespace_ident_3,
1477        ])
1478        .await;
1479
1480        assert_eq!(
1481            to_set(catalog.list_namespaces(None).await.unwrap()),
1482            to_set(vec![namespace_ident_1, namespace_ident_3])
1483        );
1484    }
1485
1486    #[tokio::test]
1487    async fn test_list_namespaces_returns_no_namespaces_under_parent() {
1488        let warehouse_loc = temp_path();
1489        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1490        let namespace_ident_1 = NamespaceIdent::new("a".into());
1491        let namespace_ident_2 = NamespaceIdent::new("b".into());
1492        create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
1493
1494        assert_eq!(
1495            catalog
1496                .list_namespaces(Some(&namespace_ident_1))
1497                .await
1498                .unwrap(),
1499            vec![]
1500        );
1501    }
1502
1503    #[tokio::test]
1504    async fn test_list_namespaces_returns_namespace_under_parent() {
1505        let warehouse_loc = temp_path();
1506        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1507        let namespace_ident_1 = NamespaceIdent::new("a".into());
1508        let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1509        let namespace_ident_3 = NamespaceIdent::new("c".into());
1510        create_namespaces(&catalog, &vec![
1511            &namespace_ident_1,
1512            &namespace_ident_2,
1513            &namespace_ident_3,
1514        ])
1515        .await;
1516
1517        assert_eq!(
1518            to_set(catalog.list_namespaces(None).await.unwrap()),
1519            to_set(vec![namespace_ident_1.clone(), namespace_ident_3])
1520        );
1521
1522        assert_eq!(
1523            catalog
1524                .list_namespaces(Some(&namespace_ident_1))
1525                .await
1526                .unwrap(),
1527            vec![NamespaceIdent::from_strs(vec!["a", "b"]).unwrap()]
1528        );
1529    }
1530
1531    #[tokio::test]
1532    async fn test_list_namespaces_returns_multiple_namespaces_under_parent() {
1533        let warehouse_loc = temp_path();
1534        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1535        let namespace_ident_1 = NamespaceIdent::new("a".to_string());
1536        let namespace_ident_2 = NamespaceIdent::from_strs(vec!["a", "a"]).unwrap();
1537        let namespace_ident_3 = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1538        let namespace_ident_4 = NamespaceIdent::from_strs(vec!["a", "c"]).unwrap();
1539        let namespace_ident_5 = NamespaceIdent::new("b".into());
1540        create_namespaces(&catalog, &vec![
1541            &namespace_ident_1,
1542            &namespace_ident_2,
1543            &namespace_ident_3,
1544            &namespace_ident_4,
1545            &namespace_ident_5,
1546        ])
1547        .await;
1548
1549        assert_eq!(
1550            to_set(
1551                catalog
1552                    .list_namespaces(Some(&namespace_ident_1))
1553                    .await
1554                    .unwrap()
1555            ),
1556            to_set(vec![
1557                NamespaceIdent::from_strs(vec!["a", "a"]).unwrap(),
1558                NamespaceIdent::from_strs(vec!["a", "b"]).unwrap(),
1559                NamespaceIdent::from_strs(vec!["a", "c"]).unwrap(),
1560            ])
1561        );
1562    }
1563
1564    #[tokio::test]
1565    async fn test_namespace_exists_returns_false() {
1566        let warehouse_loc = temp_path();
1567        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1568        let namespace_ident = NamespaceIdent::new("a".into());
1569        create_namespace(&catalog, &namespace_ident).await;
1570
1571        assert!(
1572            !catalog
1573                .namespace_exists(&NamespaceIdent::new("b".into()))
1574                .await
1575                .unwrap()
1576        );
1577    }
1578
1579    #[tokio::test]
1580    async fn test_namespace_exists_returns_true() {
1581        let warehouse_loc = temp_path();
1582        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1583        let namespace_ident = NamespaceIdent::new("a".into());
1584        create_namespace(&catalog, &namespace_ident).await;
1585
1586        assert!(catalog.namespace_exists(&namespace_ident).await.unwrap());
1587    }
1588
1589    #[tokio::test]
1590    async fn test_create_namespace_with_properties() {
1591        let warehouse_loc = temp_path();
1592        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1593        let namespace_ident = NamespaceIdent::new("abc".into());
1594
1595        let mut properties = default_properties();
1596        properties.insert("k".into(), "v".into());
1597
1598        assert_eq!(
1599            catalog
1600                .create_namespace(&namespace_ident, properties.clone())
1601                .await
1602                .unwrap(),
1603            Namespace::with_properties(namespace_ident.clone(), properties.clone())
1604        );
1605
1606        assert_eq!(
1607            catalog.get_namespace(&namespace_ident).await.unwrap(),
1608            Namespace::with_properties(namespace_ident, properties)
1609        );
1610    }
1611
1612    #[tokio::test]
1613    async fn test_create_nested_namespace() {
1614        let warehouse_loc = temp_path();
1615        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1616        let parent_namespace_ident = NamespaceIdent::new("a".into());
1617        create_namespace(&catalog, &parent_namespace_ident).await;
1618
1619        let child_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1620
1621        assert_eq!(
1622            catalog
1623                .create_namespace(&child_namespace_ident, HashMap::new())
1624                .await
1625                .unwrap(),
1626            Namespace::new(child_namespace_ident.clone())
1627        );
1628
1629        assert_eq!(
1630            catalog.get_namespace(&child_namespace_ident).await.unwrap(),
1631            Namespace::with_properties(child_namespace_ident, default_properties())
1632        );
1633    }
1634
1635    #[tokio::test]
1636    async fn test_create_deeply_nested_namespace() {
1637        let warehouse_loc = temp_path();
1638        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1639        let namespace_ident_a = NamespaceIdent::new("a".into());
1640        let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1641        create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
1642
1643        let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
1644
1645        assert_eq!(
1646            catalog
1647                .create_namespace(&namespace_ident_a_b_c, HashMap::new())
1648                .await
1649                .unwrap(),
1650            Namespace::new(namespace_ident_a_b_c.clone())
1651        );
1652
1653        assert_eq!(
1654            catalog.get_namespace(&namespace_ident_a_b_c).await.unwrap(),
1655            Namespace::with_properties(namespace_ident_a_b_c, default_properties())
1656        );
1657    }
1658
1659    #[tokio::test]
1660    async fn test_update_namespace_noop() {
1661        let warehouse_loc = temp_path();
1662        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1663        let namespace_ident = NamespaceIdent::new("a".into());
1664        create_namespace(&catalog, &namespace_ident).await;
1665
1666        catalog
1667            .update_namespace(&namespace_ident, HashMap::new())
1668            .await
1669            .unwrap();
1670
1671        assert_eq!(
1672            *catalog
1673                .get_namespace(&namespace_ident)
1674                .await
1675                .unwrap()
1676                .properties(),
1677            HashMap::from_iter([("exists".to_string(), "true".to_string())])
1678        )
1679    }
1680
1681    #[tokio::test]
1682    async fn test_update_nested_namespace() {
1683        let warehouse_loc = temp_path();
1684        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1685        let namespace_ident = NamespaceIdent::from_strs(["a", "b"]).unwrap();
1686        create_namespace(&catalog, &namespace_ident).await;
1687
1688        let mut props = HashMap::from_iter([
1689            ("prop1".to_string(), "val1".to_string()),
1690            ("prop2".into(), "val2".into()),
1691        ]);
1692
1693        catalog
1694            .update_namespace(&namespace_ident, props.clone())
1695            .await
1696            .unwrap();
1697
1698        props.insert("exists".into(), "true".into());
1699
1700        assert_eq!(
1701            *catalog
1702                .get_namespace(&namespace_ident)
1703                .await
1704                .unwrap()
1705                .properties(),
1706            props
1707        )
1708    }
1709
1710    #[tokio::test]
1711    async fn test_update_namespace_errors_if_nested_namespace_doesnt_exist() {
1712        let warehouse_loc = temp_path();
1713        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1714        let namespace_ident = NamespaceIdent::from_strs(["a", "b"]).unwrap();
1715
1716        let props = HashMap::from_iter([
1717            ("prop1".to_string(), "val1".to_string()),
1718            ("prop2".into(), "val2".into()),
1719        ]);
1720
1721        let err = catalog
1722            .update_namespace(&namespace_ident, props)
1723            .await
1724            .unwrap_err();
1725
1726        assert_eq!(
1727            err.message(),
1728            format!("No such namespace: {namespace_ident:?}")
1729        );
1730    }
1731
1732    #[tokio::test]
1733    async fn test_drop_nested_namespace() {
1734        let warehouse_loc = temp_path();
1735        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1736        let namespace_ident_a = NamespaceIdent::new("a".into());
1737        let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1738        create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
1739
1740        catalog.drop_namespace(&namespace_ident_a_b).await.unwrap();
1741
1742        assert!(
1743            !catalog
1744                .namespace_exists(&namespace_ident_a_b)
1745                .await
1746                .unwrap()
1747        );
1748
1749        assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
1750    }
1751
1752    #[tokio::test]
1753    async fn test_drop_deeply_nested_namespace() {
1754        let warehouse_loc = temp_path();
1755        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1756        let namespace_ident_a = NamespaceIdent::new("a".into());
1757        let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1758        let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
1759        create_namespaces(&catalog, &vec![
1760            &namespace_ident_a,
1761            &namespace_ident_a_b,
1762            &namespace_ident_a_b_c,
1763        ])
1764        .await;
1765
1766        catalog
1767            .drop_namespace(&namespace_ident_a_b_c)
1768            .await
1769            .unwrap();
1770
1771        assert!(
1772            !catalog
1773                .namespace_exists(&namespace_ident_a_b_c)
1774                .await
1775                .unwrap()
1776        );
1777
1778        assert!(
1779            catalog
1780                .namespace_exists(&namespace_ident_a_b)
1781                .await
1782                .unwrap()
1783        );
1784
1785        assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
1786    }
1787
1788    #[tokio::test]
1789    async fn test_drop_namespace_throws_error_if_nested_namespace_doesnt_exist() {
1790        let warehouse_loc = temp_path();
1791        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1792        create_namespace(&catalog, &NamespaceIdent::new("a".into())).await;
1793
1794        let non_existent_namespace_ident =
1795            NamespaceIdent::from_vec(vec!["a".into(), "b".into()]).unwrap();
1796        assert_eq!(
1797            catalog
1798                .drop_namespace(&non_existent_namespace_ident)
1799                .await
1800                .unwrap_err()
1801                .to_string(),
1802            format!("NamespaceNotFound => No such namespace: {non_existent_namespace_ident:?}")
1803        )
1804    }
1805
1806    #[tokio::test]
1807    async fn test_dropping_a_namespace_does_not_drop_namespaces_nested_under_that_one() {
1808        let warehouse_loc = temp_path();
1809        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1810        let namespace_ident_a = NamespaceIdent::new("a".into());
1811        let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1812        create_namespaces(&catalog, &vec![&namespace_ident_a, &namespace_ident_a_b]).await;
1813
1814        catalog.drop_namespace(&namespace_ident_a).await.unwrap();
1815
1816        assert!(!catalog.namespace_exists(&namespace_ident_a).await.unwrap());
1817
1818        assert!(
1819            catalog
1820                .namespace_exists(&namespace_ident_a_b)
1821                .await
1822                .unwrap()
1823        );
1824    }
1825
1826    #[tokio::test]
1827    async fn test_create_table_falls_back_to_namespace_location_if_table_location_is_missing() {
1828        let warehouse_loc = temp_path();
1829        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1830
1831        let namespace_ident = NamespaceIdent::new("a".into());
1832        let mut namespace_properties = HashMap::new();
1833        let namespace_location = temp_path();
1834        namespace_properties.insert(
1835            NAMESPACE_LOCATION_PROPERTY_KEY.to_string(),
1836            namespace_location.to_string(),
1837        );
1838        catalog
1839            .create_namespace(&namespace_ident, namespace_properties)
1840            .await
1841            .unwrap();
1842
1843        let table_name = "tbl1";
1844        let expected_table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
1845        let expected_table_metadata_location_regex =
1846            format!("^{namespace_location}/tbl1/metadata/00000-{UUID_REGEX_STR}.metadata.json$",);
1847
1848        let table = catalog
1849            .create_table(
1850                &namespace_ident,
1851                TableCreation::builder()
1852                    .name(table_name.into())
1853                    .schema(simple_table_schema())
1854                    // no location specified for table
1855                    .build(),
1856            )
1857            .await
1858            .unwrap();
1859        assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
1860        assert_table_metadata_location_matches(&table, &expected_table_metadata_location_regex);
1861
1862        let table = catalog.load_table(&expected_table_ident).await.unwrap();
1863        assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
1864        assert_table_metadata_location_matches(&table, &expected_table_metadata_location_regex);
1865    }
1866
1867    #[tokio::test]
1868    async fn test_create_table_in_nested_namespace_falls_back_to_nested_namespace_location_if_table_location_is_missing()
1869     {
1870        let warehouse_loc = temp_path();
1871        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
1872
1873        let namespace_ident = NamespaceIdent::new("a".into());
1874        let mut namespace_properties = HashMap::new();
1875        let namespace_location = temp_path();
1876        namespace_properties.insert(
1877            NAMESPACE_LOCATION_PROPERTY_KEY.to_string(),
1878            namespace_location.to_string(),
1879        );
1880        catalog
1881            .create_namespace(&namespace_ident, namespace_properties)
1882            .await
1883            .unwrap();
1884
1885        let nested_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1886        let mut nested_namespace_properties = HashMap::new();
1887        let nested_namespace_location = temp_path();
1888        nested_namespace_properties.insert(
1889            NAMESPACE_LOCATION_PROPERTY_KEY.to_string(),
1890            nested_namespace_location.to_string(),
1891        );
1892        catalog
1893            .create_namespace(&nested_namespace_ident, nested_namespace_properties)
1894            .await
1895            .unwrap();
1896
1897        let table_name = "tbl1";
1898        let expected_table_ident =
1899            TableIdent::new(nested_namespace_ident.clone(), table_name.into());
1900        let expected_table_metadata_location_regex = format!(
1901            "^{nested_namespace_location}/tbl1/metadata/00000-{UUID_REGEX_STR}.metadata.json$",
1902        );
1903
1904        let table = catalog
1905            .create_table(
1906                &nested_namespace_ident,
1907                TableCreation::builder()
1908                    .name(table_name.into())
1909                    .schema(simple_table_schema())
1910                    // no location specified for table
1911                    .build(),
1912            )
1913            .await
1914            .unwrap();
1915        assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
1916        assert_table_metadata_location_matches(&table, &expected_table_metadata_location_regex);
1917
1918        let table = catalog.load_table(&expected_table_ident).await.unwrap();
1919        assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
1920        assert_table_metadata_location_matches(&table, &expected_table_metadata_location_regex);
1921    }
1922
1923    #[tokio::test]
1924    async fn test_create_table_falls_back_to_warehouse_location_if_both_table_location_and_namespace_location_are_missing()
1925     {
1926        let warehouse_loc = temp_path();
1927        let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
1928
1929        let namespace_ident = NamespaceIdent::new("a".into());
1930        // note: no location specified in namespace_properties
1931        let namespace_properties = HashMap::new();
1932        catalog
1933            .create_namespace(&namespace_ident, namespace_properties)
1934            .await
1935            .unwrap();
1936
1937        let table_name = "tbl1";
1938        let expected_table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
1939        let expected_table_metadata_location_regex =
1940            format!("^{warehouse_loc}/a/tbl1/metadata/00000-{UUID_REGEX_STR}.metadata.json$");
1941
1942        let table = catalog
1943            .create_table(
1944                &namespace_ident,
1945                TableCreation::builder()
1946                    .name(table_name.into())
1947                    .schema(simple_table_schema())
1948                    // no location specified for table
1949                    .build(),
1950            )
1951            .await
1952            .unwrap();
1953        assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
1954        assert_table_metadata_location_matches(&table, &expected_table_metadata_location_regex);
1955
1956        let table = catalog.load_table(&expected_table_ident).await.unwrap();
1957        assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
1958        assert_table_metadata_location_matches(&table, &expected_table_metadata_location_regex);
1959    }
1960
1961    #[tokio::test]
1962    async fn test_create_table_in_nested_namespace_falls_back_to_warehouse_location_if_both_table_location_and_namespace_location_are_missing()
1963     {
1964        let warehouse_loc = temp_path();
1965        let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
1966
1967        let namespace_ident = NamespaceIdent::new("a".into());
1968        create_namespace(&catalog, &namespace_ident).await;
1969
1970        let nested_namespace_ident = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
1971        create_namespace(&catalog, &nested_namespace_ident).await;
1972
1973        let table_name = "tbl1";
1974        let expected_table_ident =
1975            TableIdent::new(nested_namespace_ident.clone(), table_name.into());
1976        let expected_table_metadata_location_regex =
1977            format!("^{warehouse_loc}/a/b/tbl1/metadata/00000-{UUID_REGEX_STR}.metadata.json$");
1978
1979        let table = catalog
1980            .create_table(
1981                &nested_namespace_ident,
1982                TableCreation::builder()
1983                    .name(table_name.into())
1984                    .schema(simple_table_schema())
1985                    // no location specified for table
1986                    .build(),
1987            )
1988            .await
1989            .unwrap();
1990        assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
1991        assert_table_metadata_location_matches(&table, &expected_table_metadata_location_regex);
1992
1993        let table = catalog.load_table(&expected_table_ident).await.unwrap();
1994        assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
1995        assert_table_metadata_location_matches(&table, &expected_table_metadata_location_regex);
1996    }
1997
1998    #[tokio::test]
1999    async fn test_create_table_throws_error_if_table_with_same_name_already_exists() {
2000        let warehouse_loc = temp_path();
2001        let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
2002        let namespace_ident = NamespaceIdent::new("a".into());
2003        create_namespace(&catalog, &namespace_ident).await;
2004        let table_name = "tbl1";
2005        let table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
2006        create_table(&catalog, &table_ident).await;
2007
2008        let tmp_dir = TempDir::new().unwrap();
2009        let location = tmp_dir.path().to_str().unwrap().to_string();
2010
2011        assert_eq!(
2012            catalog
2013                .create_table(
2014                    &namespace_ident,
2015                    TableCreation::builder()
2016                        .name(table_name.into())
2017                        .schema(simple_table_schema())
2018                        .location(location)
2019                        .build()
2020                )
2021                .await
2022                .unwrap_err()
2023                .to_string(),
2024            format!(
2025                "TableAlreadyExists => Table {:?} already exists.",
2026                &table_ident
2027            )
2028        );
2029    }
2030
2031    #[tokio::test]
2032    async fn test_rename_table_src_table_is_same_as_dst_table() {
2033        let warehouse_loc = temp_path();
2034        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
2035        let namespace_ident = NamespaceIdent::new("n1".into());
2036        create_namespace(&catalog, &namespace_ident).await;
2037        let table_ident = TableIdent::new(namespace_ident.clone(), "tbl".into());
2038        create_table(&catalog, &table_ident).await;
2039
2040        catalog
2041            .rename_table(&table_ident, &table_ident)
2042            .await
2043            .unwrap();
2044
2045        assert_eq!(catalog.list_tables(&namespace_ident).await.unwrap(), vec![
2046            table_ident
2047        ],);
2048    }
2049
2050    #[tokio::test]
2051    async fn test_rename_table_across_nested_namespaces() {
2052        let warehouse_loc = temp_path();
2053        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
2054        let namespace_ident_a = NamespaceIdent::new("a".into());
2055        let namespace_ident_a_b = NamespaceIdent::from_strs(vec!["a", "b"]).unwrap();
2056        let namespace_ident_a_b_c = NamespaceIdent::from_strs(vec!["a", "b", "c"]).unwrap();
2057        create_namespaces(&catalog, &vec![
2058            &namespace_ident_a,
2059            &namespace_ident_a_b,
2060            &namespace_ident_a_b_c,
2061        ])
2062        .await;
2063
2064        let src_table_ident = TableIdent::new(namespace_ident_a_b_c.clone(), "tbl1".into());
2065        create_tables(&catalog, vec![&src_table_ident]).await;
2066
2067        let dst_table_ident = TableIdent::new(namespace_ident_a_b.clone(), "tbl1".into());
2068        catalog
2069            .rename_table(&src_table_ident, &dst_table_ident)
2070            .await
2071            .unwrap();
2072
2073        assert!(!catalog.table_exists(&src_table_ident).await.unwrap());
2074
2075        assert!(catalog.table_exists(&dst_table_ident).await.unwrap());
2076    }
2077
2078    #[tokio::test]
2079    async fn test_rename_table_throws_error_if_dst_namespace_doesnt_exist() {
2080        let warehouse_loc = temp_path();
2081        let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
2082        let src_namespace_ident = NamespaceIdent::new("n1".into());
2083        let src_table_ident = TableIdent::new(src_namespace_ident.clone(), "tbl1".into());
2084        create_namespace(&catalog, &src_namespace_ident).await;
2085        create_table(&catalog, &src_table_ident).await;
2086
2087        let non_existent_dst_namespace_ident = NamespaceIdent::new("n2".into());
2088        let dst_table_ident =
2089            TableIdent::new(non_existent_dst_namespace_ident.clone(), "tbl1".into());
2090        assert_eq!(
2091            catalog
2092                .rename_table(&src_table_ident, &dst_table_ident)
2093                .await
2094                .unwrap_err()
2095                .to_string(),
2096            format!("NamespaceNotFound => No such namespace: {non_existent_dst_namespace_ident:?}"),
2097        );
2098    }
2099}