diff --git a/be/src/exec/schema_scanner.cpp b/be/src/exec/schema_scanner.cpp
index de9857bad2ce83..b78083f39e2efb 100644
--- a/be/src/exec/schema_scanner.cpp
+++ b/be/src/exec/schema_scanner.cpp
@@ -51,7 +51,10 @@
 #include "exec/schema_scanner/schema_workload_groups_scanner.h"
 #include "exec/schema_scanner/schema_workload_sched_policy_scanner.h"
 #include "olap/hll.h"
+#include "pipeline/dependency.h"
 #include "runtime/define_primitive_type.h"
+#include "runtime/fragment_mgr.h"
+#include "runtime/types.h"
 #include "util/string_util.h"
 #include "util/types.h"
 #include "vec/columns/column.h"
@@ -65,6 +68,7 @@
 #include "vec/core/column_with_type_and_name.h"
 #include "vec/core/types.h"
 #include "vec/data_types/data_type.h"
+#include "vec/data_types/data_type_factory.hpp"
 
 namespace doris {
 class ObjectPool;
@@ -85,7 +89,60 @@ Status SchemaScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaScanner::get_next_block(RuntimeState* state, vectorized::Block* block, bool* eos) {
+    if (_data_block == nullptr) {
+        return Status::InternalError("No data left!");
+    }
+    DCHECK(_async_thread_running == false);
+    RETURN_IF_ERROR(_scanner_status.status());
+    for (size_t i = 0; i < block->columns(); i++) {
+        std::move(*block->get_by_position(i).column)
+                .mutate()
+                ->insert_range_from(*_data_block->get_by_position(i).column, 0,
+                                    _data_block->rows());
+    }
+    _data_block->clear_column_data();
+    *eos = _eos;
+    if (!*eos) {
+        RETURN_IF_ERROR(get_next_block_async(state));
+    }
+    return Status::OK();
+}
+
+Status SchemaScanner::get_next_block_async(RuntimeState* state) {
+    _dependency->block();
+    auto task_ctx = state->get_task_execution_context();
+    RETURN_IF_ERROR(ExecEnv::GetInstance()->fragment_mgr()->get_thread_pool()->submit_func(
+            [this, task_ctx, state]() {
+                DCHECK(_async_thread_running == false);
+                auto task_lock = task_ctx.lock();
+                if (task_lock == nullptr) {
+                    _scanner_status.update(Status::InternalError("Task context not exists!"));
+                    return;
+                }
+                SCOPED_ATTACH_TASK(state);
+                _dependency->block();
+                _async_thread_running = true;
+                _finish_dependency->block();
+                if (!_opened) {
+                    _data_block = vectorized::Block::create_unique();
+                    _init_block(_data_block.get());
+                    _scanner_status.update(start(state));
+                    _opened = true;
+                }
+                bool eos = false;
+                _scanner_status.update(get_next_block_internal(_data_block.get(), &eos));
+                _eos = eos;
+                _async_thread_running = false;
+                _dependency->set_ready();
+                if (eos) {
+                    _finish_dependency->set_ready();
+                }
+            }));
+    return Status::OK();
+}
+
+Status SchemaScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("used before initialized.");
     }
@@ -176,6 +233,16 @@ std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type
     }
 }
 
