Merge branch 'master' into fix-replica

.github/workflows/build-extension.yml

@@ -61,7 +61,7 @@ jobs:
  uses: actions/setup-java@v2
  with:
  distribution: adopt
- java-version: '8'
+ java-version: '17'
 
  - name: Setup thrift
  run: |

be/src/agent/heartbeat_server.cpp

@@ -30,6 +30,7 @@
 #include "common/status.h"
 #include "olap/storage_engine.h"
 #include "runtime/exec_env.h"
+#include "runtime/fragment_mgr.h"
 #include "runtime/heartbeat_flags.h"
 #include "service/backend_options.h"
 #include "util/debug_util.h"
@@ -83,6 +84,10 @@ void HeartbeatServer::heartbeat(THeartbeatResult& heartbeat_result,
  heartbeat_result.backend_info.__set_be_node_role(config::be_node_role);
  // If be is gracefully stop, then k_doris_exist is set to true
  heartbeat_result.backend_info.__set_is_shutdown(doris::k_doris_exit);
+ heartbeat_result.backend_info.__set_fragment_executing_count(
+ get_fragment_executing_count());
+ heartbeat_result.backend_info.__set_fragment_last_active_time(
+ get_fragment_last_active_time());
  }
  watch.stop();
  if (watch.elapsed_time() > 1000L * 1000L * 1000L) {

be/src/agent/task_worker_pool.cpp

@@ -76,6 +76,7 @@
 #include "olap/txn_manager.h"
 #include "olap/utils.h"
 #include "runtime/exec_env.h"
+#include "runtime/fragment_mgr.h"
 #include "runtime/snapshot_loader.h"
 #include "service/backend_options.h"
 #include "util/doris_metrics.h"
@@ -446,7 +447,6 @@ void add_task_count(const TAgentTaskRequest& task, int n) {
  ADD_TASK_COUNT(PUBLISH_VERSION)
  ADD_TASK_COUNT(CLEAR_TRANSACTION_TASK)
  ADD_TASK_COUNT(UPDATE_TABLET_META_INFO)
- ADD_TASK_COUNT(ALTER)
  ADD_TASK_COUNT(CLONE)
  ADD_TASK_COUNT(STORAGE_MEDIUM_MIGRATE)
  ADD_TASK_COUNT(GC_BINLOG)
@@ -459,6 +459,17 @@ void add_task_count(const TAgentTaskRequest& task, int n) {
  DELETE_count << n;
  }
  return;
+ case TTaskType::ALTER:
+ {
+ ALTER_count << n;
+ // cloud auto stop need sc jobs, a tablet's sc can also be considered a fragment
+ doris::g_fragment_executing_count << 1;
+ int64 now = duration_cast<std::chrono::milliseconds>(
+ std::chrono::system_clock::now().time_since_epoch())
+ .count();
+ g_fragment_last_active_time.set_value(now);
+ return;
+ }
  default:
  return;
  }
@@ -1387,12 +1398,14 @@ void update_s3_resource(const TStorageResource& param, io::RemoteFileSystemSPtr
 void update_hdfs_resource(const TStorageResource& param, io::RemoteFileSystemSPtr existed_fs) {
  Status st;
  io::RemoteFileSystemSPtr fs;
+ std::string root_path =
+ param.hdfs_storage_param.__isset.root_path ? param.hdfs_storage_param.root_path : "";
 
  if (!existed_fs) {
  // No such FS instance on BE
- auto res = io::HdfsFileSystem::create(param.hdfs_storage_param,
-   param.hdfs_storage_param.fs_name,
-  std::to_string(param.id), nullptr);
+ auto res = io::HdfsFileSystem::create(
+ param.hdfs_storage_param, param.hdfs_storage_param.fs_name,
+ std::to_string(param.id), nullptr, std::move(root_path));
  if (!res.has_value()) {
  st = std::move(res).error();
  } else {
@@ -1410,7 +1423,8 @@ void update_hdfs_resource(const TStorageResource& param, io::RemoteFileSystemSPt
  } else {
  LOG_INFO("successfully update hdfs resource")
  .tag("resource_id", param.id)
- .tag("resource_name", param.name);
+ .tag("resource_name", param.name)
+ .tag("root_path", fs->root_path().string());
  put_storage_resource(param.id, {std::move(fs), param.version});
  }
 }
@@ -1851,6 +1865,11 @@ void alter_tablet_callback(StorageEngine& engine, const TAgentTaskRequest& req)
  alter_tablet(engine, req, signature, task_type, &finish_task_request);
  finish_task(finish_task_request);
  }
+ doris::g_fragment_executing_count << -1;
+ int64 now = duration_cast<std::chrono::milliseconds>(
+ std::chrono::system_clock::now().time_since_epoch())
+ .count();
+ g_fragment_last_active_time.set_value(now);
  remove_task_info(req.task_type, req.signature);
 }
 
@@ -1872,6 +1891,11 @@ void alter_cloud_tablet_callback(CloudStorageEngine& engine, const TAgentTaskReq
  alter_cloud_tablet(engine, req, signature, task_type, &finish_task_request);
  finish_task(finish_task_request);
  }
+ doris::g_fragment_executing_count << -1;
+ int64 now = duration_cast<std::chrono::milliseconds>(
+ std::chrono::system_clock::now().time_since_epoch())
+ .count();
+ g_fragment_last_active_time.set_value(now);
  remove_task_info(req.task_type, req.signature);
 }
 

be/src/cloud/cloud_rowset_builder.cpp

@@ -65,6 +65,7 @@ Status CloudRowsetBuilder::init() {
  context.mow_context = mow_context;
  context.write_file_cache = _req.write_file_cache;
  context.partial_update_info = _partial_update_info;
+ context.file_cache_ttl_sec = _tablet->ttl_seconds();
  // New loaded data is always written to latest shared storage
  // TODO(AlexYue): use the passed resource id to retrive the corresponding
  // fs to pass to the RowsetWriterContext

be/src/cloud/cloud_tablet.cpp

@@ -95,11 +95,6 @@ Status CloudTablet::capture_rs_readers(const Version& spec_version,
  return capture_rs_readers_unlocked(version_path, rs_splits);
 }
 
-Status CloudTablet::sync_meta() {
- // TODO(lightman): FileCache
- return Status::NotSupported("CloudTablet::sync_meta is not implemented");
-}
-
 // There are only two tablet_states RUNNING and NOT_READY in cloud mode
 // This function will erase the tablet from `CloudTabletMgr` when it can't find this tablet in MS.
 Status CloudTablet::sync_rowsets(int64_t query_version, bool warmup_delta_data) {
@@ -618,4 +613,82 @@ Status CloudTablet::calc_delete_bitmap_for_compaciton(
  return Status::OK();
 }
 
+Status CloudTablet::sync_meta() {
+ if (!config::enable_file_cache) {
+ return Status::OK();
+ }
+
+ TabletMetaSharedPtr tablet_meta;
+ auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta);
+ if (!st.ok()) {
+ if (st.is<ErrorCode::NOT_FOUND>()) {
+ // TODO(Lchangliang): recycle_resources_by_self();
+ }
+ return st;
+ }
+ if (tablet_meta->tablet_state() != TABLET_RUNNING) { // impossible
+ return Status::InternalError("invalid tablet state. tablet_id={}", tablet_id());
+ }
+
+ auto new_ttl_seconds = tablet_meta->ttl_seconds();
+ if (_tablet_meta->ttl_seconds() != new_ttl_seconds) {
+ _tablet_meta->set_ttl_seconds(new_ttl_seconds);
+ int64_t cur_time = UnixSeconds();
+ std::shared_lock rlock(_meta_lock);
+ for (auto& [_, rs] : _rs_version_map) {
+ for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) {
+ int64_t new_expiration_time =
+ new_ttl_seconds + rs->rowset_meta()->newest_write_timestamp();
+ new_expiration_time = new_expiration_time > cur_time ? new_expiration_time : 0;
+ auto file_key = io::BlockFileCache::hash(
+ io::Path(rs->segment_file_path(seg_id)).filename().native());
+ auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);
+ file_cache->modify_expiration_time(file_key, new_expiration_time);
+ }
+ }
+ }
+
+ auto new_compaction_policy = tablet_meta->compaction_policy();
+ if (_tablet_meta->compaction_policy() != new_compaction_policy) {
+ _tablet_meta->set_compaction_policy(new_compaction_policy);
+ }
+ auto new_time_series_compaction_goal_size_mbytes =
+ tablet_meta->time_series_compaction_goal_size_mbytes();
+ if (_tablet_meta->time_series_compaction_goal_size_mbytes() !=
+ new_time_series_compaction_goal_size_mbytes) {
+ _tablet_meta->set_time_series_compaction_goal_size_mbytes(
+ new_time_series_compaction_goal_size_mbytes);
+ }
+ auto new_time_series_compaction_file_count_threshold =
+ tablet_meta->time_series_compaction_file_count_threshold();
+ if (_tablet_meta->time_series_compaction_file_count_threshold() !=
+ new_time_series_compaction_file_count_threshold) {
+ _tablet_meta->set_time_series_compaction_file_count_threshold(
+ new_time_series_compaction_file_count_threshold);
+ }
+ auto new_time_series_compaction_time_threshold_seconds =
+ tablet_meta->time_series_compaction_time_threshold_seconds();
+ if (_tablet_meta->time_series_compaction_time_threshold_seconds() !=
+ new_time_series_compaction_time_threshold_seconds) {
+ _tablet_meta->set_time_series_compaction_time_threshold_seconds(
+ new_time_series_compaction_time_threshold_seconds);
+ }
+ auto new_time_series_compaction_empty_rowsets_threshold =
+ tablet_meta->time_series_compaction_empty_rowsets_threshold();
+ if (_tablet_meta->time_series_compaction_empty_rowsets_threshold() !=
+ new_time_series_compaction_empty_rowsets_threshold) {
+ _tablet_meta->set_time_series_compaction_empty_rowsets_threshold(
+ new_time_series_compaction_empty_rowsets_threshold);
+ }
+ auto new_time_series_compaction_level_threshold =
+ tablet_meta->time_series_compaction_level_threshold();
+ if (_tablet_meta->time_series_compaction_level_threshold() !=
+ new_time_series_compaction_level_threshold) {
+ _tablet_meta->set_time_series_compaction_level_threshold(
+ new_time_series_compaction_level_threshold);
+ }
+
+ return Status::OK();
+}
+
 } // namespace doris

