This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 6859b93f51 refactor: move `CteWorkTable`, `default_table_source` a
bunch of files out of core (#15316)
6859b93f51 is described below
commit 6859b93f5136da4cb5af102d1799a7b8a1891973
Author: logan-keede <[email protected]>
AuthorDate: Sat Mar 22 00:40:02 2025 +0530
refactor: move `CteWorkTable`, `default_table_source` a bunch of files
out of core (#15316)
* First Iteration
* fix: CI tests
* stable waypoint, documentation pending, memory moce pending
* stable waypoint 2: add document pending, get a heads up pending
* pushing for test and review
* fix:mock in test
* fix:cliipy
---
datafusion/catalog-listing/src/helpers.rs | 4 +
.../datasource => catalog/src}/cte_worktable.rs | 13 ++--
.../src}/default_table_source.rs | 4 +-
datafusion/catalog/src/lib.rs | 2 +
.../catalog/src/{memory.rs => memory/catalog.rs} | 69 +----------------
.../src/execution => catalog/src/memory}/mod.rs | 18 +----
datafusion/catalog/src/memory/schema.rs | 89 ++++++++++++++++++++++
datafusion/catalog/src/session.rs | 3 +
datafusion/core/src/datasource/memory.rs | 5 +-
datafusion/core/src/datasource/mod.rs | 13 ++--
datafusion/core/src/execution/mod.rs | 2 -
datafusion/core/src/execution/session_state.rs | 4 +
datafusion/core/src/physical_planner.rs | 41 ++--------
datafusion/physical-expr/src/lib.rs | 5 +-
datafusion/physical-expr/src/physical_expr.rs | 36 ++++++++-
15 files changed, 171 insertions(+), 137 deletions(-)
diff --git a/datafusion/catalog-listing/src/helpers.rs
b/datafusion/catalog-listing/src/helpers.rs
index 9ac8423042..7742f5f9a1 100644
--- a/datafusion/catalog-listing/src/helpers.rs
+++ b/datafusion/catalog-listing/src/helpers.rs
@@ -1079,5 +1079,9 @@ mod tests {
fn table_options_mut(&mut self) -> &mut TableOptions {
unimplemented!()
}
+
+ fn task_ctx(&self) -> Arc<datafusion_execution::TaskContext> {
+ unimplemented!()
+ }
}
}
diff --git a/datafusion/core/src/datasource/cte_worktable.rs
b/datafusion/catalog/src/cte_worktable.rs
similarity index 93%
rename from datafusion/core/src/datasource/cte_worktable.rs
rename to datafusion/catalog/src/cte_worktable.rs
index b63755f644..d72a30909c 100644
--- a/datafusion/core/src/datasource/cte_worktable.rs
+++ b/datafusion/catalog/src/cte_worktable.rs
@@ -20,18 +20,17 @@
use std::sync::Arc;
use std::{any::Any, borrow::Cow};
+use crate::Session;
use arrow::datatypes::SchemaRef;
use async_trait::async_trait;
-use datafusion_catalog::Session;
use datafusion_physical_plan::work_table::WorkTableExec;
-use crate::{
- error::Result,
- logical_expr::{Expr, LogicalPlan, TableProviderFilterPushDown},
- physical_plan::ExecutionPlan,
-};
+use datafusion_physical_plan::ExecutionPlan;
-use crate::datasource::{TableProvider, TableType};
+use datafusion_common::error::Result;
+use datafusion_expr::{Expr, LogicalPlan, TableProviderFilterPushDown,
TableType};
+
+use crate::TableProvider;
/// The temporary working table where the previous iteration of a recursive
query is stored
/// Naming is based on PostgreSQL's implementation.
diff --git a/datafusion/core/src/datasource/default_table_source.rs
b/datafusion/catalog/src/default_table_source.rs
similarity index 98%
rename from datafusion/core/src/datasource/default_table_source.rs
rename to datafusion/catalog/src/default_table_source.rs
index 541e0b6dfa..9db8242caa 100644
--- a/datafusion/core/src/datasource/default_table_source.rs
+++ b/datafusion/catalog/src/default_table_source.rs
@@ -20,7 +20,7 @@
use std::sync::Arc;
use std::{any::Any, borrow::Cow};
-use crate::datasource::TableProvider;
+use crate::TableProvider;
use arrow::datatypes::SchemaRef;
use datafusion_common::{internal_err, Constraints};
@@ -133,7 +133,7 @@ fn preserves_table_type() {
async fn scan(
&self,
- _: &dyn datafusion_catalog::Session,
+ _: &dyn crate::Session,
_: Option<&Vec<usize>>,
_: &[Expr],
_: Option<usize>,
diff --git a/datafusion/catalog/src/lib.rs b/datafusion/catalog/src/lib.rs
index 7ba97fbc9f..a1c0a6185d 100644
--- a/datafusion/catalog/src/lib.rs
+++ b/datafusion/catalog/src/lib.rs
@@ -49,6 +49,8 @@ pub use r#async::*;
pub use schema::*;
pub use session::*;
pub use table::*;
+pub mod cte_worktable;
+pub mod default_table_source;
pub mod stream;
pub mod streaming;
pub mod view;
diff --git a/datafusion/catalog/src/memory.rs
b/datafusion/catalog/src/memory/catalog.rs
similarity index 70%
rename from datafusion/catalog/src/memory.rs
rename to datafusion/catalog/src/memory/catalog.rs
index d22a98d3d0..b71888c54e 100644
--- a/datafusion/catalog/src/memory.rs
+++ b/datafusion/catalog/src/memory/catalog.rs
@@ -18,10 +18,9 @@
//! [`MemoryCatalogProvider`], [`MemoryCatalogProviderList`]: In-memory
//! implementations of [`CatalogProviderList`] and [`CatalogProvider`].
-use crate::{CatalogProvider, CatalogProviderList, SchemaProvider,
TableProvider};
-use async_trait::async_trait;
+use crate::{CatalogProvider, CatalogProviderList, SchemaProvider};
use dashmap::DashMap;
-use datafusion_common::{exec_err, DataFusionError};
+use datafusion_common::exec_err;
use std::any::Any;
use std::sync::Arc;
@@ -134,67 +133,3 @@ impl CatalogProvider for MemoryCatalogProvider {
}
}
}
-
-/// Simple in-memory implementation of a schema.
-#[derive(Debug)]
-pub struct MemorySchemaProvider {
- tables: DashMap<String, Arc<dyn TableProvider>>,
-}
-
-impl MemorySchemaProvider {
- /// Instantiates a new MemorySchemaProvider with an empty collection of
tables.
- pub fn new() -> Self {
- Self {
- tables: DashMap::new(),
- }
- }
-}
-
-impl Default for MemorySchemaProvider {
- fn default() -> Self {
- Self::new()
- }
-}
-
-#[async_trait]
-impl SchemaProvider for MemorySchemaProvider {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
- fn table_names(&self) -> Vec<String> {
- self.tables
- .iter()
- .map(|table| table.key().clone())
- .collect()
- }
-
- async fn table(
- &self,
- name: &str,
- ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>,
DataFusionError> {
- Ok(self.tables.get(name).map(|table| Arc::clone(table.value())))
- }
-
- fn register_table(
- &self,
- name: String,
- table: Arc<dyn TableProvider>,
- ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
- if self.table_exist(name.as_str()) {
- return exec_err!("The table {name} already exists");
- }
- Ok(self.tables.insert(name, table))
- }
-
- fn deregister_table(
- &self,
- name: &str,
- ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
- Ok(self.tables.remove(name).map(|(_, table)| table))
- }
-
- fn table_exist(&self, name: &str) -> bool {
- self.tables.contains_key(name)
- }
-}
diff --git a/datafusion/core/src/execution/mod.rs
b/datafusion/catalog/src/memory/mod.rs
similarity index 68%
copy from datafusion/core/src/execution/mod.rs
copy to datafusion/catalog/src/memory/mod.rs
index 10aa16ffe4..4c5cf1a9ae 100644
--- a/datafusion/core/src/execution/mod.rs
+++ b/datafusion/catalog/src/memory/mod.rs
@@ -15,18 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-//! Shared state for query planning and execution.
+pub(crate) mod catalog;
+pub(crate) mod schema;
-pub mod context;
-pub mod session_state;
-pub use session_state::{SessionState, SessionStateBuilder};
-
-mod session_state_defaults;
-
-pub use session_state_defaults::SessionStateDefaults;
-
-// backwards compatibility
-pub use crate::datasource::file_format::options;
-
-// backwards compatibility
-pub use datafusion_execution::*;
+pub use catalog::*;
+pub use schema::*;
diff --git a/datafusion/catalog/src/memory/schema.rs
b/datafusion/catalog/src/memory/schema.rs
new file mode 100644
index 0000000000..f1b3628f7a
--- /dev/null
+++ b/datafusion/catalog/src/memory/schema.rs
@@ -0,0 +1,89 @@
+// 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.
+
+//! [`MemorySchemaProvider`]: In-memory implementations of [`SchemaProvider`].
+
+use crate::{SchemaProvider, TableProvider};
+use async_trait::async_trait;
+use dashmap::DashMap;
+use datafusion_common::{exec_err, DataFusionError};
+use std::any::Any;
+use std::sync::Arc;
+
+/// Simple in-memory implementation of a schema.
+#[derive(Debug)]
+pub struct MemorySchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl MemorySchemaProvider {
+ /// Instantiates a new MemorySchemaProvider with an empty collection of
tables.
+ pub fn new() -> Self {
+ Self {
+ tables: DashMap::new(),
+ }
+ }
+}
+
+impl Default for MemorySchemaProvider {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+#[async_trait]
+impl SchemaProvider for MemorySchemaProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn table_names(&self) -> Vec<String> {
+ self.tables
+ .iter()
+ .map(|table| table.key().clone())
+ .collect()
+ }
+
+ async fn table(
+ &self,
+ name: &str,
+ ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>,
DataFusionError> {
+ Ok(self.tables.get(name).map(|table| Arc::clone(table.value())))
+ }
+
+ fn register_table(
+ &self,
+ name: String,
+ table: Arc<dyn TableProvider>,
+ ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
+ if self.table_exist(name.as_str()) {
+ return exec_err!("The table {name} already exists");
+ }
+ Ok(self.tables.insert(name, table))
+ }
+
+ fn deregister_table(
+ &self,
+ name: &str,
+ ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
+ Ok(self.tables.remove(name).map(|(_, table)| table))
+ }
+
+ fn table_exist(&self, name: &str) -> bool {
+ self.tables.contains_key(name)
+ }
+}
diff --git a/datafusion/catalog/src/session.rs
b/datafusion/catalog/src/session.rs
index 27177e2393..88b9669cff 100644
--- a/datafusion/catalog/src/session.rs
+++ b/datafusion/catalog/src/session.rs
@@ -132,6 +132,9 @@ pub trait Session: Send + Sync {
/// Returns a mutable reference to [`TableOptions`]
fn table_options_mut(&mut self) -> &mut TableOptions;
+
+ /// Get a new TaskContext to run in this session
+ fn task_ctx(&self) -> Arc<TaskContext>;
}
/// Create a new task context instance from Session
diff --git a/datafusion/core/src/datasource/memory.rs
b/datafusion/core/src/datasource/memory.rs
index 982cb79d97..23d09719f2 100644
--- a/datafusion/core/src/datasource/memory.rs
+++ b/datafusion/core/src/datasource/memory.rs
@@ -24,7 +24,6 @@ use std::sync::Arc;
use crate::datasource::{TableProvider, TableType};
use crate::error::Result;
-use crate::execution::context::SessionState;
use crate::logical_expr::Expr;
use crate::physical_plan::insert::{DataSink, DataSinkExec};
use crate::physical_plan::repartition::RepartitionExec;
@@ -129,7 +128,7 @@ impl MemTable {
pub async fn load(
t: Arc<dyn TableProvider>,
output_partitions: Option<usize>,
- state: &SessionState,
+ state: &dyn Session,
) -> Result<Self> {
let schema = t.schema();
let constraints = t.constraints();
@@ -267,6 +266,8 @@ impl TableProvider for MemTable {
/// # Returns
///
/// * A plan that returns the number of rows written.
+ ///
+ /// [`SessionState`]: crate::execution::context::SessionState
async fn insert_into(
&self,
_state: &dyn Session,
diff --git a/datafusion/core/src/datasource/mod.rs
b/datafusion/core/src/datasource/mod.rs
index 3602a603cd..a932ae76c6 100644
--- a/datafusion/core/src/datasource/mod.rs
+++ b/datafusion/core/src/datasource/mod.rs
@@ -19,8 +19,6 @@
//!
//! [`ListingTable`]: crate::datasource::listing::ListingTable
-pub mod cte_worktable;
-pub mod default_table_source;
pub mod dynamic_file;
pub mod empty;
pub mod file_format;
@@ -32,11 +30,6 @@ pub mod provider;
mod statistics;
mod view_test;
-pub use datafusion_catalog::stream;
-pub use datafusion_catalog::view;
-pub use datafusion_datasource::schema_adapter;
-pub use datafusion_datasource::source;
-
// backwards compatibility
pub use self::default_table_source::{
provider_as_source, source_as_provider, DefaultTableSource,
@@ -45,6 +38,12 @@ pub use self::memory::MemTable;
pub use self::view::ViewTable;
pub use crate::catalog::TableProvider;
pub use crate::logical_expr::TableType;
+pub use datafusion_catalog::cte_worktable;
+pub use datafusion_catalog::default_table_source;
+pub use datafusion_catalog::stream;
+pub use datafusion_catalog::view;
+pub use datafusion_datasource::schema_adapter;
+pub use datafusion_datasource::source;
pub use datafusion_execution::object_store;
pub use datafusion_physical_expr::create_ordering;
pub use statistics::get_statistics_with_limit;
diff --git a/datafusion/core/src/execution/mod.rs
b/datafusion/core/src/execution/mod.rs
index 10aa16ffe4..2e3e09685b 100644
--- a/datafusion/core/src/execution/mod.rs
+++ b/datafusion/core/src/execution/mod.rs
@@ -27,6 +27,4 @@ pub use session_state_defaults::SessionStateDefaults;
// backwards compatibility
pub use crate::datasource::file_format::options;
-
-// backwards compatibility
pub use datafusion_execution::*;
diff --git a/datafusion/core/src/execution/session_state.rs
b/datafusion/core/src/execution/session_state.rs
index d640f8e37a..515163102c 100644
--- a/datafusion/core/src/execution/session_state.rs
+++ b/datafusion/core/src/execution/session_state.rs
@@ -266,6 +266,10 @@ impl Session for SessionState {
fn table_options_mut(&mut self) -> &mut TableOptions {
self.table_options_mut()
}
+
+ fn task_ctx(&self) -> Arc<TaskContext> {
+ self.task_ctx()
+ }
}
impl SessionState {
diff --git a/datafusion/core/src/physical_planner.rs
b/datafusion/core/src/physical_planner.rs
index bc3c1d0ac9..600a8de1de 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -39,7 +39,6 @@ use crate::physical_expr::{create_physical_expr,
create_physical_exprs};
use crate::physical_plan::aggregates::{AggregateExec, AggregateMode,
PhysicalGroupBy};
use crate::physical_plan::analyze::AnalyzeExec;
use crate::physical_plan::explain::ExplainExec;
-use crate::physical_plan::expressions::PhysicalSortExpr;
use crate::physical_plan::filter::FilterExec;
use crate::physical_plan::joins::utils as join_utils;
use crate::physical_plan::joins::{
@@ -78,7 +77,7 @@ use datafusion_expr::expr_rewriter::unnormalize_cols;
use
datafusion_expr::logical_plan::builder::wrap_projection_for_join_if_necessary;
use datafusion_expr::{
Analyze, DescribeTable, DmlStatement, Explain, ExplainFormat, Extension,
FetchType,
- Filter, JoinType, RecursiveQuery, SkipType, SortExpr, StringifiedPlan,
WindowFrame,
+ Filter, JoinType, RecursiveQuery, SkipType, StringifiedPlan, WindowFrame,
WindowFrameBound, WriteOp,
};
use datafusion_physical_expr::aggregate::{AggregateExprBuilder,
AggregateFunctionExpr};
@@ -1688,37 +1687,13 @@ pub fn create_aggregate_expr_and_maybe_filter(
)
}
-/// Create a physical sort expression from a logical expression
-pub fn create_physical_sort_expr(
- e: &SortExpr,
- input_dfschema: &DFSchema,
- execution_props: &ExecutionProps,
-) -> Result<PhysicalSortExpr> {
- let SortExpr {
- expr,
- asc,
- nulls_first,
- } = e;
- Ok(PhysicalSortExpr {
- expr: create_physical_expr(expr, input_dfschema, execution_props)?,
- options: SortOptions {
- descending: !asc,
- nulls_first: *nulls_first,
- },
- })
-}
-
-/// Create vector of physical sort expression from a vector of logical
expression
-pub fn create_physical_sort_exprs(
- exprs: &[SortExpr],
- input_dfschema: &DFSchema,
- execution_props: &ExecutionProps,
-) -> Result<LexOrdering> {
- exprs
- .iter()
- .map(|expr| create_physical_sort_expr(expr, input_dfschema,
execution_props))
- .collect::<Result<LexOrdering>>()
-}
+#[deprecated(
+ since = "47.0.0",
+ note = "use datafusion::{create_physical_sort_expr,
create_physical_sort_exprs}"
+)]
+pub use datafusion_physical_expr::{
+ create_physical_sort_expr, create_physical_sort_exprs,
+};
impl DefaultPhysicalPlanner {
/// Handles capturing the various plans for EXPLAIN queries
diff --git a/datafusion/physical-expr/src/lib.rs
b/datafusion/physical-expr/src/lib.rs
index 9abaeae440..93ced2eb62 100644
--- a/datafusion/physical-expr/src/lib.rs
+++ b/datafusion/physical-expr/src/lib.rs
@@ -54,8 +54,9 @@ pub use equivalence::{
};
pub use partitioning::{Distribution, Partitioning};
pub use physical_expr::{
- create_ordering, physical_exprs_bag_equal, physical_exprs_contains,
- physical_exprs_equal, PhysicalExprRef,
+ create_ordering, create_physical_sort_expr, create_physical_sort_exprs,
+ physical_exprs_bag_equal, physical_exprs_contains, physical_exprs_equal,
+ PhysicalExprRef,
};
pub use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
diff --git a/datafusion/physical-expr/src/physical_expr.rs
b/datafusion/physical-expr/src/physical_expr.rs
index 2221bc980f..63c4ccbb4b 100644
--- a/datafusion/physical-expr/src/physical_expr.rs
+++ b/datafusion/physical-expr/src/physical_expr.rs
@@ -17,7 +17,9 @@
use std::sync::Arc;
-use datafusion_common::HashMap;
+use crate::create_physical_expr;
+use datafusion_common::{DFSchema, HashMap};
+use datafusion_expr::execution_props::ExecutionProps;
pub(crate) use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
pub use datafusion_physical_expr_common::physical_expr::PhysicalExprRef;
use itertools::izip;
@@ -146,6 +148,38 @@ pub fn create_ordering(
Ok(all_sort_orders)
}
+/// Create a physical sort expression from a logical expression
+pub fn create_physical_sort_expr(
+ e: &SortExpr,
+ input_dfschema: &DFSchema,
+ execution_props: &ExecutionProps,
+) -> Result<PhysicalSortExpr> {
+ let SortExpr {
+ expr,
+ asc,
+ nulls_first,
+ } = e;
+ Ok(PhysicalSortExpr {
+ expr: create_physical_expr(expr, input_dfschema, execution_props)?,
+ options: SortOptions {
+ descending: !asc,
+ nulls_first: *nulls_first,
+ },
+ })
+}
+
+/// Create vector of physical sort expression from a vector of logical
expression
+pub fn create_physical_sort_exprs(
+ exprs: &[SortExpr],
+ input_dfschema: &DFSchema,
+ execution_props: &ExecutionProps,
+) -> Result<LexOrdering> {
+ exprs
+ .iter()
+ .map(|expr| create_physical_sort_expr(expr, input_dfschema,
execution_props))
+ .collect::<Result<LexOrdering>>()
+}
+
#[cfg(test)]
mod tests {
use super::*;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
