refactor(deps): Bump OpenDAL and other crates (#16251)

Xuanwo · web-flow · commit eaae4e0e5242 · 2024-08-15T20:39:36.000+08:00
* deps: Bump OpenDAL and other crates

Signed-off-by: Xuanwo &lt;github@xuanwo.io&gt;

* fix test for ahash upgrade

Signed-off-by: Xuanwo &lt;github@xuanwo.io&gt;

* Fix test

Signed-off-by: Xuanwo &lt;github@xuanwo.io&gt;

---------

Signed-off-by: Xuanwo &lt;github@xuanwo.io&gt;
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -221,7 +221,7 @@ arrow-udf-js = "0.3.1"
 arrow-udf-python = "0.2.1"
 arrow-udf-wasm = "0.2.2"
 async-backtrace = "0.2"
-async-trait = { version = "0.1.77", package = "async-trait-fn" }
+async-trait = { version = "0.1.77" }
 backtrace = "0.3.73"
 base64 = "0.22"
 bincode = { version = "2.0.0-rc.3", features = ["serde", "std", "alloc"] }
@@ -256,10 +256,10 @@ jsonb = "0.4.1"
 jwt-simple = "0.11.0"
 match-template = "0.0.1"
 mysql_async = { version = "0.34", default-features = false, features = ["native-tls-tls"] }
-object_store_opendal = "0.45"
+object_store_opendal = "0.46"
 once_cell = "1.15.0"
 openai_api_rust = "0.1"
-opendal = { version = "0.48.0", features = [
+opendal = { version = "0.49.0", features = [
     "layers-fastrace",
     "layers-prometheus-client",
     "layers-async-backtrace",
diff --git a/src/common/storage/src/operator.rs b/src/common/storage/src/operator.rs
@@ -135,7 +135,7 @@ pub fn build_operator<B: Builder>(builder: B) -> Result<Operator> {
         // Add async backtrace
         .layer(AsyncBacktraceLayer)
         // Add logging
-        .layer(LoggingLayer::default().with_backtrace_output(true))
+        .layer(LoggingLayer::default())
         // Add tracing
         .layer(FastraceLayer)
         // Add PrometheusClientLayer
diff --git a/src/query/pipeline/sinks/src/async_mpsc_sink.rs b/src/query/pipeline/sinks/src/async_mpsc_sink.rs
@@ -16,7 +16,6 @@ use std::any::Any;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use async_trait::unboxed_simple;
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
 use databend_common_pipeline_core::processors::Event;
@@ -38,7 +37,6 @@ pub trait AsyncMpscSink: Send {
         Ok(())
     }
 
-    #[unboxed_simple]
     async fn consume(&mut self, data_block: DataBlock) -> Result<bool>;
 }
 
diff --git a/src/query/pipeline/sinks/src/async_sink.rs b/src/query/pipeline/sinks/src/async_sink.rs
@@ -16,7 +16,6 @@ use std::any::Any;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use async_trait::unboxed_simple;
 use databend_common_base::runtime::drop_guard;
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_base::runtime::TrySpawn;
@@ -40,7 +39,6 @@ pub trait AsyncSink: Send {
         Ok(())
     }
 
-    #[unboxed_simple]
     async fn consume(&mut self, data_block: DataBlock) -> Result<bool>;
 
     fn details_status(&self) -> Option<String> {
diff --git a/src/query/pipeline/sinks/src/union_receive_sink.rs b/src/query/pipeline/sinks/src/union_receive_sink.rs
@@ -16,7 +16,6 @@ use std::sync::Arc;
 
 use async_channel::Sender;
 use async_trait::async_trait;
-use async_trait::unboxed_simple;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
@@ -46,7 +45,6 @@ impl AsyncSink for UnionReceiveSink {
         Ok(())
     }
 
-    #[unboxed_simple]
     #[async_backtrace::framed]
     async fn consume(&mut self, data_block: DataBlock) -> Result<bool> {
         if let Some(sender) = self.sender.as_ref() {
diff --git a/src/query/pipeline/sinks/tests/it/async_mpsc_sink.rs b/src/query/pipeline/sinks/tests/it/async_mpsc_sink.rs
@@ -17,7 +17,6 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use async_trait::unboxed_simple;
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
 use databend_common_pipeline_core::processors::connect;
@@ -51,7 +50,6 @@ impl AsyncMpscSink for TestSink {
         Ok(())
     }
 
-    #[unboxed_simple]
     async fn consume(&mut self, data_block: DataBlock) -> Result<bool> {
         self.count
             .fetch_add(data_block.num_rows(), Ordering::SeqCst);
diff --git a/src/query/pipeline/sources/src/async_source.rs b/src/query/pipeline/sources/src/async_source.rs
@@ -33,7 +33,6 @@ pub trait AsyncSource: Send {
     const NAME: &'static str;
     const SKIP_EMPTY_DATA_BLOCK: bool = true;
 
-    #[async_trait::unboxed_simple]
     async fn generate(&mut self) -> Result<Option<DataBlock>>;
 
     fn un_reacted(&self) -> Result<()> {
diff --git a/src/query/pipeline/sources/src/prefetch_async_source.rs b/src/query/pipeline/sources/src/prefetch_async_source.rs
@@ -33,7 +33,6 @@ pub trait PrefetchAsyncSource: Send {
     const NAME: &'static str;
     const SKIP_EMPTY_DATA_BLOCK: bool = true;
 
-    #[async_trait::unboxed_simple]
     async fn generate(&mut self) -> Result<Option<DataBlock>>;
     fn is_full(&self, prefetched: &[DataBlock]) -> bool;
 
diff --git a/src/query/pipeline/sources/src/stream_source.rs b/src/query/pipeline/sources/src/stream_source.rs
@@ -57,7 +57,6 @@ impl<const T: bool> AsyncSource for AsyncStreamSource<T> {
     const NAME: &'static str = "stream source";
     const SKIP_EMPTY_DATA_BLOCK: bool = T;
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         match self
diff --git a/src/query/service/src/pipelines/builders/builder_replace_into.rs b/src/query/service/src/pipelines/builders/builder_replace_into.rs
@@ -408,7 +408,6 @@ impl AsyncSource for ValueSource {
     const NAME: &'static str = "ValueSource";
     const SKIP_EMPTY_DATA_BLOCK: bool = true;
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.is_finished {
@@ -478,7 +477,6 @@ impl AsyncSource for RawValueSource {
     const NAME: &'static str = "RawValueSource";
     const SKIP_EMPTY_DATA_BLOCK: bool = true;
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.is_finished {
diff --git a/src/query/service/src/pipelines/processors/transforms/transform_cache_scan.rs b/src/query/service/src/pipelines/processors/transforms/transform_cache_scan.rs
@@ -124,7 +124,6 @@ impl TransformCacheScan {
 impl AsyncSource for TransformCacheScan {
     const NAME: &'static str = "TransformCacheScan";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         let data_block = self.cache_source_state.next_data_block();
diff --git a/src/query/service/src/pipelines/processors/transforms/transform_materialized_cte.rs b/src/query/service/src/pipelines/processors/transforms/transform_materialized_cte.rs
@@ -161,7 +161,6 @@ impl MaterializedCteSource {
 impl AsyncSource for MaterializedCteSource {
     const NAME: &'static str = "MaterializedCteSource";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         self.cte_state.wait_sink_finished().await?;
diff --git a/src/query/service/src/pipelines/processors/transforms/transform_recursive_cte_scan.rs b/src/query/service/src/pipelines/processors/transforms/transform_recursive_cte_scan.rs
@@ -50,7 +50,6 @@ impl TransformRecursiveCteScan {
 impl AsyncSource for TransformRecursiveCteScan {
     const NAME: &'static str = "RecursiveCteScan";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.table.is_none() {
diff --git a/src/query/service/src/servers/flight/v1/exchange/exchange_sink_writer.rs b/src/query/service/src/servers/flight/v1/exchange/exchange_sink_writer.rs
@@ -66,7 +66,6 @@ impl AsyncSink for ExchangeWriterSink {
         Ok(())
     }
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn consume(&mut self, mut data_block: DataBlock) -> Result<bool> {
         let serialize_meta = match data_block.take_meta() {
diff --git a/src/query/service/src/table_functions/async_crash_me.rs b/src/query/service/src/table_functions/async_crash_me.rs
@@ -155,7 +155,6 @@ impl AsyncCrashMeSource {
 impl AsyncSource for AsyncCrashMeSource {
     const NAME: &'static str = "async_crash_me";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         match &self.message {
diff --git a/src/query/service/src/table_functions/cloud/task_dependents.rs b/src/query/service/src/table_functions/cloud/task_dependents.rs
@@ -240,7 +240,6 @@ impl TaskDependentsSource {
 impl AsyncSource for TaskDependentsSource {
     const NAME: &'static str = "task_dependents";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> databend_common_exception::Result<Option<DataBlock>> {
         if self.is_finished {
diff --git a/src/query/service/src/table_functions/cloud/task_dependents_enable.rs b/src/query/service/src/table_functions/cloud/task_dependents_enable.rs
@@ -158,7 +158,6 @@ impl TaskDependentsEnableSource {
 impl AsyncSource for TaskDependentsEnableSource {
     const NAME: &'static str = "task_dependents_enable";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         let config = GlobalConfig::instance();
diff --git a/src/query/service/src/table_functions/cloud/task_history.rs b/src/query/service/src/table_functions/cloud/task_history.rs
@@ -184,7 +184,6 @@ impl TaskHistorySource {
 impl AsyncSource for TaskHistorySource {
     const NAME: &'static str = "task_history";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.is_finished {
diff --git a/src/query/service/src/table_functions/infer_schema/parquet.rs b/src/query/service/src/table_functions/infer_schema/parquet.rs
@@ -65,7 +65,6 @@ impl ParquetInferSchemaSource {
 impl AsyncSource for ParquetInferSchemaSource {
     const NAME: &'static str = INFER_SCHEMA;
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.is_finished {
diff --git a/src/query/service/src/table_functions/inspect_parquet/inspect_parquet_table.rs b/src/query/service/src/table_functions/inspect_parquet/inspect_parquet_table.rs
@@ -200,7 +200,6 @@ impl InspectParquetSource {
 impl AsyncSource for InspectParquetSource {
     const NAME: &'static str = INSPECT_PARQUET;
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.is_finished {
diff --git a/src/query/service/src/table_functions/list_stage/list_stage_table.rs b/src/query/service/src/table_functions/list_stage/list_stage_table.rs
@@ -247,7 +247,6 @@ fn make_block(files: &[StageFileInfo]) -> DataBlock {
 impl AsyncSource for ListStagesSource {
     const NAME: &'static str = LIST_STAGE;
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         match &self.state {
diff --git a/src/query/service/src/table_functions/openai/ai_to_sql.rs b/src/query/service/src/table_functions/openai/ai_to_sql.rs
@@ -168,7 +168,6 @@ impl GPT2SQLSource {
 impl AsyncSource for GPT2SQLSource {
     const NAME: &'static str = "gpt_to_sql";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.finished {
diff --git a/src/query/service/src/table_functions/others/execute_background_job.rs b/src/query/service/src/table_functions/others/execute_background_job.rs
@@ -145,7 +145,6 @@ impl ExecuteBackgroundJobSource {
 impl AsyncSource for ExecuteBackgroundJobSource {
     const NAME: &'static str = "execute_job";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         let background_handler = get_background_service_handler();
diff --git a/src/query/service/src/table_functions/others/license_info.rs b/src/query/service/src/table_functions/others/license_info.rs
@@ -205,7 +205,6 @@ impl LicenseInfoSource {
 impl AsyncSource for LicenseInfoSource {
     const NAME: &'static str = "license_info";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.done {
diff --git a/src/query/service/src/table_functions/others/suggested_background_tasks.rs b/src/query/service/src/table_functions/others/suggested_background_tasks.rs
@@ -241,7 +241,6 @@ impl SuggestedBackgroundTasksSource {
 impl AsyncSource for SuggestedBackgroundTasksSource {
     const NAME: &'static str = "suggested_background_tasks";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.done {
diff --git a/src/query/service/src/table_functions/others/tenant_quota.rs b/src/query/service/src/table_functions/others/tenant_quota.rs
@@ -222,7 +222,6 @@ impl TenantQuotaSource {
 impl AsyncSource for TenantQuotaSource {
     const NAME: &'static str = "tenant_quota";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.done {
diff --git a/src/query/service/src/table_functions/show_grants/show_grants_table.rs b/src/query/service/src/table_functions/show_grants/show_grants_table.rs
@@ -226,7 +226,6 @@ impl ShowGrantsSource {
 impl AsyncSource for ShowGrantsSource {
     const NAME: &'static str = "show_grants";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.finished {
diff --git a/src/query/storages/fuse/src/operations/agg_index_sink.rs b/src/query/storages/fuse/src/operations/agg_index_sink.rs
@@ -17,7 +17,6 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use async_trait::async_trait;
-use async_trait::unboxed_simple;
 use databend_common_exception::Result;
 use databend_common_expression::types::StringType;
 use databend_common_expression::BlockRowIndex;
@@ -122,7 +121,6 @@ impl AsyncSink for AggIndexSink {
         Ok(())
     }
 
-    #[unboxed_simple]
     #[async_backtrace::framed]
     async fn consume(&mut self, data_block: DataBlock) -> Result<bool> {
         let mut block = data_block;
diff --git a/src/query/storages/fuse/src/operations/analyze.rs b/src/query/storages/fuse/src/operations/analyze.rs
@@ -20,7 +20,6 @@ use std::time::Instant;
 use async_channel::Receiver;
 use async_channel::Sender;
 use async_trait::async_trait;
-use async_trait::unboxed_simple;
 use databend_common_catalog::catalog::CatalogManager;
 use databend_common_catalog::table::Table;
 use databend_common_catalog::table_context::TableContext;
@@ -146,7 +145,6 @@ impl SinkAnalyzeState {
         Ok(table)
     }
 
-    #[unboxed_simple]
     #[async_backtrace::framed]
     pub async fn merge_analyze_states(&mut self, data_block: DataBlock) -> Result<()> {
         if data_block.num_rows() == 0 {
@@ -202,7 +200,6 @@ impl SinkAnalyzeState {
         Ok(())
     }
 
-    #[unboxed_simple]
     #[async_backtrace::framed]
     async fn create_histogram(&mut self, col_id: u32, data_block: DataBlock) -> Result<()> {
         if data_block.num_rows() == 0 {
@@ -440,7 +437,6 @@ impl AsyncSink for HistogramInfoSink {
         Ok(())
     }
 
-    #[unboxed_simple]
     #[async_backtrace::framed]
     async fn consume(&mut self, data_block: DataBlock) -> Result<bool> {
         if let Some(sender) = self.sender.as_ref() {
diff --git a/src/query/storages/fuse/src/operations/common/processors/multi_table_insert_commit.rs b/src/query/storages/fuse/src/operations/common/processors/multi_table_insert_commit.rs
@@ -17,7 +17,6 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use async_trait::async_trait;
-use async_trait::unboxed_simple;
 use backoff::backoff::Backoff;
 use databend_common_catalog::catalog::Catalog;
 use databend_common_catalog::table::Table;
@@ -203,7 +202,6 @@ impl AsyncSink for CommitMultiTableInsert {
         }
     }
 
-    #[unboxed_simple]
     #[async_backtrace::framed]
     async fn consume(&mut self, data_block: DataBlock) -> Result<bool> {
         let input_meta = data_block
diff --git a/src/query/storages/fuse/src/operations/inverted_index.rs b/src/query/storages/fuse/src/operations/inverted_index.rs
@@ -20,7 +20,6 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use async_trait::async_trait;
-use async_trait::unboxed_simple;
 use databend_common_catalog::plan::Projection;
 use databend_common_catalog::table::Table;
 use databend_common_catalog::table_context::TableContext;
@@ -218,7 +217,6 @@ impl InvertedIndexSource {
 impl AsyncSource for InvertedIndexSource {
     const NAME: &'static str = "InvertedIndexSource";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.is_finished {
@@ -328,7 +326,6 @@ impl InvertedIndexSink {
 impl AsyncSink for InvertedIndexSink {
     const NAME: &'static str = "InvertedIndexSink";
 
-    #[unboxed_simple]
     #[async_backtrace::framed]
     async fn consume(&mut self, _data_block: DataBlock) -> Result<bool> {
         let num = self.block_nums.fetch_sub(1, Ordering::SeqCst);
diff --git a/src/query/storages/fuse/src/table_functions/function_template/simple_func_template.rs b/src/query/storages/fuse/src/table_functions/function_template/simple_func_template.rs
@@ -224,7 +224,6 @@ where T: SimpleTableFunc
 {
     const NAME: &'static str = "template_func";
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         if self.finish {
diff --git a/src/query/storages/iceberg/Cargo.toml b/src/query/storages/iceberg/Cargo.toml
@@ -26,9 +26,9 @@ databend-common-storages-parquet = { workspace = true }
 databend-storages-common-pruner = { workspace = true }
 databend-storages-common-table-meta = { workspace = true }
 fastrace = { workspace = true }
-iceberg = { git = "https://github.com/apache/iceberg-rust", package = "iceberg", rev = "e8f8694" }
-iceberg-catalog-hms = { git = "https://github.com/apache/iceberg-rust", package = "iceberg-catalog-hms", rev = "e8f8694" }
-iceberg-catalog-rest = { git = "https://github.com/apache/iceberg-rust", package = "iceberg-catalog-rest", rev = "e8f8694" }
+iceberg = { git = "https://github.com/apache/iceberg-rust", package = "iceberg", rev = "v0.3.0-rc.1" }
+iceberg-catalog-hms = { git = "https://github.com/apache/iceberg-rust", package = "iceberg-catalog-hms", rev = "v0.3.0-rc.1" }
+iceberg-catalog-rest = { git = "https://github.com/apache/iceberg-rust", package = "iceberg-catalog-rest", rev = "v0.3.0-rc.1" }
 match-template = { workspace = true }
 parquet = { workspace = true }
 serde = { workspace = true }
diff --git a/src/query/storages/iceberg/src/stats.rs b/src/query/storages/iceberg/src/stats.rs
@@ -76,8 +76,6 @@ fn parse_datum(data: &Datum) -> Option<Scalar> {
         PrimitiveLiteral::Long(v) => Some(Scalar::Number(i64::upcast_scalar(*v))),
         PrimitiveLiteral::Float(v) => Some(Scalar::Number(F32::upcast_scalar(F32::from(*v)))),
         PrimitiveLiteral::Double(v) => Some(Scalar::Number(F64::upcast_scalar(F64::from(*v)))),
-        PrimitiveLiteral::Date(v) => Some(Scalar::Date(*v)),
-        PrimitiveLiteral::Timestamp(v) => Some(Scalar::Timestamp(*v)),
         PrimitiveLiteral::String(v) => Some(Scalar::String(v.clone())),
         _ => None,
     }
diff --git a/src/query/storages/orc/src/copy_into_table/processors/source.rs b/src/query/storages/orc/src/copy_into_table/processors/source.rs
@@ -98,7 +98,6 @@ impl AsyncSource for ORCSourceForCopy {
     const NAME: &'static str = "ORCSourceForCopy";
     const SKIP_EMPTY_DATA_BLOCK: bool = false;
 
-    #[async_trait::unboxed_simple]
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
         loop {
diff --git a/src/query/storages/orc/src/processors/source.rs b/src/query/storages/orc/src/processors/source.rs
diff --git a/src/query/storages/result_cache/src/write/sink.rs b/src/query/storages/result_cache/src/write/sink.rs
diff --git a/src/query/storages/stage/src/read/row_based/processors/reader.rs b/src/query/storages/stage/src/read/row_based/processors/reader.rs
diff --git a/src/query/storages/stream/src/stream_status_table_func.rs b/src/query/storages/stream/src/stream_status_table_func.rs
diff --git a/src/query/storages/system/src/table.rs b/src/query/storages/system/src/table.rs
diff --git a/tests/sqllogictests/suites/base/03_common/03_0008_select_aggregator_combinator.test b/tests/sqllogictests/suites/base/03_common/03_0008_select_aggregator_combinator.test
diff --git a/tests/sqllogictests/suites/base/20+_others/20_0001_planner.test b/tests/sqllogictests/suites/base/20+_others/20_0001_planner.test
diff --git a/tests/sqllogictests/suites/mode/standalone/explain/explain.test b/tests/sqllogictests/suites/mode/standalone/explain/explain.test

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,6 @@ use std::any::Any;`
`16`	`16`	`use std::sync::Arc;`
`17`	`17`
`18`	`18`	`use async_trait::async_trait;`
`19`		`-use async_trait::unboxed_simple;`
`20`	`19`	`use databend_common_exception::Result;`
`21`	`20`	`use databend_common_expression::DataBlock;`
`22`	`21`	`use databend_common_pipeline_core::processors::Event;`
`@@ -38,7 +37,6 @@ pub trait AsyncMpscSink: Send {`
`38`	`37`	`Ok(())`
`39`	`38`	`}`
`40`	`39`
`41`		`- #[unboxed_simple]`
`42`	`40`	`async fn consume(&mut self, data_block: DataBlock) -> Result<bool>;`
`43`	`41`	`}`
`44`	`42`