alexanderbianchi commented on code in PR #3000: URL: https://github.com/apache/iceberg-rust/pull/3000#discussion_r3974698726
########## crates/integrations/datafusion/src/catalog_provider.rs: ########## @@ -0,0 +1,335 @@ +// 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; +use std::sync::Arc; + +use datafusion::catalog::{CatalogProvider, SchemaProvider}; +use futures::future::try_join_all; +use iceberg::{Catalog, NamespaceIdent, Result, SessionCatalog, SessionContext}; + +use crate::catalog_adapter::SessionBindingCatalogAdapter; +use crate::schema_provider::IcebergSchemaProvider; + +/// Provides a DataFusion interface to schemas in an Iceberg [`Catalog`] or +/// [`SessionCatalog`]. +/// +/// Acts as a centralized catalog provider that aggregates +/// multiple [`SchemaProvider`], each associated with distinct namespaces. +#[derive(Debug)] +pub struct IcebergCatalogProvider { + /// A `HashMap` where keys are namespace names + /// and values are dynamic references to objects implementing the + /// [`SchemaProvider`] trait. + schemas: HashMap<String, Arc<dyn SchemaProvider>>, +} + +impl IcebergCatalogProvider { + /// Asynchronously constructs an [`IcebergCatalogProvider`] from a + /// [`Catalog`], fetching and initializing a schema provider for each + /// namespace. + /// + /// This method retrieves the namespace names and collects an initialized + /// schema provider for each namespace into a `HashMap`. + pub async fn try_new(catalog: Arc<dyn Catalog>) -> Result<Self> { + let session_binding_catalog = SessionBindingCatalogAdapter::new_without_context(catalog); + Self::try_new_with_binding_catalog(Arc::new(session_binding_catalog)).await + } + + /// Creates an [`IcebergCatalogProvider`] backed by a [`SessionCatalog`]. + /// + /// Each DataFusion session that has [`crate::IcebergOptions`] configured will + /// propagate an Iceberg [`SessionContext`] for scans and inserts. Provider + /// initialization, metadata-table lookup, table registration, and table + /// deregistration do not receive a DataFusion session; they share one + /// anonymous fallback context instead. + /// + /// Namespace and table discovery is performed once during construction and + /// shared by all DataFusion sessions. Catalogs with session-dependent + /// visibility must make the intended discovery set available to the + /// anonymous fallback; discovery is not repeated per DataFusion session. + pub async fn try_new_with_session_catalog(catalog: Arc<dyn SessionCatalog>) -> Result<Self> { + let shared_fallback_context = SessionContext::empty(); + let session_bound = SessionBindingCatalogAdapter::new(shared_fallback_context, catalog); + Self::try_new_with_binding_catalog(Arc::new(session_bound)).await + } + + async fn try_new_with_binding_catalog( + catalog: Arc<SessionBindingCatalogAdapter>, + ) -> Result<Self> { + // TODO: + // Schemas and providers should be cached and evicted based on time Review Comment: It seems like namespaces should also depend on the per-request auth. Calling list_namespaces now means that any namespace or table that is only accessible to certain tenants won't be discovered. It means any tenant aware catalog must either 1. fails during provider construction, or 2. must allow a shared service identity to discover a cross-tenant superset of namespaces, tables, and schemas. DataFusion’s `AsyncCatalogProviderList::resolve()` receives both the table references and SessionConfig, which may provide the required pre-resolution binding point without threading session rebinding through scan/insert implementations. Something like ```rust pub struct IcebergCatalogProvider { catalog: Arc<dyn SessionCatalog>, context: Arc<SessionContext>, } ``` ```rust async fn resolve( &self, references: &[TableReference], config: &SessionConfig, ) -> Result<Arc<dyn CatalogProviderList>> { let properties = SessionProperties::from_session_config(config)?; let context = // build auth context // let resolved = Arc::new(MemoryCatalogProviderList::new()); for catalog_name in referenced_iceberg_catalogs( references, config, &self.catalog_names, ) { let provider: Arc<dyn CatalogProvider> = Arc::new(IcebergCatalogProvider::new( Arc::clone(&self.shared_catalog), Arc::clone(&context), )); resolved.register_catalog(catalog_name, provider); } Ok(resolved) } ``` This makes the caching story a bit harder - we need to cache globally visible namespaces _and_ have a per-identity cache but that might be something better handled on the side of the REST catalog implementation anyways. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
