Skip to content

Commit 1da5252

Browse files
authored
moving memory.rs out of datafusion/core (#14332)
* moving memory.rs out of datafusion/core * fix: forgotten license * fix: taplo + documentation * fix: cargo fmt * Cargo.toml update, removing unnecessary dependency
1 parent 62000b4 commit 1da5252

File tree

14 files changed

+213
-171
lines changed

14 files changed

+213
-171
lines changed

datafusion-cli/Cargo.lock

Lines changed: 2 additions & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

datafusion/catalog/Cargo.toml

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,12 @@ version.workspace = true
3030
[dependencies]
3131
arrow-schema = { workspace = true }
3232
async-trait = { workspace = true }
33+
dashmap = { workspace = true }
3334
datafusion-common = { workspace = true }
3435
datafusion-execution = { workspace = true }
3536
datafusion-expr = { workspace = true }
3637
datafusion-physical-plan = { workspace = true }
38+
itertools = { workspace = true }
3739
parking_lot = { workspace = true }
3840

3941
[dev-dependencies]

datafusion/catalog/src/lib.rs

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,16 @@
1515
// specific language governing permissions and limitations
1616
// under the License.
1717

18+
//! Interfaces and default implementations of catalogs and schemas.
19+
//!
20+
//! Implementations
21+
//! * Simple memory based catalog: [`MemoryCatalogProviderList`], [`MemoryCatalogProvider`], [`MemorySchemaProvider`]
22+
23+
pub mod memory;
24+
pub use memory::{
25+
MemoryCatalogProvider, MemoryCatalogProviderList, MemorySchemaProvider,
26+
};
27+
1828
mod r#async;
1929
mod catalog;
2030
mod dynamic_file;

datafusion/core/src/catalog_common/memory.rs renamed to datafusion/catalog/src/memory.rs

Lines changed: 1 addition & 156 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,7 @@
1818
//! [`MemoryCatalogProvider`], [`MemoryCatalogProviderList`]: In-memory
1919
//! implementations of [`CatalogProviderList`] and [`CatalogProvider`].
2020
21-
use crate::catalog::{
22-
CatalogProvider, CatalogProviderList, SchemaProvider, TableProvider,
23-
};
21+
use crate::{CatalogProvider, CatalogProviderList, SchemaProvider, TableProvider};
2422
use async_trait::async_trait;
2523
use dashmap::DashMap;
2624
use datafusion_common::{exec_err, DataFusionError};
@@ -200,156 +198,3 @@ impl SchemaProvider for MemorySchemaProvider {
200198
self.tables.contains_key(name)
201199
}
202200
}
203-
204-
#[cfg(test)]
205-
mod test {
206-
use super::*;
207-
use crate::catalog::CatalogProvider;
208-
use crate::catalog_common::memory::MemorySchemaProvider;
209-
use crate::datasource::empty::EmptyTable;
210-
use crate::datasource::listing::{ListingTable, ListingTableConfig, ListingTableUrl};
211-
use crate::prelude::SessionContext;
212-
use arrow_schema::Schema;
213-
use datafusion_common::assert_batches_eq;
214-
use std::any::Any;
215-
use std::sync::Arc;
216-
217-
#[test]
218-
fn memory_catalog_dereg_nonempty_schema() {
219-
let cat = Arc::new(MemoryCatalogProvider::new()) as Arc<dyn CatalogProvider>;
220-
221-
let schema = Arc::new(MemorySchemaProvider::new()) as Arc<dyn SchemaProvider>;
222-
let test_table = Arc::new(EmptyTable::new(Arc::new(Schema::empty())))
223-
as Arc<dyn TableProvider>;
224-
schema.register_table("t".into(), test_table).unwrap();
225-
226-
cat.register_schema("foo", schema.clone()).unwrap();
227-
228-
assert!(
229-
cat.deregister_schema("foo", false).is_err(),
230-
"dropping empty schema without cascade should error"
231-
);
232-
assert!(cat.deregister_schema("foo", true).unwrap().is_some());
233-
}
234-
235-
#[test]
236-
fn memory_catalog_dereg_empty_schema() {
237-
let cat = Arc::new(MemoryCatalogProvider::new()) as Arc<dyn CatalogProvider>;
238-
239-
let schema = Arc::new(MemorySchemaProvider::new()) as Arc<dyn SchemaProvider>;
240-
cat.register_schema("foo", schema).unwrap();
241-
242-
assert!(cat.deregister_schema("foo", false).unwrap().is_some());
243-
}
244-
245-
#[test]
246-
fn memory_catalog_dereg_missing() {
247-
let cat = Arc::new(MemoryCatalogProvider::new()) as Arc<dyn CatalogProvider>;
248-
assert!(cat.deregister_schema("foo", false).unwrap().is_none());
249-
}
250-
251-
#[test]
252-
fn default_register_schema_not_supported() {
253-
// mimic a new CatalogProvider and ensure it does not support registering schemas
254-
#[derive(Debug)]
255-
struct TestProvider {}
256-
impl CatalogProvider for TestProvider {
257-
fn as_any(&self) -> &dyn Any {
258-
self
259-
}
260-
261-
fn schema_names(&self) -> Vec<String> {
262-
unimplemented!()
263-
}
264-
265-
fn schema(&self, _name: &str) -> Option<Arc<dyn SchemaProvider>> {
266-
unimplemented!()
267-
}
268-
}
269-
270-
let schema = Arc::new(MemorySchemaProvider::new()) as Arc<dyn SchemaProvider>;
271-
let catalog = Arc::new(TestProvider {});
272-
273-
match catalog.register_schema("foo", schema) {
274-
Ok(_) => panic!("unexpected OK"),
275-
Err(e) => assert_eq!(e.strip_backtrace(), "This feature is not implemented: Registering new schemas is not supported"),
276-
};
277-
}
278-
279-
#[tokio::test]
280-
async fn test_mem_provider() {
281-
let provider = MemorySchemaProvider::new();
282-
let table_name = "test_table_exist";
283-
assert!(!provider.table_exist(table_name));
284-
assert!(provider.deregister_table(table_name).unwrap().is_none());
285-
let test_table = EmptyTable::new(Arc::new(Schema::empty()));
286-
// register table successfully
287-
assert!(provider
288-
.register_table(table_name.to_string(), Arc::new(test_table))
289-
.unwrap()
290-
.is_none());
291-
assert!(provider.table_exist(table_name));
292-
let other_table = EmptyTable::new(Arc::new(Schema::empty()));
293-
let result =
294-
provider.register_table(table_name.to_string(), Arc::new(other_table));
295-
assert!(result.is_err());
296-
}
297-
298-
#[tokio::test]
299-
async fn test_schema_register_listing_table() {
300-
let testdata = crate::test_util::parquet_test_data();
301-
let testdir = if testdata.starts_with('/') {
302-
format!("file://{testdata}")
303-
} else {
304-
format!("file:///{testdata}")
305-
};
306-
let filename = if testdir.ends_with('/') {
307-
format!("{}{}", testdir, "alltypes_plain.parquet")
308-
} else {
309-
format!("{}/{}", testdir, "alltypes_plain.parquet")
310-
};
311-
312-
let table_path = ListingTableUrl::parse(filename).unwrap();
313-
314-
let catalog = MemoryCatalogProvider::new();
315-
let schema = MemorySchemaProvider::new();
316-
317-
let ctx = SessionContext::new();
318-
319-
let config = ListingTableConfig::new(table_path)
320-
.infer(&ctx.state())
321-
.await
322-
.unwrap();
323-
let table = ListingTable::try_new(config).unwrap();
324-
325-
schema
326-
.register_table("alltypes_plain".to_string(), Arc::new(table))
327-
.unwrap();
328-
329-
catalog.register_schema("active", Arc::new(schema)).unwrap();
330-
ctx.register_catalog("cat", Arc::new(catalog));
331-
332-
let df = ctx
333-
.sql("SELECT id, bool_col FROM cat.active.alltypes_plain")
334-
.await
335-
.unwrap();
336-
337-
let actual = df.collect().await.unwrap();
338-
339-
let expected = [
340-
"+----+----------+",
341-
"| id | bool_col |",
342-
"+----+----------+",
343-
"| 4 | true |",
344-
"| 5 | false |",
345-
"| 6 | true |",
346-
"| 7 | false |",
347-
"| 2 | true |",
348-
"| 3 | false |",
349-
"| 0 | true |",
350-
"| 1 | false |",
351-
"+----+----------+",
352-
];
353-
assert_batches_eq!(expected, &actual);
354-
}
355-
}

