refactor(query): merge pipeline core, sources and sinks crate (#18939)

* refactor(query): merge pipeline core, sources and sinks crate

* refactor(query): merge pipeline core, sources and sinks crate

Author: zhang2014

Cargo.lock

@@ -45,9 +45,7 @@ members = [
     "src/query/functions/src/scalars/arithmetic",
     "src/query/functions/src/scalars/integer_arithmetic",
     "src/query/management",
-    "src/query/pipeline/core",
-    "src/query/pipeline/sinks",
-    "src/query/pipeline/sources",
+    "src/query/pipeline",
     "src/query/pipeline/transforms",
     "src/query/script",
     "src/query/settings",
@@ -155,9 +153,7 @@ databend-common-meta-store = { path = "src/meta/store" }
 databend-common-meta-types = { path = "src/meta/types" }
 databend-common-metrics = { path = "src/common/metrics" }
 databend-common-native = { path = "src/common/native" }
-databend-common-pipeline-core = { path = "src/query/pipeline/core" }
-databend-common-pipeline-sinks = { path = "src/query/pipeline/sinks" }
-databend-common-pipeline-sources = { path = "src/query/pipeline/sources" }
+databend-common-pipeline = { path = "src/query/pipeline" }
 databend-common-pipeline-transforms = { path = "src/query/pipeline/transforms" }
 databend-common-proto-conv = { path = "src/meta/proto-conv" }
 databend-common-protos = { path = "src/meta/protos" }

Cargo.toml

@@ -23,7 +23,7 @@ databend-common-meta-api = { workspace = true }
 databend-common-meta-app = { workspace = true }
 databend-common-meta-store = { workspace = true }
 databend-common-meta-types = { workspace = true }
-databend-common-pipeline-core = { workspace = true }
+databend-common-pipeline = { workspace = true }
 databend-common-settings = { workspace = true }
 databend-common-storage = { workspace = true }
 databend-common-users = { workspace = true }

src/query/catalog/Cargo.toml

@@ -16,7 +16,7 @@ use std::sync::Arc;
 
 use databend_common_exception::Result;
 use databend_common_meta_app::schema::LockType;
-use databend_common_pipeline_core::LockGuard;
+use databend_common_pipeline::core::LockGuard;
 
 use crate::table_context::TableContext;
 

src/query/catalog/src/lock.rs

@@ -36,7 +36,7 @@ use databend_common_meta_app::schema::TableMeta;
 use databend_common_meta_app::schema::UpdateStreamMetaReq;
 use databend_common_meta_app::schema::UpsertTableCopiedFileReq;
 use databend_common_meta_types::MetaId;
-use databend_common_pipeline_core::Pipeline;
+use databend_common_pipeline::core::Pipeline;
 use databend_common_storage::Histogram;
 use databend_common_storage::StorageMetrics;
 use databend_storages_common_table_meta::meta::ClusterKey;

src/query/catalog/src/table.rs

@@ -46,9 +46,9 @@ use databend_common_meta_app::principal::UserInfo;
 use databend_common_meta_app::principal::UserPrivilegeType;
 use databend_common_meta_app::storage::StorageParams;
 use databend_common_meta_app::tenant::Tenant;
-use databend_common_pipeline_core::processors::PlanProfile;
-use databend_common_pipeline_core::InputError;
-use databend_common_pipeline_core::LockGuard;
+use databend_common_pipeline::core::InputError;
+use databend_common_pipeline::core::LockGuard;
+use databend_common_pipeline::core::PlanProfile;
 use databend_common_settings::Settings;
 use databend_common_storage::CopyStatus;
 use databend_common_storage::DataOperator;

src/query/catalog/src/table_context.rs

@@ -26,8 +26,7 @@ databend-common-meta-app = { workspace = true }
 databend-common-meta-store = { workspace = true }
 databend-common-meta-types = { workspace = true }
 databend-common-metrics = { workspace = true }
-databend-common-pipeline-core = { workspace = true }
-databend-common-pipeline-sources = { workspace = true }
+databend-common-pipeline = { workspace = true }
 databend-common-pipeline-transforms = { workspace = true }
 databend-common-sql = { workspace = true }
 databend-common-storage = { workspace = true }

src/query/ee/Cargo.toml

@@ -33,9 +33,9 @@ use databend_common_expression::TableSchema;
 use databend_common_metrics::storage::metrics_inc_block_virtual_column_write_bytes;
 use databend_common_metrics::storage::metrics_inc_block_virtual_column_write_milliseconds;
 use databend_common_metrics::storage::metrics_inc_block_virtual_column_write_nums;
-use databend_common_pipeline_core::Pipeline;
-use databend_common_pipeline_sources::AsyncSource;
-use databend_common_pipeline_sources::AsyncSourcer;
+use databend_common_pipeline::core::Pipeline;
+use databend_common_pipeline::sources::AsyncSource;
+use databend_common_pipeline::sources::AsyncSourcer;
 use databend_common_pipeline_transforms::processors::AsyncTransform;
 use databend_common_pipeline_transforms::processors::TransformPipelineHelper;
 use databend_common_sql::executor::physical_plans::MutationKind;
@@ -173,7 +173,7 @@ pub async fn do_refresh_virtual_column(
                 block_reader.clone(),
                 virtual_column_metas.clone(),
             );
-            AsyncSourcer::create(ctx.clone(), output, inner)
+            AsyncSourcer::create(ctx.get_scan_progress(), output, inner)
         },
         1,
     )?;

src/query/ee/src/storages/fuse/operations/virtual_columns.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 use databend_common_base::base::GlobalInstance;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
-use databend_common_pipeline_core::Pipeline;
+use databend_common_pipeline::core::Pipeline;
 use databend_common_storages_fuse::FuseTable;
 use databend_enterprise_virtual_column::VirtualColumnHandler;
 use databend_enterprise_virtual_column::VirtualColumnHandlerWrapper;

src/query/ee/src/virtual_column/virtual_column_handler.rs

@@ -14,7 +14,7 @@ async-trait = { workspace = true }
 databend-common-base = { workspace = true }
 databend-common-catalog = { workspace = true }
 databend-common-exception = { workspace = true }
-databend-common-pipeline-core = { workspace = true }
+databend-common-pipeline = { workspace = true }
 databend-common-storages-fuse = { workspace = true }
 
 [build-dependencies]