diff --git a/Cargo.lock b/Cargo.lock index 566cc1166813c..ed63f60b41519 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 34cdf74440cb3..07da1293a781d 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; - - /// Retrieves a specific schema from the catalog by name, provided it exists. - fn schema(&self, name: &str) -> Option>; - - /// 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, - ) -> Result>> { - // 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>> { - 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::().is_some()`. Works correctly when - /// called on `Arc` via auto-deref. - pub fn is(&self) -> bool { - (self as &dyn Any).is::() - } - - /// 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` via auto-deref, - /// unlike `(&arc as &dyn Any).downcast_ref::()` which would attempt to - /// downcast the `Arc` itself. - pub fn downcast_ref(&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, - ) -> Option>; - - /// Retrieves the list of available catalog names - fn catalog_names(&self) -> Vec; - - /// Retrieves a specific catalog by name, provided it exists. - fn catalog(&self, name: &str) -> Option>; -} - -impl dyn CatalogProviderList { - /// Returns `true` if the catalog provider list is of type `T`. - /// - /// Prefer this over `downcast_ref::().is_some()`. Works correctly when - /// called on `Arc` via auto-deref. - pub fn is(&self) -> bool { - (self as &dyn Any).is::() - } - - /// 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` via - /// auto-deref, unlike `(&arc as &dyn Any).downcast_ref::()` which would - /// attempt to downcast the `Arc` itself. - pub fn downcast_ref(&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 33d54b7cb89d5..815bfe32fac72 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 d99027593ccce..40b20caeb9bb9 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; - - /// Retrieves a specific table from the schema by name, if it exists, - /// otherwise returns `None`. - async fn table( - &self, - name: &str, - ) -> Result>, 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> { - 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, - ) -> Result>> { - 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>> { - 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::().is_some()`. Works correctly when - /// called on `Arc` via auto-deref. - pub fn is(&self) -> bool { - (self as &dyn Any).is::() - } - - /// 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` via auto-deref, - /// unlike `(&arc as &dyn Any).downcast_ref::()` which would attempt to - /// downcast the `Arc` itself. - pub fn downcast_ref(&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 c6468fd5ad131..2a10efbdcce6a 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> { - 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>, - filters: &[Expr], - limit: Option, - ) -> Result>; - - /// 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 { - 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>, f: &[Expr], l: Option) -> Result> { - /// todo!() - /// # } - /// // Override the supports_filters_pushdown to evaluate which expressions - /// // to accept as pushdown predicates. - /// fn supports_filters_pushdown(&self, filters: &[&Expr]) -> Result> { - /// // 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> { - 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 { - 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, - _insert_op: InsertOp, - ) -> Result> { - 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, - ) -> Result> { - 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, - ) -> Result> { - 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> { - 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::().is_some()`. Works correctly when - /// called on `Arc` via auto-deref. - pub fn is(&self) -> bool { - (self as &dyn Any).is::() - } - - /// 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` via auto-deref, - /// unlike `(&arc as &dyn Any).downcast_ref::()` which would attempt to - /// downcast the `Arc` itself. - pub fn downcast_ref(&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, - 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) -> 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 { - 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, -} - -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) -> 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 { - &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 { - self.plan - } -} - -impl From> for ScanResult { - fn from(plan: Arc) -> 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>; -} - -/// 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> { - 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> { - #[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, -} - -impl TableFunction { - /// Create a new table function - pub fn new(name: String, fun: Arc) -> 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 { - &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> { - #[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> { - 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 ba94d23236140..68f6743189447 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 { + 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 ff4ec20cfc1c3..dfdbb1617efde 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 { + 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 1daf12540cbe4..c50ad98dfca0b 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 { + 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 7985a29e4fd94..9ac4c5f50d1f7 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 { + 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 6ab6f0dd4ed45..0c2d9fdeee819 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, @@ -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, scalar_functions: HashMap>, higher_order_functions: HashMap>, aggregate_functions: HashMap>, @@ -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 { + 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 69104709b477e..9821c3e501f67 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>, _filters: &[Expr], _limit: Option, ) -> Result> { + 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 f1edc447309a6..86f953e262ead 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 230e26d1fc9fc..2bbbdd20df1b8 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 4bb605b1e199c..72a693e81deb4 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/session/src/catalog.rs b/datafusion/session/src/catalog.rs new file mode 100644 index 0000000000000..bd9eb781abe77 --- /dev/null +++ b/datafusion/session/src/catalog.rs @@ -0,0 +1,246 @@ +// 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::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; + +/// 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, + ) -> Option> { + None + } + + fn catalog_names(&self) -> Vec { + vec![] + } + + fn catalog(&self, _name: &str) -> Option> { + None + } +} + +/// 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; + + /// Retrieves a specific schema from the catalog by name, provided it exists. + fn schema(&self, name: &str) -> Option>; + + /// 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, + ) -> Result>> { + // 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>> { + 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::().is_some()`. Works correctly when + /// called on `Arc` via auto-deref. + pub fn is(&self) -> bool { + (self as &dyn Any).is::() + } + + /// 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` via auto-deref, + /// unlike `(&arc as &dyn Any).downcast_ref::()` which would attempt to + /// downcast the `Arc` itself. + pub fn downcast_ref(&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, + ) -> Option>; + + /// Retrieves the list of available catalog names + fn catalog_names(&self) -> Vec; + + /// Retrieves a specific catalog by name, provided it exists. + fn catalog(&self, name: &str) -> Option>; +} + +impl dyn CatalogProviderList { + /// Returns `true` if the catalog provider list is of type `T`. + /// + /// Prefer this over `downcast_ref::().is_some()`. Works correctly when + /// called on `Arc` via auto-deref. + pub fn is(&self) -> bool { + (self as &dyn Any).is::() + } + + /// 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` via + /// auto-deref, unlike `(&arc as &dyn Any).downcast_ref::()` which would + /// attempt to downcast the `Arc` itself. + pub fn downcast_ref(&self) -> Option<&T> { + (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 11f734e757452..3b9ed7dacf1a8 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/session/src/schema.rs b/datafusion/session/src/schema.rs new file mode 100644 index 0000000000000..7a66072bb4d8a --- /dev/null +++ b/datafusion/session/src/schema.rs @@ -0,0 +1,107 @@ +// 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. + +//! 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_schema.schemata`. + fn owner_name(&self) -> Option<&str> { + None + } + + /// Retrieves the list of available table names in this schema. + fn table_names(&self) -> Vec; + + /// Retrieves a specific table from the schema by name, if it exists, + /// otherwise returns `None`. + async fn table( + &self, + name: &str, + ) -> Result>, 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> { + 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, + ) -> Result>> { + 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>> { + 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::().is_some()`. Works correctly when + /// called on `Arc` via auto-deref. + pub fn is(&self) -> bool { + (self as &dyn Any).is::() + } + + /// 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` via auto-deref, + /// unlike `(&arc as &dyn Any).downcast_ref::()` which would attempt to + /// downcast the `Arc` itself. + pub fn downcast_ref(&self) -> Option<&T> { + (self as &dyn Any).downcast_ref() + } +} diff --git a/datafusion/session/src/session.rs b/datafusion/session/src/session.rs index 15ad543cf0ffb..cdac3f4bebc9e 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; + /// return the [`ConfigOptions`] fn config_options(&self) -> &ConfigOptions { self.config().options() diff --git a/datafusion/session/src/table.rs b/datafusion/session/src/table.rs new file mode 100644 index 0000000000000..8d9cd92d4c664 --- /dev/null +++ b/datafusion/session/src/table.rs @@ -0,0 +1,640 @@ +// 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::any::Any; +use std::borrow::Cow; +use std::fmt::Debug; +use std::sync::Arc; + +use crate::session::Session; +use arrow_schema::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, +}; +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> { + 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>, + filters: &[Expr], + limit: Option, + ) -> Result>; + + /// 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 { + 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_schema::SchemaRef; + /// # use async_trait::async_trait; + /// # use datafusion_session::{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>, f: &[Expr], l: Option) -> Result> { + /// todo!() + /// # } + /// // Override the supports_filters_pushdown to evaluate which expressions + /// // to accept as pushdown predicates. + /// fn supports_filters_pushdown(&self, filters: &[&Expr]) -> Result> { + /// // 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> { + 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 { + 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`]: https://docs.rs/datafusion-datasource/latest/datafusion_datasource/sink/struct.DataSinkExec.html + async fn insert_into( + &self, + _state: &dyn Session, + _input: Arc, + _insert_op: InsertOp, + ) -> Result> { + 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, + ) -> Result> { + 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, + ) -> Result> { + 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> { + 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::().is_some()`. Works correctly when + /// called on `Arc` via auto-deref. + pub fn is(&self) -> bool { + (self as &dyn Any).is::() + } + + /// 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` via auto-deref, + /// unlike `(&arc as &dyn Any).downcast_ref::()` which would attempt to + /// downcast the `Arc` itself. + pub fn downcast_ref(&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, + 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) -> 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 { + 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, +} + +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) -> 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 { + &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 { + self.plan + } +} + +impl From> for ScanResult { + fn from(plan: Arc) -> 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>; +} + +/// 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> { + 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> { + #[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, +} + +impl TableFunction { + /// Create a new table function + pub fn new(name: String, fun: Arc) -> 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 { + &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> { + #[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> { + self.fun.call_with_args(args) + } +} 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 651588c500979..57da7b7dac248 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; +``` + +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 { + 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`].