Merge branch 'master' into fdb_tool

.asf.yaml

@@ -49,12 +49,9 @@ github:
         strict: false
         contexts:
           - License Check
-          - Clang Formatter
           - CheckStyle
           - P0 Regression (Doris Regression)
-          - P1 Regression (Doris Regression)
           - External Regression (Doris External Regression)
-          - cloud_p1 (Doris Cloud Regression)
           - cloud_p0 (Doris Cloud Regression)
           - FE UT (Doris FE UT)
           - BE UT (Doris BE UT)
@@ -89,7 +86,6 @@ github:
         strict: false
         contexts:
           - License Check
-          - Clang Formatter
           - CheckStyle
           - Build Broker
           - ShellCheck
@@ -111,10 +107,8 @@ github:
         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)

.github/workflows/clang-format.yml

@@ -67,6 +67,11 @@ jobs:
           git checkout 6adbe14579e5b8e19eb3e31e5ff2479f3bd302c7
           popd &>/dev/null
 
+      - name: Install Python dependencies
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.10'  # Adjust if needed
+
       - name: "Format it!"
         if: ${{ steps.filter.outputs.changes == 'true' }}
         uses: ./.github/actions/clang-format-lint-action

.github/workflows/code-checks.yml

@@ -50,8 +50,8 @@ jobs:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
           sh_checker_comment: true
-          sh_checker_exclude: .git .github ^docker ^thirdparty/src ^thirdparty/installed ^ui ^docs/node_modules ^tools/clickbench-tools ^extension ^output ^fs_brokers/apache_hdfs_broker/output (^|.*/)Dockerfile$ ^be/src/apache-orc ^be/src/clucene ^pytest ^toos/fdb/fdb_vars.sh
-
+          sh_checker_exclude: .git .github ^docker ^thirdparty/src ^thirdparty/installed ^ui ^docs/node_modules ^tools/clickbench-tools ^extension ^output ^fs_brokers/apache_hdfs_broker/output (^|.*/)Dockerfile$ ^be/src/apache-orc ^be/src/clucene ^pytest ^samples ^toos/fdb/fdb_vars.sh
+  
   preparation:
     name: "Clang Tidy Preparation"
     if: ${{ github.event_name == 'pull_request_target' }}

.licenserc.yaml

@@ -93,4 +93,5 @@ header:
     - "pytest/sys/data"
     - "pytest/deploy/*.conf"
     - "tools/jeprof"
+    - "tools/FlameGraph/*"
   comment: on-failure

LICENSE.txt

@@ -725,4 +725,12 @@ Apache 2.0, Copyright 2023 SAP SE or an SAP affiliate company, Johannes Bechberg
 
 This project is maintained by the SapMachine team at SAP SE
 
