Merge branch 'master' into add-report-tablet-metrics

Author: xy720

be/src/common/config.cpp

@@ -1091,6 +1091,7 @@ DEFINE_Bool(enable_graceful_exit_check, "false");
 DEFINE_Bool(enable_debug_points, "false");
 
 DEFINE_Int32(pipeline_executor_size, "0");
+DEFINE_Int32(blocking_pipeline_executor_size, "0");
 DEFINE_Bool(enable_workload_group_for_scan, "false");
 DEFINE_mInt64(workload_group_scan_task_wait_timeout_ms, "10000");
 
@@ -1626,6 +1627,8 @@ DEFINE_Validator(binary_plain_encoding_default_impl, [](const std::string& confi
     return config == "v1" || config == "v2";
 });
 
+DEFINE_mBool(integer_type_default_use_plain_encoding, "true");
+
 // clang-format off
 #ifdef BE_TEST
 // test s3

be/src/common/config.h

@@ -1122,6 +1122,7 @@ DECLARE_Bool(enable_graceful_exit_check);
 DECLARE_Bool(enable_debug_points);
 
 DECLARE_Int32(pipeline_executor_size);
+DECLARE_Int32(blocking_pipeline_executor_size);
 
 // block file cache
 DECLARE_Bool(enable_file_cache);
@@ -1666,6 +1667,8 @@ DECLARE_mString(aws_credentials_provider_version);
 
 DECLARE_mString(binary_plain_encoding_default_impl);
 
+DECLARE_mBool(integer_type_default_use_plain_encoding);
+
 #ifdef BE_TEST
 // test s3
 DECLARE_String(test_s3_resource);

be/src/olap/memtable.cpp

@@ -45,6 +45,7 @@ namespace doris {
 #include "common/compile_check_begin.h"
 
 bvar::Adder<int64_t> g_memtable_cnt("memtable_cnt");
+bvar::Adder<uint64_t> g_flush_cuz_memtable_full("flush_cuz_memtable_full");
 
 using namespace ErrorCode;
 
@@ -665,7 +666,12 @@ bool MemTable::need_flush() const {
         max_size = max_size * update_columns_size / _tablet_schema->num_columns();
         max_size = max_size > min_buffer_size ? max_size : min_buffer_size;
     }
-    return memory_usage() >= max_size;
+
+    if (memory_usage() >= max_size) {
+        g_flush_cuz_memtable_full << 1;
+        return true;
+    }
+    return false;
 }
 
 int64_t MemTable::_adaptive_write_buffer_size() const {

be/src/olap/memtable_memory_limiter.cpp

@@ -40,6 +40,8 @@ bvar::Status<int64_t> g_memtable_flush_memory("mm_limiter_mem_flush", 0);
 bvar::Status<int64_t> g_memtable_load_memory("mm_limiter_mem_load", 0);
 bvar::Status<int64_t> g_load_hard_mem_limit("mm_limiter_limit_hard", 0);
 bvar::Status<int64_t> g_load_soft_mem_limit("mm_limiter_limit_soft", 0);
+bvar::Adder<uint64_t> g_flush_cuz_load_mem_exceed_hard_limit("flush_cuz_hard_limit");
+bvar::Adder<uint64_t> g_flush_cuz_sys_mem_exceed_soft_limit("flush_cuz_soft_limit");
 bvar::Adder<int> g_memtable_memory_limit_flush_memtable_count("mm_limiter_flush_memtable_count");
 bvar::LatencyRecorder g_memtable_memory_limit_flush_size_bytes("mm_limiter_flush_size_bytes");
 
@@ -115,7 +117,7 @@ int64_t MemTableMemoryLimiter::_need_flush() {
     int64_t limit1 = _mem_tracker->consumption() - _load_soft_mem_limit;
     int64_t limit2 = _sys_avail_mem_less_than_warning_water_mark();
     int64_t limit3 = _process_used_mem_more_than_soft_mem_limit();
-    int64_t need_flush = std::max(limit1, std::max(limit2, limit3));
+    int64_t need_flush = std::max({limit1, limit2, limit3});
     return need_flush - _queue_mem_usage - _flush_mem_usage;
 }
 
@@ -166,6 +168,13 @@ void MemTableMemoryLimiter::handle_memtable_flush(std::function<bool()> cancel_c
                                        ->process_memory_detail_str();
                 LOG_LONG_STRING(INFO, log_str);
             }
+            if (limit == Limit::HARD) {
+                g_flush_cuz_load_mem_exceed_hard_limit << 1;
+            } else if (limit == Limit::SOFT) {
+                g_flush_cuz_sys_mem_exceed_soft_limit << 1;
+            } else {
+                // will not reach here
+            }
             _flush_active_memtables(need_flush);
         }
     } while (_hard_limit_reached() && !_load_usage_low());

