Merge branch 'master' into master-fix12

Author: SWJTU-ZhangLei

be/src/cloud/cloud_base_compaction.cpp

@@ -302,7 +302,9 @@ Status CloudBaseCompaction::execute_compact() {
             .tag("output_segments", _output_rowset->num_segments())
             .tag("output_rowset_data_size", _output_rowset->data_disk_size())
             .tag("output_rowset_index_size", _output_rowset->index_disk_size())
-            .tag("output_rowset_total_size", _output_rowset->total_disk_size());
+            .tag("output_rowset_total_size", _output_rowset->total_disk_size())
+            .tag("local_read_bytes", _local_read_bytes_total)
+            .tag("remote_read_bytes", _remote_read_bytes_total);
 
     //_compaction_succeed = true;
     _state = CompactionState::SUCCESS;

be/src/cloud/cloud_cumulative_compaction.cpp

@@ -226,7 +226,9 @@ Status CloudCumulativeCompaction::execute_compact() {
             .tag("tablet_max_version", _tablet->max_version_unlocked())
             .tag("cumulative_point", cloud_tablet()->cumulative_layer_point())
             .tag("num_rowsets", cloud_tablet()->fetch_add_approximate_num_rowsets(0))
-            .tag("cumu_num_rowsets", cloud_tablet()->fetch_add_approximate_cumu_num_rowsets(0));
+            .tag("cumu_num_rowsets", cloud_tablet()->fetch_add_approximate_cumu_num_rowsets(0))
+            .tag("local_read_bytes", _local_read_bytes_total)
+            .tag("remote_read_bytes", _remote_read_bytes_total);
 
     _state = CompactionState::SUCCESS;
 

be/src/cloud/cloud_full_compaction.cpp

@@ -182,7 +182,9 @@ Status CloudFullCompaction::execute_compact() {
             .tag("output_segments", _output_rowset->num_segments())
             .tag("output_rowset_data_size", _output_rowset->data_disk_size())
             .tag("output_rowset_index_size", _output_rowset->index_disk_size())
-            .tag("output_rowset_total_size", _output_rowset->total_disk_size());
+            .tag("output_rowset_total_size", _output_rowset->total_disk_size())
+            .tag("local_read_bytes", _local_read_bytes_total)
+            .tag("remote_read_bytes", _remote_read_bytes_total);
 
     _state = CompactionState::SUCCESS;
 

be/src/cloud/cloud_storage_engine.cpp

@@ -58,6 +58,7 @@
 #include "olap/storage_policy.h"
 #include "runtime/memory/cache_manager.h"
 #include "util/parse_util.h"
+#include "util/time.h"
 #include "vec/common/assert_cast.h"
 
 namespace doris {
@@ -445,14 +446,14 @@ void CloudStorageEngine::_compaction_tasks_producer_callback() {
 
     int64_t interval = config::generate_compaction_tasks_interval_ms;
     do {
+        int64_t cur_time = UnixMillis();
         if (!config::disable_auto_compaction) {
             Status st = _adjust_compaction_thread_num();
             if (!st.ok()) {
                 break;
             }
 
             bool check_score = false;
-            int64_t cur_time = UnixMillis();
             if (round < config::cumulative_compaction_rounds_for_each_base_compaction_round) {
                 compaction_type = CompactionType::CUMULATIVE_COMPACTION;
                 round++;
@@ -504,6 +505,9 @@ void CloudStorageEngine::_compaction_tasks_producer_callback() {
         } else {
             interval = config::check_auto_compaction_interval_seconds * 1000;
         }
+        int64_t end_time = UnixMillis();
+        DorisMetrics::instance()->compaction_producer_callback_a_round_time->set_value(end_time -
+                                                                                       cur_time);
     } while (!_stop_background_threads_latch.wait_for(std::chrono::milliseconds(interval)));
 }
 

be/src/cloud/cloud_tablet.cpp

@@ -17,6 +17,7 @@
 
 #include "cloud/cloud_tablet.h"
 
+#include <bvar/bvar.h>
 #include <gen_cpp/olap_file.pb.h>
 #include <rapidjson/document.h>
 #include <rapidjson/encodings.h>
@@ -57,6 +58,11 @@ namespace doris {
 #include "common/compile_check_begin.h"
 using namespace ErrorCode;
 
+bvar::LatencyRecorder g_cu_compaction_get_delete_bitmap_lock_time_ms(
+        "cu_compaction_get_delete_bitmap_lock_time_ms");
+bvar::LatencyRecorder g_base_compaction_get_delete_bitmap_lock_time_ms(
+        "base_compaction_get_delete_bitmap_lock_time_ms");
+
 static constexpr int LOAD_INITIATOR_ID = -1;
 
 CloudTablet::CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta)
@@ -991,6 +997,11 @@ Status CloudTablet::calc_delete_bitmap_for_compaction(
     RETURN_IF_ERROR(_engine.meta_mgr().get_delete_bitmap_update_lock(
             *this, COMPACTION_DELETE_BITMAP_LOCK_ID, initiator));
     int64_t t2 = MonotonicMicros();
+    if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION) {
+        g_cu_compaction_get_delete_bitmap_lock_time_ms << (t2 - t1) / 1000;
+    } else if (compaction_type == ReaderType::READER_BASE_COMPACTION) {
+        g_base_compaction_get_delete_bitmap_lock_time_ms << (t2 - t1) / 1000;
+    }
     get_delete_bitmap_lock_start_time = t2;
     RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(this));
     int64_t t3 = MonotonicMicros();

be/src/common/config.cpp

@@ -614,9 +614,9 @@ DEFINE_mInt32(olap_table_sink_send_interval_microseconds, "1000");
 DEFINE_mDouble(olap_table_sink_send_interval_auto_partition_factor, "0.001");
 
 // Fragment thread pool
-DEFINE_Int32(fragment_mgr_asynic_work_pool_thread_num_min, "16");
-DEFINE_Int32(fragment_mgr_asynic_work_pool_thread_num_max, "512");
-DEFINE_Int32(fragment_mgr_asynic_work_pool_queue_size, "4096");
+DEFINE_Int32(fragment_mgr_async_work_pool_thread_num_min, "16");
+DEFINE_Int32(fragment_mgr_async_work_pool_thread_num_max, "512");
+DEFINE_Int32(fragment_mgr_async_work_pool_queue_size, "4096");
 
 // Control the number of disks on the machine.  If 0, this comes from the system settings.
 DEFINE_Int32(num_disks, "0");

be/src/common/config.h

@@ -650,9 +650,9 @@ DECLARE_mInt32(olap_table_sink_send_interval_microseconds);
 DECLARE_mDouble(olap_table_sink_send_interval_auto_partition_factor);
 
 // Fragment thread pool
-DECLARE_Int32(fragment_mgr_asynic_work_pool_thread_num_min);
-DECLARE_Int32(fragment_mgr_asynic_work_pool_thread_num_max);
-DECLARE_Int32(fragment_mgr_asynic_work_pool_queue_size);
+DECLARE_Int32(fragment_mgr_async_work_pool_thread_num_min);
+DECLARE_Int32(fragment_mgr_async_work_pool_thread_num_max);
+DECLARE_Int32(fragment_mgr_async_work_pool_queue_size);
 
 // Control the number of disks on the machine.  If 0, this comes from the system settings.
 DECLARE_Int32(num_disks);

be/src/exec/es/es_scroll_parser.cpp

@@ -847,7 +847,8 @@ Status ScrollParser::fill_columns(const TupleDescriptor* tuple_desc,
             break;
         }
         case TYPE_JSONB: {
-            JsonBinaryValue binary_val(json_value_to_string(col));
+            JsonBinaryValue binary_val;
+            RETURN_IF_ERROR(binary_val.init_from_json_string(json_value_to_string(col)));
             vectorized::JsonbField json(binary_val.value(), binary_val.size());
             col_ptr->insert(vectorized::Field::create_field<TYPE_JSONB>(json));
             break;

be/src/exec/rowid_fetcher.cpp

@@ -173,9 +173,9 @@ Status RowIDFetcher::_merge_rpc_results(const PMultiGetRequest& request,
                 }
             }
             for (int i = 0; i < resp.binary_row_data_size(); ++i) {
-                vectorized::JsonbSerializeUtil::jsonb_to_block(
+                RETURN_IF_ERROR(vectorized::JsonbSerializeUtil::jsonb_to_block(
                         serdes, resp.binary_row_data(i).data(), resp.binary_row_data(i).size(),
-                        col_uid_to_idx, *output_block, default_values, {});
+                        col_uid_to_idx, *output_block, default_values, {}));
             }
             return Status::OK();
         }
@@ -797,10 +797,10 @@ Status RowIdStorageReader::read_doris_format_row(
                 },
                 lookup_row_data_ms));
 
-        vectorized::JsonbSerializeUtil::jsonb_to_block(
+        RETURN_IF_ERROR(vectorized::JsonbSerializeUtil::jsonb_to_block(
                 row_store_read_struct.serdes, row_store_read_struct.row_store_buffer.data(),
                 row_store_read_struct.row_store_buffer.size(), row_store_read_struct.col_uid_to_idx,
-                result_block, row_store_read_struct.default_values, {});
+                result_block, row_store_read_struct.default_values, {}));
     } else {
         for (int x = 0; x < slots.size(); ++x) {
             vectorized::MutableColumnPtr column =

be/src/exec/schema_scanner.cpp

@@ -28,6 +28,7 @@
 
 #include "exec/schema_scanner/schema_active_queries_scanner.h"
 #include "exec/schema_scanner/schema_backend_active_tasks.h"
+#include "exec/schema_scanner/schema_backend_configuration_scanner.h"
 #include "exec/schema_scanner/schema_backend_kerberos_ticket_cache.h"
 #include "exec/schema_scanner/schema_catalog_meta_cache_stats_scanner.h"
 #include "exec/schema_scanner/schema_charsets_scanner.h"
@@ -69,7 +70,6 @@
 #include "vec/columns/column_nullable.h"
 #include "vec/columns/column_string.h"
 #include "vec/columns/column_vector.h"
-#include "vec/columns/columns_number.h"
 #include "vec/common/string_ref.h"
 #include "vec/core/block.h"
 #include "vec/core/column_with_type_and_name.h"
@@ -195,6 +195,8 @@ std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type
         return SchemaFilesScanner::create_unique();
     case TSchemaTableType::SCH_PARTITIONS:
         return SchemaPartitionsScanner::create_unique();
+    case TSchemaTableType::SCH_BACKEND_CONFIGURATION:
+        return SchemaBackendConfigurationScanner::create_unique();
     case TSchemaTableType::SCH_ROWSETS:
         return SchemaRowsetsScanner::create_unique();
     case TSchemaTableType::SCH_METADATA_NAME_IDS:

be/src/exec/schema_scanner/schema_backend_configuration_scanner.cpp

@@ -0,0 +1,104 @@
+// 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 "exec/schema_scanner/schema_backend_configuration_scanner.h"
+
+#include <gen_cpp/Descriptors_types.h>
+
+#include <string>
+
+#include "runtime/define_primitive_type.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_state.h"
+#include "vec/common/string_ref.h"
+#include "vec/core/block.h"
+
+namespace doris {
+
+std::vector<SchemaScanner::ColumnDesc> SchemaBackendConfigurationScanner::_s_tbls_columns = {
+        //   name,       type,          size,     is_null
+        {"BE_ID", TYPE_BIGINT, sizeof(int64_t), true},
+        {"CONFIG_NAME", TYPE_STRING, sizeof(StringRef), true},
+        {"CONFIG_TYPE", TYPE_STRING, sizeof(StringRef), true},
+        {"CONFIG_VALUE", TYPE_STRING, sizeof(StringRef), true},
+        {"IS_MUTABLE", TYPE_BOOLEAN, sizeof(bool), true}};
+
+SchemaBackendConfigurationScanner::SchemaBackendConfigurationScanner()
+        : SchemaScanner(_s_tbls_columns, TSchemaTableType::SCH_BACKEND_CONFIGURATION),
+          _backend_id(ExecEnv::GetInstance()->cluster_info()->backend_id) {}
+
+SchemaBackendConfigurationScanner::~SchemaBackendConfigurationScanner() = default;
+
+Status SchemaBackendConfigurationScanner::start(doris::RuntimeState* state) {
+    _config_infos = config::get_config_info();
+    return Status::OK();
+}
+
+Status SchemaBackendConfigurationScanner::get_next_block_internal(vectorized::Block* block,
+                                                                  bool* eos) {
+    if (!_is_init) {
+        return Status::InternalError("Used before initialized.");
+    }
+
+    if (nullptr == block || nullptr == eos) {
+        return Status::InternalError("input pointer is nullptr.");
+    }
+
+    *eos = true;
+    if (_config_infos.empty()) {
+        return Status::OK();
+    }
+
+    for (size_t col_idx = 0; col_idx < _s_tbls_columns.size(); ++col_idx) {
+        size_t row_num = _config_infos.size();
+        std::vector<StringRef> str_refs(row_num);
+        std::vector<int8_t> bool_vals(row_num);
+        std::vector<void*> datas(row_num);
+        std::vector<std::string> column_values(row_num);
+
+        for (size_t row_idx = 0; row_idx < row_num; ++row_idx) {
+            // be_id
+            if (col_idx == 0) {
+                datas[row_idx] = &_backend_id;
+            } else {
+                // config
+                const auto& row = _config_infos[row_idx];
+                if (row.size() != _s_tbls_columns.size() - 1) {
+                    return Status::InternalError(
+                            "backend configs info meet invalid schema, schema_size={}, "
+                            "input_data_size={}",
+                            _config_infos.size(), row.size());
+                }
+
+                std::string& column_value =
+                        column_values[row_idx]; // Reference to the actual string in the vector
+                column_value = row[col_idx - 1];
+                if (_s_tbls_columns[col_idx].type == TYPE_BOOLEAN) {
+                    bool_vals[row_idx] = column_value == "true" ? 1 : 0;
+                    datas[row_idx] = &bool_vals[row_idx];
+                } else {
+                    str_refs[row_idx] =
+                            StringRef(column_values[row_idx].data(), column_values[row_idx].size());
+                    datas[row_idx] = &str_refs[row_idx];
+                }
+            }
+        }
+        RETURN_IF_ERROR(fill_dest_column_for_range(block, col_idx, datas));
+    }
+    return Status::OK();
+}
+} // namespace doris

be/src/exec/schema_scanner/schema_backend_configuration_scanner.h

@@ -0,0 +1,47 @@
+// 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.
+
+#pragma once
+
+#include <vector>
+
+#include "common/status.h"
+#include "exec/schema_scanner.h"
+
+namespace doris {
+class RuntimeState;
+namespace vectorized {
+class Block;
+} // namespace vectorized
+
+class SchemaBackendConfigurationScanner : public SchemaScanner {
+    ENABLE_FACTORY_CREATOR(SchemaBackendConfigurationScanner);
+
+public:
+    SchemaBackendConfigurationScanner();
+    ~SchemaBackendConfigurationScanner() override;
+
+    Status start(RuntimeState* state) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
+
+    static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
+
+private:
+    int64_t _backend_id;
+    std::vector<std::vector<std::string>> _config_infos;
+};
+} // namespace doris

be/src/exec/schema_scanner/schema_helper.cpp

@@ -151,4 +151,13 @@ Status SchemaHelper::fetch_routine_load_job(const std::string& ip, const int32_t
             });
 }
 
+Status SchemaHelper::fetch_schema_table_data(const std::string& ip, const int32_t port,
+                                             const TFetchSchemaTableDataRequest& request,
+                                             TFetchSchemaTableDataResult* result) {
+    return ThriftRpcHelper::rpc<FrontendServiceClient>(
+            ip, port, [&request, &result](FrontendServiceConnection& client) {
+                client->fetchSchemaTableData(*result, request);
+            });
+}
+
 } // namespace doris

be/src/exec/schema_scanner/schema_helper.h

@@ -28,6 +28,8 @@ class TDescribeTablesParams;
 class TDescribeTablesResult;
 class TFetchRoutineLoadJobRequest;
 class TFetchRoutineLoadJobResult;
+class TFetchSchemaTableDataRequest;
+class TFetchSchemaTableDataResult;
 class TGetDbsParams;
 class TGetDbsResult;
 class TGetTablesParams;
@@ -90,6 +92,10 @@ class SchemaHelper {
     static Status fetch_routine_load_job(const std::string& ip, const int32_t port,
                                          const TFetchRoutineLoadJobRequest& request,
                                          TFetchRoutineLoadJobResult* result);
+
+    static Status fetch_schema_table_data(const std::string& ip, const int32_t port,
+                                          const TFetchSchemaTableDataRequest& request,
+                                          TFetchSchemaTableDataResult* result);
 };
 
 } // namespace doris