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

Optimize the load performance for large file (#1798)

Browse Source 
The current load process is:

Tablet Sink -> Tablet Channel Mgr -> Tablets Channel -> Delta Writer -> MemTable -> Flush to disk

In the path of Tablets Channel -> DeltaWriter -> MemTable -> Flush to disk, the following operations are performed:

Insert tuple into different memtables according to tablet ID
When the memtable size reaches the threshold, it is written to disk.
The above operations are equivalent to single thread execution for a single load task.
In fact, the insertion of memtable and the flush of memtable can be executed synchronously.
Perform these operation in single thread prevents the insertion of memtable from being delayed due to slow disk writing.

In the new implementation, I added a MemTableFlushExecutor class with a set of flush queues and corresponding worker threads.
By default, each data directory uses two worker threads for flush, which can be modified by the parameter flush_thread_num_per_store of BE.
DeltaWriter will push the full memtable to MemTableFlushExecutor for flush operation and generate a new memtable for receiving new data.

This design can improve the performance of load large files.
In single host testing, the time to load a 1GB text file is reduced from 48 seconds to 29 seconds.
This commit is contained in:
Mingyu Chen
2019-09-25 13:49:32 +08:00
committed by GitHub
parent dd02382abd
commit c643cbd30c
27 changed files with 1204 additions and 413 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
be/test/runtime/user_function_cache_test.cpp
View File

@ -86,7 +86,7 @@ public:
UserFunctionCacheTest() { }
virtual ~UserFunctionCacheTest() { }
static void SetUpTestCase() {
s_server = new EvHttpServer(29999);
s_server = new EvHttpServer(29987);
s_server->register_handler(GET, "/{FILE}", &s_test_handler);
s_server->start();
@ -130,7 +130,7 @@ TEST_F(UserFunctionCacheTest, download_normal) {
// get my_add
st = cache.get_function_ptr(1,
"_Z6my_addv",
"http://127.0.0.1:29999/my_add.so",
"http://127.0.0.1:29987/my_add.so",
my_add_md5sum, &fn_ptr, &entry);
ASSERT_TRUE(st.ok());
ASSERT_TRUE(k_is_downloaded);
@ -140,7 +140,7 @@ TEST_F(UserFunctionCacheTest, download_normal) {
// get my_del
st = cache.get_function_ptr(1,
"_Z6my_delv",
"http://127.0.0.1:29999/my_add.so",
"http://127.0.0.1:29987/my_add.so",
my_add_md5sum, &fn_ptr, &entry);
ASSERT_TRUE(st.ok());
ASSERT_NE(nullptr, fn_ptr);
@ -149,7 +149,7 @@ TEST_F(UserFunctionCacheTest, download_normal) {
// get my_mul
st = cache.get_function_ptr(1,
"_Z6my_mulv",
"http://127.0.0.1:29999/my_add.so",
"http://127.0.0.1:29987/my_add.so",
my_add_md5sum, &fn_ptr, &entry);
ASSERT_FALSE(st.ok());
@ -165,7 +165,7 @@ TEST_F(UserFunctionCacheTest, load_normal) {
UserFunctionCacheEntry* entry = nullptr;
st = cache.get_function_ptr(1,
"_Z6my_addv",
"http://127.0.0.1:29999/my_add.so",
"http://127.0.0.1:29987/my_add.so",
my_add_md5sum, &fn_ptr, &entry);
ASSERT_TRUE(st.ok());
ASSERT_FALSE(k_is_downloaded);
@ -183,7 +183,7 @@ TEST_F(UserFunctionCacheTest, download_fail) {
UserFunctionCacheEntry* entry = nullptr;
st = cache.get_function_ptr(2,
"_Z6my_delv",
"http://127.0.0.1:29999/my_del.so",
"http://127.0.0.1:29987/my_del.so",
my_add_md5sum, &fn_ptr, &entry);
ASSERT_FALSE(st.ok());
}
@ -199,7 +199,7 @@ TEST_F(UserFunctionCacheTest, md5_fail) {
UserFunctionCacheEntry* entry = nullptr;
st = cache.get_function_ptr(1,
"_Z6my_addv",
"http://127.0.0.1:29999/my_add.so",
"http://127.0.0.1:29987/my_add.so",
"1234", &fn_ptr, &entry);
ASSERT_FALSE(st.ok());
}
@ -218,7 +218,7 @@ TEST_F(UserFunctionCacheTest, bad_so) {
UserFunctionCacheEntry* entry = nullptr;
st = cache.get_function_ptr(2,
"_Z6my_addv",
"http://127.0.0.1:29999/my_add.so",
"http://127.0.0.1:29987/my_add.so",
"abc", &fn_ptr, &entry);
ASSERT_FALSE(st.ok());
}