You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "Mryange (via GitHub)" <gi...@apache.org> on 2023/11/12 12:51:30 UTC
[PR] [only test] pipelineX [doris]
Mryange opened a new pull request, #26839:
URL: https://github.com/apache/doris/pull/26839
## Proposed changes
Issue Number: close #xxx
<!--Describe your changes.-->
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [only test] pipelineX [doris]
Posted by "Mryange (via GitHub)" <gi...@apache.org>.
Mryange commented on PR #26839:
URL: https://github.com/apache/doris/pull/26839#issuecomment-1807118555
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [only test] pipelineX [doris]
Posted by "Mryange (via GitHub)" <gi...@apache.org>.
Mryange closed pull request #26839: [only test] pipelineX
URL: https://github.com/apache/doris/pull/26839
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [only test] pipelineX [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26839:
URL: https://github.com/apache/doris/pull/26839#issuecomment-1807194016
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 44.88 seconds
stream load tsv: 554 seconds loaded 74807831229 Bytes, about 128 MB/s
stream load json: 21 seconds loaded 2358488459 Bytes, about 107 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 33 seconds loaded 861443392 Bytes, about 24 MB/s
insert into select: 29.0 seconds inserted 10000000 Rows, about 344K ops/s
storage size: 17162327283 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [only test] pipelineX [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26839:
URL: https://github.com/apache/doris/pull/26839#issuecomment-1807385335
TeamCity be ut coverage result:
Function Coverage: 36.75% (8407/22876)
Line Coverage: 29.26% (68159/232909)
Region Coverage: 27.90% (35221/126234)
Branch Coverage: 24.72% (18006/72848)
Coverage Report: http://coverage.selectdb-in.cc/coverage/5113d9bc38bb6f1181f77f93a74c224d6730ac33_5113d9bc38bb6f1181f77f93a74c224d6730ac33/report/index.html
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [only test] pipelineX [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26839:
URL: https://github.com/apache/doris/pull/26839#discussion_r1390405900
##########
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/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/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/pipeline_x/dependency.h:
##########
@@ -899,7 +895,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/pipeline_x/dependency.h:
##########
@@ -899,7 +895,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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [only test] pipelineX [doris]
Posted by "Mryange (via GitHub)" <gi...@apache.org>.
Mryange commented on PR #26839:
URL: https://github.com/apache/doris/pull/26839#issuecomment-1807258192
run pipelinex_p0
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [only test] pipelineX [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26839:
URL: https://github.com/apache/doris/pull/26839#issuecomment-1807191414
<details>
<summary>TPC-H test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G'</summary>
```
Tpch sf100 test result on commit 5113d9bc38bb6f1181f77f93a74c224d6730ac33, data reload: false
run tpch-sf100 query with default conf and session variables
q1 5296 5056 5129 5056
q2 368 184 178 178
q3 2070 2056 2096 2056
q4 1485 1414 1419 1414
q5 4137 4170 4092 4092
q6 255 138 132 132
q7 2083 1598 1619 1598
q8 2753 2744 2758 2744
q9 10441 10370 10215 10215
q10 3501 3550 3540 3540
q11 368 254 248 248
q12 446 291 296 291
q13 4505 4128 4092 4092
q14 326 300 303 300
q15 658 582 587 582
q16 703 611 607 607
q17 1153 1094 1080 1080
q18 7796 7375 7360 7360
q19 1695 1708 1691 1691
q20 582 364 353 353
q21 4889 4576 4567 4567
q22 537 426 435 426
Total cold run time: 56047 ms
Total hot run time: 52622 ms
run tpch-sf100 query with default conf and set session variable runtime_filter_mode=off
q1 5061 4949 4951 4949
q2 324 265 246 246
q3 3983 4016 3963 3963
q4 2808 2750 2739 2739
q5 6543 6457 6472 6457
q6 243 129 129 129
q7 3183 2749 2687 2687
q8 4869 4784 4835 4784
q9 17800 17744 17685 17685
q10 4123 4193 4170 4170
q11 748 659 638 638
q12 1000 833 826 826
q13 4347 3913 3915 3913
q14 375 346 347 346
q15 638 591 564 564
q16 770 687 731 687
q17 3922 3893 3890 3890
q18 9339 9078 9314 9078
q19 1849 1800 1781 1781
q20 2368 2067 2034 2034
q21 8919 8714 8749 8714
q22 973 823 862 823
Total cold run time: 84185 ms
Total hot run time: 81103 ms
```
</details>
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org