Merge branch 'master' into fix/stale-version-count-bug

Author: LemonCL

.asf.yaml

@@ -180,5 +180,7 @@ github:
     - yuanyuan8983
     - yz-jayhua
     - ixzc
+    - deardeng
+    - wyxxxcat
 notifications:
   pullrequests_status:  commits@doris.apache.org

be/CMakeLists.txt

@@ -838,6 +838,12 @@ endif()
 option(BUILD_FILE_CACHE_MICROBENCH_TOOL "Build file cache mirobench Tool" OFF)
 if (BUILD_FILE_CACHE_MICROBENCH_TOOL)
     add_subdirectory(${SRC_DIR}/io/tools)
+    install(FILES
+        ${BASE_DIR}/../bin/start_file_cache_microbench.sh
+        PERMISSIONS OWNER_READ OWNER_WRITE OWNER_EXECUTE
+        GROUP_READ GROUP_WRITE GROUP_EXECUTE
+        WORLD_READ WORLD_EXECUTE
+        DESTINATION ${OUTPUT_DIR}/bin)
 endif()
 
 option(BUILD_INDEX_TOOL "Build index tool" OFF)

be/src/http/action/file_cache_action.cpp

@@ -61,28 +61,29 @@ constexpr static std::string_view VALUE = "value";
 constexpr static std::string_view RELOAD = "reload";
 
 Status FileCacheAction::_handle_header(HttpRequest* req, std::string* json_metrics) {
-    req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.data());
-    std::string operation = req->param(OP.data());
+    const std::string header_json(HEADER_JSON);
+    req->add_output_header(HttpHeaders::CONTENT_TYPE, header_json.c_str());
+    std::string operation = req->param(std::string(OP));
     Status st = Status::OK();
     if (operation == RELEASE) {
         size_t released = 0;
-        const std::string& base_path = req->param(BASE_PATH.data());
+        const std::string& base_path = req->param(std::string(BASE_PATH));
         if (!base_path.empty()) {
             released = io::FileCacheFactory::instance()->try_release(base_path);
         } else {
             released = io::FileCacheFactory::instance()->try_release();
         }
         EasyJson json;
-        json[RELEASED_ELEMENTS.data()] = released;
+        json[std::string(RELEASED_ELEMENTS)] = released;
         *json_metrics = json.ToString();
     } else if (operation == CLEAR) {
         DBUG_EXECUTE_IF("FileCacheAction._handle_header.ignore_clear", {
             LOG_WARNING("debug point FileCacheAction._handle_header.ignore_clear");
             st = Status::OK();
             return st;
         });
-        const std::string& sync = req->param(SYNC.data());
-        const std::string& segment_path = req->param(VALUE.data());
+        const std::string& sync = req->param(std::string(SYNC));
+        const std::string& segment_path = req->param(std::string(VALUE));
         if (segment_path.empty()) {
             io::FileCacheFactory::instance()->clear_file_caches(to_lower(sync) == "true");
         } else {
@@ -91,7 +92,7 @@ Status FileCacheAction::_handle_header(HttpRequest* req, std::string* json_metri
             cache->remove_if_cached(hash);
         }
     } else if (operation == RESET) {
-        std::string capacity = req->param(CAPACITY.data());
+        std::string capacity = req->param(std::string(CAPACITY));
         int64_t new_capacity = 0;
         bool parse = true;
         try {
@@ -105,24 +106,24 @@ Status FileCacheAction::_handle_header(HttpRequest* req, std::string* json_metri
                     "the interval (0, INT64_MAX]",
                     capacity);
         } else {
-            const std::string& path = req->param(PATH.data());
+            const std::string& path = req->param(std::string(PATH));
             auto ret = io::FileCacheFactory::instance()->reset_capacity(path, new_capacity);
             LOG(INFO) << ret;
         }
     } else if (operation == HASH) {
-        const std::string& segment_path = req->param(VALUE.data());
+        const std::string& segment_path = req->param(std::string(VALUE));
         if (segment_path.empty()) {
-            st = Status::InvalidArgument("missing parameter: {} is required", VALUE.data());
+            st = Status::InvalidArgument("missing parameter: {} is required", VALUE);
         } else {
             io::UInt128Wrapper ret = io::BlockFileCache::hash(segment_path);
             EasyJson json;
-            json[HASH.data()] = ret.to_string();
+            json[std::string(HASH)] = ret.to_string();
             *json_metrics = json.ToString();
         }
     } else if (operation == LIST_CACHE) {
-        const std::string& segment_path = req->param(VALUE.data());
+        const std::string& segment_path = req->param(std::string(VALUE));
         if (segment_path.empty()) {
-            st = Status::InvalidArgument("missing parameter: {} is required", VALUE.data());
+            st = Status::InvalidArgument("missing parameter: {} is required", VALUE);
         } else {
             io::UInt128Wrapper cache_hash = io::BlockFileCache::hash(segment_path);
             std::vector<std::string> cache_files =
@@ -145,9 +146,9 @@ Status FileCacheAction::_handle_header(HttpRequest* req, std::string* json_metri
                               [&json](auto& x) { json.PushBack(std::move(x)); });
         *json_metrics = json.ToString();
     } else if (operation == CHECK_CONSISTENCY) {
-        const std::string& cache_base_path = req->param(BASE_PATH.data());
+        const std::string& cache_base_path = req->param(std::string(BASE_PATH));
         if (cache_base_path.empty()) {
-            st = Status::InvalidArgument("missing parameter: {} is required", BASE_PATH.data());
+            st = Status::InvalidArgument("missing parameter: {} is required", BASE_PATH);
         } else {
             auto* block_file_cache = io::FileCacheFactory::instance()->get_by_path(cache_base_path);
             if (block_file_cache == nullptr) {

be/src/io/tools/file_cache_microbench.cpp

@@ -101,6 +101,27 @@ const size_t BUFFER_SIZE = 1024 * 1024;
 // Just 10^9.
 static constexpr auto NS = 1000000000UL;
 
+static std::string normalize_benchmark_prefix(std::string_view raw_prefix) {
+    std::string normalized {doris::trim(raw_prefix)};
+    while (!normalized.empty() && normalized.front() == '/') {
+        normalized.erase(normalized.begin());
+    }
+    while (!normalized.empty() && normalized.back() == '/') {
+        normalized.pop_back();
+    }
+    return normalized;
+}
+
+static std::string get_prefix() {
+    std::string prefix = HIDDEN_PREFIX;
+    std::string subdir = normalize_benchmark_prefix(doris::config::test_s3_prefix);
+    if (!subdir.empty()) {
+        prefix += subdir;
+        prefix += "/";
+    }
+    return prefix;
+}
+
 DEFINE_int32(port, 8888, "Http Port of this server");
 
 static std::string build_info() {
@@ -487,7 +508,7 @@ std::string get_usage(const std::string& progname) {
           "read_iops": <limit>,                // IOPS limit for reading per segment files
           "num_threads": <count>,              // Number of threads in the thread pool, default 200
           "num_files": <count>,                // Number of segments to write/read
-          "file_prefix": "<prefix>",           // Prefix for segment files, Notice: this tools hide prefix(test_file_cache_microbench/) before file_prefix
+          "file_prefix": "<prefix>",           // Prefix for segment files, key prefix is test_file_cache_microbench/<test_s3_prefix>/
           "write_batch_size": <size>,          // Size of data to write in each write operation
           "cache_type": <type>,                // Write or Read data enter file cache queue type, support NORMAL | TTL | INDEX | DISPOSABLE, default NORMAL
           "expiration": <timestamp>,           // File cache ttl expire time, value is a unix timestamp
@@ -720,7 +741,7 @@ struct JobConfig {
                 "repeat: {}, expiration: {}, cache_type: {}, read_offset: [{}, {}), "
                 "read_length: [{}, {})",
                 size_bytes_perfile, write_iops, read_iops, num_threads, num_files,
-                HIDDEN_PREFIX + file_prefix, write_file_cache, write_batch_size, repeat, expiration,
+                get_prefix() + file_prefix, write_file_cache, write_batch_size, repeat, expiration,
                 cache_type, read_offset_left, read_offset_right, read_length_left,
                 read_length_right);
     }
@@ -1280,7 +1301,7 @@ class JobManager {
         // If it's a read-only job, find the previously written files
         if (config.read_iops > 0 && config.write_iops == 0) {
             std::string old_job_id =
-                    s3_file_records.find_job_id_by_prefix(HIDDEN_PREFIX + config.file_prefix);
+                    s3_file_records.find_job_id_by_prefix(get_prefix() + config.file_prefix);
             if (old_job_id.empty()) {
                 throw std::runtime_error(
                         "Can't find previously job uploaded files. Please make sure read "
@@ -1293,7 +1314,7 @@ class JobManager {
 
         // Generate file keys
         for (int i = 0; i < config.num_files; ++i) {
-            keys.push_back(HIDDEN_PREFIX + config.file_prefix + "/" + rewrite_job_id + "_" +
+            keys.push_back(get_prefix() + config.file_prefix + "/" + rewrite_job_id + "_" +
                            std::to_string(i));
         }
 
@@ -1424,7 +1445,7 @@ class JobManager {
         auto start_time = std::chrono::steady_clock::now();
 
         int64_t exist_job_perfile_size = s3_file_records.get_exist_job_perfile_size_by_prefix(
-                HIDDEN_PREFIX + config.file_prefix);
+                get_prefix() + config.file_prefix);
         std::vector<std::future<void>> read_futures;
         doris::io::IOContext io_ctx;
         doris::io::FileCacheStatistics total_stats;
@@ -1447,7 +1468,7 @@ class JobManager {
         std::vector<std::string> read_files;
         if (exist_job_perfile_size != -1) {
             // read exist files
-            s3_file_records.get_exist_job_files_by_prefix(HIDDEN_PREFIX + config.file_prefix,
+            s3_file_records.get_exist_job_files_by_prefix(get_prefix() + config.file_prefix,
                                                           read_files, config.num_files);
         }
 
@@ -2662,12 +2683,8 @@ int main(int argc, char* argv[]) {
     };
     periodiccally_log_thread = std::thread {periodiccally_log};
 
-    try {
-        HttpServer http_server;
-        http_server.start(doris_home);
-    } catch (const std::exception& e) {
-        LOG(ERROR) << "Error in HTTP server: " << e.what();
-    }
+    HttpServer http_server;
+    http_server.start(doris_home);
 
     if (periodiccally_log_thread.joinable()) {
         {

be/src/olap/memtable_writer.cpp

@@ -160,9 +160,6 @@ Status MemTableWriter::_flush_memtable_async() {
         std::lock_guard<std::mutex> l(_mem_table_ptr_lock);
         memtable = _mem_table;
         _mem_table = nullptr;
-    }
-    {
-        std::lock_guard<std::mutex> l(_mem_table_ptr_lock);
         memtable->update_mem_type(MemType::WRITE_FINISHED);
         _freezed_mem_tables.push_back(memtable);
     }

be/src/pipeline/common/set_utils.h

@@ -51,9 +51,13 @@ using SetPrimaryTypeHashTableContextNullable = vectorized::MethodSingleNullableC
 using SetSerializedHashTableContext =
         vectorized::MethodSerialized<PHHashMap<StringRef, RowRefWithFlag>>;
 using SetMethodOneString = vectorized::MethodStringNoCache<PHHashMap<StringRef, RowRefWithFlag>>;
+using SetMethodOneStringNullable =
+        vectorized::MethodSingleNullableColumn<vectorized::MethodStringNoCache<
+                vectorized::DataWithNullKey<PHHashMap<StringRef, RowRefWithFlag>>>>;
 
 using SetHashTableVariants =
         std::variant<std::monostate, SetSerializedHashTableContext, SetMethodOneString,
+                     SetMethodOneStringNullable,
                      SetPrimaryTypeHashTableContextNullable<vectorized::UInt8>,
                      SetPrimaryTypeHashTableContextNullable<vectorized::UInt16>,
                      SetPrimaryTypeHashTableContextNullable<vectorized::UInt32>,
@@ -102,7 +106,11 @@ struct SetDataVariants
             emplace_single<vectorized::UInt256, SetData<vectorized::UInt256>>(nullable);
             break;
         case HashKeyType::string_key:
-            method_variant.emplace<SetMethodOneString>();
+            if (nullable) {
+                method_variant.emplace<SetMethodOneStringNullable>();
+            } else {
+                method_variant.emplace<SetMethodOneString>();
+            }
             break;
         case HashKeyType::fixed64:
             method_variant.emplace<SetFixedKeyHashTableContext<vectorized::UInt64>>(

be/src/pipeline/exec/olap_scan_operator.cpp

@@ -894,21 +894,7 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() {
                                     _scan_keys.extend_scan_key(temp_range, p._max_scan_key_num,
                                                                &exact_range, &eos, &should_break));
                             if (exact_range) {
-                                auto key = iter->first;
-                                _slot_id_to_value_range.erase(key);
-
-                                std::vector<std::shared_ptr<ColumnPredicate>> new_predicates;
-                                for (const auto& it : _slot_id_to_predicates[key]) {
-                                    if (it->type() == PredicateType::NOT_IN_LIST ||
-                                        it->type() == PredicateType::NE) {
-                                        new_predicates.push_back(it);
-                                    }
-                                }
-                                if (new_predicates.empty()) {
-                                    _slot_id_to_predicates.erase(key);
-                                } else {
-                                    _slot_id_to_predicates[key] = new_predicates;
-                                }
+                                _slot_id_to_value_range.erase(iter->first);
                             }
                         } else {
                             // if exceed max_pushdown_conditions_per_column, use whole_value_rang instead

be/src/pipeline/exec/scan_operator.cpp

@@ -458,6 +458,9 @@ Status ScanLocalState<Derived>::_normalize_bloom_filter(
         if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
+        } else {
+            // If exception occurs during processing, do not push down
+            *pdt = PushDownType::UNACCEPTABLE;
         }
     };
     DCHECK(TExprNodeType::BLOOM_PRED == root->node_type());
@@ -484,6 +487,9 @@ Status ScanLocalState<Derived>::_normalize_topn_filter(
         if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
+        } else {
+            // If exception occurs during processing, do not push down
+            *pdt = PushDownType::UNACCEPTABLE;
         }
     };
     DCHECK(root->is_topn_filter());
@@ -508,6 +514,9 @@ Status ScanLocalState<Derived>::_normalize_bitmap_filter(
         if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
+        } else {
+            // If exception occurs during processing, do not push down
+            *pdt = PushDownType::UNACCEPTABLE;
         }
     };
     DCHECK(TExprNodeType::BITMAP_PRED == root->node_type());
@@ -656,6 +665,9 @@ Status ScanLocalState<Derived>::_normalize_in_predicate(
         if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
+        } else {
+            // If exception occurs during processing, do not push down
+            *pdt = PushDownType::UNACCEPTABLE;
         }
     };
 
@@ -754,6 +766,9 @@ Status ScanLocalState<Derived>::_normalize_binary_predicate(
         if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
+        } else {
+            // If exception occurs during processing, do not push down
+            *pdt = PushDownType::UNACCEPTABLE;
         }
     };
 
@@ -931,6 +946,9 @@ Status ScanLocalState<Derived>::_normalize_is_null_predicate(
         if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
+        } else {
+            // If exception occurs during processing, do not push down
+            *pdt = PushDownType::UNACCEPTABLE;
         }
     };
     DCHECK(!root->is_rf_wrapper()) << root->debug_string();