database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[Bug] Fix DCHECK failed in runtime filter and mutable block (#8720)

Browse Source 
Co-authored-by: lihaopeng <lihaopeng@baidu.com>
This commit is contained in:
HappenLee
2022-03-31 11:13:05 +08:00
committed by GitHub
parent 3a8ca80eab
commit bf73ab69f2
2 changed files with 10 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
be/src/exprs/runtime_filter.cpp
View File

@ -452,7 +452,8 @@ public:
DCHECK(state != nullptr);
DCHECK(container != nullptr);
DCHECK(_pool != nullptr);
DCHECK(prob_expr->root()->type().type == _column_return_type);
DCHECK(prob_expr->root()->type().type == _column_return_type ||
(is_string_type(prob_expr->root()->type().type) && is_string_type(_column_return_type)));
auto real_filter_type = get_real_type();
switch (real_filter_type) {

13
be/src/vec/sink/vdata_stream_sender.cpp
View File

@ -425,9 +425,9 @@ Status VDataStreamSender::send(RuntimeState* state, Block* block) {
}
_current_channel_idx = (_current_channel_idx + 1) % _channels.size();
} else if (_part_type == TPartitionType::HASH_PARTITIONED) {
int num_channels = _channels.size();
// will only copy schema
// we don't want send temp columns
auto column_to_keep = block->columns();
int result_size = _partition_expr_ctxs.size();
int result[result_size];
@ -451,12 +451,14 @@ Status VDataStreamSender::send(RuntimeState* state, Block* block) {
hash_vals[i] = siphashs[i].get64();
}
RETURN_IF_ERROR(channel_add_rows(_channels, num_channels, hash_vals, rows, block));
Block::erase_useless_column(block, column_to_keep);
RETURN_IF_ERROR(channel_add_rows(_channels, _channels.size(), hash_vals, rows, block));
} else if (_part_type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED) {
// will only copy schema
// we don't want send temp columns
auto column_to_keep = block->columns();
// 1. calculate hash
// 2. dispatch rows to channel
int num_channels = _channel_shared_ptrs.size();
int result_size = _partition_expr_ctxs.size();
int result[result_size];
RETURN_IF_ERROR(get_partition_column_result(block, result));
@ -484,8 +486,9 @@ Status VDataStreamSender::send(RuntimeState* state, Block* block) {
}
}
Block::erase_useless_column(block, column_to_keep);
RETURN_IF_ERROR(
channel_add_rows(_channel_shared_ptrs, num_channels, hash_vals, rows, block));
channel_add_rows(_channel_shared_ptrs, _channel_shared_ptrs.size(), hash_vals, rows, block));
} else {
// Range partition
// 1. calculate range