You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/23 13:36:37 UTC
(datafusion) branch main updated: refactor: move catalog traits to session crate (#23703)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 1bdff19402 refactor: move catalog traits to session crate (#23703)
1bdff19402 is described below
commit 1bdff19402b62bad04f6c9cc8d5817f546a2fd7d
Author: Tim Saucer <ti...@gmail.com>
AuthorDate: Thu Jul 23 09:25:45 2026 -0400
refactor: move catalog traits to session crate (#23703)
## Which issue does this PR close?
- Addresses one portion of #23678.
## Rationale for this change
This will enable users to create table views using table function over
an existing table in the catalog via FFI. We are currently unable to
access the catalog other than by downcasting a `Session` to
`SessionState`, which is not effective over FFI.
In order to support this, we need to move many trait definitions out of
the `datafusion-catalog` crate and into the `datafusion-session` crate.
There is more discussion in #23678 about why this is necessary.
## What changes are included in this PR?
- Move a variety of catalog traits into `datafusion-session`.
- Add `catalog_list()` function to `Session`.
- Implement `catalog_list()` in `ForeignSession` for FFI users
## Are these changes tested?
New tests are included.
## Are there any user-facing changes?
User guide is updated.
---------
Co-authored-by: Claude Opus 4.8 (1M context) <no...@anthropic.com>
---
Cargo.lock | 1 +
datafusion/catalog/src/catalog.rs | 197 +------
datafusion/catalog/src/lib.rs | 10 +-
datafusion/catalog/src/schema.rs | 92 +--
datafusion/catalog/src/table.rs | 626 +--------------------
.../core/src/datasource/listing_table_factory.rs | 4 +
datafusion/core/src/execution/session_state.rs | 8 +
datafusion/datasource-arrow/src/file_format.rs | 5 +
datafusion/datasource/src/url.rs | 5 +
datafusion/ffi/src/session/mod.rs | 38 +-
datafusion/ffi/src/tests/async_provider.rs | 21 +-
datafusion/ffi/tests/ffi_integration.rs | 9 +
datafusion/session/Cargo.toml | 1 +
datafusion/session/README.md | 2 +-
datafusion/{catalog => session}/src/catalog.rs | 37 ++
datafusion/session/src/lib.rs | 28 +-
datafusion/{catalog => session}/src/schema.rs | 2 +-
datafusion/session/src/session.rs | 5 +
datafusion/{catalog => session}/src/table.rs | 8 +-
docs/source/library-user-guide/upgrading/55.0.0.md | 36 ++
20 files changed, 210 insertions(+), 925 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 566cc11668..ed63f60b41 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2581,6 +2581,7 @@ dependencies = [
name = "datafusion-session"
version = "54.1.0"
dependencies = [
+ "arrow-schema",
"async-trait",
"datafusion-common",
"datafusion-execution",
diff --git a/datafusion/catalog/src/catalog.rs b/datafusion/catalog/src/catalog.rs
index 34cdf74440..07da1293a7 100644
--- a/datafusion/catalog/src/catalog.rs
+++ b/datafusion/catalog/src/catalog.rs
@@ -15,195 +15,8 @@
// 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()
- }
-}
+// Re-export from this module for backwards compatibility.
+pub use datafusion_session::{CatalogProvider, CatalogProviderList};
+// Re-export so users can access this type through `datafusion_catalog` and
+// `datafusion::catalog` without depending directly on `datafusion_session`.
+pub use datafusion_session::EmptyCatalogProviderList;
diff --git a/datafusion/catalog/src/lib.rs b/datafusion/catalog/src/lib.rs
index 33d54b7cb8..815bfe32fa 100644
--- a/datafusion/catalog/src/lib.rs
+++ b/datafusion/catalog/src/lib.rs
@@ -25,7 +25,10 @@
#![cfg_attr(not(test), deny(clippy::clone_on_ref_ptr))]
#![cfg_attr(test, allow(clippy::needless_pass_by_value))]
-//! Interfaces and default implementations of catalogs and schemas.
+//! Default implementations of catalogs and schemas.
+//!
+//! The catalog interfaces are defined in [`datafusion_session`] and re-exported
+//! by this crate.
//!
//! Implementations
//! * Information schema: [`information_schema`]
@@ -57,8 +60,3 @@ pub use memory::{
};
pub use schema::*;
pub use table::*;
-
-// For backwards compatibility,
-mod session {
- pub use datafusion_session::Session;
-}
diff --git a/datafusion/catalog/src/schema.rs b/datafusion/catalog/src/schema.rs
index d99027593c..40b20caeb9 100644
--- a/datafusion/catalog/src/schema.rs
+++ b/datafusion/catalog/src/schema.rs
@@ -15,93 +15,5 @@
// 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()
- }
-}
+// Re-export from this module for backwards compatibility.
+pub use datafusion_session::SchemaProvider;
diff --git a/datafusion/catalog/src/table.rs b/datafusion/catalog/src/table.rs
index c6468fd5ad..2a10efbdcc 100644
--- a/datafusion/catalog/src/table.rs
+++ b/datafusion/catalog/src/table.rs
@@ -15,626 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-use std::any::Any;
-use std::borrow::Cow;
-use std::fmt::Debug;
-use std::sync::Arc;
-
-use crate::session::Session;
-use arrow::datatypes::SchemaRef;
-use async_trait::async_trait;
-use datafusion_common::{Constraints, Statistics, not_impl_err};
-use datafusion_common::{Result, internal_err};
-use datafusion_expr::Expr;
-use datafusion_expr::statistics::StatisticsRequest;
-
-use datafusion_expr::dml::InsertOp;
-use datafusion_expr::{
- CreateExternalTable, LogicalPlan, TableProviderFilterPushDown, TableType,
+// Re-export from this module for backwards compatibility.
+pub use datafusion_session::{
+ ScanArgs, ScanResult, TableFunction, TableFunctionArgs, TableFunctionImpl,
+ TableProvider, TableProviderFactory,
};
-use datafusion_physical_plan::ExecutionPlan;
-
-/// A table which can be queried and modified.
-///
-/// Please see [`CatalogProvider`] for details of implementing a custom catalog.
-///
-/// [`TableProvider`] represents a source of data which can provide data as
-/// Apache Arrow [`RecordBatch`]es. Implementations of this trait provide
-/// important information for planning such as:
-///
-/// 1. [`Self::schema`]: The schema (columns and their types) of the table
-/// 2. [`Self::supports_filters_pushdown`]: Should filters be pushed into this scan
-/// 2. [`Self::scan`]: An [`ExecutionPlan`] that can read data
-///
-/// [`RecordBatch`]: https://docs.rs/arrow/latest/arrow/record_batch/struct.RecordBatch.html
-/// [`CatalogProvider`]: super::CatalogProvider
-#[async_trait]
-pub trait TableProvider: Any + Debug + Sync + Send {
- /// Get a reference to the schema for this table
- fn schema(&self) -> SchemaRef;
-
- /// Get a reference to the constraints of the table.
- /// Returns:
- /// - `None` for tables that do not support constraints.
- /// - `Some(&Constraints)` for tables supporting constraints.
- /// Therefore, a `Some(&Constraints::empty())` return value indicates that
- /// this table supports constraints, but there are no constraints.
- fn constraints(&self) -> Option<&Constraints> {
- None
- }
-
- /// Get the type of this table for metadata/catalog purposes.
- fn table_type(&self) -> TableType;
-
- /// Get the create statement used to create this table, if available.
- fn get_table_definition(&self) -> Option<&str> {
- None
- }
-
- /// Get the [`LogicalPlan`] of this table, if available.
- fn get_logical_plan(&'_ self) -> Option<Cow<'_, LogicalPlan>> {
- None
- }
-
- /// Get the default value for a column, if available.
- fn get_column_default(&self, _column: &str) -> Option<&Expr> {
- None
- }
-
- /// Create an [`ExecutionPlan`] for scanning the table with optional
- /// `projection`, `filter`, and `limit`, described below.
- ///
- /// The returned `ExecutionPlan` is responsible for scanning the datasource's
- /// partitions in a streaming, parallelized fashion.
- ///
- /// # Projection
- ///
- /// If specified, only a subset of columns should be returned, in the order
- /// specified. The projection is a set of indexes of the fields in
- /// [`Self::schema`].
- ///
- /// DataFusion provides the projection so the scan reads only the columns
- /// actually used in the query, an optimization called "Projection
- /// Pushdown". Some datasources, such as Parquet, can use this information
- /// to go significantly faster when only a subset of columns is required.
- ///
- /// # Filters
- ///
- /// A list of boolean filter [`Expr`]s to evaluate *during* the scan, in the
- /// manner specified by [`Self::supports_filters_pushdown`]. Only rows for
- /// which *all* of the `Expr`s evaluate to `true` must be returned (that is,
- /// the expressions are `AND`ed together).
- ///
- /// To enable filter pushdown, override
- /// [`Self::supports_filters_pushdown`]. The default implementation does not
- /// push down filters, and `filters` will be empty.
- ///
- /// DataFusion pushes filters into scans whenever possible ("Filter
- /// Pushdown"). Depending on the data format and implementation, evaluating
- /// predicates during the scan can significantly improve performance.
- ///
- /// ## Note: Some columns may appear *only* in Filters
- ///
- /// In some cases, a query may use a column only in a filter and the
- /// projection will not contain all columns referenced by the filter
- /// expressions.
- ///
- /// For example, given the query `SELECT t.a FROM t WHERE t.b > 5`,
- ///
- /// ```text
- /// ┌────────────────────┐
- /// │ Projection(t.a) │
- /// └────────────────────┘
- /// ▲
- /// │
- /// │
- /// ┌────────────────────┐ Filter ┌────────────────────┐ Projection ┌────────────────────┐
- /// │ Filter(t.b > 5) │────Pushdown──▶ │ Projection(t.a) │ ───Pushdown───▶ │ Projection(t.a) │
- /// └────────────────────┘ └────────────────────┘ └────────────────────┘
- /// ▲ ▲ ▲
- /// │ │ │
- /// │ │ ┌────────────────────┐
- /// ┌────────────────────┐ ┌────────────────────┐ │ Scan │
- /// │ Scan │ │ Scan │ │ filter=(t.b > 5) │
- /// └────────────────────┘ │ filter=(t.b > 5) │ │ projection=(t.a) │
- /// └────────────────────┘ └────────────────────┘
- ///
- /// Initial Plan If `TableProviderFilterPushDown` Projection pushdown notes that
- /// returns true, filter pushdown the scan only needs t.a
- /// pushes the filter into the scan
- /// BUT internally evaluating the
- /// predicate still requires t.b
- /// ```
- ///
- /// # Limit
- ///
- /// If `limit` is specified, the scan must produce *at least* this many
- /// rows, though it may return more. Like Projection Pushdown and Filter
- /// Pushdown, DataFusion pushes `LIMIT`s as far down in the plan as
- /// possible. This is called "Limit Pushdown", and some sources can use the
- /// information to improve performance.
- ///
- /// Note: If any pushed-down filters are `Inexact`, the `LIMIT` cannot be
- /// pushed down. Inexact filters do not guarantee that every filtered row is
- /// removed, so applying the limit could leave too few rows to return in the
- /// final result.
- ///
- /// # Evaluation Order
- ///
- /// The logical evaluation order is `filters`, then `limit`, then
- /// `projection`.
- ///
- /// Note that `limit` applies to the filtered result, not to the unfiltered
- /// input, and `projection` affects only which columns are returned, not
- /// which rows qualify.
- ///
- /// For example, if a scan receives:
- ///
- /// - `projection = [a]`
- /// - `filters = [b > 5]`
- /// - `limit = Some(3)`
- ///
- /// It must logically produce results equivalent to:
- ///
- /// ```text
- /// PROJECTION a (LIMIT 3 (SCAN WHERE b > 5))
- /// ```
- ///
- /// As noted above, columns referenced only by pushed-down filters may be
- /// absent from `projection`.
- async fn scan(
- &self,
- state: &dyn Session,
- projection: Option<&Vec<usize>>,
- filters: &[Expr],
- limit: Option<usize>,
- ) -> Result<Arc<dyn ExecutionPlan>>;
-
- /// Create an [`ExecutionPlan`] for scanning the table using structured arguments.
- ///
- /// This method uses [`ScanArgs`] to pass scan parameters in a structured way
- /// and returns a [`ScanResult`] containing the execution plan.
- ///
- /// Table providers can override this method to take advantage of additional
- /// parameters like the upcoming `preferred_ordering` that may not be available through
- /// other scan methods.
- ///
- /// # Arguments
- /// * `state` - The session state containing configuration and context
- /// * `args` - Structured scan arguments including projection, filters, limit, and ordering preferences
- ///
- /// # Returns
- /// A [`ScanResult`] containing the [`ExecutionPlan`] for scanning the table
- ///
- /// See [`Self::scan`] for detailed documentation about projection, filters, and limits.
- async fn scan_with_args<'a>(
- &self,
- state: &dyn Session,
- args: ScanArgs<'a>,
- ) -> Result<ScanResult> {
- let filters = args.filters().unwrap_or(&[]);
- let projection = args.projection().map(|p| p.to_vec());
- let limit = args.limit();
- let plan = self
- .scan(state, projection.as_ref(), filters, limit)
- .await?;
- Ok(plan.into())
- }
-
- /// Specify if DataFusion should provide filter expressions to the
- /// TableProvider to apply *during* the scan.
- ///
- /// Some TableProviders can evaluate filters more efficiently than the
- /// `Filter` operator in DataFusion, for example by using an index.
- ///
- /// # Parameters and Return Value
- ///
- /// The return `Vec` must have one element for each element of the `filters`
- /// argument. The value of each element indicates if the TableProvider can
- /// apply the corresponding filter during the scan. The position in the return
- /// value corresponds to the expression in the `filters` parameter.
- ///
- /// If the length of the resulting `Vec` does not match the `filters` input
- /// an error will be thrown.
- ///
- /// Each element in the resulting `Vec` is one of the following:
- /// * [`Exact`] or [`Inexact`]: The TableProvider can apply the filter
- /// during scan
- /// * [`Unsupported`]: The TableProvider cannot apply the filter during scan
- ///
- /// By default, this function returns [`Unsupported`] for all filters,
- /// meaning no filters will be provided to [`Self::scan`].
- ///
- /// [`Unsupported`]: TableProviderFilterPushDown::Unsupported
- /// [`Exact`]: TableProviderFilterPushDown::Exact
- /// [`Inexact`]: TableProviderFilterPushDown::Inexact
- /// # Example
- ///
- /// ```rust
- /// # use std::any::Any;
- /// # use std::sync::Arc;
- /// # use arrow::datatypes::SchemaRef;
- /// # use async_trait::async_trait;
- /// # use datafusion_catalog::{TableProvider, Session};
- /// # use datafusion_common::Result;
- /// # use datafusion_expr::{Expr, TableProviderFilterPushDown, TableType};
- /// # use datafusion_physical_plan::ExecutionPlan;
- /// // Define a struct that implements the TableProvider trait
- /// #[derive(Debug)]
- /// struct TestDataSource {}
- ///
- /// #[async_trait]
- /// impl TableProvider for TestDataSource {
- /// # fn schema(&self) -> SchemaRef { todo!() }
- /// # fn table_type(&self) -> TableType { todo!() }
- /// # async fn scan(&self, s: &dyn Session, p: Option<&Vec<usize>>, f: &[Expr], l: Option<usize>) -> Result<Arc<dyn ExecutionPlan>> {
- /// todo!()
- /// # }
- /// // Override the supports_filters_pushdown to evaluate which expressions
- /// // to accept as pushdown predicates.
- /// fn supports_filters_pushdown(&self, filters: &[&Expr]) -> Result<Vec<TableProviderFilterPushDown>> {
- /// // Process each filter
- /// let support: Vec<_> = filters.iter().map(|expr| {
- /// match expr {
- /// // This example only supports a between expr with a single column named "c1".
- /// Expr::Between(between_expr) => {
- /// between_expr.expr
- /// .try_as_col()
- /// .map(|column| {
- /// if column.name == "c1" {
- /// TableProviderFilterPushDown::Exact
- /// } else {
- /// TableProviderFilterPushDown::Unsupported
- /// }
- /// })
- /// // If there is no column in the expr set the filter to unsupported.
- /// .unwrap_or(TableProviderFilterPushDown::Unsupported)
- /// }
- /// _ => {
- /// // For all other cases return Unsupported.
- /// TableProviderFilterPushDown::Unsupported
- /// }
- /// }
- /// }).collect();
- /// Ok(support)
- /// }
- /// }
- /// ```
- fn supports_filters_pushdown(
- &self,
- filters: &[&Expr],
- ) -> Result<Vec<TableProviderFilterPushDown>> {
- Ok(vec![
- TableProviderFilterPushDown::Unsupported;
- filters.len()
- ])
- }
-
- /// Get statistics for this table, if available
- /// Although not presently used in mainline DataFusion, this allows implementation specific
- /// behavior for downstream repositories, in conjunction with specialized optimizer rules to
- /// perform operations such as re-ordering of joins.
- fn statistics(&self) -> Option<Statistics> {
- None
- }
-
- /// Return an [`ExecutionPlan`] to insert data into this table, if
- /// supported.
- ///
- /// The returned plan should return a single row in a UInt64
- /// column called "count" such as the following
- ///
- /// ```text
- /// +-------+,
- /// | count |,
- /// +-------+,
- /// | 6 |,
- /// +-------+,
- /// ```
- ///
- /// # See Also
- ///
- /// See [`DataSinkExec`] for the common pattern of inserting a
- /// streams of `RecordBatch`es as files to an ObjectStore.
- ///
- /// [`DataSinkExec`]: datafusion_datasource::sink::DataSinkExec
- async fn insert_into(
- &self,
- _state: &dyn Session,
- _input: Arc<dyn ExecutionPlan>,
- _insert_op: InsertOp,
- ) -> Result<Arc<dyn ExecutionPlan>> {
- not_impl_err!("Insert into not implemented for this table")
- }
-
- /// Delete rows matching the filter predicates.
- ///
- /// Returns an [`ExecutionPlan`] producing a single row with `count` (UInt64).
- /// Empty `filters` deletes all rows.
- async fn delete_from(
- &self,
- _state: &dyn Session,
- _filters: Vec<Expr>,
- ) -> Result<Arc<dyn ExecutionPlan>> {
- not_impl_err!("DELETE not supported for {} table", self.table_type())
- }
-
- /// Update rows matching the filter predicates.
- ///
- /// Returns an [`ExecutionPlan`] producing a single row with `count` (UInt64).
- /// Empty `filters` updates all rows.
- async fn update(
- &self,
- _state: &dyn Session,
- _assignments: Vec<(String, Expr)>,
- _filters: Vec<Expr>,
- ) -> Result<Arc<dyn ExecutionPlan>> {
- not_impl_err!("UPDATE not supported for {} table", self.table_type())
- }
-
- /// Remove all rows from the table.
- ///
- /// Should return an [ExecutionPlan] producing a single row with count (UInt64),
- /// representing the number of rows removed.
- async fn truncate(&self, _state: &dyn Session) -> Result<Arc<dyn ExecutionPlan>> {
- not_impl_err!("TRUNCATE not supported for {} table", self.table_type())
- }
-}
-
-impl dyn TableProvider {
- /// Returns `true` if the table provider is of type `T`.
- ///
- /// Prefer this over `downcast_ref::<T>().is_some()`. Works correctly when
- /// called on `Arc<dyn TableProvider>` via auto-deref.
- pub fn is<T: TableProvider>(&self) -> bool {
- (self as &dyn Any).is::<T>()
- }
-
- /// Attempts to downcast this table provider to a concrete type `T`,
- /// returning `None` if the provider is not of that type.
- ///
- /// Works correctly when called on `Arc<dyn TableProvider>` via auto-deref,
- /// unlike `(&arc as &dyn Any).downcast_ref::<T>()` which would attempt to
- /// downcast the `Arc` itself.
- pub fn downcast_ref<T: TableProvider>(&self) -> Option<&T> {
- (self as &dyn Any).downcast_ref()
- }
-}
-
-/// Arguments for scanning a table with [`TableProvider::scan_with_args`].
-#[derive(Debug, Clone, Default)]
-pub struct ScanArgs<'a> {
- filters: Option<&'a [Expr]>,
- projection: Option<&'a [usize]>,
- limit: Option<usize>,
- statistics_requests: &'a [StatisticsRequest],
-}
-
-impl<'a> ScanArgs<'a> {
- /// Set the column projection for the scan.
- ///
- /// The projection is a list of column indices from [`TableProvider::schema`]
- /// that should be included in the scan results. If `None`, all columns are included.
- ///
- /// # Arguments
- /// * `projection` - Optional slice of column indices to project
- pub fn with_projection(mut self, projection: Option<&'a [usize]>) -> Self {
- self.projection = projection;
- self
- }
-
- /// Get the column projection for the scan.
- ///
- /// Returns a reference to the projection column indices, or `None` if
- /// no projection was specified (meaning all columns should be included).
- pub fn projection(&self) -> Option<&'a [usize]> {
- self.projection
- }
-
- /// Set the filter expressions for the scan.
- ///
- /// Filters are boolean expressions that should be evaluated during the scan
- /// to reduce the number of rows returned. All expressions are combined with AND logic.
- /// Whether filters are actually pushed down depends on [`TableProvider::supports_filters_pushdown`].
- ///
- /// # Arguments
- /// * `filters` - Optional slice of filter expressions
- pub fn with_filters(mut self, filters: Option<&'a [Expr]>) -> Self {
- self.filters = filters;
- self
- }
-
- /// Get the filter expressions for the scan.
- ///
- /// Returns a reference to the filter expressions, or `None` if no filters were specified.
- pub fn filters(&self) -> Option<&'a [Expr]> {
- self.filters
- }
-
- /// Set the maximum number of rows to return from the scan.
- ///
- /// If specified, the scan should return at most this many rows. This is typically
- /// used to optimize queries with `LIMIT` clauses.
- ///
- /// # Arguments
- /// * `limit` - Optional maximum number of rows to return
- pub fn with_limit(mut self, limit: Option<usize>) -> Self {
- self.limit = limit;
- self
- }
-
- /// Get the maximum number of rows to return from the scan.
- ///
- /// Returns the row limit, or `None` if no limit was specified.
- pub fn limit(&self) -> Option<usize> {
- self.limit
- }
-
- /// Specifies the statistics the caller may use when optimizing the query.
- ///
- /// This is intended to allow the `TableProvider` to cheaply provide
- /// statistics that may help, such as those it has in an in-memory catalog
- /// or from some other metadata source.
- ///
- /// `TableProvider`s read these via [`Self::statistics_requests()`]; anything
- /// a `TableProvider` cannot answer cheaply it simply ignores. DataFusion's
- /// own `TableProvider`s ignore this field — it exists so a request can be
- /// threaded from a custom optimizer rule (which annotates
- /// `TableScan::statistics_requests`) through to a custom `TableProvider`.
- pub fn with_statistics_requests(
- mut self,
- statistics_requests: &'a [StatisticsRequest],
- ) -> Self {
- self.statistics_requests = statistics_requests;
- self
- }
-
- /// Get the statistics requests for the scan. Empty if none were set.
- ///
- /// See [`Self::with_statistics_requests`] for more details
- pub fn statistics_requests(&self) -> &'a [StatisticsRequest] {
- self.statistics_requests
- }
-}
-
-/// Result of a table scan operation from [`TableProvider::scan_with_args`].
-#[derive(Debug, Clone)]
-pub struct ScanResult {
- /// The ExecutionPlan to run.
- plan: Arc<dyn ExecutionPlan>,
-}
-
-impl ScanResult {
- /// Create a new `ScanResult` with the given execution plan.
- ///
- /// # Arguments
- /// * `plan` - The execution plan that will perform the table scan
- pub fn new(plan: Arc<dyn ExecutionPlan>) -> Self {
- Self { plan }
- }
-
- /// Get a reference to the execution plan for this scan result.
- ///
- /// Returns a reference to the [`ExecutionPlan`] that will perform
- /// the actual table scanning and data retrieval.
- pub fn plan(&self) -> &Arc<dyn ExecutionPlan> {
- &self.plan
- }
-
- /// Consume this ScanResult and return the execution plan.
- ///
- /// Returns the owned [`ExecutionPlan`] that will perform
- /// the actual table scanning and data retrieval.
- pub fn into_inner(self) -> Arc<dyn ExecutionPlan> {
- self.plan
- }
-}
-
-impl From<Arc<dyn ExecutionPlan>> for ScanResult {
- fn from(plan: Arc<dyn ExecutionPlan>) -> Self {
- Self::new(plan)
- }
-}
-
-/// A factory which creates [`TableProvider`]s at runtime given a URL.
-///
-/// For example, this can be used to create a table "on the fly"
-/// from a directory of files only when that name is referenced.
-#[async_trait]
-pub trait TableProviderFactory: Debug + Sync + Send {
- /// Create a TableProvider with the given url
- async fn create(
- &self,
- state: &dyn Session,
- cmd: &CreateExternalTable,
- ) -> Result<Arc<dyn TableProvider>>;
-}
-
-/// Describes arguments provided to the table function call.
-pub struct TableFunctionArgs<'e, 's> {
- /// Call arguments.
- exprs: &'e [Expr],
- /// Session within which the function is called.
- session: &'s dyn Session,
-}
-
-impl<'e, 's> TableFunctionArgs<'e, 's> {
- /// Make a new [`TableFunctionArgs`].
- pub fn new(exprs: &'e [Expr], session: &'s dyn Session) -> Self {
- Self { exprs, session }
- }
-
- /// Get expressions passed as the called function arguments.
- pub fn exprs(&self) -> &'e [Expr] {
- self.exprs
- }
-
- /// Get a session where the table function is called.
- pub fn session(&self) -> &'s dyn Session {
- self.session
- }
-}
-
-/// A trait for table function implementations
-pub trait TableFunctionImpl: Debug + Sync + Send + Any {
- /// Create a table provider
- #[deprecated(
- since = "53.0.0",
- note = "Implement `TableFunctionImpl::call_with_args` instead"
- )]
- fn call(&self, _exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
- internal_err!(
- "TableFunctionImpl::call is not implemented. Implement TableFunctionImpl::call_with_args instead."
- )
- }
-
- /// Create a table provider
- fn call_with_args(&self, args: TableFunctionArgs) -> Result<Arc<dyn TableProvider>> {
- #[expect(deprecated)]
- self.call(args.exprs)
- }
-}
-
-/// A table that uses a function to generate data
-#[derive(Clone, Debug)]
-pub struct TableFunction {
- /// Name of the table function
- name: String,
- /// Function implementation
- fun: Arc<dyn TableFunctionImpl>,
-}
-
-impl TableFunction {
- /// Create a new table function
- pub fn new(name: String, fun: Arc<dyn TableFunctionImpl>) -> Self {
- Self { name, fun }
- }
-
- /// Get the name of the table function
- pub fn name(&self) -> &str {
- &self.name
- }
-
- /// Get the implementation of the table function
- pub fn function(&self) -> &Arc<dyn TableFunctionImpl> {
- &self.fun
- }
-
- /// Get the function implementation and generate a table
- #[deprecated(
- since = "53.0.0",
- note = "Use `TableFunction::create_table_provider_with_args` instead"
- )]
- pub fn create_table_provider(&self, args: &[Expr]) -> Result<Arc<dyn TableProvider>> {
- #[expect(deprecated)]
- self.fun.call(args)
- }
-
- /// Get the function implementation and generate a table
- pub fn create_table_provider_with_args(
- &self,
- args: TableFunctionArgs,
- ) -> Result<Arc<dyn TableProvider>> {
- self.fun.call_with_args(args)
- }
-}
diff --git a/datafusion/core/src/datasource/listing_table_factory.rs b/datafusion/core/src/datasource/listing_table_factory.rs
index ba94d23236..68f6743189 100644
--- a/datafusion/core/src/datasource/listing_table_factory.rs
+++ b/datafusion/core/src/datasource/listing_table_factory.rs
@@ -838,6 +838,7 @@ mod tests {
use datafusion_execution::config::SessionConfig;
use datafusion_physical_expr::PhysicalExpr;
use datafusion_physical_plan::ExecutionPlan;
+ use datafusion_session::{CatalogProviderList, EmptyCatalogProviderList};
use std::any::Any;
use std::collections::HashMap;
@@ -853,6 +854,9 @@ mod tests {
fn config(&self) -> &SessionConfig {
unimplemented!()
}
+ fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
+ Arc::new(EmptyCatalogProviderList)
+ }
async fn create_physical_plan(
&self,
_logical_plan: &datafusion_expr::LogicalPlan,
diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs
index ff4ec20cfc..dfdbb1617e 100644
--- a/datafusion/core/src/execution/session_state.rs
+++ b/datafusion/core/src/execution/session_state.rs
@@ -270,6 +270,10 @@ impl Session for SessionState {
self.config()
}
+ fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
+ Arc::clone(self.catalog_list())
+ }
+
async fn create_physical_plan(
&self,
logical_plan: &LogicalPlan,
@@ -2372,6 +2376,7 @@ mod tests {
use datafusion_optimizer::Optimizer;
use datafusion_optimizer::optimizer::OptimizerRule;
use datafusion_physical_plan::display::DisplayableExecutionPlan;
+ use datafusion_session::Session;
use datafusion_sql::planner::{PlannerContext, SqlToRel};
use std::collections::HashMap;
use std::sync::Arc;
@@ -2463,6 +2468,9 @@ mod tests {
let session_state = SessionStateBuilder::new()
.with_catalog_list(Arc::new(MemoryCatalogProviderList::new()))
.build();
+ let session_catalogs = Session::catalog_list(&session_state);
+ assert!(Arc::ptr_eq(&session_catalogs, session_state.catalog_list()));
+
let table_ref = session_state.resolve_table_ref("employee").to_string();
session_state
.schema_for_ref(&table_ref)?
diff --git a/datafusion/datasource-arrow/src/file_format.rs b/datafusion/datasource-arrow/src/file_format.rs
index 1daf12540c..c50ad98dfc 100644
--- a/datafusion/datasource-arrow/src/file_format.rs
+++ b/datafusion/datasource-arrow/src/file_format.rs
@@ -548,6 +548,7 @@ mod tests {
AggregateUDF, Expr, HigherOrderUDF, LogicalPlan, ScalarUDF, WindowUDF,
};
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
+ use datafusion_session::{CatalogProviderList, EmptyCatalogProviderList};
use object_store::{chunked::ChunkedStore, memory::InMemory};
struct MockSession {
@@ -574,6 +575,10 @@ mod tests {
&self.config
}
+ fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
+ Arc::new(EmptyCatalogProviderList)
+ }
+
async fn create_physical_plan(
&self,
_logical_plan: &LogicalPlan,
diff --git a/datafusion/datasource/src/url.rs b/datafusion/datasource/src/url.rs
index 7985a29e4f..9ac4c5f50d 100644
--- a/datafusion/datasource/src/url.rs
+++ b/datafusion/datasource/src/url.rs
@@ -523,6 +523,7 @@ mod tests {
};
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
use datafusion_physical_plan::ExecutionPlan;
+ use datafusion_session::{CatalogProviderList, EmptyCatalogProviderList};
use object_store::{
CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload,
PutMultipartOptions, PutPayload,
@@ -1191,6 +1192,10 @@ mod tests {
&self.config
}
+ fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
+ Arc::new(EmptyCatalogProviderList)
+ }
+
async fn create_physical_plan(
&self,
_logical_plan: &LogicalPlan,
diff --git a/datafusion/ffi/src/session/mod.rs b/datafusion/ffi/src/session/mod.rs
index 6ab6f0dd4e..0c2d9fdeee 100644
--- a/datafusion/ffi/src/session/mod.rs
+++ b/datafusion/ffi/src/session/mod.rs
@@ -42,7 +42,7 @@ use datafusion_proto::logical_plan::LogicalExtensionCodec;
use datafusion_proto::logical_plan::from_proto::parse_expr;
use datafusion_proto::logical_plan::to_proto::serialize_expr;
use datafusion_proto::protobuf::LogicalExprNode;
-use datafusion_session::Session;
+use datafusion_session::{CatalogProviderList, Session};
use prost::Message;
use stabby::str::Str as SStr;
@@ -51,6 +51,7 @@ use stabby::vec::Vec as SVec;
use tokio::runtime::Handle;
use crate::arrow_wrappers::WrappedSchema;
+use crate::catalog_provider_list::FFI_CatalogProviderList;
use crate::execution::FFI_TaskContext;
use crate::execution_plan::FFI_ExecutionPlan;
use crate::physical_expr::FFI_PhysicalExpr;
@@ -83,6 +84,8 @@ pub(crate) struct FFI_SessionRef {
config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+ catalog_list: unsafe extern "C" fn(&Self) -> FFI_CatalogProviderList,
+
create_physical_plan:
unsafe extern "C" fn(
&Self,
@@ -160,6 +163,16 @@ unsafe extern "C" fn config_fn_wrapper(session: &FFI_SessionRef) -> FFI_SessionC
session.config().into()
}
+unsafe extern "C" fn catalog_list_fn_wrapper(
+ session: &FFI_SessionRef,
+) -> FFI_CatalogProviderList {
+ FFI_CatalogProviderList::new_with_ffi_codec(
+ session.inner().catalog_list(),
+ unsafe { session.runtime() }.clone(),
+ session.logical_codec.clone(),
+ )
+}
+
unsafe extern "C" fn create_physical_plan_fn_wrapper(
session: &FFI_SessionRef,
logical_plan_serialized: SVec<u8>,
@@ -310,6 +323,7 @@ unsafe extern "C" fn clone_fn_wrapper(provider: &FFI_SessionRef) -> FFI_SessionR
FFI_SessionRef {
session_id: session_id_fn_wrapper,
config: config_fn_wrapper,
+ catalog_list: catalog_list_fn_wrapper,
create_physical_plan: create_physical_plan_fn_wrapper,
create_physical_expr: create_physical_expr_fn_wrapper,
scalar_functions: scalar_functions_fn_wrapper,
@@ -351,6 +365,7 @@ impl FFI_SessionRef {
Self {
session_id: session_id_fn_wrapper,
config: config_fn_wrapper,
+ catalog_list: catalog_list_fn_wrapper,
create_physical_plan: create_physical_plan_fn_wrapper,
create_physical_expr: create_physical_expr_fn_wrapper,
scalar_functions: scalar_functions_fn_wrapper,
@@ -378,6 +393,7 @@ impl FFI_SessionRef {
pub struct ForeignSession {
session: FFI_SessionRef,
config: SessionConfig,
+ catalog_list: Arc<dyn CatalogProviderList>,
scalar_functions: HashMap<String, Arc<ScalarUDF>>,
higher_order_functions: HashMap<String, Arc<HigherOrderUDF>>,
aggregate_functions: HashMap<String, Arc<AggregateUDF>>,
@@ -410,6 +426,9 @@ impl TryFrom<&FFI_SessionRef> for ForeignSession {
let config = (session.config)(session);
let config = SessionConfig::try_from(&config)?;
+ let ffi_catalog_list = (session.catalog_list)(session);
+ let catalog_list = (&ffi_catalog_list).into();
+
let scalar_functions = (session.scalar_functions)(session)
.into_iter()
.map(|kv_pair| {
@@ -447,6 +466,7 @@ impl TryFrom<&FFI_SessionRef> for ForeignSession {
Ok(Self {
session: session.clone(),
config,
+ catalog_list,
table_options,
scalar_functions,
higher_order_functions: HashMap::new(),
@@ -549,6 +569,10 @@ impl Session for ForeignSession {
self.config.options()
}
+ fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
+ Arc::clone(&self.catalog_list)
+ }
+
async fn create_physical_plan(
&self,
logical_plan: &LogicalPlan,
@@ -650,6 +674,7 @@ mod tests {
use std::sync::Arc;
use arrow_schema::{DataType, Field, Schema};
+ use datafusion::catalog::MemoryCatalogProvider;
use datafusion::execution::SessionStateBuilder;
use datafusion_common::DataFusionError;
use datafusion_expr::col;
@@ -693,6 +718,17 @@ mod tests {
assert_eq!(foreign_session.session_id(), state.session_id());
+ let foreign_catalog_list = foreign_session.catalog_list();
+ assert_eq!(
+ foreign_catalog_list.catalog_names(),
+ state.catalog_list().catalog_names()
+ );
+ foreign_catalog_list.register_catalog(
+ "foreign_registered".to_owned(),
+ Arc::new(MemoryCatalogProvider::new()),
+ );
+ assert!(state.catalog_list().catalog("foreign_registered").is_some());
+
let logical_plan = LogicalPlan::default();
let physical_plan = foreign_session.create_physical_plan(&logical_plan).await?;
assert_eq!(
diff --git a/datafusion/ffi/src/tests/async_provider.rs b/datafusion/ffi/src/tests/async_provider.rs
index 69104709b4..9821c3e501 100644
--- a/datafusion/ffi/src/tests/async_provider.rs
+++ b/datafusion/ffi/src/tests/async_provider.rs
@@ -31,7 +31,7 @@ use std::sync::Arc;
use arrow::array::RecordBatch;
use arrow::datatypes::Schema;
use async_trait::async_trait;
-use datafusion_catalog::TableProvider;
+use datafusion_catalog::{MemoryCatalogProvider, TableProvider};
use datafusion_common::{Result, exec_err};
use datafusion_execution::RecordBatchStream;
use datafusion_expr::Expr;
@@ -135,11 +135,28 @@ impl TableProvider for AsyncTableProvider {
async fn scan(
&self,
- _state: &dyn Session,
+ state: &dyn Session,
_projection: Option<&Vec<usize>>,
_filters: &[Expr],
_limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>> {
+ let catalog = state.catalog_list().catalog("datafusion").ok_or_else(|| {
+ datafusion_common::exec_datafusion_err!("missing datafusion catalog")
+ })?;
+ let schema = catalog.schema("public").ok_or_else(|| {
+ datafusion_common::exec_datafusion_err!("missing public schema")
+ })?;
+ if schema.table("external_table").await?.is_none() {
+ return exec_err!("missing external_table");
+ }
+
+ // Register a catalog from the dynamically loaded library so the host
+ // can verify that catalog mutations cross the FFI boundary as well.
+ state.catalog_list().register_catalog(
+ "ffi_registered".to_owned(),
+ Arc::new(MemoryCatalogProvider::new()),
+ );
+
Ok(Arc::new(AsyncTestExecutionPlan::new(
self.batch_request.clone(),
self.batch_receiver.resubscribe(),
diff --git a/datafusion/ffi/tests/ffi_integration.rs b/datafusion/ffi/tests/ffi_integration.rs
index f1edc44730..86f953e262 100644
--- a/datafusion/ffi/tests/ffi_integration.rs
+++ b/datafusion/ffi/tests/ffi_integration.rs
@@ -58,6 +58,15 @@ mod tests {
assert!(results.contains(&create_record_batch(6, 1)));
assert!(results.contains(&create_record_batch(7, 5)));
+ if !synchronous {
+ assert!(
+ ctx.state()
+ .catalog_list()
+ .catalog("ffi_registered")
+ .is_some()
+ );
+ }
+
Ok(())
}
diff --git a/datafusion/session/Cargo.toml b/datafusion/session/Cargo.toml
index 230e26d1fc..2bbbdd20df 100644
--- a/datafusion/session/Cargo.toml
+++ b/datafusion/session/Cargo.toml
@@ -31,6 +31,7 @@ version.workspace = true
all-features = true
[dependencies]
+arrow-schema = { workspace = true }
async-trait = { workspace = true }
datafusion-common = { workspace = true }
datafusion-execution = { workspace = true }
diff --git a/datafusion/session/README.md b/datafusion/session/README.md
index 4bb605b1e1..72a693e81d 100644
--- a/datafusion/session/README.md
+++ b/datafusion/session/README.md
@@ -21,7 +21,7 @@
[Apache DataFusion] is an extensible query execution framework, written in Rust, that uses [Apache Arrow] as its in-memory format.
-This crate provides **session-related abstractions** used in the DataFusion query engine. A _session_ represents the runtime context for query execution, including configuration, runtime environment, function registry, and planning.
+This crate defines the **session-related APIs and extension points** used in the DataFusion query engine. A _session_ represents the runtime context for query execution, including configuration, runtime environment, function registry, and planning. This crate focuses on shared interfaces; concrete query-engine implementations live in higher-level DataFusion crates.
Most projects should use the [`datafusion`] crate directly, which re-exports
this module. If you are already using the [`datafusion`] crate, there is no
diff --git a/datafusion/catalog/src/catalog.rs b/datafusion/session/src/catalog.rs
similarity index 90%
copy from datafusion/catalog/src/catalog.rs
copy to datafusion/session/src/catalog.rs
index 34cdf74440..bd9eb781ab 100644
--- a/datafusion/catalog/src/catalog.rs
+++ b/datafusion/session/src/catalog.rs
@@ -23,6 +23,31 @@ pub use crate::schema::SchemaProvider;
use datafusion_common::Result;
use datafusion_common::not_impl_err;
+/// A catalog list that contains no catalogs.
+///
+/// [`Session`](crate::Session) implementations that do not provide catalog
+/// access can return this list explicitly.
+#[derive(Debug, Default)]
+pub struct EmptyCatalogProviderList;
+
+impl CatalogProviderList for EmptyCatalogProviderList {
+ fn register_catalog(
+ &self,
+ _name: String,
+ _catalog: Arc<dyn CatalogProvider>,
+ ) -> Option<Arc<dyn CatalogProvider>> {
+ None
+ }
+
+ fn catalog_names(&self) -> Vec<String> {
+ vec![]
+ }
+
+ fn catalog(&self, _name: &str) -> Option<Arc<dyn CatalogProvider>> {
+ None
+ }
+}
+
/// Represents a catalog, comprising a number of named schemas.
///
/// # Catalog Overview
@@ -207,3 +232,15 @@ impl dyn CatalogProviderList {
(self as &dyn Any).downcast_ref()
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::{CatalogProviderList, EmptyCatalogProviderList};
+
+ #[test]
+ fn empty_catalog_provider_list_has_no_catalogs() {
+ let catalogs = EmptyCatalogProviderList;
+ assert!(catalogs.catalog_names().is_empty());
+ assert!(catalogs.catalog("missing").is_none());
+ }
+}
diff --git a/datafusion/session/src/lib.rs b/datafusion/session/src/lib.rs
index 11f734e757..3b9ed7dacf 100644
--- a/datafusion/session/src/lib.rs
+++ b/datafusion/session/src/lib.rs
@@ -15,18 +15,23 @@
// specific language governing permissions and limitations
// under the License.
+// Make sure fast / cheap clones on Arc are explicit:
+// https://github.com/apache/datafusion/issues/11143
+#![cfg_attr(not(test), deny(clippy::clone_on_ref_ptr))]
#![cfg_attr(test, allow(clippy::needless_pass_by_value))]
-//! Session management for DataFusion query execution environment
+//! Session APIs for the DataFusion query execution environment
//!
-//! This module provides the core session management functionality for DataFusion,
-//! handling both Catalog (Table) and Datasource (File) configurations. It defines
-//! the fundamental interfaces and implementations for maintaining query execution
-//! state and configurations.
+//! This crate defines shared interfaces for session-related APIs and extension
+//! points. Concrete query-engine implementations are provided by higher-level
+//! DataFusion crates.
//!
//! Key components:
-//! * [`Session`] - Manages query execution context, including configurations,
+//! * [`Session`] - Describes a query execution context, including configurations,
//! catalogs, and runtime state
+//! * [`CatalogProviderList`], [`CatalogProvider`], and [`SchemaProvider`] -
+//! Describe catalog hierarchies
+//! * [`TableProvider`] - Provides data for query planning and execution
//! * [`SessionStore`] - Handles session persistence and retrieval
//!
//! The session system enables:
@@ -36,6 +41,17 @@
//! * Runtime environment configuration
//! * Query state persistence
+pub mod catalog;
+pub mod schema;
pub mod session;
+pub mod table;
+pub use crate::catalog::{
+ CatalogProvider, CatalogProviderList, EmptyCatalogProviderList,
+};
+pub use crate::schema::SchemaProvider;
pub use crate::session::{Session, SessionStore};
+pub use crate::table::{
+ ScanArgs, ScanResult, TableFunction, TableFunctionArgs, TableFunctionImpl,
+ TableProvider, TableProviderFactory,
+};
diff --git a/datafusion/catalog/src/schema.rs b/datafusion/session/src/schema.rs
similarity index 98%
copy from datafusion/catalog/src/schema.rs
copy to datafusion/session/src/schema.rs
index d99027593c..7a66072bb4 100644
--- a/datafusion/catalog/src/schema.rs
+++ b/datafusion/session/src/schema.rs
@@ -36,7 +36,7 @@ use datafusion_expr::TableType;
#[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
+ /// as part of `information_schema.schemata`.
fn owner_name(&self) -> Option<&str> {
None
}
diff --git a/datafusion/session/src/session.rs b/datafusion/session/src/session.rs
index 15ad543cf0..cdac3f4beb 100644
--- a/datafusion/session/src/session.rs
+++ b/datafusion/session/src/session.rs
@@ -27,6 +27,8 @@ use datafusion_expr::{
AggregateUDF, Expr, HigherOrderUDF, LogicalPlan, ScalarUDF, WindowUDF,
};
use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr};
+
+use crate::CatalogProviderList;
use parking_lot::{Mutex, RwLock};
use std::any::Any;
use std::collections::HashMap;
@@ -79,6 +81,9 @@ pub trait Session: Send + Sync {
/// Return the [`SessionConfig`]
fn config(&self) -> &SessionConfig;
+ /// Return the catalogs registered with this session.
+ fn catalog_list(&self) -> Arc<dyn CatalogProviderList>;
+
/// return the [`ConfigOptions`]
fn config_options(&self) -> &ConfigOptions {
self.config().options()
diff --git a/datafusion/catalog/src/table.rs b/datafusion/session/src/table.rs
similarity index 99%
copy from datafusion/catalog/src/table.rs
copy to datafusion/session/src/table.rs
index c6468fd5ad..8d9cd92d4c 100644
--- a/datafusion/catalog/src/table.rs
+++ b/datafusion/session/src/table.rs
@@ -21,7 +21,7 @@ use std::fmt::Debug;
use std::sync::Arc;
use crate::session::Session;
-use arrow::datatypes::SchemaRef;
+use arrow_schema::SchemaRef;
use async_trait::async_trait;
use datafusion_common::{Constraints, Statistics, not_impl_err};
use datafusion_common::{Result, internal_err};
@@ -253,9 +253,9 @@ pub trait TableProvider: Any + Debug + Sync + Send {
/// ```rust
/// # use std::any::Any;
/// # use std::sync::Arc;
- /// # use arrow::datatypes::SchemaRef;
+ /// # use arrow_schema::SchemaRef;
/// # use async_trait::async_trait;
- /// # use datafusion_catalog::{TableProvider, Session};
+ /// # use datafusion_session::{TableProvider, Session};
/// # use datafusion_common::Result;
/// # use datafusion_expr::{Expr, TableProviderFilterPushDown, TableType};
/// # use datafusion_physical_plan::ExecutionPlan;
@@ -337,7 +337,7 @@ pub trait TableProvider: Any + Debug + Sync + Send {
/// See [`DataSinkExec`] for the common pattern of inserting a
/// streams of `RecordBatch`es as files to an ObjectStore.
///
- /// [`DataSinkExec`]: datafusion_datasource::sink::DataSinkExec
+ /// [`DataSinkExec`]: https://docs.rs/datafusion-datasource/latest/datafusion_datasource/sink/struct.DataSinkExec.html
async fn insert_into(
&self,
_state: &dyn Session,
diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md
index 651588c500..57da7b7dac 100644
--- a/docs/source/library-user-guide/upgrading/55.0.0.md
+++ b/docs/source/library-user-guide/upgrading/55.0.0.md
@@ -777,6 +777,42 @@ async fn plan_extension(
See [PR #23649](https://github.com/apache/datafusion/pull/23649) for details.
+### Catalog traits moved to `datafusion-session`
+
+The catalog contract traits now live in the `datafusion-session` crate so they
+can be reached without downcasting a `Session` to `SessionState` (in particular
+across the FFI boundary). The affected traits are `CatalogProviderList`,
+`CatalogProvider`, `SchemaProvider`, `TableProvider`, `TableProviderFactory`,
+and `TableFunctionImpl`. The related `TableFunction` struct also moved.
+
+The `datafusion-catalog` crate re-exports all of them from their new location,
+so paths such as `datafusion::catalog::TableProvider` and
+`datafusion_catalog::CatalogProvider` continue to work unchanged. Most users do
+not need to do anything.
+
+### `Session` gains a required `catalog_list` method
+
+The `Session` trait now requires a `catalog_list` method that returns the
+catalogs registered with the session:
+
+```rust
+fn catalog_list(&self) -> Arc<dyn CatalogProviderList>;
+```
+
+Custom `Session` implementations must add this method. Implementations that do
+not expose a catalog can return the new `EmptyCatalogProviderList`:
+
+```rust
+use std::sync::Arc;
+use datafusion_session::{CatalogProviderList, EmptyCatalogProviderList};
+
+fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
+ Arc::new(EmptyCatalogProviderList)
+}
+```
+
+See [PR #23703](https://github.com/apache/datafusion/pull/23703) for details.
+
### `MSRV` updated to 1.94.0
The Minimum Supported Rust Version (MSRV) has been updated to [`1.94.0`].
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org