Merge branch 'apache:master' into fix-complex-nested-add-regress

.asf.yaml

@@ -80,6 +80,31 @@ github:
         dismiss_stale_reviews: true
         required_approving_review_count: 1
 
+    branch-2.0:
+      required_status_checks:
+      # if strict is true, means "Require branches to be up to date before merging".
+        strict: false
+        contexts:
+          - License Check
+          - Clang Formatter
+          - CheckStyle
+          - P0 Regression (Doris Regression)
+          - P1 Regression (Doris Regression)
+          - External Regression (Doris External Regression)
+          - FE UT (Doris FE UT)
+          - BE UT (Doris BE UT)
+          - Build Broker
+          - Build Documents
+          - ShellCheck
+          - clickbench-new (clickbench)
+          - Build Third Party Libraries (Linux)
+          - Build Third Party Libraries (macOS)
+          - COMPILE (DORIS_COMPILE)
+
+        required_pull_request_reviews:
+          dismiss_stale_reviews: true
+          required_approving_review_count: 1
+
   collaborators:
     - LemonLiTree
     - Yukang-Lian

.clang-tidy

@@ -7,13 +7,19 @@ Checks: |
   modernize-*,
   -modernize-use-trailing-return-type,
   -modernize-use-nodiscard,
+  -modernize-avoid-c-arrays,
   misc-redundant-expression,
   misc-unused-*,
   -misc-unused-parameters,
   readability-*,
   -readability-identifier-length,
   -readability-implicit-bool-conversion,
   -readability-function-cognitive-complexity,
+  -readability-magic-numbers,
+  -readability-else-after-return,
+  -readability-inconsistent-declaration-parameter-name,
+  -readability-isolate-declaration,
+  -readability-named-parameter,
   portability-simd-intrinsics,
   performance-type-promotion-in-math-fn,
   performance-faster-string-find,

.github/workflows/pr-approve-status.yml

@@ -33,7 +33,7 @@ jobs:
               echo "PR number is not set"
               exit 1
           fi
-          response=$(curl -s  -H "Authorization: token ${{ secrets.GITHUB_TOKEN }} "  "https://api.github.com/repos/apache/doris/pulls/${pr_num}/reviews")
+          response=$(curl -s  -H "Authorization: token ${{ secrets.GITHUB_TOKEN }} "  "https://api.github.com/repos/apache/doris/pulls/${pr_num}/reviews?per_page=100")
           # shellcheck disable=SC2207
           reviewers=($(echo $response | jq -r '.[] | .user.login'))
           # shellcheck disable=SC2207

README.md

@@ -35,8 +35,6 @@ Apache Doris is an easy-to-use, high-performance and real-time analytical databa
 
 All this makes Apache Doris an ideal tool for scenarios including report analysis, ad-hoc query, unified data warehouse, and data lake query acceleration. On Apache Doris, users can build various applications, such as user behavior analysis, AB test platform, log retrieval analysis, user portrait analysis, and order analysis.
 
