You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2024/06/01 10:02:30 UTC
(doris) branch remove_code created (now 3d70d7ab904)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a change to branch remove_code
in repository https://gitbox.apache.org/repos/asf/doris.git
at 3d70d7ab904 [chore](cleancode) remove unused code from be
This branch includes the following new commits:
new 3d70d7ab904 [chore](cleancode) remove unused code from be
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/01: [chore](cleancode) remove unused code from be
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch remove_code
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 3d70d7ab90435a7919e49503c88e5bd0a6a7a86a
Author: yiguolei <yi...@gmail.com>
AuthorDate: Sat Jun 1 18:02:03 2024 +0800
[chore](cleancode) remove unused code from be
---
be/src/exec/data_sink.h | 2 --
be/src/exec/exec_node.h | 2 --
be/src/vec/exec/vanalytic_eval_node.cpp | 13 -------------
be/src/vec/exec/vanalytic_eval_node.h | 2 --
be/src/vec/exec/vpartition_sort_node.cpp | 5 -----
be/src/vec/exec/vpartition_sort_node.h | 1 -
be/src/vec/sink/async_writer_sink.h | 4 ----
be/src/vec/sink/multi_cast_data_stream_sink.h | 3 ---
be/src/vec/sink/vdata_stream_sender.cpp | 16 ----------------
be/src/vec/sink/vdata_stream_sender.h | 13 -------------
be/src/vec/sink/vmemory_scratch_sink.cpp | 4 ----
be/src/vec/sink/vmemory_scratch_sink.h | 2 --
be/src/vec/sink/writer/async_result_writer.h | 7 -------
13 files changed, 74 deletions(-)
diff --git a/be/src/exec/data_sink.h b/be/src/exec/data_sink.h
index 5258929ba79..97842f5acb4 100644
--- a/be/src/exec/data_sink.h
+++ b/be/src/exec/data_sink.h
@@ -71,8 +71,6 @@ public:
return send(state, block, eos);
}
- [[nodiscard]] virtual bool is_pending_finish() const { return false; }
-
// Releases all resources that were allocated in prepare()/send().
// Further send() calls are illegal after calling close().
// It must be okay to call this multiple times. Subsequent calls should
diff --git a/be/src/exec/exec_node.h b/be/src/exec/exec_node.h
index 10b035835d7..2dedee61ba5 100644
--- a/be/src/exec/exec_node.h
+++ b/be/src/exec/exec_node.h
@@ -142,8 +142,6 @@ public:
return Status::OK();
}
- bool can_read() const { return _can_read; }
-
[[nodiscard]] virtual bool can_terminate_early() { return false; }
// Sink Data to ExecNode to do some stock work, both need impl with method: get_result
diff --git a/be/src/vec/exec/vanalytic_eval_node.cpp b/be/src/vec/exec/vanalytic_eval_node.cpp
index fbd49aa145a..410964c1969 100644
--- a/be/src/vec/exec/vanalytic_eval_node.cpp
+++ b/be/src/vec/exec/vanalytic_eval_node.cpp
@@ -298,19 +298,6 @@ void VAnalyticEvalNode::release_resource(RuntimeState* state) {
return ExecNode::release_resource(state);
}
-//TODO: maybe could have better strategy, not noly when need data to sink data
-//even could get some resources in advance as soon as possible
-bool VAnalyticEvalNode::can_write() {
- return _need_more_input;
-}
-
-bool VAnalyticEvalNode::can_read() {
- if (_need_more_input) {
- return false;
- }
- return true;
-}
-
Status VAnalyticEvalNode::get_next(RuntimeState* state, vectorized::Block* block, bool* eos) {
SCOPED_TIMER(_runtime_profile->total_time_counter());
RETURN_IF_CANCELLED(state);
diff --git a/be/src/vec/exec/vanalytic_eval_node.h b/be/src/vec/exec/vanalytic_eval_node.h
index 45f7ce5b1e8..9b302b32ed6 100644
--- a/be/src/vec/exec/vanalytic_eval_node.h
+++ b/be/src/vec/exec/vanalytic_eval_node.h
@@ -83,8 +83,6 @@ public:
void release_resource(RuntimeState* state) override;
Status sink(doris::RuntimeState* state, vectorized::Block* input_block, bool eos) override;
Status pull(doris::RuntimeState* state, vectorized::Block* output_block, bool* eos) override;
- bool can_read();
- bool can_write();
protected:
using ExecNode::debug_string;
diff --git a/be/src/vec/exec/vpartition_sort_node.cpp b/be/src/vec/exec/vpartition_sort_node.cpp
index 15d8124c653..a8b986d130e 100644
--- a/be/src/vec/exec/vpartition_sort_node.cpp
+++ b/be/src/vec/exec/vpartition_sort_node.cpp
@@ -317,11 +317,6 @@ Status VPartitionSortNode::alloc_resource(RuntimeState* state) {
return Status::OK();
}
-bool VPartitionSortNode::can_read() {
- std::lock_guard<std::mutex> lock(_buffer_mutex);
- return !_blocks_buffer.empty() || _can_read;
-}
-
Status VPartitionSortNode::pull(doris::RuntimeState* state, vectorized::Block* output_block,
bool* eos) {
SCOPED_TIMER(_exec_timer);
diff --git a/be/src/vec/exec/vpartition_sort_node.h b/be/src/vec/exec/vpartition_sort_node.h
index 481a99719fb..a9edca80df9 100644
--- a/be/src/vec/exec/vpartition_sort_node.h
+++ b/be/src/vec/exec/vpartition_sort_node.h
@@ -233,7 +233,6 @@ public:
Status sink(RuntimeState* state, vectorized::Block* input_block, bool eos) override;
void debug_profile();
- bool can_read();
private:
Status _init_hash_method();
diff --git a/be/src/vec/sink/async_writer_sink.h b/be/src/vec/sink/async_writer_sink.h
index 150ebeaf4f2..526064404a4 100644
--- a/be/src/vec/sink/async_writer_sink.h
+++ b/be/src/vec/sink/async_writer_sink.h
@@ -94,8 +94,6 @@ public:
return _writer->sink(block, eos);
}
- bool can_write() override { return _writer->can_write(); }
-
Status close(RuntimeState* state, Status exec_status) override {
// if the init failed, the _writer may be nullptr. so here need check
if (_writer) {
@@ -104,8 +102,6 @@ public:
return DataSink::close(state, exec_status);
}
- [[nodiscard]] bool is_pending_finish() const override { return _writer->is_pending_finish(); }
-
protected:
const std::vector<TExpr>& _t_output_expr;
VExprContextSPtrs _output_vexpr_ctxs;
diff --git a/be/src/vec/sink/multi_cast_data_stream_sink.h b/be/src/vec/sink/multi_cast_data_stream_sink.h
index 7cc057013aa..d6b85010c5a 100644
--- a/be/src/vec/sink/multi_cast_data_stream_sink.h
+++ b/be/src/vec/sink/multi_cast_data_stream_sink.h
@@ -41,9 +41,6 @@ public:
Status open(doris::RuntimeState* state) override { return Status::OK(); };
- // use sink to check can_write, now always true after we support spill to disk
- bool can_write() override { return _multi_cast_data_streamer->can_write(); }
-
std::shared_ptr<pipeline::MultiCastDataStreamer>& get_multi_cast_data_streamer() {
return _multi_cast_data_streamer;
}
diff --git a/be/src/vec/sink/vdata_stream_sender.cpp b/be/src/vec/sink/vdata_stream_sender.cpp
index 63f2aa19515..529a8256e77 100644
--- a/be/src/vec/sink/vdata_stream_sender.cpp
+++ b/be/src/vec/sink/vdata_stream_sender.cpp
@@ -925,22 +925,6 @@ void VDataStreamSender::register_pipeline_channels(pipeline::ExchangeSinkBuffer*
}
}
-bool VDataStreamSender::channel_all_can_write() {
- if ((_part_type == TPartitionType::UNPARTITIONED || _channels.size() == 1) &&
- !_only_local_exchange) {
- // This condition means we need use broadcast buffer, so we should make sure
- // there are available buffer before running pipeline
- return !_broadcast_pb_blocks->empty();
- } else {
- for (auto channel : _channels) {
- if (!channel->can_write()) {
- return false;
- }
- }
- return true;
- }
-}
-
template class Channel<pipeline::ExchangeSinkLocalState>;
template class Channel<VDataStreamSender>;
template class Channel<pipeline::ResultFileSinkLocalState>;
diff --git a/be/src/vec/sink/vdata_stream_sender.h b/be/src/vec/sink/vdata_stream_sender.h
index 5ca31bcbe44..b6346787ebb 100644
--- a/be/src/vec/sink/vdata_stream_sender.h
+++ b/be/src/vec/sink/vdata_stream_sender.h
@@ -132,8 +132,6 @@ public:
void register_pipeline_channels(pipeline::ExchangeSinkBuffer* buffer);
- bool channel_all_can_write();
-
int sender_id() const { return _sender_id; }
RuntimeProfile::Counter* brpc_wait_timer() { return _brpc_wait_timer; }
@@ -327,17 +325,6 @@ public:
virtual void ch_roll_pb_block();
- bool can_write() {
- if (!is_local()) {
- return true;
- }
-
- // if local recvr queue mem over the exchange node mem limit, we must ensure each queue
- // has one block to do merge sort in exchange node to prevent the logic dead lock
- return !_local_recvr || _local_recvr->is_closed() || !_local_recvr->exceeds_limit(0) ||
- _local_recvr->sender_queue_empty(_parent->sender_id());
- }
-
bool is_receiver_eof() const { return _receiver_status.is<ErrorCode::END_OF_FILE>(); }
void set_receiver_eof(Status st) { _receiver_status = st; }
diff --git a/be/src/vec/sink/vmemory_scratch_sink.cpp b/be/src/vec/sink/vmemory_scratch_sink.cpp
index eca9e65ab49..95266ba6de0 100644
--- a/be/src/vec/sink/vmemory_scratch_sink.cpp
+++ b/be/src/vec/sink/vmemory_scratch_sink.cpp
@@ -99,10 +99,6 @@ Status MemoryScratchSink::open(RuntimeState* state) {
return VExpr::open(_output_vexpr_ctxs, state);
}
-bool MemoryScratchSink::can_write() {
- return _queue->size() < 10;
-}
-
Status MemoryScratchSink::close(RuntimeState* state, Status exec_status) {
if (_closed) {
return Status::OK();
diff --git a/be/src/vec/sink/vmemory_scratch_sink.h b/be/src/vec/sink/vmemory_scratch_sink.h
index 3a1dd9991d4..7e2d042cb8d 100644
--- a/be/src/vec/sink/vmemory_scratch_sink.h
+++ b/be/src/vec/sink/vmemory_scratch_sink.h
@@ -62,8 +62,6 @@ public:
Status close(RuntimeState* state, Status exec_status) override;
- bool can_write() override;
-
private:
Status _prepare_vexpr(RuntimeState* state);
cctz::time_zone _timezone_obj;
diff --git a/be/src/vec/sink/writer/async_result_writer.h b/be/src/vec/sink/writer/async_result_writer.h
index b1426a48806..5e21dc13e12 100644
--- a/be/src/vec/sink/writer/async_result_writer.h
+++ b/be/src/vec/sink/writer/async_result_writer.h
@@ -65,13 +65,6 @@ public:
virtual Status open(RuntimeState* state, RuntimeProfile* profile) = 0;
- bool can_write() {
- std::lock_guard l(_m);
- return _data_queue_is_available() || _is_finished();
- }
-
- [[nodiscard]] bool is_pending_finish() const { return !_writer_thread_closed; }
-
// sink the block date to date queue, it is async
Status sink(Block* block, bool eos);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org