Move DataSink to datasource and add session crate (#15371)

* session

* clippy

* fmt

* session

* minor

* Update README.md

* doc

* fmt

* doc

---------

Co-authored-by: berkaysynnada <[email protected]>

Author: jayzhan-synnada

Cargo.lock

@@ -47,6 +47,7 @@ members = [
     "datafusion/proto/gen",
     "datafusion/proto-common",
     "datafusion/proto-common/gen",
+    "datafusion/session",
     "datafusion/sql",
     "datafusion/sqllogictest",
     "datafusion/substrait",
@@ -136,6 +137,7 @@ datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", versio
 datafusion-physical-plan = { path = "datafusion/physical-plan", version = "46.0.1" }
 datafusion-proto = { path = "datafusion/proto", version = "46.0.1" }
 datafusion-proto-common = { path = "datafusion/proto-common", version = "46.0.1" }
+datafusion-session = { path = "datafusion/session", version = "46.0.1" }
 datafusion-sql = { path = "datafusion/sql", version = "46.0.1" }
 doc-comment = "0.3"
 env_logger = "0.11"

Cargo.toml

@@ -41,6 +41,7 @@ datafusion-expr = { workspace = true }
 datafusion-physical-expr = { workspace = true }
 datafusion-physical-expr-common = { workspace = true }
 datafusion-physical-plan = { workspace = true }
+datafusion-session = { workspace = true }
 futures = { workspace = true }
 log = { workspace = true }
 object_store = { workspace = true }

datafusion/catalog-listing/Cargo.toml

@@ -36,10 +36,12 @@ async-trait = { workspace = true }
 dashmap = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-common-runtime = { workspace = true }
+datafusion-datasource = { workspace = true }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
 datafusion-physical-expr = { workspace = true }
 datafusion-physical-plan = { workspace = true }
+datafusion-session = { workspace = true }
 datafusion-sql = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }

datafusion/catalog/Cargo.toml

@@ -31,26 +31,32 @@
 //! * Simple memory based catalog: [`MemoryCatalogProviderList`], [`MemoryCatalogProvider`], [`MemorySchemaProvider`]
 //! * Listing schema: [`listing_schema`]
 
+pub mod cte_worktable;
+pub mod default_table_source;
+pub mod information_schema;
+pub mod listing_schema;
 pub mod memory;
-pub use memory::{
-    MemoryCatalogProvider, MemoryCatalogProviderList, MemorySchemaProvider,
-};
+pub mod stream;
+pub mod streaming;
+pub mod view;
+
 mod r#async;
 mod catalog;
 mod dynamic_file;
-pub mod information_schema;
-pub mod listing_schema;
 mod schema;
-mod session;
 mod table;
+
 pub use catalog::*;
+pub use datafusion_session::Session;
 pub use dynamic_file::catalog::*;
+pub use memory::{
+    MemoryCatalogProvider, MemoryCatalogProviderList, MemorySchemaProvider,
+};
 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;
+
+// For backwards compatibility,
+mod session {
+    pub use datafusion_session::Session;
+}

datafusion/catalog/src/lib.rs

@@ -30,11 +30,11 @@ use arrow::array::{RecordBatch, RecordBatchReader, RecordBatchWriter};
 use arrow::datatypes::SchemaRef;
 use datafusion_common::{config_err, plan_err, Constraints, DataFusionError, Result};
 use datafusion_common_runtime::SpawnedTask;
+use datafusion_datasource::sink::{DataSink, DataSinkExec};
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
 use datafusion_expr::dml::InsertOp;
 use datafusion_expr::{CreateExternalTable, Expr, SortExpr, TableType};
 use datafusion_physical_expr::create_ordering;
-use datafusion_physical_plan::insert::{DataSink, DataSinkExec};
 use datafusion_physical_plan::stream::RecordBatchReceiverStreamBuilder;
 use datafusion_physical_plan::streaming::{PartitionStream, StreamingTableExec};
 use datafusion_physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan};

datafusion/catalog/src/stream.rs