be/src/common/config.cpp

@@ -603,7 +603,7 @@ DEFINE_mInt32(result_buffer_cancelled_interval_time, "300");
 DEFINE_mInt32(priority_queue_remaining_tasks_increased_frequency, "512");
 
 // sync tablet_meta when modifying meta
-DEFINE_mBool(sync_tablet_meta, "true");
+DEFINE_mBool(sync_tablet_meta, "false");
 
 // default thrift rpc timeout ms
 DEFINE_mInt32(thrift_rpc_timeout_ms, "60000");
@@ -617,9 +617,9 @@ DEFINE_Bool(enable_metric_calculator, "true");
 // max consumer num in one data consumer group, for routine load
 DEFINE_mInt32(max_consumer_num_per_group, "3");
 
-// the size of thread pool for routine load task.
+// the max size of thread pool for routine load task.
 // this should be larger than FE config 'max_routine_load_task_num_per_be' (default 5)
-DEFINE_Int32(routine_load_thread_pool_size, "10");
+DEFINE_Int32(max_routine_load_thread_pool_size, "1024");
 
 // max external scan cache batch count, means cache max_memory_cache_batch_count * batch_size row
 // default is 20, batch_size's default value is 1024 means 20 * 1024 rows will be cached
@@ -979,6 +979,7 @@ DEFINE_String(inverted_index_searcher_cache_limit, "10%");
 DEFINE_Bool(enable_write_index_searcher_cache, "true");
 DEFINE_Bool(enable_inverted_index_cache_check_timestamp, "true");
 DEFINE_Int32(inverted_index_fd_number_limit_percent, "40"); // 40%