-----------------------------------------------------------------------------------
+----------------------------------------------------------------------------------
+
+be/tools/FlameGraph/*.pl: COMMON DEVELOPMENT AND DISTRIBUTION LICENSE Version 1.0
+
+Unless otherwise noted, all files in this distribution are released
+under the Common Development and Distribution License (CDDL).
+Exceptions are noted within the associated source files.
+
+----------------------------------------------------------------------------------

be/src/agent/be_exec_version_manager.cpp

@@ -0,0 +1,92 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "agent/be_exec_version_manager.h"
+
+namespace doris {
+
+const std::map<int, const std::set<std::string>> AGGREGATION_CHANGE_MAP = {
+        {AGGREGATION_2_1_VERSION,
+         {"window_funnel", "stddev_samp", "variance_samp", "percentile_approx_weighted",
+          "percentile_approx", "covar_samp", "percentile", "percentile_array"}}};
+
+Status BeExecVersionManager::check_be_exec_version(int be_exec_version) {
+    if (be_exec_version > max_be_exec_version || be_exec_version < min_be_exec_version) {
+        return Status::InternalError(
+                "Received be_exec_version is not supported, be_exec_version={}, "
+                "min_be_exec_version={}, max_be_exec_version={}, maybe due to FE version not "
+                "match with BE.",
+                be_exec_version, min_be_exec_version, max_be_exec_version);
+    }
+    return Status::OK();
+}
+
+void BeExecVersionManager::check_agg_state_compatibility(int current_be_exec_version,
+                                                         int data_be_exec_version,
+                                                         std::string function_name) {
+    if (current_be_exec_version > AGGREGATION_2_1_VERSION &&
+        data_be_exec_version <= AGGREGATION_2_1_VERSION &&
+        AGGREGATION_CHANGE_MAP.find(AGGREGATION_2_1_VERSION)->second.contains(function_name)) {
+        throw Exception(Status::InternalError(
+                "agg state data with {} is not supported, "
+                "current_be_exec_version={}, data_be_exec_version={}, need to rebuild the data "
+                "or set the be_exec_version={} in fe.conf",
+                function_name, current_be_exec_version, data_be_exec_version,
+                AGGREGATION_2_1_VERSION));
+    }
+}
+
+/**
+ * When we have some breaking change for execute engine, we should update be_exec_version.
+ * NOTICE: The change could only be dont in X.Y.0 version. and if you introduced new version number N,
+ *  remember remove version N-1's all REUSEABLE changes in master branch only. REUSEABLE means scalar or agg functions' replacement.
+ *  If not, the old replacement will happens in the new version which is wrong.
+ *
+ * 0: not contain be_exec_version.
+ * 1: start from doris 1.2.0
+ *    a. remove ColumnString terminating zero.
+ *    b. runtime filter use new hash method.
+ * 2: start from doris 2.0.0
+ *    a. function month/day/hour/minute/second's return type is changed to smaller type.
+ *    b. in order to solve agg of sum/count is not compatibility during the upgrade process
+ *    c. change the string hash method in runtime filter
+ *    d. elt function return type change to nullable(string)
+ *    e. add repeat_max_num in repeat function
+ * 3: start from doris 2.0.0 (by some mistakes)
+ *    a. aggregation function do not serialize bitmap to string.
+ *    b. support window funnel mode.
+ * 4: start from doris 2.1.0
+ *    a. ignore this line, window funnel mode should be enabled from 2.0.
+ *    b. array contains/position/countequal function return nullable in less situations.
+ *    c. cleared old version of Version 2.
+ *    d. unix_timestamp function support timestamp with float for datetimev2, and change nullable mode.
+ *    e. change shuffle serialize/deserialize way 
+ *    f. shrink some function's nullable mode.
+ *    g. do local merge of remote runtime filter
+ *    h. "now": ALWAYS_NOT_NULLABLE -> DEPEND_ON_ARGUMENTS
+ *
+ * 5: start from doris 3.0.0
+ *    a. change the impl of percentile (need fix)
+ *    b. clear old version of version 3->4
+ *    c. change FunctionIsIPAddressInRange from AlwaysNotNullable to DependOnArguments
+ *    d. change some agg function nullable property: PR #37215
+ *    e. change variant serde to fix PR #38413
+ */
+const int BeExecVersionManager::max_be_exec_version = 7;
+const int BeExecVersionManager::min_be_exec_version = 0;
+
+} // namespace doris

be/src/agent/be_exec_version_manager.h

@@ -20,24 +20,28 @@
 #include <fmt/format.h>
 #include <glog/logging.h>
 
