Merge branch 'master' into show-columns-hide-version

Author: catpineapple

.asf.yaml

@@ -50,7 +50,7 @@ github:
           - Clang Formatter
           - CheckStyle
           - P0 Regression (Doris Regression)
-          - NonConcurrent Regression (Doris Regression)
+          - NonConcurrent Regression (nonConcurrent)
           - External Regression (Doris External Regression)
           - cloud_p0 (Doris Cloud Regression)
           - vault_p0 (Doris Cloud Regression)
@@ -83,7 +83,7 @@ github:
           - Cloud UT (Doris Cloud UT)
           - COMPILE (DORIS_COMPILE)
           - P0 Regression (Doris Regression)
-          - NonConcurrent Regression (Doris Regression)
+          - NonConcurrent Regression (nonConcurrent)
           - External Regression (Doris External Regression) 
           - cloud_p0 (Doris Cloud Regression)
           - vault_p0 (Doris Cloud Regression)
@@ -105,7 +105,7 @@ github:
           - Cloud UT (Doris Cloud UT)
           - COMPILE (DORIS_COMPILE)
           - P0 Regression (Doris Regression)
-          - NonConcurrent Regression (Doris Regression)
+          - NonConcurrent Regression (nonConcurrent)
           - External Regression (Doris External Regression) 
           - cloud_p0 (Doris Cloud Regression)
           - vault_p0 (Doris Cloud Regression)
@@ -127,7 +127,7 @@ github:
           - Cloud UT (Doris Cloud UT)
           - COMPILE (DORIS_COMPILE)
           - P0 Regression (Doris Regression)
-          - NonConcurrent Regression (Doris Regression)
+          - NonConcurrent Regression (nonConcurrent)
           - External Regression (Doris External Regression)
           - cloud_p0 (Doris Cloud Regression)
           - vault_p0 (Doris Cloud Regression)

be/src/cloud/cloud_rowset_writer.cpp

@@ -61,6 +61,7 @@ Status CloudRowsetWriter::init(const RowsetWriterContext& rowset_writer_context)
         _rowset_meta->set_newest_write_timestamp(_context.newest_write_timestamp);
     }
     _rowset_meta->set_tablet_schema(_context.tablet_schema);
