Skip to content

refactor: move catalog traits to session crate #23703

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
timsaucer wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

refactor: move catalog traits to session crate #23703

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

193 changes: 1 addition & 192 deletions datafusion/catalog/src/catalog.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,195 +15,4 @@
// specific language governing permissions and limitations
// under the License.

use std::any::Any;
use std::fmt::Debug;
use std::sync::Arc;

pub use crate::schema::SchemaProvider;
use datafusion_common::Result;
use datafusion_common::not_impl_err;

/// Represents a catalog, comprising a number of named schemas.
///
/// # Catalog Overview
///
/// To plan and execute queries, DataFusion needs a "Catalog" that provides
/// metadata such as which schemas and tables exist, their columns and data
/// types, and how to access the data.
///
/// The Catalog API consists:
/// * [`CatalogProviderList`]: a collection of `CatalogProvider`s
/// * [`CatalogProvider`]: a collection of `SchemaProvider`s (sometimes called a "database" in other systems)
/// * [`SchemaProvider`]: a collection of `TableProvider`s (often called a "schema" in other systems)
/// * [`TableProvider`]: individual tables
///
/// # Implementing Catalogs
///
/// To implement a catalog, you implement at least one of the [`CatalogProviderList`],
/// [`CatalogProvider`] and [`SchemaProvider`] traits and register them
/// appropriately in the `SessionContext`.
///
/// DataFusion comes with a simple in-memory catalog implementation,
/// `MemoryCatalogProvider`, that is used by default and has no persistence.
/// DataFusion does not include more complex Catalog implementations because
/// catalog management is a key design choice for most data systems, and thus
/// it is unlikely that any general-purpose catalog implementation will work
/// well across many use cases.
///
/// # Implementing "Remote" catalogs
///
/// See [`remote_catalog`] for an end to end example of how to implement a
/// remote catalog.
///
/// Sometimes catalog information is stored remotely and requires a network call
/// to retrieve. For example, the [Delta Lake] table format stores table
/// metadata in files on S3 that must be first downloaded to discover what
/// schemas and tables exist.
///
/// [Delta Lake]: https://delta.io/
/// [`remote_catalog`]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/data_io/remote_catalog.rs
///
/// The [`CatalogProvider`] can support this use case, but it takes some care.
/// The planning APIs in DataFusion are not `async` and thus network IO can not
/// be performed "lazily" / "on demand" during query planning. The rationale for
/// this design is that using remote procedure calls for all catalog accesses
/// required for query planning would likely result in multiple network calls
/// per plan, resulting in very poor planning performance.
///
/// To implement [`CatalogProvider`] and [`SchemaProvider`] for remote catalogs,
/// you need to provide an in memory snapshot of the required metadata. Most
/// systems typically either already have this information cached locally or can
/// batch access to the remote catalog to retrieve multiple schemas and tables
/// in a single network call.
///
/// Note that [`SchemaProvider::table`] **is** an `async` function in order to
/// simplify implementing simple [`SchemaProvider`]s. For many table formats it
/// is easy to list all available tables but there is additional non trivial
/// access required to read table details (e.g. statistics).
///
/// The pattern that DataFusion itself uses to plan SQL queries is to walk over
/// the query to find all table references, performing required remote catalog
/// lookups in parallel, storing the results in a cached snapshot, and then plans
/// the query using that snapshot.
///
/// # Example Catalog Implementations
///
/// Here are some examples of how to implement custom catalogs:
///
/// * [`datafusion-cli`]: [`DynamicFileCatalogProvider`] catalog provider
/// that treats files and directories on a filesystem as tables.
///
/// * The [`catalog.rs`]: a simple directory based catalog.
///
/// * [delta-rs]: [`UnityCatalogProvider`] implementation that can
/// read from Delta Lake tables
///
/// [`datafusion-cli`]: https://datafusion.apache.org/user-guide/cli/index.html
/// [`DynamicFileCatalogProvider`]: https://github.com/apache/datafusion/blob/31b9b48b08592b7d293f46e75707aad7dadd7cbc/datafusion-cli/src/catalog.rs#L75
/// [`catalog.rs`]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/data_io/catalog.rs
/// [delta-rs]: https://github.com/delta-io/delta-rs
/// [`UnityCatalogProvider`]: https://github.com/delta-io/delta-rs/blob/951436ecec476ce65b5ed3b58b50fb0846ca7b91/crates/deltalake-core/src/data_catalog/unity/datafusion.rs#L111-L123
///
/// [`TableProvider`]: crate::TableProvider
pub trait CatalogProvider: Any + Debug + Sync + Send {
/// Retrieves the list of available schema names in this catalog.
fn schema_names(&self) -> Vec<String>;

/// Retrieves a specific schema from the catalog by name, provided it exists.
fn schema(&self, name: &str) -> Option<Arc<dyn SchemaProvider>>;

/// Adds a new schema to this catalog.
///
/// If a schema of the same name existed before, it is replaced in
/// the catalog and returned.
///
/// By default returns a "Not Implemented" error
fn register_schema(
&self,
name: &str,
schema: Arc<dyn SchemaProvider>,
) -> Result<Option<Arc<dyn SchemaProvider>>> {
// use variables to avoid unused variable warnings
let _ = name;
let _ = schema;
not_impl_err!("Registering new schemas is not supported")
}

/// Removes a schema from this catalog. Implementations of this method should return
/// errors if the schema exists but cannot be dropped. For example, in DataFusion's
/// default in-memory catalog, `MemoryCatalogProvider`, a non-empty schema
/// will only be successfully dropped when `cascade` is true.
/// This is equivalent to how DROP SCHEMA works in PostgreSQL.
///
/// Implementations of this method should return None if schema with `name`
/// does not exist.
///
/// By default returns a "Not Implemented" error
fn deregister_schema(
&self,
_name: &str,
_cascade: bool,
) -> Result<Option<Arc<dyn SchemaProvider>>> {
not_impl_err!("Deregistering new schemas is not supported")
}
}