@@ -288,7 +288,7 @@ pub trait TableProvider: Debug + Sync + Send {
     /// See [`DataSinkExec`] for the common pattern of inserting a
     /// streams of `RecordBatch`es as files to an ObjectStore.
     ///
-    /// [`DataSinkExec`]: datafusion_physical_plan::insert::DataSinkExec
+    /// [`DataSinkExec`]: datafusion_datasource::sink::DataSinkExec
     async fn insert_into(
         &self,
         _state: &dyn Session,

datafusion/catalog/src/table.rs

@@ -123,6 +123,7 @@ datafusion-physical-expr = { workspace = true }
 datafusion-physical-expr-common = { workspace = true }
 datafusion-physical-optimizer = { workspace = true }
 datafusion-physical-plan = { workspace = true }
+datafusion-session = { workspace = true }
 datafusion-sql = { workspace = true }
 flate2 = { version = "1.1.0", optional = true }
 futures = { workspace = true }

datafusion/core/Cargo.toml

@@ -20,15 +20,17 @@
 
 use std::sync::Arc;
 
-use async_trait::async_trait;
-use datafusion_catalog::{SessionStore, UrlTableFactory};
-use datafusion_common::plan_datafusion_err;
-
 use crate::datasource::listing::{ListingTable, ListingTableConfig, ListingTableUrl};
 use crate::datasource::TableProvider;
 use crate::error::Result;
 use crate::execution::context::SessionState;
 
+use datafusion_catalog::UrlTableFactory;
+use datafusion_common::plan_datafusion_err;
+use datafusion_session::SessionStore;
+
+use async_trait::async_trait;
+
 /// [DynamicListTableFactory] is a factory that can create a [ListingTable] from the given url.
 #[derive(Default, Debug)]
 pub struct DynamicListTableFactory {

datafusion/core/src/datasource/dynamic_file.rs

@@ -50,11 +50,11 @@ use datafusion_common_runtime::{JoinSet, SpawnedTask};
 use datafusion_datasource::display::FileGroupDisplay;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
+use datafusion_datasource::sink::{DataSink, DataSinkExec};
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
 use datafusion_expr::dml::InsertOp;
 use datafusion_physical_expr::PhysicalExpr;
 use datafusion_physical_expr_common::sort_expr::LexRequirement;
-use datafusion_physical_plan::insert::{DataSink, DataSinkExec};
 
 use async_trait::async_trait;
 use bytes::Bytes;

datafusion/core/src/datasource/file_format/arrow.rs

@@ -25,7 +25,6 @@ use std::sync::Arc;
 use crate::datasource::{TableProvider, TableType};
 use crate::error::Result;
 use crate::logical_expr::Expr;
-use crate::physical_plan::insert::{DataSink, DataSinkExec};
 use crate::physical_plan::repartition::RepartitionExec;
 use crate::physical_plan::{
     common, DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties,
@@ -39,6 +38,7 @@ use datafusion_catalog::Session;
 use datafusion_common::{not_impl_err, plan_err, Constraints, DFSchema, SchemaExt};
 use datafusion_common_runtime::JoinSet;
 pub use datafusion_datasource::memory::MemorySourceConfig;
+use datafusion_datasource::sink::{DataSink, DataSinkExec};
 pub use datafusion_datasource::source::DataSourceExec;
 use datafusion_execution::TaskContext;
 use datafusion_expr::dml::InsertOp;

datafusion/core/src/datasource/memory.rs

@@ -43,6 +43,7 @@ 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::sink;
 pub use datafusion_datasource::source;
 pub use datafusion_execution::object_store;
 pub use datafusion_physical_expr::create_ordering;

datafusion/core/src/datasource/mod.rs

@@ -17,13 +17,13 @@
 
 //! [`SessionContext`] API for registering data sources and executing queries
 
-use datafusion_catalog::memory::MemorySchemaProvider;
-use datafusion_catalog::MemoryCatalogProvider;
 use std::collections::HashSet;
 use std::fmt::Debug;
 use std::sync::{Arc, Weak};
 
 use super::options::ReadOptions;
+use crate::datasource::dynamic_file::DynamicListTableFactory;
+use crate::execution::session_state::SessionStateBuilder;
 use crate::{
     catalog::listing_schema::ListingSchemaProvider,
     catalog::{
@@ -49,39 +49,40 @@ use crate::{
     variable::{VarProvider, VarType},
 };
 
+// backwards compatibility
+pub use crate::execution::session_state::SessionState;
+
 use arrow::datatypes::{Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
+use datafusion_catalog::memory::MemorySchemaProvider;
+use datafusion_catalog::MemoryCatalogProvider;
+use datafusion_catalog::{
+    DynamicFileCatalog, TableFunction, TableFunctionImpl, UrlTableFactory,
+};
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::{
     config::{ConfigExtension, TableOptions},
     exec_datafusion_err, exec_err, not_impl_err, plan_datafusion_err, plan_err,
     tree_node::{TreeNodeRecursion, TreeNodeVisitor},
     DFSchema, ParamValues, ScalarValue, SchemaReference, TableReference,
 };
+pub use datafusion_execution::config::SessionConfig;
 use datafusion_execution::registry::SerializerRegistry;
+pub use datafusion_execution::TaskContext;
+pub use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::{
     expr_rewriter::FunctionRewrite,
     logical_plan::{DdlStatement, Statement},
     planner::ExprPlanner,
     Expr, UserDefinedLogicalNode, WindowUDF,
 };
-
-// backwards compatibility
-pub use crate::execution::session_state::SessionState;
-
-use crate::datasource::dynamic_file::DynamicListTableFactory;
-use crate::execution::session_state::SessionStateBuilder;
-use async_trait::async_trait;
-use chrono::{DateTime, Utc};
-use datafusion_catalog::{
-    DynamicFileCatalog, SessionStore, TableFunction, TableFunctionImpl, UrlTableFactory,
-};
-use datafusion_common::config::ConfigOptions;
-pub use datafusion_execution::config::SessionConfig;
-pub use datafusion_execution::TaskContext;
-pub use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_optimizer::analyzer::type_coercion::TypeCoercion;
 use datafusion_optimizer::Analyzer;
 use datafusion_optimizer::{AnalyzerRule, OptimizerRule};
+use datafusion_session::SessionStore;
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
 use object_store::ObjectStore;
 use parking_lot::RwLock;
 use url::Url;

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

@@ -33,10 +33,10 @@ use crate::physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner};
 use datafusion_catalog::information_schema::{
     InformationSchemaProvider, INFORMATION_SCHEMA,
 };
-use datafusion_catalog::MemoryCatalogProviderList;
 
 use arrow::datatypes::{DataType, SchemaRef};
-use datafusion_catalog::{Session, TableFunction, TableFunctionImpl};
+use datafusion_catalog::MemoryCatalogProviderList;
+use datafusion_catalog::{TableFunction, TableFunctionImpl};
 use datafusion_common::alias::AliasGenerator;
 use datafusion_common::config::{ConfigExtension, ConfigOptions, TableOptions};
 use datafusion_common::display::{PlanType, StringifiedPlan, ToStringifiedPlan};
@@ -68,6 +68,7 @@ use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_optimizer::optimizer::PhysicalOptimizer;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
 use datafusion_physical_plan::ExecutionPlan;
+use datafusion_session::Session;
 use datafusion_sql::parser::{DFParserBuilder, Statement};
 use datafusion_sql::planner::{ContextProvider, ParserOptions, PlannerContext, SqlToRel};
 

datafusion/core/src/execution/session_state.rs

@@ -649,6 +649,8 @@
 //!
 //! * [datafusion_common]: Common traits and types
 //! * [datafusion_catalog]: Catalog APIs such as [`SchemaProvider`] and [`CatalogProvider`]
+//! * [datafusion_datasource]: File and Data IO such as [`FileSource`] and [`DataSink`]
+//! * [datafusion_session]: [`Session`] and related structures
 //! * [datafusion_execution]: State and structures needed for execution
 //! * [datafusion_expr]: [`LogicalPlan`], [`Expr`] and related logical planning structure
 //! * [datafusion_functions]: Scalar function packages
@@ -664,6 +666,9 @@
 //!
 //! [`SchemaProvider`]: datafusion_catalog::SchemaProvider
 //! [`CatalogProvider`]: datafusion_catalog::CatalogProvider
+//! [`Session`]: datafusion_session::Session
+//! [`FileSource`]: datafusion_datasource::file::FileSource
+//! [`DataSink`]: datafusion_datasource::sink::DataSink
 //!
 //! ## Citing DataFusion in Academic Papers
 //!

datafusion/core/src/lib.rs

@@ -43,6 +43,7 @@ datafusion-execution = { workspace = true }
 datafusion-physical-expr = { workspace = true }
 datafusion-physical-expr-common = { workspace = true }
 datafusion-physical-plan = { workspace = true }
+datafusion-session = { workspace = true }
 futures = { workspace = true }
 num-traits = { version = "0.2" }
 object_store = { workspace = true }