[Pick](branch-2.0) Pick from branch-2.0

be/CMakeLists.txt

@@ -559,6 +559,7 @@ set(COMMON_THIRDPARTY
     xml2
     lzma
     simdjson
+    deflate
 )
 
 if ((ARCH_AMD64 OR ARCH_AARCH64) AND OS_LINUX)

be/cmake/thirdparty.cmake

@@ -299,3 +299,9 @@ if (OS_MACOSX)
     add_library(intl STATIC IMPORTED)
     set_target_properties(intl PROPERTIES IMPORTED_LOCATION "${THIRDPARTY_DIR}/lib/libintl.a")
 endif()
+
+# Only used on x86 or x86_64
+if ("${CMAKE_BUILD_TARGET_ARCH}" STREQUAL "x86" OR "${CMAKE_BUILD_TARGET_ARCH}" STREQUAL "x86_64")
+    add_library(deflate STATIC IMPORTED)
+    set_target_properties(deflate PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/libdeflate.a)
+endif()

be/src/common/config.cpp

@@ -1086,7 +1086,7 @@ DEFINE_Int32(group_commit_insert_threads, "10");
 
 DEFINE_mInt32(scan_thread_nice_value, "0");
 
-DEFINE_mInt32(tablet_schema_cache_recycle_interval, "86400");
+DEFINE_mInt32(tablet_schema_cache_recycle_interval, "3600");
 
 DEFINE_Bool(exit_on_exception, "false");
 
@@ -1113,6 +1113,7 @@ DEFINE_mInt32(variant_max_merged_tablet_schema_size, "2048");
 // then the new created tablet will not locate in the high use disk.
 // range: 0 ~ 100
 DEFINE_mInt32(disk_diff_usage_percentage_for_create_tablet, "20");
+DEFINE_Bool(enable_snapshot_action, "false");
 
 // clang-format off
 #ifdef BE_TEST

be/src/common/config.h

@@ -1174,6 +1174,8 @@ DECLARE_mInt32(variant_max_merged_tablet_schema_size);
 // then the new created tablet will not locate in the high use disk.
 // range: 0 ~ 100
 DECLARE_mInt32(disk_diff_usage_percentage_for_create_tablet);
+// whether to enable /api/snapshot api
+DECLARE_Bool(enable_snapshot_action);
 
 #ifdef BE_TEST
 // test s3

be/src/common/status.h

@@ -272,9 +272,10 @@ E(INVERTED_INDEX_NO_TERMS, -6005);
 E(INVERTED_INDEX_RENAME_FILE_FAILED, -6006);
 E(INVERTED_INDEX_EVALUATE_SKIPPED, -6007);
 E(INVERTED_INDEX_BUILD_WAITTING, -6008);
-E(KEY_NOT_FOUND, -6009);
-E(KEY_ALREADY_EXISTS, -6010);
-E(ENTRY_NOT_FOUND, -6011);
+E(INVERTED_INDEX_NOT_IMPLEMENTED, -6009);
+E(KEY_NOT_FOUND, -7000);
+E(KEY_ALREADY_EXISTS, -7001);
+E(ENTRY_NOT_FOUND, -7002);
 #undef E
 } // namespace ErrorCode
 
