Remove iceberg call from create table statement (#21)

* Remove iceberg call from create table

* Remove invalidate_cache

Author: osipovartem

crates/catalog/src/catalog_list.rs

@@ -14,7 +14,6 @@ use aws_credential_types::provider::SharedCredentialsProvider;
 use catalog_metastore::{
     AwsCredentials, Database, Metastore, RwObject, S3TablesVolume, VolumeType,
 };
-use catalog_metastore::{SchemaIdent, TableIdent};
 use dashmap::DashMap;
 use datafusion::{
     catalog::{CatalogProvider, CatalogProviderList},
@@ -37,22 +36,6 @@ use url::Url;
 
 pub const DEFAULT_CATALOG: &str = "embucket";
 
-#[derive(Debug, Eq, PartialEq)]
-pub enum CachedEntity {
-    Schema(SchemaIdent),
-    Table(TableIdent),
-}
-
-impl CachedEntity {
-    #[must_use]
-    pub fn normalized(&self) -> Self {
-        match self {
-            Self::Schema(ident) => Self::Schema(ident.normalized()),
-            Self::Table(ident) => Self::Table(ident.normalized()),
-        }
-    }
-}
-
 pub struct EmbucketCatalogList {
     pub metastore: Arc<dyn Metastore>,
     pub table_object_store: Arc<DashMap<String, Arc<dyn ObjectStore>>>,
@@ -262,109 +245,6 @@ impl EmbucketCatalogList {
         )
     }
 
-    /// Do not keep returned references to avoid deadlocks
-    fn catalog_ref_by_name(
-        &self,
-        name: &str,
-    ) -> Result<dashmap::mapref::one::Ref<'_, String, Arc<CachingCatalog>>> {
-        self.catalogs.get(name).ok_or_else(|| {
-            InvalidCacheSnafu {
-                entity: "catalog",
-                name,
-            }
-            .build()
-        })
-    }
-
-    /// Invalidates the cache for a specific catalog entity (schema or table).
-    ///
-    /// This method ensures that the cache for the specified entity is refreshed or cleared as appropriate.
-    /// - For a schema: If the schema exists in the underlying catalog, it is (re-)cached; if it does not exist, the cache entry is removed.
-    /// - For a table: The table cache is invalidated. If the table exists in the underlying schema, it is (re-)cached; otherwise, the cache entry is removed.
-    ///
-    /// # Arguments
-    /// * `entity` - The cached entity to invalidate, which can be either a schema or a table.
-    ///
-    /// # Errors
-    /// Returns an error if:
-    /// - The specified catalog or schema does not exist in the cache.
-    /// - There is a failure when accessing the underlying catalog or schema provider.
-    #[allow(clippy::as_conversions, clippy::too_many_lines)]
-    #[tracing::instrument(
-        name = "EmbucketCatalogList::refresh_schema",
-        level = "debug",
-        skip(self),
-        err
-    )]
-    pub async fn invalidate_cache(&self, entity: CachedEntity) -> Result<()> {
-        match entity {
-            CachedEntity::Schema(schema_ident) => {
-                let SchemaIdent { schema, database } = schema_ident;
-                let catalog_ref = self.catalog_ref_by_name(&database)?;
-                if !catalog_ref.should_refresh {
-                    return Ok(());
-                }
-                if let Some(schema_provider) = catalog_ref.catalog.schema(&schema) {
-                    // schema exists -> ensure it's cached
-                    if catalog_ref.schemas_cache.get(&schema).is_none() {
-                        let schema = CachingSchema {
-                            schema: schema_provider,
-                            tables_cache: DashMap::default(),
-                            name: schema.clone(),
-                            iceberg_catalog: catalog_ref.iceberg_catalog.clone(),
-                        };
-                        catalog_ref
-                            .schemas_cache
-                            .insert(schema.name.clone(), Arc::new(schema));
-                    }
-                } else {
-                    // no schema exists -> ensure cache is empty
-                    catalog_ref.schemas_cache.remove(&schema);
-                }
-            }
-            CachedEntity::Table(table_ident) => {
-                let TableIdent {
-                    database,
-                    schema,
-                    table,
-                } = table_ident;
-                let catalog_ref = self.catalog_ref_by_name(&database)?;
-                if !catalog_ref.should_refresh {
-                    return Ok(());
-                }
-                let schema_ref = catalog_ref.schemas_cache.get(&schema).ok_or_else(|| {
-                    InvalidCacheSnafu {
-                        entity: "schema",
-                        name: format!("{database}.{schema}"),
-                    }
-                    .build()
-                })?;
-
-                // invalidate table cache if table exists, noop if doesn't
-                schema_ref.tables_cache.remove(&table);
-
-                if let Some(table_provider) = schema_ref
-                    .schema
-                    .table(&table)
-                    .await
-                    .context(catalog_error::DataFusionSnafu)?
-                {
-                    // ensure table is cached
-                    schema_ref.tables_cache.insert(
-                        table.clone(),
-                        Arc::new(CachingTable::new_with_schema(
-                            table.clone(),
-                            table_provider.schema(),
-                            Arc::clone(&table_provider),
-                        )),
-                    );
-                }
-            }
-        }
-
-        Ok(())
-    }
-
     #[allow(clippy::as_conversions, clippy::too_many_lines)]
     #[tracing::instrument(
         name = "EmbucketCatalogList::refresh",

