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

iceberg_catalog_sql/
catalog.rs