+    _rowset_meta->set_job_id(_context.job_id);
     _context.segment_collector = std::make_shared<SegmentCollectorT<BaseBetaRowsetWriter>>(this);
     _context.file_writer_creator = std::make_shared<FileWriterCreatorT<BaseBetaRowsetWriter>>(this);
     if (_context.mow_context != nullptr) {

be/src/cloud/cloud_schema_change_job.cpp

@@ -326,6 +326,7 @@ Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParam
         context.tablet_schema = _new_tablet->tablet_schema();
         context.newest_write_timestamp = rs_reader->newest_write_timestamp();
         context.storage_resource = _cloud_storage_engine.get_storage_resource(sc_params.vault_id);
+        context.job_id = _job_id;
         context.write_file_cache = sc_params.output_to_file_cache;
         context.tablet = _new_tablet;
         if (!context.storage_resource) {

be/src/cloud/cloud_stream_load_executor.cpp

@@ -104,6 +104,11 @@ Status CloudStreamLoadExecutor::operate_txn_2pc(StreamLoadContext* ctx) {
 
 Status CloudStreamLoadExecutor::commit_txn(StreamLoadContext* ctx) {
     DBUG_EXECUTE_IF("StreamLoadExecutor.commit_txn.block", DBUG_BLOCK);
+    DBUG_EXECUTE_IF("StreamLoadExecutor.commit_txn.crash", {
+        LOG(INFO) << "debug point " << DP_NAME << " trigger crash";
+        volatile int* p = nullptr;
+        *p = 1;
+    });
     // forward to fe to excute commit transaction for MoW table
     if (ctx->is_mow_table() || !config::enable_stream_load_commit_txn_on_be ||
         ctx->load_type == TLoadType::ROUTINE_LOAD) {

be/src/cloud/cloud_tablet_mgr.cpp

@@ -29,6 +29,7 @@
 #include "common/status.h"
 #include "olap/lru_cache.h"
 #include "runtime/memory/cache_policy.h"
+#include "util/debug_points.h"
 #include "util/stack_util.h"
 
 namespace doris {
@@ -148,7 +149,9 @@ CloudTabletMgr::CloudTabletMgr(CloudStorageEngine& engine)
           _tablet_map(std::make_unique<TabletMap>()),
           _cache(std::make_unique<LRUCachePolicy>(
                   CachePolicy::CacheType::CLOUD_TABLET_CACHE, config::tablet_cache_capacity,
-                  LRUCacheType::NUMBER, 0, config::tablet_cache_shards, false /*enable_prune*/)) {}
+                  LRUCacheType::NUMBER, /*sweep time*/ 0, config::tablet_cache_shards,
+                  /*element_count_capacity*/ 0, /*enable_prune*/ false,
+                  /*is_lru_k*/ false)) {}
 
 CloudTabletMgr::~CloudTabletMgr() = default;
 
@@ -163,6 +166,7 @@ Result<std::shared_ptr<CloudTablet>> CloudTabletMgr::get_tablet(int64_t tablet_i
                                                                 SyncRowsetStats* sync_stats,
                                                                 bool force_use_only_cached,
                                                                 bool cache_on_miss) {
+    DBUG_EXECUTE_IF("CloudTabletMgr::get_tablet.block", DBUG_BLOCK);
     // LRU value type. `Value`'s lifetime MUST NOT be longer than `CloudTabletMgr`
     class Value : public LRUCacheValueBase {
     public:

be/src/cloud/cloud_txn_delete_bitmap_cache.cpp

@@ -34,7 +34,9 @@ namespace doris {
 
 CloudTxnDeleteBitmapCache::CloudTxnDeleteBitmapCache(size_t size_in_bytes)
         : LRUCachePolicy(CachePolicy::CacheType::CLOUD_TXN_DELETE_BITMAP_CACHE, size_in_bytes,
-                         LRUCacheType::SIZE, 86400, 4),
+                         LRUCacheType::SIZE, /*stale_sweep_time_s*/ 86400, /*num_shards*/ 4,
+                         /*element_count_capacity*/ 0, /*enable_prune*/ true,
+                         /*is_lru_k*/ false),
           _stop_latch(1) {}
 
 CloudTxnDeleteBitmapCache::~CloudTxnDeleteBitmapCache() {

be/src/cloud/config.cpp

@@ -110,7 +110,7 @@ DEFINE_mBool(enable_cloud_tablet_report, "true");
 
 DEFINE_mInt32(delete_bitmap_rpc_retry_times, "25");
 
-DEFINE_mInt64(meta_service_rpc_reconnect_interval_ms, "5000");
+DEFINE_mInt64(meta_service_rpc_reconnect_interval_ms, "100");
 
 DEFINE_mInt32(meta_service_conflict_error_retry_times, "10");
 

be/src/cloud/injection_point_action.cpp

@@ -279,6 +279,17 @@ void set_return_error(const std::string& point, HttpRequest* req) {
     HttpChannel::send_reply(req, HttpStatus::OK, "OK");
 }
 
+void set_segfault(const std::string& point, HttpRequest* req) {
+    auto sp = SyncPoint::get_instance();
+    sp->set_call_back(point, [point](auto&&) {
+        LOG(INFO) << "injection point hit, point=" << point << " trigger segfault";
+        // Intentional null dereference to crash the BE for testing.
+        volatile int* p = nullptr;
+        *p = 1;
+    });
+    HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
 void handle_set(HttpRequest* req) {
     auto& point = req->param("name");
     if (point.empty()) {
@@ -302,6 +313,9 @@ void handle_set(HttpRequest* req) {
     } else if (behavior == "return_error") {
         set_return_error(point, req);
         return;
+    } else if (behavior == "segfault") {
+        set_segfault(point, req);
+        return;
     }
     HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "unknown behavior: " + behavior);
 }
@@ -377,13 +391,15 @@ InjectionPointAction::InjectionPointAction() = default;
 //                 which is an int, valid values can be found in status.h, e.g. -235 or -230,
 //                 if `code` is not present return Status::InternalError. Optional `probability`
 //                 determines the percentage of times to inject the error (default 100).
+// * segfault: dereference a null pointer to crash BE intentionally
 // ```
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=sleep&duration=${x_millsec}" # sleep x millisecs
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return" # return void
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return_ok" # return ok
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return_error" # internal error
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return_error&code=${code}" # -235
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return_error&code=${code}&probability=50" # inject with 50% probability
+// curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=segfault" # crash BE
 // ```
 void InjectionPointAction::handle(HttpRequest* req) {
     LOG(INFO) << "handle InjectionPointAction " << req->debug_string();

be/src/cloud/pb_convert.cpp

@@ -105,6 +105,12 @@ void doris_rowset_meta_to_cloud(RowsetMetaCloudPB* out, const RowsetMetaPB& in)
     auto* slice_locations = out->mutable_packed_slice_locations();
     slice_locations->clear();
     slice_locations->insert(in.packed_slice_locations().begin(), in.packed_slice_locations().end());
+    if (in.has_is_recycled()) {
+        out->set_is_recycled(in.is_recycled());
+    }
+    if (in.has_job_id()) {
+        out->set_job_id(in.job_id());
+    }
 }
 
 void doris_rowset_meta_to_cloud(RowsetMetaCloudPB* out, RowsetMetaPB&& in) {
@@ -177,6 +183,12 @@ void doris_rowset_meta_to_cloud(RowsetMetaCloudPB* out, RowsetMetaPB&& in) {
     auto* slice_locations = out->mutable_packed_slice_locations();
     slice_locations->clear();
     slice_locations->insert(in.packed_slice_locations().begin(), in.packed_slice_locations().end());
+    if (in.has_is_recycled()) {
+        out->set_is_recycled(in.is_recycled());
+    }
+    if (in.has_job_id()) {
+        out->set_job_id(in.job_id());
+    }
 }
 
 RowsetMetaPB cloud_rowset_meta_to_doris(const RowsetMetaCloudPB& in) {
@@ -259,6 +271,12 @@ void cloud_rowset_meta_to_doris(RowsetMetaPB* out, const RowsetMetaCloudPB& in)
     auto* slice_locations = out->mutable_packed_slice_locations();
     slice_locations->clear();
     slice_locations->insert(in.packed_slice_locations().begin(), in.packed_slice_locations().end());
+    if (in.has_is_recycled()) {
+        out->set_is_recycled(in.is_recycled());
+    }
+    if (in.has_job_id()) {
+        out->set_job_id(in.job_id());
+    }
 }
 
 void cloud_rowset_meta_to_doris(RowsetMetaPB* out, RowsetMetaCloudPB&& in) {
@@ -330,6 +348,12 @@ void cloud_rowset_meta_to_doris(RowsetMetaPB* out, RowsetMetaCloudPB&& in) {
     auto* slice_locations = out->mutable_packed_slice_locations();
     slice_locations->clear();
     slice_locations->insert(in.packed_slice_locations().begin(), in.packed_slice_locations().end());
+    if (in.has_is_recycled()) {
+        out->set_is_recycled(in.is_recycled());
+    }
+    if (in.has_job_id()) {
+        out->set_job_id(in.job_id());
+    }
 }
 
 TabletSchemaCloudPB doris_tablet_schema_to_cloud(const TabletSchemaPB& in) {

be/src/common/config.cpp

@@ -308,7 +308,7 @@ DEFINE_Int32(task_executor_max_concurrency_per_task, "-1");
 DEFINE_Int32(task_executor_initial_max_concurrency_per_task, "-1");
 
 // Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "false");
+DEFINE_Bool(enable_task_executor_in_internal_table, "true");
 // Enable task executor in external table scan.
 DEFINE_Bool(enable_task_executor_in_external_table, "true");