Repository navigation
Draft: resolve Iceberg session catalogs per DataFusion query - #1
alexanderbianchi wants to merge 5 commits into
Conversation
d764ad2 to
2ec37da
Compare
| /// The returned ordinary [`Catalog`] forwards every operation with `context`. | ||
| /// Binding performs no catalog I/O, and clones of `inner` may be bound to | ||
| /// independent contexts. | ||
| pub fn bind_session_catalog( |
There was a problem hiding this comment.
Nice I think this may be useful for usptream too! API-wise, I've also played with this but a pub fn might be simpler
impl dyn SessionCatalog {
/// Bind this catalog to a session, exposing the ordinary Catalog API.
///
/// # Example
/// ```
/// # fn into_catalog(session_catalog: Arc<dyn SessionCatalog>, id: String) {
/// let session = SessionContext::builder().session_id(id).build();
///
/// // Use the plain catalog API for the duration of this session.
/// let catalog = session_catalog.into_catalog(session);
/// # let _ = catalog;
/// # }
/// ```
pub fn into_catalog(self: Arc<Self>, session: SessionContext) -> Arc<dyn Catalog> {
Arc::new(SessionBoundCatalog {
inner: self,
session,
})
}
}There was a problem hiding this comment.
What I was trying to play with was is there was a way to kinda work around datafusion's (limited) catalog API and just have it handled in the service layer. apache#3081 is still going to be needed from what I can tell. There is this trait
pub trait AsyncCatalogProviderList: Send + Sync {
// Required method
fn catalog<'life0, 'life1, 'async_trait>(
&'life0 self,
name: &'life1 str,
) -> Pin<Box<dyn Future<Output = Result<Option<Arc<dyn AsyncCatalogProvider>>, DataFusionError>> + Send + 'async_trait>>
where 'life0: 'async_trait,
'life1: 'async_trait,
Self: 'async_trait;
// Provided method
fn resolve<'life0, 'life1, 'life2, 'async_trait>(
&'life0 self,
references: &'life1 [TableReference],
config: &'life2 SessionConfig,
) -> Pin<Box<dyn Future<Output = Result<Arc<dyn CatalogProviderList>, DataFusionError>> + Send + 'async_trait>>
where 'life0: 'async_trait,
'life1: 'async_trait,
'life2: 'async_trait,
Self: 'async_trait { ... }
}which I think we could use build the "bound" catalog.
2ec37da to
1ad09a8
Compare
| } | ||
|
|
||
| #[async_trait::async_trait] | ||
| impl AsyncCatalogProviderList for IcebergSessionCatalogProviderList { |
There was a problem hiding this comment.
https://docs.rs/datafusion/latest/datafusion/catalog/trait.AsyncCatalogProvider.html
maybe should just do the provider not the List
| #[derive(Clone, Debug)] | ||
| pub(crate) enum CatalogAccess { | ||
| Plain(Arc<dyn Catalog>), | ||
| Session { |
There was a problem hiding this comment.
Hm... At first I didn't like the enum but it's growing on me.
| #[async_trait] | ||
| impl SchemaProvider for IcebergSchemaProvider { | ||
| fn table_names(&self) -> Vec<String> { | ||
| self.tables |
There was a problem hiding this comment.
Lots of stripping of the schema provider here isn't great
1ad09a8 to
6094367
Compare
| /// produced by [`crate::IcebergSessionCatalogProvider`] to retain the exact | ||
| /// [`SessionContext`] supplied by a DataFusion query. | ||
| #[derive(Clone, Debug)] | ||
| pub(crate) enum CatalogAccess { |
There was a problem hiding this comment.
I think this might be 1 extra level of indirection I think we could probably do something closer to the actual datafusion catalog provider
| } | ||
|
|
||
| #[derive(Clone, Copy)] | ||
| enum TransactionCatalog<'a> { |
There was a problem hiding this comment.
I don't want to have to touch this file
|
|
||
| pub(crate) async fn load_table(&self, ident: &TableIdent) -> Result<Table> { | ||
| match self { | ||
| Self::Plain(catalog) => catalog.load_table(ident).await, |
There was a problem hiding this comment.
why do we ever have to load table after async provider
There was a problem hiding this comment.
AI is informing me it's becuase the table provider currently reloads the table whenever it's called so table provider are setup to refresh. THis... seems silly to me. SHould be table provider per request
Purpose
Draft a query-scoped alternative to the scan/insert-time rebinding in apache/iceberg-rust#3000.
This personal-fork PR is based directly on Apache iceberg-rust
mainand contains no commits from apache#3000.Model
Applications place Iceberg's existing
Arc<iceberg::SessionContext>directly into DataFusion's typedSessionConfigextensions:One shared resolver owns the session-aware catalog:
Before SQL/Substrait planning, the caller resolves this catalog's references:
resolveretrieves the exactArc<SessionContext>fromSessionConfig, loads only the referenced tables under that immutable context, and returns an ordinary query-scopedCatalogProvider. The same context reaches initial table loading, metadata-table loading, scan refresh, insert loading, and commit/update.Catalog-name routing remains the caller's responsibility. An application with multiple catalogs can register the returned provider in its query-local ordinary
CatalogProviderList. DataFusion does not automatically invoke async providers.Preserve the existing plain provider
The existing
IcebergCatalogProvider::try_new(Arc<dyn Catalog>)and its eager namespace/table listing behavior are unchanged. Existingschema_names,table_names,table_exist, registration, and deregistration behavior and tests are preserved.The new session-aware path binds the request context into private
AsyncCatalogProvider/AsyncSchemaProviderimplementations, then delegates grouping, deduplication, and immutable query-local cache construction to DataFusion's defaultAsyncCatalogProvider::resolve. No mutable custom provider maps are maintained by this integration.Shared table and commit implementation
The existing
IcebergCatalogProviderremains plain-only and unchanged.A single private
IcebergTableSourcerepresents the two valid table-provider lifetimes:This collapses the former independent catalog-access and metadata-lifetime modes, making invalid combinations unrepresentable. It deliberately does not implement
Catalog, so there is noSessionCatalog + SessionContext -> Catalogadapter.Existing plain providers continue refreshing metadata on every scan/write. Query-resolved session providers retain and reuse the exact
Tableloaded during async resolution for metadata tables, scans, and insert planning. This avoids redundant requests and prevents schema/metadata drift between logical and physical planning.Transaction::commitand the newTransaction::commit_with_sessionare thin entry points into one shared private transaction algorithm. Refresh, retries, action application, requirements, updates, and final commit construction are identical; only contextualload_tableandupdate_tabledispatch differs.No persistent session cache
The session-aware resolver does not enumerate all namespaces/tables and does not retain cross-query providers or metadata. Persistent cache scope, authorization keys freshness, invalidation, and vended credentials remain the responsibility of
SessionCatalogusers and concrete REST/catalog implementations.DataFusion trait behavior
IcebergSessionCatalogProviderimplementsAsyncCatalogProvider, which represents one catalog. Itsresolve(references, config, catalog_name)is the authenticated entry point.The required
schema()method cannot safely resolve a session catalog because DataFusion does not passSessionConfigto it. It therefore reports thatresolve()is required rather than introducing an anonymous fallback. References belonging to another catalog return an empty resolved provider without requiring an Iceberg context.Authentication boundary
This change propagates Iceberg
SessionContext; it does not turn credentials into HTTP headers. The contextual REST authentication hook proposed in #3081, or an equivalent context-aware REST implementation, remains required.With apache#3081:
SessionContextcredentials;/v1/configuses service/bootstrap authentication;No application-specific JWT, attribution, or organization fields are introduced in OSS.
Validation
cargo test -p iceberg-datafusion: 94 unit tests + 9 integration tests + 1 doctest passed; 5 existing doctests ignored.cargo clippy -p iceberg-datafusion --all-targets -- -D warnings: passed.cargo fmt -p iceberg-datafusion -- --check: passed.cargo public-api -p iceberg-datafusion --all-features -ss: matches the updated snapshot.New tests verify:
SessionContextfails before catalog I/O instead of falling back anonymously;Arc<SessionContext>installed by the application;update_table;SessionCatalogdo not mix identities;Scope and limitations
TableReferences, callresolve, and install the returned provider before planning.Change size
Production Rust changes are approximately
+334/-44. Existing plain catalog/schema behavior remains intact; the new code adds DataFusion-native async resolution, one table-source lifecycle abstraction, and shared contextual transaction dispatch. The remaining lines are tests and generated API output.AI disclosure
Implemented and validated with an AI coding assistant; all commands reported above were run locally.