You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "HappenLee (via GitHub)" <gi...@apache.org> on 2023/02/27 07:53:57 UTC
[GitHub] [doris] HappenLee opened a new pull request, #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
HappenLee opened a new pull request, #17187:
URL: https://github.com/apache/doris/pull/17187
# Proposed changes
clickbench 1 tablet：
<img width="903" alt="image" src="https://user-images.githubusercontent.com/10553413/221505402-92e8e835-7278-4e54-bb6a-8a7017b78f35.png">
## Problem summary
Describe your changes.
## Checklist(Required)
* [ ] Does it affect the original behavior
* [ ] Has unit tests been added
* [ ] Has document been added or modified
* [ ] Does it need to update dependencies
* [ ] Is this PR support rollback (If NO, please explain WHY)
## 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
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #17187:
URL: https://github.com/apache/doris/pull/17187#discussion_r1118466007
##########
be/src/vec/functions/function.cpp:
##########
@@ -37,48 +37,58 @@ namespace doris::vectorized {
ColumnPtr wrap_in_nullable(const ColumnPtr& src, const Block& block, const ColumnNumbers& args,
size_t result, size_t input_rows_count) {
ColumnPtr result_null_map_column;
-
/// If result is already nullable.
ColumnPtr src_not_nullable = src;
+ MutableColumnPtr mutable_result_null_map_column;
- if (src->only_null())
- return src;
- else if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
src_not_nullable = nullable->get_nested_column_ptr();
result_null_map_column = nullable->get_null_map_column_ptr();
}
for (const auto& arg : args) {
const ColumnWithTypeAndName& elem = block.get_by_position(arg);
- if (!elem.type->is_nullable()) continue;
+ if (!elem.column->is_nullable()) {
+ continue;
+ }
+ bool is_const = is_column_const(*elem.column);
/// Const Nullable that are NULL.
- if (elem.column->only_null())
+ if (is_const && assert_cast<const ColumnConst*>(elem.column.get())->only_null())
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (is_const && assert_cast<const ColumnConst*>(elem.column.get())->only_null()) {
```
be/src/vec/functions/function.cpp:58:
```diff
- Null());
+ Null());
+ }
```
##########
be/src/vec/functions/function.cpp:
##########
@@ -37,48 +37,58 @@
ColumnPtr wrap_in_nullable(const ColumnPtr& src, const Block& block, const ColumnNumbers& args,
size_t result, size_t input_rows_count) {
ColumnPtr result_null_map_column;
-
/// If result is already nullable.
ColumnPtr src_not_nullable = src;
+ MutableColumnPtr mutable_result_null_map_column;
- if (src->only_null())
- return src;
- else if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
src_not_nullable = nullable->get_nested_column_ptr();
result_null_map_column = nullable->get_null_map_column_ptr();
}
for (const auto& arg : args) {
const ColumnWithTypeAndName& elem = block.get_by_position(arg);
- if (!elem.type->is_nullable()) continue;
+ if (!elem.column->is_nullable()) {
+ continue;
+ }
+ bool is_const = is_column_const(*elem.column);
/// Const Nullable that are NULL.
- if (elem.column->only_null())
+ if (is_const && assert_cast<const ColumnConst*>(elem.column.get())->only_null())
return block.get_by_position(result).type->create_column_const(input_rows_count,
Null());
+ if (is_const) {
+ continue;
+ }
- if (is_column_const(*elem.column)) continue;
-
- if (auto* nullable = check_and_get_column<ColumnNullable>(*elem.column)) {
+ if (auto* nullable = assert_cast<const ColumnNullable*>(elem.column.get())) {
const ColumnPtr& null_map_column = nullable->get_null_map_column_ptr();
if (!result_null_map_column) {
- result_null_map_column = null_map_column->clone_resized(null_map_column->size());
+ result_null_map_column = null_map_column->clone_resized(input_rows_count);
} else {
- MutableColumnPtr mutable_result_null_map_column =
- (*std::move(result_null_map_column)).assume_mutable();
+ if (!mutable_result_null_map_column)
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!mutable_result_null_map_column) {
```
be/src/vec/functions/function.cpp:70:
```diff
- (*std::move(result_null_map_column)).assume_mutable();
+ (*std::move(result_null_map_column)).assume_mutable();
+ }
```
--
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1451722515
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1453409057
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1462351463
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
[GitHub] [doris] HappenLee commented on a diff in pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on code in PR #17187:
URL: https://github.com/apache/doris/pull/17187#discussion_r1133374909
##########
be/src/vec/exec/scan/pip_scanner_context.h:
##########
@@ -33,20 +33,68 @@ class PipScannerContext : public vectorized::ScannerContext {
: vectorized::ScannerContext(state, parent, input_tuple_desc, output_tuple_desc,
scanners, limit, max_bytes_in_blocks_queue) {}
- void _update_block_queue_empty() override { _blocks_queue_empty = _blocks_queue.empty(); }
+ Status get_block_from_queue(RuntimeState* state, vectorized::BlockUPtr* block, bool* eos,
+ int id, bool wait = false) override {
+ {
+ std::unique_lock<std::mutex> l(_transfer_lock);
+ if (state->is_cancelled()) {
+ _process_status = Status::Cancelled("cancelled");
+ }
- Status get_block_from_queue(vectorized::BlockUPtr* block, bool* eos,
- bool wait = false) override {
- return vectorized::ScannerContext::get_block_from_queue(block, eos, false);
+ if (!_process_status.ok()) {
+ return _process_status;
+ }
+ }
+
+ {
+ std::unique_lock<std::mutex> l(*_queue_mutexs[id]);
+ if (!_blocks_queues[id].empty()) {
+ *block = std::move(_blocks_queues[id].front());
+ _blocks_queues[id].pop_front();
+ return Status::OK();
+ } else {
+ *eos = _is_finished || _should_stop;
+ }
+ }
+ return Status::OK();
}
// We should make those method lock free.
bool done() override { return _is_finished || _should_stop || _status_error; }
- bool no_schedule() override { return _num_running_scanners == 0 && _num_scheduling_ctx == 0; }
- bool empty_in_queue() override { return _blocks_queue_empty; }
+
+ void append_blocks_to_queue(std::vector<vectorized::BlockUPtr>& blocks) override {
+ const int queue_size = _queue_mutexs.size();
+ const int block_size = blocks.size();
+ for (int i = 0; i < queue_size && i < block_size; ++i) {
+ int queue = _next_queue_to_feed;
+ {
+ std::lock_guard<std::mutex> l(*_queue_mutexs[queue]);
+ for (int j = i; j < block_size; j += queue_size) {
+ _blocks_queues[queue].emplace_back(std::move(blocks[j]));
+ }
+ }
+ _next_queue_to_feed = queue + 1 < queue_size ? queue + 1 : 0;
+ }
+ }
+
+ bool empty_in_queue(int id) override {
+ std::unique_lock<std::mutex> l(*_queue_mutexs[id]);
+ return _blocks_queues[id].empty();
+ }
+
+ void set_max_queue_size(int max_queue_size) override {
+ for (int i = 0; i < max_queue_size; ++i) {
+ _blocks_queue_empty.emplace_back(true);
+ _queue_mutexs.emplace_back(new std::mutex);
+ _blocks_queues.emplace_back(std::list<vectorized::BlockUPtr>());
+ }
+ }
private:
- std::atomic_bool _blocks_queue_empty = true;
+ int _next_queue_to_feed = 0;
+ std::vector<bool> _blocks_queue_empty;
Review Comment:
will remove next pr
--
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
[GitHub] [doris] hello-stephen commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1446621822
TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 34.8 seconds
stream load tsv: 478 seconds loaded 74807831229 Bytes, about 149 MB/s
stream load json: 41 seconds loaded 2358488459 Bytes, about 54 MB/s
stream load orc: 69 seconds loaded 1101869774 Bytes, about 15 MB/s
stream load parquet: 28 seconds loaded 861443392 Bytes, about 29 MB/s
https://doris-community-test-1308700295.cos.ap-hongkong.myqcloud.com/tmp/20230227161544_clickbench_pr_105205.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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1453350268
run allbuild
--
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
[GitHub] [doris] Gabriel39 commented on a diff in pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "Gabriel39 (via GitHub)" <gi...@apache.org>.
Gabriel39 commented on code in PR #17187:
URL: https://github.com/apache/doris/pull/17187#discussion_r1133385741
##########
be/src/common/status.h:
##########
@@ -402,6 +404,8 @@ class Status {
bool ok() const { return _code == ErrorCode::OK; }
+ bool is_blocked_by_sc() const { return _code == ErrorCode::PIP_WAIT_FOR_SC; }
Review Comment:
use is<ErrorCode::PIP_WAIT_FOR_SC>() instead of is_blocked_by_sc
--
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
[GitHub] [doris] github-actions[bot] commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1465419062
PR approved by at least one committer and no changes requested.
--
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1453350003
run allbuild
--
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1446513459
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
[GitHub] [doris] github-actions[bot] commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1465419108
PR approved by anyone and no changes requested.
--
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1461800097
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1463381726
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1453350220
run allbuild
--
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
[GitHub] [doris] HappenLee merged pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee merged PR #17187:
URL: https://github.com/apache/doris/pull/17187
--
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
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #17187:
URL: https://github.com/apache/doris/pull/17187#discussion_r1118393210
##########
be/src/runtime/runtime_state.h:
##########
@@ -195,6 +196,10 @@ class RuntimeState {
int64_t load_job_id() const { return _load_job_id; }
+ void set_shared_scan_opt(bool shared_scan_opt) { _shared_scan_opt = shared_scan_opt; }
+
+ bool shared_scan_opt() { return _shared_scan_opt; }
Review Comment:
warning: method 'shared_scan_opt' can be made const [readability-make-member-function-const]
```suggestion
bool shared_scan_opt() const { return _shared_scan_opt; }
```
##########
be/src/vec/exec/scan/pip_scanner_context.h:
##########
@@ -33,20 +33,74 @@ class PipScannerContext : public vectorized::ScannerContext {
: vectorized::ScannerContext(state, parent, input_tuple_desc, output_tuple_desc,
scanners, limit, max_bytes_in_blocks_queue) {}
- void _update_block_queue_empty() override { _blocks_queue_empty = _blocks_queue.empty(); }
+ void _update_block_queue_empty(int id) override {
+ _blocks_queue_empty[id] = _blocks_queues[id].empty();
+ }
+
+ Status get_block_from_queue(RuntimeState* state, vectorized::BlockUPtr* block, bool* eos,
+ int id, bool wait = false) override {
+ {
+ std::unique_lock<std::mutex> l(_transfer_lock);
+ if (state->is_cancelled()) {
Review Comment:
warning: member access into incomplete type 'doris::RuntimeState' [clang-diagnostic-error]
```cpp
if (state->is_cancelled()) {
^
```
**be/src/runtime/memory/mem_tracker_limiter.h:31:** forward declaration of 'doris::RuntimeState'
```cpp
class RuntimeState;
^
```
##########
be/src/vec/exec/scan/vscan_node.cpp:
##########
@@ -117,13 +134,25 @@ Status VScanNode::alloc_resource(RuntimeState* state) {
return Status::OK();
}
- std::list<VScanner*> scanners;
- RETURN_IF_ERROR(_init_scanners(&scanners));
- if (scanners.empty()) {
- _eos = true;
+ if (_is_pipeline_scan) {
+ if (_should_create_scanner) {
+ auto status = _prepare_scanners();
+ if (_shared_scan_opt) {
+ _shared_scanner_controller->set_scanner_context(
+ id(), _num_scanners->value() == 0 ? nullptr : _scanner_ctx);
+ RETURN_IF_ERROR(status);
+ if (!_eos) _scanner_ctx->set_max_queue_size(state->query_parallel_instance_num());
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!_eos) { _scanner_ctx->set_max_queue_size(state->query_parallel_instance_num());
}
```
##########
be/src/vec/functions/function.cpp:
##########
@@ -37,48 +37,53 @@ namespace doris::vectorized {
ColumnPtr wrap_in_nullable(const ColumnPtr& src, const Block& block, const ColumnNumbers& args,
size_t result, size_t input_rows_count) {
ColumnPtr result_null_map_column;
-
/// If result is already nullable.
ColumnPtr src_not_nullable = src;
+ MutableColumnPtr mutable_result_null_map_column;
- if (src->only_null())
- return src;
- else if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
src_not_nullable = nullable->get_nested_column_ptr();
result_null_map_column = nullable->get_null_map_column_ptr();
}
for (const auto& arg : args) {
const ColumnWithTypeAndName& elem = block.get_by_position(arg);
- if (!elem.type->is_nullable()) continue;
+ if (!elem.column->is_nullable()) continue;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!elem.column->is_nullable()) { continue;
}
```
##########
be/src/vec/functions/function.cpp:
##########
@@ -37,48 +37,53 @@
ColumnPtr wrap_in_nullable(const ColumnPtr& src, const Block& block, const ColumnNumbers& args,
size_t result, size_t input_rows_count) {
ColumnPtr result_null_map_column;
-
/// If result is already nullable.
ColumnPtr src_not_nullable = src;
+ MutableColumnPtr mutable_result_null_map_column;
- if (src->only_null())
- return src;
- else if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
src_not_nullable = nullable->get_nested_column_ptr();
result_null_map_column = nullable->get_null_map_column_ptr();
}
for (const auto& arg : args) {
const ColumnWithTypeAndName& elem = block.get_by_position(arg);
- if (!elem.type->is_nullable()) continue;
+ if (!elem.column->is_nullable()) continue;
+ bool is_const = is_column_const(*elem.column);
/// Const Nullable that are NULL.
- if (elem.column->only_null())
+ if (is_const && assert_cast<const ColumnConst*>(elem.column.get())->only_null())
return block.get_by_position(result).type->create_column_const(input_rows_count,
Null());
+ if (is_const) continue;
- if (is_column_const(*elem.column)) continue;
-
- if (auto* nullable = check_and_get_column<ColumnNullable>(*elem.column)) {
+ if (auto* nullable = assert_cast<const ColumnNullable*>(elem.column.get())) {
const ColumnPtr& null_map_column = nullable->get_null_map_column_ptr();
if (!result_null_map_column) {
- result_null_map_column = null_map_column->clone_resized(null_map_column->size());
+ result_null_map_column = null_map_column->clone_resized(input_rows_count);
} else {
- MutableColumnPtr mutable_result_null_map_column =
- (*std::move(result_null_map_column)).assume_mutable();
+ if (!mutable_result_null_map_column)
+ mutable_result_null_map_column =
+ (*std::move(result_null_map_column)).assume_mutable();
NullMap& result_null_map =
assert_cast<ColumnUInt8&>(*mutable_result_null_map_column).get_data();
const NullMap& src_null_map =
assert_cast<const ColumnUInt8&>(*null_map_column).get_data();
VectorizedUtils::update_null_map(result_null_map, src_null_map);
- result_null_map_column = std::move(mutable_result_null_map_column);
}
}
}
- if (!result_null_map_column) return make_nullable(src);
+ if (!result_null_map_column) {
+ if (is_column_const(*src)) {
+ return ColumnConst::create(
+ make_nullable(assert_cast<const ColumnConst&>(*src).get_data_column_ptr(), 0),
Review Comment:
warning: converting integer literal to bool, use bool literal instead [modernize-use-bool-literals]
```suggestion
make_nullable(assert_cast<const ColumnConst&>(*src).get_data_column_ptr(), false),
```
##########
be/src/vec/functions/function.cpp:
##########
@@ -37,48 +37,53 @@
ColumnPtr wrap_in_nullable(const ColumnPtr& src, const Block& block, const ColumnNumbers& args,
size_t result, size_t input_rows_count) {
ColumnPtr result_null_map_column;
-
/// If result is already nullable.
ColumnPtr src_not_nullable = src;
+ MutableColumnPtr mutable_result_null_map_column;
- if (src->only_null())
- return src;
- else if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
src_not_nullable = nullable->get_nested_column_ptr();
result_null_map_column = nullable->get_null_map_column_ptr();
}
for (const auto& arg : args) {
const ColumnWithTypeAndName& elem = block.get_by_position(arg);
- if (!elem.type->is_nullable()) continue;
+ if (!elem.column->is_nullable()) continue;
+ bool is_const = is_column_const(*elem.column);
/// Const Nullable that are NULL.
- if (elem.column->only_null())
+ if (is_const && assert_cast<const ColumnConst*>(elem.column.get())->only_null())
return block.get_by_position(result).type->create_column_const(input_rows_count,
Null());
+ if (is_const) continue;
- if (is_column_const(*elem.column)) continue;
-
- if (auto* nullable = check_and_get_column<ColumnNullable>(*elem.column)) {
+ if (auto* nullable = assert_cast<const ColumnNullable*>(elem.column.get())) {
const ColumnPtr& null_map_column = nullable->get_null_map_column_ptr();
if (!result_null_map_column) {
- result_null_map_column = null_map_column->clone_resized(null_map_column->size());
+ result_null_map_column = null_map_column->clone_resized(input_rows_count);
} else {
- MutableColumnPtr mutable_result_null_map_column =
- (*std::move(result_null_map_column)).assume_mutable();
+ if (!mutable_result_null_map_column)
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!mutable_result_null_map_column) {
```
be/src/vec/functions/function.cpp:66:
```diff
- (*std::move(result_null_map_column)).assume_mutable();
+ (*std::move(result_null_map_column)).assume_mutable();
+ }
```
##########
be/src/vec/exec/scan/vscan_node.cpp:
##########
@@ -117,13 +134,25 @@
return Status::OK();
}
- std::list<VScanner*> scanners;
- RETURN_IF_ERROR(_init_scanners(&scanners));
- if (scanners.empty()) {
- _eos = true;
+ if (_is_pipeline_scan) {
+ if (_should_create_scanner) {
+ auto status = _prepare_scanners();
+ if (_shared_scan_opt) {
+ _shared_scanner_controller->set_scanner_context(
+ id(), _num_scanners->value() == 0 ? nullptr : _scanner_ctx);
+ RETURN_IF_ERROR(status);
+ if (!_eos) _scanner_ctx->set_max_queue_size(state->query_parallel_instance_num());
+ } else {
+ RETURN_IF_ERROR(status);
+ }
+ } else if (_shared_scanner_controller->scanner_context_is_ready(id())) {
+ _scanner_ctx = _shared_scanner_controller->get_scanner_context(id());
+ if (!_scanner_ctx) _eos = true;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!_scanner_ctx) { _eos = true;
}
```
##########
be/src/vec/functions/function.cpp:
##########
@@ -37,48 +37,53 @@
ColumnPtr wrap_in_nullable(const ColumnPtr& src, const Block& block, const ColumnNumbers& args,
size_t result, size_t input_rows_count) {
ColumnPtr result_null_map_column;
-
/// If result is already nullable.
ColumnPtr src_not_nullable = src;
+ MutableColumnPtr mutable_result_null_map_column;
- if (src->only_null())
- return src;
- else if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
src_not_nullable = nullable->get_nested_column_ptr();
result_null_map_column = nullable->get_null_map_column_ptr();
}
for (const auto& arg : args) {
const ColumnWithTypeAndName& elem = block.get_by_position(arg);
- if (!elem.type->is_nullable()) continue;
+ if (!elem.column->is_nullable()) continue;
+ bool is_const = is_column_const(*elem.column);
/// Const Nullable that are NULL.
- if (elem.column->only_null())
+ if (is_const && assert_cast<const ColumnConst*>(elem.column.get())->only_null())
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (is_const && assert_cast<const ColumnConst*>(elem.column.get())->only_null()) {
```
be/src/vec/functions/function.cpp:56:
```diff
- Null());
+ Null());
+ }
```
##########
be/src/vec/functions/function.cpp:
##########
@@ -37,48 +37,53 @@
ColumnPtr wrap_in_nullable(const ColumnPtr& src, const Block& block, const ColumnNumbers& args,
size_t result, size_t input_rows_count) {
ColumnPtr result_null_map_column;
-
/// If result is already nullable.
ColumnPtr src_not_nullable = src;
+ MutableColumnPtr mutable_result_null_map_column;
- if (src->only_null())
- return src;
- else if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*src)) {
src_not_nullable = nullable->get_nested_column_ptr();
result_null_map_column = nullable->get_null_map_column_ptr();
}
for (const auto& arg : args) {
const ColumnWithTypeAndName& elem = block.get_by_position(arg);
- if (!elem.type->is_nullable()) continue;
+ if (!elem.column->is_nullable()) continue;
+ bool is_const = is_column_const(*elem.column);
/// Const Nullable that are NULL.
- if (elem.column->only_null())
+ if (is_const && assert_cast<const ColumnConst*>(elem.column.get())->only_null())
return block.get_by_position(result).type->create_column_const(input_rows_count,
Null());
+ if (is_const) continue;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (is_const) { continue;
}
```
--
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1447987224
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1453408904
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1458444503
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1455983927
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1446513394
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
[GitHub] [doris] HappenLee commented on pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on PR #17187:
URL: https://github.com/apache/doris/pull/17187#issuecomment-1450362852
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
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #17187:
URL: https://github.com/apache/doris/pull/17187#discussion_r1130831673
##########
be/src/pipeline/pipeline_task.cpp:
##########
@@ -135,6 +135,9 @@ Status PipelineTask::execute(bool* eos) {
if (st.is<ErrorCode::PIP_WAIT_FOR_RF>()) {
set_state(PipelineTaskState::BLOCKED_FOR_RF);
return Status::OK();
+ } else if (st.is_blocked_by_sc()) {
+ set_state(BLOCKED_FOR_SOURCE);
Review Comment:
warning: use of undeclared identifier 'BLOCKED_FOR_SOURCE' [clang-diagnostic-error]
```cpp
set_state(BLOCKED_FOR_SOURCE);
^
```
--
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
[GitHub] [doris] wangbo commented on a diff in pull request #17187: [Pipeline](shared_scan_opt) Support shared scan opt in pipeline exec engine
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #17187:
URL: https://github.com/apache/doris/pull/17187#discussion_r1133061779
##########
be/src/vec/exec/scan/pip_scanner_context.h:
##########
@@ -33,20 +33,68 @@ class PipScannerContext : public vectorized::ScannerContext {
: vectorized::ScannerContext(state, parent, input_tuple_desc, output_tuple_desc,
scanners, limit, max_bytes_in_blocks_queue) {}
- void _update_block_queue_empty() override { _blocks_queue_empty = _blocks_queue.empty(); }
+ Status get_block_from_queue(RuntimeState* state, vectorized::BlockUPtr* block, bool* eos,
+ int id, bool wait = false) override {
+ {
+ std::unique_lock<std::mutex> l(_transfer_lock);
+ if (state->is_cancelled()) {
+ _process_status = Status::Cancelled("cancelled");
+ }
- Status get_block_from_queue(vectorized::BlockUPtr* block, bool* eos,
- bool wait = false) override {
- return vectorized::ScannerContext::get_block_from_queue(block, eos, false);
+ if (!_process_status.ok()) {
+ return _process_status;
+ }
+ }
+
+ {
+ std::unique_lock<std::mutex> l(*_queue_mutexs[id]);
+ if (!_blocks_queues[id].empty()) {
+ *block = std::move(_blocks_queues[id].front());
+ _blocks_queues[id].pop_front();
+ return Status::OK();
+ } else {
+ *eos = _is_finished || _should_stop;
+ }
+ }
+ return Status::OK();
}
// We should make those method lock free.
bool done() override { return _is_finished || _should_stop || _status_error; }
- bool no_schedule() override { return _num_running_scanners == 0 && _num_scheduling_ctx == 0; }
- bool empty_in_queue() override { return _blocks_queue_empty; }
+
+ void append_blocks_to_queue(std::vector<vectorized::BlockUPtr>& blocks) override {
+ const int queue_size = _queue_mutexs.size();
+ const int block_size = blocks.size();
+ for (int i = 0; i < queue_size && i < block_size; ++i) {
+ int queue = _next_queue_to_feed;
+ {
+ std::lock_guard<std::mutex> l(*_queue_mutexs[queue]);
+ for (int j = i; j < block_size; j += queue_size) {
+ _blocks_queues[queue].emplace_back(std::move(blocks[j]));
+ }
+ }
+ _next_queue_to_feed = queue + 1 < queue_size ? queue + 1 : 0;
+ }
+ }
+
+ bool empty_in_queue(int id) override {
+ std::unique_lock<std::mutex> l(*_queue_mutexs[id]);
+ return _blocks_queues[id].empty();
+ }
+
+ void set_max_queue_size(int max_queue_size) override {
+ for (int i = 0; i < max_queue_size; ++i) {
+ _blocks_queue_empty.emplace_back(true);
+ _queue_mutexs.emplace_back(new std::mutex);
+ _blocks_queues.emplace_back(std::list<vectorized::BlockUPtr>());
+ }
+ }
private:
- std::atomic_bool _blocks_queue_empty = true;
+ int _next_queue_to_feed = 0;
+ std::vector<bool> _blocks_queue_empty;
Review Comment:
What is ```_blocks_queue_empty``` used for ?
It seems that no place using it?
--
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