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

Fix memory_scratch_sink_test in debug mode (#1906)

Browse Source
This commit is contained in:
kangkaisen
2019-09-28 10:33:24 +08:00
committed by ZHAO Chun
parent 1131f53420
commit d3a445ee09
3 changed files with 21 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
be/src/runtime/runtime_state.cpp
View File

@ -264,6 +264,11 @@ Status RuntimeState::init_mem_trackers(const TUniqueId& query_id) {
return Status::OK();
}
Status RuntimeState::init_instance_mem_tracker() {
_instance_mem_tracker.reset(new MemTracker(-1));
return Status::OK();
}
Status RuntimeState::init_buffer_poolstate() {
ExecEnv* exec_env = ExecEnv::GetInstance();
int64_t mem_limit = _query_mem_tracker->lowest_limit();

3
be/src/runtime/runtime_state.h
View File

@ -93,6 +93,9 @@ public:
// This function also initializes a user function mem tracker (in the fourth level).
Status init_mem_trackers(const TUniqueId& query_id);
// for ut only
Status init_instance_mem_tracker();
/// Called from Init() to set up buffer reservations and the file group.
Status init_buffer_poolstate();

22
be/test/runtime/memory_scratch_sink_test.cpp
View File

@ -37,17 +37,14 @@
#include "runtime/tuple_row.h"
#include "util/blocking_queue.hpp"
#include "util/logging.h"
#include "testutil/desc_tbl_builder.h"
namespace doris {
class MemoryScratchSinkTest : public testing::Test {
public:
MemoryScratchSinkTest() {
// _exec_env = doris::ExecEnv::GetInstance();
// std::vector<doris::StorePath> paths;
// doris::parse_conf_store_paths(doris::config::storage_root_path, &paths);
// doris::ExecEnv::init(_exec_env, paths);
// all below is just only for test MemoryScratchSink
// all below is just only for test MemoryScratchSink
ResultQueueMgr* result_queue_mgr = new ResultQueueMgr();
ThreadResourceMgr* thread_mgr = new ThreadResourceMgr();
_exec_env._result_queue_mgr = result_queue_mgr;
@ -58,6 +55,7 @@ public:
query_id.lo = 10;
query_id.hi = 100;
_runtime_state = new RuntimeState(query_id, query_options, TQueryGlobals(), &_exec_env);
_runtime_state->init_instance_mem_tracker();
{
TExpr expr;
{
@ -75,6 +73,12 @@ public:
}
_mem_tracker = new MemTracker(-1, "MemoryScratchSinkTest", _runtime_state->instance_mem_tracker());
// init _row_desc
vector<bool> nullable_tuples(1, false);
vector<TTupleId> tuple_ids(1, static_cast<TTupleId>(0));
DescriptorTblBuilder int_builder(&_pool);
int_builder.declare_tuple() << TYPE_INT;
_row_desc = _pool.add(new RowDescriptor(*int_builder.build(), tuple_ids, nullable_tuples));
}
virtual ~MemoryScratchSinkTest() {
delete _runtime_state;
@ -86,22 +90,22 @@ protected:
}
private:
ObjectPool _pool;
ExecEnv _exec_env;
std::vector<TExpr> _exprs;
RuntimeState* _runtime_state;
RowDescriptor _row_desc;
RowDescriptor* _row_desc;
TMemoryScratchSink _tsink;
MemTracker *_mem_tracker;
};
TEST_F(MemoryScratchSinkTest, work_flow_normal) {
MemoryScratchSink sink(_row_desc, _exprs, _tsink);
MemoryScratchSink sink(*_row_desc, _exprs, _tsink);
TDataSink data_sink;
data_sink.memory_scratch_sink = _tsink;
ASSERT_TRUE(sink.init(data_sink).ok());
ASSERT_TRUE(sink.prepare(_runtime_state).ok());
ASSERT_TRUE(sink.prepare(_runtime_state).ok());
RowBatch row_batch(_row_desc, 1024, _mem_tracker);
RowBatch row_batch(*_row_desc, 1024, _mem_tracker);
row_batch.add_row();
row_batch.commit_last_row();
ASSERT_TRUE(sink.send(_runtime_state, &row_batch).ok());