impl dyn CatalogProvider {
/// Returns `true` if the catalog provider is of type `T`.
///
/// Prefer this over `downcast_ref::<T>().is_some()`. Works correctly when
/// called on `Arc<dyn CatalogProvider>` via auto-deref.
pub fn is<T: CatalogProvider>(&self) -> bool {
(self as &dyn Any).is::<T>()
}

/// Attempts to downcast this catalog provider to a concrete type `T`,
/// returning `None` if the provider is not of that type.
///
/// Works correctly when called on `Arc<dyn CatalogProvider>` via auto-deref,
/// unlike `(&arc as &dyn Any).downcast_ref::<T>()` which would attempt to
/// downcast the `Arc` itself.
pub fn downcast_ref<T: CatalogProvider>(&self) -> Option<&T> {
(self as &dyn Any).downcast_ref()
}
}

/// Represent a list of named [`CatalogProvider`]s.
///
/// Please see the documentation on [`CatalogProvider`] for details of
/// implementing a custom catalog.
pub trait CatalogProviderList: Any + Debug + Sync + Send {
/// Adds a new catalog to this catalog list
/// If a catalog of the same name existed before, it is replaced in the list and returned.
fn register_catalog(
&self,
name: String,
catalog: Arc<dyn CatalogProvider>,
) -> Option<Arc<dyn CatalogProvider>>;

/// Retrieves the list of available catalog names
fn catalog_names(&self) -> Vec<String>;

/// Retrieves a specific catalog by name, provided it exists.
fn catalog(&self, name: &str) -> Option<Arc<dyn CatalogProvider>>;
}