datafusion/core/Cargo.toml

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,6 @@ async-trait = { workspace = true }
9999
bytes = { workspace = true }
100100
bzip2 = { version = "0.5.0", optional = true }
101101
chrono = { workspace = true }
102-
dashmap = { workspace = true }
103102
datafusion-catalog = { workspace = true }
104103
datafusion-common = { workspace = true, features = ["object_store"] }
105104
datafusion-common-runtime = { workspace = true }

datafusion/core/src/catalog_common/mod.rs

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -18,18 +18,12 @@
1818
//! Interfaces and default implementations of catalogs and schemas.
1919
//!
2020
//! Implementations
21-
//! * Simple memory based catalog: [`MemoryCatalogProviderList`], [`MemoryCatalogProvider`], [`MemorySchemaProvider`]
2221
//! * Information schema: [`information_schema`]
2322
//! * Listing schema: [`listing_schema`]
2423
2524
pub mod information_schema;
2625
pub mod listing_schema;
27-
pub mod memory;
28-
2926
pub use crate::catalog::{CatalogProvider, CatalogProviderList, SchemaProvider};
30-
pub use memory::{
31-
MemoryCatalogProvider, MemoryCatalogProviderList, MemorySchemaProvider,
32-
};
3327

3428
pub use datafusion_sql::{ResolvedTableReference, TableReference};
3529