crates/catalog/src/schema.rs

@@ -1,13 +1,16 @@
 use crate::df_error;
-use crate::table::CachingTable;
+use crate::table::{CachingTable, IcebergTableBuilder};
 use async_trait::async_trait;
 use dashmap::DashMap;
 use datafusion::catalog::{SchemaProvider, TableProvider};
 use datafusion_common::DataFusionError;
 use datafusion_expr::TableType;
+use datafusion_iceberg::DataFusionTable;
 use futures::executor::block_on;
 use iceberg_rust::catalog::Catalog;
+use iceberg_rust::catalog::tabular::Tabular as IcebergTabular;
 use iceberg_rust_spec::identifier::Identifier;
+use snafu::ResultExt;
 use snafu::futures::TryFutureExt;
 use std::any::Any;
 use std::sync::Arc;
@@ -81,12 +84,29 @@ impl SchemaProvider for CachingSchema {
         name: String,
         table: Arc<dyn TableProvider>,
     ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
-        let caching_table = Arc::new(CachingTable::new(name.clone(), Arc::clone(&table)));
-        self.tables_cache.insert(name.clone(), caching_table);
-        if table.table_type() != TableType::View {
-            return self.schema.register_table(name, table);
-        }
-        Ok(Some(table))
+        let table_provider: Arc<dyn TableProvider> = if let Some(catalog) = &self.iceberg_catalog
+            && let Some(iceberg_builder) = table.as_any().downcast_ref::<IcebergTableBuilder>()
+            && table.table_type() != TableType::View
+        {
+            let ident = Identifier::new(std::slice::from_ref(&self.name), &name);
+            block_on(async move {
+                let mut builder = iceberg_builder.builder.clone();
+                let iceberg_table = builder
+                    .build(ident.namespace(), catalog.clone())
+                    .await
+                    .context(df_error::IcebergSnafu)?;
+                let tabular = IcebergTabular::Table(iceberg_table);
+                let table_provider: Arc<dyn TableProvider> =
+                    Arc::new(DataFusionTable::new(tabular, None, None, None));
+                Ok::<Arc<dyn TableProvider>, DataFusionError>(table_provider)
+            })?
+        } else {
+            table
+        };
+
+        let caching_table = Arc::new(CachingTable::new(name.clone(), Arc::clone(&table_provider)));
+        self.tables_cache.insert(name, caching_table);
+        Ok(Some(table_provider))
     }
 
     #[allow(clippy::as_conversions)]

crates/catalog/src/table.rs

@@ -1,18 +1,20 @@
 use crate::df_error;
 use async_trait::async_trait;
-use datafusion::arrow::datatypes::SchemaRef;
+use datafusion::arrow::datatypes::{Schema, SchemaRef};
 use datafusion::catalog::{Session, TableProvider};
 use datafusion::datasource::{ViewTable, provider_as_source};
 use datafusion::execution::SessionState;
-use datafusion_common::Statistics;
 use datafusion_common::tree_node::{Transformed, TreeNode};
+use datafusion_common::{Statistics, plan_err};
 use datafusion_expr::dml::InsertOp;
 use datafusion_expr::{Expr, LogicalPlan, TableProviderFilterPushDown, TableScan, TableType};
 use datafusion_physical_plan::ExecutionPlan;
+use iceberg_rust::catalog::create::CreateTableBuilder;
 use once_cell::sync::OnceCell;
 use snafu::OptionExt;
 use std::any::Any;
 use std::collections::HashMap;
+use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
 pub struct CachingTable {
@@ -38,8 +40,8 @@ impl CachingTable {
     }
 }
 
-impl std::fmt::Debug for CachingTable {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+impl Debug for CachingTable {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Table")
             .field("schema", &"")
             .field("name", &self.name)
@@ -176,3 +178,47 @@ async fn rewrite_view_source(
         .data;
     Ok(new_plan)
 }
+
+pub struct IcebergTableBuilder {
+    pub builder: CreateTableBuilder,
+}
+
+impl IcebergTableBuilder {
+    #[must_use]
+    pub const fn new(builder: CreateTableBuilder) -> Self {
+        Self { builder }
+    }
+}
+
+impl Debug for IcebergTableBuilder {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("IcebergTableBuilder")
+            .field("builder", &"")
+            .finish()
+    }
+}
+
+#[async_trait]
+impl TableProvider for IcebergTableBuilder {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn schema(&self) -> SchemaRef {
+        SchemaRef::from(Schema::empty())
+    }
+
+    fn table_type(&self) -> TableType {
+        TableType::Base
+    }
+
+    async fn scan(
+        &self,
+        _state: &dyn Session,
+        _projection: Option<&Vec<usize>>,
+        _filters: &[Expr],
+        _limit: Option<usize>,
+    ) -> datafusion_common::Result<Arc<dyn ExecutionPlan>> {
+        plan_err!("Iceberg table builder cannot be scanned")
+    }
+}