This is an automated email from the ASF dual-hosted git repository.
gabriellee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new b8dcac05bae [fix](scanner) Delete meaningless finish dependency in
schema scanner (#44915)
b8dcac05bae is described below
commit b8dcac05bae57f2a2c87f5f2f2d1acdcf5912850
Author: Gabriel <liwenqi...@selectdb.com>
AuthorDate: Wed Dec 4 10:03:54 2024 +0800
[fix](scanner) Delete meaningless finish dependency in schema scanner
(#44915)
Finish dependency is used to block `close` phase in order to release
some necessary resources safely. In schema scanner, an async thread is
used to do scanning and finish dependency is used to wait it done. But
in fact, we could close schema scanner directly now so this PR delete
this finish dependency.
---
be/src/exec/schema_scanner.cpp | 4 ----
be/src/exec/schema_scanner.h | 7 +------
be/src/pipeline/exec/schema_scan_operator.cpp | 5 +----
be/src/pipeline/exec/schema_scan_operator.h | 5 -----
4 files changed, 2 insertions(+), 19 deletions(-)
diff --git a/be/src/exec/schema_scanner.cpp b/be/src/exec/schema_scanner.cpp
index 39dd4516332..4b430f04289 100644
--- a/be/src/exec/schema_scanner.cpp
+++ b/be/src/exec/schema_scanner.cpp
@@ -124,7 +124,6 @@ Status SchemaScanner::get_next_block_async(RuntimeState*
state) {
}
SCOPED_ATTACH_TASK(state);
_async_thread_running = true;
- _finish_dependency->block();
if (!_opened) {
_data_block = vectorized::Block::create_unique();
_init_block(_data_block.get());
@@ -140,9 +139,6 @@ Status SchemaScanner::get_next_block_async(RuntimeState*
state) {
_eos = eos;
_async_thread_running = false;
_dependency->set_ready();
- if (eos) {
- _finish_dependency->set_ready();
- }
}));
return Status::OK();
}
diff --git a/be/src/exec/schema_scanner.h b/be/src/exec/schema_scanner.h
index 440912bff1d..6e7a229b7fd 100644
--- a/be/src/exec/schema_scanner.h
+++ b/be/src/exec/schema_scanner.h
@@ -106,11 +106,7 @@ public:
// 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;
- }
+ void set_dependency(std::shared_ptr<pipeline::Dependency> dep) {
_dependency = dep; }
Status get_next_block_async(RuntimeState* state);
protected:
@@ -139,7 +135,6 @@ protected:
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;
diff --git a/be/src/pipeline/exec/schema_scan_operator.cpp
b/be/src/pipeline/exec/schema_scan_operator.cpp
index 006ecf8ad82..ddc2821cac1 100644
--- a/be/src/pipeline/exec/schema_scan_operator.cpp
+++ b/be/src/pipeline/exec/schema_scan_operator.cpp
@@ -48,7 +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);
+ _schema_scanner->set_dependency(_data_dependency);
if (nullptr == _schema_scanner) {
return Status::InternalError("schema scanner get nullptr pointer.");
}
@@ -266,9 +266,6 @@ Status SchemaScanOperatorX::get_block(RuntimeState* state,
vectorized::Block* bl
} while (block->rows() == 0 && !*eos);
local_state.reached_limit(block, eos);
- if (*eos) {
- local_state._finish_dependency->set_always_ready();
- }
return Status::OK();
}
diff --git a/be/src/pipeline/exec/schema_scan_operator.h
b/be/src/pipeline/exec/schema_scan_operator.h
index 03cf422fbc5..c8ddf885e98 100644
--- a/be/src/pipeline/exec/schema_scan_operator.h
+++ b/be/src/pipeline/exec/schema_scan_operator.h
@@ -36,9 +36,6 @@ public:
SchemaScanLocalState(RuntimeState* state, OperatorXBase* 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);
}
@@ -48,7 +45,6 @@ public:
Status open(RuntimeState* state) override;
- Dependency* finishdependency() override { return _finish_dependency.get();
}
std::vector<Dependency*> dependencies() const override { return
{_data_dependency.get()}; }
private:
@@ -57,7 +53,6 @@ private:
SchemaScannerParam _scanner_param;
std::unique_ptr<SchemaScanner> _schema_scanner;
- std::shared_ptr<Dependency> _finish_dependency;
std::shared_ptr<Dependency> _data_dependency;
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