-Doris Summit Asia 2023 is coming and warmly invite you to join! Click Now 🔗[doris-summit.org.cn](https://doris-summit.org.cn/?utm_source=website&utm_medium=readme&utm_campaign=2023&utm_id=2023)
-
 🎉 Version 2.0.2 version released now. The 2.0.2 version has achieved over 10x performance improvements on standard Benchmark, comprehensive enhancement in log analysis and lakehouse scenarios, more efficient and stable data update and write efficiency, support for more comprehensive multi-tenant and resource isolation mechanisms, and take a new step in the direction of resource elasticity and storage computing separation. It has also been added a series of usability features for enterprise users. We welcome all users who have requirements for the new features of the 2.0 version to deploy and upgrade. Check out the 🔗[Release Notes](https://github.com/apache/doris/issues/25011) here. 
 
 🎉 Version 1.2.7 released now! It is fully evolved release and all users are encouraged to upgrade to this release. Check out the 🔗[Release Notes](https://doris.apache.org/docs/dev/releasenotes/release-1.2.7) here. 

be/src/common/config.cpp

@@ -81,7 +81,7 @@ DEFINE_String(memory_mode, "moderate");
 // defaults to bytes if no unit is given"
 // must larger than 0. and if larger than physical memory size,
 // it will be set to physical memory size.
-DEFINE_String(mem_limit, "80%");
+DEFINE_String(mem_limit, "90%");
 
 // Soft memory limit as a fraction of hard memory limit.
 DEFINE_Double(soft_mem_limit_frac, "0.9");
@@ -319,7 +319,7 @@ DEFINE_Int32(index_page_cache_percentage, "10");
 // whether to disable page cache feature in storage
 DEFINE_Bool(disable_storage_page_cache, "false");
 // whether to disable row cache feature in storage
-DEFINE_Bool(disable_storage_row_cache, "true");
+DEFINE_mBool(disable_storage_row_cache, "true");
 // whether to disable pk page cache feature in storage
 DEFINE_Bool(disable_pk_storage_page_cache, "false");
 
@@ -351,6 +351,9 @@ DEFINE_Int32(vertical_compaction_max_row_source_memory_mb, "200");
 // In vertical compaction, max dest segment file size
 DEFINE_mInt64(vertical_compaction_max_segment_size, "268435456");
 
+// If enabled, segments will be flushed column by column
+DEFINE_mBool(enable_vertical_segment_writer, "true");
+
 // In ordered data compaction, min segment size for input rowset
 DEFINE_mInt32(ordered_data_compaction_min_segment_size, "10485760");
 
@@ -1074,6 +1077,7 @@ DEFINE_String(group_commit_replay_wal_dir, "./wal");
 DEFINE_Int32(group_commit_replay_wal_retry_num, "10");
 DEFINE_Int32(group_commit_replay_wal_retry_interval_seconds, "5");
 DEFINE_Int32(group_commit_sync_wal_batch, "10");
+DEFINE_Bool(wait_internal_group_commit_finish, "false");
 
 // the count of thread to group commit insert
 DEFINE_Int32(group_commit_insert_threads, "10");

be/src/common/config.h

@@ -106,7 +106,7 @@ DECLARE_Int32(brpc_num_threads);
 
 // Declare a selection strategy for those servers have many ips.
 // Note that there should at most one ip match this list.
-// this is a list in semicolon-delimited format, in CIDR notation, e.g. 10.10.10.0/24
+// This is a list in semicolon-delimited format, in CIDR notation, e.g. 10.10.10.0/24
 // If no ip match this rule, will choose one randomly.
 DECLARE_String(priority_networks);
 
@@ -369,7 +369,7 @@ DECLARE_Int32(index_page_cache_percentage);
 // TODO delete it. Divided into Data page, Index page, pk index page
 DECLARE_Bool(disable_storage_page_cache);
 // whether to disable row cache feature in storage
-DECLARE_Bool(disable_storage_row_cache);
+DECLARE_mBool(disable_storage_row_cache);
 // whether to disable pk page cache feature in storage
 DECLARE_Bool(disable_pk_storage_page_cache);
 
@@ -404,6 +404,9 @@ DECLARE_Int32(vertical_compaction_max_row_source_memory_mb);
 // In vertical compaction, max dest segment file size
 DECLARE_mInt64(vertical_compaction_max_segment_size);
 
+// If enabled, segments will be flushed column by column
+DECLARE_mBool(enable_vertical_segment_writer);
+
 // In ordered data compaction, min segment size for input rowset
 DECLARE_mInt32(ordered_data_compaction_min_segment_size);
 
@@ -1143,6 +1146,7 @@ DECLARE_String(group_commit_replay_wal_dir);
 DECLARE_Int32(group_commit_replay_wal_retry_num);
 DECLARE_Int32(group_commit_replay_wal_retry_interval_seconds);
 DECLARE_Int32(group_commit_sync_wal_batch);
+DECLARE_Bool(wait_internal_group_commit_finish);
 
 // This config can be set to limit thread number in group commit insert thread pool.
 DECLARE_mInt32(group_commit_insert_threads);

be/src/exprs/create_predicate_function.h

@@ -17,11 +17,11 @@
 
 #pragma once
 
-#include "bloom_filter_func.h"
 #include "exprs/hybrid_set.h"
 #include "exprs/minmax_predicate.h"
 #include "function_filter.h"
 #include "olap/bitmap_filter_predicate.h"
+#include "olap/bloom_filter_predicate.h"
 #include "olap/column_predicate.h"
 #include "olap/in_list_predicate.h"
 #include "olap/like_column_predicate.h"
@@ -225,6 +225,16 @@ inline auto create_bitmap_filter(PrimitiveType type) {
     return create_bitmap_predicate_function<BitmapFilterTraits>(type);
 }
 
+template <PrimitiveType PT>
+ColumnPredicate* create_olap_column_predicate(uint32_t column_id,
+                                              const std::shared_ptr<BloomFilterFuncBase>& filter,
+                                              int be_exec_version, const TabletColumn*) {
+    std::shared_ptr<BloomFilterFuncBase> filter_olap;
+    filter_olap.reset(create_bloom_filter(PT));
+    filter_olap->light_copy(filter.get());
+    return new BloomFilterColumnPredicate<PT>(column_id, filter, be_exec_version);
+}
+
 template <PrimitiveType PT>
 ColumnPredicate* create_olap_column_predicate(uint32_t column_id,
                                               const std::shared_ptr<BitmapFilterFuncBase>& filter,

be/src/exprs/runtime_filter.cpp

@@ -1185,17 +1185,17 @@ bool IRuntimeFilter::await() {
     return true;
 }
 
+// NOTE: Wait infinitely will not make scan task wait really forever.
+// Because BlockTaskSchedule will make it run when query is timedout.
+bool IRuntimeFilter::wait_infinitely() const {
+    // bitmap filter is precise filter and only filter once, so it must be applied.
+    return _wait_infinitely ||
+           (_wrapper != nullptr && _wrapper->get_real_type() == RuntimeFilterType::BITMAP_FILTER);
+}
+
 bool IRuntimeFilter::is_ready_or_timeout() {
     DCHECK(is_consumer());
     auto cur_state = _rf_state_atomic.load(std::memory_order_acquire);
-    auto execution_timeout = _state == nullptr ? _query_ctx->execution_timeout() * 1000
-                                               : _state->execution_timeout() * 1000;
-    auto runtime_filter_wait_time_ms = _state == nullptr ? _query_ctx->runtime_filter_wait_time_ms()
-                                                         : _state->runtime_filter_wait_time_ms();
-    // bitmap filter is precise filter and only filter once, so it must be applied.
-    int64_t wait_times_ms = _wrapper->get_real_type() == RuntimeFilterType::BITMAP_FILTER
-                                    ? execution_timeout
-                                    : runtime_filter_wait_time_ms;
     int64_t ms_since_registration = MonotonicMillis() - registration_time_;
     if (!_enable_pipeline_exec) {
         _rf_state = RuntimeFilterState::TIME_OUT;
@@ -1212,7 +1212,7 @@ bool IRuntimeFilter::is_ready_or_timeout() {
         if (is_ready()) {
             return true;
         }
-        bool timeout = wait_times_ms <= ms_since_registration;
+        bool timeout = wait_infinitely() ? false : _rf_wait_time_ms <= ms_since_registration;
         auto expected = RuntimeFilterState::NOT_READY;
         if (timeout) {
             if (!_rf_state_atomic.compare_exchange_strong(expected, RuntimeFilterState::TIME_OUT,

be/src/exprs/runtime_filter.h

@@ -211,6 +211,8 @@ class IRuntimeFilter {
               _always_true(false),
               _is_ignored(false),
               registration_time_(MonotonicMillis()),
+              _wait_infinitely(_state->runtime_filter_wait_infinitely()),
+              _rf_wait_time_ms(_state->runtime_filter_wait_time_ms()),
               _enable_pipeline_exec(_state->enable_pipeline_exec()),
               _profile(new RuntimeProfile(_name)) {
         if (desc->__isset.min_max_type && desc->type == TRuntimeFilterType::MIN_MAX) {
@@ -236,6 +238,8 @@ class IRuntimeFilter {
               _always_true(false),
               _is_ignored(false),
               registration_time_(MonotonicMillis()),
+              _wait_infinitely(query_ctx->runtime_filter_wait_infinitely()),
+              _rf_wait_time_ms(query_ctx->runtime_filter_wait_time_ms()),
               _enable_pipeline_exec(query_ctx->enable_pipeline_exec()),
               _profile(new RuntimeProfile(_name)) {
         if (desc->__isset.min_max_type && desc->type == TRuntimeFilterType::MIN_MAX) {
@@ -388,13 +392,21 @@ class IRuntimeFilter {
         }
     }
 
-    int32_t wait_time_ms() {
-        auto runtime_filter_wait_time_ms = _state == nullptr
-                                                   ? _query_ctx->runtime_filter_wait_time_ms()
-                                                   : _state->runtime_filter_wait_time_ms();
-        return runtime_filter_wait_time_ms;
+    // For pipelineX & Producer
+    int32_t wait_time_ms() const {
+        int32_t res = 0;
+        if (wait_infinitely()) {
+            res = _state == nullptr ? _query_ctx->execution_timeout() : _state->execution_timeout();
+            // Convert to ms
+            res *= 1000;
+        } else {
+            res = _rf_wait_time_ms;
+        }
+        return res;
     }
 
+    bool wait_infinitely() const;
+
     int64_t registration_time() const { return registration_time_; }
 
     void set_filter_timer(std::shared_ptr<pipeline::RuntimeFilterTimer>);
@@ -479,6 +491,9 @@ class IRuntimeFilter {
 
     /// Time in ms (from MonotonicMillis()), that the filter was registered.
     const int64_t registration_time_;
+    /// runtime filter wait time will be ignored if wait_infinitly is true
+    const bool _wait_infinitely;
+    const int32_t _rf_wait_time_ms;
 
     const bool _enable_pipeline_exec;
 

be/src/exprs/runtime_filter_rpc.cpp

@@ -74,7 +74,7 @@ Status IRuntimeFilter::push_to_remote(RuntimeState* state, const TNetworkAddress
     _rpc_context->request.set_filter_id(_filter_id);
     _rpc_context->request.set_opt_remote_rf(opt_remote_rf);
     _rpc_context->request.set_is_pipeline(state->enable_pipeline_exec());
-    _rpc_context->cntl.set_timeout_ms(state->runtime_filter_wait_time_ms());
+    _rpc_context->cntl.set_timeout_ms(wait_time_ms());
     _rpc_context->cid = _rpc_context->cntl.call_id();
 
     Status serialize_status = serialize(&_rpc_context->request, &data, &len);

be/src/http/action/http_stream.cpp

@@ -167,7 +167,8 @@ int HttpStreamAction::on_header(HttpRequest* req) {
     ctx->load_type = TLoadType::MANUL_LOAD;
     ctx->load_src_type = TLoadSourceType::RAW;
 
-    ctx->group_commit = iequal(req->header(HTTP_GROUP_COMMIT), "true");
+    ctx->group_commit = iequal(req->header(HTTP_GROUP_COMMIT), "true") ||
+                        config::wait_internal_group_commit_finish;
 
     ctx->two_phase_commit = req->header(HTTP_TWO_PHASE_COMMIT) == "true" ? true : false;
 

be/src/http/action/stream_load.cpp

@@ -188,8 +188,9 @@ int StreamLoadAction::on_header(HttpRequest* req) {
     url_decode(req->param(HTTP_TABLE_KEY), &ctx->table);
     ctx->label = req->header(HTTP_LABEL_KEY);
     Status st = Status::OK();
-    if (iequal(req->header(HTTP_GROUP_COMMIT), "true")) {
-        if (!ctx->label.empty()) {
+    if (iequal(req->header(HTTP_GROUP_COMMIT), "true") ||
+        config::wait_internal_group_commit_finish) {
+        if (iequal(req->header(HTTP_GROUP_COMMIT), "true") && !ctx->label.empty()) {
             st = Status::InternalError("label and group_commit can't be set at the same time");
         }
         ctx->group_commit = true;
@@ -522,7 +523,10 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req,
             request.__set_send_batch_parallelism(
                     std::stoi(http_req->header(HTTP_SEND_BATCH_PARALLELISM)));
         } catch (const std::invalid_argument& e) {
-            return Status::InvalidArgument("Invalid send_batch_parallelism format, {}", e.what());
+            return Status::InvalidArgument("send_batch_parallelism must be an integer, {}",
+                                           e.what());
+        } catch (const std::out_of_range& e) {
+            return Status::InvalidArgument("send_batch_parallelism out of range, {}", e.what());
         }
     }