From c8c9171e558ed00a703141148e6a2f6576d833db Mon Sep 17 00:00:00 2001 From: Noah Date: Sat, 10 Oct 2026 23:50:12 -0400 Subject: [PATCH] feat: add IcebergCatalogTableProvider::new_with_schema Builds a catalog-backed provider from its catalog, table identifier and schema without loading the table, so a codec can rebuild a provider in another process that declares exactly the original's schema. --- .../table_provider/catalog_table_provider.rs | 21 +++++++- .../tests/integration_datafusion_test.rs | 52 +++++++++++-------- 2 files changed, 49 insertions(+), 24 deletions(-) diff --git a/crates/datafusion/src/table_provider/catalog_table_provider.rs b/crates/datafusion/src/table_provider/catalog_table_provider.rs index 5145b42..7d21319 100644 --- a/crates/datafusion/src/table_provider/catalog_table_provider.rs +++ b/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, 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, + table_ident: TableIdent, + schema: ArrowSchemaRef, + ) -> Self { + IcebergCatalogTableProvider { catalog, table_ident, schema, - }) + } } pub(crate) async fn metadata_table( diff --git a/crates/datafusion/tests/integration_datafusion_test.rs b/crates/datafusion/tests/integration_datafusion_test.rs index 6511a15..8ef0add 100644 --- a/crates/datafusion/tests/integration_datafusion_test.rs +++ b/crates/datafusion/tests/integration_datafusion_test.rs @@ -1065,12 +1065,11 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box> { .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(()) }