fallback to empty mirror for sync datafusion cataloglist

Author: JanKaul

datafusion-iceberg-sql/src/context.rs

@@ -28,7 +28,6 @@ impl IcebergContext {
         for (catalog_name, namespace, name) in tables {
             let catalog = catalogs
                 .catalog(catalog_name)
-                .await
                 .ok_or(DataFusionError::Internal(format!(
                     "Catalog {} was not provided",
                     &catalog_name

datafusion_iceberg/examples/refresh_materialized_view.rs

@@ -26,7 +26,7 @@ pub(crate) async fn main() {
             .unwrap(),
     );
 
-    let catalog = catalog_list.catalog("iceberg").await.unwrap();
+    let catalog = catalog_list.catalog("iceberg").unwrap();
 
     let schema = Schema::builder()
         .with_fields(

datafusion_iceberg/src/catalog/catalog.rs

@@ -19,6 +19,12 @@ impl IcebergCatalog {
         })
     }
 
+    pub fn new_sync(catalog: Arc<dyn Catalog>, branch: Option<&str>) -> Self {
+        IcebergCatalog {
+            catalog: Arc::new(Mirror::new_sync(catalog, branch.map(ToOwned::to_owned))),
+        }
+    }
+
     pub fn catalog(&self) -> Arc<dyn Catalog> {
         self.catalog.catalog()
     }

datafusion_iceberg/src/catalog/catalog_list.rs

@@ -9,7 +9,10 @@ use crate::error::Error;
 
 use super::catalog::IcebergCatalog;
 
-pub struct IcebergCatalogList(DashMap<String, Arc<dyn CatalogProvider>>);
+pub struct IcebergCatalogList {
+    catalogs: DashMap<String, Arc<dyn CatalogProvider>>,
+    catalog_list: Arc<dyn CatalogList>,
+}
 
 impl IcebergCatalogList {
     pub async fn new(catalog_list: Arc<dyn CatalogList>) -> Result<Self, Error> {
@@ -19,7 +22,7 @@ impl IcebergCatalogList {
             .then(|x| {
                 let catalog_list = catalog_list.clone();
                 async move {
-                    let catalog = catalog_list.catalog(&x).await?;
+                    let catalog = catalog_list.catalog(&x)?;
                     Some((
                         x,
                         Arc::new(IcebergCatalog::new(catalog, None).await.ok()?)
@@ -31,7 +34,10 @@ impl IcebergCatalogList {
             .collect::<DashMap<_, _>>()
             .await;
 
-        Ok(IcebergCatalogList(map))
+        Ok(IcebergCatalogList {
+            catalogs: map,
+            catalog_list,
+        })
     }
 }
 
@@ -41,18 +47,22 @@ impl CatalogProviderList for IcebergCatalogList {
     }
 
     fn catalog(&self, name: &str) -> Option<Arc<dyn CatalogProvider>> {
-        self.0.get(name).as_deref().cloned()
+        self.catalogs.get(name).as_deref().cloned().or_else(|| {
+            self.catalog_list.catalog(name).map(|catalog| {
+                Arc::new(IcebergCatalog::new_sync(catalog, None)) as Arc<dyn CatalogProvider>
+            })
+        })
     }
 
     fn catalog_names(&self) -> Vec<String> {
-        self.0.iter().map(|c| c.key().clone()).collect()
+        self.catalogs.iter().map(|c| c.key().clone()).collect()
     }
 
     fn register_catalog(
         &self,
         name: String,
         catalog: Arc<dyn CatalogProvider>,
     ) -> Option<Arc<dyn CatalogProvider>> {
-        self.0.insert(name, catalog)
+        self.catalogs.insert(name, catalog)
     }
 }

datafusion_iceberg/src/catalog/mirror.rs

@@ -1,7 +1,6 @@
 use dashmap::DashMap;
 use datafusion::{datasource::TableProvider, error::DataFusionError};
 use futures::{executor::LocalPool, task::LocalSpawnExt};
-use std::ops::Deref;
 use std::{collections::HashSet, sync::Arc};
 
 use iceberg_rust::spec::{tabular::TabularMetadata, view_metadata::REF_PREFIX};
@@ -66,6 +65,15 @@ impl Mirror {
             branch,
         })
     }
+    pub fn new_sync(catalog: Arc<dyn Catalog>, branch: Option<String>) -> Self {
+        let storage = DashMap::new();
+
+        Mirror {
+            storage,
+            catalog,
+            branch,
+        }
+    }
     /// Lists all tables in the given namespace.
     pub fn table_names(&self, namespace: &Namespace) -> Result<Vec<Identifier>, DataFusionError> {
         let node = self

datafusion_iceberg/src/materialized_view.rs

@@ -62,13 +62,10 @@ pub async fn refresh_materialized_view(
             async move {
                 let catalog_name = full_identifier.catalog().to_string();
                 let identifier: Identifier = full_identifier.into();
-                let catalog = catalog_list
-                    .catalog(&catalog_name)
-                    .await
-                    .ok_or(Error::NotFound(
-                        "Catalog".to_owned(),
-                        catalog_name.to_owned(),
-                    ))?;
+                let catalog = catalog_list.catalog(&catalog_name).ok_or(Error::NotFound(
+                    "Catalog".to_owned(),
+                    catalog_name.to_owned(),
+                ))?;
 
                 let tabular = match catalog.load_tabular(&identifier).await? {
                     Tabular::View(_) => {
@@ -243,7 +240,7 @@ mod tests {
                 .unwrap(),
         );
 
-        let catalog = catalog_list.catalog("iceberg").await.unwrap();
+        let catalog = catalog_list.catalog("iceberg").unwrap();
 
         let schema = Schema::builder()
             .with_schema_id(0)

iceberg-rest-catalog/src/catalog.rs

@@ -497,7 +497,7 @@ impl RestCatalogList {
 
 #[async_trait]
 impl CatalogList for RestCatalogList {
-    async fn catalog(&self, name: &str) -> Option<Arc<dyn Catalog>> {
+    fn catalog(&self, name: &str) -> Option<Arc<dyn Catalog>> {
         Some(Arc::new(RestCatalog::new(
             Some(name),
             self.configuration.clone(),

iceberg-rust/src/catalog/mod.rs

@@ -109,7 +109,7 @@ pub trait Catalog: Send + Sync + Debug {
 #[async_trait::async_trait]
 pub trait CatalogList: Send + Sync + Debug {
     /// Get catalog from list by name
-    async fn catalog(&self, name: &str) -> Option<Arc<dyn Catalog>>;
+    fn catalog(&self, name: &str) -> Option<Arc<dyn Catalog>>;
     /// Get the list of available catalogs
     async fn list_catalogs(&self) -> Vec<String>;
 }

iceberg-sql-catalog/src/lib.rs

@@ -711,7 +711,7 @@ impl SqlCatalogList {
 
 #[async_trait]
 impl CatalogList for SqlCatalogList {
-    async fn catalog(&self, name: &str) -> Option<Arc<dyn Catalog>> {
+    fn catalog(&self, name: &str) -> Option<Arc<dyn Catalog>> {
         Some(Arc::new(SqlCatalog {
             name: name.to_owned(),
             pool: self.pool.clone(),