feat(datafusion): Split IcebergTableProvider into static and non-static table provider

bindings/python/src/datafusion_table_provider.rs

@@ -23,7 +23,7 @@ use datafusion_ffi::table_provider::FFI_TableProvider;
 use iceberg::TableIdent;
 use iceberg::io::FileIO;
 use iceberg::table::StaticTable;
-use iceberg_datafusion::table::IcebergTableProvider;
+use iceberg_datafusion::table::IcebergStaticTableProvider;
 use pyo3::exceptions::PyRuntimeError;
 use pyo3::prelude::*;
 use pyo3::types::PyCapsule;
@@ -32,7 +32,7 @@ use crate::runtime::runtime;
 
 #[pyclass(name = "IcebergDataFusionTable")]
 pub struct PyIcebergDataFusionTable {
-    inner: Arc<IcebergTableProvider>,
+    inner: Arc<IcebergStaticTableProvider>,
 }
 
 #[pymethods]
@@ -69,7 +69,7 @@ impl PyIcebergDataFusionTable {
 
             let table = static_table.into_table();
 
-            IcebergTableProvider::try_new_from_table(table)
+            IcebergStaticTableProvider::try_new_from_table(table)
                 .await
                 .map_err(|e| {
                     PyRuntimeError::new_err(format!("Failed to create table provider: {e}"))

crates/integration_tests/tests/shared_tests/datafusion.rs

@@ -26,7 +26,7 @@ use datafusion::error::DataFusionError;
 use datafusion::prelude::SessionContext;
 use iceberg::{Catalog, CatalogBuilder, TableIdent};
 use iceberg_catalog_rest::RestCatalogBuilder;
-use iceberg_datafusion::IcebergTableProvider;
+use iceberg_datafusion::IcebergStaticTableProvider;
 use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
 
 use crate::get_shared_containers;
@@ -47,7 +47,7 @@ async fn test_basic_queries() -> Result<(), DataFusionError> {
     let ctx = SessionContext::new();
 
     let table_provider = Arc::new(
-        IcebergTableProvider::try_new_from_table(table)
+        IcebergStaticTableProvider::try_new_from_table(table)
             .await
             .unwrap(),
     );

crates/integrations/datafusion/src/schema.rs

@@ -28,6 +28,7 @@ use iceberg::inspect::MetadataTableType;
 use iceberg::{Catalog, NamespaceIdent, Result};
 
 use crate::table::IcebergTableProvider;
+use crate::to_datafusion_error;
 
 /// Represents a [`SchemaProvider`] for the Iceberg [`Catalog`], managing
 /// access to table providers within a specific namespace.
@@ -113,7 +114,10 @@ impl SchemaProvider for IcebergSchemaProvider {
             let metadata_table_type =
                 MetadataTableType::try_from(metadata_table_name).map_err(DataFusionError::Plan)?;
             if let Some(table) = self.tables.get(table_name) {
-                let metadata_table = table.metadata_table(metadata_table_type);
+                let metadata_table = table
+                    .metadata_table(metadata_table_type)
+                    .await
+                    .map_err(to_datafusion_error)?;
                 return Ok(Some(Arc::new(metadata_table)));
             } else {
                 return Ok(None);