github-actions[bot] commented on code in PR #26799:
URL: https://github.com/apache/doris/pull/26799#discussion_r1390068641
##########
be/src/pipeline/exec/multi_cast_data_streamer.cpp:
##########
@@ -87,10 +92,39 @@
for (int i = 0; i < _sender_pos_to_read.size(); ++i) {
if (_sender_pos_to_read[i] == _multi_cast_blocks.end()) {
_sender_pos_to_read[i] = end;
+ _set_ready_for_read(i);
}
}
_eos = eos;
return Status::OK();
}
+void MultiCastDataStreamer::_set_ready_for_read(int sender_idx) {
+ if (!_has_dependencys) {
+ return;
+ }
+ auto* dep = _dependencys[sender_idx];
+ DCHECK(dep);
+ dep->set_ready_for_read();
+}
+
+void MultiCastDataStreamer::_set_ready_for_read() {
Review Comment:
warning: method '_set_ready_for_read' can be made const
[readability-make-member-function-const]
```suggestion
void MultiCastDataStreamer::_set_ready_for_read() const {
```
be/src/pipeline/exec/multi_cast_data_streamer.h:80:
```diff
- void _set_ready_for_read();
+ void _set_ready_for_read() const;
```
##########
be/src/pipeline/exec/multi_cast_data_streamer.cpp:
##########
@@ -87,10 +92,39 @@ Status MultiCastDataStreamer::push(RuntimeState* state,
doris::vectorized::Block
for (int i = 0; i < _sender_pos_to_read.size(); ++i) {
if (_sender_pos_to_read[i] == _multi_cast_blocks.end()) {
_sender_pos_to_read[i] = end;
+ _set_ready_for_read(i);
}
}
_eos = eos;
return Status::OK();
}
+void MultiCastDataStreamer::_set_ready_for_read(int sender_idx) {
Review Comment:
warning: method '_set_ready_for_read' can be made const
[readability-make-member-function-const]
```suggestion
void MultiCastDataStreamer::_set_ready_for_read(int sender_idx) const {
```
be/src/pipeline/exec/multi_cast_data_streamer.h:79:
```diff
- void _set_ready_for_read(int sender_idx);
+ void _set_ready_for_read(int sender_idx) const;
```
##########
be/src/pipeline/pipeline_x/pipeline_x_task.h:
##########
@@ -70,56 +70,64 @@
// must be call after all pipeline task is finish to release resource
Status close(Status exec_status) override;
+ Dependency* read_blocked_dependency() {
+ for (auto* op_dep : _read_dependencies) {
+ _blocked_dep = op_dep->read_blocked_by();
+ if (_blocked_dep != nullptr) {
+ _blocked_dep->start_read_watcher();
+ return _blocked_dep;
+ }
+ }
+ return nullptr;
+ }
+
bool source_can_read() override {
if (_dry_run) {
return true;
}
- for (auto* op_dep : _read_dependencies) {
- auto* dep = op_dep->read_blocked_by();
- if (dep != nullptr) {
- dep->start_read_watcher();
- push_blocked_task_to_dependency(dep);
- return false;
- }
+ return read_blocked_dependency() == nullptr;
+ }
+
+ Dependency* filter_blocked_dependency() {
+ _blocked_dep = _filter_dependency->filter_blocked_by();
+ if (_blocked_dep != nullptr) {
+ return _blocked_dep;
}
- return true;
+ return nullptr;
}
bool runtime_filters_are_ready_or_timeout() override {
- auto* dep = _filter_dependency->filter_blocked_by();
- if (dep != nullptr) {
- push_blocked_task_to_dependency(dep);
- return false;
- }
- return true;
+ return filter_blocked_dependency() == nullptr;
}
- bool sink_can_write() override {
- auto* dep = _write_dependencies->write_blocked_by();
- if (dep != nullptr) {
- dep->start_write_watcher();
- push_blocked_task_to_dependency(dep);
- return false;
+ Dependency* write_blocked_dependency() {
+ _blocked_dep = _write_dependencies->write_blocked_by();
+ if (_blocked_dep != nullptr) {
+ static_cast<WriteDependency*>(_blocked_dep)->start_write_watcher();
+ return _blocked_dep;
}
- return true;
+ return nullptr;
}
+ bool sink_can_write() override { return write_blocked_dependency() ==
nullptr; }
+
Status finalize() override;
std::string debug_string() override;
- bool is_pending_finish() override {
+ Dependency* finish_blocked_dependency() {
Review Comment:
warning: method 'finish_blocked_dependency' can be made static
[readability-convert-member-functions-to-static]
```suggestion
static Dependency* finish_blocked_dependency() {
```
##########
be/src/pipeline/pipeline_x/pipeline_x_task.h:
##########
@@ -70,56 +70,64 @@ class PipelineXTask : public PipelineTask {
// must be call after all pipeline task is finish to release resource
Status close(Status exec_status) override;
+ Dependency* read_blocked_dependency() {
Review Comment:
warning: method 'read_blocked_dependency' can be made static
[readability-convert-member-functions-to-static]
```suggestion
static Dependency* read_blocked_dependency() {
```
##########
be/src/pipeline/pipeline_x/dependency.h:
##########
@@ -899,7 +896,30 @@ struct LocalExchangeSharedState {
public:
ENABLE_FACTORY_CREATOR(LocalExchangeSharedState);
std::vector<moodycamel::ConcurrentQueue<PartitionedBlock>> data_queue;
+ std::vector<Dependency*> source_dependencies;
std::atomic<int> running_sink_operators = 0;
+ void add_running_sink_operators() { running_sink_operators++; }
+ void sub_running_sink_operators() {
+ running_sink_operators--;
+ if (running_sink_operators == 0) {
+ _set_ready_for_read();
+ }
+ }
+ void _set_ready_for_read() {
+ for (auto* dep : source_dependencies) {
+ DCHECK(dep);
+ dep->set_ready_for_read();
+ }
+ }
+ void set_dep_by_channel_id(Dependency* dep, int channel_id) {
Review Comment:
warning: method 'set_dep_by_channel_id' can be made static
[readability-convert-member-functions-to-static]
```suggestion
static void set_dep_by_channel_id(Dependency* dep, int channel_id) {
```
##########
be/src/pipeline/exec/multi_cast_data_streamer.cpp:
##########
@@ -87,10 +92,39 @@
for (int i = 0; i < _sender_pos_to_read.size(); ++i) {
if (_sender_pos_to_read[i] == _multi_cast_blocks.end()) {
_sender_pos_to_read[i] = end;
+ _set_ready_for_read(i);
}
}
_eos = eos;
return Status::OK();
}
+void MultiCastDataStreamer::_set_ready_for_read(int sender_idx) {
+ if (!_has_dependencys) {
+ return;
+ }
+ auto* dep = _dependencys[sender_idx];
+ DCHECK(dep);
+ dep->set_ready_for_read();
+}
+
+void MultiCastDataStreamer::_set_ready_for_read() {
+ if (!_has_dependencys) {
+ return;
+ }
+ for (auto* dep : _dependencys) {
+ DCHECK(dep);
+ dep->set_ready_for_read();
+ }
+}
+
+void MultiCastDataStreamer::_block_reading(int sender_idx) {
Review Comment:
warning: method '_block_reading' can be made const
[readability-make-member-function-const]
```suggestion
void MultiCastDataStreamer::_block_reading(int sender_idx) const {
```
be/src/pipeline/exec/multi_cast_data_streamer.h:81:
```diff
- void _block_reading(int sender_idx);
+ void _block_reading(int sender_idx) const;
```
##########
be/src/pipeline/pipeline_x/dependency.h:
##########
@@ -899,7 +896,30 @@
public:
ENABLE_FACTORY_CREATOR(LocalExchangeSharedState);
std::vector<moodycamel::ConcurrentQueue<PartitionedBlock>> data_queue;
+ std::vector<Dependency*> source_dependencies;
std::atomic<int> running_sink_operators = 0;
+ void add_running_sink_operators() { running_sink_operators++; }
+ void sub_running_sink_operators() {
+ running_sink_operators--;
+ if (running_sink_operators == 0) {
+ _set_ready_for_read();
+ }
+ }
+ void _set_ready_for_read() {
+ for (auto* dep : source_dependencies) {
+ DCHECK(dep);
+ dep->set_ready_for_read();
+ }
+ }
+ void set_dep_by_channel_id(Dependency* dep, int channel_id) {
+ source_dependencies[channel_id] = dep;
+ dep->block_reading();
+ }
+ void _set_ready_for_read(int channel_id) {
Review Comment:
warning: method '_set_ready_for_read' can be made static
[readability-convert-member-functions-to-static]
```suggestion
static void _set_ready_for_read(int channel_id) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
