feat: add IcebergCatalogTableProvider::new_with_schema

crates/datafusion/src/table_provider/catalog_table_provider.rs

@@ -66,6 +66,11 @@ impl IcebergCatalogTableProvider {
     ///
     /// Loads the table once to get the initial schema, then stores the catalog
     /// reference for future metadata refreshes on each operation.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the catalog cannot load the table, or its schema has
+    /// no Arrow equivalent.
     pub async fn try_new(
         catalog: Arc<dyn Catalog>,
         namespace: NamespaceIdent,
@@ -83,11 +88,23 @@ impl IcebergCatalogTableProvider {
                 .map_err(to_datafusion_error)?,
         );
 
-        Ok(IcebergCatalogTableProvider {
+        Ok(Self::new_with_schema(catalog, table_ident, schema))
+    }
+
+    /// Creates a provider for `table_ident` that declares `schema`, without loading
+    /// the table. For rebuilding a provider in another process from [`Self::catalog`],
+    /// [`Self::table_ident`] and [`TableProvider::schema`]. `schema` is not checked
+    /// against the table; scans read its columns by name from the snapshot they scan.
+    pub fn new_with_schema(
+        catalog: Arc<dyn Catalog>,
+        table_ident: TableIdent,
+        schema: ArrowSchemaRef,
+    ) -> Self {
+        IcebergCatalogTableProvider {
             catalog,
             table_ident,
             schema,
-        })
+        }
     }
 
     pub(crate) async fn metadata_table(

crates/datafusion/tests/integration_datafusion_test.rs

@@ -1065,12 +1065,11 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
         .expect("a catalog-backed provider");
     assert_eq!(provider.table_ident(), &ident);
     assert!(Arc::ptr_eq(provider.catalog(), &client));
-    let rebuilt = IcebergCatalogTableProvider::try_new(
+    let rebuilt = IcebergCatalogTableProvider::new_with_schema(
         provider.catalog().clone(),
-        provider.table_ident().namespace().clone(),
-        provider.table_ident().name(),
-    )
-    .await?;
+        provider.table_ident().clone(),
+        provider.schema(),
+    );
     assert_eq!(rebuilt.table_ident(), &ident);
     assert_eq!(rebuilt.schema(), provider.schema());
 
@@ -1395,23 +1394,32 @@ async fn test_scan_after_schema_evolution_reads_provider_columns()
         .collect()
         .await?;
 
-    let plan = provider.scan(&ctx.state(), None, &[], None).await?;
-    assert_eq!(plan.schema(), provider.schema());
-    // The rows come from two data files, which may be read in either order.
-    // They are joined under their own schema, not the plan's, so that a column
-    // the plan does not report would show.
-    let batches = run_batches(plan.as_ref(), &ctx).await?;
-    let batch = concat_batches(&batches[0].schema(), &batches)?;
-    let order = sort_to_indices(batch.column_by_name("foo1").unwrap(), None, None)?;
-    let sorted = take_record_batch(&batch, &order)?;
-    expect![[r#"
-        +------+--------+
-        | foo1 | foo2   |
-        +------+--------+
-        | 1    | alan   |
-        | 2    | turing |
-        +------+--------+"#]]
-    .assert_eq(&pretty_format_batches(&[sorted])?.to_string());
+    // A provider rebuilt from the original's parts, as a codec does, scans the
+    // same columns as the original, not the table's current ones.
+    let rebuilt = IcebergCatalogTableProvider::new_with_schema(
+        provider.catalog().clone(),
+        provider.table_ident().clone(),
+        provider.schema(),
+    );
+    for provider in [provider.as_ref(), &rebuilt] {
+        let plan = provider.scan(&ctx.state(), None, &[], None).await?;
+        assert_eq!(plan.schema(), provider.schema());
+        // The rows come from two data files, which may be read in either order.
+        // They are joined under their own schema, not the plan's, so that a
+        // column the plan does not report would show.
+        let batches = run_batches(plan.as_ref(), &ctx).await?;
+        let batch = concat_batches(&batches[0].schema(), &batches)?;
+        let order = sort_to_indices(batch.column_by_name("foo1").unwrap(), None, None)?;
+        let sorted = take_record_batch(&batch, &order)?;
+        expect![[r#"
+            +------+--------+
+            | foo1 | foo2   |
+            +------+--------+
+            | 1    | alan   |
+            | 2    | turing |
+            +------+--------+"#]]
+        .assert_eq(&pretty_format_batches(&[sorted])?.to_string());
+    }
 
     Ok(())
 }