+DEFINE_Int32(inverted_index_query_cache_shards, "256");
 
 // inverted index match bitmap cache size
 DEFINE_String(inverted_index_query_cache_limit, "10%");

be/src/common/config.h

@@ -667,9 +667,9 @@ DECLARE_Bool(enable_metric_calculator);
 // max consumer num in one data consumer group, for routine load
 DECLARE_mInt32(max_consumer_num_per_group);
 
-// the size of thread pool for routine load task.
+// the max size of thread pool for routine load task.
 // this should be larger than FE config 'max_routine_load_task_num_per_be' (default 5)
-DECLARE_Int32(routine_load_thread_pool_size);
+DECLARE_Int32(max_routine_load_thread_pool_size);
 
 // max external scan cache batch count, means cache max_memory_cache_batch_count * batch_size row
 // default is 20, batch_size's default value is 1024 means 20 * 1024 rows will be cached
@@ -1027,6 +1027,7 @@ DECLARE_String(inverted_index_searcher_cache_limit);
 DECLARE_Bool(enable_write_index_searcher_cache);
 DECLARE_Bool(enable_inverted_index_cache_check_timestamp);
 DECLARE_Int32(inverted_index_fd_number_limit_percent); // 50%
+DECLARE_Int32(inverted_index_query_cache_shards);
 
 // inverted index match bitmap cache size
 DECLARE_String(inverted_index_query_cache_limit);

be/src/io/cache/block_file_cache.h

@@ -21,6 +21,7 @@
 
 #include <memory>
 #include <optional>
+#include <thread>
 
 #include "io/cache/file_block.h"
 #include "io/cache/file_cache_common.h"

be/src/io/fs/hdfs_file_system.cpp

