refactor(connector): rename source crate to dml

Cargo.toml

@@ -13,6 +13,7 @@ members = [
   "src/connector",
   "src/connector/with_options",
   "src/ctl",
+  "src/dml",
   "src/error",
   "src/expr/core",
   "src/expr/impl",
@@ -33,7 +34,6 @@ members = [
   "src/risedevtool",
   "src/risedevtool/config",
   "src/rpc_client",
-  "src/source",
   "src/sqlparser",
   "src/sqlparser/test_runner",
   "src/storage",
@@ -164,6 +164,7 @@ risingwave_compactor = { path = "./src/storage/compactor" }
 risingwave_compute = { path = "./src/compute" }
 risingwave_ctl = { path = "./src/ctl" }
 risingwave_connector = { path = "./src/connector" }
+risingwave_dml = { path = "./src/dml" }
 risingwave_error = { path = "./src/error" }
 risingwave_expr = { path = "./src/expr/core" }
 risingwave_expr_impl = { path = "./src/expr/impl" }
@@ -181,7 +182,6 @@ risingwave_object_store = { path = "./src/object_store" }
 risingwave_pb = { path = "./src/prost" }
 risingwave_rpc_client = { path = "./src/rpc_client" }
 risingwave_rt = { path = "./src/utils/runtime" }
-risingwave_source = { path = "./src/source" }
 risingwave_sqlparser = { path = "./src/sqlparser" }
 risingwave_sqlsmith = { path = "./src/tests/sqlsmith" }
 risingwave_storage = { path = "./src/storage" }

Makefile.toml

@@ -919,10 +919,10 @@ cargo build \
   -p risingwave_common \
   -p risingwave_compute \
   -p risingwave_connector \
+  -p risingwave_dml \
   -p risingwave_frontend \
   -p risingwave_meta \
   -p risingwave_object_store \
-  -p risingwave_source \
   -p risingwave_storage \
   -p risingwave_stream \
   -p pgwire \
@@ -944,10 +944,10 @@ cargo nextest run \
   -p risingwave_common \
   -p risingwave_compute \
   -p risingwave_connector \
+  -p risingwave_dml \
   -p risingwave_frontend \
   -p risingwave_meta \
   -p risingwave_object_store \
-  -p risingwave_source \
   -p risingwave_storage \
   -p risingwave_stream \
   -p pgwire \
@@ -1000,10 +1000,10 @@ cargo check \
   -p risingwave_common \
   -p risingwave_compute \
   -p risingwave_connector \
+  -p risingwave_dml \
   -p risingwave_frontend \
   -p risingwave_meta \
   -p risingwave_object_store \
-  -p risingwave_source \
   -p risingwave_storage \
   -p risingwave_stream \
   -p pgwire \

docs/rustdoc/index.md

@@ -30,9 +30,9 @@ There's also a [Dash](https://kapeli.com/dash) docset feed containing all crate
   - [risingwave_hummock_sdk](risingwave_hummock_sdk/index.html)
   - [risingwave_object_store](risingwave_object_store/index.html)
 
-### Source and Connector
+### DML and Connectors (Source and Sink)
 
-- [risingwave_source](risingwave_source/index.html)
+- [risingwave_dml](risingwave_dml/index.html)
 - [risingwave_connector](risingwave_connector/index.html)
 
 ### Common

src/batch/Cargo.toml

@@ -31,11 +31,11 @@ paste = "1"
 prometheus = { version = "0.13", features = ["process"] }
 risingwave_common = { workspace = true }
 risingwave_connector = { workspace = true }
+risingwave_dml = { workspace = true }
 risingwave_expr = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
-risingwave_source = { workspace = true }
 risingwave_storage = { workspace = true }
 rw_futures_util = { workspace = true }
 scopeguard = "1"

src/batch/src/executor/delete.rs

@@ -21,8 +21,8 @@ use risingwave_common::catalog::{Field, Schema, TableId, TableVersionId};
 use risingwave_common::transaction::transaction_id::TxnId;
 use risingwave_common::types::DataType;
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
+use risingwave_dml::dml_manager::DmlManagerRef;
 use risingwave_pb::batch_plan::plan_node::NodeBody;
-use risingwave_source::dml_manager::DmlManagerRef;
 
 use crate::error::{BatchError, Result};
 use crate::executor::{
@@ -201,7 +201,7 @@ mod tests {
         schema_test_utils, ColumnDesc, ColumnId, INITIAL_TABLE_VERSION_ID,
     };
     use risingwave_common::test_prelude::DataChunkTestExt;
-    use risingwave_source::dml_manager::DmlManager;
+    use risingwave_dml::dml_manager::DmlManager;
 
     use super::*;
     use crate::executor::test_utils::MockExecutor;

src/batch/src/executor/insert.rs

@@ -25,10 +25,10 @@ use risingwave_common::catalog::{Field, Schema, TableId, TableVersionId};
 use risingwave_common::transaction::transaction_id::TxnId;
 use risingwave_common::types::DataType;
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
+use risingwave_dml::dml_manager::DmlManagerRef;
 use risingwave_expr::expr::{build_from_prost, BoxedExpression};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 use risingwave_pb::plan_common::IndexAndExpr;
-use risingwave_source::dml_manager::DmlManagerRef;
 
 use crate::error::{BatchError, Result};
 use crate::executor::{
@@ -276,7 +276,7 @@ mod tests {
     };
     use risingwave_common::transaction::transaction_message::TxnMsg;
     use risingwave_common::types::{DataType, StructType};
-    use risingwave_source::dml_manager::DmlManager;
+    use risingwave_dml::dml_manager::DmlManager;
     use risingwave_storage::hummock::CachePolicy;
     use risingwave_storage::memory::MemoryStateStore;
     use risingwave_storage::store::{ReadOptions, StateStoreReadExt};

src/batch/src/executor/source.rs

@@ -22,19 +22,19 @@ use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema, TableId};
 use risingwave_common::types::DataType;
 use risingwave_connector::parser::SpecificParserConfig;
 use risingwave_connector::source::monitor::SourceMetrics;
+use risingwave_connector::source::reader::reader::SourceReader;
 use risingwave_connector::source::{
     ConnectorProperties, SourceColumnDesc, SourceContext, SourceCtrlOpts, SplitImpl, SplitMetaData,
 };
 use risingwave_pb::batch_plan::plan_node::NodeBody;
-use risingwave_source::connector_source::ConnectorSource;
 
 use super::Executor;
 use crate::error::{BatchError, Result};
 use crate::executor::{BoxedExecutor, BoxedExecutorBuilder, ExecutorBuilder};
 use crate::task::BatchTaskContext;
 
 pub struct SourceExecutor {
-    connector_source: ConnectorSource,
+    source: SourceReader,
 
     // used to create reader
     column_ids: Vec<ColumnId>,
@@ -75,7 +75,7 @@ impl BoxedExecutorBuilder for SourceExecutor {
             .map(|c| SourceColumnDesc::from(&ColumnDesc::from(c.column_desc.as_ref().unwrap())))
             .collect();
 
-        let connector_source = ConnectorSource {
+        let source_reader = SourceReader {
             config,
             columns,
             parser_config,
@@ -110,7 +110,7 @@ impl BoxedExecutorBuilder for SourceExecutor {
         let schema = Schema::new(fields);
 
         Ok(Box::new(SourceExecutor {
-            connector_source,
+            source: source_reader,
             column_ids,
             metrics: source.context().source_metrics(),
             source_id: TableId::new(source_node.source_id),
@@ -149,8 +149,8 @@ impl SourceExecutor {
             ConnectorProperties::default(),
         ));
         let stream = self
-            .connector_source
-            .stream_reader(Some(vec![self.split]), self.column_ids, source_ctx)
+            .source
+            .to_stream(Some(vec![self.split]), self.column_ids, source_ctx)
             .await?;
 
         #[for_await]

src/batch/src/executor/update.rs

@@ -22,9 +22,9 @@ use risingwave_common::transaction::transaction_id::TxnId;
 use risingwave_common::types::DataType;
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::iter_util::ZipEqDebug;
+use risingwave_dml::dml_manager::DmlManagerRef;
 use risingwave_expr::expr::{build_from_prost, BoxedExpression};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
-use risingwave_source::dml_manager::DmlManagerRef;
 
 use crate::error::{BatchError, Result};
 use crate::executor::{
@@ -136,7 +136,7 @@ impl UpdateExecutor {
 
         let mut builder = DataChunkBuilder::new(data_types, self.chunk_size);
 
-        let mut write_handle: risingwave_source::WriteHandle =
+        let mut write_handle: risingwave_dml::WriteHandle =
             table_dml_handle.write_handle(self.session_id, self.txn_id)?;
         write_handle.begin()?;
 
@@ -264,8 +264,8 @@ mod tests {
         schema_test_utils, ColumnDesc, ColumnId, INITIAL_TABLE_VERSION_ID,
     };
     use risingwave_common::test_prelude::DataChunkTestExt;
+    use risingwave_dml::dml_manager::DmlManager;
     use risingwave_expr::expr::InputRefExpression;
-    use risingwave_source::dml_manager::DmlManager;
 
     use super::*;
     use crate::executor::test_utils::MockExecutor;

src/batch/src/task/context.rs

@@ -19,8 +19,8 @@ use risingwave_common::config::BatchConfig;
 use risingwave_common::memory::MemoryContext;
 use risingwave_common::util::addr::{is_local_address, HostAddr};
 use risingwave_connector::source::monitor::SourceMetrics;
+use risingwave_dml::dml_manager::DmlManagerRef;
 use risingwave_rpc_client::ComputeClientPoolRef;
-use risingwave_source::dml_manager::DmlManagerRef;
 use risingwave_storage::StateStoreImpl;
 
 use super::TaskId;

src/batch/src/task/env.rs

@@ -18,8 +18,8 @@ use risingwave_common::config::BatchConfig;
 use risingwave_common::util::addr::HostAddr;
 use risingwave_common::util::worker_util::WorkerNodeId;
 use risingwave_connector::source::monitor::SourceMetrics;
+use risingwave_dml::dml_manager::DmlManagerRef;
 use risingwave_rpc_client::ComputeClientPoolRef;
-use risingwave_source::dml_manager::DmlManagerRef;
 use risingwave_storage::StateStoreImpl;
 
 use crate::monitor::{BatchExecutorMetrics, BatchManagerMetrics, BatchTaskMetrics};
@@ -91,8 +91,8 @@ impl BatchEnvironment {
     // Create an instance for testing purpose.
     #[cfg(test)]
     pub fn for_test() -> Self {
+        use risingwave_dml::dml_manager::DmlManager;
         use risingwave_rpc_client::ComputeClientPool;
-        use risingwave_source::dml_manager::DmlManager;
         use risingwave_storage::monitor::MonitoredStorageMetrics;
 
         BatchEnvironment {

src/compute/Cargo.toml

@@ -32,11 +32,11 @@ risingwave_common = { workspace = true }
 risingwave_common_heap_profiling = { workspace = true }
 risingwave_common_service = { workspace = true }
 risingwave_connector = { workspace = true }
+risingwave_dml = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_jni_core = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
-risingwave_source = { workspace = true }
 risingwave_storage = { workspace = true }
 risingwave_stream = { workspace = true }
 serde = { version = "1", features = ["derive"] }

src/compute/src/server.rs

@@ -39,6 +39,7 @@ use risingwave_common_service::metrics_manager::MetricsManager;
 use risingwave_common_service::observer_manager::ObserverManager;
 use risingwave_common_service::tracing::TracingExtractLayer;
 use risingwave_connector::source::monitor::GLOBAL_SOURCE_METRICS;
+use risingwave_dml::dml_manager::DmlManager;
 use risingwave_pb::common::WorkerType;
 use risingwave_pb::compute::config_service_server::ConfigServiceServer;
 use risingwave_pb::connector_service::SinkPayloadFormat;
@@ -49,7 +50,6 @@ use risingwave_pb::stream_service::stream_service_server::StreamServiceServer;
 use risingwave_pb::task_service::exchange_service_server::ExchangeServiceServer;
 use risingwave_pb::task_service::task_service_server::TaskServiceServer;
 use risingwave_rpc_client::{ComputeClientPool, ConnectorClient, ExtraInfoSourceRef, MetaClient};
-use risingwave_source::dml_manager::DmlManager;
 use risingwave_storage::hummock::compactor::{
     start_compactor, CompactionExecutor, CompactorContext,
 };