datafusion_catalog/catalog.rs
1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements. See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership. The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License. You may obtain a copy of the License at
8//
9// http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied. See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::any::Any;
19use std::fmt::Debug;
20use std::sync::Arc;
21
22pub use crate::schema::SchemaProvider;
23use datafusion_common::Result;
24use datafusion_common::not_impl_err;
25
26/// Represents a catalog, comprising a number of named schemas.
27///
28/// # Catalog Overview
29///
30/// To plan and execute queries, DataFusion needs a "Catalog" that provides
31/// metadata such as which schemas and tables exist, their columns and data
32/// types, and how to access the data.
33///
34/// The Catalog API consists:
35/// * [`CatalogProviderList`]: a collection of `CatalogProvider`s
36/// * [`CatalogProvider`]: a collection of `SchemaProvider`s (sometimes called a "database" in other systems)
37/// * [`SchemaProvider`]: a collection of `TableProvider`s (often called a "schema" in other systems)
38/// * [`TableProvider`]: individual tables
39///
40/// # Implementing Catalogs
41///
42/// To implement a catalog, you implement at least one of the [`CatalogProviderList`],
43/// [`CatalogProvider`] and [`SchemaProvider`] traits and register them
44/// appropriately in the `SessionContext`.
45///
46/// DataFusion comes with a simple in-memory catalog implementation,
47/// `MemoryCatalogProvider`, that is used by default and has no persistence.
48/// DataFusion does not include more complex Catalog implementations because
49/// catalog management is a key design choice for most data systems, and thus
50/// it is unlikely that any general-purpose catalog implementation will work
51/// well across many use cases.
52///
53/// # Implementing "Remote" catalogs
54///
55/// See [`remote_catalog`] for an end to end example of how to implement a
56/// remote catalog.
57///
58/// Sometimes catalog information is stored remotely and requires a network call
59/// to retrieve. For example, the [Delta Lake] table format stores table
60/// metadata in files on S3 that must be first downloaded to discover what
61/// schemas and tables exist.
62///
63/// [Delta Lake]: https://delta.io/
64/// [`remote_catalog`]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/data_io/remote_catalog.rs
65///
66/// The [`CatalogProvider`] can support this use case, but it takes some care.
67/// The planning APIs in DataFusion are not `async` and thus network IO can not
68/// be performed "lazily" / "on demand" during query planning. The rationale for
69/// this design is that using remote procedure calls for all catalog accesses
70/// required for query planning would likely result in multiple network calls
71/// per plan, resulting in very poor planning performance.
72///
73/// To implement [`CatalogProvider`] and [`SchemaProvider`] for remote catalogs,
74/// you need to provide an in memory snapshot of the required metadata. Most
75/// systems typically either already have this information cached locally or can
76/// batch access to the remote catalog to retrieve multiple schemas and tables
77/// in a single network call.
78///
79/// Note that [`SchemaProvider::table`] **is** an `async` function in order to
80/// simplify implementing simple [`SchemaProvider`]s. For many table formats it
81/// is easy to list all available tables but there is additional non trivial
82/// access required to read table details (e.g. statistics).
83///
84/// The pattern that DataFusion itself uses to plan SQL queries is to walk over
85/// the query to find all table references, performing required remote catalog
86/// lookups in parallel, storing the results in a cached snapshot, and then plans
87/// the query using that snapshot.
88///
89/// # Example Catalog Implementations
90///
91/// Here are some examples of how to implement custom catalogs:
92///
93/// * [`datafusion-cli`]: [`DynamicFileCatalogProvider`] catalog provider
94/// that treats files and directories on a filesystem as tables.
95///
96/// * The [`catalog.rs`]: a simple directory based catalog.
97///
98/// * [delta-rs]: [`UnityCatalogProvider`] implementation that can
99/// read from Delta Lake tables
100///
101/// [`datafusion-cli`]: https://datafusion.apache.org/user-guide/cli/index.html
102/// [`DynamicFileCatalogProvider`]: https://github.com/apache/datafusion/blob/31b9b48b08592b7d293f46e75707aad7dadd7cbc/datafusion-cli/src/catalog.rs#L75
103/// [`catalog.rs`]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/data_io/catalog.rs
104/// [delta-rs]: https://github.com/delta-io/delta-rs
105/// [`UnityCatalogProvider`]: https://github.com/delta-io/delta-rs/blob/951436ecec476ce65b5ed3b58b50fb0846ca7b91/crates/deltalake-core/src/data_catalog/unity/datafusion.rs#L111-L123
106///
107/// [`TableProvider`]: crate::TableProvider
108pub trait CatalogProvider: Any + Debug + Sync + Send {
109 /// Retrieves the list of available schema names in this catalog.
110 fn schema_names(&self) -> Vec<String>;
111
112 /// Retrieves a specific schema from the catalog by name, provided it exists.
113 fn schema(&self, name: &str) -> Option<Arc<dyn SchemaProvider>>;
114
115 /// Adds a new schema to this catalog.
116 ///
117 /// If a schema of the same name existed before, it is replaced in
118 /// the catalog and returned.
119 ///
120 /// By default returns a "Not Implemented" error
121 fn register_schema(
122 &self,
123 name: &str,
124 schema: Arc<dyn SchemaProvider>,
125 ) -> Result<Option<Arc<dyn SchemaProvider>>> {
126 // use variables to avoid unused variable warnings
127 let _ = name;
128 let _ = schema;
129 not_impl_err!("Registering new schemas is not supported")
130 }
131
132 /// Removes a schema from this catalog. Implementations of this method should return
133 /// errors if the schema exists but cannot be dropped. For example, in DataFusion's
134 /// default in-memory catalog, `MemoryCatalogProvider`, a non-empty schema
135 /// will only be successfully dropped when `cascade` is true.
136 /// This is equivalent to how DROP SCHEMA works in PostgreSQL.
137 ///
138 /// Implementations of this method should return None if schema with `name`
139 /// does not exist.
140 ///
141 /// By default returns a "Not Implemented" error
142 fn deregister_schema(
143 &self,
144 _name: &str,
145 _cascade: bool,
146 ) -> Result<Option<Arc<dyn SchemaProvider>>> {
147 not_impl_err!("Deregistering new schemas is not supported")
148 }
149}
150
151impl dyn CatalogProvider {
152 /// Returns `true` if the catalog provider is of type `T`.
153 ///
154 /// Prefer this over `downcast_ref::<T>().is_some()`. Works correctly when
155 /// called on `Arc<dyn CatalogProvider>` via auto-deref.
156 pub fn is<T: CatalogProvider>(&self) -> bool {
157 (self as &dyn Any).is::<T>()
158 }
159
160 /// Attempts to downcast this catalog provider to a concrete type `T`,
161 /// returning `None` if the provider is not of that type.
162 ///
163 /// Works correctly when called on `Arc<dyn CatalogProvider>` via auto-deref,
164 /// unlike `(&arc as &dyn Any).downcast_ref::<T>()` which would attempt to
165 /// downcast the `Arc` itself.
166 pub fn downcast_ref<T: CatalogProvider>(&self) -> Option<&T> {
167 (self as &dyn Any).downcast_ref()
168 }
169}
170
171/// Represent a list of named [`CatalogProvider`]s.
172///
173/// Please see the documentation on [`CatalogProvider`] for details of
174/// implementing a custom catalog.
175pub trait CatalogProviderList: Any + Debug + Sync + Send {
176 /// Adds a new catalog to this catalog list
177 /// If a catalog of the same name existed before, it is replaced in the list and returned.
178 fn register_catalog(
179 &self,
180 name: String,
181 catalog: Arc<dyn CatalogProvider>,
182 ) -> Option<Arc<dyn CatalogProvider>>;
183
184 /// Retrieves the list of available catalog names
185 fn catalog_names(&self) -> Vec<String>;
186
187 /// Retrieves a specific catalog by name, provided it exists.
188 fn catalog(&self, name: &str) -> Option<Arc<dyn CatalogProvider>>;
189}
190
191impl dyn CatalogProviderList {
192 /// Returns `true` if the catalog provider list is of type `T`.
193 ///
194 /// Prefer this over `downcast_ref::<T>().is_some()`. Works correctly when
195 /// called on `Arc<dyn CatalogProviderList>` via auto-deref.
196 pub fn is<T: CatalogProviderList>(&self) -> bool {
197 (self as &dyn Any).is::<T>()
198 }
199
200 /// Attempts to downcast this catalog provider list to a concrete type `T`,
201 /// returning `None` if the provider list is not of that type.
202 ///
203 /// Works correctly when called on `Arc<dyn CatalogProviderList>` via
204 /// auto-deref, unlike `(&arc as &dyn Any).downcast_ref::<T>()` which would
205 /// attempt to downcast the `Arc` itself.
206 pub fn downcast_ref<T: CatalogProviderList>(&self) -> Option<&T> {
207 (self as &dyn Any).downcast_ref()
208 }
209}