+void SchemaScanner::_init_block(vectorized::Block* src_block) {
+    const std::vector<SchemaScanner::ColumnDesc>& columns_desc(get_column_desc());
+    for (int i = 0; i < columns_desc.size(); ++i) {
+        TypeDescriptor descriptor(columns_desc[i].type);
+        auto data_type = vectorized::DataTypeFactory::instance().create_data_type(descriptor, true);
+        src_block->insert(vectorized::ColumnWithTypeAndName(data_type->create_column(), data_type,
+                                                            columns_desc[i].name));
+    }
+}
+
 Status SchemaScanner::fill_dest_column_for_range(vectorized::Block* block, size_t pos,
                                                  const std::vector<void*>& datas) {
     const ColumnDesc& col_desc = _columns[pos];
diff --git a/be/src/exec/schema_scanner.h b/be/src/exec/schema_scanner.h
index a23706ac6a440a..4666657af21b5f 100644
--- a/be/src/exec/schema_scanner.h
+++ b/be/src/exec/schema_scanner.h
@@ -22,6 +22,7 @@
 #include <stddef.h>
 #include <stdint.h>
 
+#include <condition_variable>
 #include <memory>
 #include <string>
 #include <vector>
@@ -43,6 +44,10 @@ namespace vectorized {
 class Block;
 }
 
+namespace pipeline {
+class Dependency;
+}
+
 struct SchemaScannerCommonParam {
     SchemaScannerCommonParam()
             : db(nullptr),
@@ -94,15 +99,23 @@ class SchemaScanner {
 
     // init object need information, schema etc.
     virtual Status init(SchemaScannerParam* param, ObjectPool* pool);
+    Status get_next_block(RuntimeState* state, vectorized::Block* block, bool* eos);
     // Start to work
     virtual Status start(RuntimeState* state);
-    virtual Status get_next_block(vectorized::Block* block, bool* eos);
+    virtual Status get_next_block_internal(vectorized::Block* block, bool* eos);
     const std::vector<ColumnDesc>& get_column_desc() const { return _columns; }
     // factory function
     static std::unique_ptr<SchemaScanner> create(TSchemaTableType::type type);
     TSchemaTableType::type type() const { return _schema_table_type; }
+    void set_dependency(std::shared_ptr<pipeline::Dependency> dep,
+                        std::shared_ptr<pipeline::Dependency> fin_dep) {
+        _dependency = dep;
+        _finish_dependency = fin_dep;
+    }
+    Status get_next_block_async(RuntimeState* state);
 
 protected:
+    void _init_block(vectorized::Block* src_block);
     Status fill_dest_column_for_range(vectorized::Block* block, size_t pos,
                                       const std::vector<void*>& datas);
 
@@ -125,6 +138,15 @@ class SchemaScanner {
     RuntimeProfile::Counter* _get_table_timer = nullptr;
     RuntimeProfile::Counter* _get_describe_timer = nullptr;
     RuntimeProfile::Counter* _fill_block_timer = nullptr;
+
+    std::shared_ptr<pipeline::Dependency> _dependency = nullptr;
+    std::shared_ptr<pipeline::Dependency> _finish_dependency = nullptr;
+
+    std::unique_ptr<vectorized::Block> _data_block;
+    AtomicStatus _scanner_status;
+    std::atomic<bool> _eos = false;
+    std::atomic<bool> _opened = false;
+    std::atomic<bool> _async_thread_running = false;
 };
 
 } // namespace doris
diff --git a/be/src/exec/schema_scanner/schema_active_queries_scanner.cpp b/be/src/exec/schema_scanner/schema_active_queries_scanner.cpp
index 2115a38a6ebce3..46522a36242fc1 100644
--- a/be/src/exec/schema_scanner/schema_active_queries_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_active_queries_scanner.cpp
@@ -137,7 +137,7 @@ Status SchemaActiveQueriesScanner::_get_active_queries_block_from_fe() {
     return Status::OK();
 }
 
-Status SchemaActiveQueriesScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaActiveQueriesScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_active_queries_scanner.h b/be/src/exec/schema_scanner/schema_active_queries_scanner.h
index 1df5b1f9d7402d..7e9ae4b8034083 100644
--- a/be/src/exec/schema_scanner/schema_active_queries_scanner.h
+++ b/be/src/exec/schema_scanner/schema_active_queries_scanner.h
@@ -36,7 +36,7 @@ class SchemaActiveQueriesScanner : public SchemaScanner {
     ~SchemaActiveQueriesScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
 
diff --git a/be/src/exec/schema_scanner/schema_backend_active_tasks.cpp b/be/src/exec/schema_scanner/schema_backend_active_tasks.cpp
index f1155796ed434d..b35e84a9f9c9f4 100644
--- a/be/src/exec/schema_scanner/schema_backend_active_tasks.cpp
+++ b/be/src/exec/schema_scanner/schema_backend_active_tasks.cpp
@@ -51,7 +51,8 @@ Status SchemaBackendActiveTasksScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaBackendActiveTasksScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaBackendActiveTasksScanner::get_next_block_internal(vectorized::Block* block,
+                                                                bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_backend_active_tasks.h b/be/src/exec/schema_scanner/schema_backend_active_tasks.h
index d8a2a1ffa3f96a..43819818b57f69 100644
--- a/be/src/exec/schema_scanner/schema_backend_active_tasks.h
+++ b/be/src/exec/schema_scanner/schema_backend_active_tasks.h
@@ -36,7 +36,7 @@ class SchemaBackendActiveTasksScanner : public SchemaScanner {
     ~SchemaBackendActiveTasksScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
 
diff --git a/be/src/exec/schema_scanner/schema_charsets_scanner.cpp b/be/src/exec/schema_scanner/schema_charsets_scanner.cpp
index 534f045341b7e3..d06cd8fa745634 100644
--- a/be/src/exec/schema_scanner/schema_charsets_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_charsets_scanner.cpp
@@ -48,7 +48,7 @@ SchemaCharsetsScanner::SchemaCharsetsScanner()
 
 SchemaCharsetsScanner::~SchemaCharsetsScanner() {}
 
-Status SchemaCharsetsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaCharsetsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("call this before initial.");
     }
diff --git a/be/src/exec/schema_scanner/schema_charsets_scanner.h b/be/src/exec/schema_scanner/schema_charsets_scanner.h
index 1f01070875ccf6..d5089c62826b0b 100644
--- a/be/src/exec/schema_scanner/schema_charsets_scanner.h
+++ b/be/src/exec/schema_scanner/schema_charsets_scanner.h
@@ -36,7 +36,7 @@ class SchemaCharsetsScanner : public SchemaScanner {
     SchemaCharsetsScanner();
     ~SchemaCharsetsScanner() override;
 
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     struct CharsetStruct {
diff --git a/be/src/exec/schema_scanner/schema_collations_scanner.cpp b/be/src/exec/schema_scanner/schema_collations_scanner.cpp
index 9d50b5216303d8..8592eb7575c387 100644
--- a/be/src/exec/schema_scanner/schema_collations_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_collations_scanner.cpp
@@ -50,7 +50,7 @@ SchemaCollationsScanner::SchemaCollationsScanner()
 
 SchemaCollationsScanner::~SchemaCollationsScanner() {}
 
-Status SchemaCollationsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaCollationsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("call this before initial.");
     }
diff --git a/be/src/exec/schema_scanner/schema_collations_scanner.h b/be/src/exec/schema_scanner/schema_collations_scanner.h
index f0f60538cacce0..2fe200da78d04d 100644
--- a/be/src/exec/schema_scanner/schema_collations_scanner.h
+++ b/be/src/exec/schema_scanner/schema_collations_scanner.h
@@ -36,7 +36,7 @@ class SchemaCollationsScanner : public SchemaScanner {
     SchemaCollationsScanner();
     ~SchemaCollationsScanner() override;
 
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     struct CollationStruct {
diff --git a/be/src/exec/schema_scanner/schema_columns_scanner.cpp b/be/src/exec/schema_scanner/schema_columns_scanner.cpp
index deda8af7d8de58..f4e15d2aef0af2 100644
--- a/be/src/exec/schema_scanner/schema_columns_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_columns_scanner.cpp
@@ -347,7 +347,7 @@ Status SchemaColumnsScanner::_get_new_table() {
     return Status::OK();
 }
 
-Status SchemaColumnsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaColumnsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("use this class before inited.");
     }
diff --git a/be/src/exec/schema_scanner/schema_columns_scanner.h b/be/src/exec/schema_scanner/schema_columns_scanner.h
index 2499db7ed82a2b..99150c36d109a2 100644
--- a/be/src/exec/schema_scanner/schema_columns_scanner.h
+++ b/be/src/exec/schema_scanner/schema_columns_scanner.h
@@ -38,7 +38,7 @@ class SchemaColumnsScanner : public SchemaScanner {
     SchemaColumnsScanner();
     ~SchemaColumnsScanner() override;
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _get_new_table();
diff --git a/be/src/exec/schema_scanner/schema_dummy_scanner.cpp b/be/src/exec/schema_scanner/schema_dummy_scanner.cpp
index 1d5956f390ea26..9e3a703d9fb5d6 100644
--- a/be/src/exec/schema_scanner/schema_dummy_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_dummy_scanner.cpp
@@ -40,7 +40,7 @@ Status SchemaDummyScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaDummyScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaDummyScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     *eos = true;
     return Status::OK();
 }
diff --git a/be/src/exec/schema_scanner/schema_dummy_scanner.h b/be/src/exec/schema_scanner/schema_dummy_scanner.h
index a67f6fa25c1648..0c5e4aabe357e4 100644
--- a/be/src/exec/schema_scanner/schema_dummy_scanner.h
+++ b/be/src/exec/schema_scanner/schema_dummy_scanner.h
@@ -33,7 +33,7 @@ class SchemaDummyScanner : public SchemaScanner {
     SchemaDummyScanner();
     ~SchemaDummyScanner() override;
     Status start(RuntimeState* state = nullptr) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 };
 
 } // namespace doris
diff --git a/be/src/exec/schema_scanner/schema_files_scanner.cpp b/be/src/exec/schema_scanner/schema_files_scanner.cpp
index 55b7a338c319e8..20aa07fa69116c 100644
--- a/be/src/exec/schema_scanner/schema_files_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_files_scanner.cpp
@@ -113,7 +113,7 @@ Status SchemaFilesScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaFilesScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaFilesScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_files_scanner.h b/be/src/exec/schema_scanner/schema_files_scanner.h
index 6805a04be4aacc..bb3b2d68493147 100644
--- a/be/src/exec/schema_scanner/schema_files_scanner.h
+++ b/be/src/exec/schema_scanner/schema_files_scanner.h
@@ -38,7 +38,7 @@ class SchemaFilesScanner : public SchemaScanner {
     ~SchemaFilesScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     int _db_index;
     int _table_index;
diff --git a/be/src/exec/schema_scanner/schema_metadata_name_ids_scanner.cpp b/be/src/exec/schema_scanner/schema_metadata_name_ids_scanner.cpp
index 928567a2e4a99e..1267c32c8d8dfb 100644
--- a/be/src/exec/schema_scanner/schema_metadata_name_ids_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_metadata_name_ids_scanner.cpp
@@ -225,7 +225,7 @@ Status SchemaMetadataNameIdsScanner::_fill_block_impl(vectorized::Block* block)
     return Status::OK();
 }
 
-Status SchemaMetadataNameIdsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaMetadataNameIdsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_metadata_name_ids_scanner.h b/be/src/exec/schema_scanner/schema_metadata_name_ids_scanner.h
index 9981d441d856aa..c3beea7769754d 100644
--- a/be/src/exec/schema_scanner/schema_metadata_name_ids_scanner.h
+++ b/be/src/exec/schema_scanner/schema_metadata_name_ids_scanner.h
@@ -39,7 +39,7 @@ class SchemaMetadataNameIdsScanner : public SchemaScanner {
     ~SchemaMetadataNameIdsScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _get_new_table();
diff --git a/be/src/exec/schema_scanner/schema_partitions_scanner.cpp b/be/src/exec/schema_scanner/schema_partitions_scanner.cpp
index f1ad1f594f883f..ea7394e15e12d2 100644
--- a/be/src/exec/schema_scanner/schema_partitions_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_partitions_scanner.cpp
@@ -101,7 +101,7 @@ Status SchemaPartitionsScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaPartitionsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaPartitionsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_partitions_scanner.h b/be/src/exec/schema_scanner/schema_partitions_scanner.h
index 47e1d1fcf87d15..87e55db984a3de 100644
--- a/be/src/exec/schema_scanner/schema_partitions_scanner.h
+++ b/be/src/exec/schema_scanner/schema_partitions_scanner.h
@@ -38,7 +38,7 @@ class SchemaPartitionsScanner : public SchemaScanner {
     ~SchemaPartitionsScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     int _db_index;
     int _table_index;
diff --git a/be/src/exec/schema_scanner/schema_processlist_scanner.cpp b/be/src/exec/schema_scanner/schema_processlist_scanner.cpp
index 0f270a6a8c1777..caf3dcc5af99b2 100644
--- a/be/src/exec/schema_scanner/schema_processlist_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_processlist_scanner.cpp
@@ -63,7 +63,7 @@ Status SchemaProcessListScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaProcessListScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaProcessListScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("call this before initial.");
     }
diff --git a/be/src/exec/schema_scanner/schema_processlist_scanner.h b/be/src/exec/schema_scanner/schema_processlist_scanner.h
index 8aae87e1ef6d0f..c0b0a47f6154ee 100644
--- a/be/src/exec/schema_scanner/schema_processlist_scanner.h
+++ b/be/src/exec/schema_scanner/schema_processlist_scanner.h
@@ -40,7 +40,7 @@ class SchemaProcessListScanner : public SchemaScanner {
     ~SchemaProcessListScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_processlist_columns;
 
diff --git a/be/src/exec/schema_scanner/schema_profiling_scanner.cpp b/be/src/exec/schema_scanner/schema_profiling_scanner.cpp
index 2f71eb96f2613a..0a2a64330bb018 100644
--- a/be/src/exec/schema_scanner/schema_profiling_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_profiling_scanner.cpp
@@ -88,7 +88,7 @@ Status SchemaProfilingScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaProfilingScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaProfilingScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_profiling_scanner.h b/be/src/exec/schema_scanner/schema_profiling_scanner.h
index 5399cb14eb43f5..6b969a478aca69 100644
--- a/be/src/exec/schema_scanner/schema_profiling_scanner.h
+++ b/be/src/exec/schema_scanner/schema_profiling_scanner.h
@@ -38,7 +38,7 @@ class SchemaProfilingScanner : public SchemaScanner {
     ~SchemaProfilingScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
 };
diff --git a/be/src/exec/schema_scanner/schema_routine_scanner.cpp b/be/src/exec/schema_scanner/schema_routine_scanner.cpp
index 3d55addee6c093..8c263c99d2d6c8 100644
--- a/be/src/exec/schema_scanner/schema_routine_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_routine_scanner.cpp
@@ -141,7 +141,7 @@ Status SchemaRoutinesScanner::get_block_from_fe() {
     return Status::OK();
 }
 
-Status SchemaRoutinesScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaRoutinesScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_routine_scanner.h b/be/src/exec/schema_scanner/schema_routine_scanner.h
index 543f9e8e8f684a..c60d72340e1104 100644
--- a/be/src/exec/schema_scanner/schema_routine_scanner.h
+++ b/be/src/exec/schema_scanner/schema_routine_scanner.h
@@ -36,7 +36,7 @@ class SchemaRoutinesScanner : public SchemaScanner {
     ~SchemaRoutinesScanner() override = default;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
 
diff --git a/be/src/exec/schema_scanner/schema_rowsets_scanner.cpp b/be/src/exec/schema_scanner/schema_rowsets_scanner.cpp
index 6ece8e22331e38..16d5f2daba61e7 100644
--- a/be/src/exec/schema_scanner/schema_rowsets_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_rowsets_scanner.cpp
@@ -97,7 +97,7 @@ Status SchemaRowsetsScanner::_get_all_rowsets() {
     return Status::OK();
 }
 
-Status SchemaRowsetsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaRowsetsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_rowsets_scanner.h b/be/src/exec/schema_scanner/schema_rowsets_scanner.h
index b975cc4231bc20..cad34fc04945e4 100644
--- a/be/src/exec/schema_scanner/schema_rowsets_scanner.h
+++ b/be/src/exec/schema_scanner/schema_rowsets_scanner.h
@@ -40,7 +40,7 @@ class SchemaRowsetsScanner : public SchemaScanner {
     ~SchemaRowsetsScanner() override = default;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _get_all_rowsets();
diff --git a/be/src/exec/schema_scanner/schema_schema_privileges_scanner.cpp b/be/src/exec/schema_scanner/schema_schema_privileges_scanner.cpp
index 9789b6c72d6f30..f529821e5a54e2 100644
--- a/be/src/exec/schema_scanner/schema_schema_privileges_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_schema_privileges_scanner.cpp
@@ -82,7 +82,7 @@ Status SchemaSchemaPrivilegesScanner::_get_new_table() {
     return Status::OK();
 }
 
-Status SchemaSchemaPrivilegesScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaSchemaPrivilegesScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_schema_privileges_scanner.h b/be/src/exec/schema_scanner/schema_schema_privileges_scanner.h
index af2ad49634bd49..9522fba908bb2a 100644
--- a/be/src/exec/schema_scanner/schema_schema_privileges_scanner.h
+++ b/be/src/exec/schema_scanner/schema_schema_privileges_scanner.h
@@ -38,7 +38,7 @@ class SchemaSchemaPrivilegesScanner : public SchemaScanner {
     ~SchemaSchemaPrivilegesScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _get_new_table();
diff --git a/be/src/exec/schema_scanner/schema_schemata_scanner.cpp b/be/src/exec/schema_scanner/schema_schemata_scanner.cpp
index 1854e4f2b54af1..618e831c90e219 100644
--- a/be/src/exec/schema_scanner/schema_schemata_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_schemata_scanner.cpp
@@ -81,7 +81,7 @@ Status SchemaSchemataScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaSchemataScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaSchemataScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before Initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_schemata_scanner.h b/be/src/exec/schema_scanner/schema_schemata_scanner.h
index 46fad31af1fd5e..39a5ddda495bdd 100644
--- a/be/src/exec/schema_scanner/schema_schemata_scanner.h
+++ b/be/src/exec/schema_scanner/schema_schemata_scanner.h
@@ -38,7 +38,7 @@ class SchemaSchemataScanner : public SchemaScanner {
     ~SchemaSchemataScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _fill_block_impl(vectorized::Block* block);
diff --git a/be/src/exec/schema_scanner/schema_table_options_scanner.cpp b/be/src/exec/schema_scanner/schema_table_options_scanner.cpp
index 604da59b6377b6..e0481599a2f004 100644
--- a/be/src/exec/schema_scanner/schema_table_options_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_table_options_scanner.cpp
@@ -103,7 +103,7 @@ Status SchemaTableOptionsScanner::get_block_from_fe() {
     return Status::OK();
 }
 
-Status SchemaTableOptionsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaTableOptionsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_table_options_scanner.h b/be/src/exec/schema_scanner/schema_table_options_scanner.h
index d40f1b73c633c5..95c8bdb89076d7 100644
--- a/be/src/exec/schema_scanner/schema_table_options_scanner.h
+++ b/be/src/exec/schema_scanner/schema_table_options_scanner.h
@@ -36,7 +36,7 @@ class SchemaTableOptionsScanner : public SchemaScanner {
     ~SchemaTableOptionsScanner() override = default;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
 
diff --git a/be/src/exec/schema_scanner/schema_table_privileges_scanner.cpp b/be/src/exec/schema_scanner/schema_table_privileges_scanner.cpp
index fe8aa725b73b80..cdeac2b70dcadd 100644
--- a/be/src/exec/schema_scanner/schema_table_privileges_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_table_privileges_scanner.cpp
@@ -84,7 +84,7 @@ Status SchemaTablePrivilegesScanner::_get_new_table() {
     return Status::OK();
 }
 
-Status SchemaTablePrivilegesScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaTablePrivilegesScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_table_privileges_scanner.h b/be/src/exec/schema_scanner/schema_table_privileges_scanner.h
index aa79c88304b7c5..4cfcc16d3583ce 100644
--- a/be/src/exec/schema_scanner/schema_table_privileges_scanner.h
+++ b/be/src/exec/schema_scanner/schema_table_privileges_scanner.h
@@ -38,7 +38,7 @@ class SchemaTablePrivilegesScanner : public SchemaScanner {
     ~SchemaTablePrivilegesScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _get_new_table();
diff --git a/be/src/exec/schema_scanner/schema_tables_scanner.cpp b/be/src/exec/schema_scanner/schema_tables_scanner.cpp
index 093acf9cecbcb1..23710b81971c15 100644
--- a/be/src/exec/schema_scanner/schema_tables_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_tables_scanner.cpp
@@ -342,7 +342,7 @@ Status SchemaTablesScanner::_fill_block_impl(vectorized::Block* block) {
     return Status::OK();
 }
 
-Status SchemaTablesScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaTablesScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_tables_scanner.h b/be/src/exec/schema_scanner/schema_tables_scanner.h
index 11a96bf65d5271..7f8eb11f397e06 100644
--- a/be/src/exec/schema_scanner/schema_tables_scanner.h
+++ b/be/src/exec/schema_scanner/schema_tables_scanner.h
@@ -39,7 +39,7 @@ class SchemaTablesScanner : public SchemaScanner {
     ~SchemaTablesScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _get_new_table();
diff --git a/be/src/exec/schema_scanner/schema_user_privileges_scanner.cpp b/be/src/exec/schema_scanner/schema_user_privileges_scanner.cpp
index 6a12d846fbd560..3eeabc0e4a0917 100644
--- a/be/src/exec/schema_scanner/schema_user_privileges_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_user_privileges_scanner.cpp
@@ -81,7 +81,7 @@ Status SchemaUserPrivilegesScanner::_get_new_table() {
     return Status::OK();
 }
 
-Status SchemaUserPrivilegesScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaUserPrivilegesScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_user_privileges_scanner.h b/be/src/exec/schema_scanner/schema_user_privileges_scanner.h
index eb8f3c63f1433b..ffc3840db676c4 100644
--- a/be/src/exec/schema_scanner/schema_user_privileges_scanner.h
+++ b/be/src/exec/schema_scanner/schema_user_privileges_scanner.h
@@ -38,7 +38,7 @@ class SchemaUserPrivilegesScanner : public SchemaScanner {
     ~SchemaUserPrivilegesScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _get_new_table();
diff --git a/be/src/exec/schema_scanner/schema_user_scanner.cpp b/be/src/exec/schema_scanner/schema_user_scanner.cpp
index 9b153414380350..e56f18f05aea93 100644
--- a/be/src/exec/schema_scanner/schema_user_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_user_scanner.cpp
@@ -76,7 +76,7 @@ Status SchemaUserScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaUserScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaUserScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("call this before initial.");
     }
diff --git a/be/src/exec/schema_scanner/schema_user_scanner.h b/be/src/exec/schema_scanner/schema_user_scanner.h
index c55f216804d5dd..bdc618eb5a0332 100644
--- a/be/src/exec/schema_scanner/schema_user_scanner.h
+++ b/be/src/exec/schema_scanner/schema_user_scanner.h
@@ -40,7 +40,7 @@ class SchemaUserScanner : public SchemaScanner {
     ~SchemaUserScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_user_columns;
 
diff --git a/be/src/exec/schema_scanner/schema_variables_scanner.cpp b/be/src/exec/schema_scanner/schema_variables_scanner.cpp
index 546a0a471cfb01..9ebd4d7781f7c2 100644
--- a/be/src/exec/schema_scanner/schema_variables_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_variables_scanner.cpp
@@ -70,7 +70,7 @@ Status SchemaVariablesScanner::start(RuntimeState* state) {
     return Status::OK();
 }
 
-Status SchemaVariablesScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaVariablesScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("call this before initial.");
     }
diff --git a/be/src/exec/schema_scanner/schema_variables_scanner.h b/be/src/exec/schema_scanner/schema_variables_scanner.h
index 2d207ff8b2e6c2..31bbacf713be0f 100644
--- a/be/src/exec/schema_scanner/schema_variables_scanner.h
+++ b/be/src/exec/schema_scanner/schema_variables_scanner.h
@@ -40,7 +40,7 @@ class SchemaVariablesScanner : public SchemaScanner {
     ~SchemaVariablesScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     struct VariableStruct {
diff --git a/be/src/exec/schema_scanner/schema_views_scanner.cpp b/be/src/exec/schema_scanner/schema_views_scanner.cpp
index 6c3b5f2e21bc3a..f47766ef3567ad 100644
--- a/be/src/exec/schema_scanner/schema_views_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_views_scanner.cpp
@@ -113,7 +113,7 @@ Status SchemaViewsScanner::_get_new_table() {
     return Status::OK();
 }
 
-Status SchemaViewsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaViewsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_views_scanner.h b/be/src/exec/schema_scanner/schema_views_scanner.h
index bc473057905a12..b86ad922e5e76a 100644
--- a/be/src/exec/schema_scanner/schema_views_scanner.h
+++ b/be/src/exec/schema_scanner/schema_views_scanner.h
@@ -38,7 +38,7 @@ class SchemaViewsScanner : public SchemaScanner {
     ~SchemaViewsScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
 private:
     Status _get_new_table();
diff --git a/be/src/exec/schema_scanner/schema_workload_groups_scanner.cpp b/be/src/exec/schema_scanner/schema_workload_groups_scanner.cpp
index ad9be85ad2e832..dd81a3ecb267da 100644
--- a/be/src/exec/schema_scanner/schema_workload_groups_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_workload_groups_scanner.cpp
@@ -116,7 +116,7 @@ Status SchemaWorkloadGroupsScanner::_get_workload_groups_block_from_fe() {
     return Status::OK();
 }
 
-Status SchemaWorkloadGroupsScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaWorkloadGroupsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_workload_groups_scanner.h b/be/src/exec/schema_scanner/schema_workload_groups_scanner.h
index bf7a103526dc80..3121c4dbac149e 100644
--- a/be/src/exec/schema_scanner/schema_workload_groups_scanner.h
+++ b/be/src/exec/schema_scanner/schema_workload_groups_scanner.h
@@ -36,7 +36,7 @@ class SchemaWorkloadGroupsScanner : public SchemaScanner {
     ~SchemaWorkloadGroupsScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
 
diff --git a/be/src/exec/schema_scanner/schema_workload_sched_policy_scanner.cpp b/be/src/exec/schema_scanner/schema_workload_sched_policy_scanner.cpp
index 035d3bfe217aec..2d91f151f5f2bb 100644
--- a/be/src/exec/schema_scanner/schema_workload_sched_policy_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_workload_sched_policy_scanner.cpp
@@ -106,7 +106,8 @@ Status SchemaWorkloadSchedulePolicyScanner::_get_workload_schedule_policy_block_
     return Status::OK();
 }
 
-Status SchemaWorkloadSchedulePolicyScanner::get_next_block(vectorized::Block* block, bool* eos) {
+Status SchemaWorkloadSchedulePolicyScanner::get_next_block_internal(vectorized::Block* block,
+                                                                    bool* eos) {
     if (!_is_init) {
         return Status::InternalError("Used before initialized.");
     }
diff --git a/be/src/exec/schema_scanner/schema_workload_sched_policy_scanner.h b/be/src/exec/schema_scanner/schema_workload_sched_policy_scanner.h
index 5284975fe66b31..da8d9f15c4989e 100644
--- a/be/src/exec/schema_scanner/schema_workload_sched_policy_scanner.h
+++ b/be/src/exec/schema_scanner/schema_workload_sched_policy_scanner.h
@@ -36,7 +36,7 @@ class SchemaWorkloadSchedulePolicyScanner : public SchemaScanner {
     ~SchemaWorkloadSchedulePolicyScanner() override;
 
     Status start(RuntimeState* state) override;
-    Status get_next_block(vectorized::Block* block, bool* eos) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
 
     static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
 
diff --git a/be/src/pipeline/dependency.h b/be/src/pipeline/dependency.h
index 8adc24d3b4ed7c..ab0f5216581c74 100644
--- a/be/src/pipeline/dependency.h
+++ b/be/src/pipeline/dependency.h
@@ -88,20 +88,11 @@ class Dependency : public std::enable_shared_from_this<Dependency> {
 public:
     ENABLE_FACTORY_CREATOR(Dependency);
     Dependency(int id, int node_id, std::string name)
-            : _id(id),
-              _node_id(node_id),
-              _name(std::move(name)),
-              _is_write_dependency(false),
-              _ready(false) {}
+            : _id(id), _node_id(node_id), _name(std::move(name)), _ready(false) {}
     Dependency(int id, int node_id, std::string name, bool ready)
-            : _id(id),
-              _node_id(node_id),
-              _name(std::move(name)),
-              _is_write_dependency(true),
-              _ready(ready) {}
+            : _id(id), _node_id(node_id), _name(std::move(name)), _ready(ready) {}
     virtual ~Dependency() = default;
 
-    bool is_write_dependency() const { return _is_write_dependency; }
     [[nodiscard]] int id() const { return _id; }
     [[nodiscard]] virtual std::string name() const { return _name; }
     BasicSharedState* shared_state() { return _shared_state; }
@@ -118,12 +109,10 @@ class Dependency : public std::enable_shared_from_this<Dependency> {
     // Notify downstream pipeline tasks this dependency is ready.
     void set_ready();
     void set_ready_to_read() {
-        DCHECK(_is_write_dependency) << debug_string();
         DCHECK(_shared_state->source_deps.size() == 1) << debug_string();
         _shared_state->source_deps.front()->set_ready();
     }
     void set_block_to_read() {
-        DCHECK(_is_write_dependency) << debug_string();
         DCHECK(_shared_state->source_deps.size() == 1) << debug_string();
         _shared_state->source_deps.front()->block();
     }
@@ -166,7 +155,6 @@ class Dependency : public std::enable_shared_from_this<Dependency> {
     const int _id;
     const int _node_id;
     const std::string _name;
-    const bool _is_write_dependency;
     std::atomic<bool> _ready;
 
     BasicSharedState* _shared_state = nullptr;
diff --git a/be/src/pipeline/exec/schema_scan_operator.cpp b/be/src/pipeline/exec/schema_scan_operator.cpp
index f46589880958ee..f1f3608aa20b0d 100644
--- a/be/src/pipeline/exec/schema_scan_operator.cpp
+++ b/be/src/pipeline/exec/schema_scan_operator.cpp
@@ -48,6 +48,7 @@ Status SchemaScanLocalState::init(RuntimeState* state, LocalStateInfo& info) {
     // new one scanner
     _schema_scanner = SchemaScanner::create(schema_table->schema_table_type());
 
+    _schema_scanner->set_dependency(_data_dependency, _finish_dependency);
     if (nullptr == _schema_scanner) {
         return Status::InternalError("schema scanner get nullptr pointer.");
     }
@@ -59,7 +60,7 @@ Status SchemaScanLocalState::open(RuntimeState* state) {
     SCOPED_TIMER(exec_time_counter());
     SCOPED_TIMER(_open_timer);
     RETURN_IF_ERROR(PipelineXLocalState<>::open(state));
-    return _schema_scanner->start(state);
+    return _schema_scanner->get_next_block_async(state);
 }
 
 SchemaScanOperatorX::SchemaScanOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
@@ -226,8 +227,12 @@ Status SchemaScanOperatorX::get_block(RuntimeState* state, vectorized::Block* bl
         while (true) {
             RETURN_IF_CANCELLED(state);
 
+            if (local_state._data_dependency->is_blocked_by() != nullptr) {
+                break;
+            }
             // get all slots from schema table.
-            RETURN_IF_ERROR(local_state._schema_scanner->get_next_block(&src_block, &schema_eos));
+            RETURN_IF_ERROR(
+                    local_state._schema_scanner->get_next_block(state, &src_block, &schema_eos));
 
             if (schema_eos) {
                 *eos = true;
diff --git a/be/src/pipeline/exec/schema_scan_operator.h b/be/src/pipeline/exec/schema_scan_operator.h
index 8f2b73f5123f0d..aa2bff7e6440a2 100644
--- a/be/src/pipeline/exec/schema_scan_operator.h
+++ b/be/src/pipeline/exec/schema_scan_operator.h
@@ -35,18 +35,30 @@ class SchemaScanLocalState final : public PipelineXLocalState<> {
     ENABLE_FACTORY_CREATOR(SchemaScanLocalState);
 
     SchemaScanLocalState(RuntimeState* state, OperatorXBase* parent)
-            : PipelineXLocalState<>(state, parent) {}
+            : PipelineXLocalState<>(state, parent) {
+        _finish_dependency =
+                std::make_shared<Dependency>(parent->operator_id(), parent->node_id(),
+                                             parent->get_name() + "_FINISH_DEPENDENCY", true);
+        _data_dependency = std::make_shared<Dependency>(parent->operator_id(), parent->node_id(),
+                                                        parent->get_name() + "_DEPENDENCY", true);
+    }
     ~SchemaScanLocalState() override = default;
 
     Status init(RuntimeState* state, LocalStateInfo& info) override;
 
     Status open(RuntimeState* state) override;
 
+    Dependency* finishdependency() override { return _finish_dependency.get(); }
+    std::vector<Dependency*> dependencies() const override { return {_data_dependency.get()}; }
+
 private:
     friend class SchemaScanOperatorX;
 
     SchemaScannerParam _scanner_param;
     std::unique_ptr<SchemaScanner> _schema_scanner;
+
+    std::shared_ptr<Dependency> _finish_dependency;
+    std::shared_ptr<Dependency> _data_dependency;
 };
 
 class SchemaScanOperatorX final : public OperatorX<SchemaScanLocalState> {
diff --git a/be/src/pipeline/pipeline_task.cpp b/be/src/pipeline/pipeline_task.cpp
index b8a52575b97682..9a287dec6386dd 100644
--- a/be/src/pipeline/pipeline_task.cpp
+++ b/be/src/pipeline/pipeline_task.cpp
@@ -153,8 +153,6 @@ Status PipelineTask::_extract_dependencies() {
     {
         auto* local_state = _state->get_sink_local_state();
         write_dependencies = local_state->dependencies();
-        DCHECK(std::all_of(write_dependencies.begin(), write_dependencies.end(),
-                           [](auto* dep) { return dep->is_write_dependency(); }));
         auto* fin_dep = local_state->finishdependency();
         if (fin_dep) {
             finish_dependencies.push_back(fin_dep);