refactor: delegate catalog scans to static table provider

crates/datafusion/src/table_provider/catalog_table_provider.rs

@@ -22,7 +22,6 @@ use async_trait::async_trait;
 use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
 use datafusion::catalog::Session;
 use datafusion::common::{config_datafusion_err, not_impl_err};
-use datafusion::datasource::source::DataSourceExec;
 use datafusion::datasource::{TableProvider, TableType};
 use datafusion::error::Result;
 use datafusion::logical_expr::dml::InsertOp;
@@ -36,13 +35,14 @@ use iceberg::{Catalog, NamespaceIdent, TableIdent};
 
 use crate::error::to_datafusion_error;
 use crate::metadata::table::IcebergMetadataTableProvider;
-use crate::read::IcebergDataSource;
 use crate::write::commit::IcebergCommitExec;
 use crate::write::exec::IcebergWriteExec;
 use crate::write::project::project_with_partition;
 use crate::write::repartition::repartition;
 use crate::write::sort::sort_by_partition;
 
+use super::IcebergStaticTableProvider;
+
 /// Catalog-backed table provider with automatic metadata refresh.
 ///
 /// This provider loads fresh table metadata from the catalog on every scan and write
@@ -126,7 +126,7 @@ impl TableProvider for IcebergCatalogTableProvider {
 
     async fn scan(
         &self,
-        _state: &dyn Session,
+        state: &dyn Session,
         projection: Option<&Vec<usize>>,
         filters: &[Expr],
         limit: Option<usize>,
@@ -138,13 +138,9 @@ impl TableProvider for IcebergCatalogTableProvider {
             .await
             .map_err(to_datafusion_error)?;
 
-        // Create scan with fresh metadata (always use current snapshot)
-        let source = IcebergDataSource::builder(table, self.schema.clone())
-            .with_projection(projection)
-            .with_filters(filters)
-            .with_fetch(limit)
-            .build()?;
-        Ok(DataSourceExec::from_data_source(source))
+        IcebergStaticTableProvider::from_table_with_schema(table, self.schema.clone())
+            .scan(state, projection, filters, limit)
+            .await
     }
 
     fn supports_filters_pushdown(
@@ -233,6 +229,7 @@ mod tests {
     use std::sync::Arc;
 
     use datafusion::datasource::TableProvider;
+    use datafusion::datasource::source::DataSourceExec;
     use datafusion::error::DataFusionError;
     use datafusion::physical_plan::ExecutionPlan;
     use datafusion::prelude::SessionContext;
@@ -242,6 +239,7 @@ mod tests {
     use tempfile::TempDir;
 
     use super::*;
+    use crate::read::IcebergDataSource;
 
     #[tokio::test]
     async fn test_catalog_backed_provider_creation() {

crates/datafusion/src/table_provider/static_table_provider.rs

@@ -60,11 +60,17 @@ impl IcebergStaticTableProvider {
             schema_to_arrow_schema(table.metadata().current_schema())
                 .map_err(to_datafusion_error)?,
         );
-        Ok(IcebergStaticTableProvider {
+        Ok(Self::from_table_with_schema(table, schema))
+    }
+
+    // Catalog scans must use the catalog provider's cached schema even when the
+    // loaded table has newer metadata.
+    pub(crate) fn from_table_with_schema(table: Table, schema: ArrowSchemaRef) -> Self {
+        Self {
             table,
             snapshot_id: None,
             schema,
-        })
+        }
     }
 
     /// Creates a static provider for a specific table snapshot.

crates/datafusion/tests/integration_datafusion_test.rs

@@ -1408,5 +1408,25 @@ async fn test_scan_after_schema_evolution_reads_provider_columns()
         +------+--------+"#]]
     .assert_eq(&pretty_format_batches(&[sorted])?.to_string());
 
+    // Projection indices still refer to the cached provider schema, not the
+    // newly loaded table schema.
+    let projection = vec![1];
+    let plan = provider
+        .scan(&ctx.state(), Some(&projection), &[], None)
+        .await?;
+    assert_eq!(plan.schema().field(0).name(), "foo2");
+    let batches = run_batches(plan.as_ref(), &ctx).await?;
+    let batch = concat_batches(&batches[0].schema(), &batches)?;
+    let order = sort_to_indices(batch.column(0), None, None)?;
+    let sorted = take_record_batch(&batch, &order)?;
+    expect![[r#"
+        +--------+
+        | foo2   |
+        +--------+
+        | alan   |
+        | turing |
+        +--------+"#]]
+    .assert_eq(&pretty_format_batches(&[sorted])?.to_string());
+
     Ok(())
 }