datafusion/core/src/execution/context/mod.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
//! [`SessionContext`] API for registering data sources and executing queries
1919
20+
use datafusion_catalog::memory::MemorySchemaProvider;
21+
use datafusion_catalog::MemoryCatalogProvider;
2022
use std::collections::HashSet;
2123
use std::fmt::Debug;
2224
use std::sync::{Arc, Weak};
@@ -27,8 +29,6 @@ use crate::{
2729
CatalogProvider, CatalogProviderList, TableProvider, TableProviderFactory,
2830
},
2931
catalog_common::listing_schema::ListingSchemaProvider,
30-
catalog_common::memory::MemorySchemaProvider,
31-
catalog_common::MemoryCatalogProvider,
3232
dataframe::DataFrame,
3333
datasource::listing::{
3434
ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl,

datafusion/core/src/execution/session_state.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,13 +27,13 @@ use crate::catalog::{CatalogProviderList, SchemaProvider, TableProviderFactory};
2727
use crate::catalog_common::information_schema::{
2828
InformationSchemaProvider, INFORMATION_SCHEMA,
2929
};
30-
use crate::catalog_common::MemoryCatalogProviderList;
3130
use crate::datasource::cte_worktable::CteWorkTable;
3231
use crate::datasource::file_format::{format_as_file_type, FileFormatFactory};
3332
use crate::datasource::provider_as_source;
3433
use crate::execution::context::{EmptySerializerRegistry, FunctionFactory, QueryPlanner};
3534
use crate::execution::SessionStateDefaults;
3635
use crate::physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner};
36+
use datafusion_catalog::MemoryCatalogProviderList;
3737

3838
use arrow_schema::{DataType, SchemaRef};
3939
use datafusion_catalog::{Session, TableFunction, TableFunctionImpl};
@@ -1987,11 +1987,11 @@ pub(crate) struct PreparedPlan {
19871987
#[cfg(test)]
19881988
mod tests {
19891989
use super::{SessionContextProvider, SessionStateBuilder};
1990-
use crate::catalog_common::MemoryCatalogProviderList;
19911990
use crate::datasource::MemTable;
19921991
use crate::execution::context::SessionState;
19931992
use arrow_array::{ArrayRef, Int32Array, RecordBatch, StringArray};
19941993
use arrow_schema::{DataType, Field, Schema};
1994+
use datafusion_catalog::MemoryCatalogProviderList;
19951995
use datafusion_common::DFSchema;
19961996
use datafusion_common::Result;
19971997
use datafusion_execution::config::SessionConfig;

datafusion/core/src/execution/session_state_defaults.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717

1818
use crate::catalog::{CatalogProvider, TableProviderFactory};
1919
use crate::catalog_common::listing_schema::ListingSchemaProvider;
20-
use crate::catalog_common::{MemoryCatalogProvider, MemorySchemaProvider};
2120
use crate::datasource::file_format::arrow::ArrowFormatFactory;
2221
use crate::datasource::file_format::avro::AvroFormatFactory;
2322
use crate::datasource::file_format::csv::CsvFormatFactory;
@@ -31,6 +30,7 @@ use crate::execution::context::SessionState;
3130
use crate::functions_nested;
3231
use crate::{functions, functions_aggregate, functions_table, functions_window};
3332
use datafusion_catalog::TableFunction;
33+
use datafusion_catalog::{MemoryCatalogProvider, MemorySchemaProvider};
3434
use datafusion_execution::config::SessionConfig;
3535
use datafusion_execution::object_store::ObjectStoreUrl;
3636
use datafusion_execution::runtime_env::RuntimeEnv;

0 commit comments

Comments
 (0)