@@ -300,6 +301,7 @@ constexpr bool capture_stacktrace(int code) {
         && code != ErrorCode::SEGCOMPACTION_INIT_READER
         && code != ErrorCode::SEGCOMPACTION_INIT_WRITER
         && code != ErrorCode::SEGCOMPACTION_FAILED
+        && code != ErrorCode::INVALID_ARGUMENT
         && code != ErrorCode::INVERTED_INDEX_INVALID_PARAMETERS
         && code != ErrorCode::INVERTED_INDEX_NOT_SUPPORTED
         && code != ErrorCode::INVERTED_INDEX_CLUCENE_ERROR
@@ -308,6 +310,7 @@ constexpr bool capture_stacktrace(int code) {
         && code != ErrorCode::INVERTED_INDEX_NO_TERMS
         && code != ErrorCode::INVERTED_INDEX_EVALUATE_SKIPPED
         && code != ErrorCode::INVERTED_INDEX_BUILD_WAITTING
+        && code != ErrorCode::INVERTED_INDEX_NOT_IMPLEMENTED
         && code != ErrorCode::META_KEY_NOT_FOUND
         && code != ErrorCode::PUSH_VERSION_ALREADY_EXIST
         && code != ErrorCode::VERSION_NOT_EXIST

be/src/http/action/snapshot_action.cpp

@@ -41,6 +41,10 @@ SnapshotAction::SnapshotAction(ExecEnv* exec_env, TPrivilegeHier::type hier,
         : HttpHandlerWithAuth(exec_env, hier, type) {}
 
 void SnapshotAction::handle(HttpRequest* req) {
+    if (!config::enable_snapshot_action) {
+        HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "feature disabled");
+        return;
+    }
     LOG(INFO) << "accept one request " << req->debug_string();
     // Get tablet id
     const std::string& tablet_id_str = req->param(TABLET_ID);

be/src/io/fs/buffered_reader.cpp

@@ -476,6 +476,10 @@ void PrefetchBuffer::prefetch_buffer() {
         return;
     }
     if (!s.ok() && _offset < _reader->size()) {
+        // We should print the error msg since this buffer might not be accessed by the consumer
+        // which would result in the status being missed
+        LOG_WARNING("prefetch path {} failed, offset {}, error {}", _reader->path().native(),
+                    _offset, s.to_string());
         _prefetch_status = std::move(s);
     }
     _buffer_status = BufferStatus::PREFETCHED;

be/src/io/fs/s3_file_reader.cpp

@@ -98,8 +98,10 @@ Status S3FileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_rea
     auto outcome = client->GetObject(request);
     s3_bvar::s3_get_total << 1;
     if (!outcome.IsSuccess()) {
-        return Status::IOError("failed to read from {}: {}", _path.native(),
-                               outcome.GetError().GetMessage());
+        return Status::IOError("failed to read from {}: {}, exception {}, error code {}",
+                               _path.native(), outcome.GetError().GetMessage(),
+                               outcome.GetError().GetExceptionName(),
+                               outcome.GetError().GetResponseCode());
     }
     *bytes_read = outcome.GetResult().GetContentLength();
     if (*bytes_read != bytes_req) {

be/src/io/fs/s3_file_system.cpp

@@ -538,9 +538,10 @@ Status S3FileSystem::get_key(const Path& path, std::string* key) const {
 
 template <typename AwsOutcome>
 std::string S3FileSystem::error_msg(const std::string& key, const AwsOutcome& outcome) const {
-    return fmt::format("(endpoint: {}, bucket: {}, key:{}, {}), {}", _s3_conf.endpoint,
-                       _s3_conf.bucket, key, outcome.GetError().GetExceptionName(),
-                       outcome.GetError().GetMessage());
+    return fmt::format("(endpoint: {}, bucket: {}, key:{}, {}), {}, error code {}",
+                       _s3_conf.endpoint, _s3_conf.bucket, key,
+                       outcome.GetError().GetExceptionName(), outcome.GetError().GetMessage(),
+                       outcome.GetError().GetResponseCode());
 }
 
 std::string S3FileSystem::error_msg(const std::string& key, const std::string& err) const {

be/src/io/fs/s3_file_writer.cpp

@@ -119,8 +119,11 @@ Status S3FileWriter::_create_multi_upload_request() {
         _upload_id = outcome.GetResult().GetUploadId();
         return Status::OK();
     }
-    return Status::IOError("failed to create multipart upload(bucket={}, key={}, upload_id={}): {}",
-                           _bucket, _path.native(), _upload_id, outcome.GetError().GetMessage());
+    return Status::IOError(
+            "failed to create multipart upload(bucket={}, key={}, upload_id={}): {}, exception {}, "
+            "error code {}",
+            _bucket, _path.native(), _upload_id, outcome.GetError().GetMessage(),
+            outcome.GetError().GetExceptionName(), outcome.GetError().GetResponseCode());
 }
 
 void S3FileWriter::_wait_until_finish(std::string_view task_name) {
@@ -171,8 +174,11 @@ Status S3FileWriter::abort() {
         _aborted = true;
         return Status::OK();
     }
-    return Status::IOError("failed to abort multipart upload(bucket={}, key={}, upload_id={}): {}",
-                           _bucket, _path.native(), _upload_id, outcome.GetError().GetMessage());
+    return Status::IOError(
+            "failed to abort multipart upload(bucket={}, key={}, upload_id={}): {}, exception {}, "
+            "error code {}",
+            _bucket, _path.native(), _upload_id, outcome.GetError().GetMessage(),
+            outcome.GetError().GetExceptionName(), outcome.GetError().GetResponseCode());
 }
 
 Status S3FileWriter::close() {
@@ -281,9 +287,12 @@ void S3FileWriter::_upload_one_part(int64_t part_num, S3FileBuffer& buf) {
     UploadPartOutcome upload_part_outcome = upload_part_callable.get();
     if (!upload_part_outcome.IsSuccess()) {
         auto s = Status::IOError(
-                "failed to upload part (bucket={}, key={}, part_num={}, up_load_id={}): {}",
+                "failed to upload part (bucket={}, key={}, part_num={}, up_load_id={}): {}, "
+                "exception {}, error code {}",
                 _bucket, _path.native(), part_num, _upload_id,
-                upload_part_outcome.GetError().GetMessage());
+                upload_part_outcome.GetError().GetMessage(),
+                upload_part_outcome.GetError().GetExceptionName(),
+                upload_part_outcome.GetError().GetResponseCode());
         LOG_WARNING(s.to_string());
         buf._on_failed(s);
         return;
@@ -331,8 +340,11 @@ Status S3FileWriter::_complete() {
 
     if (!compute_outcome.IsSuccess()) {
         auto s = Status::IOError(
-                "failed to create complete multi part upload (bucket={}, key={}): {}", _bucket,
-                _path.native(), compute_outcome.GetError().GetMessage());
+                "failed to create complete multi part upload (bucket={}, key={}): {}, exception "
+                "{}, error code {}",
+                _bucket, _path.native(), compute_outcome.GetError().GetMessage(),
+                compute_outcome.GetError().GetExceptionName(),
+                compute_outcome.GetError().GetResponseCode());
         LOG_WARNING(s.to_string());
         return s;
     }
@@ -371,12 +383,12 @@ void S3FileWriter::_put_object(S3FileBuffer& buf) {
     auto response = _client->PutObject(request);
     s3_bvar::s3_put_total << 1;
     if (!response.IsSuccess()) {
-        _st = Status::InternalError("Error: [{}:{}, responseCode:{}]",
-                                    response.GetError().GetExceptionName(),
-                                    response.GetError().GetMessage(),
-                                    static_cast<int>(response.GetError().GetResponseCode()));
-        buf._on_failed(_st);
+        _st = Status::InternalError(
+                "failed to put object (bucket={}, key={}), Error: [{}:{}, responseCode:{}]",
+                _bucket, _path.native(), response.GetError().GetExceptionName(),
+                response.GetError().GetMessage(), response.GetError().GetResponseCode());
         LOG(WARNING) << _st;
+        buf._on_failed(_st);
         return;
     }
     _bytes_written += buf.get_size();

be/src/olap/block_column_predicate.cpp

@@ -216,7 +216,7 @@ void AndBlockColumnPredicate::evaluate_vec(vectorized::MutableColumns& block, ui
 Status AndBlockColumnPredicate::evaluate(const std::string& column_name,
                                          InvertedIndexIterator* iterator, uint32_t num_rows,
                                          roaring::Roaring* bitmap) const {
-    return Status::NotSupported(
+    return Status::Error<ErrorCode::INVERTED_INDEX_NOT_IMPLEMENTED>(
             "Not Implemented evaluate with inverted index, please check the predicate");
 }
 

be/src/olap/block_column_predicate.h

@@ -94,7 +94,7 @@ class BlockColumnPredicate {
     //evaluate predicate on inverted
     virtual Status evaluate(const std::string& column_name, InvertedIndexIterator* iterator,
                             uint32_t num_rows, roaring::Roaring* bitmap) const {
-        return Status::NotSupported(
+        return Status::Error<ErrorCode::INVERTED_INDEX_NOT_IMPLEMENTED>(
                 "Not Implemented evaluate with inverted index, please check the predicate");
     }
 };

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

@@ -1028,7 +1028,7 @@ Status SegmentIterator::_apply_inverted_index_on_block_column_predicate(
             return res;
         } else {
             //TODO:mock until AndBlockColumnPredicate evaluate is ok.
-            if (res.code() == ErrorCode::NOT_IMPLEMENTED_ERROR) {
+            if (res.code() == ErrorCode::INVERTED_INDEX_NOT_IMPLEMENTED) {
                 return Status::OK();
             }
             LOG(WARNING) << "failed to evaluate index"

be/src/pipeline/task_scheduler.cpp

@@ -319,6 +319,11 @@ void TaskScheduler::_do_work(size_t index) {
 }
 
 void TaskScheduler::_try_close_task(PipelineTask* task, PipelineTaskState state) {
+    // close_a_pipeline may delete fragment context and will core in some defer
+    // code, because the defer code will access fragment context it self.
+    std::shared_ptr<PipelineFragmentContext> lock_for_context =
+            task->fragment_context()->shared_from_this();
+
     if (task->is_pending_finish()) {
         task->set_state(PipelineTaskState::PENDING_FINISH);
         _blocked_task_scheduler->add_blocked_task(task);

be/src/runtime/descriptors.cpp

@@ -192,7 +192,6 @@ MaxComputeTableDescriptor::MaxComputeTableDescriptor(const TTableDescriptor& tde
           _table(tdesc.mcTable.table),
           _access_key(tdesc.mcTable.access_key),
           _secret_key(tdesc.mcTable.secret_key),
-          _partition_spec(tdesc.mcTable.partition_spec),
           _public_access(tdesc.mcTable.public_access) {}
 
 MaxComputeTableDescriptor::~MaxComputeTableDescriptor() = default;

be/src/runtime/descriptors.h

@@ -238,7 +238,6 @@ class MaxComputeTableDescriptor : public TableDescriptor {
     const std::string table() const { return _table; }
     const std::string access_key() const { return _access_key; }
     const std::string secret_key() const { return _secret_key; }
-    const std::string partition_spec() const { return _partition_spec; }
     const std::string public_access() const { return _public_access; }
 
 private:
@@ -247,7 +246,6 @@ class MaxComputeTableDescriptor : public TableDescriptor {
     std::string _table;
     std::string _access_key;
     std::string _secret_key;
-    std::string _partition_spec;
     std::string _public_access;
 };
 

be/src/runtime/runtime_filter_mgr.cpp

@@ -39,7 +39,6 @@
 #include "runtime/runtime_state.h"
 #include "runtime/thread_context.h"
 #include "util/brpc_client_cache.h"
-#include "util/spinlock.h"
 
 namespace doris {
 
@@ -227,7 +226,7 @@ Status RuntimeFilterMergeControllerEntity::_init_with_desc(
     auto filter_id = runtime_filter_desc->filter_id;
     // LOG(INFO) << "entity filter id:" << filter_id;
     cntVal->filter->init_with_desc(&cntVal->runtime_filter_desc, query_options, -1, false);
-    _filter_map.emplace(filter_id, CntlValwithLock {cntVal, std::make_unique<SpinLock>()});
+    _filter_map.emplace(filter_id, CntlValwithLock {cntVal, std::make_unique<std::mutex>()});
     return Status::OK();
 }
 
@@ -249,7 +248,7 @@ Status RuntimeFilterMergeControllerEntity::_init_with_desc(
     auto filter_id = runtime_filter_desc->filter_id;
     // LOG(INFO) << "entity filter id:" << filter_id;
     cntVal->filter->init_with_desc(&cntVal->runtime_filter_desc, query_options);
-    _filter_map.emplace(filter_id, CntlValwithLock {cntVal, std::make_unique<SpinLock>()});
+    _filter_map.emplace(filter_id, CntlValwithLock {cntVal, std::make_unique<std::mutex>()});
     return Status::OK();
 }
 
@@ -323,7 +322,7 @@ Status RuntimeFilterMergeControllerEntity::merge(const PMergeFilterRequest* requ
     // iter->second = pair{CntlVal,SpinLock}
     cntVal = iter->second.first;
     {
-        std::lock_guard<SpinLock> l(*iter->second.second);
+        std::lock_guard<std::mutex> l(*iter->second.second);
         MergeRuntimeFilterParams params(request, attach_data);
         ObjectPool* pool = cntVal->pool.get();
         RuntimeFilterWrapperHolder holder;

be/src/runtime/runtime_filter_mgr.h

@@ -170,7 +170,7 @@ class RuntimeFilterMergeControllerEntity {
     std::shared_mutex _filter_map_mutex;
     std::shared_ptr<MemTracker> _mem_tracker;
     using CntlValwithLock =
-            std::pair<std::shared_ptr<RuntimeFilterCntlVal>, std::unique_ptr<SpinLock>>;
+            std::pair<std::shared_ptr<RuntimeFilterCntlVal>, std::unique_ptr<std::mutex>>;
     std::map<int, CntlValwithLock> _filter_map;
     RuntimeState* _state;
     bool _opt_remote_rf = true;

be/src/service/internal_service.cpp

@@ -461,10 +461,22 @@ Status PInternalServiceImpl::_exec_plan_fragment_impl(const std::string& ser_req
             uint32_t len = ser_request.size();
             RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, compact, &t_request));
         }
+        const auto& fragment_list = t_request.paramsList;
+        MonotonicStopWatch timer;
+        timer.start();
 
         for (const TExecPlanFragmentParams& params : t_request.paramsList) {
             RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(params));
         }
+
+        timer.stop();
+        double cost_secs = static_cast<double>(timer.elapsed_time()) / 1000000000ULL;
+        if (cost_secs > 5) {
+            LOG_WARNING("Prepare {} fragments of query {} costs {} seconds, it costs too much",
+                        fragment_list.size(), print_id(fragment_list.front().params.query_id),
+                        cost_secs);
+        }
+
         return Status::OK();
     } else if (version == PFragmentRequestVersion::VERSION_3) {
         TPipelineFragmentParamsList t_request;
@@ -474,9 +486,21 @@ Status PInternalServiceImpl::_exec_plan_fragment_impl(const std::string& ser_req
             RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, compact, &t_request));
         }
 
-        for (const TPipelineFragmentParams& params : t_request.params_list) {
-            RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(params));
+        const auto& fragment_list = t_request.params_list;
+        MonotonicStopWatch timer;
+        timer.start();
+
+        for (const TPipelineFragmentParams& fragment : fragment_list) {
+            RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(fragment));
+        }
+
+        timer.stop();
+        double cost_secs = static_cast<double>(timer.elapsed_time()) / 1000000000ULL;
+        if (cost_secs > 5) {
+            LOG_WARNING("Prepare {} fragments of query {} costs {} seconds, it costs too much",
+                        fragment_list.size(), print_id(fragment_list.front().query_id), cost_secs);
         }
+
         return Status::OK();
     } else {
         return Status::InternalError("invalid version");