impl dyn CatalogProviderList {
/// Returns `true` if the catalog provider list is of type `T`.
///
/// Prefer this over `downcast_ref::<T>().is_some()`. Works correctly when
/// called on `Arc<dyn CatalogProviderList>` via auto-deref.
pub fn is<T: CatalogProviderList>(&self) -> bool {
(self as &dyn Any).is::<T>()
}

/// Attempts to downcast this catalog provider list to a concrete type `T`,
/// returning `None` if the provider list is not of that type.
///
/// Works correctly when called on `Arc<dyn CatalogProviderList>` via
/// auto-deref, unlike `(&arc as &dyn Any).downcast_ref::<T>()` which would
/// attempt to downcast the `Arc` itself.
pub fn downcast_ref<T: CatalogProviderList>(&self) -> Option<&T> {
(self as &dyn Any).downcast_ref()
}
}
pub use datafusion_session::{CatalogProvider, CatalogProviderList};
5 changes: 0 additions & 5 deletions datafusion/catalog/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,3 @@ pub use memory::{
};
pub use schema::*;
pub use table::*;

// For backwards compatibility,
mod session {
pub use datafusion_session::Session;
}
91 changes: 1 addition & 90 deletions datafusion/catalog/src/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,93 +15,4 @@
// specific language governing permissions and limitations
// under the License.

//! Describes the interface and built-in implementations of schemas,
//! representing collections of named tables.

use async_trait::async_trait;
use datafusion_common::{DataFusionError, exec_err};
use std::any::Any;
use std::fmt::Debug;
use std::sync::Arc;

use crate::table::TableProvider;
use datafusion_common::Result;
use datafusion_expr::TableType;

/// Represents a schema, comprising a number of named tables.
///
/// Please see [`CatalogProvider`] for details of implementing a custom catalog.
///
/// [`CatalogProvider`]: super::CatalogProvider
#[async_trait]
pub trait SchemaProvider: Any + Debug + Sync + Send {
/// Returns the owner of the Schema, default is None. This value is reported
/// as part of `information_tables.schemata
fn owner_name(&self) -> Option<&str> {
None
}

/// Retrieves the list of available table names in this schema.
fn table_names(&self) -> Vec<String>;

/// Retrieves a specific table from the schema by name, if it exists,
/// otherwise returns `None`.
async fn table(
&self,
name: &str,
) -> Result<Option<Arc<dyn TableProvider>>, DataFusionError>;

/// Retrieves the type of a specific table from the schema by name, if it exists, otherwise
/// returns `None`. Implementations for which this operation is cheap but [Self::table] is
/// expensive can override this to improve operations that only need the type, e.g.
/// `SELECT * FROM information_schema.tables`.
async fn table_type(&self, name: &str) -> Result<Option<TableType>> {
self.table(name).await.map(|o| o.map(|t| t.table_type()))
}

/// If supported by the implementation, adds a new table named `name` to
/// this schema.
///
/// If a table of the same name was already registered, returns "Table
/// already exists" error.
#[expect(unused_variables)]
fn register_table(
&self,
name: String,
table: Arc<dyn TableProvider>,
) -> Result<Option<Arc<dyn TableProvider>>> {
exec_err!("schema provider does not support registering tables")
}

/// If supported by the implementation, removes the `name` table from this
/// schema and returns the previously registered [`TableProvider`], if any.
///
/// If no `name` table exists, returns Ok(None).
#[expect(unused_variables)]
fn deregister_table(&self, name: &str) -> Result<Option<Arc<dyn TableProvider>>> {
exec_err!("schema provider does not support deregistering tables")
}

/// Returns true if table exist in the schema provider, false otherwise.
fn table_exist(&self, name: &str) -> bool;
}

impl dyn SchemaProvider {
/// Returns `true` if the schema provider is of type `T`.
///
/// Prefer this over `downcast_ref::<T>().is_some()`. Works correctly when
/// called on `Arc<dyn SchemaProvider>` via auto-deref.
pub fn is<T: SchemaProvider>(&self) -> bool {
(self as &dyn Any).is::<T>()
}

/// Attempts to downcast this schema provider to a concrete type `T`,
/// returning `None` if the provider is not of that type.
///
/// Works correctly when called on `Arc<dyn SchemaProvider>` via auto-deref,
/// unlike `(&arc as &dyn Any).downcast_ref::<T>()` which would attempt to
/// downcast the `Arc` itself.
pub fn downcast_ref<T: SchemaProvider>(&self) -> Option<&T> {
(self as &dyn Any).downcast_ref()
}
}
pub use datafusion_session::SchemaProvider;
Loading
Loading