[improve](load) limit delta writer flush task parallelism (#28883)

apache · Dec 22, 2023 · f781f0c · f781f0c
1 parent b1c5747
commit f781f0c
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 0 deletions.
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
@@ -592,6 +592,8 @@ DEFINE_mDouble(memtable_insert_memory_ratio, "1.4");
 DEFINE_mInt64(write_buffer_size, "209715200");
 // max buffer size used in memtable for the aggregated table, default 400MB
 DEFINE_mInt64(write_buffer_size_for_agg, "419430400");
+// max parallel flush task per memtable writer
+DEFINE_mInt32(memtable_flush_running_count_limit, "5");
 
 DEFINE_Int32(load_process_max_memory_limit_percent, "50"); // 50%
 

diff --git a/be/src/common/config.h b/be/src/common/config.h
@@ -651,6 +651,8 @@ DECLARE_mDouble(memtable_insert_memory_ratio);
 DECLARE_mInt64(write_buffer_size);
 // max buffer size used in memtable for the aggregated table, default 400MB
 DECLARE_mInt64(write_buffer_size_for_agg);
+// max parallel flush task per memtable writer
+DECLARE_mInt32(memtable_flush_running_count_limit);
 
 DECLARE_Int32(load_process_max_memory_limit_percent); // 50%
 

diff --git a/be/src/olap/delta_writer.cpp b/be/src/olap/delta_writer.cpp
@@ -126,6 +126,10 @@ Status BaseDeltaWriter::write(const vectorized::Block* block, const std::vector<
  if (!_is_init && !_is_cancelled) {
  RETURN_IF_ERROR(init());
  }
+ while (_memtable_writer->get_flush_token_stats().flush_running_count >=
+ config::memtable_flush_running_count_limit) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
  return _memtable_writer->write(block, row_idxs, is_append);
 }
 Status BaseDeltaWriter::wait_flush() {

diff --git a/be/src/olap/delta_writer_v2.cpp b/be/src/olap/delta_writer_v2.cpp
@@ -37,6 +37,7 @@
 #include "gutil/strings/numbers.h"
 #include "io/fs/file_writer.h" // IWYU pragma: keep
 #include "olap/data_dir.h"
+#include "olap/memtable_flush_executor.h"
 #include "olap/olap_define.h"
 #include "olap/rowset/beta_rowset.h"
 #include "olap/rowset/beta_rowset_writer_v2.h"
@@ -152,6 +153,10 @@ Status DeltaWriterV2::write(const vectorized::Block* block, const std::vector<ui
  if (!_is_init && !_is_cancelled) {
  RETURN_IF_ERROR(init());
  }
+ while (_memtable_writer->get_flush_token_stats().flush_running_count >=
+ config::memtable_flush_running_count_limit) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
  SCOPED_RAW_TIMER(&_write_memtable_time);
  return _memtable_writer->write(block, row_idxs, is_append);
 }