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

Fix output results may incorrect when using intersect and except statements (#3228)

Browse Source 
output results may  incorrect  when using intersect and except statements
This commit is contained in:
yangzhg
2020-04-01 20:58:43 +08:00
committed by GitHub
parent 34993a69a8
commit 63cee94c5c
7 changed files with 256 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
be/src/exec/except_node.cpp
View File

@ -19,19 +19,15 @@
#include "exec/hash_table.hpp"
#include "exprs/expr.h"
#include "runtime/raw_value.h"
#include "runtime/row_batch.h"
#include "runtime/runtime_state.h"
namespace doris {
ExceptNode::ExceptNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs)
: ExecNode(pool, tnode, descs) {}
: SetOperationNode(pool, tnode, descs, tnode.except_node.tuple_id) {}
Status ExceptNode::init(const TPlanNode& tnode, RuntimeState* state) {
RETURN_IF_ERROR(ExecNode::init(tnode, state));
DCHECK(tnode.__isset.except_node);
DCHECK_EQ(_conjunct_ctxs.size(), 0);
DCHECK_GE(_children.size(), 2);
RETURN_IF_ERROR(SetOperationNode::init(tnode, state));
// Create result_expr_ctx_lists_ from thrift exprs.
auto& result_texpr_lists = tnode.except_node.result_expr_lists;
for (auto& texprs : result_texpr_lists) {
@ -42,51 +38,6 @@ Status ExceptNode::init(const TPlanNode& tnode, RuntimeState* state) {
return Status::OK();
}
Status ExceptNode::prepare(RuntimeState* state) {
RETURN_IF_ERROR(ExecNode::prepare(state));
_build_timer = ADD_TIMER(runtime_profile(), "BuildTime");
_probe_timer = ADD_TIMER(runtime_profile(), "ProbeTime");
_build_pool.reset(new MemPool(mem_tracker()));
SCOPED_TIMER(_runtime_profile->total_time_counter());
for (size_t i = 0; i < _child_expr_lists.size(); ++i) {
RETURN_IF_ERROR(Expr::prepare(_child_expr_lists[i], state, child(i)->row_desc(),
expr_mem_tracker()));
}
_build_tuple_size = child(0)->row_desc().tuple_descriptors().size();
_build_tuple_row_size = _build_tuple_size * sizeof(Tuple*);
_build_tuple_idx.reserve(_build_tuple_size);
for (int i = 0; i < _build_tuple_size; ++i) {
TupleDescriptor* build_tuple_desc = child(0)->row_desc().tuple_descriptors()[i];
_build_tuple_idx.push_back(_row_descriptor.get_tuple_idx(build_tuple_desc->id()));
}
_find_nulls = std::vector<bool>(_build_tuple_size, true);
return Status::OK();
}
Status ExceptNode::close(RuntimeState* state) {
if (is_closed()) {
return Status::OK();
}
for (auto& exprs : _child_expr_lists) {
Expr::close(exprs, state);
}
RETURN_IF_ERROR(exec_debug_action(TExecNodePhase::CLOSE));
// Must reset _probe_batch in close() to release resources
_probe_batch.reset(NULL);
if (_memory_used_counter != NULL && _hash_tbl.get() != NULL) {
COUNTER_UPDATE(_memory_used_counter, _build_pool->peak_allocated_bytes());
COUNTER_UPDATE(_memory_used_counter, _hash_tbl->byte_size());
}
if (_hash_tbl.get() != NULL) {
_hash_tbl->close();
}
if (_build_pool.get() != NULL) {
_build_pool->free_all();
}
return ExecNode::close(state);
}
Status ExceptNode::open(RuntimeState* state) {
RETURN_IF_ERROR(ExecNode::open(state));
RETURN_IF_ERROR(exec_debug_action(TExecNodePhase::OPEN));
@ -110,8 +61,10 @@ Status ExceptNode::open(RuntimeState* state) {
// take ownership of tuple data of build_batch
_build_pool->acquire_data(build_batch.tuple_data_pool(), false);
RETURN_IF_LIMIT_EXCEEDED(state, " Except, while constructing the hash table.");
// build hash table and remvoe duplicate items
// build hash table and remove duplicate items
for (int i = 0; i < build_batch.num_rows(); ++i) {
VLOG_ROW << "build row: "
<< get_row_output_string(build_batch.get_row(i), child(0)->row_desc());
_hash_tbl->insert_unique(build_batch.get_row(i));
}
VLOG_ROW << "hash table content: " << _hash_tbl->debug_string(true, &child(0)->row_desc());
@ -137,6 +90,9 @@ Status ExceptNode::open(RuntimeState* state) {
if (previous_hash != _hash_tbl_iterator.get_hash()) {
previous_hash = _hash_tbl_iterator.get_hash();
if (!_hash_tbl_iterator.matched()) {
VLOG_ROW << "rebuild row: "
<< get_row_output_string(_hash_tbl_iterator.get_row(),
child(0)->row_desc());
temp_tbl->insert(_hash_tbl_iterator.get_row());
}
}
@ -155,6 +111,8 @@ Status ExceptNode::open(RuntimeState* state) {
RETURN_IF_ERROR(child(i)->get_next(state, _probe_batch.get(), &eos));
RETURN_IF_LIMIT_EXCEEDED(state, " Except , while probing the hash table.");
for (int j = 0; j < _probe_batch->num_rows(); ++j) {
VLOG_ROW << "probe row: "
<< get_row_output_string(_probe_batch->get_row(j), child(i)->row_desc());
_hash_tbl_iterator = _hash_tbl->find(_probe_batch->get_row(j));
if (_hash_tbl_iterator != _hash_tbl->end()) {
_hash_tbl_iterator.set_matched();
@ -179,24 +137,21 @@ Status ExceptNode::get_next(RuntimeState* state, RowBatch* out_batch, bool* eos)
if (reached_limit()) {
return Status::OK();
}
uint32_t previous_hash = -1;
TupleRow* previous_row = nullptr;
int64_t tuple_buf_size;
uint8_t* tuple_buf;
RETURN_IF_ERROR(
out_batch->resize_and_allocate_tuple_buffer(state, &tuple_buf_size, &tuple_buf));
memset(tuple_buf, 0, tuple_buf_size);
while (_hash_tbl_iterator.has_next()) {
VLOG_ROW << "find row: "
<< get_row_output_string(_hash_tbl_iterator.get_row(), child(0)->row_desc())
<< " matched: " << _hash_tbl_iterator.matched();
if (!_hash_tbl_iterator.matched()) {
if (previous_hash != _hash_tbl_iterator.get_hash() ||
!equals(previous_row, _hash_tbl_iterator.get_row())) {
int row_idx = out_batch->add_row();
TupleRow* out_row = out_batch->get_row(row_idx);
uint8_t* out_ptr = reinterpret_cast<uint8_t*>(out_row);
memcpy(out_ptr, _hash_tbl_iterator.get_row(), _build_tuple_row_size);
out_batch->commit_last_row();
++_num_rows_returned;
}
create_output_row(_hash_tbl_iterator.get_row(), out_batch, tuple_buf);
tuple_buf += _tuple_desc->byte_size();
++_num_rows_returned;
}
previous_hash = _hash_tbl_iterator.get_hash();
previous_row = _hash_tbl_iterator.get_row();
_hash_tbl_iterator.next<false>();
*eos = !_hash_tbl_iterator.has_next() || reached_limit();
if (out_batch->is_full() || out_batch->at_resource_limit() || *eos) {
return Status::OK();
@ -205,24 +160,4 @@ Status ExceptNode::get_next(RuntimeState* state, RowBatch* out_batch, bool* eos)
return Status::OK();
}
bool ExceptNode::equals(TupleRow* row, TupleRow* other) {
DCHECK(!(row == nullptr && other == nullptr));
if (row == nullptr || other == nullptr) {
return false;
}
for (int i = 0; i < _child_expr_lists[0].size(); ++i) {
void* val_row = _child_expr_lists[0][i]->get_value(row);
void* val_other = _child_expr_lists[0][i]->get_value(other);
if (val_row == nullptr && val_other == nullptr) {
continue;
} else if (val_row == nullptr || val_other == nullptr) {
return false;
} else if (!RawValue::eq(val_row, val_other, _child_expr_lists[0][i]->root()->type())) {
return false;
}
}
return true;
}
} // namespace doris