dataroaring commented on code in PR #15832:
URL: https://github.com/apache/doris/pull/15832#discussion_r1067666463
##########
be/src/common/config.h:
##########
@@ -815,6 +815,7 @@ CONF_Int32(concurrency_per_dir, "2");
CONF_mInt64(cooldown_lag_time_sec, "10800"); // 3h
CONF_mInt64(max_sub_cache_file_size, "104857600"); // 100MB
CONF_mInt64(file_cache_alive_time_sec, "604800"); // 1 week
+CONF_Bool(cooldown_single_remote_file, "false"); // create single remote
file when cooldown.
Review Comment:
We should not add the config. If it is used during development, we can
rename it to internal_cooldown_single_remote_replica.
##########
be/src/io/fs/file_reader_options.h:
##########
@@ -39,6 +39,7 @@ class CachePathPolicy {
// path: the path of file which will be cached
// return value: the cache path of the given file.
virtual std::string get_cache_path(const std::string& path) const { return
""; }
+ std::string cache_path;
Review Comment:
Please refer to 29492f0d6c455f22aef0ae0b19f436fcd18b7d52.
##########
be/src/olap/rowset/segment_v2/segment.cpp:
##########
@@ -46,9 +46,8 @@ using io::FileCacheManager;
Status Segment::open(io::FileSystemSPtr fs, const std::string& path, uint32_t
segment_id,
RowsetId rowset_id, TabletSchemaSPtr tablet_schema,
+ const io::FileReaderOptions& reader_options,
std::shared_ptr<Segment>* output) {
- io::FileReaderOptions
reader_options(io::cache_type_from_string(config::file_cache_type),
- io::SegmentCachePathPolicy());
Review Comment:
Please refer to 29492f0d6c455f22aef0ae0b19f436fcd18b7d52.
##########
be/src/olap/rowset/beta_rowset_writer.cpp:
##########
@@ -385,8 +385,10 @@ Status BetaRowsetWriter::_load_noncompacted_segments(
auto seg_path =
BetaRowset::segment_file_path(_context.rowset_dir,
_context.rowset_id, seg_id);
std::shared_ptr<segment_v2::Segment> segment;
+ io::FileReaderOptions
reader_options(io::cache_type_from_string(config::file_cache_type),
+ io::SegmentCachePathPolicy());
auto s = segment_v2::Segment::open(fs, seg_path, seg_id, rowset_id(),
Review Comment:
Please refer to 29492f0d6c455f22aef0ae0b19f436fcd18b7d52.
##########
be/src/olap/push_handler.cpp:
##########
@@ -203,6 +203,8 @@ Status PushHandler::_convert_v2(TabletSharedPtr cur_tablet,
RowsetSharedPtr* cur
context.rowset_state = PREPARED;
context.segments_overlap = OVERLAP_UNKNOWN;
context.tablet_schema = tablet_schema;
+ context.oldest_write_timestamp = UnixSeconds();
+ context.newest_write_timestamp = UnixSeconds();
Review Comment:
timestamps are set in BetaRowsetWriter::build, we can remove either?
##########
be/src/olap/tablet_meta.h:
##########
@@ -113,7 +113,7 @@ class TabletMeta {
// Init `RowsetMeta._fs` if rowset is local.
void init_rs_metas_fs(const io::FileSystemSPtr& fs);
- void to_meta_pb(TabletMetaPB* tablet_meta_pb);
+ void to_meta_pb(bool only_remote, TabletMetaPB* tablet_meta_pb);
Review Comment:
only_include_remote_rowset is a much more specific name.
##########
be/src/olap/tablet.cpp:
##########
@@ -1683,11 +1690,23 @@ Status Tablet::cooldown() {
LOG(WARNING) << "Failed to own cumu_compaction_lock. tablet=" <<
tablet_id();
return Status::Error<TRY_LOCK_FAILED>();
}
+
+ if (!config::cooldown_single_remote_file ||
+ _tablet_meta->cooldown_replica_id() == _tablet_meta->replica_id()) {
+ RETURN_IF_ERROR(_cooldown_upload_data());
+ } else {
+ RETURN_IF_ERROR(_cooldown_use_remote_data());
+ }
+ return Status::OK();
+}
+
+Status Tablet::_cooldown_upload_data() {
Review Comment:
cooldown_data is a better name?
##########
be/src/olap/tablet.cpp:
##########
@@ -1729,18 +1757,118 @@ Status Tablet::cooldown() {
has_shutdown = tablet_state() == TABLET_SHUTDOWN;
if (!has_shutdown) {
modify_rowsets(to_add, to_delete);
- _self_owned_remote_rowsets.insert(to_add.front());
+ if (!config::cooldown_single_remote_file) {
+ _self_owned_remote_rowsets.insert(to_add.front());
+ }
save_meta();
}
}
- if (has_shutdown) {
+ if (has_shutdown && !config::cooldown_single_remote_file) {
record_unused_remote_rowset(new_rowset_id, dest_fs->resource_id(),
to_add.front()->num_segments());
return Status::Aborted("tablet {} has shutdown", tablet_id());
}
return Status::OK();
}
+Status Tablet::_read_remote_tablet_meta(FileSystemSPtr fs, TabletMetaPB*
tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->cooldown_replica_id());
+ bool exist = false;
+ RETURN_IF_ERROR(fs->exists(remote_meta_path, &exist));
+ if (exist) {
+ io::FileReaderSPtr tablet_meta_reader;
+ RETURN_IF_ERROR(fs->open_file(remote_meta_path, &tablet_meta_reader));
+ if (tablet_meta_reader == nullptr) {
+ return Status::InternalError("tablet_meta_reader is null");
+ }
+ auto file_size = tablet_meta_reader->size();
+ size_t bytes_read;
+ uint8_t* buf = new uint8_t[file_size];
+ Slice slice(buf, file_size);
+ IOContext io_ctx;
+ Status st = tablet_meta_reader->read_at(0, slice, io_ctx, &bytes_read);
+ if (!st.ok()) {
+ tablet_meta_reader->close();
+ return st;
+ }
+ tablet_meta_reader->close();
+ if (!tablet_meta_pb->ParseFromString(slice.to_string())) {
+ LOG(WARNING) << "parse tablet meta failed";
+ return Status::InternalError("parse tablet meta failed");
+ }
+ }
+ LOG(INFO) << "No tablet meta file founded, init needed. tablet_id: " <<
tablet_id();
+ return Status::InternalError("No tablet meta file founded, init needed.");
+}
+
+Status Tablet::_write_remote_tablet_meta(FileSystemSPtr fs, const
TabletMetaPB& tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->replica_id());
+ io::FileWriterPtr tablet_meta_writer;
+ RETURN_IF_ERROR(fs->create_file(remote_meta_path, &tablet_meta_writer));
+ if (tablet_meta_writer == nullptr) {
+ return Status::InternalError("tablet_meta_writer is null");
+ }
+ string value;
+ tablet_meta_pb.SerializeToString(&value);
+ uint8_t* buf = new uint8_t[value.size()];
+ memcpy(buf, value.c_str(), value.size());
+ Slice slice(buf, value.size());
+ Status st = tablet_meta_writer->appendv(&slice, 1);
+ if (!st.ok()) {
+ tablet_meta_writer->close();
+ return st;
+ }
+ tablet_meta_writer->close();
+ return Status::OK();
+}
+
+Status Tablet::_cooldown_use_remote_data() {
+ auto dest_fs = io::FileSystemMap::instance()->get(storage_policy());
+ if (!dest_fs) {
+ return Status::InternalError("storage_policy doesn't exist: " +
storage_policy());
+ }
+ DCHECK(dest_fs->type() == io::FileSystemType::S3);
+ TabletMetaPB remote_tablet_meta_pb;
+ RETURN_IF_ERROR(_read_remote_tablet_meta(dest_fs, &remote_tablet_meta_pb));
+ std::vector<RowsetSharedPtr> to_add;
+ std::vector<RowsetSharedPtr> to_delete;
+ int64_t max_version = -1;
+ for (auto& rowset_meta_pb : remote_tablet_meta_pb.rs_metas()) {
+ RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
+ rowset_meta->init_from_pb(rowset_meta_pb);
+ RowsetSharedPtr new_rowset;
+ RowsetFactory::create_rowset(_schema, _tablet_path, rowset_meta,
&new_rowset);
+ to_add.push_back(new_rowset);
+ if (rowset_meta_pb.end_version() > max_version) {
+ max_version = rowset_meta_pb.end_version();
+ }
+ }
+
+ {
+ std::shared_lock meta_rlock(_meta_lock);
+ for (const auto& it : _rs_version_map) {
+ auto& rs = it.second;
+ if (rs->end_version() <= max_version) {
+ to_delete.push_back(rs);
+ }
+ }
+ }
+
+ bool has_shutdown = false;
+ {
+ std::unique_lock meta_wlock(_meta_lock);
+ has_shutdown = tablet_state() == TABLET_SHUTDOWN;
+ if (!has_shutdown) {
Review Comment:
We can remove has_shutdown and just just use tablet_state() ==
TABLET_SHUTDOWN;
##########
be/src/olap/tablet.cpp:
##########
@@ -1729,18 +1757,118 @@ Status Tablet::cooldown() {
has_shutdown = tablet_state() == TABLET_SHUTDOWN;
if (!has_shutdown) {
modify_rowsets(to_add, to_delete);
- _self_owned_remote_rowsets.insert(to_add.front());
+ if (!config::cooldown_single_remote_file) {
+ _self_owned_remote_rowsets.insert(to_add.front());
+ }
save_meta();
}
}
- if (has_shutdown) {
+ if (has_shutdown && !config::cooldown_single_remote_file) {
record_unused_remote_rowset(new_rowset_id, dest_fs->resource_id(),
to_add.front()->num_segments());
return Status::Aborted("tablet {} has shutdown", tablet_id());
}
return Status::OK();
}
+Status Tablet::_read_remote_tablet_meta(FileSystemSPtr fs, TabletMetaPB*
tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->cooldown_replica_id());
+ bool exist = false;
+ RETURN_IF_ERROR(fs->exists(remote_meta_path, &exist));
+ if (exist) {
+ io::FileReaderSPtr tablet_meta_reader;
+ RETURN_IF_ERROR(fs->open_file(remote_meta_path, &tablet_meta_reader));
+ if (tablet_meta_reader == nullptr) {
+ return Status::InternalError("tablet_meta_reader is null");
+ }
+ auto file_size = tablet_meta_reader->size();
+ size_t bytes_read;
+ uint8_t* buf = new uint8_t[file_size];
+ Slice slice(buf, file_size);
+ IOContext io_ctx;
+ Status st = tablet_meta_reader->read_at(0, slice, io_ctx, &bytes_read);
+ if (!st.ok()) {
+ tablet_meta_reader->close();
+ return st;
+ }
+ tablet_meta_reader->close();
+ if (!tablet_meta_pb->ParseFromString(slice.to_string())) {
+ LOG(WARNING) << "parse tablet meta failed";
+ return Status::InternalError("parse tablet meta failed");
+ }
+ }
+ LOG(INFO) << "No tablet meta file founded, init needed. tablet_id: " <<
tablet_id();
+ return Status::InternalError("No tablet meta file founded, init needed.");
+}
+
+Status Tablet::_write_remote_tablet_meta(FileSystemSPtr fs, const
TabletMetaPB& tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->replica_id());
+ io::FileWriterPtr tablet_meta_writer;
+ RETURN_IF_ERROR(fs->create_file(remote_meta_path, &tablet_meta_writer));
+ if (tablet_meta_writer == nullptr) {
+ return Status::InternalError("tablet_meta_writer is null");
+ }
+ string value;
+ tablet_meta_pb.SerializeToString(&value);
+ uint8_t* buf = new uint8_t[value.size()];
+ memcpy(buf, value.c_str(), value.size());
+ Slice slice(buf, value.size());
+ Status st = tablet_meta_writer->appendv(&slice, 1);
+ if (!st.ok()) {
+ tablet_meta_writer->close();
+ return st;
+ }
+ tablet_meta_writer->close();
+ return Status::OK();
+}
+
+Status Tablet::_cooldown_use_remote_data() {
+ auto dest_fs = io::FileSystemMap::instance()->get(storage_policy());
+ if (!dest_fs) {
+ return Status::InternalError("storage_policy doesn't exist: " +
storage_policy());
+ }
+ DCHECK(dest_fs->type() == io::FileSystemType::S3);
+ TabletMetaPB remote_tablet_meta_pb;
+ RETURN_IF_ERROR(_read_remote_tablet_meta(dest_fs, &remote_tablet_meta_pb));
+ std::vector<RowsetSharedPtr> to_add;
+ std::vector<RowsetSharedPtr> to_delete;
+ int64_t max_version = -1;
+ for (auto& rowset_meta_pb : remote_tablet_meta_pb.rs_metas()) {
+ RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
+ rowset_meta->init_from_pb(rowset_meta_pb);
+ RowsetSharedPtr new_rowset;
+ RowsetFactory::create_rowset(_schema, _tablet_path, rowset_meta,
&new_rowset);
+ to_add.push_back(new_rowset);
+ if (rowset_meta_pb.end_version() > max_version) {
+ max_version = rowset_meta_pb.end_version();
+ }
+ }
+
+ {
+ std::shared_lock meta_rlock(_meta_lock);
+ for (const auto& it : _rs_version_map) {
+ auto& rs = it.second;
+ if (rs->end_version() <= max_version) {
+ to_delete.push_back(rs);
+ }
+ }
+ }
+
+ bool has_shutdown = false;
+ {
+ std::unique_lock meta_wlock(_meta_lock);
+ has_shutdown = tablet_state() == TABLET_SHUTDOWN;
+ if (!has_shutdown) {
+ modify_rowsets(to_add, to_delete);
+ save_meta();
Review Comment:
We should handle failure of saving meta.
##########
be/src/agent/task_worker_pool.cpp:
##########
@@ -1693,4 +1697,43 @@ void
TaskWorkerPool::_storage_update_storage_policy_worker_thread_callback() {
}
}
+void TaskWorkerPool::_push_cooldown_conf_worker_thread_callback() {
+ while (_is_work) {
+ TAgentTaskRequest agent_task_req;
+ TPushCooldownConfReq push_cooldown_conf_req;
+ {
+ std::unique_lock<std::mutex>
worker_thread_lock(_worker_thread_lock);
+ while (_is_work && _tasks.empty()) {
+ _worker_thread_condition_variable.wait(worker_thread_lock);
+ }
+ if (!_is_work) {
+ return;
+ }
Review Comment:
we just check _is_work in while , we can skip checking it here.
##########
be/src/olap/rowset/beta_rowset.cpp:
##########
@@ -134,7 +139,11 @@ Status
BetaRowset::load_segments(std::vector<segment_v2::SegmentSharedPtr>* segm
for (int seg_id = 0; seg_id < num_segments(); ++seg_id) {
auto seg_path = segment_file_path(seg_id);
std::shared_ptr<segment_v2::Segment> segment;
- auto s = segment_v2::Segment::open(fs, seg_path, seg_id, rowset_id(),
_schema, &segment);
+ io::FileReaderOptions
reader_options(io::cache_type_from_string(config::file_cache_type),
+ io::SegmentCachePathPolicy());
+ reader_options.path_policy.cache_path = segment_cache_path(seg_id);
+ auto s = segment_v2::Segment::open(fs, seg_path, seg_id, rowset_id(),
_schema,
Review Comment:
please refer to 29492f0d6c455f22aef0ae0b19f436fcd18b7d52.
##########
be/src/olap/tablet.cpp:
##########
@@ -1683,11 +1690,23 @@ Status Tablet::cooldown() {
LOG(WARNING) << "Failed to own cumu_compaction_lock. tablet=" <<
tablet_id();
return Status::Error<TRY_LOCK_FAILED>();
}
+
+ if (!config::cooldown_single_remote_file ||
+ _tablet_meta->cooldown_replica_id() == _tablet_meta->replica_id()) {
+ RETURN_IF_ERROR(_cooldown_upload_data());
+ } else {
+ RETURN_IF_ERROR(_cooldown_use_remote_data());
Review Comment:
follow_cooldowned_data is a better name?
##########
be/src/olap/tablet_meta.cpp:
##########
@@ -573,7 +577,9 @@ void TabletMeta::to_meta_pb(TabletMetaPB* tablet_meta_pb) {
}
for (auto& rs : _rs_metas) {
- rs->to_rowset_pb(tablet_meta_pb->add_rs_metas());
+ if (!only_remote || !rs->is_local()) {
Review Comment:
This if is not easy to be understood.
##########
be/src/olap/tablet.cpp:
##########
@@ -1717,6 +1736,15 @@ Status Tablet::cooldown() {
new_rowset_meta->set_resource_id(dest_fs->resource_id());
new_rowset_meta->set_fs(dest_fs);
new_rowset_meta->set_creation_time(time(nullptr));
+
+ if (config::cooldown_single_remote_file) {
Review Comment:
We can put meta unconditionally?
##########
be/src/olap/tablet.cpp:
##########
@@ -1729,18 +1757,118 @@ Status Tablet::cooldown() {
has_shutdown = tablet_state() == TABLET_SHUTDOWN;
if (!has_shutdown) {
modify_rowsets(to_add, to_delete);
- _self_owned_remote_rowsets.insert(to_add.front());
+ if (!config::cooldown_single_remote_file) {
+ _self_owned_remote_rowsets.insert(to_add.front());
+ }
save_meta();
}
}
- if (has_shutdown) {
+ if (has_shutdown && !config::cooldown_single_remote_file) {
record_unused_remote_rowset(new_rowset_id, dest_fs->resource_id(),
to_add.front()->num_segments());
return Status::Aborted("tablet {} has shutdown", tablet_id());
}
return Status::OK();
}
+Status Tablet::_read_remote_tablet_meta(FileSystemSPtr fs, TabletMetaPB*
tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->cooldown_replica_id());
+ bool exist = false;
+ RETURN_IF_ERROR(fs->exists(remote_meta_path, &exist));
+ if (exist) {
+ io::FileReaderSPtr tablet_meta_reader;
+ RETURN_IF_ERROR(fs->open_file(remote_meta_path, &tablet_meta_reader));
+ if (tablet_meta_reader == nullptr) {
+ return Status::InternalError("tablet_meta_reader is null");
+ }
+ auto file_size = tablet_meta_reader->size();
+ size_t bytes_read;
+ uint8_t* buf = new uint8_t[file_size];
+ Slice slice(buf, file_size);
+ IOContext io_ctx;
+ Status st = tablet_meta_reader->read_at(0, slice, io_ctx, &bytes_read);
+ if (!st.ok()) {
+ tablet_meta_reader->close();
+ return st;
+ }
+ tablet_meta_reader->close();
+ if (!tablet_meta_pb->ParseFromString(slice.to_string())) {
+ LOG(WARNING) << "parse tablet meta failed";
+ return Status::InternalError("parse tablet meta failed");
+ }
+ }
+ LOG(INFO) << "No tablet meta file founded, init needed. tablet_id: " <<
tablet_id();
+ return Status::InternalError("No tablet meta file founded, init needed.");
+}
+
+Status Tablet::_write_remote_tablet_meta(FileSystemSPtr fs, const
TabletMetaPB& tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->replica_id());
+ io::FileWriterPtr tablet_meta_writer;
+ RETURN_IF_ERROR(fs->create_file(remote_meta_path, &tablet_meta_writer));
+ if (tablet_meta_writer == nullptr) {
+ return Status::InternalError("tablet_meta_writer is null");
+ }
+ string value;
+ tablet_meta_pb.SerializeToString(&value);
+ uint8_t* buf = new uint8_t[value.size()];
+ memcpy(buf, value.c_str(), value.size());
+ Slice slice(buf, value.size());
+ Status st = tablet_meta_writer->appendv(&slice, 1);
+ if (!st.ok()) {
+ tablet_meta_writer->close();
+ return st;
+ }
+ tablet_meta_writer->close();
+ return Status::OK();
+}
+
+Status Tablet::_cooldown_use_remote_data() {
+ auto dest_fs = io::FileSystemMap::instance()->get(storage_policy());
+ if (!dest_fs) {
+ return Status::InternalError("storage_policy doesn't exist: " +
storage_policy());
+ }
+ DCHECK(dest_fs->type() == io::FileSystemType::S3);
+ TabletMetaPB remote_tablet_meta_pb;
+ RETURN_IF_ERROR(_read_remote_tablet_meta(dest_fs, &remote_tablet_meta_pb));
+ std::vector<RowsetSharedPtr> to_add;
+ std::vector<RowsetSharedPtr> to_delete;
+ int64_t max_version = -1;
+ for (auto& rowset_meta_pb : remote_tablet_meta_pb.rs_metas()) {
+ RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
+ rowset_meta->init_from_pb(rowset_meta_pb);
+ RowsetSharedPtr new_rowset;
+ RowsetFactory::create_rowset(_schema, _tablet_path, rowset_meta,
&new_rowset);
+ to_add.push_back(new_rowset);
+ if (rowset_meta_pb.end_version() > max_version) {
+ max_version = rowset_meta_pb.end_version();
+ }
+ }
+
+ {
+ std::shared_lock meta_rlock(_meta_lock);
+ for (const auto& it : _rs_version_map) {
+ auto& rs = it.second;
+ if (rs->end_version() <= max_version) {
+ to_delete.push_back(rs);
+ }
Review Comment:
Please add a check versions beyond max_version is local, otherwise we can
not follow.
##########
be/src/olap/tablet.cpp:
##########
@@ -1729,18 +1757,119 @@ Status Tablet::cooldown() {
has_shutdown = tablet_state() == TABLET_SHUTDOWN;
if (!has_shutdown) {
modify_rowsets(to_add, to_delete);
- _self_owned_remote_rowsets.insert(to_add.front());
+ if (!config::cooldown_single_remote_file) {
+ _self_owned_remote_rowsets.insert(to_add.front());
+ }
save_meta();
}
}
- if (has_shutdown) {
+ if (has_shutdown && !config::cooldown_single_remote_file) {
record_unused_remote_rowset(new_rowset_id, dest_fs->resource_id(),
to_add.front()->num_segments());
return Status::Aborted("tablet {} has shutdown", tablet_id());
}
return Status::OK();
}
+Status Tablet::_read_remote_tablet_meta(FileSystemSPtr fs, TabletMetaPB*
tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->cooldown_replica_id());
+ bool exist = false;
+ RETURN_IF_ERROR(fs->exists(remote_meta_path, &exist));
+ if (exist) {
+ IOContext io_ctx;
+ io::FileReaderSPtr tablet_meta_reader;
+ RETURN_IF_ERROR(fs->open_file(remote_meta_path, &tablet_meta_reader,
&io_ctx));
+ if (tablet_meta_reader == nullptr) {
+ return Status::InternalError("tablet_meta_reader is null");
+ }
+ auto file_size = tablet_meta_reader->size();
+ size_t bytes_read;
+ uint8_t* buf = new uint8_t[file_size];
+ Slice slice(buf, file_size);
+ IOContext io_ctx;
+ Status st = tablet_meta_reader->read_at(0, slice, io_ctx, &bytes_read);
+ if (!st.ok()) {
+ tablet_meta_reader->close();
+ return st;
+ }
+ tablet_meta_reader->close();
+ if (!tablet_meta_pb->ParseFromString(slice.to_string())) {
+ LOG(WARNING) << "parse tablet meta failed";
+ return Status::InternalError("parse tablet meta failed");
+ }
+ }
+ LOG(INFO) << "No tablet meta file founded, init needed. tablet_id: " <<
tablet_id();
+ return Status::InternalError("No tablet meta file founded, init needed.");
+}
+
+Status Tablet::_write_remote_tablet_meta(FileSystemSPtr fs, const
TabletMetaPB& tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->replica_id());
+ io::FileWriterPtr tablet_meta_writer;
+ RETURN_IF_ERROR(fs->create_file(remote_meta_path, &tablet_meta_writer));
+ if (tablet_meta_writer == nullptr) {
+ return Status::InternalError("tablet_meta_writer is null");
+ }
+ string value;
+ tablet_meta_pb.SerializeToString(&value);
+ uint8_t* buf = new uint8_t[value.size()];
+ memcpy(buf, value.c_str(), value.size());
+ Slice slice(buf, value.size());
+ Status st = tablet_meta_writer->appendv(&slice, 1);
+ if (!st.ok()) {
+ tablet_meta_writer->close();
+ return st;
+ }
+ tablet_meta_writer->close();
+ return Status::OK();
+}
+
+Status Tablet::_cooldown_use_remote_data() {
+ auto dest_fs = io::FileSystemMap::instance()->get(storage_policy());
+ if (!dest_fs) {
Review Comment:
We should check that version is contiguous.
##########
be/src/olap/tablet.cpp:
##########
@@ -1729,18 +1757,118 @@ Status Tablet::cooldown() {
has_shutdown = tablet_state() == TABLET_SHUTDOWN;
if (!has_shutdown) {
modify_rowsets(to_add, to_delete);
- _self_owned_remote_rowsets.insert(to_add.front());
+ if (!config::cooldown_single_remote_file) {
+ _self_owned_remote_rowsets.insert(to_add.front());
+ }
save_meta();
}
}
- if (has_shutdown) {
+ if (has_shutdown && !config::cooldown_single_remote_file) {
record_unused_remote_rowset(new_rowset_id, dest_fs->resource_id(),
to_add.front()->num_segments());
return Status::Aborted("tablet {} has shutdown", tablet_id());
}
return Status::OK();
}
+Status Tablet::_read_remote_tablet_meta(FileSystemSPtr fs, TabletMetaPB*
tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->cooldown_replica_id());
+ bool exist = false;
+ RETURN_IF_ERROR(fs->exists(remote_meta_path, &exist));
+ if (exist) {
+ io::FileReaderSPtr tablet_meta_reader;
+ RETURN_IF_ERROR(fs->open_file(remote_meta_path, &tablet_meta_reader));
+ if (tablet_meta_reader == nullptr) {
+ return Status::InternalError("tablet_meta_reader is null");
+ }
+ auto file_size = tablet_meta_reader->size();
+ size_t bytes_read;
+ uint8_t* buf = new uint8_t[file_size];
+ Slice slice(buf, file_size);
+ IOContext io_ctx;
+ Status st = tablet_meta_reader->read_at(0, slice, io_ctx, &bytes_read);
+ if (!st.ok()) {
+ tablet_meta_reader->close();
+ return st;
+ }
+ tablet_meta_reader->close();
+ if (!tablet_meta_pb->ParseFromString(slice.to_string())) {
+ LOG(WARNING) << "parse tablet meta failed";
+ return Status::InternalError("parse tablet meta failed");
+ }
+ }
+ LOG(INFO) << "No tablet meta file founded, init needed. tablet_id: " <<
tablet_id();
+ return Status::InternalError("No tablet meta file founded, init needed.");
+}
+
+Status Tablet::_write_remote_tablet_meta(FileSystemSPtr fs, const
TabletMetaPB& tablet_meta_pb) {
+ std::string remote_meta_path =
+ BetaRowset::remote_tablet_meta_path(tablet_id(),
_tablet_meta->replica_id());
+ io::FileWriterPtr tablet_meta_writer;
+ RETURN_IF_ERROR(fs->create_file(remote_meta_path, &tablet_meta_writer));
+ if (tablet_meta_writer == nullptr) {
+ return Status::InternalError("tablet_meta_writer is null");
+ }
+ string value;
+ tablet_meta_pb.SerializeToString(&value);
+ uint8_t* buf = new uint8_t[value.size()];
+ memcpy(buf, value.c_str(), value.size());
+ Slice slice(buf, value.size());
+ Status st = tablet_meta_writer->appendv(&slice, 1);
+ if (!st.ok()) {
+ tablet_meta_writer->close();
+ return st;
+ }
+ tablet_meta_writer->close();
+ return Status::OK();
+}
+
+Status Tablet::_cooldown_use_remote_data() {
+ auto dest_fs = io::FileSystemMap::instance()->get(storage_policy());
+ if (!dest_fs) {
+ return Status::InternalError("storage_policy doesn't exist: " +
storage_policy());
+ }
+ DCHECK(dest_fs->type() == io::FileSystemType::S3);
+ TabletMetaPB remote_tablet_meta_pb;
+ RETURN_IF_ERROR(_read_remote_tablet_meta(dest_fs, &remote_tablet_meta_pb));
+ std::vector<RowsetSharedPtr> to_add;
+ std::vector<RowsetSharedPtr> to_delete;
+ int64_t max_version = -1;
+ for (auto& rowset_meta_pb : remote_tablet_meta_pb.rs_metas()) {
+ RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
+ rowset_meta->init_from_pb(rowset_meta_pb);
+ RowsetSharedPtr new_rowset;
+ RowsetFactory::create_rowset(_schema, _tablet_path, rowset_meta,
&new_rowset);
+ to_add.push_back(new_rowset);
+ if (rowset_meta_pb.end_version() > max_version) {
+ max_version = rowset_meta_pb.end_version();
+ }
Review Comment:
Please check versions which are continuous.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