be/src/olap/memtable_writer.cpp

@@ -46,6 +46,8 @@
 #include "vec/core/block.h"
 
 namespace doris {
+bvar::Adder<uint64_t> g_flush_cuz_rowscnt_oveflow("flush_cuz_rowscnt_oveflow");
+
 using namespace ErrorCode;
 
 MemTableWriter::MemTableWriter(const WriteRequest& req) : _req(req) {}
@@ -108,6 +110,7 @@ Status MemTableWriter::write(const vectorized::Block* block,
     DBUG_EXECUTE_IF("MemTableWriter.too_many_raws",
                     { raw_rows = std::numeric_limits<int32_t>::max(); });
     if (raw_rows + row_idxs.size() > std::numeric_limits<int32_t>::max()) {
+        g_flush_cuz_rowscnt_oveflow << 1;
         RETURN_IF_ERROR(_flush_memtable());
     }
 

be/src/olap/rowset/segment_v2/encoding_info.cpp

@@ -19,6 +19,7 @@
 
 #include <gen_cpp/segment_v2.pb.h>
 
+#include <array>
 #include <iterator>
 #include <type_traits>
 #include <unordered_map>
@@ -317,6 +318,80 @@ EncodingInfoResolver::~EncodingInfoResolver() {
     _encoding_map.clear();
 }
 
+namespace {
+bool is_integer_type(FieldType type) {
+    return type == FieldType::OLAP_FIELD_TYPE_TINYINT ||
+           type == FieldType::OLAP_FIELD_TYPE_SMALLINT || type == FieldType::OLAP_FIELD_TYPE_INT ||
+           type == FieldType::OLAP_FIELD_TYPE_BIGINT || type == FieldType::OLAP_FIELD_TYPE_LARGEINT;
+}
+
+bool is_binary_type(FieldType type) {
+    return type == FieldType::OLAP_FIELD_TYPE_CHAR || type == FieldType::OLAP_FIELD_TYPE_VARCHAR ||
+           type == FieldType::OLAP_FIELD_TYPE_STRING || type == FieldType::OLAP_FIELD_TYPE_JSONB ||
+           type == FieldType::OLAP_FIELD_TYPE_VARIANT || type == FieldType::OLAP_FIELD_TYPE_HLL ||
+           type == FieldType::OLAP_FIELD_TYPE_BITMAP ||
+           type == FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE ||
+           type == FieldType::OLAP_FIELD_TYPE_AGG_STATE;
+}
+} // namespace
+
+EncodingTypePB EncodingInfoResolver::get_default_encoding(FieldType type,
+                                                          bool optimize_value_seek) const {
+    // Predicate for default encoding transformation
+    // Parameters: (type, current_default_encoding, optimize_value_seek)
+    // Returns: true if the transformation should be applied
+    using Predicate = std::function<bool(FieldType, EncodingTypePB, bool)>;
+
+    // Hook for transforming default encoding: predicate -> target encoding
+    struct EncodingTransform {
+        Predicate predicate;
+        EncodingTypePB target_encoding;
+    };
+
+    // Static array of hooks for default encoding transformations
+    static const std::vector<EncodingTransform> hooks = {
+            // Hook 1: Binary types - PLAIN_ENCODING -> PLAIN_ENCODING_V2
+            // Applies when: type is binary, encoding is PLAIN_ENCODING, and config enables v2
+            EncodingTransform {
+                    .predicate =
+                            [](FieldType type, EncodingTypePB encoding, bool optimize_value_seek) {
+                                return encoding == PLAIN_ENCODING && is_binary_type(type) &&
+                                       config::binary_plain_encoding_default_impl == "v2";
+                            },
+                    .target_encoding = PLAIN_ENCODING_V2},
+
+            // Hook 2: Integer types - any encoding -> PLAIN_ENCODING
+            // Applies when: type is integer and config enables plain encoding for integers
+            EncodingTransform {
+                    .predicate =
+                            [](FieldType type, EncodingTypePB encoding, bool optimize_value_seek) {
+                                return is_integer_type(type) &&
+                                       config::integer_type_default_use_plain_encoding;
+                            },
+                    .target_encoding = PLAIN_ENCODING}};
+
+    auto& encoding_map =
+            optimize_value_seek ? _value_seek_encoding_map : _default_encoding_type_map;
+    auto it = encoding_map.find(type);
+    if (it != encoding_map.end()) {
+        EncodingTypePB encoding = it->second;
+
+        // Apply hooks in order to transform the default encoding
+        for (const auto& hook : hooks) {
+            if (hook.predicate(type, encoding, optimize_value_seek)) {
+                // Verify target encoding is available for this type
+                if (_encoding_map.contains(std::make_pair(type, hook.target_encoding))) {
+                    encoding = hook.target_encoding;
+                    break; // Apply only the first matching hook
+                }
+            }
+        }
+
+        return encoding;
+    }
+    return UNKNOWN_ENCODING;
+}
+
 Status EncodingInfoResolver::get(FieldType data_type, EncodingTypePB encoding_type,
                                  const EncodingInfo** out) {
     if (encoding_type == DEFAULT_ENCODING) {

be/src/olap/rowset/segment_v2/encoding_info.h

@@ -103,21 +103,7 @@ class EncodingInfoResolver {
     EncodingInfoResolver();
     ~EncodingInfoResolver();
 
-    EncodingTypePB get_default_encoding(FieldType type, bool optimize_value_seek) const {
-        auto& encoding_map =
-                optimize_value_seek ? _value_seek_encoding_map : _default_encoding_type_map;
-        auto it = encoding_map.find(type);
-        if (it != encoding_map.end()) {
-            EncodingTypePB encoding = it->second;
-            // For binary types, use PLAIN_ENCODING_V2 if config::binary_plain_encoding_default_impl is "v2"
-            if (encoding == PLAIN_ENCODING && config::binary_plain_encoding_default_impl == "v2" &&
-                _encoding_map.contains(std::make_pair(type, PLAIN_ENCODING_V2))) {
-                return PLAIN_ENCODING_V2;
-            }
-            return encoding;
-        }
-        return UNKNOWN_ENCODING;
-    }
+    EncodingTypePB get_default_encoding(FieldType type, bool optimize_value_seek) const;
 
     Status get(FieldType data_type, EncodingTypePB encoding_type, const EncodingInfo** out);
 

be/src/olap/rowset/segment_v2/plain_page.h

@@ -25,6 +25,7 @@
 #include "olap/types.h"
 #include "util/coding.h"
 #include "util/faststring.h"
+#include "vec/common/unaligned.h"
 
 namespace doris {
 #include "common/compile_check_begin.h"
@@ -33,32 +34,36 @@ namespace segment_v2 {
 static const size_t PLAIN_PAGE_HEADER_SIZE = sizeof(uint32_t);
 
 template <FieldType Type>
-class PlainPageBuilder : public PageBuilderHelper<PlainPageBuilder<Type> > {
+class PlainPageBuilder : public PageBuilderHelper<PlainPageBuilder<Type>> {
 public:
     using Self = PlainPageBuilder<Type>;
     friend class PageBuilderHelper<Self>;
 
     Status init() override {
         // Reserve enough space for the page, plus a bit of slop since
         // we often overrun the page by a few values.
-        RETURN_IF_CATCH_EXCEPTION(_buffer.reserve(_options.data_page_size + 1024));
+        RETURN_IF_CATCH_EXCEPTION(_buffer.reserve(_options.data_page_size));
         return reset();
     }
 
-    bool is_page_full() override { return _buffer.size() > _options.data_page_size; }
+    bool is_page_full() override { return _remain_element_capacity == 0; }
 
     Status add(const uint8_t* vals, size_t* count) override {
-        if (is_page_full()) {
+        if (is_page_full() || *count == 0) {
             *count = 0;
             return Status::OK();
         }
         size_t old_size = _buffer.size();
+        size_t to_add = std::min(_remain_element_capacity, *count);
         // This may need a large memory, should return error if could not allocated
         // successfully, to avoid BE OOM.
-        RETURN_IF_CATCH_EXCEPTION(_buffer.resize(old_size + *count * SIZE_OF_TYPE));
-        memcpy(&_buffer[old_size], vals, *count * SIZE_OF_TYPE);
-        _count += *count;
-        _raw_data_size += *count * SIZE_OF_TYPE;
+        RETURN_IF_CATCH_EXCEPTION(_buffer.resize(old_size + to_add * SIZE_OF_TYPE));
+        memcpy(&_buffer[old_size], vals, to_add * SIZE_OF_TYPE);
+        _count += to_add;
+        _raw_data_size += to_add * SIZE_OF_TYPE;
+
+        *count = to_add;
+        _remain_element_capacity -= to_add;
         return Status::OK();
     }
 
@@ -78,11 +83,12 @@ class PlainPageBuilder : public PageBuilderHelper<PlainPageBuilder<Type> > {
 
     Status reset() override {
         RETURN_IF_CATCH_EXCEPTION({
-            _buffer.reserve(_options.data_page_size + 1024);
+            _buffer.reserve(_options.data_page_size);
             _count = 0;
             _raw_data_size = 0;
             _buffer.clear();
             _buffer.resize(PLAIN_PAGE_HEADER_SIZE);
+            _remain_element_capacity = _options.data_page_size / SIZE_OF_TYPE;
         });
         return Status::OK();
     }
@@ -115,6 +121,7 @@ class PlainPageBuilder : public PageBuilderHelper<PlainPageBuilder<Type> > {
     faststring _buffer;
     PageBuilderOptions _options;
     size_t _count;
+    size_t _remain_element_capacity {0};
     uint64_t _raw_data_size = 0;
     typedef typename TypeTraits<Type>::CppType CppType;
     enum { SIZE_OF_TYPE = TypeTraits<Type>::size };
@@ -204,7 +211,48 @@ class PlainPageDecoder : public PageDecoder {
     }
 
     Status next_batch(size_t* n, vectorized::MutableColumnPtr& dst) override {
-        return Status::NotSupported("plain page not implement vec op now");
+        DCHECK(_parsed);
+        if (*n == 0 || _cur_idx >= _num_elems) [[unlikely]] {
+            return Status::OK();
+        }
+
+        size_t max_fetch = std::min(*n, static_cast<size_t>(_num_elems - _cur_idx));
+        const void* src_data = &_data[PLAIN_PAGE_HEADER_SIZE + _cur_idx * SIZE_OF_TYPE];
+
+        dst->insert_many_fix_len_data((const char*)src_data, max_fetch);
+
+        *n = max_fetch;
+        _cur_idx += max_fetch;
+
+        return Status::OK();
+    }
+
+    Status read_by_rowids(const rowid_t* rowids, ordinal_t page_first_ordinal, size_t* n,
+                          vectorized::MutableColumnPtr& dst) override {
+        DCHECK(_parsed);
+        if (*n == 0) [[unlikely]] {
+            return Status::OK();
+        }
+
+        auto total = *n;
+        auto read_count = 0;
+        _buffer.resize(total);
+        for (size_t i = 0; i < total; ++i) {
+            ordinal_t ord = rowids[i] - page_first_ordinal;
+            if (UNLIKELY(ord >= _num_elems)) {
+                break;
+            }
+
+            _buffer[read_count++] =
+                    unaligned_load<CppType>(&_data[PLAIN_PAGE_HEADER_SIZE + ord * SIZE_OF_TYPE]);
+        }
+
+        if (LIKELY(read_count > 0)) {
+            dst->insert_many_fix_len_data((char*)_buffer.data(), read_count);
+        }
+
+        *n = read_count;
+        return Status::OK();
     }
 
     size_t count() const override {
@@ -225,6 +273,8 @@ class PlainPageDecoder : public PageDecoder {
     uint32_t _cur_idx;
     typedef typename TypeTraits<Type>::CppType CppType;
     enum { SIZE_OF_TYPE = TypeTraits<Type>::size };
+
+    std::vector<std::conditional_t<std::is_same_v<CppType, bool>, uint8_t, CppType>> _buffer;
 };
 
 } // namespace segment_v2

be/src/pipeline/task_scheduler.h

@@ -80,10 +80,11 @@ class TaskScheduler {
 
 class HybridTaskScheduler MOCK_REMOVE(final) : public TaskScheduler {
 public:
-    HybridTaskScheduler(int core_num, std::string name,
+    HybridTaskScheduler(int exec_thread_num, int blocking_exec_thread_num, std::string name,
                         std::shared_ptr<CgroupCpuCtl> cgroup_cpu_ctl)
-            : _blocking_scheduler(core_num * 2, name + "_blocking_scheduler", cgroup_cpu_ctl),
-              _simple_scheduler(core_num, name + "_simple_scheduler", cgroup_cpu_ctl) {}
+            : _blocking_scheduler(blocking_exec_thread_num, name + "_blocking_scheduler",
+                                  cgroup_cpu_ctl),
+              _simple_scheduler(exec_thread_num, name + "_simple_scheduler", cgroup_cpu_ctl) {}
 
     Status submit(PipelineTaskSPtr task) override;