+#include "common/exception.h"
 #include "common/status.h"
 
 namespace doris {
 
+/// functional
+constexpr inline int BITMAP_SERDE = 3;
+constexpr inline int USE_NEW_SERDE = 4;         // release on DORIS version 2.1
+constexpr inline int OLD_WAL_SERDE = 3;         // use to solve compatibility issues, see pr #32299
+constexpr inline int AGG_FUNCTION_NULLABLE = 5; // change some agg nullable property: PR #37215
+constexpr inline int VARIANT_SERDE = 6;         // change variant serde to fix PR #38413
+constexpr inline int AGGREGATION_2_1_VERSION =
+        4; // some aggregation changed the data format after this version
+
 class BeExecVersionManager {
 public:
     BeExecVersionManager() = delete;
 
-    static Status check_be_exec_version(int be_exec_version) {
-        if (be_exec_version > max_be_exec_version || be_exec_version < min_be_exec_version) {
-            return Status::InternalError(
-                    "Received be_exec_version is not supported, be_exec_version={}, "
-                    "min_be_exec_version={}, max_be_exec_version={}, maybe due to FE version not "
-                    "match with BE.",
-                    be_exec_version, min_be_exec_version, max_be_exec_version);
-        }
-        return Status::OK();
-    }
+    static Status check_be_exec_version(int be_exec_version);
+
+    static void check_agg_state_compatibility(int current_be_exec_version, int data_be_exec_version,
+                                              std::string function_name);
 
     static int get_newest_version() { return max_be_exec_version; }
 
@@ -46,50 +50,4 @@ class BeExecVersionManager {
     static const int min_be_exec_version;
 };
 
-/**
- * When we have some breaking change for execute engine, we should update be_exec_version.
- * NOTICE: The change could only be dont in X.Y.0 version. and if you introduced new version number N,
- *  remember remove version N-1's all REUSEABLE changes in master branch only. REUSEABLE means scalar or agg functions' replacement.
- *  If not, the old replacement will happens in the new version which is wrong.
- *
- * 0: not contain be_exec_version.
- * 1: start from doris 1.2.0
- *    a. remove ColumnString terminating zero.
- *    b. runtime filter use new hash method.
- * 2: start from doris 2.0.0
- *    a. function month/day/hour/minute/second's return type is changed to smaller type.
- *    b. in order to solve agg of sum/count is not compatibility during the upgrade process
- *    c. change the string hash method in runtime filter
- *    d. elt function return type change to nullable(string)
- *    e. add repeat_max_num in repeat function
- * 3: start from doris 2.0.0 (by some mistakes)
- *    a. aggregation function do not serialize bitmap to string.
- *    b. support window funnel mode.
- * 4: start from doris 2.1.0
- *    a. ignore this line, window funnel mode should be enabled from 2.0.
- *    b. array contains/position/countequal function return nullable in less situations.
- *    c. cleared old version of Version 2.
- *    d. unix_timestamp function support timestamp with float for datetimev2, and change nullable mode.
- *    e. change shuffle serialize/deserialize way 
- *    f. shrink some function's nullable mode.
- *    g. do local merge of remote runtime filter
- *    h. "now": ALWAYS_NOT_NULLABLE -> DEPEND_ON_ARGUMENTS
- *
- * 5: start from doris 3.0.0
- *    a. change the impl of percentile (need fix)
- *    b. clear old version of version 3->4
- *    c. change FunctionIsIPAddressInRange from AlwaysNotNullable to DependOnArguments
- *    d. change some agg function nullable property: PR #37215
- *    e. change variant serde to fix PR #38413
- */
-constexpr inline int BeExecVersionManager::max_be_exec_version = 7;
-constexpr inline int BeExecVersionManager::min_be_exec_version = 0;
-
-/// functional
-constexpr inline int BITMAP_SERDE = 3;
-constexpr inline int USE_NEW_SERDE = 4;         // release on DORIS version 2.1
-constexpr inline int OLD_WAL_SERDE = 3;         // use to solve compatibility issues, see pr #32299
-constexpr inline int AGG_FUNCTION_NULLABLE = 5; // change some agg nullable property: PR #37215
-constexpr inline int VARIANT_SERDE = 6;         // change variant serde to fix PR #38413
-
 } // namespace doris

be/src/cloud/cloud_backend_service.cpp

@@ -180,6 +180,11 @@ void CloudBackendService::check_warm_up_cache_async(TCheckWarmUpCacheAsyncRespon
                                                     const TCheckWarmUpCacheAsyncRequest& request) {
     std::map<int64_t, bool> task_done;
     _engine.file_cache_block_downloader().check_download_task(request.tablets, &task_done);
+    DBUG_EXECUTE_IF("CloudBackendService.check_warm_up_cache_async.return_task_false", {
+        for (auto& it : task_done) {
+            it.second = false;
+        }
+    });
     response.__set_task_done(task_done);
 
     Status st = Status::OK();

be/src/cloud/cloud_base_compaction.cpp

@@ -29,6 +29,7 @@
 #include "service/backend_options.h"
 #include "util/thread.h"
 #include "util/uuid_generator.h"
+#include "vec/runtime/vdatetime_value.h"
 
 namespace doris {
 using namespace ErrorCode;
@@ -82,21 +83,40 @@ Status CloudBaseCompaction::prepare_compact() {
     compaction_job->set_type(cloud::TabletCompactionJobPB::BASE);
     compaction_job->set_base_compaction_cnt(_base_compaction_cnt);
     compaction_job->set_cumulative_compaction_cnt(_cumulative_compaction_cnt);
+    compaction_job->add_input_versions(_input_rowsets.front()->start_version());
+    compaction_job->add_input_versions(_input_rowsets.back()->end_version());
     using namespace std::chrono;
     int64_t now = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();
     _expiration = now + config::compaction_timeout_seconds;
     compaction_job->set_expiration(_expiration);
     compaction_job->set_lease(now + config::lease_compaction_interval_seconds * 4);
     cloud::StartTabletJobResponse resp;
-    //auto st = cloud::meta_mgr()->prepare_tablet_job(job, &resp);
     auto st = _engine.meta_mgr().prepare_tablet_job(job, &resp);
+    if (resp.has_alter_version()) {
+        (static_cast<CloudTablet*>(_tablet.get()))->set_alter_version(resp.alter_version());
+    }
     if (!st.ok()) {
         if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
             // set last_sync_time to 0 to force sync tablet next time
             cloud_tablet()->last_sync_time_s = 0;
         } else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
             // tablet not found
             cloud_tablet()->clear_cache();
+        } else if (resp.status().code() == cloud::JOB_CHECK_ALTER_VERSION) {
+            auto* cloud_tablet = (static_cast<CloudTablet*>(_tablet.get()));
+            std::stringstream ss;
+            ss << "failed to prepare cumu compaction. Check compaction input versions "
+                  "failed in schema change. The input version end must "
+                  "less than or equal to alter_version."
+                  "current alter version in BE is not correct."
+                  "input_version_start="
+               << compaction_job->input_versions(0)
+               << " input_version_end=" << compaction_job->input_versions(1)
+               << " current alter_version=" << cloud_tablet->alter_version()
+               << " schema_change_alter_version=" << resp.alter_version();
+            std::string msg = ss.str();
+            LOG(WARNING) << msg;
+            return Status::InternalError(msg);
         }
         return st;
     }
@@ -320,6 +340,22 @@ Status CloudBaseCompaction::modify_rowsets() {
     if (!st.ok()) {
         if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
             cloud_tablet()->clear_cache();
+        } else if (resp.status().code() == cloud::JOB_CHECK_ALTER_VERSION) {
+            auto* cloud_tablet = (static_cast<CloudTablet*>(_tablet.get()));
+            std::stringstream ss;
+            ss << "failed to prepare cumu compaction. Check compaction input versions "
+                  "failed in schema change. The input version end must "
+                  "less than or equal to alter_version."
+                  "current alter version in BE is not correct."
+                  "input_version_start="
+               << compaction_job->input_versions(0)
+               << " input_version_end=" << compaction_job->input_versions(1)
+               << " current alter_version=" << cloud_tablet->alter_version()
+               << " schema_change_alter_version=" << resp.alter_version();
+            std::string msg = ss.str();
+            LOG(WARNING) << msg;
+            cloud_tablet->set_alter_version(resp.alter_version());
+            return Status::InternalError(msg);
         }
         return st;
     }

be/src/cloud/cloud_compaction_action.cpp

@@ -62,7 +62,7 @@ const static std::string HEADER_JSON = "application/json";
 CloudCompactionAction::CloudCompactionAction(CompactionActionType ctype, ExecEnv* exec_env,
                                              CloudStorageEngine& engine, TPrivilegeHier::type hier,
                                              TPrivilegeType::type ptype)
-        : HttpHandlerWithAuth(exec_env, hier, ptype), _engine(engine), _type(ctype) {}
+        : HttpHandlerWithAuth(exec_env, hier, ptype), _engine(engine), _compaction_type(ctype) {}
 
 /// check param and fetch tablet_id & table_id from req
 static Status _check_param(HttpRequest* req, uint64_t* tablet_id, uint64_t* table_id) {
@@ -233,15 +233,15 @@ Status CloudCompactionAction::_handle_run_status_compaction(HttpRequest* req,
 void CloudCompactionAction::handle(HttpRequest* req) {
     req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.c_str());
 
-    if (_type == CompactionActionType::SHOW_INFO) {
+    if (_compaction_type == CompactionActionType::SHOW_INFO) {
         std::string json_result;
         Status st = _handle_show_compaction(req, &json_result);
         if (!st.ok()) {
             HttpChannel::send_reply(req, HttpStatus::OK, st.to_json());
         } else {
             HttpChannel::send_reply(req, HttpStatus::OK, json_result);
         }
-    } else if (_type == CompactionActionType::RUN_COMPACTION) {
+    } else if (_compaction_type == CompactionActionType::RUN_COMPACTION) {
         std::string json_result;
         Status st = _handle_run_compaction(req, &json_result);
         if (!st.ok()) {

be/src/cloud/cloud_compaction_action.h

@@ -70,7 +70,7 @@ class CloudCompactionAction : public HttpHandlerWithAuth {
 
 private:
     CloudStorageEngine& _engine;
-    CompactionActionType _type;
+    CompactionActionType _compaction_type;
 };
 
 } // end namespace doris