From cdc9dd86854e56e75acb85859c5976fa83742cf6 Mon Sep 17 00:00:00 2001 From: Noah Date: Tue, 21 Jul 2026 22:55:08 -0400 Subject: [PATCH 1/3] init --- crates/integrations/datafusion/public-api.txt | 150 ++++++++++++ crates/integrations/datafusion/src/catalog.rs | 20 ++ .../datafusion/src/catalog_config.rs | 54 +++++ crates/integrations/datafusion/src/lib.rs | 3 + .../datafusion/src/physical_plan/commit.rs | 42 +++- .../src/physical_plan/metadata_scan.rs | 7 + .../datafusion/src/physical_plan/mod.rs | 5 +- .../datafusion/src/physical_plan/project.rs | 51 ++-- .../datafusion/src/physical_plan/scan.rs | 33 ++- .../datafusion/src/physical_plan/write.rs | 36 ++- crates/integrations/datafusion/src/schema.rs | 103 +++++++- .../datafusion/src/table/metadata_table.rs | 41 ++++ .../integrations/datafusion/src/table/mod.rs | 222 ++++++++++++++---- 13 files changed, 682 insertions(+), 85 deletions(-) create mode 100644 crates/integrations/datafusion/src/catalog_config.rs diff --git a/crates/integrations/datafusion/public-api.txt b/crates/integrations/datafusion/public-api.txt index e197c2057d..760a923e13 100644 --- a/crates/integrations/datafusion/public-api.txt +++ b/crates/integrations/datafusion/public-api.txt @@ -2,6 +2,12 @@ pub mod iceberg_datafusion pub mod iceberg_datafusion::metadata_table pub struct iceberg_datafusion::metadata_table::IcebergMetadataTableProvider impl iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::metadata_type(&self) -> &iceberg::inspect::metadata_table::MetadataTableType +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::new(table: iceberg::table::Table, type: iceberg::inspect::metadata_table::MetadataTableType) -> Self +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table(&self) -> &iceberg::table::Table +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::with_catalog_config(self, catalog_config: core::option::Option) -> Self +impl iceberg_datafusion::metadata_table::IcebergMetadataTableProvider pub async fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan(self) -> datafusion_common::error::Result>> impl core::clone::Clone for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::clone(&self) -> iceberg_datafusion::metadata_table::IcebergMetadataTableProvider @@ -12,13 +18,49 @@ pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan<'l pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType pub mod iceberg_datafusion::physical_plan +pub struct iceberg_datafusion::physical_plan::IcebergCommitExec +impl iceberg_datafusion::physical_plan::IcebergCommitExec +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::new(table: iceberg::table::Table, catalog: alloc::sync::Arc, input: alloc::sync::Arc, schema: arrow_schema::schema::SchemaRef) -> Self +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::table(&self) -> &iceberg::table::Table +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::with_catalog_config(self, catalog_config: core::option::Option) -> Self +impl core::fmt::Debug for iceberg_datafusion::physical_plan::IcebergCommitExec +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl datafusion_physical_plan::display::DisplayAs for iceberg_datafusion::physical_plan::IcebergCommitExec +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::fmt_as(&self, t: datafusion_physical_plan::display::DisplayFormatType, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl datafusion_physical_plan::execution_plan::ExecutionPlan for iceberg_datafusion::physical_plan::IcebergCommitExec +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::benefits_from_input_partitioning(&self) -> alloc::vec::Vec +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::children(&self) -> alloc::vec::Vec<&alloc::sync::Arc> +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::execute(&self, partition: usize, context: alloc::sync::Arc) -> datafusion_common::error::Result +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::name(&self) -> &str +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::properties(&self) -> &alloc::sync::Arc +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::required_input_distribution(&self) -> alloc::vec::Vec +pub fn iceberg_datafusion::physical_plan::IcebergCommitExec::with_new_children(self: alloc::sync::Arc, children: alloc::vec::Vec>) -> datafusion_common::error::Result> +pub struct iceberg_datafusion::physical_plan::IcebergMetadataScan +impl iceberg_datafusion::physical_plan::IcebergMetadataScan +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::new(provider: iceberg_datafusion::metadata_table::IcebergMetadataTableProvider) -> Self +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::provider(&self) -> &iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +impl core::fmt::Debug for iceberg_datafusion::physical_plan::IcebergMetadataScan +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl datafusion_physical_plan::display::DisplayAs for iceberg_datafusion::physical_plan::IcebergMetadataScan +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::fmt_as(&self, _t: datafusion_physical_plan::display::DisplayFormatType, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl datafusion_physical_plan::execution_plan::ExecutionPlan for iceberg_datafusion::physical_plan::IcebergMetadataScan +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::children(&self) -> alloc::vec::Vec<&alloc::sync::Arc> +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::execute(&self, _partition: usize, _context: alloc::sync::Arc) -> datafusion_common::error::Result +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::name(&self) -> &str +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::properties(&self) -> &alloc::sync::Arc +pub fn iceberg_datafusion::physical_plan::IcebergMetadataScan::with_new_children(self: alloc::sync::Arc, _children: alloc::vec::Vec>) -> datafusion_common::error::Result> pub struct iceberg_datafusion::physical_plan::IcebergTableScan impl iceberg_datafusion::physical_plan::IcebergTableScan +pub fn iceberg_datafusion::physical_plan::IcebergTableScan::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> pub fn iceberg_datafusion::physical_plan::IcebergTableScan::limit(&self) -> core::option::Option +pub fn iceberg_datafusion::physical_plan::IcebergTableScan::new(table: iceberg::table::Table, snapshot_id: core::option::Option, schema: arrow_schema::schema::SchemaRef, projection: core::option::Option<&alloc::vec::Vec>, filters: &[datafusion_expr::expr::Expr], limit: core::option::Option) -> Self pub fn iceberg_datafusion::physical_plan::IcebergTableScan::predicates(&self) -> core::option::Option<&iceberg::expr::predicate::Predicate> pub fn iceberg_datafusion::physical_plan::IcebergTableScan::projection(&self) -> core::option::Option<&[alloc::string::String]> pub fn iceberg_datafusion::physical_plan::IcebergTableScan::snapshot_id(&self) -> core::option::Option pub fn iceberg_datafusion::physical_plan::IcebergTableScan::table(&self) -> &iceberg::table::Table +pub fn iceberg_datafusion::physical_plan::IcebergTableScan::with_catalog_config(self, catalog_config: core::option::Option) -> Self +pub fn iceberg_datafusion::physical_plan::IcebergTableScan::with_predicates(self, predicates: core::option::Option) -> Self impl core::fmt::Debug for iceberg_datafusion::physical_plan::IcebergTableScan pub fn iceberg_datafusion::physical_plan::IcebergTableScan::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl datafusion_physical_plan::display::DisplayAs for iceberg_datafusion::physical_plan::IcebergTableScan @@ -29,12 +71,59 @@ pub fn iceberg_datafusion::physical_plan::IcebergTableScan::execute(&self, _part pub fn iceberg_datafusion::physical_plan::IcebergTableScan::name(&self) -> &str pub fn iceberg_datafusion::physical_plan::IcebergTableScan::properties(&self) -> &alloc::sync::Arc pub fn iceberg_datafusion::physical_plan::IcebergTableScan::with_new_children(self: alloc::sync::Arc, _children: alloc::vec::Vec>) -> datafusion_common::error::Result> +pub struct iceberg_datafusion::physical_plan::IcebergWriteExec +impl iceberg_datafusion::physical_plan::IcebergWriteExec +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::new(table: iceberg::table::Table, input: alloc::sync::Arc, schema: arrow_schema::schema::SchemaRef) -> Self +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::table(&self) -> &iceberg::table::Table +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::with_catalog_config(self, catalog_config: core::option::Option) -> Self +impl core::fmt::Debug for iceberg_datafusion::physical_plan::IcebergWriteExec +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl datafusion_physical_plan::display::DisplayAs for iceberg_datafusion::physical_plan::IcebergWriteExec +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::fmt_as(&self, t: datafusion_physical_plan::display::DisplayFormatType, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl datafusion_physical_plan::execution_plan::ExecutionPlan for iceberg_datafusion::physical_plan::IcebergWriteExec +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::benefits_from_input_partitioning(&self) -> alloc::vec::Vec +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::children(&self) -> alloc::vec::Vec<&alloc::sync::Arc> +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::execute(&self, partition: usize, context: alloc::sync::Arc) -> datafusion_common::error::Result +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::maintains_input_order(&self) -> alloc::vec::Vec +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::name(&self) -> &str +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::properties(&self) -> &alloc::sync::Arc +pub fn iceberg_datafusion::physical_plan::IcebergWriteExec::with_new_children(self: alloc::sync::Arc, children: alloc::vec::Vec>) -> datafusion_common::error::Result> +pub struct iceberg_datafusion::physical_plan::PartitionExpr +impl iceberg_datafusion::physical_plan::PartitionExpr +pub fn iceberg_datafusion::physical_plan::PartitionExpr::partition_spec(&self) -> &alloc::sync::Arc +pub fn iceberg_datafusion::physical_plan::PartitionExpr::table_schema(&self) -> &iceberg::spec::schema::SchemaRef +pub fn iceberg_datafusion::physical_plan::PartitionExpr::try_new(partition_spec: alloc::sync::Arc, table_schema: iceberg::spec::schema::SchemaRef) -> datafusion_common::error::Result +impl core::clone::Clone for iceberg_datafusion::physical_plan::PartitionExpr +pub fn iceberg_datafusion::physical_plan::PartitionExpr::clone(&self) -> iceberg_datafusion::physical_plan::PartitionExpr +impl core::cmp::Eq for iceberg_datafusion::physical_plan::PartitionExpr +impl core::cmp::PartialEq for iceberg_datafusion::physical_plan::PartitionExpr +pub fn iceberg_datafusion::physical_plan::PartitionExpr::eq(&self, other: &Self) -> bool +impl core::fmt::Debug for iceberg_datafusion::physical_plan::PartitionExpr +pub fn iceberg_datafusion::physical_plan::PartitionExpr::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::fmt::Display for iceberg_datafusion::physical_plan::PartitionExpr +pub fn iceberg_datafusion::physical_plan::PartitionExpr::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::hash::Hash for iceberg_datafusion::physical_plan::PartitionExpr +pub fn iceberg_datafusion::physical_plan::PartitionExpr::hash(&self, state: &mut H) +impl datafusion_physical_expr_common::physical_expr::PhysicalExpr for iceberg_datafusion::physical_plan::PartitionExpr +pub fn iceberg_datafusion::physical_plan::PartitionExpr::children(&self) -> alloc::vec::Vec<&alloc::sync::Arc> +pub fn iceberg_datafusion::physical_plan::PartitionExpr::data_type(&self, _input_schema: &arrow_schema::schema::Schema) -> datafusion_common::error::Result +pub fn iceberg_datafusion::physical_plan::PartitionExpr::evaluate(&self, batch: &arrow_array::record_batch::RecordBatch) -> datafusion_common::error::Result +pub fn iceberg_datafusion::physical_plan::PartitionExpr::fmt_sql(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +pub fn iceberg_datafusion::physical_plan::PartitionExpr::nullable(&self, _input_schema: &arrow_schema::schema::Schema) -> datafusion_common::error::Result +pub fn iceberg_datafusion::physical_plan::PartitionExpr::with_new_children(self: alloc::sync::Arc, _children: alloc::vec::Vec>) -> datafusion_common::error::Result> pub fn iceberg_datafusion::physical_plan::convert_filters_to_predicate(filters: &[datafusion_expr::expr::Expr]) -> core::option::Option pub fn iceberg_datafusion::physical_plan::project_with_partition(input: alloc::sync::Arc, table: &iceberg::table::Table) -> datafusion_common::error::Result> pub mod iceberg_datafusion::table pub mod iceberg_datafusion::table::metadata_table pub struct iceberg_datafusion::table::metadata_table::IcebergMetadataTableProvider impl iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::metadata_type(&self) -> &iceberg::inspect::metadata_table::MetadataTableType +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::new(table: iceberg::table::Table, type: iceberg::inspect::metadata_table::MetadataTableType) -> Self +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table(&self) -> &iceberg::table::Table +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::with_catalog_config(self, catalog_config: core::option::Option) -> Self +impl iceberg_datafusion::metadata_table::IcebergMetadataTableProvider pub async fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan(self) -> datafusion_common::error::Result>> impl core::clone::Clone for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::clone(&self) -> iceberg_datafusion::metadata_table::IcebergMetadataTableProvider @@ -55,6 +144,23 @@ impl core::fmt::Debug for iceberg_datafusion::table_provider_factory::IcebergTab pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl datafusion_catalog::table::TableProviderFactory for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub struct iceberg_datafusion::table::IcebergMetadataTableProvider +impl iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::metadata_type(&self) -> &iceberg::inspect::metadata_table::MetadataTableType +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::new(table: iceberg::table::Table, type: iceberg::inspect::metadata_table::MetadataTableType) -> Self +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table(&self) -> &iceberg::table::Table +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::with_catalog_config(self, catalog_config: core::option::Option) -> Self +impl iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub async fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan(self) -> datafusion_common::error::Result>> +impl core::clone::Clone for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::clone(&self) -> iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +impl core::fmt::Debug for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl datafusion_catalog::table::TableProvider for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, _projection: core::option::Option<&'life2 alloc::vec::Vec>, _filters: &'life3 [datafusion_expr::expr::Expr], _limit: core::option::Option) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType pub struct iceberg_datafusion::table::IcebergStaticTableProvider impl iceberg_datafusion::IcebergStaticTableProvider pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_from_table(table: iceberg::table::Table) -> iceberg::error::Result @@ -70,6 +176,12 @@ pub fn iceberg_datafusion::IcebergStaticTableProvider::schema(&self) -> arrow_sc pub fn iceberg_datafusion::IcebergStaticTableProvider::supports_filters_pushdown(&self, filters: &[&datafusion_expr::expr::Expr]) -> datafusion_common::error::Result> pub fn iceberg_datafusion::IcebergStaticTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType pub struct iceberg_datafusion::table::IcebergTableProvider +impl iceberg_datafusion::IcebergTableProvider +pub fn iceberg_datafusion::IcebergTableProvider::config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> +pub fn iceberg_datafusion::IcebergTableProvider::snapshot_id(&self) -> core::option::Option +pub fn iceberg_datafusion::IcebergTableProvider::table_ident(&self) -> &iceberg::catalog::TableIdent +pub async fn iceberg_datafusion::IcebergTableProvider::try_new_with_config(catalog: alloc::sync::Arc, config: iceberg_datafusion::IcebergCatalogConfig, namespace: iceberg::catalog::NamespaceIdent, name: impl core::convert::Into) -> iceberg::error::Result +pub fn iceberg_datafusion::IcebergTableProvider::with_snapshot_id(self, snapshot_id: core::option::Option) -> Self impl core::clone::Clone for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::clone(&self) -> iceberg_datafusion::IcebergTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergTableProvider @@ -91,14 +203,46 @@ impl core::fmt::Debug for iceberg_datafusion::table_provider_factory::IcebergTab pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl datafusion_catalog::table::TableProviderFactory for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub struct iceberg_datafusion::IcebergCatalogConfig +pub iceberg_datafusion::IcebergCatalogConfig::name: alloc::string::String +pub iceberg_datafusion::IcebergCatalogConfig::props: std::collections::hash::map::HashMap +pub iceberg_datafusion::IcebergCatalogConfig::type: alloc::string::String +impl iceberg_datafusion::IcebergCatalogConfig +pub fn iceberg_datafusion::IcebergCatalogConfig::new(type: impl core::convert::Into, name: impl core::convert::Into, props: std::collections::hash::map::HashMap) -> Self +impl core::clone::Clone for iceberg_datafusion::IcebergCatalogConfig +pub fn iceberg_datafusion::IcebergCatalogConfig::clone(&self) -> iceberg_datafusion::IcebergCatalogConfig +impl core::cmp::Eq for iceberg_datafusion::IcebergCatalogConfig +impl core::cmp::PartialEq for iceberg_datafusion::IcebergCatalogConfig +pub fn iceberg_datafusion::IcebergCatalogConfig::eq(&self, other: &iceberg_datafusion::IcebergCatalogConfig) -> bool +impl core::fmt::Debug for iceberg_datafusion::IcebergCatalogConfig +pub fn iceberg_datafusion::IcebergCatalogConfig::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg_datafusion::IcebergCatalogConfig pub struct iceberg_datafusion::IcebergCatalogProvider impl iceberg_datafusion::IcebergCatalogProvider pub async fn iceberg_datafusion::IcebergCatalogProvider::try_new(client: alloc::sync::Arc) -> iceberg::error::Result +pub async fn iceberg_datafusion::IcebergCatalogProvider::try_new_with_config(client: alloc::sync::Arc, config: iceberg_datafusion::IcebergCatalogConfig) -> iceberg::error::Result impl core::fmt::Debug for iceberg_datafusion::IcebergCatalogProvider pub fn iceberg_datafusion::IcebergCatalogProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl datafusion_catalog::catalog::CatalogProvider for iceberg_datafusion::IcebergCatalogProvider pub fn iceberg_datafusion::IcebergCatalogProvider::schema(&self, name: &str) -> core::option::Option> pub fn iceberg_datafusion::IcebergCatalogProvider::schema_names(&self) -> alloc::vec::Vec +pub struct iceberg_datafusion::IcebergMetadataTableProvider +impl iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::metadata_type(&self) -> &iceberg::inspect::metadata_table::MetadataTableType +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::new(table: iceberg::table::Table, type: iceberg::inspect::metadata_table::MetadataTableType) -> Self +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table(&self) -> &iceberg::table::Table +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::with_catalog_config(self, catalog_config: core::option::Option) -> Self +impl iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub async fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan(self) -> datafusion_common::error::Result>> +impl core::clone::Clone for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::clone(&self) -> iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +impl core::fmt::Debug for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl datafusion_catalog::table::TableProvider for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, _projection: core::option::Option<&'life2 alloc::vec::Vec>, _filters: &'life3 [datafusion_expr::expr::Expr], _limit: core::option::Option) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef +pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType pub struct iceberg_datafusion::IcebergStaticTableProvider impl iceberg_datafusion::IcebergStaticTableProvider pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_from_table(table: iceberg::table::Table) -> iceberg::error::Result @@ -114,6 +258,12 @@ pub fn iceberg_datafusion::IcebergStaticTableProvider::schema(&self) -> arrow_sc pub fn iceberg_datafusion::IcebergStaticTableProvider::supports_filters_pushdown(&self, filters: &[&datafusion_expr::expr::Expr]) -> datafusion_common::error::Result> pub fn iceberg_datafusion::IcebergStaticTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType pub struct iceberg_datafusion::IcebergTableProvider +impl iceberg_datafusion::IcebergTableProvider +pub fn iceberg_datafusion::IcebergTableProvider::config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> +pub fn iceberg_datafusion::IcebergTableProvider::snapshot_id(&self) -> core::option::Option +pub fn iceberg_datafusion::IcebergTableProvider::table_ident(&self) -> &iceberg::catalog::TableIdent +pub async fn iceberg_datafusion::IcebergTableProvider::try_new_with_config(catalog: alloc::sync::Arc, config: iceberg_datafusion::IcebergCatalogConfig, namespace: iceberg::catalog::NamespaceIdent, name: impl core::convert::Into) -> iceberg::error::Result +pub fn iceberg_datafusion::IcebergTableProvider::with_snapshot_id(self, snapshot_id: core::option::Option) -> Self impl core::clone::Clone for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::clone(&self) -> iceberg_datafusion::IcebergTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergTableProvider diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index 2c6e1ff002..a838ce2b47 100644 --- a/crates/integrations/datafusion/src/catalog.rs +++ b/crates/integrations/datafusion/src/catalog.rs @@ -22,6 +22,7 @@ use datafusion::catalog::{CatalogProvider, SchemaProvider}; use futures::future::try_join_all; use iceberg::{Catalog, NamespaceIdent, Result}; +use crate::IcebergCatalogConfig; use crate::schema::IcebergSchemaProvider; /// Provides an interface to manage and access multiple schemas @@ -46,6 +47,24 @@ impl IcebergCatalogProvider { /// attempts to create a schema provider for each namespace, and /// collects these providers into a `HashMap`. pub async fn try_new(client: Arc) -> Result { + Self::try_new_impl(client, None).await + } + + /// Like [`try_new`](Self::try_new), but threads a serializable + /// [`IcebergCatalogConfig`] into every schema and table provider it creates, + /// so the catalog's tables can be queried by a distributed engine such as + /// Ballista. The `client` must already be built from the same `config`. + pub async fn try_new_with_config( + client: Arc, + config: IcebergCatalogConfig, + ) -> Result { + Self::try_new_impl(client, Some(config)).await + } + + async fn try_new_impl( + client: Arc, + config: Option, + ) -> Result { // TODO: // Schemas and providers should be cached and evicted based on time // As of right now; schemas might become stale. @@ -62,6 +81,7 @@ impl IcebergCatalogProvider { .map(|name| { IcebergSchemaProvider::try_new( client.clone(), + config.clone(), NamespaceIdent::new(name.clone()), ) }) diff --git a/crates/integrations/datafusion/src/catalog_config.rs b/crates/integrations/datafusion/src/catalog_config.rs new file mode 100644 index 0000000000..b62e500ffe --- /dev/null +++ b/crates/integrations/datafusion/src/catalog_config.rs @@ -0,0 +1,54 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::HashMap; + +/// A serializable description of the catalog (and storage) that backs an +/// [`IcebergTableProvider`](crate::table::IcebergTableProvider). +/// +/// This is the minimal, self-contained handle needed to *reconstruct* a catalog +/// and its associated `FileIO` on a remote node. It deliberately holds only +/// plain data (no live connections) so that distributed query engines such as +/// Ballista can serialize it, ship it to executors, and rebuild the catalog +/// there via a catalog loader (e.g. `iceberg-catalog-loader`) and the storage +/// via `FileIOBuilder::with_props`. +/// +/// The `props` map carries both the catalog connection properties (e.g. the +/// REST catalog URI) and the storage/`FileIO` properties (e.g. S3 endpoint and +/// credentials); in practice these live together in a single map. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct IcebergCatalogConfig { + /// The catalog type, e.g. `"rest"`, `"sql"`, `"glue"`. + pub r#type: String, + pub name: String, + /// Catalog connection and storage properties. + pub props: HashMap, +} + +impl IcebergCatalogConfig { + pub fn new( + r#type: impl Into, + name: impl Into, + props: HashMap, + ) -> Self { + Self { + r#type: r#type.into(), + name: name.into(), + props, + } + } +} diff --git a/crates/integrations/datafusion/src/lib.rs b/crates/integrations/datafusion/src/lib.rs index 4b0ea8606d..19475aeda2 100644 --- a/crates/integrations/datafusion/src/lib.rs +++ b/crates/integrations/datafusion/src/lib.rs @@ -18,6 +18,9 @@ mod catalog; pub use catalog::*; +mod catalog_config; +pub use catalog_config::*; + mod error; pub use error::*; diff --git a/crates/integrations/datafusion/src/physical_plan/commit.rs b/crates/integrations/datafusion/src/physical_plan/commit.rs index 9ae8b845ce..a9cb294122 100644 --- a/crates/integrations/datafusion/src/physical_plan/commit.rs +++ b/crates/integrations/datafusion/src/physical_plan/commit.rs @@ -40,13 +40,17 @@ use crate::to_datafusion_error; /// IcebergCommitExec is responsible for collecting the files written and use /// [`Transaction::fast_append`] to commit the data files written. #[derive(Debug)] -pub(crate) struct IcebergCommitExec { +pub struct IcebergCommitExec { table: Table, catalog: Arc, input: Arc, schema: ArrowSchemaRef, count_schema: ArrowSchemaRef, plan_properties: Arc, + /// Optional serializable catalog/storage config, populated when this node is + /// built through a config-backed provider so it can be reconstructed on a + /// remote node by a distributed engine. + catalog_config: Option, } impl IcebergCommitExec { @@ -67,9 +71,29 @@ impl IcebergCommitExec { schema, count_schema, plan_properties, + catalog_config: None, } } + /// Attaches a serializable catalog/storage config to this node so that a + /// distributed engine can reconstruct it (including the catalog) on a remote + /// node. + pub fn with_catalog_config( + mut self, + catalog_config: Option, + ) -> Self { + self.catalog_config = catalog_config; + self + } + + pub fn catalog_config(&self) -> Option<&crate::IcebergCatalogConfig> { + self.catalog_config.as_ref() + } + + pub fn table(&self) -> &Table { + &self.table + } + // Compute the plan properties for this execution plan fn compute_properties(schema: ArrowSchemaRef) -> Arc { Arc::new(PlanProperties::new( @@ -155,12 +179,15 @@ impl ExecutionPlan for IcebergCommitExec { ))); } - Ok(Arc::new(IcebergCommitExec::new( - self.table.clone(), - self.catalog.clone(), - children[0].clone(), - self.schema.clone(), - ))) + Ok(Arc::new( + IcebergCommitExec::new( + self.table.clone(), + self.catalog.clone(), + children[0].clone(), + self.schema.clone(), + ) + .with_catalog_config(self.catalog_config.clone()), + )) } fn execute( @@ -659,6 +686,7 @@ mod tests { let iceberg_table_provider = IcebergTableProvider::try_new( catalog.clone(), + None, namespace.clone(), table_name.to_string(), ) diff --git a/crates/integrations/datafusion/src/physical_plan/metadata_scan.rs b/crates/integrations/datafusion/src/physical_plan/metadata_scan.rs index f15acf807b..d4a2884d46 100644 --- a/crates/integrations/datafusion/src/physical_plan/metadata_scan.rs +++ b/crates/integrations/datafusion/src/physical_plan/metadata_scan.rs @@ -45,6 +45,13 @@ impl IcebergMetadataScan { properties, } } + + /// Returns the metadata-table provider this node scans, so a distributed + /// engine can serialize the catalog config + table identifier + metadata type + /// it carries and rebuild it on a remote node. + pub fn provider(&self) -> &IcebergMetadataTableProvider { + &self.provider + } } impl DisplayAs for IcebergMetadataScan { diff --git a/crates/integrations/datafusion/src/physical_plan/mod.rs b/crates/integrations/datafusion/src/physical_plan/mod.rs index aeac30de32..024266a4e9 100644 --- a/crates/integrations/datafusion/src/physical_plan/mod.rs +++ b/crates/integrations/datafusion/src/physical_plan/mod.rs @@ -26,6 +26,9 @@ pub(crate) mod write; pub(crate) const DATA_FILES_COL_NAME: &str = "data_files"; +pub use commit::IcebergCommitExec; pub use expr_to_predicate::convert_filters_to_predicate; -pub use project::project_with_partition; +pub use metadata_scan::IcebergMetadataScan; +pub use project::{PartitionExpr, project_with_partition}; pub use scan::IcebergTableScan; +pub use write::IcebergWriteExec; diff --git a/crates/integrations/datafusion/src/physical_plan/project.rs b/crates/integrations/datafusion/src/physical_plan/project.rs index a77abd0eef..e4bb204dd5 100644 --- a/crates/integrations/datafusion/src/physical_plan/project.rs +++ b/crates/integrations/datafusion/src/physical_plan/project.rs @@ -30,7 +30,7 @@ use iceberg::arrow::{ PROJECTED_PARTITION_VALUE_COLUMN, PartitionValueCalculator, schema_to_arrow_schema, strip_metadata_from_schema, }; -use iceberg::spec::PartitionSpec; +use iceberg::spec::{PartitionSpec, SchemaRef}; use iceberg::table::Table; use crate::to_datafusion_error; @@ -79,10 +79,6 @@ pub fn project_with_partition( ))); } - let calculator = - PartitionValueCalculator::try_new(partition_spec.as_ref(), table_schema.as_ref()) - .map_err(to_datafusion_error)?; - let mut projection_exprs: Vec<(Arc, String)> = Vec::with_capacity(input_schema.fields().len() + 1); @@ -91,26 +87,51 @@ pub fn project_with_partition( projection_exprs.push((column_expr, field.name().clone())); } - let partition_expr = Arc::new(PartitionExpr::new(calculator, partition_spec.clone())); + let partition_expr = Arc::new(PartitionExpr::try_new( + partition_spec.clone(), + table_schema.clone(), + )?); projection_exprs.push((partition_expr, PROJECTED_PARTITION_VALUE_COLUMN.to_string())); let projection = ProjectionExec::try_new(projection_exprs, input)?; Ok(Arc::new(projection)) } -/// PhysicalExpr implementation for partition value calculation +/// `PhysicalExpr` that computes Iceberg partition values for each input row. +/// +/// Alongside the live (non-serializable) [`PartitionValueCalculator`], it retains +/// the [`PartitionSpec`] and table schema it was built from. A distributed engine +/// can serialize those two — both are self-contained iceberg spec types — and +/// rebuild an equivalent expression on a remote node via [`PartitionExpr::try_new`]. #[derive(Debug, Clone)] -struct PartitionExpr { +pub struct PartitionExpr { calculator: Arc, partition_spec: Arc, + table_schema: SchemaRef, } impl PartitionExpr { - fn new(calculator: PartitionValueCalculator, partition_spec: Arc) -> Self { - Self { + /// Builds a partition expression from a partition spec and the table schema + /// it is bound to, constructing the underlying [`PartitionValueCalculator`]. + pub fn try_new(partition_spec: Arc, table_schema: SchemaRef) -> DFResult { + let calculator = + PartitionValueCalculator::try_new(partition_spec.as_ref(), table_schema.as_ref()) + .map_err(to_datafusion_error)?; + Ok(Self { calculator: Arc::new(calculator), partition_spec, - } + table_schema, + }) + } + + /// The partition spec whose values this expression computes. + pub fn partition_spec(&self) -> &Arc { + &self.partition_spec + } + + /// The table schema the partition values are derived from. + pub fn table_schema(&self) -> &SchemaRef { + &self.table_schema } } @@ -244,8 +265,6 @@ mod tests { let input = Arc::new(EmptyExec::new(arrow_schema.clone())); - let calculator = PartitionValueCalculator::try_new(&partition_spec, &table_schema).unwrap(); - let mut projection_exprs: Vec<(Arc, String)> = Vec::with_capacity(arrow_schema.fields().len() + 1); for (i, field) in arrow_schema.fields().iter().enumerate() { @@ -253,7 +272,9 @@ mod tests { projection_exprs.push((column_expr, field.name().clone())); } - let partition_expr = Arc::new(PartitionExpr::new(calculator, partition_spec)); + let partition_expr = Arc::new( + PartitionExpr::try_new(partition_spec, Arc::new(table_schema.clone())).unwrap(), + ); projection_exprs.push((partition_expr, PROJECTED_PARTITION_VALUE_COLUMN.to_string())); let projection = ProjectionExec::try_new(projection_exprs, input).unwrap(); @@ -298,7 +319,7 @@ mod tests { let partition_spec = Arc::new(partition_spec); let calculator = PartitionValueCalculator::try_new(&partition_spec, &table_schema).unwrap(); let partition_type = calculator.partition_arrow_type().clone(); - let expr = PartitionExpr::new(calculator, partition_spec); + let expr = PartitionExpr::try_new(partition_spec, Arc::new(table_schema.clone())).unwrap(); assert_eq!(expr.data_type(&arrow_schema).unwrap(), partition_type); assert!(!expr.nullable(&arrow_schema).unwrap()); diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs b/crates/integrations/datafusion/src/physical_plan/scan.rs index 1e53c20cec..12adbac43f 100644 --- a/crates/integrations/datafusion/src/physical_plan/scan.rs +++ b/crates/integrations/datafusion/src/physical_plan/scan.rs @@ -52,11 +52,15 @@ pub struct IcebergTableScan { predicates: Option, /// Optional limit on the number of rows to return limit: Option, + /// Optional serializable catalog/storage config, populated when this scan is + /// built through a config-backed provider so it can be reconstructed on a + /// remote node by a distributed engine. + catalog_config: Option, } impl IcebergTableScan { /// Creates a new [`IcebergTableScan`] object. - pub(crate) fn new( + pub fn new( table: Table, snapshot_id: Option, schema: ArrowSchemaRef, @@ -79,9 +83,36 @@ impl IcebergTableScan { projection, predicates, limit, + catalog_config: None, } } + /// Attaches a serializable catalog/storage config to this scan so that a + /// distributed engine can reconstruct it on a remote node. + pub fn with_catalog_config( + mut self, + catalog_config: Option, + ) -> Self { + self.catalog_config = catalog_config; + self + } + + /// Returns the serializable catalog/storage config, if any. + pub fn catalog_config(&self) -> Option<&crate::IcebergCatalogConfig> { + self.catalog_config.as_ref() + } + + /// Replaces the scan's pushed-down filter predicate. + /// + /// `IcebergTableScan::new` derives the predicate from DataFusion `Expr` + /// filters; this setter lets a distributed engine restore an already-built + /// [`Predicate`] directly (e.g. after deserializing it), so file pruning is + /// preserved on remote nodes. + pub fn with_predicates(mut self, predicates: Option) -> Self { + self.predicates = predicates; + self + } + pub fn table(&self) -> &Table { &self.table } diff --git a/crates/integrations/datafusion/src/physical_plan/write.rs b/crates/integrations/datafusion/src/physical_plan/write.rs index a7d771ec1b..fded5266fb 100644 --- a/crates/integrations/datafusion/src/physical_plan/write.rs +++ b/crates/integrations/datafusion/src/physical_plan/write.rs @@ -58,11 +58,15 @@ use crate::to_datafusion_error; /// The output of this execution plan is a record batch containing a single column with serialized /// data file information that can be used for committing the write operation to the table. #[derive(Debug)] -pub(crate) struct IcebergWriteExec { +pub struct IcebergWriteExec { table: Table, input: Arc, result_schema: ArrowSchemaRef, plan_properties: Arc, + /// Optional serializable catalog/storage config, populated when this node is + /// built through a config-backed provider so it can be reconstructed on a + /// remote node by a distributed engine. + catalog_config: Option, } impl IcebergWriteExec { @@ -74,9 +78,30 @@ impl IcebergWriteExec { input, result_schema: Self::make_result_schema(), plan_properties, + catalog_config: None, } } + /// Attaches a serializable catalog/storage config to this node so that a + /// distributed engine can reconstruct it on a remote node. + pub fn with_catalog_config( + mut self, + catalog_config: Option, + ) -> Self { + self.catalog_config = catalog_config; + self + } + + /// Returns the serializable catalog/storage config, if any. + pub fn catalog_config(&self) -> Option<&crate::IcebergCatalogConfig> { + self.catalog_config.as_ref() + } + + /// Returns the table this node writes to. + pub fn table(&self) -> &Table { + &self.table + } + fn compute_properties( input: &Arc, schema: ArrowSchemaRef, @@ -166,11 +191,10 @@ impl ExecutionPlan for IcebergWriteExec { ))); } - Ok(Arc::new(Self::new( - self.table.clone(), - Arc::clone(&children[0]), - self.schema(), - ))) + Ok(Arc::new( + Self::new(self.table.clone(), Arc::clone(&children[0]), self.schema()) + .with_catalog_config(self.catalog_config.clone()), + )) } /// Executes the write operation for the given partition. diff --git a/crates/integrations/datafusion/src/schema.rs b/crates/integrations/datafusion/src/schema.rs index 545863f8c6..839c984602 100644 --- a/crates/integrations/datafusion/src/schema.rs +++ b/crates/integrations/datafusion/src/schema.rs @@ -32,7 +32,7 @@ use iceberg::spec::FormatVersion; use iceberg::{Catalog, Error, ErrorKind, NamespaceIdent, Result, TableCreation, TableIdent}; use crate::table::IcebergTableProvider; -use crate::to_datafusion_error; +use crate::{IcebergCatalogConfig, to_datafusion_error}; /// Represents a [`SchemaProvider`] for the Iceberg [`Catalog`], managing /// access to table providers within a specific namespace. @@ -42,6 +42,10 @@ pub(crate) struct IcebergSchemaProvider { catalog: Arc, /// The namespace this schema represents namespace: NamespaceIdent, + /// Optional serializable catalog/storage config. When present, every table + /// provider this schema creates carries it, so catalog-registered tables can + /// be queried by a distributed engine. + config: Option, /// A concurrent map where keys are table names /// and values are dynamic references to objects implementing the /// [`TableProvider`] trait. @@ -57,8 +61,12 @@ impl IcebergSchemaProvider { /// This method retrieves a list of table names /// attempts to create a table provider for each table name, and /// collects these providers into a `HashMap`. + /// + /// When `config` is present it is threaded into every table provider this + /// schema creates, so the tables can be queried by a distributed engine. pub(crate) async fn try_new( client: Arc, + config: Option, namespace: NamespaceIdent, ) -> Result { // TODO: @@ -75,7 +83,14 @@ impl IcebergSchemaProvider { let providers = try_join_all( table_names .iter() - .map(|name| IcebergTableProvider::try_new(client.clone(), namespace.clone(), name)) + .map(|name| { + IcebergTableProvider::try_new( + client.clone(), + config.clone(), + namespace.clone(), + name.clone(), + ) + }) .collect::>(), ) .await?; @@ -88,6 +103,7 @@ impl IcebergSchemaProvider { Ok(IcebergSchemaProvider { catalog: client, namespace, + config, tables, }) } @@ -173,6 +189,7 @@ impl SchemaProvider for IcebergSchemaProvider { let catalog = self.catalog.clone(); let namespace = self.namespace.clone(); + let config = self.config.clone(); let tables = self.tables.clone(); let name_clone = name.clone(); @@ -191,9 +208,9 @@ impl SchemaProvider for IcebergSchemaProvider { .await .map_err(to_datafusion_error)?; - // Create a new table provider using the catalog reference let table_provider = IcebergTableProvider::try_new( catalog.clone(), + config.clone(), namespace.clone(), name_clone.clone(), ) @@ -317,13 +334,91 @@ mod tests { .await .unwrap(); - let provider = IcebergSchemaProvider::try_new(Arc::new(catalog), namespace) + let provider = IcebergSchemaProvider::try_new(Arc::new(catalog), None, namespace) .await .unwrap(); (provider, temp_dir) } + #[tokio::test] + async fn test_schema_provider_with_config_propagates_to_tables() { + use iceberg::TableCreation; + use iceberg::spec::{NestedField, PrimitiveType, Schema, Type}; + + let temp_dir = TempDir::new().unwrap(); + let warehouse_path = temp_dir.path().to_str().unwrap().to_string(); + let catalog = Arc::new( + MemoryCatalogBuilder::default() + .load( + "memory", + HashMap::from([(MEMORY_CATALOG_WAREHOUSE.to_string(), warehouse_path.clone())]), + ) + .await + .unwrap(), + ); + + let namespace = NamespaceIdent::new("test_ns".to_string()); + catalog + .create_namespace(&namespace, HashMap::new()) + .await + .unwrap(); + + let schema = Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + catalog + .create_table( + &namespace, + TableCreation::builder() + .name("t".to_string()) + .location(format!("{warehouse_path}/t")) + .schema(schema) + .properties(HashMap::new()) + .build(), + ) + .await + .unwrap(); + + // With config: the table provider carries it (and is therefore distributable). + let config = IcebergCatalogConfig::new("memory", "memory", HashMap::new()); + let with_config = + IcebergSchemaProvider::try_new(catalog.clone(), Some(config), namespace.clone()) + .await + .unwrap(); + let provider = with_config + .table("t") + .await + .unwrap() + .expect("table provider"); + let iceberg = provider + .downcast_ref::() + .expect("IcebergTableProvider"); + assert!( + iceberg.config().is_some(), + "try_new_with_config should propagate the config to its tables" + ); + + // Without config: providers stay config-less (legacy behavior). + let without_config = + IcebergSchemaProvider::try_new(catalog.clone(), None, namespace.clone()) + .await + .unwrap(); + let provider = without_config + .table("t") + .await + .unwrap() + .expect("table provider"); + let iceberg = provider + .downcast_ref::() + .expect("IcebergTableProvider"); + assert!(iceberg.config().is_none()); + } + #[tokio::test] async fn test_register_table_with_data_fails() { let (schema_provider, _temp_dir) = create_test_schema_provider().await; diff --git a/crates/integrations/datafusion/src/table/metadata_table.rs b/crates/integrations/datafusion/src/table/metadata_table.rs index 43e069cf5a..95d582ebf2 100644 --- a/crates/integrations/datafusion/src/table/metadata_table.rs +++ b/crates/integrations/datafusion/src/table/metadata_table.rs @@ -40,6 +40,47 @@ use crate::to_datafusion_error; pub struct IcebergMetadataTableProvider { pub(crate) table: Table, pub(crate) r#type: MetadataTableType, + /// Optional serializable catalog/storage config, populated when this provider + /// is built through a config-backed table provider so that a distributed + /// engine can reconstruct it (reload the table from the catalog) on a remote + /// node. + catalog_config: Option, +} + +impl IcebergMetadataTableProvider { + /// Creates a metadata-table provider over an already-loaded table. + pub fn new(table: Table, r#type: MetadataTableType) -> Self { + Self { + table, + r#type, + catalog_config: None, + } + } + + /// Attaches a serializable catalog/storage config so that a distributed engine + /// can reconstruct this provider on a remote node. + pub fn with_catalog_config( + mut self, + catalog_config: Option, + ) -> Self { + self.catalog_config = catalog_config; + self + } + + /// Returns the serializable catalog/storage config, if any. + pub fn catalog_config(&self) -> Option<&crate::IcebergCatalogConfig> { + self.catalog_config.as_ref() + } + + /// Returns the table this provider inspects. + pub fn table(&self) -> &Table { + &self.table + } + + /// Returns which metadata table this provider serves. + pub fn metadata_type(&self) -> &MetadataTableType { + &self.r#type + } } #[async_trait] diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index 2fd958dff4..7ec978687b 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -46,7 +46,7 @@ use iceberg::inspect::MetadataTableType; use iceberg::spec::TableProperties; use iceberg::table::Table; use iceberg::{Catalog, Error, ErrorKind, NamespaceIdent, Result, TableIdent}; -use metadata_table::IcebergMetadataTableProvider; +pub use metadata_table::IcebergMetadataTableProvider; use crate::error::to_datafusion_error; use crate::physical_plan::commit::IcebergCommitExec; @@ -72,6 +72,15 @@ pub struct IcebergTableProvider { table_ident: TableIdent, /// A reference-counted arrow `Schema` (cached at construction) schema: ArrowSchemaRef, + /// Optional serializable catalog/storage config. When present, it is + /// threaded into the execution plan nodes produced by `scan`/`insert_into` + /// so that a distributed engine can reconstruct them (and their catalog and + /// storage) on remote nodes. + config: Option, + /// Optional snapshot to read. `None` reads the current snapshot (refreshed + /// from the catalog on each scan); `Some` pins reads to that snapshot for + /// time-travel. Writes always target the current table state. + snapshot_id: Option, } impl IcebergTableProvider { @@ -81,6 +90,7 @@ impl IcebergTableProvider { /// reference for future metadata refreshes on each operation. pub(crate) async fn try_new( catalog: Arc, + config: Option, namespace: NamespaceIdent, name: impl Into, ) -> Result { @@ -94,16 +104,59 @@ impl IcebergTableProvider { catalog, table_ident, schema, + config, + snapshot_id: None, }) } + /// Creates a catalog-backed table provider that carries a serializable + /// [`IcebergCatalogConfig`](crate::IcebergCatalogConfig). + /// + /// The `catalog` must already be built from the same `config`. The config is + /// threaded into the execution plan nodes this provider produces so that a + /// distributed engine (e.g. Ballista) can serialize those nodes and rebuild + /// the catalog/storage on remote executors. + pub async fn try_new_with_config( + catalog: Arc, + config: crate::IcebergCatalogConfig, + namespace: NamespaceIdent, + name: impl Into, + ) -> Result { + Self::try_new(catalog, Some(config), namespace, name).await + } + + /// Pins reads to a specific snapshot for time-travel. `None` (the default) + /// reads the current snapshot. The snapshot id is threaded into the scan + /// node, so it is serialized and honored by a distributed engine as well. + pub fn with_snapshot_id(mut self, snapshot_id: Option) -> Self { + self.snapshot_id = snapshot_id; + self + } + + /// Returns the snapshot this provider reads, if pinned for time-travel. + pub fn snapshot_id(&self) -> Option { + self.snapshot_id + } + + /// Returns the serializable catalog/storage config, if this provider was + /// created with one. + pub fn config(&self) -> Option<&crate::IcebergCatalogConfig> { + self.config.as_ref() + } + + /// Returns the identifier of the table this provider serves. + pub fn table_ident(&self) -> &TableIdent { + &self.table_ident + } + pub(crate) async fn metadata_table( &self, r#type: MetadataTableType, ) -> Result { // Load fresh table metadata for metadata table access let table = self.catalog.load_table(&self.table_ident).await?; - Ok(IcebergMetadataTableProvider { table, r#type }) + Ok(IcebergMetadataTableProvider::new(table, r#type) + .with_catalog_config(self.config.clone())) } } @@ -131,15 +184,17 @@ impl TableProvider for IcebergTableProvider { .await .map_err(to_datafusion_error)?; - // Create scan with fresh metadata (always use current snapshot) - Ok(Arc::new(IcebergTableScan::new( - table, - None, // Always use current snapshot for catalog-backed provider - self.schema.clone(), - projection, - filters, - limit, - ))) + Ok(Arc::new( + IcebergTableScan::new( + table, + self.snapshot_id, + self.schema.clone(), + projection, + filters, + limit, + ) + .with_catalog_config(self.config.clone()), + )) } fn supports_filters_pushdown( @@ -218,21 +273,23 @@ impl TableProvider for IcebergTableProvider { sort_by_partition(repartitioned_plan)? }; - let write_plan = Arc::new(IcebergWriteExec::new( - table.clone(), - write_input, - self.schema.clone(), - )); + let write_plan = Arc::new( + IcebergWriteExec::new(table.clone(), write_input, self.schema.clone()) + .with_catalog_config(self.config.clone()), + ); // Merge the outputs of write_plan into one so we can commit all files together let coalesce_partitions = Arc::new(CoalescePartitionsExec::new(write_plan)); - Ok(Arc::new(IcebergCommitExec::new( - table, - self.catalog.clone(), - coalesce_partitions, - self.schema.clone(), - ))) + Ok(Arc::new( + IcebergCommitExec::new( + table, + self.catalog.clone(), + coalesce_partitions, + self.schema.clone(), + ) + .with_catalog_config(self.config.clone()), + )) } } @@ -523,10 +580,14 @@ mod tests { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; // Test creating a catalog-backed provider - let provider = - IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone()) - .await - .unwrap(); + let provider = IcebergTableProvider::try_new( + catalog.clone(), + None, + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); // Verify the schema is loaded correctly let schema = provider.schema(); @@ -539,10 +600,14 @@ mod tests { async fn test_catalog_backed_provider_scan() { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = - IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone()) - .await - .unwrap(); + let provider = IcebergTableProvider::try_new( + catalog.clone(), + None, + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); let ctx = SessionContext::new(); ctx.register_table("test_table", Arc::new(provider)) @@ -565,10 +630,14 @@ mod tests { async fn test_catalog_backed_provider_insert() { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = - IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone()) - .await - .unwrap(); + let provider = IcebergTableProvider::try_new( + catalog.clone(), + None, + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); let ctx = SessionContext::new(); ctx.register_table("test_table", Arc::new(provider)) @@ -592,10 +661,14 @@ mod tests { async fn test_physical_input_schema_consistent_with_logical_input_schema() { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = - IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone()) - .await - .unwrap(); + let provider = IcebergTableProvider::try_new( + catalog.clone(), + None, + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); let ctx = SessionContext::new(); ctx.register_table("test_table", Arc::new(provider)) @@ -711,7 +784,7 @@ mod tests { use datafusion::physical_plan::empty::EmptyExec; let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = IcebergTableProvider::try_new(catalog, namespace, table_name) + let provider = IcebergTableProvider::try_new(catalog, None, namespace, table_name) .await .unwrap(); let ctx = SessionContext::new(); @@ -752,10 +825,14 @@ mod tests { let (catalog, namespace, table_name, _temp_dir) = get_partitioned_test_catalog_and_table(Some(true)).await; - let provider = - IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone()) - .await - .unwrap(); + let provider = IcebergTableProvider::try_new( + catalog.clone(), + None, + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); let ctx = SessionContext::new(); let input_schema = provider.schema(); @@ -784,10 +861,14 @@ mod tests { let (catalog, namespace, table_name, _temp_dir) = get_partitioned_test_catalog_and_table(Some(false)).await; - let provider = - IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone()) - .await - .unwrap(); + let provider = IcebergTableProvider::try_new( + catalog.clone(), + None, + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); let ctx = SessionContext::new(); let input_schema = provider.schema(); @@ -843,10 +924,14 @@ mod tests { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = - IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone()) - .await - .unwrap(); + let provider = IcebergTableProvider::try_new( + catalog.clone(), + None, + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); let ctx = SessionContext::new(); let state = ctx.state(); @@ -894,4 +979,39 @@ mod tests { "Limit should be None when not specified" ); } + + #[tokio::test] + async fn test_with_snapshot_id_pins_scan() { + use datafusion::datasource::TableProvider; + + let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; + + // Default provider reads the current snapshot (None in the scan). + let provider = IcebergTableProvider::try_new( + catalog.clone(), + None, + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); + assert_eq!(provider.snapshot_id(), None); + + // Pinning a snapshot threads it into the scan node, where the codec reads + // it — so time-travel is honored locally and distributed. + let pinned = provider.with_snapshot_id(Some(123)); + assert_eq!(pinned.snapshot_id(), Some(123)); + + let ctx = SessionContext::new(); + let state = ctx.state(); + let scan_plan = pinned.scan(&state, None, &[], None).await.unwrap(); + let iceberg_scan = scan_plan + .downcast_ref::() + .expect("Expected IcebergTableScan"); + assert_eq!( + iceberg_scan.snapshot_id(), + Some(123), + "pinned snapshot should propagate to the scan node" + ); + } } From e85f190484154c13e49e39b1a2cb95a38fe20c50 Mon Sep 17 00:00:00 2001 From: Noah Date: Tue, 21 Jul 2026 23:15:50 -0400 Subject: [PATCH 2/3] cleanup --- .../integrations/datafusion/src/table/mod.rs | 35 ------------------- 1 file changed, 35 deletions(-) diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index 7ec978687b..7f57d18ff3 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -979,39 +979,4 @@ mod tests { "Limit should be None when not specified" ); } - - #[tokio::test] - async fn test_with_snapshot_id_pins_scan() { - use datafusion::datasource::TableProvider; - - let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - - // Default provider reads the current snapshot (None in the scan). - let provider = IcebergTableProvider::try_new( - catalog.clone(), - None, - namespace.clone(), - table_name.clone(), - ) - .await - .unwrap(); - assert_eq!(provider.snapshot_id(), None); - - // Pinning a snapshot threads it into the scan node, where the codec reads - // it — so time-travel is honored locally and distributed. - let pinned = provider.with_snapshot_id(Some(123)); - assert_eq!(pinned.snapshot_id(), Some(123)); - - let ctx = SessionContext::new(); - let state = ctx.state(); - let scan_plan = pinned.scan(&state, None, &[], None).await.unwrap(); - let iceberg_scan = scan_plan - .downcast_ref::() - .expect("Expected IcebergTableScan"); - assert_eq!( - iceberg_scan.snapshot_id(), - Some(123), - "pinned snapshot should propagate to the scan node" - ); - } } From c823ce628fb438ab64827e135045025c7d4e50af Mon Sep 17 00:00:00 2001 From: Noah Date: Wed, 22 Jul 2026 00:27:20 -0400 Subject: [PATCH 3/3] cleanup --- crates/integrations/datafusion/public-api.txt | 8 +- crates/integrations/datafusion/src/catalog.rs | 90 ++++++ .../datafusion/src/catalog_config.rs | 2 + .../datafusion/src/physical_plan/project.rs | 9 +- crates/integrations/datafusion/src/schema.rs | 4 +- .../integrations/datafusion/src/table/mod.rs | 278 ++++++++++++------ 6 files changed, 296 insertions(+), 95 deletions(-) diff --git a/crates/integrations/datafusion/public-api.txt b/crates/integrations/datafusion/public-api.txt index 760a923e13..b575d14c22 100644 --- a/crates/integrations/datafusion/public-api.txt +++ b/crates/integrations/datafusion/public-api.txt @@ -177,11 +177,9 @@ pub fn iceberg_datafusion::IcebergStaticTableProvider::supports_filters_pushdown pub fn iceberg_datafusion::IcebergStaticTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType pub struct iceberg_datafusion::table::IcebergTableProvider impl iceberg_datafusion::IcebergTableProvider -pub fn iceberg_datafusion::IcebergTableProvider::config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> -pub fn iceberg_datafusion::IcebergTableProvider::snapshot_id(&self) -> core::option::Option +pub fn iceberg_datafusion::IcebergTableProvider::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> pub fn iceberg_datafusion::IcebergTableProvider::table_ident(&self) -> &iceberg::catalog::TableIdent pub async fn iceberg_datafusion::IcebergTableProvider::try_new_with_config(catalog: alloc::sync::Arc, config: iceberg_datafusion::IcebergCatalogConfig, namespace: iceberg::catalog::NamespaceIdent, name: impl core::convert::Into) -> iceberg::error::Result -pub fn iceberg_datafusion::IcebergTableProvider::with_snapshot_id(self, snapshot_id: core::option::Option) -> Self impl core::clone::Clone for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::clone(&self) -> iceberg_datafusion::IcebergTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergTableProvider @@ -259,11 +257,9 @@ pub fn iceberg_datafusion::IcebergStaticTableProvider::supports_filters_pushdown pub fn iceberg_datafusion::IcebergStaticTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType pub struct iceberg_datafusion::IcebergTableProvider impl iceberg_datafusion::IcebergTableProvider -pub fn iceberg_datafusion::IcebergTableProvider::config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> -pub fn iceberg_datafusion::IcebergTableProvider::snapshot_id(&self) -> core::option::Option +pub fn iceberg_datafusion::IcebergTableProvider::catalog_config(&self) -> core::option::Option<&iceberg_datafusion::IcebergCatalogConfig> pub fn iceberg_datafusion::IcebergTableProvider::table_ident(&self) -> &iceberg::catalog::TableIdent pub async fn iceberg_datafusion::IcebergTableProvider::try_new_with_config(catalog: alloc::sync::Arc, config: iceberg_datafusion::IcebergCatalogConfig, namespace: iceberg::catalog::NamespaceIdent, name: impl core::convert::Into) -> iceberg::error::Result -pub fn iceberg_datafusion::IcebergTableProvider::with_snapshot_id(self, snapshot_id: core::option::Option) -> Self impl core::clone::Clone for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::clone(&self) -> iceberg_datafusion::IcebergTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergTableProvider diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index a838ce2b47..68f4a40289 100644 --- a/crates/integrations/datafusion/src/catalog.rs +++ b/crates/integrations/datafusion/src/catalog.rs @@ -111,3 +111,93 @@ impl CatalogProvider for IcebergCatalogProvider { self.schemas.get(name).cloned() } } + +#[cfg(test)] +mod tests { + use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; + use iceberg::spec::{NestedField, PrimitiveType, Schema, Type}; + use iceberg::{CatalogBuilder, TableCreation}; + use tempfile::TempDir; + + use super::*; + use crate::table::IcebergTableProvider; + + async fn catalog_with_one_table() -> (Arc, TempDir) { + let temp_dir = TempDir::new().unwrap(); + let warehouse_path = temp_dir.path().to_str().unwrap().to_string(); + let catalog = MemoryCatalogBuilder::default() + .load( + "memory", + HashMap::from([(MEMORY_CATALOG_WAREHOUSE.to_string(), warehouse_path.clone())]), + ) + .await + .unwrap(); + + let namespace = NamespaceIdent::new("test_ns".to_string()); + catalog + .create_namespace(&namespace, HashMap::new()) + .await + .unwrap(); + + let schema = Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + catalog + .create_table( + &namespace, + TableCreation::builder() + .name("t".to_string()) + .location(format!("{warehouse_path}/t")) + .schema(schema) + .properties(HashMap::new()) + .build(), + ) + .await + .unwrap(); + + (Arc::new(catalog), temp_dir) + } + + /// Downcasts the `t` table provider in the `test_ns` schema and returns + /// whether it carries an [`IcebergCatalogConfig`]. + async fn table_provider_has_config(provider: &IcebergCatalogProvider) -> bool { + let schema = provider.schema("test_ns").expect("test_ns schema"); + let table = schema.table("t").await.unwrap().expect("table provider"); + table + .downcast_ref::() + .expect("IcebergTableProvider") + .catalog_config() + .is_some() + } + + #[tokio::test] + async fn test_try_new_with_config_propagates_to_tables() { + let (catalog, _temp_dir) = catalog_with_one_table().await; + let config = IcebergCatalogConfig::new("memory", "memory", HashMap::new()); + + let provider = IcebergCatalogProvider::try_new_with_config(catalog, config) + .await + .unwrap(); + + assert!( + table_provider_has_config(&provider).await, + "try_new_with_config should thread the config down to table providers" + ); + } + + #[tokio::test] + async fn test_try_new_leaves_tables_config_less() { + let (catalog, _temp_dir) = catalog_with_one_table().await; + + let provider = IcebergCatalogProvider::try_new(catalog).await.unwrap(); + + assert!( + !table_provider_has_config(&provider).await, + "try_new should leave table providers without a config" + ); + } +} diff --git a/crates/integrations/datafusion/src/catalog_config.rs b/crates/integrations/datafusion/src/catalog_config.rs index b62e500ffe..ae01fb4508 100644 --- a/crates/integrations/datafusion/src/catalog_config.rs +++ b/crates/integrations/datafusion/src/catalog_config.rs @@ -34,6 +34,8 @@ use std::collections::HashMap; pub struct IcebergCatalogConfig { /// The catalog type, e.g. `"rest"`, `"sql"`, `"glue"`. pub r#type: String, + /// The catalog name it is registered under (used to look it up when + /// reconstructing the catalog on a remote node). pub name: String, /// Catalog connection and storage properties. pub props: HashMap, diff --git a/crates/integrations/datafusion/src/physical_plan/project.rs b/crates/integrations/datafusion/src/physical_plan/project.rs index e4bb204dd5..bd6b33f040 100644 --- a/crates/integrations/datafusion/src/physical_plan/project.rs +++ b/crates/integrations/datafusion/src/physical_plan/project.rs @@ -319,7 +319,14 @@ mod tests { let partition_spec = Arc::new(partition_spec); let calculator = PartitionValueCalculator::try_new(&partition_spec, &table_schema).unwrap(); let partition_type = calculator.partition_arrow_type().clone(); - let expr = PartitionExpr::try_new(partition_spec, Arc::new(table_schema.clone())).unwrap(); + let table_schema_ref = Arc::new(table_schema.clone()); + let expr = + PartitionExpr::try_new(partition_spec.clone(), table_schema_ref.clone()).unwrap(); + + // The getters expose the spec and schema the expression was built from, + // which a distributed engine serializes and rebuilds via `try_new`. + assert!(Arc::ptr_eq(expr.partition_spec(), &partition_spec)); + assert!(Arc::ptr_eq(expr.table_schema(), &table_schema_ref)); assert_eq!(expr.data_type(&arrow_schema).unwrap(), partition_type); assert!(!expr.nullable(&arrow_schema).unwrap()); diff --git a/crates/integrations/datafusion/src/schema.rs b/crates/integrations/datafusion/src/schema.rs index 839c984602..d82755c9e8 100644 --- a/crates/integrations/datafusion/src/schema.rs +++ b/crates/integrations/datafusion/src/schema.rs @@ -399,7 +399,7 @@ mod tests { .downcast_ref::() .expect("IcebergTableProvider"); assert!( - iceberg.config().is_some(), + iceberg.catalog_config().is_some(), "try_new_with_config should propagate the config to its tables" ); @@ -416,7 +416,7 @@ mod tests { let iceberg = provider .downcast_ref::() .expect("IcebergTableProvider"); - assert!(iceberg.config().is_none()); + assert!(iceberg.catalog_config().is_none()); } #[tokio::test] diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index 7f57d18ff3..ac3b644dff 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -76,11 +76,7 @@ pub struct IcebergTableProvider { /// threaded into the execution plan nodes produced by `scan`/`insert_into` /// so that a distributed engine can reconstruct them (and their catalog and /// storage) on remote nodes. - config: Option, - /// Optional snapshot to read. `None` reads the current snapshot (refreshed - /// from the catalog on each scan); `Some` pins reads to that snapshot for - /// time-travel. Writes always target the current table state. - snapshot_id: Option, + catalog_config: Option, } impl IcebergTableProvider { @@ -104,8 +100,7 @@ impl IcebergTableProvider { catalog, table_ident, schema, - config, - snapshot_id: None, + catalog_config: config, }) } @@ -125,23 +120,10 @@ impl IcebergTableProvider { Self::try_new(catalog, Some(config), namespace, name).await } - /// Pins reads to a specific snapshot for time-travel. `None` (the default) - /// reads the current snapshot. The snapshot id is threaded into the scan - /// node, so it is serialized and honored by a distributed engine as well. - pub fn with_snapshot_id(mut self, snapshot_id: Option) -> Self { - self.snapshot_id = snapshot_id; - self - } - - /// Returns the snapshot this provider reads, if pinned for time-travel. - pub fn snapshot_id(&self) -> Option { - self.snapshot_id - } - /// Returns the serializable catalog/storage config, if this provider was /// created with one. - pub fn config(&self) -> Option<&crate::IcebergCatalogConfig> { - self.config.as_ref() + pub fn catalog_config(&self) -> Option<&crate::IcebergCatalogConfig> { + self.catalog_config.as_ref() } /// Returns the identifier of the table this provider serves. @@ -156,7 +138,7 @@ impl IcebergTableProvider { // Load fresh table metadata for metadata table access let table = self.catalog.load_table(&self.table_ident).await?; Ok(IcebergMetadataTableProvider::new(table, r#type) - .with_catalog_config(self.config.clone())) + .with_catalog_config(self.catalog_config.clone())) } } @@ -187,13 +169,13 @@ impl TableProvider for IcebergTableProvider { Ok(Arc::new( IcebergTableScan::new( table, - self.snapshot_id, + None, // Always use current snapshot for catalog-backed provider self.schema.clone(), projection, filters, limit, ) - .with_catalog_config(self.config.clone()), + .with_catalog_config(self.catalog_config.clone()), )) } @@ -275,7 +257,7 @@ impl TableProvider for IcebergTableProvider { let write_plan = Arc::new( IcebergWriteExec::new(table.clone(), write_input, self.schema.clone()) - .with_catalog_config(self.config.clone()), + .with_catalog_config(self.catalog_config.clone()), ); // Merge the outputs of write_plan into one so we can commit all files together @@ -288,7 +270,7 @@ impl TableProvider for IcebergTableProvider { coalesce_partitions, self.schema.clone(), ) - .with_catalog_config(self.config.clone()), + .with_catalog_config(self.catalog_config.clone()), )) } } @@ -482,6 +464,24 @@ mod tests { ) } + /// Builds a config-less catalog-backed [`IcebergTableProvider`] over the + /// given table. Most catalog-backed tests don't care about the config, so + /// this keeps them free of the `try_new(.., None, ..)` boilerplate. + async fn catalog_backed_provider( + catalog: Arc, + namespace: NamespaceIdent, + table_name: String, + ) -> IcebergTableProvider { + IcebergTableProvider::try_new(catalog, None, namespace, table_name) + .await + .unwrap() + } + + /// A sample [`IcebergCatalogConfig`] for tests that exercise config threading. + fn sample_catalog_config() -> crate::IcebergCatalogConfig { + crate::IcebergCatalogConfig::new("memory", "memory", HashMap::new()) + } + // Tests for IcebergStaticTableProvider #[tokio::test] @@ -580,14 +580,8 @@ mod tests { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; // Test creating a catalog-backed provider - let provider = IcebergTableProvider::try_new( - catalog.clone(), - None, - namespace.clone(), - table_name.clone(), - ) - .await - .unwrap(); + let provider = + catalog_backed_provider(catalog.clone(), namespace.clone(), table_name.clone()).await; // Verify the schema is loaded correctly let schema = provider.schema(); @@ -600,14 +594,8 @@ mod tests { async fn test_catalog_backed_provider_scan() { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = IcebergTableProvider::try_new( - catalog.clone(), - None, - namespace.clone(), - table_name.clone(), - ) - .await - .unwrap(); + let provider = + catalog_backed_provider(catalog.clone(), namespace.clone(), table_name.clone()).await; let ctx = SessionContext::new(); ctx.register_table("test_table", Arc::new(provider)) @@ -630,14 +618,8 @@ mod tests { async fn test_catalog_backed_provider_insert() { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = IcebergTableProvider::try_new( - catalog.clone(), - None, - namespace.clone(), - table_name.clone(), - ) - .await - .unwrap(); + let provider = + catalog_backed_provider(catalog.clone(), namespace.clone(), table_name.clone()).await; let ctx = SessionContext::new(); ctx.register_table("test_table", Arc::new(provider)) @@ -661,14 +643,8 @@ mod tests { async fn test_physical_input_schema_consistent_with_logical_input_schema() { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = IcebergTableProvider::try_new( - catalog.clone(), - None, - namespace.clone(), - table_name.clone(), - ) - .await - .unwrap(); + let provider = + catalog_backed_provider(catalog.clone(), namespace.clone(), table_name.clone()).await; let ctx = SessionContext::new(); ctx.register_table("test_table", Arc::new(provider)) @@ -784,9 +760,7 @@ mod tests { use datafusion::physical_plan::empty::EmptyExec; let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = IcebergTableProvider::try_new(catalog, None, namespace, table_name) - .await - .unwrap(); + let provider = catalog_backed_provider(catalog, namespace, table_name).await; let ctx = SessionContext::new(); for (insert_op, expected_message) in [ @@ -825,14 +799,8 @@ mod tests { let (catalog, namespace, table_name, _temp_dir) = get_partitioned_test_catalog_and_table(Some(true)).await; - let provider = IcebergTableProvider::try_new( - catalog.clone(), - None, - namespace.clone(), - table_name.clone(), - ) - .await - .unwrap(); + let provider = + catalog_backed_provider(catalog.clone(), namespace.clone(), table_name.clone()).await; let ctx = SessionContext::new(); let input_schema = provider.schema(); @@ -861,14 +829,8 @@ mod tests { let (catalog, namespace, table_name, _temp_dir) = get_partitioned_test_catalog_and_table(Some(false)).await; - let provider = IcebergTableProvider::try_new( - catalog.clone(), - None, - namespace.clone(), - table_name.clone(), - ) - .await - .unwrap(); + let provider = + catalog_backed_provider(catalog.clone(), namespace.clone(), table_name.clone()).await; let ctx = SessionContext::new(); let input_schema = provider.schema(); @@ -924,14 +886,8 @@ mod tests { let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; - let provider = IcebergTableProvider::try_new( - catalog.clone(), - None, - namespace.clone(), - table_name.clone(), - ) - .await - .unwrap(); + let provider = + catalog_backed_provider(catalog.clone(), namespace.clone(), table_name.clone()).await; let ctx = SessionContext::new(); let state = ctx.state(); @@ -979,4 +935,154 @@ mod tests { "Limit should be None when not specified" ); } + + // Tests for catalog-config threading (the surface a distributed engine + // relies on to reconstruct these providers and plan nodes remotely). + + #[tokio::test] + async fn test_try_new_with_config_carries_config() { + let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; + + let provider = IcebergTableProvider::try_new_with_config( + catalog, + sample_catalog_config(), + namespace, + table_name, + ) + .await + .unwrap(); + + assert_eq!(provider.catalog_config(), Some(&sample_catalog_config())); + } + + #[tokio::test] + async fn test_scan_threads_catalog_config_into_scan_node() { + let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; + + let provider = IcebergTableProvider::try_new_with_config( + catalog, + sample_catalog_config(), + namespace, + table_name, + ) + .await + .unwrap(); + + let state = SessionContext::new().state(); + let scan_plan = provider.scan(&state, None, &[], None).await.unwrap(); + let scan = scan_plan + .downcast_ref::() + .expect("Expected IcebergTableScan"); + + assert_eq!(scan.catalog_config(), Some(&sample_catalog_config())); + } + + #[tokio::test] + async fn test_scan_omits_catalog_config_when_provider_has_none() { + let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; + let provider = catalog_backed_provider(catalog, namespace, table_name).await; + + let state = SessionContext::new().state(); + let scan_plan = provider.scan(&state, None, &[], None).await.unwrap(); + let scan = scan_plan + .downcast_ref::() + .expect("Expected IcebergTableScan"); + + assert!(scan.catalog_config().is_none()); + } + + #[tokio::test] + async fn test_insert_threads_catalog_config_into_write_and_commit_nodes() { + use datafusion::physical_plan::empty::EmptyExec; + + let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; + let provider = IcebergTableProvider::try_new_with_config( + catalog, + sample_catalog_config(), + namespace, + table_name, + ) + .await + .unwrap(); + + let input = Arc::new(EmptyExec::new(provider.schema())) as Arc; + let state = SessionContext::new().state(); + let insert_plan = provider + .insert_into(&state, input, InsertOp::Append) + .await + .unwrap(); + + // The top node is the commit; it must carry the config. + let commit = insert_plan + .downcast_ref::() + .expect("Expected IcebergCommitExec at the root of the insert plan"); + assert_eq!(commit.catalog_config(), Some(&sample_catalog_config())); + + // The write node buried in the plan must carry it too. + assert!( + plan_has_write_exec_with_config(&insert_plan), + "IcebergWriteExec in the insert plan should carry the catalog config" + ); + } + + fn plan_has_write_exec_with_config(plan: &Arc) -> bool { + if let Some(write) = plan.downcast_ref::() { + return write.catalog_config() == Some(&sample_catalog_config()); + } + plan.children() + .into_iter() + .any(plan_has_write_exec_with_config) + } + + #[tokio::test] + async fn test_scan_with_predicates_overrides_predicate() { + use iceberg::expr::Reference; + use iceberg::spec::Datum; + + let table = get_test_table_from_metadata_file().await; + let arrow_schema = + Arc::new(schema_to_arrow_schema(table.metadata().current_schema()).unwrap()); + + // Built with no filters, the scan carries no predicate. + let scan = IcebergTableScan::new(table, None, arrow_schema, None, &[], None); + assert!(scan.predicates().is_none()); + + // `with_predicates` installs one directly, as a distributed engine would + // after deserializing an already-built predicate. + let predicate = Reference::new("x").less_than(Datum::long(10)); + let scan = scan.with_predicates(Some(predicate.clone())); + assert_eq!(scan.predicates(), Some(&predicate)); + + // ...and can clear it again. + let scan = scan.with_predicates(None); + assert!(scan.predicates().is_none()); + } + + #[tokio::test] + async fn test_metadata_table_propagates_catalog_config() { + let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await; + + // With config: the metadata-table provider carries it. + let with_config = IcebergTableProvider::try_new_with_config( + catalog.clone(), + sample_catalog_config(), + namespace.clone(), + table_name.clone(), + ) + .await + .unwrap(); + let meta = with_config + .metadata_table(MetadataTableType::Snapshots) + .await + .unwrap(); + assert_eq!(meta.catalog_config(), Some(&sample_catalog_config())); + + // Without config: the metadata-table provider stays config-less. + let without_config = catalog_backed_provider(catalog, namespace, table_name).await; + let meta = without_config + .metadata_table(MetadataTableType::Snapshots) + .await + .unwrap(); + assert!(meta.catalog_config().is_none()); + } }