From e63bd8f48be41b9fe6068d019c52031c294080b3 Mon Sep 17 00:00:00 2001 From: sandudb Date: Fri, 11 Sep 2026 14:07:59 +0300 Subject: [PATCH 1/2] Added both catalog AND schema providers. Updated the testing suite to use in-memory catalog builder. --- Cargo.lock | 24 +++- iceberg/Cargo.toml | 2 + iceberg/src/catalog.rs | 49 +++++++ iceberg/src/lib.rs | 4 + iceberg/src/schema_provider.rs | 75 ++++++++++ iceberg/src/test_utils/harness.rs | 85 ++++++++++- iceberg/tests/catalog.rs | 230 ++++++++++++++++++++++++++++++ 7 files changed, 462 insertions(+), 7 deletions(-) create mode 100644 iceberg/src/catalog.rs create mode 100644 iceberg/src/schema_provider.rs create mode 100644 iceberg/tests/catalog.rs diff --git a/Cargo.lock b/Cargo.lock index 528f0a99c..da292a110 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2497,12 +2497,14 @@ version = "0.1.0" dependencies = [ "async-trait", "bytes", + "dashmap", "datafusion", "datafusion-distributed", "datafusion-proto", "delegate", "futures", "iceberg", + "iceberg-catalog-rest", "iceberg-storage-opendal", "insta", "parquet", @@ -4047,6 +4049,26 @@ dependencies = [ "zstd 0.13.3", ] +[[package]] +name = "iceberg-catalog-rest" +version = "0.10.1" +source = "git+https://github.com/apache/iceberg-rust.git?rev=4d83bc77dc10cff851a3ef427c451ff977bc3643#4d83bc77dc10cff851a3ef427c451ff977bc3643" +dependencies = [ + "async-trait", + "chrono", + "http 1.5.0", + "iceberg", + "itertools 0.13.0", + "reqwest 0.12.28", + "serde", + "serde_derive", + "serde_json", + "tokio", + "tracing", + "typed-builder", + "uuid", +] + [[package]] name = "iceberg-property-macro" version = "0.10.1" @@ -5859,7 +5881,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.119", diff --git a/iceberg/Cargo.toml b/iceberg/Cargo.toml index acb37ab62..a42bcfd25 100644 --- a/iceberg/Cargo.toml +++ b/iceberg/Cargo.toml @@ -27,8 +27,10 @@ delegate = "0.13" prost = "0.14.1" tokio = { version = "1.48", features = ["full"] } tokio-stream = "0.1" +dashmap = "6.0.1" [dev-dependencies] +iceberg-catalog-rest = { git = "https://github.com/apache/iceberg-rust.git", rev = "4d83bc77dc10cff851a3ef427c451ff977bc3643" } insta = "1" parquet = "59" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } diff --git a/iceberg/src/catalog.rs b/iceberg/src/catalog.rs new file mode 100644 index 000000000..ccfc0da15 --- /dev/null +++ b/iceberg/src/catalog.rs @@ -0,0 +1,49 @@ +// Iceberg's Catalog backed into DataFusion catalog + +use std::{collections::HashMap, sync::Arc}; + +use datafusion::{ + catalog::{CatalogProvider, SchemaProvider}, + error::Result, +}; +use futures::future::try_join_all; +use iceberg::{Catalog, Runtime}; + +use crate::{common::df_err, schema_provider::IcebergSchemaProvider}; + +#[derive(Debug)] +pub struct IcebergCatalog { + schemas: HashMap>, +} + +impl IcebergCatalog { + pub async fn try_new(client: Arc, iceberg_runtime: Runtime) -> Result { + let namespaces = client.list_namespaces(None).await.map_err(df_err)?; + + let schema_providers = try_join_all(namespaces.iter().map(|ns| { + IcebergSchemaProvider::try_new(client.clone(), ns.clone(), iceberg_runtime.clone()) + })) + .await?; + + let schemas: HashMap> = namespaces + .into_iter() + .zip(schema_providers) + .map(|(name, provider)| { + let key = name.as_ref().join("."); + (key, Arc::new(provider) as Arc) + }) + .collect(); + + Ok(Self { schemas }) + } +} + +impl CatalogProvider for IcebergCatalog { + fn schema(&self, name: &str) -> Option> { + self.schemas.get(name).cloned() + } + + fn schema_names(&self) -> Vec { + self.schemas.keys().cloned().collect() + } +} diff --git a/iceberg/src/lib.rs b/iceberg/src/lib.rs index 8879ea3db..f6ef48c1f 100644 --- a/iceberg/src/lib.rs +++ b/iceberg/src/lib.rs @@ -6,12 +6,14 @@ //! integration. It deliberately contains no distributed execution adaptation //! and no Iceberg write or commit implementation. +mod catalog; mod common; mod config; mod data_source; mod distributed_desired_task_count_handler; mod iceberg_ext; mod proto; +mod schema_provider; mod table_provider; mod work_unit_feed; mod work_unit_wire; @@ -20,12 +22,14 @@ mod codec; #[doc(hidden)] pub mod test_utils; +pub use catalog::IcebergCatalog; pub use codec::IcebergCodec; pub use config::IcebergConfig; pub use data_source::IcebergDataSource; pub use distributed_desired_task_count_handler::iceberg_desired_task_count; pub use iceberg_ext::IcebergExt; pub use iceberg_ext::IcebergIntegrationOptions; +pub use schema_provider::IcebergSchemaProvider; pub use table_provider::IcebergCatalogTableProvider; pub use table_provider::IcebergStaticTableProvider; pub use table_provider::IcebergTableProviderFactory; diff --git a/iceberg/src/schema_provider.rs b/iceberg/src/schema_provider.rs new file mode 100644 index 000000000..fd8eb809b --- /dev/null +++ b/iceberg/src/schema_provider.rs @@ -0,0 +1,75 @@ +use std::sync::Arc; + +use dashmap::DashMap; +use datafusion::{ + catalog::{SchemaProvider, TableProvider}, + error::Result, +}; +use futures::future::try_join_all; +use iceberg::{Catalog, NamespaceIdent, Runtime}; + +use crate::{IcebergCatalogTableProvider, common::df_err}; + +#[derive(Debug)] +pub struct IcebergSchemaProvider { + // Using Arc + DashMap for cheap clones of the tables in this schema + tables: Arc>>, +} + +impl IcebergSchemaProvider { + pub async fn try_new( + client: Arc, + namespace: NamespaceIdent, + iceberg_runtime: Runtime, + ) -> Result { + let table_names: Vec<_> = client + .list_tables(&namespace) + .await + .map_err(df_err)? + .into_iter() + .map(|ident| ident.name().to_string()) + .collect(); + + let table_providers = try_join_all( + table_names + .iter() + .map(|name| { + IcebergCatalogTableProvider::try_new( + client.clone(), + namespace.clone(), + name, + iceberg_runtime.clone(), + ) + }) + .collect::>(), + ) + .await?; + + let tables = Arc::new(DashMap::new()); + + // Getting a map of: + for (name, provider) in table_names.into_iter().zip(table_providers) { + tables.insert(name, Arc::new(provider)); + } + + Ok(IcebergSchemaProvider { tables }) + } +} + +#[async_trait::async_trait] +impl SchemaProvider for IcebergSchemaProvider { + async fn table(&self, name: &str) -> Result>> { + Ok(self + .tables + .get(name) + .map(|provider| provider.clone() as Arc)) + } + + fn table_names(&self) -> Vec { + self.tables.iter().map(|k| k.key().clone()).collect() + } + + fn table_exist(&self, name: &str) -> bool { + self.tables.contains_key(name) + } +} diff --git a/iceberg/src/test_utils/harness.rs b/iceberg/src/test_utils/harness.rs index 0520fa37c..815982256 100644 --- a/iceberg/src/test_utils/harness.rs +++ b/iceberg/src/test_utils/harness.rs @@ -25,19 +25,26 @@ use iceberg::io::{ FileMetadata, FileRead, FileWrite, InputFile, LocalFsStorage, OutputFile, Storage, StorageConfig, StorageFactory, }; +use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; use iceberg::spec::TableMetadata; -use iceberg::{Error, ErrorKind, Result as IcebergResult}; +use iceberg::{ + Catalog, CatalogBuilder, Error, ErrorKind, NamespaceIdent, Result as IcebergResult, TableIdent, +}; use serde::{Deserialize, Serialize}; use super::taxi_metadata; -use crate::{IcebergExt, IcebergIntegrationOptions, iceberg_desired_task_count}; +use crate::common::df_err; +use crate::{IcebergCatalog, IcebergExt, IcebergIntegrationOptions, iceberg_desired_task_count}; +pub const FIXTURE_CATALOG: &str = "iceberg"; +pub const FIXTURE_NAMESPACE: &str = "nyc"; pub const FIXTURE_URI: &str = "s3://iceberg-test/warehouse/taxi"; const WAREHOUSE_URI: &str = "s3://iceberg-test/warehouse/"; const FIXTURE_METADATA_URI: &str = "s3://iceberg-test/warehouse/taxi/metadata/v1.metadata.json"; pub struct IcebergTestHarness { ctx: SessionContext, + catalog: Option>, } impl IcebergTestHarness { @@ -61,6 +68,14 @@ impl IcebergTestHarness { self.ctx.sql(sql).await?.create_physical_plan().await } + /// Returns the in-memory Iceberg catalog backing a fixture built with + /// [`IcebergTestHarnessBuilder::with_catalog`]. + pub fn iceberg_catalog(&self) -> Result> { + self.catalog.clone().ok_or_else(|| { + DataFusionError::Plan("the fixture was not built with a catalog".to_string()) + }) + } + /// Returns the fixture scan without SQL optimization, including for an empty table. pub async fn scan(&self) -> Result> { self.ctx @@ -94,6 +109,7 @@ pub struct IcebergTestHarnessBuilder { metadata: TableMetadata, table_options: BTreeMap, files: HashMap>, + catalog: bool, #[cfg(feature = "integration")] workers: Option, } @@ -108,6 +124,7 @@ impl Default for IcebergTestHarnessBuilder { metadata: taxi_metadata(), table_options: BTreeMap::new(), files: HashMap::new(), + catalog: false, #[cfg(feature = "integration")] workers: None, } @@ -144,6 +161,14 @@ impl IcebergTestHarnessBuilder { self } + /// Registers the fixture through an in-memory Iceberg catalog exposed as + /// [`FIXTURE_CATALOG`].[`FIXTURE_NAMESPACE`] instead of `CREATE EXTERNAL TABLE`. + /// Table options are not applied in this mode. + pub fn with_catalog(mut self) -> Self { + self.catalog = true; + self + } + /// Enables distributed planning with logical workers backed by in-memory gRPC. #[cfg(feature = "integration")] pub fn with_workers(mut self, workers: usize) -> Self { @@ -162,12 +187,19 @@ impl IcebergTestHarnessBuilder { ..FixtureStorageFactory::default() }; let options = IcebergIntegrationOptions { - storage_factory: Arc::new(storage_factory), + storage_factory: Arc::new(storage_factory.clone()), iceberg_runtime: iceberg::Runtime::current(), }; - let state = self + let iceberg_runtime = options.iceberg_runtime.clone(); + let mut state = self .session_builder .with_iceberg_integration(options.clone()); + if self.catalog { + // Makes `taxi` resolve through the registered Iceberg catalog. + let config = state.config().get_or_insert_default(); + *config = std::mem::take(config) + .with_default_catalog_and_schema(FIXTURE_CATALOG, FIXTURE_NAMESPACE); + } #[cfg(feature = "integration")] let state = if let Some(workers) = self.workers { let resolver = @@ -186,6 +218,11 @@ impl IcebergTestHarnessBuilder { state }; let ctx = SessionContext::new_with_state(state.build()); + if self.catalog && !self.table_options.is_empty() { + return Err(DataFusionError::Plan( + "table options are not supported with a catalog-backed fixture".to_string(), + )); + } let mut statement = format!( "CREATE EXTERNAL TABLE taxi STORED AS ICEBERG \ LOCATION '{FIXTURE_METADATA_URI}'" @@ -205,8 +242,44 @@ impl IcebergTestHarnessBuilder { .join(", "); statement.push_str(&format!(" OPTIONS ({options})")); } - ctx.sql(&statement).await?.collect().await?; - Ok(IcebergTestHarness { ctx }) + + let mut iceberg_catalog = None; + if self.catalog { + let catalog = MemoryCatalogBuilder::default() + .with_storage_factory(Arc::new(storage_factory)) + .with_runtime(iceberg_runtime.clone()) + .load( + "memory", + HashMap::from([( + MEMORY_CATALOG_WAREHOUSE.to_string(), + WAREHOUSE_URI.to_string(), + )]), + ) + .await + .map_err(df_err)?; + let namespace = NamespaceIdent::new(FIXTURE_NAMESPACE.to_string()); + catalog + .create_namespace(&namespace, HashMap::new()) + .await + .map_err(df_err)?; + catalog + .register_table( + &TableIdent::new(namespace, "taxi".to_string()), + FIXTURE_METADATA_URI.to_string(), + ) + .await + .map_err(df_err)?; + let catalog: Arc = Arc::new(catalog); + let provider = IcebergCatalog::try_new(catalog.clone(), iceberg_runtime).await?; + ctx.register_catalog(FIXTURE_CATALOG, Arc::new(provider)); + iceberg_catalog = Some(catalog); + } else { + ctx.sql(&statement).await?.collect().await?; + } + Ok(IcebergTestHarness { + ctx, + catalog: iceberg_catalog, + }) } } diff --git a/iceberg/tests/catalog.rs b/iceberg/tests/catalog.rs new file mode 100644 index 000000000..d74ed2d02 --- /dev/null +++ b/iceberg/tests/catalog.rs @@ -0,0 +1,230 @@ +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use datafusion::catalog::CatalogProvider; + use datafusion::error::{DataFusionError, Result}; + use datafusion_distributed_iceberg::IcebergCatalog; + use datafusion_distributed_iceberg::test_utils::IcebergTestHarness; + use iceberg::{NamespaceIdent, Runtime}; + + #[tokio::test] + async fn exposes_namespaces_as_schemas() -> Result<()> { + let harness = catalog_harness().await?; + let (_, batches) = harness + .query( + "SELECT table_catalog, table_schema, table_name, table_type \ + FROM information_schema.tables \ + WHERE table_catalog = 'iceberg' AND table_schema <> 'information_schema'", + ) + .await?; + + insta::assert_snapshot!(batches, @" + +---------------+--------------+------------+------------+ + | table_catalog | table_schema | table_name | table_type | + +---------------+--------------+------------+------------+ + | iceberg | nyc | taxi | BASE TABLE | + +---------------+--------------+------------+------------+ + "); + Ok(()) + } + + #[tokio::test] + async fn resolves_a_fully_qualified_table() -> Result<()> { + let harness = catalog_harness().await?; + let (_, batches) = harness + .query("SELECT COUNT(*) AS trips FROM iceberg.nyc.taxi") + .await?; + + insta::assert_snapshot!(batches, @" + +--------+ + | trips | + +--------+ + | 175000 | + +--------+ + "); + Ok(()) + } + + #[tokio::test] + async fn resolves_a_table_through_the_default_catalog_and_schema() -> Result<()> { + let harness = catalog_harness().await?; + let (_, batches) = harness.query("DESCRIBE taxi").await?; + + insta::assert_snapshot!(batches, @" + +---------------------+---------------+-------------+ + | column_name | data_type | is_nullable | + +---------------------+---------------+-------------+ + | vendor_id | Int32 | YES | + | pickup_at | Timestamp(µs) | YES | + | dropoff_at | Timestamp(µs) | YES | + | passenger_count | Int64 | YES | + | trip_distance | Float64 | YES | + | pickup_location_id | Int32 | YES | + | dropoff_location_id | Int32 | YES | + | payment_type | Int64 | YES | + | fare_amount | Float64 | YES | + | tip_amount | Float64 | YES | + | tolls_amount | Float64 | YES | + | total_amount | Float64 | YES | + | pickup_date | Date32 | YES | + +---------------------+---------------+-------------+ + "); + Ok(()) + } + + #[tokio::test] + async fn scans_through_the_catalog_table_provider() -> Result<()> { + let harness = catalog_harness().await?; + let (plan, batches) = harness + .query( + "SELECT vendor_id, pickup_date FROM nyc.taxi \ + WHERE pickup_date = DATE '2024-01-10' LIMIT 3", + ) + .await?; + + insta::assert_snapshot!(plan + &batches, @" + CoalescePartitionsExec: fetch=3 + FilterExec: pickup_date@1 = 2024-01-10, fetch=3 + DataSourceExec: format=iceberg, projection=[vendor_id, pickup_date], predicate=pickup_date = 2024-01-10 + +-----------+-------------+ + | vendor_id | pickup_date | + +-----------+-------------+ + | 2 | 2024-01-10 | + | 1 | 2024-01-10 | + | 2 | 2024-01-10 | + +-----------+-------------+ + "); + Ok(()) + } + + #[tokio::test] + async fn rejects_an_unknown_table() -> Result<()> { + let harness = catalog_harness().await?; + let error = harness + .query("SELECT * FROM nyc.missing") + .await + .unwrap_err(); + + insta::assert_snapshot!(error.to_string(), @"Error during planning: table 'iceberg.nyc.missing' not found"); + Ok(()) + } + + #[tokio::test] + async fn rejects_an_unknown_namespace() -> Result<()> { + let harness = catalog_harness().await?; + let error = harness + .query("SELECT * FROM iceberg.missing.taxi") + .await + .unwrap_err(); + + insta::assert_snapshot!(error.to_string(), @"Error during planning: table 'iceberg.missing.taxi' not found"); + Ok(()) + } + + #[tokio::test] + async fn exposes_a_namespace_without_tables_as_an_empty_schema() -> Result<()> { + let harness = catalog_harness().await?; + let client = harness.iceberg_catalog()?; + client + .create_namespace(&NamespaceIdent::new("empty".to_string()), HashMap::new()) + .await + .map_err(|error| DataFusionError::External(Box::new(error)))?; + let catalog = IcebergCatalog::try_new(client, Runtime::current()).await?; + + let mut schema_names = catalog.schema_names(); + schema_names.sort(); + assert_eq!(schema_names, vec!["empty", "nyc"]); + let empty = catalog + .schema("empty") + .expect("empty namespace must resolve"); + assert!(empty.table_names().is_empty()); + Ok(()) + } + + #[tokio::test] + async fn rejects_registering_tables_through_sql() -> Result<()> { + let harness = catalog_harness().await?; + let error = harness + .query("CREATE TABLE nyc.trips (id INT)") + .await + .unwrap_err(); + + insta::assert_snapshot!(error.to_string(), @"Execution error: schema provider does not support registering tables"); + Ok(()) + } + + #[tokio::test] + async fn rejects_dropping_tables_through_sql() -> Result<()> { + let harness = catalog_harness().await?; + let error = harness.query("DROP TABLE nyc.taxi").await.unwrap_err(); + + // DataFusion reports any deregistration failure as a missing table. + insta::assert_snapshot!(error.to_string(), @"Execution error: Table 'nyc.taxi' doesn't exist."); + assert!( + harness + .query("SELECT 1 FROM nyc.taxi LIMIT 1") + .await + .is_ok() + ); + Ok(()) + } + + #[tokio::test] + async fn rejects_table_options_with_a_catalog() -> Result<()> { + let error = IcebergTestHarness::builder() + .with_catalog() + .with_table_option("iceberg.snapshot_id", "42") + .build() + .await + .err() + .expect("table options must be rejected with a catalog-backed fixture"); + + insta::assert_snapshot!(error.to_string(), @"Error during planning: table options are not supported with a catalog-backed fixture"); + Ok(()) + } + + #[cfg(feature = "integration")] + #[tokio::test] + async fn executes_a_catalog_table_across_workers() -> Result<()> { + let harness = IcebergTestHarness::builder() + .with_catalog() + .with_workers(2) + .build() + .await?; + // Grouping prevents COUNT(*) from being answered from snapshot metadata alone. + let (plan, batches) = harness + .query( + "SELECT vendor_id, COUNT(*) AS trips FROM iceberg.nyc.taxi \ + GROUP BY vendor_id ORDER BY vendor_id", + ) + .await?; + + insta::assert_snapshot!(plan + &batches, @" + SortPreservingMergeExec: [vendor_id@0 ASC NULLS LAST] + ProjectionExec: expr=[vendor_id@0 as vendor_id, count(Int64(1))@1 as trips] + SortExec: expr=[vendor_id@0 ASC NULLS LAST], preserve_partitioning=[true] + AggregateExec: mode=FinalPartitioned, gby=[vendor_id@0 as vendor_id], aggr=[count(Int64(1))] + RepartitionExec: partitioning=Hash([vendor_id@0], 4), input_partitions=4 + AggregateExec: mode=Partial, gby=[vendor_id@0 as vendor_id], aggr=[count(Int64(1))] + DataSourceExec: format=iceberg, projection=[vendor_id] + +-----------+--------+ + | vendor_id | trips | + +-----------+--------+ + | 1 | 45743 | + | 2 | 129212 | + | 6 | 45 | + +-----------+--------+ + "); + Ok(()) + } + + /// Builds a catalog-backed fixture with `information_schema` enabled. + async fn catalog_harness() -> Result { + let harness = IcebergTestHarness::builder().with_catalog().build().await?; + harness + .query("SET datafusion.catalog.information_schema = true") + .await?; + Ok(harness) + } +} From b8fbfee96ed75bfedb936f82e67e4703e4c69551 Mon Sep 17 00:00:00 2001 From: sandudb Date: Fri, 11 Sep 2026 15:54:10 +0300 Subject: [PATCH 2/2] Switched the naming to be more explicit --- iceberg/src/catalog.rs | 6 +++--- iceberg/src/schema_provider.rs | 6 +++--- iceberg/src/test_utils/harness.rs | 6 +++--- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/iceberg/src/catalog.rs b/iceberg/src/catalog.rs index ccfc0da15..b8a5ab32c 100644 --- a/iceberg/src/catalog.rs +++ b/iceberg/src/catalog.rs @@ -17,11 +17,11 @@ pub struct IcebergCatalog { } impl IcebergCatalog { - pub async fn try_new(client: Arc, iceberg_runtime: Runtime) -> Result { - let namespaces = client.list_namespaces(None).await.map_err(df_err)?; + pub async fn try_new(catalog: Arc, iceberg_runtime: Runtime) -> Result { + let namespaces = catalog.list_namespaces(None).await.map_err(df_err)?; let schema_providers = try_join_all(namespaces.iter().map(|ns| { - IcebergSchemaProvider::try_new(client.clone(), ns.clone(), iceberg_runtime.clone()) + IcebergSchemaProvider::try_new(catalog.clone(), ns.clone(), iceberg_runtime.clone()) })) .await?; diff --git a/iceberg/src/schema_provider.rs b/iceberg/src/schema_provider.rs index fd8eb809b..de1b20a18 100644 --- a/iceberg/src/schema_provider.rs +++ b/iceberg/src/schema_provider.rs @@ -18,11 +18,11 @@ pub struct IcebergSchemaProvider { impl IcebergSchemaProvider { pub async fn try_new( - client: Arc, + catalog: Arc, namespace: NamespaceIdent, iceberg_runtime: Runtime, ) -> Result { - let table_names: Vec<_> = client + let table_names: Vec<_> = catalog .list_tables(&namespace) .await .map_err(df_err)? @@ -35,7 +35,7 @@ impl IcebergSchemaProvider { .iter() .map(|name| { IcebergCatalogTableProvider::try_new( - client.clone(), + catalog.clone(), namespace.clone(), name, iceberg_runtime.clone(), diff --git a/iceberg/src/test_utils/harness.rs b/iceberg/src/test_utils/harness.rs index 815982256..a42cce0c2 100644 --- a/iceberg/src/test_utils/harness.rs +++ b/iceberg/src/test_utils/harness.rs @@ -44,7 +44,7 @@ const FIXTURE_METADATA_URI: &str = "s3://iceberg-test/warehouse/taxi/metadata/v1 pub struct IcebergTestHarness { ctx: SessionContext, - catalog: Option>, + iceberg_catalog: Option>, } impl IcebergTestHarness { @@ -71,7 +71,7 @@ impl IcebergTestHarness { /// Returns the in-memory Iceberg catalog backing a fixture built with /// [`IcebergTestHarnessBuilder::with_catalog`]. pub fn iceberg_catalog(&self) -> Result> { - self.catalog.clone().ok_or_else(|| { + self.iceberg_catalog.clone().ok_or_else(|| { DataFusionError::Plan("the fixture was not built with a catalog".to_string()) }) } @@ -278,7 +278,7 @@ impl IcebergTestHarnessBuilder { } Ok(IcebergTestHarness { ctx, - catalog: iceberg_catalog, + iceberg_catalog, }) } }