From 5c06d15127757c169e6825cdb4258155a42b2968 Mon Sep 17 00:00:00 2001
From: "alex.bianchi" <alex.bianchi@datadoghq.com>
Date: Mon, 12 Oct 2026 12:31:39 -0400
Subject: [PATCH] refactor: delegate catalog scans to a static table provider

---
 .../table_provider/catalog_table_provider.rs  | 18 ++++++++---------
 .../table_provider/static_table_provider.rs   | 10 ++++++++--
 .../tests/integration_datafusion_test.rs      | 20 +++++++++++++++++++
 3 files changed, 36 insertions(+), 12 deletions(-)

diff --git a/crates/datafusion/src/table_provider/catalog_table_provider.rs b/crates/datafusion/src/table_provider/catalog_table_provider.rs
index 6dbb79d..8c9e024 100644
--- a/crates/datafusion/src/table_provider/catalog_table_provider.rs
+++ b/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() {
diff --git a/crates/datafusion/src/table_provider/static_table_provider.rs b/crates/datafusion/src/table_provider/static_table_provider.rs
index 5ebabc8..a343399 100644
--- a/crates/datafusion/src/table_provider/static_table_provider.rs
+++ b/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.
diff --git a/crates/datafusion/tests/integration_datafusion_test.rs b/crates/datafusion/tests/integration_datafusion_test.rs
index 8d300a3..fc3d62b 100644
--- a/crates/datafusion/tests/integration_datafusion_test.rs
+++ b/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(())
 }