@@ -53,30 +53,31 @@ namespace doris::io {
 
 Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(
  const std::map<std::string, std::string>& properties, std::string fs_name, std::string id,
- RuntimeProfile* profile) {
+ RuntimeProfile* profile, std::string root_path) {
  return HdfsFileSystem::create(parse_properties(properties), std::move(fs_name), std::move(id),
- profile);
+ profile, std::move(root_path));
 }
 
 Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(const THdfsParams& hdfs_params,
  std::string fs_name, std::string id,
- RuntimeProfile* profile) {
+ RuntimeProfile* profile,
+ std::string root_path) {
 #ifdef USE_HADOOP_HDFS
  if (!config::enable_java_support) {
  return ResultError(Status::InternalError(
  "hdfs file system is not enabled, you can change be config enable_java_support to "
  "true."));
  }
 #endif
- std::shared_ptr<HdfsFileSystem> fs(
- new HdfsFileSystem(hdfs_params, std::move(fs_name), std::move(id), profile));
+ std::shared_ptr<HdfsFileSystem> fs(new HdfsFileSystem(
+ hdfs_params, std::move(fs_name), std::move(id), profile, std::move(root_path)));
  RETURN_IF_ERROR_RESULT(fs->init());
  return fs;
 }
 
 HdfsFileSystem::HdfsFileSystem(const THdfsParams& hdfs_params, std::string fs_name, std::string id,
- RuntimeProfile* profile)
- : RemoteFileSystem("", std::move(id), FileSystemType::HDFS),
+ RuntimeProfile* profile, std::string root_path)
+ : RemoteFileSystem(root_path, std::move(id), FileSystemType::HDFS),
  _hdfs_params(hdfs_params),
  _fs_name(std::move(fs_name)),
  _profile(profile) {

be/src/io/fs/hdfs_file_system.h

@@ -48,11 +48,12 @@ class HdfsFileSystem final : public RemoteFileSystem {
 public:
  static Result<std::shared_ptr<HdfsFileSystem>> create(const THdfsParams& hdfs_params,
  std::string fs_name, std::string id,
- RuntimeProfile* profile);
+ RuntimeProfile* profile,
+ std::string root_path = "");
 
  static Result<std::shared_ptr<HdfsFileSystem>> create(
  const std::map<std::string, std::string>& properties, std::string fs_name,
- std::string id, RuntimeProfile* profile);
+ std::string id, RuntimeProfile* profile, std::string root_path = "");
 
  ~HdfsFileSystem() override;
 
@@ -84,7 +85,7 @@ class HdfsFileSystem final : public RemoteFileSystem {
 private:
  friend class HdfsFileWriter;
  HdfsFileSystem(const THdfsParams& hdfs_params, std::string fs_name, std::string id,
- RuntimeProfile* profile);
+ RuntimeProfile* profile, std::string root_path);
  const THdfsParams& _hdfs_params; // Only used in init, so we can use reference here
  std::string _fs_name;
  // do not use std::shared_ptr or std::unique_ptr

be/src/io/fs/s3_file_writer.cpp

@@ -508,11 +508,12 @@ void S3FileWriter::_put_object(UploadFileBuffer& buf) {
 }
 
 std::string S3FileWriter::_dump_completed_part() const {
- std::string view;
+ std::stringstream ss;
+ ss << "part_numbers:";
  for (const auto& part : _completed_parts) {
- view.append(fmt::format("part {}, ", view, part->GetPartNumber()));
+ ss << " " << part->GetPartNumber();
  }
- return view;
+ return ss.str();
 }
 
 } // namespace doris::io

be/src/olap/base_tablet.h

@@ -60,6 +60,7 @@ class BaseTablet {
  int32_t schema_hash() const { return _tablet_meta->schema_hash(); }
  KeysType keys_type() const { return _tablet_meta->tablet_schema()->keys_type(); }
  size_t num_key_columns() const { return _tablet_meta->tablet_schema()->num_key_columns(); }
+ int64_t ttl_seconds() const { return _tablet_meta->ttl_seconds(); }
  std::mutex& get_schema_change_lock() { return _schema_change_lock; }
  bool enable_unique_key_merge_on_write() const {
 #ifdef BE_TEST

be/src/olap/compaction.cpp

@@ -617,7 +617,8 @@ Status CompactionMixin::do_inverted_index_compaction() {
 
  // we choose the first destination segment name as the temporary index writer path
  // Used to distinguish between different index compaction
- auto index_tmp_path = tablet_path + "/" + dest_rowset_id.to_string() + "_" + "tmp";
+ auto tmp_file_dir = ExecEnv::GetInstance()->get_tmp_file_dirs()->get_tmp_file_dir();
+ auto index_tmp_path = tmp_file_dir / dest_rowset_id.to_string();
  LOG(INFO) << "start index compaction"
  << ". tablet=" << _tablet->tablet_id() << ", source index size=" << src_segment_num
  << ", destination index size=" << dest_segment_num << ".";