Move DataSink to datasource and add session crate (apache#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;